224 lines
12 KiB
Python
224 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify render manifests and media using local ffprobe/ffmpeg only."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Verify one edit or highlight render without modifying it.")
|
|
parser.add_argument("--repo", type=Path, default=Path("."), help="Repository root used to resolve manifest paths.")
|
|
parser.add_argument("--project-dir", type=Path, required=True, help="Directory containing render-manifest.json.")
|
|
parser.add_argument("--manifest", type=Path, help="Manifest override (default: PROJECT_DIR/render-manifest.json).")
|
|
parser.add_argument("--qa-report", type=Path, help="QA report override (default: PROJECT_DIR/qa-report.json).")
|
|
parser.add_argument("--media", type=Path, help="Media override; otherwise use manifest outputPath.")
|
|
parser.add_argument("--duration-tolerance", type=float, default=0.10, help="Allowed manifest/probe difference in seconds.")
|
|
parser.add_argument("--deep", action="store_true", help="Also run read-only black, silence, and audio-peak FFmpeg probes.")
|
|
parser.add_argument("--format", choices=("text", "json"), default="text")
|
|
return parser.parse_args()
|
|
|
|
|
|
def read_object(path: Path) -> dict:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"top-level JSON is not an object: {path}")
|
|
return value
|
|
|
|
|
|
def run(command: list[str]) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(command, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
|
|
|
|
|
|
def resolve_media(value: str, repo: Path, project: Path) -> tuple[Path | None, list[str], list[str]]:
|
|
raw = Path(value)
|
|
# Current manifests record either an absolute path, a repository-relative path such as
|
|
# output/edit-projects/<id>/final.mp4, or a project-relative path such as final.mp4.
|
|
candidates = [raw] if raw.is_absolute() else [repo / raw, project / raw, project / raw.name]
|
|
normalized: list[Path] = []
|
|
for candidate in candidates:
|
|
resolved = candidate.resolve()
|
|
if resolved not in normalized:
|
|
normalized.append(resolved)
|
|
existing = [candidate for candidate in normalized if candidate.is_file()]
|
|
chosen = existing[0] if len(existing) == 1 else None
|
|
return chosen, [str(path) for path in normalized], [str(path) for path in existing]
|
|
|
|
|
|
def qa_summary(path: Path) -> dict:
|
|
if not path.is_file():
|
|
return {"present": False, "mode": "missing", "declaredPassed": None, "failedChecks": []}
|
|
report = read_object(path)
|
|
checks = [c for c in report.get("checks", []) if isinstance(c, dict)]
|
|
names = {str(c.get("name")) for c in checks}
|
|
expected = {"black_frames", "long_silence", "audio_clipping"}
|
|
mode = "measured" if expected <= names else "partial"
|
|
return {
|
|
"present": True,
|
|
"mode": mode,
|
|
"declaredPassed": report.get("passed"),
|
|
"failedChecks": [str(c.get("name")) for c in checks if c.get("passed") is False],
|
|
"invalidChecks": [str(c.get("name")) for c in checks if not isinstance(c.get("passed"), bool)],
|
|
"missingProbes": sorted(expected - names),
|
|
}
|
|
|
|
|
|
def deep_probes(ffmpeg: str, media: Path) -> tuple[dict, list[dict]]:
|
|
findings: list[dict] = []
|
|
probes: dict = {}
|
|
commands = {
|
|
"black": [ffmpeg, "-hide_banner", "-v", "info", "-i", str(media), "-vf", "blackdetect=d=0.5:pic_th=0.98", "-an", "-f", "null", "-"],
|
|
"silence": [ffmpeg, "-hide_banner", "-v", "info", "-i", str(media), "-af", "silencedetect=noise=-45dB:d=2", "-vn", "-f", "null", "-"],
|
|
"peaks": [ffmpeg, "-hide_banner", "-v", "info", "-i", str(media), "-af", "astats=metadata=1:reset=1", "-vn", "-f", "null", "-"],
|
|
}
|
|
for name, command in commands.items():
|
|
result = run(command)
|
|
output = result.stdout + result.stderr
|
|
probes[name] = {"exitCode": result.returncode}
|
|
if result.returncode != 0:
|
|
findings.append({"severity": "error", "code": f"FFMPEG_{name.upper()}_FAILED", "detail": f"exit={result.returncode}"})
|
|
continue
|
|
if name == "black":
|
|
count = len(re.findall(r"black_start:", output))
|
|
probes[name]["rangeCount"] = count
|
|
if count:
|
|
findings.append({"severity": "error", "code": "BLACK_RANGE_DETECTED", "detail": f"ranges={count}; threshold d=0.5,pic_th=0.98"})
|
|
elif name == "silence":
|
|
count = len(re.findall(r"silence_start:", output))
|
|
probes[name]["rangeCount"] = count
|
|
if count:
|
|
findings.append({"severity": "warning", "code": "LONG_SILENCE_DETECTED", "detail": f"ranges={count}; threshold -45dB for 2s"})
|
|
else:
|
|
peaks = []
|
|
for value in re.findall(r"Peak level dB:\s*([^\s]+)", output):
|
|
try:
|
|
peaks.append(float(value))
|
|
except ValueError:
|
|
pass
|
|
maximum = max(peaks) if peaks else None
|
|
probes[name]["maximumPeakDbfs"] = maximum
|
|
if maximum is None:
|
|
findings.append({"severity": "warning", "code": "AUDIO_PEAK_UNMEASURED", "detail": "astats emitted no numeric peak"})
|
|
elif maximum >= -0.1:
|
|
findings.append({"severity": "warning", "code": "AUDIO_PEAK_UNSAFE", "detail": f"maximum={maximum} dBFS"})
|
|
return probes, findings
|
|
|
|
|
|
def main() -> int:
|
|
opts = args()
|
|
if opts.duration_tolerance < 0:
|
|
print("error: --duration-tolerance must be nonnegative", file=sys.stderr)
|
|
return 3
|
|
repo = opts.repo.resolve()
|
|
project = opts.project_dir.resolve()
|
|
manifest_path = (opts.manifest or project / "render-manifest.json").resolve()
|
|
qa_path = (opts.qa_report or project / "qa-report.json").resolve()
|
|
findings: list[dict] = []
|
|
try:
|
|
manifest = read_object(manifest_path)
|
|
except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 3
|
|
try:
|
|
qa = qa_summary(qa_path)
|
|
except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
|
|
qa = {"present": True, "mode": "invalid", "declaredPassed": None, "failedChecks": [],
|
|
"invalidChecks": [], "missingProbes": []}
|
|
findings.append({"severity": "error", "code": "QA_REPORT_INVALID", "detail": str(exc)})
|
|
if qa["mode"] == "missing":
|
|
findings.append({"severity": "error", "code": "QA_REPORT_MISSING", "detail": str(qa_path)})
|
|
elif qa["mode"] == "partial":
|
|
findings.append({"severity": "warning", "code": "QA_PARTIAL", "detail": f"missing probes={','.join(qa['missingProbes'])}"})
|
|
if qa["present"] and not isinstance(qa["declaredPassed"], bool):
|
|
findings.append({"severity": "error", "code": "QA_PASSED_INVALID", "detail": "passed must be a JSON boolean"})
|
|
if qa.get("invalidChecks"):
|
|
findings.append({"severity": "error", "code": "QA_CHECK_RESULT_INVALID", "detail": ",".join(qa["invalidChecks"])})
|
|
if qa["declaredPassed"] is True and qa["failedChecks"]:
|
|
findings.append({"severity": "error", "code": "QA_RESULT_INCONSISTENT", "detail": "passed=true with failed checks: " + ",".join(qa["failedChecks"])})
|
|
if qa["declaredPassed"] is False:
|
|
findings.append({"severity": "error", "code": "QA_DECLARED_FAILED", "detail": ",".join(qa["failedChecks"]) or "report passed=false"})
|
|
media_value = str(opts.media) if opts.media else str(manifest.get("outputPath") or "")
|
|
if not media_value:
|
|
print("error: no --media and manifest outputPath is empty", file=sys.stderr)
|
|
return 3
|
|
if opts.media:
|
|
explicit = opts.media.resolve()
|
|
media, candidates, existing = (explicit if explicit.is_file() else None), [str(explicit)], ([str(explicit)] if explicit.is_file() else [])
|
|
else:
|
|
media, candidates, existing = resolve_media(media_value, repo, project)
|
|
if len(existing) > 1:
|
|
findings.append({"severity": "error", "code": "MEDIA_RESOLUTION_AMBIGUOUS", "detail": "multiple candidates exist: " + ",".join(existing)})
|
|
if media is None:
|
|
if not existing:
|
|
findings.append({"severity": "error", "code": "MANIFEST_OUTPUT_MISSING", "detail": f"candidates={','.join(candidates)}"})
|
|
report = {"manifest": str(manifest_path), "media": None, "resolutionCandidates": candidates,
|
|
"qa": qa, "probe": None, "deepProbes": None, "findings": findings}
|
|
if opts.format == "json":
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
else:
|
|
for item in findings:
|
|
print(f"{item['severity'].upper()} {item['code']} {item['detail']}")
|
|
return 2
|
|
ffprobe = shutil.which("ffprobe")
|
|
if not ffprobe:
|
|
print("error: ffprobe is not available on PATH", file=sys.stderr)
|
|
return 3
|
|
result = run([ffprobe, "-v", "error", "-show_streams", "-show_format", "-of", "json", str(media)])
|
|
if result.returncode != 0:
|
|
print(f"error: ffprobe exited {result.returncode}: {result.stderr.strip()}", file=sys.stderr)
|
|
return 3
|
|
try:
|
|
probe = json.loads(result.stdout)
|
|
duration = float(probe.get("format", {}).get("duration"))
|
|
except (TypeError, ValueError, json.JSONDecodeError) as exc:
|
|
print(f"error: invalid ffprobe result: {exc}", file=sys.stderr)
|
|
return 3
|
|
expected = manifest.get("durationSeconds")
|
|
if isinstance(expected, bool) or not isinstance(expected, (int, float)) or float(expected) < 0:
|
|
findings.append({"severity": "error", "code": "MANIFEST_DURATION_INVALID", "detail": f"durationSeconds={expected!r}"})
|
|
elif abs(float(expected) - duration) > opts.duration_tolerance:
|
|
findings.append({"severity": "error", "code": "DURATION_MISMATCH", "detail": f"manifest={expected} probe={duration:.6f} tolerance={opts.duration_tolerance}"})
|
|
streams = probe.get("streams", [])
|
|
video_streams = [s for s in streams if s.get("codec_type") == "video"]
|
|
audio_streams = [s for s in streams if s.get("codec_type") == "audio"]
|
|
if not video_streams:
|
|
findings.append({"severity": "error", "code": "VIDEO_STREAM_MISSING", "detail": "ffprobe found no video stream"})
|
|
if not audio_streams:
|
|
findings.append({"severity": "warning", "code": "AUDIO_STREAM_MISSING", "detail": "ffprobe found no audio stream"})
|
|
deep = None
|
|
if opts.deep:
|
|
ffmpeg = shutil.which("ffmpeg")
|
|
if not ffmpeg:
|
|
print("error: ffmpeg is not available on PATH", file=sys.stderr)
|
|
return 3
|
|
deep, deep_findings = deep_probes(ffmpeg, media)
|
|
findings.extend(deep_findings)
|
|
summary = {
|
|
"durationSeconds": duration,
|
|
"sizeBytes": int(probe.get("format", {}).get("size", media.stat().st_size)),
|
|
"video": [{k: s.get(k) for k in ("codec_name", "width", "height", "pix_fmt", "avg_frame_rate")} for s in video_streams],
|
|
"audio": [{k: s.get(k) for k in ("codec_name", "sample_rate", "channels", "channel_layout")} for s in audio_streams],
|
|
}
|
|
report = {"manifest": str(manifest_path), "media": str(media), "resolutionCandidates": candidates,
|
|
"qa": qa, "probe": summary, "deepProbes": deep, "findings": findings}
|
|
if opts.format == "json":
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
else:
|
|
print(f"media={media}")
|
|
print(f"duration_seconds={duration:.6f} video_streams={len(video_streams)} audio_streams={len(audio_streams)} qa_mode={qa['mode']}")
|
|
for item in findings:
|
|
print(f"{item['severity'].upper()} {item['code']} {item['detail']}")
|
|
if not findings:
|
|
print("OK manifest and measured media properties agree")
|
|
return 2 if any(item["severity"] == "error" for item in findings) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|