xiaoxia-saas/tests/render_compare/runner.py

"""灰度对比测试 Runner — 新旧引擎批量对比 + 报告生成.

使用方法：
    # 配置环境变量
    export STAGING_API_URL=https://api.staging.example.com
    export STAGING_API_KEY=your_key

    # 运行全部 P0 场景
    python -m tests.render_compare.runner --priority P0 --output ./report/

    # 只跑指定场景
    python -m tests.render_compare.runner --scenario simple_pass_through,subtitle_rendering

对比流程：
1. 对每个场景，分别提交到 legacy 和 unified 引擎（通过 Feature Flag 白名单/百分比控制）
   - 方式A：通过内部 API 临时切换 flag（需要 admin key）
   - 方式B：提交任务时指定 engine 参数（如果 API 支持）
2. 等待任务完成，下载输出视频
3. 像素对比（SSIM + PSNR）+ 音频对比（差值RMS）
4. 生成 HTML 对比报告

注意：默认假设 API 支持 `engine` 参数来指定渲染引擎。
如果不支持，需要先通过内部 API 切换 Feature Flag，然后提交任务。
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

import httpx

# 确保项目根目录在 path 中
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from .audio_diff import AudioDiffResult, compute_audio_diff
from .scenarios import SCENARIOS, CompareScenario, get_scenarios_by_priority
from .video_diff import VideoDiffResult, compute_video_diff


@dataclass
class ScenarioResult:
    """单个场景的对比结果."""

    scenario: CompareScenario
    legacy_task_id: str = ""
    unified_task_id: str = ""
    legacy_video_path: str = ""
    unified_video_path: str = ""
    legacy_duration_sec: float = 0.0
    unified_duration_sec: float = 0.0
    video_diff: VideoDiffResult | None = None
    audio_diff: AudioDiffResult | None = None
    legacy_success: bool = False
    unified_success: bool = False
    error: str = ""

    @property
    def passed(self) -> bool:
        if not (self.legacy_success and self.unified_success):
            return False
        if self.video_diff and not self.video_diff.passed:
            return False
        if self.audio_diff and not self.audio_diff.passed:
            return False
        return True


class StagingAPI:
    """Staging 环境 API 客户端."""

    def __init__(self, base_url: str, api_key: str, internal_api_key: str = ""):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.internal_api_key = internal_api_key
        self.client = httpx.Client(timeout=30.0)

    def _headers(self, internal: bool = False) -> dict[str, str]:
        headers = {"Authorization": f"Bearer {self.api_key}"}
        if internal and self.internal_api_key:
            headers["X-API-Key"] = self.internal_api_key
        return headers

    def submit_render_task(self, plan_payload: dict[str, Any], engine: str = "") -> str:
        """提交渲染任务，返回 task_id.

        Args:
            plan_payload: EditPlan payload
            engine: 可选，指定引擎（"legacy" / "unified"）

        Returns:
            task_id
        """
        url = f"{self.base_url}/api/v1/render/compose"
        payload = dict(plan_payload)
        if engine:
            payload["engine"] = engine
        resp = self.client.post(url, json=payload, headers=self._headers())
        resp.raise_for_status()
        data = resp.json()
        return data.get("task_id") or data.get("id", "")

    def get_task_status(self, task_id: str) -> dict[str, Any]:
        """获取任务状态."""
        url = f"{self.base_url}/api/v1/tasks/{task_id}"
        resp = self.client.get(url, headers=self._headers())
        resp.raise_for_status()
        return resp.json()

    def wait_for_task(self, task_id: str, timeout: float = 300.0, poll_interval: float = 3.0) -> dict[str, Any]:
        """等待任务完成.

        Returns:
            最终任务状态

        Raises:
            TimeoutError: 超时
        """
        start = time.time()
        while time.time() - start < timeout:
            status = self.get_task_status(task_id)
            state = status.get("status", "")
            if state in ("completed", "success", "done", "failed", "error"):
                return status
            time.sleep(poll_interval)
        raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

    def set_feature_flag(self, flag_name: str, enabled: bool, percentage: int = 0, whitelist: list[str] | None = None):
        """通过内部 API 设置 Feature Flag.

        用于不支持 engine 参数的场景，切换全局灰度比例。
        """
        if not self.internal_api_key:
            raise ValueError("internal_api_key is required for feature flag operations")
        url = f"{self.base_url}/api/v1/internal/feature-flags/{flag_name}"
        body: dict[str, Any] = {"enabled": enabled, "percentage": percentage}
        if whitelist is not None:
            body["whitelist"] = whitelist
        resp = self.client.put(url, json=body, headers=self._headers(internal=True))
        resp.raise_for_status()
        return resp.json()

    def get_feature_flag(self, flag_name: str) -> dict[str, Any]:
        """获取 Feature Flag 配置."""
        if not self.internal_api_key:
            raise ValueError("internal_api_key is required")
        url = f"{self.base_url}/api/v1/internal/feature-flags/{flag_name}"
        resp = self.client.get(url, headers=self._headers(internal=True))
        resp.raise_for_status()
        return resp.json()

    def download_video(self, video_url: str, output_path: str | Path) -> Path:
        """下载视频文件."""
        output_path = Path(output_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        with self.client.stream("GET", video_url, timeout=60.0) as resp:
            resp.raise_for_status()
            with open(output_path, "wb") as f:
                for chunk in resp.iter_bytes():
                    f.write(chunk)
        return output_path


class CompareRunner:
    """新旧引擎对比 Runner."""

    # 全局默认阈值（唯一真实来源，所有入口统一引用）
    DEFAULT_SSIM_THRESHOLD: float = 0.95
    DEFAULT_PSNR_THRESHOLD: float = 28.0
    DEFAULT_AUDIO_SIMILARITY_THRESHOLD: float = 0.90
    DEFAULT_DURATION_TOLERANCE: float = 0.1
    DEFAULT_TASK_TIMEOUT: float = 300.0

    def __init__(
        self,
        api: StagingAPI,
        output_dir: Path,
        *,
        ssim_threshold: float | None = None,
        psnr_threshold: float | None = None,
        audio_similarity_threshold: float | None = None,
        task_timeout: float | None = None,
        flag_mode: bool = False,  # 是否使用 Feature Flag 方式切换引擎
        duration_tolerance: float | None = None,
    ):
        self.api = api
        self.output_dir = output_dir
        self.ssim_threshold = ssim_threshold if ssim_threshold is not None else self.DEFAULT_SSIM_THRESHOLD
        self.psnr_threshold = psnr_threshold if psnr_threshold is not None else self.DEFAULT_PSNR_THRESHOLD
        self.audio_similarity_threshold = (
            audio_similarity_threshold
            if audio_similarity_threshold is not None
            else self.DEFAULT_AUDIO_SIMILARITY_THRESHOLD
        )
        self.duration_tolerance = (
            duration_tolerance if duration_tolerance is not None else self.DEFAULT_DURATION_TOLERANCE
        )
        self.task_timeout = task_timeout if task_timeout is not None else self.DEFAULT_TASK_TIMEOUT
        self.flag_mode = flag_mode
        self.results: list[ScenarioResult] = []
        # flag_mode 下保存原始配置，测试结束后恢复（防污染线上）
        self._original_flag_config: dict[str, Any] | None = None

    def run_scenario(self, scenario: CompareScenario) -> ScenarioResult:
        """运行单个场景对比."""
        print(f"\n{'='*60}")
        print(f"[{scenario.priority}] {scenario.id}: {scenario.name}")
        print(f"  {scenario.description}")

        result = ScenarioResult(scenario=scenario)
        scenario_dir = self.output_dir / scenario.id
        scenario_dir.mkdir(parents=True, exist_ok=True)

        try:
            # 1. 提交两个引擎的任务
            legacy_task_id = self._submit_with_engine(scenario, "legacy")
            unified_task_id = self._submit_with_engine(scenario, "unified")
            result.legacy_task_id = legacy_task_id
            result.unified_task_id = unified_task_id
            print(f"  legacy task:  {legacy_task_id}")
            print(f"  unified task: {unified_task_id}")

            # 2. 等待完成
            print("  waiting for legacy...", end="", flush=True)
            legacy_status = self.api.wait_for_task(legacy_task_id, timeout=self.task_timeout)
            result.legacy_success = legacy_status.get("status") in ("completed", "success", "done")
            legacy_video_url = legacy_status.get("output_url", "") or legacy_status.get("video_url", "")
            print(f" {'✅' if result.legacy_success else '❌'} ({legacy_status.get('duration_sec', '?')}s)")

            print("  waiting for unified...", end="", flush=True)
            unified_status = self.api.wait_for_task(unified_task_id, timeout=self.task_timeout)
            result.unified_success = unified_status.get("status") in ("completed", "success", "done")
            unified_video_url = unified_status.get("output_url", "") or unified_status.get("video_url", "")
            print(f" {'✅' if result.unified_success else '❌'} ({unified_status.get('duration_sec', '?')}s)")

            result.legacy_duration_sec = float(legacy_status.get("duration_sec", 0))
            result.unified_duration_sec = float(unified_status.get("duration_sec", 0))

            if not (result.legacy_success and result.unified_success):
                result.error = f"Legacy success={result.legacy_success}, Unified success={result.unified_success}"
                print("  ⚠️  任务未全部成功，跳过对比")
                return result

            # 3. 下载视频
            print("  downloading...", end="", flush=True)
            legacy_path = self.api.download_video(legacy_video_url, scenario_dir / "legacy.mp4")
            unified_path = self.api.download_video(unified_video_url, scenario_dir / "unified.mp4")
            result.legacy_video_path = str(legacy_path)
            result.unified_video_path = str(unified_path)
            print(" ✅")

            # 4. 像素对比
            print("  computing video diff...", end="", flush=True)
            result.video_diff = compute_video_diff(
                legacy_path,
                unified_path,
                ssim_threshold=self.ssim_threshold,
                psnr_threshold=self.psnr_threshold,
                duration_tolerance=self.duration_tolerance,
            )
            print(
                f" SSIM={result.video_diff.avg_ssim:.4f} PSNR={result.video_diff.avg_psnr:.2f}dB {'✅' if result.video_diff.passed else '❌'}"
            )

            # 5. 音频对比（仅当都有音频时）
            from .audio_diff import probe_has_audio

            legacy_has_audio = probe_has_audio(legacy_path)
            unified_has_audio = probe_has_audio(unified_path)

            if legacy_has_audio and unified_has_audio:
                print("  computing audio diff...", end="", flush=True)
                result.audio_diff = compute_audio_diff(
                    legacy_path,
                    unified_path,
                    similarity_threshold=self.audio_similarity_threshold,
                )
                print(
                    f" similarity={result.audio_diff.similarity_score:.4f} {'✅' if result.audio_diff.passed else '❌'}"
                )
            elif legacy_has_audio != unified_has_audio:
                result.error = f"音频不一致: legacy_has_audio={legacy_has_audio}, unified_has_audio={unified_has_audio}"
                print(f"  ⚠️  音频不一致: legacy={legacy_has_audio}, unified={unified_has_audio}")
            else:
                print("  audio: both silent (skip)")

        except Exception as e:
            result.error = str(e)
            print(f"  ❌ 错误: {e}")

        self.results.append(result)
        return result

    def _submit_with_engine(self, scenario: CompareScenario, engine: str) -> str:
        """提交指定引擎的任务.

        如果 flag_mode=True，通过 Feature Flag 切换，否则通过 engine 参数。
        """
        if self.flag_mode:
            # 先设置 flag（用白名单方式，确保只有当前测试用户命中）
            percentage = 0 if engine == "legacy" else 100
            self.api.set_feature_flag("render_engine", enabled=True, percentage=percentage)
            time.sleep(1)  # 给 worker 一点时间刷新配置
            return self.api.submit_render_task(scenario.plan_payload)
        else:
            return self.api.submit_render_task(scenario.plan_payload, engine=engine)

    def run_all(self, scenarios: list[CompareScenario]) -> list[ScenarioResult]:
        """运行所有场景.

        flag_mode=True 时，测试开始前保存原始 Feature Flag 配置，
        结束后（无论成功失败）自动恢复，避免污染线上环境。
        """
        print(f"\n灰度对比测试开始 - {len(scenarios)} 个场景")
        print(f"输出目录: {self.output_dir}")
        print(f"视频阈值: SSIM>={self.ssim_threshold}, PSNR>={self.psnr_threshold}dB")
        print(f"音频阈值: similarity>={self.audio_similarity_threshold}")

        # flag_mode：保存原始配置，测试结束后恢复（防污染）
        if self.flag_mode:
            try:
                self._original_flag_config = self.api.get_feature_flag("render_engine")
                print(f"  [flag_mode] 已保存原始配置: {self._original_flag_config}")
            except Exception as e:
                print(f"  ⚠️  [flag_mode] 保存原始配置失败: {e}")
                print("  为避免污染线上，将中止测试。请检查 internal_api_key 配置。")
                return self.results

        try:
            for i, scenario in enumerate(scenarios):
                print(f"\n进度: {i+1}/{len(scenarios)}")
                self.run_scenario(scenario)
        finally:
            # 始终恢复原始 flag 配置
            if self.flag_mode and self._original_flag_config:
                try:
                    orig = self._original_flag_config
                    self.api.set_feature_flag(
                        "render_engine",
                        enabled=orig.get("enabled", False),
                        percentage=orig.get("percentage", 0),
                        whitelist=orig.get("whitelist"),
                    )
                    print("\n[flag_mode] ✅ 已恢复原始 Feature Flag 配置")
                except Exception as e:
                    print(f"\n[flag_mode] ❌ 恢复 Feature Flag 失败: {e}")
                    print("  请手动检查并恢复 render_engine flag 配置！")

        return self.results

    def summary(self) -> dict[str, Any]:
        """生成汇总统计."""
        total = len(self.results)
        passed = sum(1 for r in self.results if r.passed)
        failed = total - passed

        # 性能对比
        perf_diffs = []
        for r in self.results:
            if r.legacy_success and r.unified_success and r.legacy_duration_sec > 0:
                diff_pct = (r.unified_duration_sec - r.legacy_duration_sec) / r.legacy_duration_sec * 100
                perf_diffs.append(diff_pct)
        avg_perf_diff = sum(perf_diffs) / len(perf_diffs) if perf_diffs else 0.0

        return {
            "total": total,
            "passed": passed,
            "failed": failed,
            "pass_rate": f"{passed/total*100:.1f}%" if total > 0 else "0%",
            "avg_perf_diff_pct": round(avg_perf_diff, 2),
            "scenarios": [self._result_to_dict(r) for r in self.results],
            "timestamp": datetime.now().isoformat(),
            "ssim_threshold": self.ssim_threshold,
            "psnr_threshold": self.psnr_threshold,
            "audio_threshold": self.audio_similarity_threshold,
        }

    def _result_to_dict(self, r: ScenarioResult) -> dict[str, Any]:
        return {
            "id": r.scenario.id,
            "name": r.scenario.name,
            "priority": r.scenario.priority,
            "passed": r.passed,
            "legacy_success": r.legacy_success,
            "unified_success": r.unified_success,
            "legacy_duration_sec": r.legacy_duration_sec,
            "unified_duration_sec": r.unified_duration_sec,
            "video_diff": r.video_diff.to_dict() if r.video_diff else None,
            "audio_diff": r.audio_diff.to_dict() if r.audio_diff else None,
            "error": r.error,
        }


def generate_html_report(summary: dict[str, Any], output_path: Path):
    """生成 HTML 对比报告."""
    scenarios = summary["scenarios"]

    # 按通过/失败分组
    # 构建场景卡片
    scenario_cards = ""
    for s in scenarios:
        status_class = "pass" if s["passed"] else "fail"
        status_text = "✅ 通过" if s["passed"] else "❌ 失败"

        vdiff = s.get("video_diff") or {}
        adiff = s.get("audio_diff") or {}

        video_info = ""
        if vdiff:
            video_info = f"""
            <div class="metric-row">
              <span>SSIM:</span>
              <span class="{'good' if vdiff.get('avg_ssim', 0) >= 0.95 else 'warn'}">{vdiff.get('avg_ssim', 0):.4f}</span>
            </div>
            <div class="metric-row">
              <span>PSNR:</span>
              <span>{vdiff.get('avg_psnr', 0):.2f} dB</span>
            </div>
            <div class="metric-row">
              <span>时长差:</span>
              <span>{vdiff.get('duration_diff', 0):.3f}s</span>
            </div>
            """

        audio_info = ""
        if adiff:
            audio_info = f"""
            <div class="metric-row">
              <span>音频相似度:</span>
              <span class="{'good' if adiff.get('similarity_score', 0) >= 0.9 else 'warn'}">{adiff.get('similarity_score', 0):.4f}</span>
            </div>
            <div class="metric-row">
              <span>差值 RMS:</span>
              <span>{adiff.get('diff_rms_db', 0):.2f} dB</span>
            </div>
            """

        perf_info = ""
        if s["legacy_duration_sec"] and s["unified_duration_sec"]:
            diff = s["unified_duration_sec"] - s["legacy_duration_sec"]
            pct = diff / s["legacy_duration_sec"] * 100 if s["legacy_duration_sec"] else 0
            trend = "🔴" if pct > 10 else ("🟡" if pct > 0 else "🟢")
            perf_info = f"""
            <div class="perf-row">
              <span>Legacy: {s['legacy_duration_sec']:.2f}s</span>
              <span>Unified: {s['unified_duration_sec']:.2f}s</span>
              <span>{trend} {pct:+.1f}%</span>
            </div>
            """

        error_info = f'<div class="error-box">{s["error"]}</div>' if s["error"] else ""

        scenario_cards += f"""
        <div class="card {status_class}">
          <div class="card-header">
            <span class="badge">{s['priority']}</span>
            <span class="scenario-name">{s['name']}</span>
            <span class="status {status_class}">{status_text}</span>
          </div>
          <div class="card-body">
            <div class="grid-2">
              <div>
                <h4>视频质量</h4>
                {video_info or '<p class="muted">无数据</p>'}
              </div>
              <div>
                <h4>音频质量</h4>
                {audio_info or '<p class="muted">无音频或跳过</p>'}
              </div>
            </div>
            <div>
              <h4>性能对比</h4>
              {perf_info or '<p class="muted">无数据</p>'}
            </div>
            {error_info}
          </div>
        </div>
        """

    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>统一渲染引擎灰度对比报告</title>
  <style>
    * {{ box-sizing: border-box; margin: 0; padding: 0; }}
    body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #f5f5f5; color: #333; padding: 20px; }}
    .container {{ max-width: 1200px; margin: 0 auto; }}
    h1 {{ margin-bottom: 20px; font-size: 24px; }}
    .summary {{ background: white; border-radius: 12px; padding: 24px; margin-bottom: 24px; display: flex; gap: 32px; flex-wrap: wrap; }}
    .summary-item {{ text-align: center; }}
    .summary-item .value {{ font-size: 32px; font-weight: bold; margin-bottom: 4px; }}
    .summary-item .label {{ color: #666; font-size: 14px; }}
    .pass .value {{ color: #10b981; }}
    .fail .value {{ color: #ef4444; }}
    .card {{ background: white; border-radius: 12px; margin-bottom: 16px; overflow: hidden; border-left: 4px solid #10b981; }}
    .card.fail {{ border-left-color: #ef4444; }}
    .card-header {{ padding: 16px 20px; background: #fafafa; display: flex; align-items: center; gap: 12px; border-bottom: 1px solid #eee; }}
    .badge {{ background: #e5e7eb; color: #374151; padding: 2px 8px; border-radius: 4px; font-size: 12px; font-weight: 600; }}
    .scenario-name {{ flex: 1; font-weight: 600; }}
    .status {{ font-weight: 600; }}
    .status.pass {{ color: #10b981; }}
    .status.fail {{ color: #ef4444; }}
    .card-body {{ padding: 20px; }}
    .grid-2 {{ display: grid; grid-template-columns: 1fr 1fr; gap: 24px; margin-bottom: 16px; }}
    h4 {{ margin-bottom: 12px; color: #374151; font-size: 14px; }}
    .metric-row {{ display: flex; justify-content: space-between; padding: 6px 0; font-size: 14px; }}
    .metric-row .good {{ color: #10b981; font-weight: 600; }}
    .metric-row .warn {{ color: #f59e0b; font-weight: 600; }}
    .perf-row {{ display: flex; gap: 24px; padding: 8px 0; font-size: 14px; background: #f9fafb; padding: 12px; border-radius: 8px; }}
    .error-box {{ background: #fef2f2; color: #dc2626; padding: 12px; border-radius: 8px; margin-top: 12px; font-size: 13px; }}
    .muted {{ color: #9ca3af; font-size: 14px; }}
    .timestamp {{ text-align: center; color: #9ca3af; font-size: 12px; margin-top: 24px; }}
  </style>
</head>
<body>
  <div class="container">
    <h1>🎬 统一渲染引擎灰度对比报告</h1>
    <div class="summary">
      <div class="summary-item">
        <div class="value">{summary['total']}</div>
        <div class="label">总场景数</div>
      </div>
      <div class="summary-item pass">
        <div class="value">{summary['passed']}</div>
        <div class="label">通过</div>
      </div>
      <div class="summary-item fail">
        <div class="value">{summary['failed']}</div>
        <div class="label">失败</div>
      </div>
      <div class="summary-item">
        <div class="value">{summary['pass_rate']}</div>
        <div class="label">通过率</div>
      </div>
      <div class="summary-item">
        <div class="value {'good' if summary['avg_perf_diff_pct'] <= 0 else 'warn'}" style="font-size: 24px; color: {'#10b981' if summary['avg_perf_diff_pct'] <= 0 else '#f59e0b'}">{summary['avg_perf_diff_pct']:+.1f}%</div>
        <div class="label">平均性能差异</div>
      </div>
    </div>
    {scenario_cards}
    <div class="timestamp">生成时间: {summary['timestamp']}</div>
  </div>
</body>
</html>"""

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(html, encoding="utf-8")
    return output_path


def main():
    parser = argparse.ArgumentParser(description="统一渲染引擎灰度对比测试")
    parser.add_argument("--priority", default="P0", choices=["P0", "P1", "P2"], help="最低优先级")
    parser.add_argument("--scenarios", default="", help="指定场景ID，逗号分隔")
    parser.add_argument("--output", default="./gray_compare_report", help="输出目录")
    parser.add_argument("--ssim-threshold", type=float, default=None, help="SSIM阈值（默认0.95）")
    parser.add_argument("--psnr-threshold", type=float, default=None, help="PSNR阈值(dB)（默认28.0）")
    parser.add_argument("--audio-threshold", type=float, default=None, help="音频相似度阈值（默认0.90）")
    parser.add_argument("--flag-mode", action="store_true", help="使用Feature Flag方式切换引擎")
    parser.add_argument("--task-timeout", type=float, default=300.0, help="单任务超时时间(秒)")
    args = parser.parse_args()

    base_url = os.environ.get("STAGING_API_URL", "")
    api_key = os.environ.get("STAGING_API_KEY", "")
    internal_key = os.environ.get("STAGING_INTERNAL_API_KEY", "")

    if not base_url or not api_key:
        print("❌ 请设置环境变量 STAGING_API_URL 和 STAGING_API_KEY")
        sys.exit(1)

    # 选择场景
    if args.scenarios:
        scenario_ids = [s.strip() for s in args.scenarios.split(",")]
        selected = [s for s in SCENARIOS if s.id in scenario_ids]
        if not selected:
            print(f"❌ 未找到匹配的场景: {scenario_ids}")
            print(f"可用场景: {[s.id for s in SCENARIOS]}")
            sys.exit(1)
    else:
        selected = get_scenarios_by_priority(args.priority)

    output_dir = Path(args.output).resolve()
    output_dir.mkdir(parents=True, exist_ok=True)

    api = StagingAPI(base_url, api_key, internal_key)
    runner = CompareRunner(
        api,
        output_dir,
        ssim_threshold=args.ssim_threshold,
        psnr_threshold=args.psnr_threshold,
        audio_similarity_threshold=args.audio_threshold,
        flag_mode=args.flag_mode,
        task_timeout=args.task_timeout,
    )

    runner.run_all(selected)

    # 生成报告
    summary = runner.summary()

    # JSON 报告
    json_path = output_dir / "report.json"
    json_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8")

    # HTML 报告
    html_path = output_dir / "report.html"
    generate_html_report(summary, html_path)

    print(f"\n{'='*60}")
    print(f"对比完成: {summary['passed']}/{summary['total']} 通过 ({summary['pass_rate']})")
    print(f"报告: {html_path}")
    print(f"JSON: {json_path}")


if __name__ == "__main__":
    main()