#!/usr/bin/env python3 """Harvest a whole reel's worth of b-roll from the transcript, in one pass. Placing one insert by hand is fine. Placing eighteen is not, and eighteen is what a 100-second reel needs at the measured cadence of one cutaway every 5–7 seconds. Doing that by hand is where b-roll goes wrong: the searches drift toward whatever the stock site ranks first, and the result is a reel of interchangeable abstract loops that illustrate nothing. So the queries come from the words that were actually said. Each caption block is scanned for a **concrete, picturable noun**; that noun is mapped to a search phrase; the phrase goes to every provider at once. What comes back is downloaded and contact-sheeted — never auto-placed, because the last two times material was chosen on tags alone it had to be thrown out after the render. The output is a shortlist plus a plan skeleton with the anchor times already filled in. A human picks which rows survive. """ from __future__ import annotations import argparse import json import re import subprocess import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from skill_config import emit, require_binary, utf8_stdout # noqa: E402 HERE = Path(__file__).resolve().parent # Concept -> what to actually search for. The map is the editorial decision: a # stock site has nothing for "нейросеть", but "server room data center" and # "circuit board macro" are what that idea looks like on screen. CONCEPTS: list[tuple[re.Pattern, str, str]] = [ (re.compile(r"нейросет|нейронн|искусственн", re.I), "neural network data visualization dark", "нейросеть"), (re.compile(r"видеокарт|gpu|gtx", re.I), "graphics card gpu close up macro", "видеокарта"), (re.compile(r"компьютер|процессор|чип|железо", re.I), "circuit board macro electronics", "железо"), (re.compile(r"код|программ|скрипт|движок", re.I), "programming code on screen close up", "код"), (re.compile(r"данн|датасет|обуча|тренир", re.I), "big data flowing numbers dark screen", "данные"), (re.compile(r"изображен|фотограф|распозна|камер", re.I), "camera lens photography close up", "камера"), (re.compile(r"лаборатор|исследов|учён|наук", re.I), "science laboratory research microscope", "лаборатория"), (re.compile(r"сервер|облак|дата.?центр", re.I), "server room data center racks", "сервер"), (re.compile(r"город|улиц|транспорт", re.I), "city street night timelapse", "город"), (re.compile(r"человек|люди|команд|зритель", re.I), "people working office team", "люди"), (re.compile(r"скорост|быстр|гонк|разгон", re.I), "speed motion blur highway night", "скорость"), (re.compile(r"монтаж|видео|ролик|контент", re.I), "video editing timeline workstation", "монтаж"), (re.compile(r"график|диаграмм|процент|результат|ошибк", re.I), "financial chart graph rising screen", "график"), (re.compile(r"мозг|мышлен|ум|интеллект", re.I), "human brain hologram abstract", "мозг"), (re.compile(r"сет[ьия]|связ|соединен", re.I), "network connections nodes glowing", "сеть"), ] def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("timeline") parser.add_argument("--out", required=True, help="Where footage is downloaded") parser.add_argument("--every", type=float, default=6.0, help="Target seconds between cutaways (measured optimum 5-7)") parser.add_argument("--per-query", type=int, default=4) parser.add_argument("--orientation", default="portrait") parser.add_argument("--providers", default="pexels,pixabay") parser.add_argument("--duration", default="3-12") parser.add_argument("--plan-out", help="Write the plan skeleton here") parser.add_argument("--search-only", action="store_true", help="Shortlist without downloading") args = parser.parse_args() require_binary("ffmpeg") timeline = json.loads(Path(args.timeline).read_text(encoding="utf-8")) words = timeline.get("words") or [] if not words: raise SystemExit(f"No words in {args.timeline}") duration = float(timeline.get("expected_duration") or words[-1]["end"]) # Walk the timeline and take the first matching concept in each window, so # the cutaways land at the target cadence rather than clustering wherever the # vocabulary happens to be dense. slots: list[dict] = [] cursor = 2.0 while cursor < duration - 3.0: window_end = cursor + args.every picked = None for word in words: start = float(word["start"]) if start < cursor or start > window_end: continue for pattern, query, label in CONCEPTS: if pattern.search(word["word"]): picked = {"at": round(start, 3), "word": word["word"], "query": query, "concept": label} break if picked: break if picked and (not slots or picked["at"] - slots[-1]["at"] >= args.every * 0.7): slots.append(picked) cursor = picked["at"] + args.every else: cursor = window_end queries = sorted({slot["query"] for slot in slots}) out_dir = Path(args.out).resolve() out_dir.mkdir(parents=True, exist_ok=True) found: dict[str, list[dict]] = {} for query in queries: command = [sys.executable, str(HERE / "stock_media.py"), query, "--kind", "video", "--orientation", args.orientation, "--count", str(args.per_query * 3), "--duration", args.duration, "--providers", args.providers] done = subprocess.run(command, capture_output=True, text=True, errors="replace") try: payload = json.loads(done.stdout or "{}") except json.JSONDecodeError: found[query] = [] continue shortlist = payload.get("shortlist") or [] # Native vertical beats a cropped landscape every time: cover-cropping # 16:9 to 9:16 throws away 68% of the frame and magnifies whatever is # left, so a clean wide shot arrives as a soft close-up of nothing. vertical = [item for item in shortlist if isinstance(item.get("size"), str) and "x" in item["size"] and (lambda w, h: h > w)(*[int(v) for v in item["size"].split("x") if v.isdigit()] or [1, 1])] found[query] = (vertical or shortlist)[:args.per_query] picks = [item["pick"] for items in found.values() for item in items] report = { "status": "ok", "timeline": args.timeline, "duration": round(duration, 2), "target_cadence_seconds": args.every, "slots_wanted": len(slots), "slots": slots, "queries": {query: [{"pick": i["pick"], "size": i.get("size"), "duration": i.get("duration"), "creator": i.get("creator")} for i in items] for query, items in found.items()}, "candidates": len(picks), } if not args.search_only and picks: for query, items in found.items(): if not items: continue command = [sys.executable, str(HERE / "stock_media.py"), query, "--kind", "video", "--orientation", args.orientation, "--count", str(args.per_query * 3), "--duration", args.duration, "--providers", args.providers, "--ids", ",".join(i["pick"] for i in items), "--out", str(out_dir)] subprocess.run(command, capture_output=True, text=True, errors="replace") report["downloaded_to"] = str(out_dir) if args.plan_out: # Skeleton only: `path` is left empty on purpose. Filling it from the # download order would be exactly the "the API returned it first" choice # this whole pipeline refuses to make. Path(args.plan_out).write_text(json.dumps({ "broll": [{"start": slot["at"], "duration": 2.4, "source_in": 0.5, "motion": "in" if index % 2 == 0 else "drift", "fade": 0.2, "anchor": slot["word"], "concept": slot["concept"], "query": slot["query"], "path": ""} for index, slot in enumerate(slots)] }, ensure_ascii=False, indent=2), encoding="utf-8") report["plan_skeleton"] = args.plan_out report["next"] = ( "Собери контактный лист по скачанному, посмотри КАДРЫ и впиши path в скелет плана. " "Материал, выбранный по тегам, уже дважды пришлось выбрасывать после рендера.") emit(report) if __name__ == "__main__": main()