#!/usr/bin/env python3 """Choose a transition for every cut, with an energy budget instead of a favourite. What makes an edit look amateur is not a bad transition, it is the same transition every time, or ten loud ones in a row. So selection is constrained: * a recipe is never used twice consecutively, and no family dominates; * each recipe's own `max_per_minute` is enforced over a rolling window; * the total "energy" per minute is capped, so a reel gets a couple of loud moments and a lot of quiet ones rather than uniform noise; * a transition that needs a pause in the speech is only placed where the timeline actually has one — over a spoken word it would bury the word; * the default is a straight cut. Anything else has to be earned by the cut's own context, which is why `--reason` output exists. Input is the list of cuts (from a recut plan, a stitch plan, or a b-roll plan) plus the word timeline used to find speech pauses. """ from __future__ import annotations import argparse import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from skill_config import canvas as resolve_canvas, emit, utf8_stdout # noqa: E402 from transition_lib import load_library, recipes # noqa: E402 # Energy points allowed per minute of finished video. A cut costs 1, a whip 4, # a zoom punch 5 — so a 60 s reel can afford roughly three strong moments. ENERGY_BUDGET_PER_MINUTE = 14 INTENT_FAMILIES = { "calm": ("cut", "light", "organic"), "editorial": ("cut", "light", "motion", "mask"), "dynamic": ("motion", "zoom", "cut", "light"), "tech": ("digital", "mechanical", "motion", "cut"), "premium": ("light", "mask", "dimensional", "cut"), "street": ("motion", "digital", "zoom", "cut"), } DEFAULT_INTENT = "editorial" def load_cuts(path: Path) -> tuple[list[dict], dict]: """Accept a recut plan, a transition plan, or a bare list of cut times.""" payload = json.loads(path.read_text(encoding="utf-8")) if isinstance(payload, list): return [{"at": float(value)} for value in payload], {} if "segments" in payload: # A recut plan: a cut sits between every pair of consecutive segments. cuts = [] cursor = 0.0 segments = payload["segments"] for index, segment in enumerate(segments[:-1]): cursor += float(segment.get("output_seconds") or (segment["source_out"] - segment["source_in"])) cuts.append({ "at": round(cursor, 4), "index": index, "reason": segment.get("cut_reason"), "removed": segment.get("removed_after"), }) return cuts, payload if "cut_points" in payload: return [{"at": float(item["centre"]), "index": item.get("index")} for item in payload["cut_points"]], payload if "clips" in payload: cuts = [] cursor = 0.0 for index, clip in enumerate(payload["clips"][:-1]): cursor += float(clip.get("source_out", 0)) - float(clip.get("source_in", 0)) cuts.append({"at": round(cursor, 4), "index": index}) return cuts, payload raise SystemExit( "Could not find cuts in the input. Expected 'segments', 'cut_points', " "'clips', or a list of times." ) def load_pauses(timeline_path: Path | None, minimum: float) -> list[tuple[float, float]]: """Speech gaps long enough to hide a transition in.""" if timeline_path is None or not timeline_path.exists(): return [] payload = json.loads(timeline_path.read_text(encoding="utf-8")) words = payload.get("words") if isinstance(payload, dict) else payload pauses = [] for index in range(len(words or []) - 1): start = float(words[index]["end"]) end = float(words[index + 1]["start"]) if end - start >= minimum: pauses.append((start, end)) return pauses def in_pause(at: float, pauses: list[tuple[float, float]], slack: float = 0.15) -> bool: return any(start - slack <= at <= end + slack for start, end in pauses) def plan(cuts: list[dict], pauses: list[tuple[float, float]], intent: str, fps: int, allow_energy: int, seed_order: list[str]) -> tuple[list[dict], list[str]]: table = recipes() families = INTENT_FAMILIES.get(intent, INTENT_FAMILIES[DEFAULT_INTENT]) notes: list[str] = [] # Candidates for this intent, quietest first: the default should be cheap. candidates = [ item for item in table.values() if item.family in families ] candidates.sort(key=lambda item: (item.energy, item.id)) if not candidates: raise SystemExit(f"No transitions match intent '{intent}'") total_seconds = max(cuts[-1]["at"] if cuts else 0.0, 1.0) minutes = max(total_seconds / 60.0, 1e-6) energy_cap = allow_energy * minutes used_energy = 0.0 history: list[dict] = [] chosen: list[dict] = [] def recent_count(recipe_id: str, at: float, window: float = 60.0) -> int: return sum(1 for item in history if item["id"] == recipe_id and at - item["at"] <= window) for position, cut in enumerate(cuts): at = float(cut["at"]) pause_here = in_pause(at, pauses) if pauses else False previous = history[-1]["id"] if history else None previous_family = history[-1]["family"] if history else None allowed = [] for item in candidates: if item.id == previous: continue if item.needs_speech_pause and pauses and not pause_here: continue if recent_count(item.id, at) >= item.max_per_minute: continue if item.family == previous_family and item.family != "cut" and \ len(history) >= 2 and history[-2]["family"] == item.family: continue # three of a family in a row reads as a template # Only the plain cut is free. `cut_sfx` still spends energy: an # accent on every join is exactly the "uniform noise" this budget # exists to prevent. if used_energy + item.energy > energy_cap and item.id != "cut": continue allowed.append(item) if not allowed: pick = table["cut"] why = "бюджет энергии или правила разнообразия исчерпаны — прямая склейка" elif seed_order: wanted = seed_order[position % len(seed_order)] match = next((item for item in allowed if item.id == wanted), None) pick = match or allowed[0] why = ("задан явным порядком" if match else f"{wanted} недоступен здесь, взят {allowed[0].id}") else: # Rotate through the allowed set so neighbouring cuts differ, and bias # towards the quiet end unless this cut sits in a real pause. # Over a spoken word only the quietest transitions are allowed; the # loud ones are saved for the gaps the speech already gives us. pool = allowed if pause_here else [ item for item in allowed if item.energy <= 2 ] or allowed pick = pool[position % len(pool)] why = ("в паузе речи можно взять заметный переход" if pause_here else "переход поверх речи — держим низкую энергию") frames = pick.frames_default duration = 0.0 if pick.is_cut else frames / fps used_energy += pick.energy history.append({"id": pick.id, "family": pick.family, "at": at}) chosen.append({ "index": cut.get("index", position), "at": round(at, 4), "id": pick.id, "label": pick.label, "family": pick.family, "energy": pick.energy, "frames": frames, "seconds": round(duration, 4), "in_speech_pause": pause_here, "why": why, "when_to_use": pick.when, "avoid": pick.avoid, "sfx": [dict(cue) for cue in pick.sfx], }) if used_energy > energy_cap: notes.append( f"Суммарная энергия {used_energy:.0f} превысила бюджет {energy_cap:.0f} — " f"проверь, не слишком ли шумно." ) if pauses: placed_in_pause = sum(1 for item in chosen if item["in_speech_pause"]) notes.append( f"{placed_in_pause} из {len(chosen)} переходов попали в паузу речи; " f"остальные держат низкую энергию, чтобы не глушить слово." ) else: notes.append( "Таймлайн слов не передан, паузы речи неизвестны — переходы, требующие " "паузы, исключены. Передай --timeline, чтобы их разрешить." ) return chosen, notes def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("cuts", nargs="?", help="recut_plan.json, transition report, or a JSON list of times") parser.add_argument("--out", default="transition_plan.json") parser.add_argument("--timeline", help="Word timeline used to locate speech pauses") parser.add_argument("--intent", default=DEFAULT_INTENT, choices=list(INTENT_FAMILIES)) parser.add_argument("--canvas", default="reels") parser.add_argument("--fps", type=int) parser.add_argument("--energy-per-minute", type=int, default=ENERGY_BUDGET_PER_MINUTE) parser.add_argument("--pause-minimum", type=float, default=0.22, help="Shortest speech gap that counts as a pause") parser.add_argument("--order", default="", help="Comma-separated recipe ids to cycle through explicitly") parser.add_argument("--list", action="store_true", help="Print the library and exit") args = parser.parse_args() if args.list: library = load_library() emit({ "families": library["families"], "doctrine": library["doctrine"], "transitions": [ {"id": item.id, "label": item.label, "family": item.family, "energy": item.energy, "frames": item.frames_default, "max_per_minute": item.max_per_minute, "needs_speech_pause": item.needs_speech_pause, "when": item.when, "avoid": item.avoid} for item in sorted(recipes().values(), key=lambda r: (r.family, r.energy)) ], }) return if not args.cuts: parser.error("cuts is required unless --list is used") board = resolve_canvas(args.canvas) fps = args.fps or board.fps cuts, source = load_cuts(Path(args.cuts)) if not cuts: raise SystemExit("Nothing to plan: the input has no cuts") pauses = load_pauses(Path(args.timeline) if args.timeline else None, args.pause_minimum) order = [value.strip() for value in args.order.split(",") if value.strip()] chosen, notes = plan(cuts, pauses, args.intent, fps, args.energy_per_minute, order) plan_document = { "version": "1.0", "intent": args.intent, "canvas": args.canvas, "fps": fps, "energy_per_minute": args.energy_per_minute, "source_plan": str(Path(args.cuts).resolve()), "clips": source.get("clips"), "transitions": [ {"id": item["id"], "frames": item["frames"]} for item in chosen ], "decisions": chosen, "notes": notes, } Path(args.out).parent.mkdir(parents=True, exist_ok=True) Path(args.out).write_text(json.dumps(plan_document, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") from collections import Counter emit({ "status": "ok", "plan": str(Path(args.out).resolve()), "cuts": len(chosen), "intent": args.intent, "by_recipe": dict(Counter(item["id"] for item in chosen)), "by_family": dict(Counter(item["family"] for item in chosen)), "total_energy": sum(item["energy"] for item in chosen), "energy_budget": round(args.energy_per_minute * max(chosen[-1]["at"], 1.0) / 60.0, 1), "straight_cuts": sum(1 for item in chosen if item["id"].startswith("cut")), "sfx_cues": sum(len(item["sfx"]) for item in chosen), "notes": notes, "next": ( f"Добавь в план clips и запусти: python scripts/render_transitions.py " f"{args.out} OUT.mp4" ), }) if __name__ == "__main__": main()