LQAgentPlatform/utils_test/Semantic_Logic_Test/semantic_logic_server.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
语义逻辑审查 — 前端测试服务器
提供独立HTTP API，直接调用 SemanticLogicReviewer，无需文件上传流程
"""

import sys
import os
import json
import time
import asyncio
import threading
from http.server import HTTPServer, SimpleHTTPRequestHandler
from urllib.parse import parse_qs, urlparse

# 添加项目根目录到路径
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, PROJECT_ROOT)
os.chdir(PROJECT_ROOT)

from foundation.observability.logger.loggering import review_logger as logger
from core.construction_review.component.reviewers.semantic_logic import SemanticLogicReviewer


# ─── 异步工具 ──────────────────────────────────────────────────────────────────

def run_async(coro):
    """在同步上下文中运行异步协程"""
    try:
        loop = asyncio.get_running_loop()
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor() as executor:
            future = executor.submit(asyncio.run, coro)
            return future.result()
    except RuntimeError:
        return asyncio.run(coro)


# ─── 审查逻辑 ──────────────────────────────────────────────────────────────────

async def do_semantic_logic_check(review_content: str) -> dict:
    """
    执行语义逻辑审查
    直接调用 SemanticLogicReviewer.check_semantic_logic
    """
    trace_id = f"semantic_logic_web_{int(time.time() * 1000)}"
    reviewer = SemanticLogicReviewer()

    logger.info(f"[语义逻辑Web测试] trace_id={trace_id}, content_length={len(review_content)}")

    start = time.time()
    result = await reviewer.check_semantic_logic(
        trace_id=trace_id,
        review_content=review_content,
        state=None,
        stage_name=None,
    )
    wall_time = time.time() - start

    return {
        "trace_id": trace_id,
        "success": result.success,
        "details": result.details,
        "error_message": result.error_message,
        "model_execution_time": result.execution_time,
        "wall_time": round(wall_time, 3),
        "content_length": len(review_content),
    }


# ─── HTTP Handler ──────────────────────────────────────────────────────────────

class SemanticLogicHandler(SimpleHTTPRequestHandler):
    """HTTP请求处理器"""

    def do_GET(self):
        parsed = urlparse(self.path)

        if parsed.path == '/api/health':
            self.send_json_response({"status": "ok"})
        elif parsed.path in ('', '/', '/index.html'):
            # 返回前端页面
            index_path = os.path.join(os.path.dirname(__file__), 'index.html')
            self.serve_file(index_path, 'text/html')
        else:
            # 静态文件服务
            super().do_GET()

    def do_POST(self):
        parsed = urlparse(self.path)

        if parsed.path == '/api/semantic_logic':
            content_length = int(self.headers.get('Content-Length', 0))
            post_data = self.rfile.read(content_length)

            try:
                body = json.loads(post_data.decode('utf-8'))
                review_content = body.get('content', '')

                if not review_content:
                    self.send_json_response({"error": "请提供 content 参数"}, 400)
                    return

                print(f"\n[语义逻辑Web测试] 收到请求, content_length={len(review_content)}")
                result = run_async(do_semantic_logic_check(review_content))
                print(f"[语义逻辑Web测试] 完成, success={result['success']}, wall_time={result['wall_time']}s")

                self.send_json_response(result)

            except json.JSONDecodeError:
                self.send_json_response({"error": "JSON解析失败"}, 400)
            except Exception as e:
                logger.error(f"[语义逻辑Web测试] 处理失败: {e}", exc_info=True)
                self.send_json_response({"error": str(e)}, 500)
        else:
            self.send_json_response({"error": "Not Found"}, 404)

    def do_OPTIONS(self):
        self.send_response(200)
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.send_header('Access-Control-Allow-Headers', 'Content-Type')
        self.end_headers()

    def send_json_response(self, data, status=200):
        self.send_response(status)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.end_headers()
        self.wfile.write(json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8'))

    def serve_file(self, filepath: str, content_type: str):
        if os.path.exists(filepath):
            self.send_response(200)
            self.send_header('Content-Type', f'{content_type}; charset=utf-8')
            self.end_headers()
            with open(filepath, 'rb') as f:
                self.wfile.write(f.read())
        else:
            self.send_json_response({"error": f"文件不存在: {filepath}"}, 404)

    def end_headers(self):
        self.send_header('Access-Control-Allow-Origin', '*')
        super().end_headers()


# ─── 启动服务器 ────────────────────────────────────────────────────────────────

def run_server(port=8766):
    server = HTTPServer(('0.0.0.0', port), SemanticLogicHandler)
    print(f"\n{'='*70}")
    print(f" 语义逻辑审查 — 前端测试服务器")
    print(f"{'='*70}")
    print(f" 访问地址: http://localhost:{port}")
    print(f" API端点:  POST /api/semantic_logic")
    print(f"{'='*70}\n")
    server.serve_forever()


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description='语义逻辑审查前端测试服务器')
    parser.add_argument('--port', type=int, default=8766, help='服务端口 (默认: 8766)')
    args = parser.parse_args()
    run_server(args.port)