#!/usr/bin/env python3 """Overlay b-roll onto a finished base in one pass, with motion, fades and cover fit. This is the step between "the talking head is cut" and "the reel is finished". It does not re-cut anything: the base keeps its own time map, and each b-roll is placed at an absolute moment on that map, so caption timings stay valid. Decisions worth stating: * **Fades, not hard cuts, by default.** A b-roll that appears and vanishes on a frame boundary reads as a glitch on a talking head, because the speaker's voice continues underneath. A 4-6 frame fade reads as an intentional cut-away. * **Slow motion on every insert.** A static stock clip inside a moving edit looks frozen. Each insert gets a slight eased zoom, the same principle as `animate_stills.py`. * **B-roll audio is dropped.** The base audio is the voice and continues under the insert; stock audio would fight it. Interface or ambience sound is a separate decision, made on the SFX bus. * **Coverage is reported and capped.** Past roughly half the runtime the speaker stops being the subject. The report says how much of the video the inserts cover and warns before that becomes a problem. """ 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 probe # noqa: E402 from render_core import PROFILES, capabilities, output_arguments # noqa: E402 from skill_config import canvas as resolve_canvas, emit, require_binary, utf8_stdout # noqa: E402 MAX_INSERTS = 40 COVERAGE_WARNING = 0.5 MOTIONS = { "none": None, "in": (1.00, 1.06), "in_strong": (1.00, 1.12), "out": (1.06, 1.00), "drift": (1.02, 1.05), } def ff(value: float) -> str: return f"{value:.6f}".rstrip("0").rstrip(".") or "0" def motion_chain(kind: str, frames: int, width: int, height: int) -> str: """Eased zoom across `frames`, implemented with an expression-driven crop. `scale` then `crop` with a smoothstep position: `zoompan` would be shorter but it interpolates linearly, so the move starts and stops with a visible jerk. """ spec = MOTIONS.get(kind) if not spec: return "" start, end = spec last = max(1, frames - 1) ease = f"((n/{last})*(n/{last})*(3-2*(n/{last})))" zoom = f"({start:.4f}+({end - start:.4f})*{ease})" # Scale up by the maximum zoom first so the crop always has pixels to take. upscale = max(start, end) big_width = int(round(width * upscale / 2) * 2) big_height = int(round(height * upscale / 2) * 2) crop_width = f"({width}*{upscale:.4f}/{zoom})" crop_height = f"({height}*{upscale:.4f}/{zoom})" return ( f",scale={big_width}:{big_height}:flags=lanczos" f",crop=w='{crop_width}':h='{crop_height}'" f":x='({big_width}-{crop_width})/2':y='({big_height}-{crop_height})/2'" f",scale={width}:{height}:flags=lanczos" ) def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("base", help="Готовая база: речь уже нарезана и сгрейжена") parser.add_argument("plan", help="broll_plan.json со вставками в ВЫХОДНОМ времени базы") parser.add_argument("output") parser.add_argument("--canvas", default="reels") parser.add_argument("--fade", type=float, default=0.18, help="Длительность появления и исчезновения вставки") parser.add_argument("--motion", default="in", choices=list(MOTIONS), help="Движение по умолчанию для вставок без своего") parser.add_argument("--grade", help="Фильтр, применяемый к каждой вставке") parser.add_argument("--profile", default="delivery", choices=list(PROFILES)) parser.add_argument("--codec", default="h264", choices=["h264", "hevc"]) parser.add_argument("--cpu", action="store_true") parser.add_argument("--filter-report") parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() require_binary("ffmpeg") base = Path(args.base).resolve() if not base.is_file(): raise SystemExit(f"Not found: {base}") base_info = probe(base) duration = float((base_info.get("video") or {}).get("duration") or base_info.get("duration") or 0.0) board = resolve_canvas(args.canvas) payload = json.loads(Path(args.plan).read_text(encoding="utf-8")) entries = payload.get("broll") if isinstance(payload, dict) else payload if not entries: raise SystemExit("В плане нет вставок (ключ 'broll')") if len(entries) > MAX_INSERTS: raise SystemExit(f"{len(entries)} вставок больше, чем {MAX_INSERTS} в один проход") inserts = [] problems: list[str] = [] for index, entry in enumerate(sorted(entries, key=lambda item: float(item["start"]))): path = Path(entry["path"]).resolve() if not path.is_file(): raise SystemExit(f"Вставка {index}: файл не найден — {path}") info = probe(path) available = float(info.get("duration") or 0.0) start = max(0.0, float(entry["start"])) length = float(entry.get("duration", 2.6)) source_in = float(entry.get("source_in", 0.0)) if start >= duration: problems.append(f"Вставка {index} начинается за концом базы, пропущена") continue length = min(length, duration - start) if source_in + length > available: trimmed = max(0.0, available - source_in) if trimmed < 0.4: problems.append( f"Вставка {index}: в источнике осталось {trimmed:.2f}s после " f"source_in={source_in} — пропущена" ) continue problems.append( f"Вставка {index}: укорочена с {length:.2f}s до {trimmed:.2f}s, " f"столько есть в источнике" ) length = trimmed inserts.append({ "index": len(inserts), "path": path, "start": round(start, 4), "end": round(start + length, 4), "duration": round(length, 4), "source_in": round(source_in, 4), "motion": entry.get("motion", args.motion), "note": entry.get("note"), "anchor": entry.get("anchor"), "fade": float(entry.get("fade", args.fade)), }) # Overlapping inserts would both be visible; the later one wins silently. for first, second in zip(inserts, inserts[1:]): if second["start"] < first["end"] - 0.01: problems.append( f"Вставки {first['index']} и {second['index']} перекрываются " f"({first['end']:.2f} > {second['start']:.2f}) — верхняя закроет нижнюю" ) covered = sum(item["duration"] for item in inserts) coverage = covered / duration if duration else 0.0 if coverage > COVERAGE_WARNING: problems.append( f"Вставки занимают {coverage:.0%} ролика: спикер перестаёт быть героем. " f"Держи ниже {COVERAGE_WARNING:.0%}" ) # A cut-away in the first seconds hides the establishing shot. early = [item for item in inserts if item["start"] < 3.0] if early: problems.append( f"{len(early)} вставок в первые 3 секунды: зритель ещё не понял, кто говорит" ) report = { "base": str(base), "base_duration": round(duration, 3), "canvas": f"{board.width}x{board.height}@{board.fps}", "inserts": [ {key: (str(value) if isinstance(value, Path) else value) for key, value in item.items()} for item in inserts ], "coverage": round(coverage, 4), "covered_seconds": round(covered, 2), "problems": problems, } if args.dry_run: emit(report) return command = ["ffmpeg", "-y", "-v", "warning", "-nostdin", "-i", str(base)] for item in inserts: command += ["-ss", ff(item["source_in"]), "-t", ff(item["duration"] + 0.2), "-i", str(item["path"])] graph: list[str] = [] graph.append(f"[0:v]fps={board.fps},setpts=PTS-STARTPTS[base]") previous = "base" for item in inserts: stream = item["index"] + 1 frames = max(2, int(round(item["duration"] * board.fps))) fade = max(0.0, min(item["fade"], item["duration"] / 2.5)) chain = ( f"[{stream}:v]setpts=PTS-STARTPTS,fps={board.fps}," f"scale={board.width}:{board.height}:force_original_aspect_ratio=increase" f":flags=lanczos,crop={board.width}:{board.height},setsar=1" ) chain += motion_chain(item["motion"], frames, board.width, board.height) if args.grade: chain += f",{args.grade}" chain += ",format=yuva420p" if fade > 0.02: # Fade the *alpha*, so the base shows through instead of the insert # fading to black and punching a hole in the picture. chain += ( f",fade=t=in:st=0:d={ff(fade)}:alpha=1" f",fade=t=out:st={ff(max(0.0, item['duration'] - fade))}:d={ff(fade)}:alpha=1" ) chain += f",setpts=PTS+{ff(item['start'])}/TB[b{item['index']}]" graph.append(chain) out = f"ov{item['index']}" graph.append( f"[{previous}][b{item['index']}]overlay=0:0:" f"enable='between(t,{ff(item['start'])},{ff(item['end'])})':" f"eof_action=pass:format=auto[{out}]" ) previous = out graph.append(f"[{previous}]format=yuv420p[vout]") filter_graph = ";\n".join(graph) report_path = (Path(args.filter_report) if args.filter_report else Path(args.output).with_suffix(".broll_filter.txt")) report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(filter_graph + "\n", encoding="utf-8") caps = capabilities() encoder_arguments, selection = output_arguments( args.codec, args.profile, prefer_gpu=not args.cpu, caps=caps ) output = Path(args.output).resolve() output.parent.mkdir(parents=True, exist_ok=True) command += [ "-filter_complex_script", str(report_path), "-map", "[vout]", "-map", "0:a?", *encoder_arguments, "-c:a", "copy", "-t", ff(duration), str(output), ] completed = subprocess.run(command, capture_output=True, text=True, errors="replace") if completed.returncode: report.update({"status": "render_failed", "filter_report": str(report_path), "tail": (completed.stderr or "").strip()[-1600:]}) emit(report) raise SystemExit(completed.returncode) result = probe(output) drift = round(float(result.get("duration") or 0.0) - duration, 4) report.update({ "status": "ok" if abs(drift) <= 0.12 else "duration_mismatch", "output": str(output), "actual_duration": result.get("duration"), "duration_drift": drift, "encoder": selection, "filter_report": str(report_path), "audio": "база скопирована без перекодирования, звук вставок отброшен", }) emit(report) if abs(drift) > 0.12: raise SystemExit(2) if __name__ == "__main__": main()