html-explainer/scripts/qc_check.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
qc_check.py —— 成片体检 + 抽帧速览图。

检查项（全部来自两个血统项目的实战教训）：
  ① 流：必须有 h264 video + aac audio 两条流
  ② 时长：成片 vs 音轨 vs layout.json 期望帧数 —— 差 >0.5s 判 FAIL
     （差 ~0.9s×N 段 = 帧时长用了词边界而非容器时长的典型症状）
  ③ 音量：成片 mean_volume 与源音轨差 >3dB 判 FAIL（静音轨/占位轨）
  ④ 抽帧：按字幕切换点 + 每场景中段抽帧，纯背景帧（体积过小）判 FAIL
     （1920×1080 纯背景 PNG ≈ 90KB 上下；有内容通常 220KB+）
  ⑤ 速览图：6 列 contact sheet → out/qc_sheet.jpg（肉眼过一遍字幕带/构图）

用法：
  python qc_check.py --project . [--samples 12]
产出：
  out/qc_report.md + out/qc_frames/*.jpg + out/qc_sheet.jpg
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys


def ffmpeg_exe() -> str:
    from shutil import which
    p = which("ffmpeg")
    if p:
        return p
    import imageio_ffmpeg
    return imageio_ffmpeg.get_ffmpeg_exe()


def probe(ff, path):
    """返回 (duration_sec, streams[], 容器行列表)。"""
    r = subprocess.run([ff, "-hide_banner", "-i", path],
                       capture_output=True, text=True, errors="replace")
    lines = r.stderr.splitlines()
    dur = 0.0
    for line in lines:
        if "Duration:" in line:
            t = line.split("Duration:")[1].split(",")[0].strip()
            h, m, s = t.split(":")
            dur = int(h) * 3600 + int(m) * 60 + float(s)
    streams = [l.strip() for l in lines if "Stream #" in l]
    return dur, streams, lines


def mean_volume(ff, path):
    r = subprocess.run([ff, "-hide_banner", "-i", path, "-af", "volumedetect",
                        "-f", "null", "-"], capture_output=True, text=True, errors="replace")
    for line in r.stderr.splitlines():
        if "mean_volume:" in line:
            return float(line.split("mean_volume:")[1].replace("dB", "").strip())
    return None


def extract_frame(ff, video, sec, out_png):
    r = subprocess.run([ff, "-hide_banner", "-loglevel", "error", "-y",
                        "-i", video, "-ss", f"{sec:.3f}", "-frames:v", "1", out_png],
                       capture_output=True)
    return r.returncode == 0 and os.path.exists(out_png)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--project", default=".")
    ap.add_argument("--samples", type=int, default=12)
    args = ap.parse_args()

    proj = os.path.abspath(args.project)
    pj = json.load(open(os.path.join(proj, "project.json"), encoding="utf-8"))
    slug = pj.get("slug") or "video"
    ff = ffmpeg_exe()
    video = os.path.join(proj, "out", f"{slug}.mp4")
    audio = os.path.join(proj, "audio", "narration-full.mp3")
    layout = os.path.join(proj, "layout.json")
    subs = os.path.join(proj, "subs.json")

    report, fails, warns = [], [], []

    def sec(title):
        report.append(f"\n## {title}\n")

    if not os.path.exists(video):
        print(f"✗ 找不到成片 {video}（先渲染）", file=sys.stderr)
        return 1

    # ① 流
    sec("流")
    vdur, streams, _ = probe(ff, video)
    has_v = any("Video:" in s and "h264" in s for s in streams)
    has_a = any("Audio:" in s and "aac" in s for s in streams)
    for s in streams:
        report.append(f"- `{s}`")
    (report if has_v else fails).append("✓ 视频流 h264" if has_v else "video 流缺失或非 h264")
    if not has_a:
        warns.append("无 aac 音轨（还没配音？）")
    report.append(f"- 成片时长 {vdur:.2f}s")

    # ② 时长对齐
    sec("时长对齐（两级时钟口径）")
    expect = None
    if os.path.exists(layout):
        lay = json.load(open(layout, encoding="utf-8"))
        expect = lay.get("_total", {})
        report.append(f"- layout 期望：视频 {expect.get('video_duration_sec')}s / "
                      f"{expect.get('total_frames')} 帧 · 音轨 {expect.get('audio_duration_sec')}s")
        d = abs(vdur - float(expect.get("video_duration_sec", vdur)))
        if d > 0.5:
            fails.append(f"成片与期望时长差 {d:.2f}s（>0.5s）—— 查帧时长口径是否用了容器时长")
        else:
            report.append(f"- ✓ 与期望差 {d:.2f}s")
    if os.path.exists(audio):
        adur, _, _ = probe(ff, audio)
        d = abs(vdur - adur)
        report.append(f"- 音轨 {adur:.2f}s，成片差 {d:.2f}s")
        if d > 0.5:
            fails.append(f"音画时长差 {d:.2f}s —— 典型原因：帧时长用了词边界时长（每段少 ~0.86s）")

    # ③ 音量
    sec("音量")
    if os.path.exists(audio):
        mv_v, mv_a = mean_volume(ff, video), mean_volume(ff, audio)
        report.append(f"- 成片 mean_volume {mv_v} dB · 源 {mv_a} dB")
        if mv_v is not None and mv_a is not None and abs(mv_v - mv_a) > 3:
            fails.append(f"音量差 {abs(mv_v - mv_a):.1f} dB（静音轨或占位轨？）")

    # ④ 抽帧（字幕切换点 + 场景中段）
    sec("抽帧体检")
    qc_dir = os.path.join(proj, "out", "qc_frames")
    os.makedirs(qc_dir, exist_ok=True)
    times = []
    if os.path.exists(subs):
        sd = json.load(open(subs, encoding="utf-8"))
        pts = sorted({round(b['global_start_sec'] + 0.25, 2)
                      for s in sd['segments'] for b in s['blocks']})
        # ★ 全片均匀取 N 个（不是前 N 个）：块数多于抽样数时按步长挑，
        #   否则 QC 只覆盖前 ~1/4 片，后半段场景（本例 world→outro）抽不到。
        #   见 lessons.md #29。
        if len(pts) > args.samples:
            step = len(pts) / args.samples
            pts = [pts[int(i * step)] for i in range(args.samples)]
        times = pts
    if not times:
        times = [round(vdur * (i + 1) / (args.samples + 1), 2) for i in range(args.samples)]

    sizes = []
    for i, t in enumerate(times):
        if t >= vdur - 0.1:
            continue
        p = os.path.join(qc_dir, f"q{i:02d}_{int(t * 1000):07d}ms.jpg")
        if extract_frame(ff, video, t, p):
            sz = os.path.getsize(p)
            sizes.append((t, sz, p))

    # 阈值必须随主题底色自适应：暗底主题压缩后体积天然很小，
    # 用固定 100KB 会把「黑底 + 稀疏文字」的合规画面全判成空画面（假阳性）。
    # 做法：取本片所有抽帧体积的中位数作参考，低于中位数 35% 的帧才是可疑帧。
    # 若全部帧都极暗（中位数本身就小），再叠加一个绝对下限兜底。
    if sizes:
        med = sorted(s for _, s, _ in sizes)[len(sizes) // 2]
        floor = max(12_000, int(med * 0.35))
        report.append(f"- 抽了 {len(sizes)} 帧 → out/qc_frames/"
                      f"（本片抽帧体积中位数 {med // 1024}KB，可疑线 {floor // 1024}KB）")
        for t, sz, _ in sizes:
            flag = "  ← 可疑" if sz < floor else ""
            report.append(f"  - {t:>6.2f}s  {sz // 1024:>4}KB{flag}")
            if sz < floor:
                warns.append(f"{t}s 抽帧 {sz // 1024}KB，低于本片中位数 {med // 1024}KB 的 35%"
                             f" —— 疑似空画面（主角没渲出来？核对 out/qc_frames/ 该帧）")
    else:
        report.append("- 未能抽到任何帧")

    # ⑤ contact sheet
    sec("速览图")
    sheet = os.path.join(proj, "out", "qc_sheet.jpg")
    try:
        from PIL import Image
        imgs = [(t, p) for t, _, p in sizes]
        if imgs:
            cols = 6
            rows = (len(imgs) + cols - 1) // cols
            tw, th = 480, 270
            sheet_im = Image.new("RGB", (cols * tw, rows * th), (16, 16, 16))
            for i, (t, p) in enumerate(imgs):
                im = Image.open(p).resize((tw, th))
                sheet_im.paste(im, ((i % cols) * tw, (i // cols) * th))
            sheet_im.save(sheet, quality=85)
            report.append(f"- out/qc_sheet.jpg（{rows}×{cols}，时间点见 qc_frames 文件名）")
    except Exception as e:
        warns.append(f"速览图失败：{e}")

    # 汇总
    head = [f"# QC 报告 · {slug}", f"成片：{video}", f"时长 {vdur:.2f}s"]
    rep = "\n".join(head + report)
    if fails:
        rep += "\n\n## FAIL\n" + "\n".join(f"- ✗ {f}" for f in fails)
    if warns:
        rep += "\n\n## WARN\n" + "\n".join(f"- ⚠ {w}" for w in warns)
    if not fails and not warns:
        rep += "\n\n## 结论：全部通过 ✓（速览图仍需肉眼过一遍字幕带与构图）"
    open(os.path.join(proj, "out", "qc_report.md"), "w", encoding="utf-8", newline="\n").write(rep)
    print(rep)
    return 1 if fails else 0


if __name__ == "__main__":
    raise SystemExit(main())