#!/usr/bin/env python3 """Проверка готового файла перед сдачей: громкость, геометрия, хвост, шрифты. Проверяет то, что ломается уже после успешного рендера и поэтому не ловится предыдущими воротами: * громкость против цели платформы — отдавать громче бессмысленно, платформа приглушит и микс потеряет динамику; * геометрия и частота кадров совпадают с заявленным канвасом; * последний кадр декодируется, последний аудиопакет на месте, речь не обрезана; * поток не HDR и не с неквадратными пикселями — и то и другое ломает превью; * если передан caption_map, титры остались внутри безопасной зоны платформы. """ from __future__ import annotations import argparse import json import subprocess import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from media_probe import loudness, probe # noqa: E402 from skill_config import LOUDNESS, canvas as resolve_canvas, emit, utf8_stdout # noqa: E402 def last_frame_decodes(video: Path, duration: float) -> tuple[bool, str]: """Декодирует последнюю секунду: обрезанный хвост виден только так.""" start = max(0.0, duration - 1.0) completed = subprocess.run( ["ffmpeg", "-v", "error", "-nostdin", "-ss", f"{start:.3f}", "-i", str(video), "-map", "0:v:0", "-f", "null", "-"], capture_output=True, text=True, errors="replace", ) return completed.returncode == 0, (completed.stderr or "").strip()[-300:] def last_audio_packet(video: Path, duration: float) -> tuple[bool, str]: start = max(0.0, duration - 1.0) completed = subprocess.run( ["ffmpeg", "-v", "error", "-nostdin", "-ss", f"{start:.3f}", "-i", str(video), "-map", "0:a:0", "-f", "null", "-"], capture_output=True, text=True, errors="replace", ) return completed.returncode == 0, (completed.stderr or "").strip()[-300:] def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("video") parser.add_argument("--canvas", default="reels") parser.add_argument("--loudness", default="reels", choices=list(LOUDNESS)) parser.add_argument("--loudness-tolerance", type=float, default=1.0) parser.add_argument("--timeline", help="Проверить длительность против ожидаемой") parser.add_argument("--duration-tolerance", type=float, default=0.12) parser.add_argument("--caption-map", help="Проверить титры на попадание в безопасную зону") parser.add_argument("--report", default="delivery_report.json") args = parser.parse_args() video = Path(args.video).resolve() if not video.is_file(): raise SystemExit(f"Not found: {video}") info = probe(video) board = resolve_canvas(args.canvas) target = LOUDNESS[args.loudness] duration = float(info.get("duration") or 0.0) video_stream = info.get("video") or {} errors: list[str] = [] warnings: list[str] = [] if not info.get("has_video"): errors.append("В файле нет видеопотока") if not info.get("has_audio"): errors.append("В файле нет аудиопотока") width = video_stream.get("display_width") height = video_stream.get("display_height") if width and height and (width, height) != (board.width, board.height): errors.append( f"Геометрия {width}x{height} не совпадает с канвасом " f"{board.width}x{board.height}" ) rate = video_stream.get("avg_frame_rate") if rate and abs(rate - board.fps) > 0.6: warnings.append(f"Частота кадров {rate} против {board.fps} у канваса") if video_stream.get("non_square_pixels"): errors.append( f"Неквадратные пиксели ({video_stream.get('sample_aspect_ratio')}) — " f"платформы покажут кадр растянутым" ) if video_stream.get("hdr"): errors.append("Поток помечен как HDR: превью и большинство плееров покажут его тускло") if video_stream.get("variable_frame_rate"): warnings.append("Переменная частота кадров в готовом файле — приведи к постоянной") measured = loudness(video) if info.get("has_audio") else {"measured": False} integrated = measured.get("integrated_lufs") if integrated is None: warnings.append("Не удалось измерить громкость") else: difference = integrated - target["i"] if abs(difference) > args.loudness_tolerance: errors.append( f"Громкость {integrated} LUFS против цели {target['i']} " f"(расхождение {difference:+.1f} LU)" ) peak = measured.get("true_peak_dbtp") if peak is not None and peak > target["tp"] + 0.5: errors.append(f"Истинный пик {peak} dBTP выше допустимого {target['tp']}") video_ok, video_tail = last_frame_decodes(video, duration) if not video_ok: errors.append(f"Последняя секунда видео не декодируется: {video_tail}") audio_ok, audio_tail = (True, "") if info.get("has_audio"): audio_ok, audio_tail = last_audio_packet(video, duration) if not audio_ok: errors.append(f"Последняя секунда звука не декодируется: {audio_tail}") expected_duration = None if args.timeline: timeline = json.loads(Path(args.timeline).read_text(encoding="utf-8")) expected_duration = float(timeline.get("expected_duration") or 0.0) if expected_duration: drift = duration - expected_duration if abs(drift) > args.duration_tolerance: errors.append( f"Длительность {duration:.3f} против ожидаемой " f"{expected_duration:.3f} (расхождение {drift:+.3f} с)" ) last_word_end = timeline.get("last_word_end") if last_word_end and duration < float(last_word_end) - 0.02: errors.append( f"Файл короче последнего слова: {duration:.3f} < {last_word_end}. " f"Последнее слово обрезано." ) caption_check = None if args.caption_map: caption_map = json.loads(Path(args.caption_map).read_text(encoding="utf-8")) safe = board.safe_box outside = [] for block in caption_map.get("blocks", []): for word in block.get("words", []): y = word.get("y") if y is None: continue if y < safe["top"] or y > safe["bottom"]: outside.append({"block": block["id"], "word": word.get("word"), "y": y}) overflow = [block["id"] for block in caption_map.get("blocks", []) if block.get("overflow")] caption_check = { "blocks": len(caption_map.get("blocks", [])), "preset": caption_map.get("preset"), "font_resolved": caption_map.get("font_resolved"), "safe_box": safe, "outside_safe_area": outside[:10], "overflow_blocks": overflow, } if outside: errors.append( f"{len(outside)} слов(а) вне безопасной зоны платформы — " f"их закроет интерфейс" ) if overflow: errors.append(f"Блоки не влезли в кадр: {overflow[:8]}") report = { "status": "ok" if not errors else "failed", "video": str(video), "canvas_expected": f"{board.width}x{board.height}@{board.fps}", "geometry": f"{width}x{height}@{rate}" if width else None, "duration": duration, "expected_duration": expected_duration, "loudness_measured": measured, "loudness_target": target, "tail_video_decodes": video_ok, "tail_audio_decodes": audio_ok, "codec": video_stream.get("codec"), "pix_fmt": video_stream.get("pix_fmt"), "captions": caption_check, "errors": errors, "warnings": warnings, } Path(args.report).parent.mkdir(parents=True, exist_ok=True) Path(args.report).write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") emit(report) if errors: raise SystemExit(2) if __name__ == "__main__": main()