#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""QA 机器人本地评估工具 — 在本机运行，调用内网机器人 API，结果自动上传 VPS

用法:
  # 首次使用：下载测试集
  python3 qa_eval.py --init

  # Titan 模式（推荐）：交互式输入凭证
  python3 qa_eval.py

  # 旧版兼容模式：命令行传参
  python3 qa_eval.py --url http://内网地址/api/chat --token xxx

  # 只跑正样本（快速验证）
  python3 qa_eval.py --positive-only

  # 自定义测试集路径
  python3 qa_eval.py --testset /path/to/testset.json

  # 不自动上传（只本地出报告）
  python3 qa_eval.py --no-upload

环境要求:
  - Python 3.8+（MacBook Pro 自带）
  - ssh / scp（MacBook Pro 自带）
  - 无需安装任何 pip 包（纯标准库）

安全:
  - 运行时交互输入凭证（测试地址/tenantId/robotId/token）
  - 凭证绝不记录到日志/结果/报告中（SensitiveDataFilter 自动过滤）
  - token 输入时不回显（getpass）

输出:
  - results-<timestamp>.json   评估原始结果（不含凭证）
  - report-<timestamp>.html    评估报告（可在浏览器打开）
  - eval-<timestamp>.log       执行日志（不含凭证，排查问题用）
"""
import argparse
import getpass
import json
import logging
import os
import subprocess
import sys
import time
import urllib.request
import urllib.error
from datetime import datetime

# ── 常量 ──
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DEFAULT_TESTSET_URL = "https://wechat-qa-code.pages.dev/testset.json"
DEFAULT_TESTSET_PATH = os.path.join(SCRIPT_DIR, "testset.json")
VPS_HOST = "ubuntu@localhost"  # 占位，实际用 --vps-host 覆盖
VPS_REMOTE_DIR = "~/wechat-qa-eval-runs"
FIXED_FALLBACK = "抱歉，暂时没有找到您问题的答案"
NO_MATCH_MARKERS = ["__NO_MATCH__", FIXED_FALLBACK, "非常抱歉"]

# Titan 平台常量
TITAN_PATH_OPEN = "/titan/flow_proc_v3/session/open"
TITAN_PATH_INTERACT = "/titan/flow_proc_v3/session/interact"
TITAN_PATH_CLOSE = "/titan/flow_proc_v3/session/close"
TITAN_REJECTION_TYPES = {"stRejection", "stSlotRejection", "stTimeout", "stSlotTimeout"}
REDACTED = "***REDACTED***"


# ── 敏感信息日志过滤器 ──
class SensitiveDataFilter(logging.Filter):
    """日志过滤器：自动屏蔽敏感信息（URL/tenantId/robotId/token）

    防御纵深：即使代码意外将敏感值写入日志消息，
    过滤器也会在写入前替换为 ***REDACTED***。
    """

    def __init__(self):
        super().__init__()
        self._sensitive_values = []

    def add_sensitive(self, value):
        """注册一个需要屏蔽的敏感值（长度 > 2 才注册，避免误伤短字符串）"""
        if value and len(str(value)) > 2:
            self._sensitive_values.append(str(value))

    def filter(self, record):
        msg = str(record.getMessage())
        changed = False
        for v in self._sensitive_values:
            if v in msg:
                msg = msg.replace(v, REDACTED)
                changed = True
        if changed:
            record.msg = msg
            record.args = ()
        return True


# ── 日志 ──
_sensitive_filter = None  # 全局引用，供动态添加敏感值

def setup_logging(log_path, sensitive_values=None):
    """配置双输出日志：文件 + 控制台，自动过滤敏感信息"""
    global _sensitive_filter
    logger = logging.getLogger("qa_eval")
    logger.setLevel(logging.DEBUG)
    logger.handlers.clear()

    _sensitive_filter = SensitiveDataFilter()
    if sensitive_values:
        for v in sensitive_values:
            _sensitive_filter.add_sensitive(v)

    fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s",
                             datefmt="%Y-%m-%d %H:%M:%S")

    fh = logging.FileHandler(log_path, encoding="utf-8")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(fmt)
    fh.addFilter(_sensitive_filter)
    logger.addHandler(fh)

    ch = logging.StreamHandler(sys.stdout)
    ch.setLevel(logging.INFO)
    ch.setFormatter(fmt)
    ch.addFilter(_sensitive_filter)
    logger.addHandler(ch)

    return logger


def add_sensitive_to_filter(value):
    """动态添加敏感值到已配置的日志过滤器"""
    global _sensitive_filter
    if _sensitive_filter and value:
        _sensitive_filter.add_sensitive(value)


# ── Titan 平台适配器 ──
class TitanAdapter:
    """Titan 平台 API 适配器

    封装 open → interact → close 三步流程。
    敏感信息存储在实例属性中，__repr__ 不暴露任何凭证。
    每个问题独立一个 session，保证测试隔离性。
    """

    def __init__(self, base_url, tenant_id, robot_id, token, timeout=30):
        self._base_url = base_url.rstrip("/")
        self._tenant_id = tenant_id
        self._robot_id = robot_id
        self._token = token
        self._timeout = timeout
        self._user_id = tenant_id  # API 文档: userId 与 tenantId 保持一致
        self._order_counter = 0

    def __repr__(self):
        return "TitanAdapter(credentials=***)"

    def _payload_base(self):
        """构造基础请求体（含凭证）"""
        return {
            "tenantId": self._tenant_id,
            "basic": {
                "robotId": self._robot_id,
                "userId": self._user_id,
                "token": self._token,
            }
        }

    def _post(self, path, biz_data):
        """发送 POST 请求，返回解析后的 JSON（code != 0 抛异常）"""
        url = self._base_url + path
        payload = self._payload_base()
        payload["biz"] = biz_data
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            url, data=data, method="POST",
            headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(req, timeout=self._timeout) as resp:
            body = resp.read().decode("utf-8")
        result = json.loads(body)
        code = result.get("code", -999)
        if code != 0:
            raise RuntimeError(f"Titan API code={code} msg={result.get('msg', '')}")
        return result

    def open_session(self):
        """开启会话，返回 (sessionId, orderId)"""
        self._order_counter += 1
        order_id = f"qa_eval_{datetime.now().strftime('%Y%m%d%H%M%S')}_{self._order_counter}"
        result = self._post(TITAN_PATH_OPEN, {
            "orderId": order_id,
            "bizInfo": {},
        })
        data = result.get("data", {})
        session_id = data.get("sessionId", "")
        if not session_id:
            raise RuntimeError("会话开启失败: 未返回 sessionId")
        return session_id, order_id

    def interact(self, session_id, order_id, question):
        """发送问题文本，返回 (answer, output_type, action, elapsed_ms)"""
        t0 = time.time()
        result = self._post(TITAN_PATH_INTERACT, {
            "orderId": order_id,
            "sessionId": session_id,
            "uid": "qa_eval",
            "uname": "qa_eval",
            "input": {
                "category": "text",
                "type": "string",
                "value": question,
            }
        })
        elapsed = int((time.time() - t0) * 1000)
        current = result.get("data", {}).get("current", {})
        output = current.get("output", {})
        answer = output.get("value", "") or output.get("display", "")
        output_type = output.get("type", "")
        action = current.get("action", "")
        return answer, output_type, action, elapsed

    def close_session(self, session_id, order_id):
        """关闭会话（best-effort，异常静默忽略）"""
        try:
            self._post(TITAN_PATH_CLOSE, {
                "orderId": order_id,
                "sessionId": session_id,
            })
        except Exception:
            pass

    def ask(self, question, log=None):
        """高层接口：open → interact → close

        Returns:
            (answer_text, elapsed_ms, error_or_none)
        """
        # 1. 开启会话
        try:
            session_id, order_id = self.open_session()
        except Exception as e:
            err = str(e)
            if log:
                log.error(f"开启会话失败: {err}")
            return "", 0, f"session_open_failed"

        # 2. 交互
        try:
            answer, output_type, action, elapsed = self.interact(
                session_id, order_id, question)
        except Exception as e:
            err = str(e)
            if log:
                log.error(f"交互失败: {err}")
            self.close_session(session_id, order_id)
            return "", 0, f"interact_failed"

        # 3. 关闭会话（action=finish 时会话已自动结束）
        if action != "finish":
            self.close_session(session_id, order_id)

        return answer, elapsed, None


# ── 凭证输入 ──
def prompt_credentials():
    """运行时交互式输入敏感凭证（不写入任何文件）

    Returns:
        (base_url, tenant_id, robot_id, token)
    """
    print("\n" + "=" * 50)
    print("Titan 平台凭证输入")
    print("（以下信息不会记录到日志/结果/报告中）")
    print("=" * 50)
    base_url = input("机器人测试地址 (如 http://10.x.x.x): ").strip()
    tenant_id = input("tenantId (租户ID): ").strip()
    robot_id = input("robotId (机器人ID): ").strip()
    token = getpass.getpass("token (输入时不显示): ").strip()
    print("=" * 50)

    missing = []
    if not base_url:
        missing.append("测试地址")
    if not tenant_id:
        missing.append("tenantId")
    if not robot_id:
        missing.append("robotId")
    if not token:
        missing.append("token")

    if missing:
        print(f"以下字段未填写: {', '.join(missing)}")
        sys.exit(1)

    return base_url, tenant_id, robot_id, token


# ── 测试集加载 ──
def download_testset(url, dest_path, log):
    """从远程下载测试集 JSON"""
    log.info(f"下载测试集: {url}")
    try:
        req = urllib.request.Request(url)
        with urllib.request.urlopen(req, timeout=30) as resp:
            data = resp.read()
        with open(dest_path, "wb") as f:
            f.write(data)
        ts = json.loads(data.decode("utf-8"))
        pos = sum(len(g["variants"]) + 1 for g in ts["positives"])
        neg = len(ts["negatives"])
        log.info(f"下载成功: {pos} 正样本 + {neg} 负样本 = {pos + neg} 条")
        return ts
    except Exception as e:
        log.error(f"下载失败: {e}")
        raise


def load_testset(path, log):
    """加载本地测试集"""
    log.info(f"加载测试集: {path}")
    with open(path, encoding="utf-8") as f:
        ts = json.load(f)
    pos = sum(len(g["variants"]) + 1 for g in ts["positives"])
    neg = len(ts["negatives"])
    log.info(f"测试集: {pos} 正样本 + {neg} 负样本 = {pos + neg} 条")
    log.info(f"  标准版本: {ts['meta'].get('standard', 'N/A')}")
    log.info(f"  QA 版本: {ts['meta'].get('qa_version', 'N/A')}")
    return ts


# ── 机器人调用 ──
def call_bot(url, question, token=None, timeout=30, log=None):
    """调用机器人 API，返回 (response_text, elapsed_ms, error)

    payload 格式可在此适配平台实际接口。
    当前格式: {"question": "...", "free_text": "..."} → 解析 JSON 取 answer/output/response/result
    """
    payload = json.dumps({"question": question, "free_text": question}).encode("utf-8")
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    req = urllib.request.Request(url, data=payload, method="POST", headers=headers)
    t0 = time.time()
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = resp.read().decode("utf-8")
            elapsed = int((time.time() - t0) * 1000)
            # 尝试解析 JSON
            try:
                data = json.loads(body)
                if isinstance(data, dict):
                    for k in ("answer", "output", "response", "result", "data", "reply"):
                        if k in data:
                            return str(data[k]), elapsed, None
                    # JSON 但无已知字段 → 返回原文
                    return body, elapsed, None
            except json.JSONDecodeError:
                pass
            return body, elapsed, None
    except urllib.error.HTTPError as e:
        elapsed = int((time.time() - t0) * 1000)
        err = f"HTTP {e.code}: {e.reason}"
        if log:
            log.error(f"API 调用失败 [{question[:30]}…]: {err}")
        return "", elapsed, err
    except Exception as e:
        elapsed = int((time.time() - t0) * 1000)
        err = str(e)
        if log:
            log.error(f"API 调用异常 [{question[:30]}…]: {err}")
        return "", elapsed, err


# ── 判定逻辑 ──
def is_fallback(text):
    """判断是否固定话术/未命中"""
    t = (text or "").strip()
    if not t:
        return True
    for m in NO_MATCH_MARKERS:
        if m in t:
            return True
    return False


def eval_positive(question, expected_answer, variant_type, variant_difficulty,
                  url, token, log, sleep_s):
    """评估正样本：返回结果字典"""
    text, elapsed, error = call_bot(url, question, token, log=log)
    time.sleep(sleep_s)

    if error:
        return {"q": question, "type": variant_type, "difficulty": variant_difficulty,
                "hit": False, "error": error, "response": "", "elapsed_ms": elapsed}

    hit = (not is_fallback(text)) and (expected_answer.strip() in (text or "").strip())
    status = "✅ 命中" if hit else "❌ 未命中"
    log.debug(f"  {status} [{variant_type}/{variant_difficulty}] {question[:40]}… → {text[:60]}…")
    return {"q": question, "type": variant_type, "difficulty": variant_difficulty,
            "hit": hit, "response": (text or "")[:300], "elapsed_ms": elapsed}


def eval_negative(question, neg_type, intent, url, token, log, sleep_s):
    """评估负样本：返回结果字典"""
    text, elapsed, error = call_bot(url, question, token, log=log)
    time.sleep(sleep_s)

    if error:
        return {"q": question, "type": neg_type, "intent": intent,
                "false_hit": False, "error": error, "response": "", "elapsed_ms": elapsed}

    false_hit = not is_fallback(text)
    status = "❌ 误命中" if false_hit else "✅ 正确拒绝"
    log.debug(f"  {status} [{neg_type}] {question[:40]}… → {text[:60]}…")
    return {"q": question, "type": neg_type, "intent": intent,
            "false_hit": false_hit, "response": (text or "")[:300], "elapsed_ms": elapsed}



def is_fallback_titan(text, output_type=None):
    """Titan 平台拒识判定：output_type 为拒识类型，或文本命中兜底话术"""
    if output_type and output_type in TITAN_REJECTION_TYPES:
        return True
    return is_fallback(text)


# ── 评估执行 ──
def eval_positive(question, expected_answer, variant_type, variant_difficulty,
                  bot_fn, log, sleep_s):
    """评估正样本：返回结果字典

    bot_fn: callable(question) -> (text, elapsed_ms, error)
    """
    text, elapsed, error = bot_fn(question)
    time.sleep(sleep_s)

    if error:
        return {"q": question, "type": variant_type, "difficulty": variant_difficulty,
                "hit": False, "error": error, "response": "", "elapsed_ms": elapsed}

    hit = (not is_fallback(text)) and (expected_answer.strip() in (text or "").strip())
    status = "命中" if hit else "未命中"
    log.debug(f"  {status} [{variant_type}/{variant_difficulty}] {question[:40]}... -> {text[:60]}...")
    return {"q": question, "type": variant_type, "difficulty": variant_difficulty,
            "hit": hit, "response": (text or "")[:300], "elapsed_ms": elapsed}


def eval_negative(question, neg_type, intent, bot_fn, log, sleep_s):
    """评估负样本：返回结果字典

    bot_fn: callable(question) -> (text, elapsed_ms, error)
    """
    text, elapsed, error = bot_fn(question)
    time.sleep(sleep_s)

    if error:
        return {"q": question, "type": neg_type, "intent": intent,
                "false_hit": False, "error": error, "response": "", "elapsed_ms": elapsed}

    false_hit = not is_fallback(text)
    status = "误命中" if false_hit else "正确拒绝"
    log.debug(f"  {status} [{neg_type}] {question[:40]}... -> {text[:60]}...")
    return {"q": question, "type": neg_type, "intent": intent,
            "false_hit": false_hit, "response": (text or "")[:300], "elapsed_ms": elapsed}


# ── 评估执行 ──
def run_evaluation(ts, bot_fn, sleep_s, positive_only, log):
    """逐条跑测试集，返回完整结果

    bot_fn: callable(question) -> (text, elapsed_ms, error)
    """
    meta = ts["meta"]
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    results = {
        "meta": meta,
        "eval_time": timestamp,
        "api": REDACTED,  # 不记录实际 API 地址
        "sleep_s": sleep_s,
        "positive_items": [],
        "negative_items": [],
    }

    # 正样本
    total_pos = sum(len(g["variants"]) + 1 for g in ts["positives"])
    done = 0
    log.info(f"=== 正样本评估开始 ({total_pos} 条) ===")
    for grp in ts["positives"]:
        scene = grp.get("scene", "")
        std_q = grp["standard_q"]
        expected = grp["expected_answer"]
        group_result = {"id": grp["id"], "scene": scene, "standard_q": std_q, "items": []}

        # 标准 Q
        item = eval_positive(std_q, expected, "standard", "L1", bot_fn, log, sleep_s)
        item["group_id"] = grp["id"]
        group_result["items"].append(item)
        done += 1

        # 变体
        for v in grp["variants"]:
            item = eval_positive(v["q"], expected, v.get("type", "unknown"),
                                 v.get("difficulty", "L2"), bot_fn, log, sleep_s)
            item["group_id"] = grp["id"]
            group_result["items"].append(item)
            done += 1

        results["positive_items"].append(group_result)
        if done % 50 == 0 or done == total_pos:
            log.info(f"  正样本进度: {done}/{total_pos}")

    # 负样本
    if not positive_only:
        negs = ts["negatives"]
        log.info(f"=== 负样本评估开始 ({len(negs)} 条) ===")
        for i, n in enumerate(negs, 1):
            item = eval_negative(n["q"], n.get("type", "unknown"),
                                 n.get("intent", ""), bot_fn, log, sleep_s)
            results["negative_items"].append(item)
            if i % 10 == 0 or i == len(negs):
                log.info(f"  负样本进度: {i}/{len(negs)}")

    log.info("=== 评估完成 ===")
    return results


# ── 指标计算 ──
def compute_metrics(results, log):
    """计算全量指标 + 分维度指标（对齐行业标准）

    行业标准对齐：
    - IR (TREC/MS MARCO): Hit Rate, Precision, Recall, F1, Graded Relevance
    - NLU (Banking77/ATIS): Intent Accuracy, Out-of-Scope Detection (FAR)
    - RAGAS: Context Relevance (命中正确 QA = 检索相关)
    - 生产指标: P50/P95/P99 延迟
    """
    pos_items = [it for g in results["positive_items"] for it in g["items"]]
    neg_items = results["negative_items"]

    # ── IR 指标 ──
    tp = sum(1 for it in pos_items if it.get("hit"))         # True Positive: 正样本命中
    fn = sum(1 for it in pos_items if not it.get("hit") and not it.get("error"))  # False Negative: 正样本漏报
    fp = sum(1 for n in neg_items if n.get("false_hit"))     # False Positive: 负样本误命中
    # True Negative: 负样本正确拒绝
    tn = len(neg_items) - fp

    total_pos = len(pos_items)
    total_neg = len(neg_items)
    errors = sum(1 for it in pos_items if it.get("error"))

    # Hit Rate (TREC relevance ≥ 1 的比例)
    hit_rate = round(tp / total_pos * 100, 1) if total_pos else 0

    # Precision = TP / (TP + FP) — 命中的有多少是真正正确的
    precision = round(tp / (tp + fp) * 100, 1) if (tp + fp) else 100.0

    # Recall = TP / (TP + FN) — 应命中的有多少被命中了
    recall = round(tp / (tp + fn) * 100, 1) if (tp + fn) else 100.0

    # F1 = 2 * P * R / (P + R)
    f1 = round(2 * precision * recall / (precision + recall), 1) if (precision + recall) else 0

    # False Acceptance Rate (NLU out-of-scope detection) — 负样本误命中率
    far = round(fp / total_neg * 100, 1) if total_neg else 0

    # 按变体类型
    by_type = {}
    for it in pos_items:
        t = it.get("type", "unknown")
        by_type.setdefault(t, {"total": 0, "hit": 0})
        by_type[t]["total"] += 1
        if it.get("hit"):
            by_type[t]["hit"] += 1

    # 按难度
    by_difficulty = {}
    for it in pos_items:
        d = it.get("difficulty", "L2")
        by_difficulty.setdefault(d, {"total": 0, "hit": 0})
        by_difficulty[d]["total"] += 1
        if it.get("hit"):
            by_difficulty[d]["hit"] += 1

    # 按场景
    by_scene = {}
    for g in results["positive_items"]:
        scene = g.get("scene", "")
        items = g["items"]
        by_scene.setdefault(scene, {"total": 0, "hit": 0})
        for it in items:
            by_scene[scene]["total"] += 1
            if it.get("hit"):
                by_scene[scene]["hit"] += 1

    # 按 QA（每条 QA 的命中率，方便定位哪条 QA 出问题）
    by_qa = []
    for g in results["positive_items"]:
        items = g["items"]
        t = len(items)
        h = sum(1 for it in items if it.get("hit"))
        by_qa.append({
            "id": g["id"], "scene": g.get("scene", ""),
            "standard_q": g["standard_q"],
            "total": t, "hits": h,
            "hit_rate": round(h / t * 100, 1) if t else 0,
        })

    # 平均延迟 + 百分位（生产指标：P50/P95/P99）
    latencies = sorted([it.get("elapsed_ms", 0) for it in pos_items if it.get("elapsed_ms")])
    avg_latency = round(sum(latencies) / len(latencies)) if latencies else 0
    p50 = latencies[len(latencies) // 2] if latencies else 0
    p95 = latencies[int(len(latencies) * 0.95)] if latencies else 0
    p99 = latencies[min(int(len(latencies) * 0.99), len(latencies) - 1)] if latencies else 0

    # 漏报 & 误命中明细
    misses = [{"q": it["q"], "type": it.get("type", ""),
               "difficulty": it.get("difficulty", ""),
               "scene": next((g.get("scene", "") for g in results["positive_items"]
                             if g["id"] == it.get("group_id")), ""),
               "response": it.get("response", "")}
              for it in pos_items if not it.get("hit") and not it.get("error")]
    false_detail = [{"q": n["q"], "type": n.get("type", ""), "response": n.get("response", "")}
                    for n in neg_items if n.get("false_hit")]

    metrics = {
        "total": total_pos, "hits": tp,
        "hit_rate": hit_rate,
        "precision": precision,
        "recall": recall,
        "f1": f1,
        "false_acceptance_rate": far,
        "errors": errors, "false_hits": fp,
        "false_total": total_neg,
        "false_rate": far,
        "avg_latency_ms": avg_latency,
        "p50_latency_ms": p50,
        "p95_latency_ms": p95,
        "p99_latency_ms": p99,
        "by_type": by_type,
        "by_difficulty": by_difficulty,
        "by_scene": by_scene,
        "by_qa": by_qa,
        "misses": misses,
        "false_detail": false_detail,
    }

    log.info(f"指标: HitRate={hit_rate}% P={precision}% R={recall}% F1={f1} FAR={far}%")
    log.info(f"  延迟: avg={avg_latency}ms P50={p50}ms P95={p95}ms P99={p99}ms")
    return metrics


# ── HTML 报告 ──
def _hr_cls(rate):
    """命中率颜色类"""
    if rate >= 85: return "good"
    if rate >= 70: return "warn"
    return "bad"


def _judge(name, value, target, op=">="):
    """达标判定：返回 (name, value, target, passed, op)"""
    if op == ">=":
        ok = value >= target
    elif op == "<=":
        ok = value <= target
    elif op == "==":
        ok = value == target
    else:
        ok = False
    return (name, value, target, ok, op)


def render_report(results, metrics, log):
    """生成增强版 HTML 评估报告（含达标判定 + 低命中QA + 嵌入评测体系说明）

    报告分 3 层：
    1. 结论摘要（达标判定表 + 指标卡）
    2. 分维度拆解（场景/难度/类型 + 低命中QA定位 + 漏报/误命中明细）
    3. 嵌入评测体系说明（指标定义 + 行业标准 + 混淆矩阵 + 如何阅读）
    """
    meta = results["meta"]
    eval_time = results.get("eval_time", "")

    m = metrics

    # ── 达标判定 ──
    judgments = [
        _judge("Hit Rate（命中率）", m["hit_rate"], 85),
        _judge("Precision（精确率）", m.get("precision", 0), 90),
        _judge("Recall（召回率）", m.get("recall", 0), 85),
        _judge("F1 Score", m.get("f1", 0), 87),
        _judge("FAR（误命中率）", m.get("false_acceptance_rate", 0), 5, "<="),
        _judge("P95 延迟", m.get("p95_latency_ms", 0), 2000, "<="),
        _judge("调用错误数", m.get("errors", 0), 0, "=="),
    ]
    all_pass = all(j[3] for j in judgments)
    overall_icon = "✅" if all_pass else "⚠️"
    overall_text = "全部达标" if all_pass else "部分指标未达标"

    judgment_rows = ""
    for name, value, target, ok, op in judgments:
        icon = "✅" if ok else "❌"
        cmp_str = f"{op} {target}" if op != "==" else f"= {target}"
        cls = "good" if ok else "bad"
        label = "达标" if ok else "未达标"
        judgment_rows += f'<tr><td>{icon}</td><td>{name}</td><td>{value}</td><td>{cmp_str}</td><td class="{cls}"><b>{label}</b></td></tr>\n'

    # ── 指标卡 ──
    hit_class = _hr_cls(m["hit_rate"])

    # ── 分场景 ──
    scene_rows = ""
    for scene, v in sorted(m["by_scene"].items()):
        rate = round(v["hit"] / v["total"] * 100, 1) if v["total"] else 0
        scene_rows += f'<tr><td>{scene}</td><td>{v["hit"]}/{v["total"]}</td><td class="{_hr_cls(rate)}"><b>{rate}%</b></td></tr>\n'

    # ── 分难度 ──
    diff_names = {"L1": "L1 高相似（≤40%）", "L2": "L2 中等 (≀40%)", "L3": "L3 高难度 (≀20%)"}
    diff_rows = ""
    for d in ["L1", "L2", "L3"]:
        v = m["by_difficulty"].get(d, {"total": 0, "hit": 0})
        rate = round(v["hit"] / v["total"] * 100, 1) if v["total"] else 0
        diff_rows += f'<tr><td>{diff_names.get(d, d)}</td><td>{v["hit"]}/{v["total"]}</td><td class="{_hr_cls(rate)}"><b>{rate}%</b></td></tr>\n'

    # ── 分类型 ──
    type_names = {
        "standard": "标准 Q", "同义改写": "同义改写", "口语化": "口语化",
        "句式变换": "句式变换", "话题前置": "话题前置",
        "术语俗称互换": "术语俗称互换", "带语气词": "带语气词", "场景化求助": "场景化求助",
    }
    type_rows = ""
    for t, v in sorted(m["by_type"].items(), key=lambda x: -x[1]["total"]):
        rate = round(v["hit"] / v["total"] * 100, 1) if v["total"] else 0
        type_rows += f'<tr><td>{type_names.get(t, t)}</td><td>{v["hit"]}/{v["total"]}</td><td class="{_hr_cls(rate)}"><b>{rate}%</b></td></tr>\n'

    # ── 低命中 QA 定位 ──
    problem_qas = [q for q in m.get("by_qa", []) if q["hit_rate"] < 80]
    problem_qas.sort(key=lambda x: x["hit_rate"])
    qa_rows = ""
    for q in problem_qas[:15]:
        qa_rows += f'<tr><td>{q["id"]}</td><td>{q["scene"]}</td><td>{q["standard_q"][:30]}</td><td>{q["hits"]}/{q["total"]}</td><td class="{_hr_cls(q["hit_rate"])}"><b>{q["hit_rate"]}%</b></td></tr>\n'
    if not qa_rows:
        qa_rows = '<tr><td colspan="5" class="good">🎉 所有 QA 命中率 ≥ 80%</td></tr>'

    # ── 漏报明细 ──
    miss_rows = ""
    for item in m["misses"][:80]:
        miss_rows += (f'<tr><td>{item["q"]}</td><td>{type_names.get(item.get("type", ""), item.get("type", ""))}</td>'
                      f'<td>{item.get("difficulty", "")}</td><td>{item.get("scene", "")}</td>'
                      f'<td class="mono">{item["response"][:120]}</td></tr>\n')
    if not miss_rows:
        miss_rows = '<tr><td colspan="5" class="good">🎉 全部命中，无漏报</td></tr>'

    # ── 误命中明细 ──
    false_rows = ""
    for item in m["false_detail"][:30]:
        false_rows += (f'<tr><td>{item["q"]}</td><td>{item["type"]}</td>'
                       f'<td class="mono">{item["response"][:120]}</td></tr>\n')
    if not false_rows:
        false_rows = '<tr><td colspan="3" class="good">🎉 无误命中</td></tr>'

    # ── 安全/模拟标记 ──
    is_mock = "模拟数据预览" in eval_time or "(模拟)" in eval_time
    mock_note = '<br>🎨 本报告为模拟数据预览，实际运行时将替换为真实评估结果' if is_mock else ''

    html = f"""<!DOCTYPE html>
<html lang="zh-CN"><head><meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>QA 机器人评估报告 · {eval_time}</title>
<style>
:root {{ --bg:#0d1117; --card:#161b22; --border:#30363d; --text:#e6edf3; --text2:#8b949e; --blue:#58A6FF; --green:#3FB950; --orange:#D29922; --red:#F85149; --purple:#BC8CFF; }}
* {{ margin:0; padding:0; box-sizing:border-box; }}
body {{ background:var(--bg); color:var(--text); font-family:-apple-system,"PingFang SC","Microsoft YaHei",sans-serif; line-height:1.6; padding:1rem; max-width:960px; margin:0 auto; font-size:15px; }}
h1 {{ font-size:1.3rem; margin-bottom:0.3rem; }}
h2 {{ font-size:1.05rem; color:var(--blue); margin:1.5rem 0 0.5rem; padding-bottom:0.3rem; border-bottom:1px solid var(--border); }}
h3 {{ font-size:0.9rem; color:var(--text); margin:1rem 0 0.4rem; }}
.sub {{ color:var(--text2); font-size:0.82rem; margin-bottom:1rem; }}
.badge {{ display:inline-block; padding:0.15rem 0.6rem; border-radius:20px; font-size:0.75rem; font-weight:600; }}
.badge.pass {{ background:rgba(63,185,80,0.15); color:var(--green); border:1px solid rgba(63,185,80,0.3); }}
.badge.warn {{ background:rgba(210,153,22,0.15); color:var(--orange); border:1px solid rgba(210,153,22,0.3); }}
.cards {{ display:flex; flex-wrap:wrap; gap:0.6rem; margin-bottom:1.2rem; }}
.card {{ background:var(--card); border:1px solid var(--border); border-radius:10px; padding:0.7rem 1rem; min-width:120px; flex:1; }}
.card .num {{ font-size:1.5rem; font-weight:700; color:var(--blue); }}
.card.good .num {{ color:var(--green); }}
.card.warn .num {{ color:var(--orange); }}
.card.bad .num {{ color:var(--red); }}
.card .label {{ font-size:0.72rem; color:var(--text2); margin-top:0.2rem; }}
table {{ width:100%; border-collapse:collapse; font-size:0.82rem; margin-bottom:0.5rem; }}
th,td {{ border:1px solid var(--border); padding:0.4rem 0.5rem; text-align:left; }}
th {{ background:rgba(88,166,255,0.08); color:var(--blue); white-space:nowrap; }}
td.good {{ color:var(--green); font-weight:600; }} td.warn {{ color:var(--orange); font-weight:600; }} td.bad {{ color:var(--red); font-weight:600; }}
.mono {{ font-family:ui-monospace,monospace; font-size:0.76rem; color:var(--text2); word-break:break-all; }}
.info-box {{ background:rgba(88,166,255,0.06); border-radius:8px; padding:0.7rem 1rem; margin:0.8rem 0; border-left:3px solid var(--blue); }}
.info-box.green {{ background:rgba(63,185,80,0.06); border-left-color:var(--green); }}
.info-box.orange {{ background:rgba(210,153,22,0.06); border-left-color:var(--orange); }}
.info-box p {{ margin:0; font-size:0.82rem; color:var(--text2); }}
.info-box p b {{ color:var(--text); }}
.eval-doc {{ background:rgba(188,140,255,0.04); border:1px solid rgba(188,140,255,0.15); border-radius:12px; padding:1rem 1.2rem; margin:1.5rem 0; }}
.eval-doc h2 {{ color:var(--purple); border-bottom-color:rgba(188,140,255,0.2); }}
.eval-doc h3 {{ color:var(--purple); }}
.toc {{ background:var(--card); border:1px solid var(--border); border-radius:8px; padding:0.8rem 1rem; margin-bottom:1.2rem; }}
.toc-title {{ font-size:0.85rem; color:var(--blue); margin-bottom:0.4rem; font-weight:600; }}
.toc a {{ display:block; font-size:0.8rem; color:var(--text2); text-decoration:none; padding:0.15rem 0; }}
.toc a:hover {{ color:var(--blue); }}
.footer {{ color:var(--text2); font-size:0.72rem; text-align:center; margin-top:2rem; padding-top:0.8rem; border-top:1px solid var(--border); }}
@media (max-width:600px) {{ .cards {{ gap:0.4rem; }} .card {{ min-width:100px; padding:0.5rem 0.7rem; }} .card .num {{ font-size:1.2rem; }} }}
</style></head><body>

<h1>QA 问答机器人评估报告</h1>
<div class="sub">评估时间: {eval_time} · 测试集: {meta.get("standard","N/A")} · QA 版本: {meta.get("qa_version","N/A")} · 共 {m["total"]} 正 + {m.get("false_total",0)} 负 · <span class="badge {"pass" if all_pass else "warn"}">{overall_icon} {overall_text}</span>{mock_note}</div>

<div class="toc">
  <div class="toc-title">📋 报告目录</div>
  <a href="#summary">① 指标总览与达标判定</a>
  <a href="#scene">② 分场景命中率</a>
  <a href="#difficulty">③ 分难度命中率</a>
  <a href="#type">④ 分变体类型命中率</a>
  <a href="#problem">⑤ 低命中 QA 定位</a>
  <a href="#misses">⑥ 正样本漏报明细</a>
  <a href="#false">⑦ 负样本误命中明细</a>
  <a href="#eval-system">📖 评测体系说明（嵌入）</a>
</div>

<h2 id="summary">① 指标总览与达标判定</h2>
<div class="cards">
  <div class="card {hit_class}"><div class="num">{m["hit_rate"]}%</div><div class="label">Hit Rate（命中率）</div></div>
  <div class="card {_hr_cls(m.get("precision",0))}"><div class="num">{m.get("precision",0)}%</div><div class="label">Precision（精确率）</div></div>
  <div class="card {_hr_cls(m.get("recall",0))}"><div class="num">{m.get("recall",0)}%</div><div class="label">Recall（召回率）</div></div>
  <div class="card {_hr_cls(m.get("f1",0))}"><div class="num">{m.get("f1",0)}</div><div class="label">F1 Score</div></div>
  <div class="card {"bad" if m.get("false_hits",0) else "good"}"><div class="num">{m.get("false_hits",0)}/{m.get("false_total",0)}</div><div class="label">误命中 (FAR={m.get("false_acceptance_rate",0)}%)</div></div>
  <div class="card {"bad" if m.get("errors",0) else "good"}"><div class="num">{m.get("errors",0)}</div><div class="label">调用错误</div></div>
</div>
<div class="cards">
  <div class="card"><div class="num">{m.get("avg_latency_ms",0)}ms</div><div class="label">平均延迟</div></div>
  <div class="card"><div class="num">{m.get("p50_latency_ms",0)}ms</div><div class="label">P50 延迟</div></div>
  <div class="card"><div class="num">{m.get("p95_latency_ms",0)}ms</div><div class="label">P95 延迟</div></div>
  <div class="card"><div class="num">{m.get("p99_latency_ms",0)}ms</div><div class="label">P99 延迟</div></div>
</div>

<h3>达标判定（7 项必过）</h3>
<table>
<tr><th></th><th>指标</th><th>实际值</th><th>达标线</th><th>结果</th></tr>
{judgment_rows}
</table>
<div class="info-box green">
  <p><b>说明</b>：达标线基于行业标准设定（详见下方<a href="#eval-system">评测体系说明</a>）。Hit Rate ≥ 85% 且 FAR ≤ 5% 为核心红线。</p>
</div>

<h2 id="scene">② 分场景命中率</h2>
<table><tr><th>场景</th><th>命中/总数</th><th>命中率</th></tr>{scene_rows}</table>

<h2 id="difficulty">③ 分难度命中率</h2>
<table><tr><th>难度</th><th>命中/总数</th><th>命中率</th></tr>{diff_rows}</table>
<div class="info-box">
  <p><b>难度定义</b>（基于 TREC QA 分级相关性）：L1 = 高相似度变体（期望必命中）、L2 = 中等改写（应命中）、L3 = 高难度表达（允许降级）。L3 低命中率不一定代表问题，但 L1 低命中率是严重信号。</p>
</div>

<h2 id="type">④ 分变体类型命中率</h2>
<table><tr><th>类型</th><th>命中/总数</th><th>命中率</th></tr>{type_rows}</table>

<h2 id="problem">⑤ 低命中 QA 定位（命中率 &lt; 80%）</h2>
<table><tr><th>QA ID</th><th>场景</th><th>标准问题</th><th>命中/总数</th><th>命中率</th></tr>{qa_rows}</table>
<div class="info-box orange">
  <p><b>行动建议</b>：以上 QA 的命中率低于 80%，建议检查：①知识库中该 QA 的相似问法是否覆盖了用户真实表达 ②阈值设置是否过高 ③答案文本是否与变体表达匹配。优先处理命中率最低的 QA。</p>
</div>

<h2 id="misses">⑥ 正样本漏报明细（{len(m["misses"])} 条）</h2>
<table><tr><th>输入</th><th>类型</th><th>难度</th><th>场景</th><th>机器人响应</th></tr>{miss_rows}</table>
<div class="info-box">
  <p><b>排查方法</b>：如果响应是兜底话术（"抱歉，暂时没有找到"），说明知识库未命中 → 需增加相似问法。如果响应是其他 QA 的答案，说明跨场景串答 → 需检查阈值。</p>
</div>

<h2 id="false">⑦ 负样本误命中明细（{len(m["false_detail"])} 条）</h2>
<table><tr><th>输入</th><th>类型</th><th>机器人响应</th></tr>{false_rows}</table>
<div class="info-box {"orange" if m.get("false_hits",0) else "green"}">
  <p>{'<b>⚠️ 注意</b>：存在误命中，需检查知识库是否过于宽泛或阈值过低。' if m.get('false_hits',0) else '<b>✅ 表现优秀</b>：所有负样本均正确返回兜底话术，无误命中。'}</p>
</div>

<div class="eval-doc" id="eval-system">
<h2>📖 评测体系说明</h2>
<p style="font-size:0.82rem;color:var(--text2);margin-bottom:1rem">本报告的所有指标均基于以下行业标准设计。以下说明帮助你正确解读报告数据。</p>

<h3>行业标准对齐</h3>
<table>
<tr><th>参考体系</th><th>本项目对应指标</th><th>说明</th></tr>
<tr><td><b>TREC QA</b>（信息检索）</td><td>Hit Rate + L1/L2/L3 分级</td><td>分级相关性：不是所有命中都同等重要</td></tr>
<tr><td><b>Banking77</b>（NLU 意图分类）</td><td>FAR（误命中率）</td><td>越界检测：负样本应被拒绝</td></tr>
<tr><td><b>RAGAS</b>（RAG 评测）</td><td>命中正确 QA = 上下文相关</td><td>检索到的 QA 必须是正确的那一条</td></tr>
<tr><td><b>ANLI</b>（对抗性评测）</td><td>42 条对抗负样本</td><td>关键词陷阱、近义混淆、跨场景串答检测</td></tr>
<tr><td><b>SRE 实践</b>（生产监控）</td><td>P50/P95/P99 延迟</td><td>尾部延迟影响用户体验</td></tr>
</table>

<h3>指标定义</h3>
<table>
<tr><th>指标</th><th>公式</th><th>含义</th><th>达标线</th></tr>
<tr><td><b>Hit Rate</b></td><td>TP / 总正样本</td><td>正样本中有多少命中了正确答案</td><td class="good">≥ 85%</td></tr>
<tr><td><b>Precision</b></td><td>TP / (TP + FP)</td><td>所有命中中有多少是真正正确的</td><td class="good">≥ 90%</td></tr>
<tr><td><b>Recall</b></td><td>TP / (TP + FN)</td><td>所有应该命中的有多少被命中了</td><td class="good">≥ 85%</td></tr>
<tr><td><b>F1</b></td><td>2×P×R / (P+R)</td><td>Precision 和 Recall 的调和平均</td><td class="good">≥ 87</td></tr>
<tr><td><b>FAR</b></td><td>FP / 总负样本</td><td>负样本中有多少被错误命中（越低越好）</td><td class="good">≤ 5%</td></tr>
<tr><td><b>P95 延迟</b></td><td>95 百分位响应时间</td><td>95% 的请求在此时长内完成</td><td class="good">≤ 2000ms</td></tr>
</table>

<h3>混淆矩阵</h3>
<table>
<tr><th></th><th>正样本（有答案）</th><th>负样本（无答案）</th></tr>
<tr><td><b>机器人命中</b></td><td class="good">TP（正确命中）</td><td class="bad">FP（误命中）</td></tr>
<tr><td><b>机器人拒识</b></td><td class="bad">FN（漏报）</td><td class="good">TN（正确拒绝）</td></tr>
</table>

<h3>如何阅读本报告</h3>
<ol style="font-size:0.82rem;color:var(--text2);padding-left:1.5rem;line-height:1.8">
  <li><b>先看达标判定</b>：7 项指标全绿 = 整体健康。有红色 = 需要排查。</li>
  <li><b>看分难度</b>：L1 低 = 严重问题。L3 低 = 可接受。</li>
  <li><b>看低命中 QA</b>：命中率 &lt; 80% 的 QA 是改进重点。</li>
  <li><b>看漏报明细</b>：兜底话术 = 加相似问法。别的 QA 答案 = 降阈值。</li>
  <li><b>看误命中</b>：FAR &gt; 0 说明负样本被错误回答了。</li>
  <li><b>看延迟</b>：P95 &gt; 2000ms 说明平台响应慢。</li>
</ol>

<h3>测试集结构</h3>
<table>
<tr><th>分类</th><th>子类</th><th>数量</th><th>说明</th></tr>
<tr><td rowspan="3"><b>正样本</b></td><td>L1 高相似</td><td>{m["by_difficulty"].get("L1",{}).get("total",0)} 条</td><td>标准问法 + 高相似变体</td></tr>
<tr><td>L2 中等</td><td>{m["by_difficulty"].get("L2",{}).get("total",0)} 条</td><td>口语化/句式变换</td></tr>
<tr><td>L3 高难度</td><td>{m["by_difficulty"].get("L3",{}).get("total",0)} 条</td><td>话题前置/场景化求助</td></tr>
<tr><td rowspan="4"><b>负样本</b></td><td>无关输入</td><td>与 QA 完全无关</td><td>应正确拒识</td></tr>
<tr><td>关键词陷阱</td><td>含关键词但意图不同</td><td>检测跨场景串答</td></tr>
<tr><td>近义混淆</td><td>近义词但不同业务</td><td>检测语义分辨力</td></tr>
<tr><td>边界输入</td><td>空串/纯标点/emoji</td><td>检测鲁棒性</td></tr>
</table>

<h3>评测工具</h3>
<div class="info-box">
  <p><b>qa_eval.py</b>（纯 Python 标准库，MacBook Pro 本地运行）：<br>
  ① 交互式输入凭证 ② 对每个问题走 Titan open→interact→close 三步流<br>
  ③ 自动计算全部指标 + P50/P95/P99 延迟 ④ 生成本报告 + JSON + LOG ⑤ scp 上传 VPS</p>
</div>

</div>

<div class="footer">由 qa_eval.py 自动生成 · {eval_time} · 测试集 {meta.get("standard","N/A")} · v2.2 标准（7 项评审必过）{mock_note}</div>

</body></html>"""
    return html


# ── VPS 上传 ──
def upload_to_vps(local_files, vps_host, remote_dir, log):
    """通过 scp 上传文件到 VPS"""
    log.info(f"上传到 VPS: {vps_host}:{remote_dir}")
    # 先创建远程目录
    try:
        subprocess.run(["ssh", vps_host, f"mkdir -p {remote_dir}"],
                       check=True, capture_output=True, timeout=15)
        log.debug("远程目录已创建")
    except Exception as e:
        log.error(f"SSH 连接失败（创建远程目录）: {e}")
        log.error("请检查: 1) SSH 密钥是否配置 2) VPS 地址是否正确 3) 网络是否可达")
        return False

    # scp 上传
    try:
        cmd = ["scp"] + local_files + [f"{vps_host}:{remote_dir}/"]
        log.debug(f"执行: {' '.join(cmd)}")
        result = subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=60)
        for f in local_files:
            log.info(f"  已上传: {os.path.basename(f)}")
        return True
    except subprocess.CalledProcessError as e:
        log.error(f"scp 失败: {e.stderr}")
        return False
    except Exception as e:
        log.error(f"scp 异常: {e}")
        return False


# ── 主流程 ──
def main():
    ap = argparse.ArgumentParser(
        description="QA 机器人本地评估工具（支持 Titan 平台）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="示例:\n"
               "  python3 qa_eval.py --init\n"
               "  python3 qa_eval.py                    # Titan 交互模式（推荐）\n"
               "  python3 qa_eval.py --url <URL> --token <TOKEN>  # 旧版兼容\n"
               "  python3 qa_eval.py --positive-only\n"
               "  python3 qa_eval.py --no-upload\n")
    ap.add_argument("--init", action="store_true", help="首次使用：下载测试集到本地")
    ap.add_argument("--url", help="机器人 API 地址（旧版兼容模式；不传则进入 Titan 交互模式）")
    ap.add_argument("--token", help="API token（旧版兼容模式）")
    ap.add_argument("--testset", default=DEFAULT_TESTSET_PATH, help=f"测试集路径（默认 {DEFAULT_TESTSET_PATH}）")
    ap.add_argument("--positive-only", action="store_true", help="只跑正样本（跳过负样本）")
    ap.add_argument("--sleep", type=float, default=0.5, help="请求间隔秒（默认 0.5）")
    ap.add_argument("--timeout", type=int, default=30, help="单次请求超时秒（默认 30）")
    ap.add_argument("--no-upload", action="store_true", help="不上传到 VPS（只本地出报告）")
    ap.add_argument("--vps-host", default=VPS_HOST, help=f"VPS SSH 地址（默认 {VPS_HOST}）")
    ap.add_argument("--vps-dir", default=VPS_REMOTE_DIR, help=f"VPS 远程目录（默认 {VPS_REMOTE_DIR}）")
    args = ap.parse_args()

    # 时间戳（用于文件命名）
    ts_str = datetime.now().strftime("%Y%m%d_%H%M%S")
    log_path = os.path.join(SCRIPT_DIR, f"eval-{ts_str}.log")

    # ── 模式一：初始化 ──
    if args.init:
        log = setup_logging(log_path)
        log.info("=" * 50)
        log.info(f"QA 机器人评估工具 · 初始化 · {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        log.info("=" * 50)
        download_testset(DEFAULT_TESTSET_URL, args.testset, log)
        log.info(f"测试集已保存到 {args.testset}")
        log.info("初始化完成。接下来运行:")
        log.info("  python3 qa_eval.py  (Titan 交互模式)")
        return

    # ── 判断运行模式 ──
    is_titan_mode = not args.url

    # 凭证收集（在 setup_logging 之前收集，确保敏感值可注册到过滤器）
    sensitive_values = []
    if is_titan_mode:
        # Titan 交互模式
        base_url, tenant_id, robot_id, token = prompt_credentials()
        sensitive_values = [base_url, tenant_id, robot_id, token]
    else:
        # 旧版兼容模式
        if args.url:
            sensitive_values.append(args.url)
        if args.token:
            sensitive_values.append(args.token)

    # 配置日志（注册敏感值到过滤器）
    log = setup_logging(log_path, sensitive_values=sensitive_values)

    log.info("=" * 50)
    log.info(f"QA 机器人评估工具启动 · {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    log.info("=" * 50)
    log.info(f"运行模式: {'Titan 平台（交互式凭证）' if is_titan_mode else '旧版兼容（命令行参数）'}")
    log.info("凭证: 已配置（不记录到日志）")

    # 加载测试集
    if not os.path.exists(args.testset):
        log.info("测试集不存在，自动下载...")
        download_testset(DEFAULT_TESTSET_URL, args.testset, log)
    ts = load_testset(args.testset, log)

    # 构造 bot_fn
    if is_titan_mode:
        adapter = TitanAdapter(base_url, tenant_id, robot_id, token, timeout=args.timeout)
        def bot_fn(question):
            return adapter.ask(question, log=log)
    else:
        def bot_fn(question):
            return call_bot(args.url, question, args.token, timeout=args.timeout, log=log)

    # 运行评估
    log.info(f"请求间隔: {args.sleep}s, 超时: {args.timeout}s")
    log.info(f"负样本: {'跳过' if args.positive_only else '包含'}")

    results = run_evaluation(ts, bot_fn, args.sleep, args.positive_only, log)

    # 计算指标
    metrics = compute_metrics(results, log)

    # 保存结果 JSON（不含任何凭证）
    results_path = os.path.join(SCRIPT_DIR, f"results-{ts_str}.json")
    with open(results_path, "w", encoding="utf-8") as f:
        json.dump({"results": results, "metrics": metrics}, f, ensure_ascii=False, indent=1)
    log.info(f"结果已保存: {results_path}")

    # 生成 HTML 报告
    html = render_report(results, metrics, log)
    report_path = os.path.join(SCRIPT_DIR, f"report-{ts_str}.html")
    with open(report_path, "w", encoding="utf-8") as f:
        f.write(html)
    log.info(f"报告已生成: {report_path}")
    log.info(f"日志文件: {log_path}")

    # 安全验证：检查日志文件不含敏感值
    with open(log_path, encoding="utf-8") as f:
        log_content = f.read()
    for sv in sensitive_values:
        if sv and len(sv) > 2 and sv in log_content:
            log.error(f"安全警告: 敏感值出现在日志文件中！已自动过滤。")

    # 上传到 VPS
    if not args.no_upload:
        files_to_upload = [results_path, report_path, log_path]
        log.info(f"上传文件清单: {len(files_to_upload)} 个")
        for fpath in files_to_upload:
            log.info(f"  - {os.path.basename(fpath)}")
        success = upload_to_vps(files_to_upload, args.vps_host, args.vps_dir, log)
        if success:
            log.info("上传完成，VPS 可进行后续分析")
        else:
            log.warning("上传失败，本地结果仍可用。可手动传文件给我排查。")
    else:
        log.info("--no-upload 模式，跳过上传")

    # 总结
    log.info("=" * 50)
    log.info(f"评估完成: 命中率 {metrics['hit_rate']}% ({metrics['hits']}/{metrics['total']})")
    log.info(f"  误命中: {metrics['false_hits']}/{metrics['false_total']}")
    log.info(f"  错误: {metrics['errors']}")
    log.info(f"  Precision: {metrics.get('precision', 0)}% | Recall: {metrics.get('recall', 0)}% | F1: {metrics.get('f1', 0)}")
    log.info(f"  延迟: avg={metrics['avg_latency_ms']}ms P50={metrics.get('p50_latency_ms', 0)}ms P95={metrics.get('p95_latency_ms', 0)}ms P99={metrics.get('p99_latency_ms', 0)}ms")
    log.info(f"  报告: {report_path}")
    log.info("=" * 50)

    # 打开浏览器（Mac）
    try:
        subprocess.run(["open", report_path], capture_output=True, timeout=5)
        log.info("已在浏览器中打开报告")
    except Exception:
        pass  # 非 Mac 环境忽略


if __name__ == "__main__":
    main()