web_app.py

# -*- coding: utf-8 -*-
"""
Flask Web 应用

提供 Web 界面用于测试 AI Agent
"""

import os
import json
import time
import uuid
from datetime import datetime
from flask import Flask, render_template, jsonify, request, Response, stream_with_context
from core.agent_client import AgentClient
from core.evaluator import Evaluator
from history import HistoryManager

app = Flask(__name__)

client = AgentClient()
evaluator = Evaluator()
history_mgr = HistoryManager()

TESTCASES_DIR = "testcases"
REPORTS_DIR = "reports"
API_CONFIG_DIR = "config"

USE_LLM_EVAL = False
LLM_API_URL = "http://localhost:8000"


def load_test_cases():
    """加载测试用例"""
    json_file = os.path.join(TESTCASES_DIR, "testcases.json")
    if not os.path.exists(json_file):
        return []

    with open(json_file, "r", encoding="utf-8") as f:
        data = json.load(f)

    all_cases = []
    for category, cases in data.items():
        for case in cases:
            keywords_str = "|".join(case.get("expect_keywords", []))
            all_cases.append({
                "id": case.get("id", ""),
                "question": case.get("question", ""),
                "expect_intent": case.get("expect_intent", ""),
                "expect_keywords": keywords_str,
                "category": category
            })
    return all_cases


def save_test_cases(data):
    """保存测试用例"""
    json_file = os.path.join(TESTCASES_DIR, "testcases.json")
    with open(json_file, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


@app.route("/")
def index():
    """主页"""
    return render_template("index.html")


@app.route("/api/testcases")
def get_testcases():
    """获取测试用例列表（支持分页）"""
    page = request.args.get("page", 1, type=int)
    size = request.args.get("size", 10, type=int)
    category = request.args.get("category", "")
    
    testcases = load_test_cases()
    
    # 过滤分类
    if category:
        testcases = [c for c in testcases if c.get("category") == category]
    
    total = len(testcases)
    start = (page - 1) * size
    end = start + size
    
    return jsonify({
        "cases": testcases[start:end],
        "total": total,
        "page": page,
        "size": size
    })


@app.route("/api/testcases", methods=["POST"])
def add_testcase():
    """添加测试用例"""
    data = request.json
    json_file = os.path.join(TESTCASES_DIR, "testcases.json")
    
    if not os.path.exists(json_file):
        return jsonify({"success": False, "error": "测试用例文件不存在"}), 400
    
    with open(json_file, "r", encoding="utf-8") as f:
        testcases = json.load(f)
    
    category = data.get("category", "default")
    if category not in testcases:
        testcases[category] = []
    
    # 生成全局唯一 ID (UUID 前 8 位)
    new_id = str(uuid.uuid4())[:8]
    new_case = {
        "id": new_id,
        "question": data.get("question", ""),
        "expect_intent": data.get("expect_intent", ""),
        "expect_keywords": data.get("expect_keywords", [])
    }
    
    testcases[category].append(new_case)
    save_test_cases(testcases)
    
    return jsonify({"success": True, "case": new_case, "category": category})


@app.route("/api/testcases/<category>/<case_id>", methods=["PUT"])
def update_testcase(category, case_id):
    """更新测试用例"""
    data = request.json
    json_file = os.path.join(TESTCASES_DIR, "testcases.json")
    
    if not os.path.exists(json_file):
        return jsonify({"success": False, "error": "测试用例文件不存在"}), 400
    
    with open(json_file, "r", encoding="utf-8") as f:
        testcases = json.load(f)
    
    if category not in testcases:
        return jsonify({"success": False, "error": "分类不存在"}), 404
    
    for case in testcases[category]:
        if str(case.get("id")) == str(case_id):
            case["question"] = data.get("question", case.get("question", ""))
            case["expect_intent"] = data.get("expect_intent", case.get("expect_intent", ""))
            case["expect_keywords"] = data.get("expect_keywords", case.get("expect_keywords", []))
            save_test_cases(testcases)
            return jsonify({"success": True, "case": case})
    
    return jsonify({"success": False, "error": "测试用例不存在"}), 404


@app.route("/api/testcases/<category>/<case_id>", methods=["DELETE"])
def delete_testcase(category, case_id):
    """删除测试用例"""
    json_file = os.path.join(TESTCASES_DIR, "testcases.json")
    
    if not os.path.exists(json_file):
        return jsonify({"success": False, "error": "测试用例文件不存在"}), 400
    
    with open(json_file, "r", encoding="utf-8") as f:
        testcases = json.load(f)
    
    if category not in testcases:
        return jsonify({"success": False, "error": "分类不存在"}), 404
    
    testcases[category] = [c for c in testcases[category] if str(c.get("id")) != str(case_id)]
    save_test_cases(testcases)
    
    return jsonify({"success": True})


@app.route("/api/categories")
def get_categories():
    """获取测试用例分类"""
    json_file = os.path.join(TESTCASES_DIR, "testcases.json")
    if not os.path.exists(json_file):
        return jsonify(["default"])
    
    with open(json_file, "r", encoding="utf-8") as f:
        testcases = json.load(f)
    
    return jsonify(list(testcases.keys()))


@app.route("/api/config", methods=["GET"])
def get_config():
    """获取当前配置"""
    return jsonify({
        "use_llm_eval": USE_LLM_EVAL,
        "llm_api_url": LLM_API_URL
    })


@app.route("/api/config", methods=["POST"])
def set_config():
    """设置评测配置"""
    global USE_LLM_EVAL, LLM_API_URL
    data = request.json
    USE_LLM_EVAL = data.get("use_llm", False)
    LLM_API_URL = data.get("llm_api_url", "http://localhost:8000")
    evaluator.set_llm_mode(USE_LLM_EVAL, LLM_API_URL)
    return jsonify({"success": True})


@app.route("/api/api-configs", methods=["GET"])
def get_api_configs():
    """获取API配置列表"""
    config_file = os.path.join(API_CONFIG_DIR, "api_configs.json")
    if not os.path.exists(config_file):
        return jsonify([])
    with open(config_file, "r", encoding="utf-8") as f:
        configs = json.load(f)
    return jsonify(configs)


@app.route("/api/api-configs", methods=["POST"])
def add_api_config():
    """添加API配置"""
    config_file = os.path.join(API_CONFIG_DIR, "api_configs.json")
    data = request.json
    
    if not os.path.exists(config_file):
        configs = []
    else:
        with open(config_file, "r", encoding="utf-8") as f:
            configs = json.load(f)
    
    new_config = {
        "id": str(uuid.uuid4())[:8],
        "name": data.get("name", "新API"),
        "url": data.get("url", "http://localhost:8000"),
        "description": data.get("description", "")
    }
    configs.append(new_config)
    
    with open(config_file, "w", encoding="utf-8") as f:
        json.dump(configs, f, ensure_ascii=False, indent=2)
    
    return jsonify(new_config)


@app.route("/api/api-configs/<config_id>", methods=["PUT"])
def update_api_config(config_id):
    """更新API配置"""
    config_file = os.path.join(API_CONFIG_DIR, "api_configs.json")
    data = request.json
    
    if not os.path.exists(config_file):
        return jsonify({"error": "配置文件不存在"}), 404
    
    with open(config_file, "r", encoding="utf-8") as f:
        configs = json.load(f)
    
    for config in configs:
        if config["id"] == config_id:
            config["name"] = data.get("name", config["name"])
            config["url"] = data.get("url", config["url"])
            config["description"] = data.get("description", config.get("description", ""))
            break
    else:
        return jsonify({"error": "配置不存在"}), 404
    
    with open(config_file, "w", encoding="utf-8") as f:
        json.dump(configs, f, ensure_ascii=False, indent=2)
    
    return jsonify({"success": True})


@app.route("/api/api-configs/<config_id>", methods=["DELETE"])
def delete_api_config(config_id):
    """删除API配置"""
    config_file = os.path.join(API_CONFIG_DIR, "api_configs.json")
    
    if not os.path.exists(config_file):
        return jsonify({"error": "配置文件不存在"}), 404
    
    with open(config_file, "r", encoding="utf-8") as f:
        configs = json.load(f)
    
    configs = [c for c in configs if c["id"] != config_id]
    
    with open(config_file, "w", encoding="utf-8") as f:
        json.dump(configs, f, ensure_ascii=False, indent=2)
    
    return jsonify({"success": True})


@app.route("/api/run/stream", methods=["POST"])
def run_tests_stream():
    """运行测试（实时进度）"""
    global USE_LLM_EVAL, LLM_API_URL
    
    data = request.json
    api_url = data.get("api_url", "http://localhost:8000")
    use_llm = data.get("use_llm", USE_LLM_EVAL)
    llm_api_url = data.get("llm_api_url", LLM_API_URL)
    case_ids = data.get("case_ids", [])
    
    client.base_url = api_url
    client.chat_endpoint = f"{api_url}/chat"
    
    if use_llm:
        test_evaluator = Evaluator(api_url=llm_api_url, use_llm=True)
    else:
        test_evaluator = Evaluator()
    
    test_cases = load_test_cases()
    if case_ids:
        test_cases = [c for c in test_cases if str(c["id"]) in [str(cid) for cid in case_ids]]
    
    results = []
    start_time = time.time()
    
    def generate():
        start_data = json.dumps({"total": len(test_cases)})
        yield f"event: start\ndata: {start_data}\n\n"
        
        for i, case in enumerate(test_cases):
            question = case["question"]
            expect_keywords = case["expect_keywords"]
            expect_intent = case.get("expect_intent", "")
            
            try:
                answer = client.send_message(question)
                
                if use_llm:
                    llm_result = test_evaluator.evaluate_with_llm(
                        question, answer, expect_intent, expect_keywords
                    )
                    result = {
                        "id": case["id"],
                        "category": case["category"],
                        "question": question,
                        "answer": answer,
                        "expect_keywords": expect_keywords,
                        "expect_intent": expect_intent,
                        "is_pass": llm_result.get("final_pass", False),
                        "matched_keyword": None,
                        "llm_evaluation": llm_result
                    }
                else:
                    is_pass, matched_keyword = test_evaluator.evaluate(answer, expect_keywords)
                    result = {
                        "id": case["id"],
                        "category": case["category"],
                        "question": question,
                        "answer": answer,
                        "expect_keywords": expect_keywords,
                        "is_pass": is_pass,
                        "matched_keyword": matched_keyword
                    }
            except Exception as e:
                result = {
                    "id": case["id"],
                    "category": case["category"],
                    "question": question,
                    "answer": "",
                    "expect_keywords": expect_keywords,
                    "is_pass": False,
                    "matched_keyword": None,
                    "error": str(e)
                }
            
            results.append(result)
            
            progress = {
                "current": i + 1,
                "total": len(test_cases),
                "result": result
            }
            yield f"event: progress\ndata: {json.dumps(progress)}\n\n"
        
        duration = time.time() - start_time
        history_mgr.save_run(results, use_llm, api_url, duration)
        
        summary = {
            "total": len(results),
            "passed": sum(1 for r in results if r.get("is_pass", False)),
            "failed": sum(1 for r in results if not r.get("is_pass", True)),
            "duration": round(duration, 2)
        }
        yield f"event: complete\ndata: {json.dumps(summary)}\n\n"
    
    return Response(stream_with_context(generate()), mimetype='text/event-stream')


@app.route("/api/run", methods=["POST"])
def run_tests():
    """运行测试（非流式）"""
    global USE_LLM_EVAL, LLM_API_URL
    
    data = request.json
    api_url = data.get("api_url", "http://localhost:8000")
    use_llm = data.get("use_llm", USE_LLM_EVAL)
    llm_api_url = data.get("llm_api_url", LLM_API_URL)
    
    client.base_url = api_url
    client.chat_endpoint = f"{api_url}/chat"
    
    if use_llm:
        test_evaluator = Evaluator(api_url=llm_api_url, use_llm=True)
    else:
        test_evaluator = Evaluator()
    
    test_cases = load_test_cases()
    results = []
    start_time = time.time()
    
    for case in test_cases:
        question = case["question"]
        expect_keywords = case["expect_keywords"]
        expect_intent = case.get("expect_intent", "")
        
        try:
            answer = client.send_message(question)
            
            if use_llm:
                llm_result = test_evaluator.evaluate_with_llm(
                    question, answer, expect_intent, expect_keywords
                )
                results.append({
                    "id": case["id"],
                    "category": case["category"],
                    "question": question,
                    "answer": answer,
                    "expect_keywords": expect_keywords,
                    "expect_intent": expect_intent,
                    "is_pass": llm_result.get("final_pass", False),
                    "matched_keyword": None,
                    "llm_evaluation": llm_result
                })
            else:
                is_pass, matched_keyword = test_evaluator.evaluate(answer, expect_keywords)
                results.append({
                    "id": case["id"],
                    "category": case["category"],
                    "question": question,
                    "answer": answer,
                    "expect_keywords": expect_keywords,
                    "is_pass": is_pass,
                    "matched_keyword": matched_keyword
                })
        except Exception as e:
            results.append({
                "id": case["id"],
                "category": case["category"],
                "question": question,
                "answer": "",
                "expect_keywords": expect_keywords,
                "is_pass": False,
                "matched_keyword": None,
                "error": str(e)
            })
    
    duration = time.time() - start_time
    history_mgr.save_run(results, use_llm, api_url, duration)
    
    return jsonify({
        "success": True,
        "results": results,
        "use_llm": use_llm,
        "duration": round(duration, 2)
    })


@app.route("/api/history")
def get_history():
    """获取测试历史（支持分页）"""
    page = request.args.get("page", 1, type=int)
    size = request.args.get("size", 10, type=int)
    
    offset = (page - 1) * size
    runs = history_mgr.get_runs(limit=size, offset=offset)
    total = history_mgr.get_runs_count()
    
    return jsonify({
        "history": runs,
        "total": total,
        "page": page,
        "size": size
    })


@app.route("/api/history/<int:run_id>")
def get_history_detail(run_id):
    """获取历史测试详情"""
    results = history_mgr.get_run_results(run_id)
    return jsonify(results)


@app.route("/api/history/<int:run_id>", methods=["DELETE"])
def delete_history(run_id):
    """删除历史记录"""
    history_mgr.delete_run(run_id)
    return jsonify({"success": True})


@app.route("/api/history/stats")
def get_history_stats():
    """获取统计信息"""
    stats = history_mgr.get_statistics()
    return jsonify(stats)


@app.route("/api/results")
def get_results():
    """获取上次测试结果"""
    return jsonify([])


if __name__ == "__main__":
    print("=" * 50)
    print("Web 服务器启动: http://localhost:8080")
    print("=" * 50)
    app.run(host="0.0.0.0", port=8080, debug=False)