#!/usr/bin/env python3 """Пакетная обработка всех MP4 из маршруты/медиа: 320x180, 15 к/с, длительность без изменений. Внешняя программа кодирования: ffmpeg/libx264. Исходные файлы только читаются. Для каждого файла в журнал пишутся входной и выходной путь, параметры кодирования, фактические параметры результата и статус проверки. """ from __future__ import annotations import hashlib import json import shlex import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent SRC_DIR = ROOT / "маршруты" / "медиа" OUT_DIR = ROOT / "результаты" LOG = ROOT / "сборка" / "видео-масштаб-журнал.json" WIDTH, HEIGHT, FPS = 320, 180, 15 SUFFIX = f"-{WIDTH}x{HEIGHT}-{FPS}fps" CRF = 18 PRESET = "medium" def run(cmd: list[str]) -> subprocess.CompletedProcess: return subprocess.run(cmd, capture_output=True, text=True) def ffmpeg_version() -> str: return run(["ffmpeg", "-version"]).stdout.splitlines()[0] def sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def probe(path: Path) -> dict: proc = run(["ffprobe", "-v", "error", "-print_format", "json", "-show_format", "-show_streams", str(path)]) if proc.returncode != 0: raise RuntimeError(proc.stderr.strip()) return json.loads(proc.stdout) def describe(path: Path) -> dict: """Фактические параметры файла: разрешение, к/с, длительность, кадры, дорожки.""" data = probe(path) video = [s for s in data["streams"] if s.get("codec_type") == "video"] audio = [s for s in data["streams"] if s.get("codec_type") == "audio"] v = video[0] if video else {} nb = v.get("nb_frames") return { "длительность_с": round(float(data["format"]["duration"]), 6), "размер_байт": int(data["format"]["size"]), "разрешение": f"{v.get('width')}x{v.get('height')}" if v else None, "частота_кадров": v.get("avg_frame_rate"), "кадров": int(nb) if nb not in (None, "N/A") else None, "видеодорожек": len(video), "аудиодорожек": len(audio), "кодек": v.get("codec_name"), "контейнер": data["format"].get("format_name"), } def build_cmd(src: Path, out: Path) -> list[str]: """Привести к 320x180 и 15 к/с; PTS не трогаются, поэтому длительность сохраняется.""" vf = f"scale={WIDTH}:{HEIGHT}:flags=lanczos,setsar=1,fps={FPS}" return [ "ffmpeg", "-y", "-hide_banner", "-v", "error", "-i", str(src), "-vf", vf, "-c:v", "libx264", "-preset", PRESET, "-crf", str(CRF), "-pix_fmt", "yuv420p", "-movflags", "+faststart", "-an", str(out), ] def check(before: dict, after: dict) -> list[dict]: """Сверить результат с ожидаемыми параметрами; допуск длительности — один кадр.""" tolerance = 1.0 / FPS rows = [ ("разрешение", after["разрешение"], f"{WIDTH}x{HEIGHT}", after["разрешение"] == f"{WIDTH}x{HEIGHT}"), ("частота кадров", after["частота_кадров"], f"{FPS}/1", after["частота_кадров"] == f"{FPS}/1"), ("длительность, с", after["длительность_с"], f"{before['длительность_с']} ± {tolerance:.4f}", abs(after["длительность_с"] - before["длительность_с"]) <= tolerance + 1e-9), ("кадров", after["кадров"], round(before["длительность_с"] * FPS), after["кадров"] is not None and abs(after["кадров"] - before["длительность_с"] * FPS) <= 1), ("видеодорожек", after["видеодорожек"], before["видеодорожек"], after["видеодорожек"] == before["видеодорожек"] == 1), ("аудиодорожек (в исходнике звука нет)", after["аудиодорожек"], before["аудиодорожек"], after["аудиодорожек"] == before["аудиодорожек"]), ("контейнер/кодек", f"{after['контейнер']}/{after['кодек']}", "mp4/h264", after["кодек"] == "h264" and "mp4" in (after["контейнер"] or "")), ] return [{"проверка": name, "факт": fact, "ожидание": exp, "итог": "ок" if ok else "ошибка"} for name, fact, exp, ok in rows] def main() -> int: sources = sorted(SRC_DIR.glob("*.mp4")) if not sources: print(f"нет MP4 в {SRC_DIR}", file=sys.stderr) return 1 OUT_DIR.mkdir(parents=True, exist_ok=True) records = [] failed = 0 for src in sources: out = OUT_DIR / f"{src.stem}{SUFFIX}.mp4" before = describe(src) src_hash = sha256(src) cmd = build_cmd(src, out) proc = run(cmd) if proc.returncode != 0 or not out.is_file(): failed += 1 records.append({ "вход": str(src.relative_to(ROOT)), "выход": str(out.relative_to(ROOT)), "параметры": {"разрешение": f"{WIDTH}x{HEIGHT}", "кадров_в_секунду": FPS, "кодек": "libx264", "preset": PRESET, "crf": CRF, "pix_fmt": "yuv420p", "звук": "не переносится"}, "исходник": before, "статус": "ошибка", "ошибка": proc.stderr.strip(), "команда": " ".join(shlex.quote(a) for a in cmd), }) continue after = describe(out) checks = check(before, after) ok = all(c["итог"] == "ок" for c in checks) and sha256(src) == src_hash if not ok: failed += 1 records.append({ "вход": str(src.relative_to(ROOT)), "выход": str(out.relative_to(ROOT)), "параметры": {"разрешение": f"{WIDTH}x{HEIGHT}", "кадров_в_секунду": FPS, "кодек": "libx264", "preset": PRESET, "crf": CRF, "pix_fmt": "yuv420p", "звук": "не переносится"}, "исходник": before, "результат": after, "статус": "ок" if ok else "ошибка", "проверки": checks, "исходник_не_изменён": sha256(src) == src_hash, "команда": " ".join(shlex.quote(a) for a in cmd), }) log = { "задача": f"все MP4 из маршруты/медиа → {WIDTH}x{HEIGHT}, {FPS} к/с, длительность сохраняется", "внешние_программы": { "ffmpeg": ffmpeg_version(), "кодек": "libx264 (H.264), pixel format yuv420p", }, "python": sys.version.split()[0], "всего_файлов": len(sources), "успешно": len(sources) - failed, "с_ошибкой": failed, "файлы": records, } LOG.write_text(json.dumps(log, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") for r in records: print(f"{r['статус']:>8} {r['вход']} → {r['выход']}") print(f"журнал: {LOG.relative_to(ROOT)}") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())