291 lines
11 KiB
Python
Executable File
291 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Fail closed until a downloaded video job reaches its required deliverable."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
from typing import Any, Sequence
|
|
|
|
|
|
class DeliveryError(RuntimeError):
|
|
"""A malformed or unreadable delivery job."""
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise DeliveryError(f"manifest not found: {path}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise DeliveryError(f"invalid JSON in {path}: {exc}") from exc
|
|
if not isinstance(value, dict):
|
|
raise DeliveryError(f"manifest root must be an object: {path}")
|
|
return value
|
|
|
|
|
|
def _artifact_path(job_dir: Path, value: Any) -> Path | None:
|
|
if not isinstance(value, dict) or not isinstance(value.get("path"), str):
|
|
return None
|
|
path = Path(value["path"])
|
|
return path if path.is_absolute() else job_dir / path
|
|
|
|
|
|
def _sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _verify_citation_watermark(job_dir: Path, download: dict[str, Any], burned: Path) -> None:
|
|
citation = download.get("citation_watermark")
|
|
if not isinstance(citation, dict) or citation.get("enabled") is not True:
|
|
return
|
|
citation_name = citation.get("citation_file")
|
|
citation_hash = citation.get("citation_sha256")
|
|
if not isinstance(citation_name, str) or Path(citation_name).name != citation_name:
|
|
raise DeliveryError("citation_watermark.citation_file must be a plain filename")
|
|
if not isinstance(citation_hash, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", citation_hash):
|
|
raise DeliveryError("citation_watermark.citation_sha256 is invalid")
|
|
citation_path = job_dir / citation_name
|
|
if not citation_path.is_file() or _sha256_file(citation_path) != citation_hash.lower():
|
|
raise DeliveryError("citation watermark text is missing or its checksum changed")
|
|
receipt_path = burned.with_suffix(burned.suffix + ".citation.json")
|
|
if not receipt_path.is_file():
|
|
raise DeliveryError(f"citation watermark burn receipt is missing: {receipt_path}")
|
|
receipt = _read_json(receipt_path)
|
|
expected = {
|
|
"output_file": burned.name,
|
|
"output_sha256": _sha256_file(burned),
|
|
"citation_sha256": citation_hash.lower(),
|
|
"position": "top-left",
|
|
}
|
|
for key, value in expected.items():
|
|
if receipt.get(key) != value:
|
|
raise DeliveryError(f"citation watermark receipt has a stale or invalid {key}")
|
|
|
|
|
|
def _existing_video_artifact(job_dir: Path, artifacts: dict[str, Any]) -> Path | None:
|
|
records = [artifacts.get("lossless_mp4_master"), artifacts.get("intermediate")]
|
|
fallback = artifacts.get("lossy_mp4_fallback")
|
|
if isinstance(fallback, dict):
|
|
records.append(fallback.get("created"))
|
|
for record in records:
|
|
path = _artifact_path(job_dir, record)
|
|
if path is not None and path.is_file() and path.stat().st_size:
|
|
return path
|
|
return None
|
|
|
|
|
|
DELIVERABLES = ("full", "video", "subs", "bilingual-subs")
|
|
|
|
|
|
def assess_delivery(download_manifest: Path) -> dict[str, Any]:
|
|
download_manifest = download_manifest.expanduser().resolve()
|
|
download = _read_json(download_manifest)
|
|
configured_dir = download.get("output_directory")
|
|
job_dir = (
|
|
Path(configured_dir).expanduser().resolve()
|
|
if isinstance(configured_dir, str)
|
|
else download_manifest.parent
|
|
)
|
|
deliverable = download.get("deliverable")
|
|
if deliverable not in DELIVERABLES:
|
|
deliverable = "full"
|
|
artifacts = download.get("artifacts")
|
|
if not isinstance(artifacts, dict):
|
|
raise DeliveryError("download manifest has no artifacts object")
|
|
if deliverable in ("full", "video") and _existing_video_artifact(job_dir, artifacts) is None:
|
|
raise DeliveryError("no declared video artifact exists on disk")
|
|
|
|
def complete(stage: str, **extra: Any) -> dict[str, Any]:
|
|
return {
|
|
"complete": True,
|
|
"stage": stage,
|
|
"deliverable": deliverable,
|
|
"job_dir": str(job_dir),
|
|
"missing": [],
|
|
**extra,
|
|
}
|
|
|
|
subtitle_record = artifacts.get("subtitle")
|
|
subtitle = (
|
|
_artifact_path(job_dir, subtitle_record.get("source_srt"))
|
|
if isinstance(subtitle_record, dict)
|
|
else None
|
|
)
|
|
if subtitle is not None and not subtitle.is_file():
|
|
raise DeliveryError(f"declared source subtitle is missing: {subtitle}")
|
|
if deliverable in ("subs", "bilingual-subs") and subtitle is None:
|
|
raise DeliveryError(
|
|
"a subtitle delivery was requested, but the manifest declares no source subtitle"
|
|
)
|
|
has_dialogue = (
|
|
subtitle is not None
|
|
and isinstance(subtitle_record, dict)
|
|
and subtitle_record.get("dialogue") is not False
|
|
)
|
|
|
|
if deliverable == "video":
|
|
return complete("video_complete")
|
|
if deliverable == "subs":
|
|
return complete("subs_complete")
|
|
if not has_dialogue:
|
|
# full falls back to plain video; bilingual-subs still delivered the
|
|
# source subtitle files even though nothing was translatable.
|
|
stage = "video_only_complete" if deliverable == "full" else "subs_complete"
|
|
return complete(stage)
|
|
|
|
subtitle_dir = job_dir / "subtitles"
|
|
subtitle_manifest_path = subtitle_dir / "subtitle-manifest.json"
|
|
if not subtitle_manifest_path.is_file():
|
|
return {
|
|
"complete": False,
|
|
"stage": "subtitle_prepare_required",
|
|
"job_dir": str(job_dir),
|
|
"missing": [str(subtitle_manifest_path)],
|
|
}
|
|
|
|
subtitle_manifest = _read_json(subtitle_manifest_path)
|
|
batches = subtitle_manifest.get("translation_batches")
|
|
if not isinstance(batches, list) or not batches:
|
|
raise DeliveryError("subtitle manifest has no translation batches")
|
|
output_dir_value = subtitle_manifest.get("translation_output_dir")
|
|
translation_output_dir = (
|
|
Path(output_dir_value)
|
|
if isinstance(output_dir_value, str)
|
|
else subtitle_dir / "translation-output"
|
|
)
|
|
missing_batches: list[str] = []
|
|
for batch in batches:
|
|
if not isinstance(batch, dict) or not isinstance(batch.get("path"), str):
|
|
raise DeliveryError("subtitle manifest has an invalid translation batch")
|
|
name = Path(batch["path"]).name
|
|
if not (translation_output_dir / name).is_file():
|
|
missing_batches.append(name)
|
|
if missing_batches:
|
|
return {
|
|
"complete": False,
|
|
"stage": "translation_required",
|
|
"job_dir": str(job_dir),
|
|
"missing": missing_batches,
|
|
}
|
|
|
|
execution = download.get("execution")
|
|
scientific_review_required = (
|
|
isinstance(execution, dict)
|
|
and execution.get("scientific_review_required") is True
|
|
)
|
|
scientific_review_report_path = subtitle_dir / "scientific-review" / "report.json"
|
|
scientific_review: dict[str, Any] | None = None
|
|
if scientific_review_required:
|
|
reviewed_translations = (
|
|
subtitle_dir
|
|
/ "scientific-review"
|
|
/ "reviewed-translations"
|
|
/ "translations.json"
|
|
)
|
|
missing_review = [
|
|
str(path)
|
|
for path in (scientific_review_report_path, reviewed_translations)
|
|
if not path.is_file()
|
|
]
|
|
if missing_review:
|
|
return {
|
|
"complete": False,
|
|
"stage": "scientific_review_required",
|
|
"job_dir": str(job_dir),
|
|
"missing": missing_review,
|
|
}
|
|
scientific_review = _read_json(scientific_review_report_path)
|
|
if scientific_review.get("status") != "complete":
|
|
raise DeliveryError("scientific review report is incomplete")
|
|
if scientific_review.get("human_expert_reviewed") is not False:
|
|
raise DeliveryError("scientific review report makes an invalid human-review claim")
|
|
if scientific_review.get("subtitle_manifest_sha256") != _sha256_file(
|
|
subtitle_manifest_path
|
|
):
|
|
raise DeliveryError("scientific review report is bound to a different subtitle manifest")
|
|
|
|
rendered_dir = subtitle_dir / "rendered"
|
|
required_rendered = [rendered_dir / "bilingual.ass", rendered_dir / "validation.json"]
|
|
missing_rendered = [str(path) for path in required_rendered if not path.is_file()]
|
|
if missing_rendered:
|
|
return {
|
|
"complete": False,
|
|
"stage": "render_required",
|
|
"job_dir": str(job_dir),
|
|
"missing": missing_rendered,
|
|
}
|
|
if scientific_review_required:
|
|
validation = _read_json(rendered_dir / "validation.json")
|
|
binding = validation.get("scientific_review")
|
|
if validation.get("translation_quality_reviewed") is not True or not isinstance(binding, dict):
|
|
raise DeliveryError("rendered subtitles did not use the scientific-review gate")
|
|
if binding.get("report_sha256") != _sha256_file(scientific_review_report_path):
|
|
raise DeliveryError("rendered subtitles use a stale scientific review report")
|
|
if deliverable == "bilingual-subs":
|
|
return complete("bilingual_subs_complete", rendered_dir=str(rendered_dir))
|
|
|
|
delivery_names = download.get("delivery_names")
|
|
burned_name = (
|
|
delivery_names.get("bilingual_video")
|
|
if isinstance(delivery_names, dict)
|
|
else None
|
|
)
|
|
if isinstance(burned_name, str) and burned_name:
|
|
if Path(burned_name).name != burned_name:
|
|
raise DeliveryError("delivery_names.bilingual_video must be a plain filename")
|
|
burned = job_dir / burned_name
|
|
if burned.is_file() and burned.stat().st_size:
|
|
_verify_citation_watermark(job_dir, download, burned)
|
|
return complete("bilingual_complete", burned_video=str(burned))
|
|
missing = [str(burned)]
|
|
else:
|
|
legacy = sorted(
|
|
path
|
|
for path in job_dir.glob("*.bilingual.mp4")
|
|
if path.is_file() and path.stat().st_size
|
|
)
|
|
if legacy:
|
|
_verify_citation_watermark(job_dir, download, legacy[-1])
|
|
return complete("bilingual_complete", burned_video=str(legacy[-1]))
|
|
missing = ["*.bilingual.mp4"]
|
|
if missing:
|
|
return {
|
|
"complete": False,
|
|
"stage": "burn_required",
|
|
"job_dir": str(job_dir),
|
|
"missing": missing,
|
|
}
|
|
raise AssertionError("unreachable")
|
|
|
|
|
|
def _parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Check whether a video job is video-only complete or bilingual complete."
|
|
)
|
|
parser.add_argument("download_manifest", type=Path)
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
args = _parser().parse_args(argv)
|
|
try:
|
|
result = assess_delivery(args.download_manifest)
|
|
except (DeliveryError, OSError) as exc:
|
|
print(f"delivery verification error: {exc}", file=sys.stderr)
|
|
return 2
|
|
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
|
|
return 0 if result["complete"] else 3
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|