#!/usr/bin/env python3 """Put the hook in front of the cut and keep one timeline. `plan_cold_open.py` decides *which* phrase opens. This renders it and — the part that actually matters — rewrites the word timeline so everything downstream still reads a single shared clock. Two things go wrong when a cold open is added by hand, and both are silent: * **the captions slide.** Every word in the body now happens later by the length of the hook, minus the overlap the transition eats. Burn captions from the old timeline and the whole file is off by a second and a half. * **the hook has no captions at all**, because its words were never added to the timeline — they exist only in the source transcript. Both are handled here: the hook's own words are emitted at their new positions and the body is shifted by exactly `hook_length − transition_length`. Movement in the first second is not decoration. A static opening frame is the most common reason a viewer leaves before the first sentence lands, so the hook is rendered with a measured punch-in: a scale ramp on an `expo.out` curve, done with `zoompan` at the canvas resolution so it does not soften the picture. """ 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 skill_config import canvas as resolve_canvas, emit, require_binary, utf8_stdout # noqa: E402 from transition_lib import needs_full_chroma, recipe, xfade_arguments # noqa: E402 def ff(value: float) -> str: return f"{value:.6f}".rstrip("0").rstrip(".") or "0" def punch_chain(width: int, height: int, fps: int, frames: int, scale_from: float, ramp_frames: int) -> str: """Cover-crop to the canvas, then ease a scale ramp down to 1.0. `zoompan` works on the already-scaled frame at 2x so the crop is taken from real pixels: ramping the scale on the final-size frame resamples an upscaled image and the open comes out visibly softer than the rest of the cut, which reads as a compression artefact. """ ratio = max(1.0, scale_from) ramp = max(1, ramp_frames) # expo.out: fast at the start, settling — matches how a camera push feels. zoom = (f"if(gte(on,{ramp}),1," f"1+{ratio - 1:.4f}*pow(2,-9*on/{ramp}))") return ( f"scale={width * 2}:{height * 2}:force_original_aspect_ratio=increase:flags=lanczos," f"crop={width * 2}:{height * 2},setsar=1,fps={fps}," f"zoompan=z='{zoom}':x='iw/2-(iw/zoom/2)':y='ih/2-(ih/zoom/2)'" f":d=1:s={width}x{height}:fps={fps}," f"format=yuv420p" ) def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("plan", help="cold_open.json from plan_cold_open.py") parser.add_argument("body", help="The already-cut main video") parser.add_argument("output") parser.add_argument("--source", help="Override the source the hook is taken from") parser.add_argument("--words", help="*.words.json — needed to caption the hook") parser.add_argument("--body-timeline", help="timeline.json of the body, to be shifted") parser.add_argument("--out-timeline", help="Where the merged timeline is written") parser.add_argument("--pick", type=int, default=0, help="0 = the plan's own choice, 1..N = an entry from alternatives") parser.add_argument("--canvas", default="reels") parser.add_argument("--crf", type=int, default=17) parser.add_argument("--preset", default="slow") args = parser.parse_args() require_binary("ffmpeg") plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) board = resolve_canvas(args.canvas) width, height, fps = board.width, board.height, board.fps chosen = plan["cold_open"] if args.pick: alternatives = plan.get("alternatives") or [] if args.pick > len(alternatives): raise SystemExit(f"--pick {args.pick} but only {len(alternatives)} alternatives") alternative = alternatives[args.pick - 1] chosen = {**chosen, "source_in": alternative["start"], "source_out": alternative["end"], "text": alternative["text"], "why": f"выбран вручную из shortlist " f"(score {alternative['score']})"} source = Path(args.source or plan["source"]).resolve() body = Path(args.body).resolve() for path in (source, body): if not path.is_file(): raise SystemExit(f"Not found: {path}") hook_in = float(chosen["source_in"]) hook_out = float(chosen["source_out"]) hook_length = hook_out - hook_in if hook_length <= 0.4: raise SystemExit(f"Hook is only {hook_length:.2f}s long") item = recipe(plan.get("rejoin", {}).get("transition", "flash_punch")) tx_frames = item.frames_default tx_duration = tx_frames / fps motion = plan.get("motion", {}) punch_from = float(motion.get("punch_in_from", 1.12)) punch_seconds = float(motion.get("punch_in_seconds", 1.2)) body_info = probe(body) body_duration = float((body_info.get("video") or {}).get("duration") or body_info.get("duration") or 0.0) # xfade overlaps: the join costs one transition length off the total. shift = round(hook_length - tx_duration, 6) expected = round(hook_length + body_duration - tx_duration, 3) hook_chain = punch_chain(width, height, fps, int(hook_length * fps), punch_from, int(punch_seconds * fps)) body_chain = (f"scale={width}:{height}:force_original_aspect_ratio=increase:flags=lanczos," f"crop={width}:{height},setsar=1,fps={fps},format=yuv420p") full_chroma = needs_full_chroma(item.ffmpeg) graph = [ f"[0:v]{hook_chain}[hv]", f"[1:v]{body_chain}[bv]", "[0:a]aformat=sample_rates=48000:channel_layouts=stereo,asetpts=PTS-STARTPTS[ha]", "[1:a]aformat=sample_rates=48000:channel_layouts=stereo,asetpts=PTS-STARTPTS[ba]", ] left, right = "hv", "bv" if full_chroma: graph += ["[hv]format=yuv444p[hv4]", "[bv]format=yuv444p[bv4]"] left, right = "hv4", "bv4" graph.append(f"[{left}][{right}]" f"{xfade_arguments(item.ffmpeg, tx_duration, hook_length - tx_duration)}" f"[vx]") graph.append("[vx]format=yuv420p[vout]" if full_chroma else "[vx]null[vout]") graph.append(f"[ha][ba]acrossfade=d={ff(tx_duration)}:c1=tri:c2=tri[aout]") command = ["ffmpeg", "-y", "-v", "warning"] if full_chroma: # Same trap as render_transitions: st()/ld() registers are shared across # the filter's threads, so a per-pixel store is overwritten mid-frame. command += ["-filter_complex_threads", "1"] command += [ "-ss", ff(hook_in), "-to", ff(hook_out), "-i", str(source), "-i", str(body), "-filter_complex", ";".join(graph), "-map", "[vout]", "-map", "[aout]", "-c:v", "libx264", "-preset", args.preset, "-crf", str(args.crf), "-profile:v", "high", "-pix_fmt", "yuv420p", "-r", str(fps), "-c:a", "aac", "-b:a", "224k", "-ar", "48000", "-ac", "2", "-movflags", "+faststart", str(Path(args.output).resolve()), ] done = subprocess.run(command, capture_output=True, text=True, errors="replace") if done.returncode: emit({"status": "failed", "graph": ";".join(graph), "tail": (done.stderr or "").strip()[-1800:]}) raise SystemExit(done.returncode) output = Path(args.output).resolve() actual = probe(output) report = { "status": "ok", "output": str(output), "hook": {"text": chosen["text"], "source_in": hook_in, "source_out": hook_out, "length": round(hook_length, 3), "why": chosen.get("why")}, "punch_in": {"from": punch_from, "to": 1.0, "seconds": punch_seconds, "curve": "expo.out"}, "rejoin": {"transition": item.id, "frames": tx_frames, "duration": round(tx_duration, 4)}, "body_duration": round(body_duration, 3), "expected_duration": expected, "actual_duration": actual.get("duration"), "body_shifted_by": shift, } drift = (round(actual["duration"] - expected, 4) if actual.get("duration") else None) report["duration_drift"] = drift report["drift_ok"] = drift is not None and abs(drift) <= max(0.08, 2.0 / fps) # ------------------------------------------------------------- timeline if args.body_timeline and args.out_timeline: timeline = json.loads(Path(args.body_timeline).read_text(encoding="utf-8")) hook_words: list[dict] = [] if args.words: raw = json.loads(Path(args.words).read_text(encoding="utf-8")) source_words = raw.get("words") if isinstance(raw, dict) else raw for entry in source_words or []: start, end = float(entry["start"]), float(entry["end"]) if start >= hook_in - 0.01 and end <= hook_out + 0.01: hook_words.append({ "word": entry["word"], "start": round(start - hook_in, 3), "end": round(min(end, hook_out) - hook_in, 3), "source": "cold_open", }) shifted = [] for entry in timeline.get("words") or []: shifted.append({**entry, "start": round(float(entry["start"]) + shift, 3), "end": round(float(entry["end"]) + shift, 3)}) for clip in timeline.get("clips") or []: clip["global_start"] = round(float(clip["global_start"]) + shift, 4) clip["global_end"] = round(float(clip["global_end"]) + shift, 4) merged = {**timeline} merged["origin"] = "render_cold_open" merged["cold_open"] = {"length": round(hook_length, 3), "transition": item.id, "shift_applied": shift} merged["words"] = [{**word, "id": index} for index, word in enumerate(hook_words + shifted)] merged["expected_duration"] = expected # Derived scalars have to be recomputed, not carried over. `**timeline` # copies `first_word_start` and `last_word_end` describing the *body*, # and after the open they are both wrong — the first word is now the # hook's, and the last is later by the shift. Nothing downstream reads # them as obviously stale: the caption gate compares the first title # against `first_word_start` and reports the file as out of sync when it # is the timeline that is inconsistent. if merged["words"]: merged["first_word_start"] = round(min(float(w["start"]) for w in merged["words"]), 3) merged["last_word_end"] = round(max(float(w["end"]) for w in merged["words"]), 3) merged["tail_after_speech"] = round(expected - merged["last_word_end"], 3) Path(args.out_timeline).write_text( json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8") report["timeline"] = str(args.out_timeline) report["hook_words"] = len(hook_words) report["body_words_shifted"] = len(shifted) if args.words and not hook_words: report["warning"] = ("В окно хука не попало ни одного слова из транскрипта — " "субтитры на холодном старте будут пустыми. Проверь, что " "--words относится к тому же исходнику.") emit(report) if not report["drift_ok"]: raise SystemExit(2) if __name__ == "__main__": main()