"""灰度对比测试 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"""
SSIM: {vdiff.get('avg_ssim', 0):.4f}
PSNR: {vdiff.get('avg_psnr', 0):.2f} dB
时长差: {vdiff.get('duration_diff', 0):.3f}s
""" audio_info = "" if adiff: audio_info = f"""
音频相似度: {adiff.get('similarity_score', 0):.4f}
差值 RMS: {adiff.get('diff_rms_db', 0):.2f} dB
""" 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"""
Legacy: {s['legacy_duration_sec']:.2f}s Unified: {s['unified_duration_sec']:.2f}s {trend} {pct:+.1f}%
""" error_info = f'
{s["error"]}
' if s["error"] else "" scenario_cards += f"""
{s['priority']} {s['name']} {status_text}

视频质量

{video_info or '

无数据

'}

音频质量

{audio_info or '

无音频或跳过

'}

性能对比

{perf_info or '

无数据

'}
{error_info}
""" html = f""" 统一渲染引擎灰度对比报告

🎬 统一渲染引擎灰度对比报告

{summary['total']}
总场景数
{summary['passed']}
通过
{summary['failed']}
失败
{summary['pass_rate']}
通过率
{summary['avg_perf_diff_pct']:+.1f}%
平均性能差异
{scenario_cards}
生成时间: {summary['timestamp']}
""" 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()