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

from sqlalchemy import text

from app.core.database import AsyncSessionLocal

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, source_message_id: str) -> dict[str, Any]:
    payload = {
        "group_chat_id": CHAT_ID,
        "message_content": message,
        "sender": SENDER,
        "feishu_message_id": source_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_body = resp.read().decode("utf-8")
    try:
        return json.loads(text_body)
    except Exception:
        return {"raw": text_body}


def detect_prompt_scene(prompt: str) -> 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 "unknown"


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 get_max_ai_call_id() -> int:
    async with AsyncSessionLocal() as db:
        row = (
            await db.execute(text("SELECT COALESCE(MAX(id), 0) AS v FROM ai_call_logs"))
        ).mappings().first()
    return int((row or {}).get("v") or 0)


async def wait_run(source_message_id: str, timeout_sec: int = 600) -> dict[str, Any] | None:
    deadline = time.time() + timeout_sec
    last = None
    while time.time() < deadline:
        async with AsyncSessionLocal() as db:
            row = (
                await db.execute(
                    text(
                        """
                        SELECT id, status, trace_id, source_message_id, created_at, updated_at
                        FROM ai_orchestration_runs
                        WHERE source_message_id = :sid
                        ORDER BY id DESC
                        LIMIT 1
                        """
                    ),
                    {"sid": source_message_id},
                )
            ).mappings().first()
        if row:
            last = dict(row)
            status = str(last.get("status") or "").lower()
            if status in {"succeeded", "failed", "cancelled", "waiting_manual"}:
                return last
        await asyncio.sleep(2)
    return last


async def wait_outbox(source_message_id: str, expected_sent: int, timeout_sec: int = 300) -> list[dict[str, Any]]:
    deadline = time.time() + timeout_sec
    latest: list[dict[str, Any]] = []
    while time.time() < deadline:
        async with AsyncSessionLocal() as db:
            rows = (
                await db.execute(
                    text(
                        """
                        SELECT id, status, attempts, message_text, created_at
                        FROM feishu_reply_outbox
                        WHERE source_message_id = :sid
                        ORDER BY id ASC
                        """
                    ),
                    {"sid": source_message_id},
                )
            ).mappings().all()
        latest = [dict(r) for r in rows]
        sent_count = sum(1 for r in latest if str(r.get("status") or "") == "sent")
        if sent_count >= expected_sent:
            return latest
        await asyncio.sleep(2)
    return latest


async def fetch_processing(source_message_id: str) -> dict[str, Any]:
    async with AsyncSessionLocal() as db:
        row = (
            await db.execute(
                text(
                    """
                    SELECT id, route, intent, processing_status, duration_ms, error_message, reply_source, created_at
                    FROM ai_processing_logs
                    WHERE feishu_message_id = :sid
                    ORDER BY id DESC
                    LIMIT 1
                    """
                ),
                {"sid": source_message_id},
            )
        ).mappings().first()
    return dict(row or {})


async def fetch_prompt_scene(min_id: int, keys: list[str]) -> tuple[str, list[dict[str, Any]]]:
    async with AsyncSessionLocal() as db:
        rows = (
            await db.execute(
                text(
                    """
                    SELECT id, function_name, prompt, created_at
                    FROM ai_call_logs
                    WHERE id > :min_id
                      AND function_name IN ('adaptive_rag_answer_normal', 'adaptive_rag_answer_deep')
                    ORDER BY id ASC
                    LIMIT 400
                    """
                ),
                {"min_id": min_id},
            )
        ).mappings().all()

    lowered = [str(k).lower() for k in (keys or []) if str(k).strip()]
    picked = "unknown"
    candidates: list[dict[str, Any]] = []
    for r in rows:
        prompt = str(r.get("prompt") or "")
        pl = prompt.lower()
        if lowered and not any(k in pl for k in lowered):
            continue
        scene = detect_prompt_scene(prompt)
        candidates.append(
            {
                "id": int(r.get("id") or 0),
                "scene": scene,
                "function_name": str(r.get("function_name") or ""),
                "created_at": str(r.get("created_at") or ""),
            }
        )
        if scene != "unknown":
            picked = scene

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


def evaluate(case_idx: int, scene: str, reply: str, run_status: str) -> tuple[bool, 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]) -> dict[str, Any]:
    idx = int(case["idx"])
    source_message_id = f"real-scene-p1-v4-{idx}-{int(time.time())}"
    message = str(case["message"])

    before_call_id = await get_max_ai_call_id()
    capture = post_capture(token, message, source_message_id)

    run_row = await wait_run(source_message_id)
    run_status = str((run_row or {}).get("status") or "missing")
    expected_sent = 2 if run_status == "succeeded" else 1
    outbox_rows = await wait_outbox(source_message_id, expected_sent)

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

    processing = await fetch_processing(source_message_id)
    prompt_scene, candidates = await fetch_prompt_scene(before_call_id, list(case.get("keys") or []))
    inferred_scene = infer_scene_from_reply(final_reply)
    actual_scene = prompt_scene if prompt_scene != "unknown" else inferred_scene
    passed, pass_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"]),
        "capture_status": str(capture.get("status") or ""),
        "run_status": run_status,
        "prompt_scene": prompt_scene,
        "inferred_scene": inferred_scene,
        "actual_scene": actual_scene,
        "ack": ack,
        "final_reply": final_reply,
        "thinking_leak": has_thinking_leak(final_reply),
        "has_capacity_warning": "⚠ 产能提示" in final_reply,
        "processing": {
            "route": str(processing.get("route") or ""),
            "intent": str(processing.get("intent") or ""),
            "status": str(processing.get("processing_status") or ""),
            "duration_ms": int(processing.get("duration_ms") or 0),
            "error_message": str(processing.get("error_message") or ""),
            "reply_source": str(processing.get("reply_source") or ""),
        },
        "outbox": outbox_rows,
        "call_log_candidates": candidates,
        "pass": passed,
        "pass_rule": pass_rule,
    }


async def main() -> None:
    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']} leak={one['thinking_leak']}",
            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===")


if __name__ == "__main__":
    asyncio.run(main())