PMFlow/backend/scripts/p1_real_validate.py

import asyncio
import json
import time
import urllib.request
import urllib.parse
from datetime import datetime
from typing import Any, Optional

from sqlalchemy import select, func

from app.core.database import AsyncSessionLocal
from app.models import AIOrchestrationRun, FeishuReplyOutbox, AIProcessingLog, AICallLog

BASE = "http://127.0.0.1:8000"
CHAT_ID = "oc_c487d87563e48c2f9566ccd5a758212c"
SENDER = "P1验收机器人"

CASES = [
    {"idx": 1, "message": "杨宇现在负载多少", "expected_scene": "member_workload", "keys": ["杨宇", "负载"]},
    {"idx": 2, "message": "最近AI行业有什么动态", "expected_scene": "external_trend", "keys": ["AI行业", "行业", "动态", "趋势"]},
    {"idx": 3, "message": "帮我创建个项目叫路由验收测试，顺便问下现在有几个进行中的项目", "expected_scene": "mixed", "keys": ["顺便", "问题部分", "需求部分", "进行中的项目"]},
    {"idx": 4, "message": "AI劝退助手项目进展怎么样", "expected_scene": "default", "keys": ["AI劝退助手", "进展怎么样", "项目进展"]},
    {"idx": 5, "message": "帮我看看宠物行业最新动态", "expected_scene": "external_trend", "keys": ["宠物行业", "最新动态", "行业", "动态"]},
]

def login_token() -> str:
    data = urllib.parse.urlencode({"username": "admin", "password": "admin123"}).encode()
    req = urllib.request.Request(
        f"{BASE}/auth/login",
        data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        payload = json.loads(resp.read().decode("utf-8"))
    return str(payload.get("access_token") or "")


def post_capture(token: str, *, message: str, message_id: str) -> dict[str, Any]:
    payload = {
        "group_chat_id": CHAT_ID,
        "message_content": message,
        "sender": SENDER,
        "feishu_message_id": message_id,
    }
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        f"{BASE}/ai-steward/ai-steward/signal/capture-feishu",
        data=body,
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=240) as resp:
        text = resp.read().decode("utf-8")
    try:
        return json.loads(text)
    except Exception:
        return {"raw": text}


def detect_prompt_scene(prompt: str) -> Optional[str]:
    p = str(prompt or "")
    if "你是 PMFlow 的团队协作助手。" in p:
        return "member_workload"
    if "你是 PMFlow 的市场情报助手。" in p:
        return "external_trend"
    if "用户消息同时包含问题和需求。" in p:
        return "mixed"
    if "你是 PMFlow 的 AI 项目管理官。" in p:
        return "ai_pmo"
    if "用户通过飞书直接提问。" in p:
        return "feishu_direct"
    if "你是 PMFlow 的 RAG 问答助手" in p:
        return "default"
    return None


def infer_scene_from_reply(reply: str) -> str:
    t = str(reply or "")
    if "[回答]" in t and "[已收到需求]" in t:
        return "mixed"
    if "[行业动态]" in t and "[建议]" in t:
        return "external_trend"
    if ("负载" in t or "工作量" in t) and "%" in t and "项目列表" not in t:
        return "member_workload"
    if "完成率" in t:
        return "default"
    return "unknown"


def has_thinking_leak(reply: str) -> bool:
    for line in str(reply or "").splitlines():
        s = line.strip()
        if not s:
            continue
        if len(s) <= 12 and s.startswith("我") and "先" in s:
            return True
    return False


async def fetch_max_ai_call_id(db) -> int:
    m = (await db.execute(select(func.max(AICallLog.id)))).scalar()
    return int(m or 0)


async def wait_run(db, source_message_id: str, timeout: int = 540) -> Optional[AIOrchestrationRun]:
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        row = (
            await db.execute(
                select(AIOrchestrationRun)
                .where(AIOrchestrationRun.source_message_id == source_message_id)
                .order_by(AIOrchestrationRun.id.desc())
                .limit(1)
            )
        ).scalar_one_or_none()
        if row is not None:
            last = row
            s = str(row.status or "").lower()
            if s in {"succeeded", "failed", "cancelled"}:
                return row
        await asyncio.sleep(2)
    return last


async def fetch_outbox(db, source_message_id: str):
    return (
        await db.execute(
            select(FeishuReplyOutbox)
            .where(FeishuReplyOutbox.source_message_id == source_message_id)
            .order_by(FeishuReplyOutbox.id.asc())
        )
    ).scalars().all()


async def wait_outbox(db, source_message_id: str, run_status: str, timeout: int = 240):
    deadline = time.time() + timeout
    expected = 2 if run_status == "succeeded" else 1
    latest = []
    while time.time() < deadline:
        latest = await fetch_outbox(db, source_message_id)
        sent = [r for r in latest if str(r.status or "") == "sent"]
        if len(sent) >= expected:
            return latest
        await asyncio.sleep(2)
    return latest


async def fetch_processing(db, source_message_id: str):
    return (
        await db.execute(
            select(AIProcessingLog)
            .where(AIProcessingLog.feishu_message_id == source_message_id)
            .order_by(AIProcessingLog.id.desc())
            .limit(1)
        )
    ).scalar_one_or_none()


async def fetch_scene_from_call_logs(db, min_id: int, keys: list[str]):
    rows = (
        await db.execute(
            select(AICallLog)
            .where(
                AICallLog.id > min_id,
                AICallLog.function_name.in_(["adaptive_rag_answer_normal", "adaptive_rag_answer_deep"]),
            )
            .order_by(AICallLog.id.asc())
            .limit(160)
        )
    ).scalars().all()

    picked = "unknown"
    inspected = []
    lk = [k.lower() for k in keys if k]
    for r in rows:
        prompt = str(r.prompt or "")
        pl = prompt.lower()
        if lk and not any(k in pl for k in lk):
            continue
        sc = detect_prompt_scene(prompt)
        inspected.append(
            {
                "id": int(r.id),
                "scene": sc or "unknown",
                "function_name": str(r.function_name or ""),
                "created_at": str(r.created_at or ""),
            }
        )
        if sc:
            picked = sc

    if picked == "unknown" and inspected:
        picked = str(inspected[-1].get("scene") or "unknown")
    return picked, inspected


def evaluate(case_idx: int, scene: str, reply: str, run_status: str):
    leak = has_thinking_leak(reply)
    if case_idx == 1:
        ok = scene == "member_workload" and "%" in reply and "项目列表" not in reply and not leak
        return ok, "member_workload+百分比+无项目列表"
    if case_idx == 2:
        sec = "[行业动态]" in reply and ("[对项目影响]" in reply or "[对项目的影响]" in reply) and "[建议]" in reply
        src = "未接入实时数据" in reply or "训练数据" in reply or "数据来源" in reply
        ok = scene == "external_trend" and sec and src and not leak
        return ok, "external_trend+三段+来源限制"
    if case_idx == 3:
        mixed = "[回答]" in reply and "[已收到需求]" in reply
        prog = "进行中" in reply
        ok = scene == "mixed" and run_status == "succeeded" and mixed and prog and not leak
        return ok, "mixed+创建+进行中数量"
    if case_idx == 4:
        ok = scene == "default" and "完成率" in reply and "%" in reply and not leak
        return ok, "default+完成率"
    if case_idx == 5:
        ok = scene == "external_trend" and not leak
        return ok, "external_trend+无思考外漏"
    return False, "unknown"


async def run_case(token: str, case: dict[str, Any]):
    idx = int(case["idx"])
    message = str(case["message"])
    source_message_id = f"real-scene-p1-v2-{idx}-{int(time.time())}"

    async with AsyncSessionLocal() as db:
        before_call_id = await fetch_max_ai_call_id(db)

    capture = post_capture(token, message=message, message_id=source_message_id)

    async with AsyncSessionLocal() as db:
        run_row = await wait_run(db, source_message_id)
        run_status = str(getattr(run_row, "status", "missing") or "missing") if run_row is not None else "missing"
        outbox_rows = await wait_outbox(db, source_message_id, run_status)
        processing = await fetch_processing(db, source_message_id)
        prompt_scene, candidates = await fetch_scene_from_call_logs(db, before_call_id, list(case.get("keys") or []))

    sent = [r for r in outbox_rows if str(r.status or "") == "sent"]
    ack = str(sent[0].message_text or "").strip() if sent else ""
    final_reply = str(sent[-1].message_text or "").strip() if sent else ""
    if not final_reply:
        final_reply = str(capture.get("direct_reply_text") or "").strip()

    inferred_scene = infer_scene_from_reply(final_reply)
    actual_scene = prompt_scene if prompt_scene != "unknown" else inferred_scene

    passed, rule = evaluate(idx, actual_scene, final_reply, run_status)

    return {
        "idx": idx,
        "source_message_id": source_message_id,
        "message": message,
        "expected_scene": str(case["expected_scene"]),
        "prompt_scene": prompt_scene,
        "inferred_scene": inferred_scene,
        "actual_scene": actual_scene,
        "capture_status": str(capture.get("status") or ""),
        "run_status": run_status,
        "ack": ack,
        "final_reply": final_reply,
        "has_capacity_warning": "⚠ 产能提示" in final_reply,
        "thinking_leak": has_thinking_leak(final_reply),
        "processing": {
            "route": str(getattr(processing, "route", "") or ""),
            "intent": str(getattr(processing, "intent", "") or ""),
            "status": str(getattr(processing, "processing_status", "") or ""),
            "duration_ms": int(getattr(processing, "duration_ms", 0) or 0),
            "error_message": str(getattr(processing, "error_message", "") or ""),
            "reply_source": str(getattr(processing, "reply_source", "") or ""),
        },
        "outbox": [
            {
                "id": int(r.id),
                "status": str(r.status or ""),
                "attempts": int(r.attempts or 0),
                "created_at": str(r.created_at or ""),
                "text": str(r.message_text or ""),
            }
            for r in outbox_rows
        ],
        "call_log_candidates": candidates,
        "pass": passed,
        "pass_rule": rule,
    }


async def main():
    token = login_token()
    if not token:
        raise RuntimeError("token empty")

    results = []
    for case in CASES:
        print(f"[RUN] case {case['idx']} -> {case['message']}", flush=True)
        one = await run_case(token, case)
        print(f"[DONE] case {one['idx']} scene={one['actual_scene']} pass={one['pass']} run={one['run_status']}", flush=True)
        results.append(one)
        await asyncio.sleep(2)

    payload = {
        "generated_at": datetime.now().isoformat(),
        "chat_id": CHAT_ID,
        "results": results,
    }
    print("===RESULT_JSON_START===")
    print(json.dumps(payload, ensure_ascii=False))
    print("===RESULT_JSON_END===")


asyncio.run(main())