"""Create reproducible correctness evidence for media transcription acceptance."""
|
|
from __future__ import annotations
|
|
import argparse
|
import hashlib
|
import json
|
import re
|
import subprocess
|
import sys
|
from datetime import datetime, timezone
|
from pathlib import Path
|
from typing import Any, Sequence
|
|
|
def _fingerprint(path: Path) -> dict[str, Any]:
|
stat = path.stat()
|
digest = hashlib.sha256()
|
with path.open("rb") as stream:
|
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
digest.update(block)
|
return {
|
"path": str(path.resolve()),
|
"size_bytes": stat.st_size,
|
"creation_time_utc": datetime.fromtimestamp(
|
stat.st_ctime, timezone.utc
|
).isoformat(),
|
"last_write_time_utc": datetime.fromtimestamp(
|
stat.st_mtime, timezone.utc
|
).isoformat(),
|
"sha256": digest.hexdigest().upper(),
|
}
|
|
|
def _write_json(path: Path, payload: dict[str, Any], replace: bool = False) -> None:
|
if path.exists() and not replace:
|
raise RuntimeError(f"证据文件已存在，拒绝覆盖：{path}")
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(
|
json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
)
|
|
|
def _run_json(command: Sequence[str]) -> dict[str, Any]:
|
result = subprocess.run(
|
list(command),
|
check=False,
|
capture_output=True,
|
text=True,
|
encoding="utf-8",
|
errors="replace",
|
)
|
if result.returncode != 0:
|
raise RuntimeError(
|
f"命令失败（exit={result.returncode}）：{' '.join(command)}\n"
|
f"{(result.stderr or result.stdout).strip()}"
|
)
|
return json.loads(result.stdout)
|
|
|
def _probe_media(path: Path, *, audio_only: bool = False) -> dict[str, Any]:
|
command = [
|
"ffprobe",
|
"-v",
|
"error",
|
"-show_entries",
|
"format=duration:stream=codec_type,codec_name,sample_rate,channels,duration",
|
"-of",
|
"json",
|
]
|
if audio_only:
|
command.extend(["-select_streams", "a:0"])
|
command.append(str(path))
|
return _run_json(command)
|
|
|
def _duration(probe: dict[str, Any]) -> float:
|
format_duration = probe.get("format", {}).get("duration")
|
if format_duration is not None:
|
return float(format_duration)
|
for stream in probe.get("streams", []):
|
if stream.get("duration") is not None:
|
return float(stream["duration"])
|
raise RuntimeError("FFprobe 证据中没有可用时长。")
|
|
|
def _parse_srt_timestamp(value: str) -> float:
|
match = re.fullmatch(r"(\d+):(\d{2}):(\d{2}),(\d{3})", value)
|
if not match:
|
raise RuntimeError(f"非法 SRT 时间戳：{value}")
|
hours, minutes, seconds, milliseconds = (int(item) for item in match.groups())
|
return hours * 3600 + minutes * 60 + seconds + milliseconds / 1000.0
|
|
|
def _parse_srt(path: Path) -> list[dict[str, Any]]:
|
content = path.read_text(encoding="utf-8").strip()
|
if not content:
|
return []
|
blocks = re.split(r"\r?\n\s*\r?\n", content)
|
parsed: list[dict[str, Any]] = []
|
for block in blocks:
|
lines = block.splitlines()
|
if len(lines) < 3:
|
raise RuntimeError(f"SRT 块结构不完整：{block!r}")
|
timing = lines[1].split(" --> ")
|
if len(timing) != 2:
|
raise RuntimeError(f"SRT 时间行非法：{lines[1]}")
|
parsed.append(
|
{
|
"index": int(lines[0]),
|
"start": _parse_srt_timestamp(timing[0]),
|
"end": _parse_srt_timestamp(timing[1]),
|
"text": "\n".join(lines[2:]).strip(),
|
}
|
)
|
return parsed
|
|
|
def _parse_txt(path: Path) -> list[str]:
|
texts: list[str] = []
|
for line in path.read_text(encoding="utf-8").splitlines():
|
if not line.strip():
|
continue
|
match = re.fullmatch(r"\[\d+:\d{2}:\d{2}\]\s(.*)", line)
|
if not match:
|
raise RuntimeError(f"TXT 行结构非法：{line!r}")
|
texts.append(match.group(1))
|
return texts
|
|
|
def _segment_checks(
|
segments: list[dict[str, Any]],
|
srt: list[dict[str, Any]],
|
txt: list[str],
|
flac_duration: float,
|
) -> dict[str, Any]:
|
legal_ranges = all(
|
0.0 <= float(item["start"]) <= float(item["end"]) <= flac_duration
|
for item in segments
|
)
|
keys = [
|
(float(item["start"]), float(item["end"]), str(item["text"]))
|
for item in segments
|
]
|
ordered = keys == sorted(keys)
|
indices = [item["index"] for item in srt]
|
continuous_indices = indices == list(range(1, len(srt) + 1))
|
counts_match = len(segments) == len(srt) == len(txt)
|
text_match = counts_match and all(
|
str(segment["text"]) == srt_item["text"] == txt_item
|
for segment, srt_item, txt_item in zip(segments, srt, txt, strict=True)
|
)
|
srt_time_match = counts_match and all(
|
abs(float(segment["start"]) - srt_item["start"]) <= 0.00051
|
and abs(float(segment["end"]) - srt_item["end"]) <= 0.00051
|
for segment, srt_item in zip(segments, srt, strict=True)
|
)
|
duplicate_rows: list[int] = []
|
seen: set[tuple[float, float, str]] = set()
|
for index, key in enumerate(keys, start=1):
|
if key in seen:
|
duplicate_rows.append(index)
|
seen.add(key)
|
return {
|
"segment_count": len(segments),
|
"srt_count": len(srt),
|
"txt_count": len(txt),
|
"legal_time_ranges": legal_ranges,
|
"sorted_by_start_end_text": ordered,
|
"continuous_srt_indices": continuous_indices,
|
"counts_match": counts_match,
|
"texts_match": text_match,
|
"srt_times_match_json": srt_time_match,
|
"exact_duplicate_segment_indices": duplicate_rows,
|
"pass": all(
|
[
|
legal_ranges,
|
ordered,
|
continuous_indices,
|
counts_match,
|
text_match,
|
srt_time_match,
|
not duplicate_rows,
|
]
|
),
|
}
|
|
|
def _boundary_checks(
|
segments: list[dict[str, Any]],
|
duration: float,
|
manual_pass: bool,
|
reviewer: str | None,
|
) -> list[dict[str, Any]]:
|
boundaries: list[dict[str, Any]] = []
|
boundary = 1200
|
while boundary < duration:
|
rows = []
|
lower = boundary - 15.0
|
upper = boundary + 15.0
|
for index, segment in enumerate(segments, start=1):
|
start = float(segment["start"])
|
end = float(segment["end"])
|
if end >= lower and start <= upper:
|
rows.append(
|
{
|
"index": index,
|
"start": start,
|
"end": end,
|
"text": str(segment["text"]),
|
}
|
)
|
rows_are_ordered = rows == sorted(
|
rows, key=lambda item: (item["start"], item["end"], item["text"])
|
)
|
keys = [(row["start"], row["end"], row["text"]) for row in rows]
|
boundaries.append(
|
{
|
"boundary_seconds": boundary,
|
"window_start": lower,
|
"window_end": upper,
|
"segments": rows,
|
"automatic_no_exact_duplicates": len(keys) == len(set(keys)),
|
"automatic_ordered": rows_are_ordered,
|
"manual_readability_pass": manual_pass,
|
"manual_reviewer": reviewer if manual_pass else None,
|
"manual_reviewed_at_utc": (
|
datetime.now(timezone.utc).isoformat() if manual_pass else None
|
),
|
}
|
)
|
boundary += 1200
|
return boundaries
|
|
|
def _snapshot(args: argparse.Namespace) -> int:
|
source = Path(args.source).resolve()
|
_write_json(Path(args.output).resolve(), _fingerprint(source), args.replace)
|
return 0
|
|
|
def _validate(args: argparse.Namespace) -> int:
|
original_source = Path(args.original_source).resolve()
|
media = Path(args.media).resolve()
|
output_dir = Path(args.output_dir).resolve()
|
evidence_path = Path(args.evidence).resolve()
|
before = json.loads(Path(args.source_before).read_text(encoding="utf-8-sig"))
|
after = _fingerprint(original_source)
|
source_fields = (
|
"size_bytes",
|
"creation_time_utc",
|
"last_write_time_utc",
|
"sha256",
|
)
|
source_unchanged = all(before.get(field) == after.get(field) for field in source_fields)
|
|
stem = media.stem
|
paths = {
|
"flac": output_dir / f"{stem}.audio.flac",
|
"txt": output_dir / f"{stem}.txt",
|
"srt": output_dir / f"{stem}.srt",
|
"json": output_dir / f"{stem}.json",
|
}
|
missing = [str(path) for path in paths.values() if not path.is_file()]
|
failures: list[str] = []
|
if missing:
|
failures.append("missing outputs: " + ", ".join(missing))
|
|
result_payload: dict[str, Any] = {}
|
media_probe: dict[str, Any] = {}
|
flac_probe: dict[str, Any] = {}
|
segment_evidence: dict[str, Any] = {"pass": False}
|
boundaries: list[dict[str, Any]] = []
|
flac_duration = 0.0
|
media_duration = 0.0
|
|
if not missing:
|
try:
|
result_payload = json.loads(paths["json"].read_text(encoding="utf-8"))
|
media_probe = _probe_media(media)
|
flac_probe = _probe_media(paths["flac"], audio_only=True)
|
media_duration = _duration(media_probe)
|
flac_duration = _duration(flac_probe)
|
srt = _parse_srt(paths["srt"])
|
txt = _parse_txt(paths["txt"])
|
segments = result_payload.get("segments", [])
|
segment_evidence = _segment_checks(
|
segments, srt, txt, flac_duration
|
)
|
boundaries = _boundary_checks(
|
segments,
|
flac_duration,
|
args.manual_boundaries_pass,
|
args.reviewer,
|
)
|
except Exception as exc:
|
failures.append(str(exc))
|
|
streams = flac_probe.get("streams", [])
|
audio_stream = streams[0] if streams else {}
|
flac_contract = {
|
"codec_name": audio_stream.get("codec_name"),
|
"sample_rate": audio_stream.get("sample_rate"),
|
"channels": audio_stream.get("channels"),
|
"duration_seconds": flac_duration,
|
"media_duration_seconds": media_duration,
|
"duration_delta_seconds": abs(flac_duration - media_duration),
|
}
|
flac_contract["pass"] = (
|
flac_contract["codec_name"] == "flac"
|
and str(flac_contract["sample_rate"]) == "16000"
|
and flac_contract["channels"] == 1
|
and flac_contract["duration_delta_seconds"] <= 1.0
|
)
|
|
metadata_contract = {
|
"model": result_payload.get("model"),
|
"device": result_payload.get("device"),
|
"compute_type": result_payload.get("compute_type"),
|
"vad_filter": result_payload.get("vad_filter"),
|
}
|
metadata_contract["pass"] = metadata_contract == {
|
"model": "large-v3",
|
"device": "cuda",
|
"compute_type": "float16",
|
"vad_filter": True,
|
}
|
|
expected_segments_pass = (
|
args.expected_segments is None
|
or segment_evidence.get("segment_count") == args.expected_segments
|
)
|
boundary_automatic_pass = all(
|
item["automatic_no_exact_duplicates"] and item["automatic_ordered"]
|
for item in boundaries
|
)
|
boundary_manual_pass = not boundaries or all(
|
item["manual_readability_pass"] for item in boundaries
|
)
|
|
resource_summary = None
|
resource_pass = True
|
if args.resource_summary:
|
resource_summary = json.loads(
|
Path(args.resource_summary).read_text(encoding="utf-8-sig")
|
)
|
resource_pass = bool(resource_summary.get("resource_gates_pass"))
|
|
automatic_pass = all(
|
[
|
not failures,
|
source_unchanged,
|
flac_contract["pass"],
|
metadata_contract["pass"],
|
bool(segment_evidence.get("pass")),
|
expected_segments_pass,
|
boundary_automatic_pass,
|
resource_pass,
|
]
|
)
|
status = "PASS" if automatic_pass and boundary_manual_pass else (
|
"PENDING_MANUAL_BOUNDARY_REVIEW"
|
if automatic_pass and boundaries
|
else "FAIL"
|
)
|
payload = {
|
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
|
"status": status,
|
"failures": failures,
|
"source": {
|
"before": before,
|
"after": after,
|
"compared_fields": list(source_fields),
|
"unchanged": source_unchanged,
|
},
|
"media": {
|
"path": str(media),
|
"fingerprint": _fingerprint(media),
|
"probe": media_probe,
|
},
|
"outputs": {
|
name: {"path": str(path), "size_bytes": path.stat().st_size}
|
for name, path in paths.items()
|
if path.is_file()
|
},
|
"flac_contract": flac_contract,
|
"metadata_contract": metadata_contract,
|
"segments": segment_evidence,
|
"expected_segments": args.expected_segments,
|
"expected_segments_pass": expected_segments_pass,
|
"boundaries": boundaries,
|
"boundary_automatic_pass": boundary_automatic_pass,
|
"boundary_manual_pass": boundary_manual_pass,
|
"resource_summary": resource_summary,
|
"automatic_pass": automatic_pass,
|
}
|
_write_json(evidence_path, payload, args.replace)
|
return 0 if status == "PASS" else (2 if status == "PENDING_MANUAL_BOUNDARY_REVIEW" else 1)
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(description=__doc__)
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
snapshot = subparsers.add_parser("snapshot", help="记录源文件只读指纹。")
|
snapshot.add_argument("--source", required=True)
|
snapshot.add_argument("--output", required=True)
|
snapshot.add_argument("--replace", action="store_true")
|
snapshot.set_defaults(func=_snapshot)
|
|
validate = subparsers.add_parser("validate", help="生成转写正确性验收证据。")
|
validate.add_argument("--original-source", required=True)
|
validate.add_argument("--source-before", required=True)
|
validate.add_argument("--media", required=True)
|
validate.add_argument("--output-dir", required=True)
|
validate.add_argument("--evidence", required=True)
|
validate.add_argument("--resource-summary")
|
validate.add_argument("--expected-segments", type=int)
|
validate.add_argument("--manual-boundaries-pass", action="store_true")
|
validate.add_argument("--reviewer")
|
validate.add_argument("--replace", action="store_true")
|
validate.set_defaults(func=_validate)
|
return parser
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
args = _build_parser().parse_args(argv)
|
try:
|
return int(args.func(args))
|
except Exception as exc:
|
print(f"验收证据生成失败：{exc}", file=sys.stderr)
|
return 1
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
