Blame for skill/scripts/validate_release.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""用 ffprobe 检查成品，生成报告并创建独立、无密钥的源码包和完整发布包。"""
import argparse
import hashlib
import json
import re
import subprocess
import zipfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
STORY = json.loads((ROOT / "story.json").read_text(encoding="utf-8"))
TIMING = json.loads((ROOT / "timing.json").read_text(encoding="utf-8"))
OUT = ROOT / "output"
VIDEO = OUT / f"{STORY['slug']}.mp4"


def parse_probe(path: Path) -> dict:
    stdout = subprocess.check_output(["ffprobe", "-v", "error", "-show_format", "-show_streams",
                                     "-of", "json", str(path)], text=True)
    return json.loads(stdout)


def inspect() -> dict:
    if not VIDEO.is_file() or VIDEO.stat().st_size < 100_000:
        raise FileNotFoundError(f"真正渲染的视频不存在或不完整: {VIDEO}")
    data = parse_probe(VIDEO)
    video = next((s for s in data["streams"] if s["codec_type"] == "video"), None)
    audio = next((s for s in data["streams"] if s["codec_type"] == "audio"), None)
    duration = float(data["format"]["duration"])
    width = int(STORY["video"]["width"])
    height = int(STORY["video"]["height"])
    fps = int(STORY["video"]["fps"])
    actual_fps = (video.get("avg_frame_rate") or video.get("r_frame_rate") or "0/1") if video else "0/1"
    values = actual_fps.split("/")
    actual_rate = float(values[0]) / max(float(values[1]), .0001)
    caption_file = OUT / f"{STORY['slug']}.srt"
    snapshots = list((OUT / "snapshots").glob("*.png"))
    snapshots += list((OUT / "snapshots-fixed").glob("*.png"))
    technical = bool(
        video and audio and video.get("codec_name") == "h264" and
        audio.get("codec_name") == "aac" and int(video["width"]) == width and
        int(video["height"]) == height and abs(actual_rate - fps) < .1 and
        abs(duration - float(TIMING["duration"])) < .12 and
        caption_file.is_file() and caption_file.stat().st_size > 10
    )
    from narrate import validate_story, clean
    subtitles = json.loads((ROOT/'captions.json').read_text(encoding='utf-8'))
    expected = clean(''.join(line['speech'] for group in validate_story(STORY) for line in group))
    emitted = clean(''.join(c.get('text','') for c in subtitles))
    exact_subtitles = bool(expected == emitted and len(subtitles)>0 and
                           all(c.get('precise_tts_boundaries') is True for c in subtitles) and
                           TIMING.get('verbatim_captions') is True and not TIMING.get('warnings'))
    if not exact_subtitles:
        raise ValueError('配音原稿与字幕文字/词边界不一致，拒绝打包。请重新生成真实配音字幕。')
    has_snapshots = len(snapshots) >= 3
    has_real_voice = not TIMING.get("mock", False)
    h = hashlib.sha256()
    with VIDEO.open("rb") as f:
        for chunk in iter(lambda: f.read(2 * 1024 * 1024), b""):
            h.update(chunk)
    result = {
        "file": str(VIDEO), "bytes": VIDEO.stat().st_size, "sha256": h.hexdigest(),
        "duration": round(duration, 6), "width": video.get("width") if video else None,
        "height": video.get("height") if video else None, "fps": actual_fps,
        "video_codec": video.get("codec_name") if video else None,
        "audio_codec": audio.get("codec_name") if audio else None,
        "audio_rate": audio.get("sample_rate") if audio else None,
        "snapshot_count": len(snapshots), "tts_mock": TIMING.get("mock", False),
        "subtitle_verbatim_pass": exact_subtitles,
        "technical_pass": technical,
        "has_story_snapshots": has_snapshots,
        "manual_science_review_required": True,
        "qa_pass": bool(technical and has_snapshots and has_real_voice),
        "manual_review": "required: watch video and listen before classroom use",
        "warnings": TIMING.get("warnings", [])
    }
    OUT.mkdir(exist_ok=True)
    (OUT / "validation.json").write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    return result


def eligible_files(full: bool):
    allowed = ["README.md", "requirements.txt", "story.json", "timing.json", "captions.json", "meta.json",
               "package.json", "index.html"]
    found = [ROOT / n for n in allowed if (ROOT / n).is_file()]
    for dirname in ("scripts", "assets"):
        found.extend(p for p in (ROOT / dirname).rglob("*") if p.is_file() and "__pycache__" not in p.parts)
    found.extend([OUT / f"{STORY['slug']}.srt", OUT / "validation.json"])
    if full:
        found.append(VIDEO)
        for sn in ("snapshots", "snapshots-fixed"):
            found.extend(p for p in (OUT / sn).glob("*.png") if p.is_file())
    for p in found:
        if (p.is_file() and not p.name.startswith(".env") and p.suffix.lower() not in
                (".pyc", ".pem", ".key", ".pfx", ".ttf", ".otf", ".woff", ".woff2") and ".git" not in p.parts):
            yield p


def pack_release() -> list[dict]:
    packages = []
    for full in (False, True):
        target = OUT / f"{STORY['slug']}-{'complete' if full else 'source'}.zip"
        members = list(eligible_files(full))
        with zipfile.ZipFile(target, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as z:
            for p in members:
                z.write(p, (ROOT.name + "/" + p.relative_to(ROOT).as_posix()))
        with zipfile.ZipFile(target) as z:
            if z.testzip() is not None:
                raise ValueError(f"损坏的 ZIP: {target}")
        packages.append({"file": str(target), "size_bytes": target.stat().st_size,
                         "entries": len(members)})
    return packages


def main() -> None:
    parser = argparse.ArgumentParser(description="检查成片、生成 validation.json、打包")
    parser.add_argument("--inspect-only", action="store_true", help="只检查和输出报告，不打包")
    opts = parser.parse_args()
    result = inspect()
    print(json.dumps(result, ensure_ascii=False, indent=2))
    if not result["qa_pass"]:
        issues = []
        if not result["technical_pass"]:
            issues.append("ffprobe 媒体规格/时长/字幕文件错误")
        if not result["has_story_snapshots"]:
            issues.append("缺少至少3张视频关键帧")
        if result["tts_mock"]:
            issues.append("仍使用模拟配音，禁止发布")
        raise SystemExit("不能发布：" + "；".join(issues))
    if not opts.inspect_only:
        print(json.dumps({"packages": pack_release()}, ensure_ascii=False, indent=2))
    print("技术检查通过；仍需人工完整观看试听后才能作为课堂正式成品。")


if __name__ == "__main__":
    main()