#!/usr/bin/env python3 """Plan a talking-head recut: trim non-speech, retime the pace, keep every word. The plan is a list of source segments with a speed each. Two rules make it safe: * A cut never lands inside a word, and never removes a whole gap. Trimming takes the *middle* of a gap and leaves a sliver on both sides, so the release of the previous word and the air before the next one both survive — that is the difference between a tightened edit and an obviously spliced one. * Speed changes only ever happen at a cut. A ramp inside continuous audio is audible; at a splice it is not. Video (`setpts`) and audio (`rubberband`) get the same factor per segment, so lip sync is preserved by construction. Every word from the transcript survives, and the report proves it: the remapped timeline is emitted with a word count and a per-word before/after time, so the caption engine reads the retimed positions rather than the original ones. """ from __future__ import annotations import argparse import json 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 emit, utf8_stdout # noqa: E402 from speech_analysis import DEFAULT_STYLE, STYLES, analyse # noqa: E402 # Speed limits. Beyond about 1.15x Russian speech starts to sound rushed even with # formant-preserving retiming, and below 0.92x it sounds like a dramatisation. SPEED_MIN = 0.90 SPEED_MAX = 1.20 def load_words(path: Path) -> list[dict]: payload = json.loads(path.read_text(encoding="utf-8")) words = payload.get("words") if isinstance(payload, dict) else payload cleaned = [] for index, word in enumerate(words or []): token = str(word.get("word", "")).strip() if not token: continue try: start = float(word["start"]) end = float(word["end"]) except (KeyError, TypeError, ValueError): continue cleaned.append({**word, "word": token, "start": start, "end": max(end, start + 0.02), "id": word.get("id", index)}) cleaned.sort(key=lambda item: item["start"]) return cleaned def build_segments(words: list[dict], gaps, lead_in: float, tail: float, media_duration: float) -> list[dict]: """Contiguous source intervals after trimming the middle of each gap.""" by_index = {gap.index: gap for gap in gaps} start = max(0.0, float(words[0]["start"]) - lead_in) segments: list[dict] = [] current_start = start for index in range(len(words) - 1): gap = by_index.get(index) if gap is None: continue removed = gap.duration - gap.keep if removed <= 0.012: continue # nothing worth cutting; the segment simply continues half = gap.keep / 2.0 segment_end = gap.start + half next_start = gap.end - half if segment_end <= current_start + 0.05: # Degenerate: the previous cut already consumed this region. current_start = min(current_start, next_start) continue segments.append({ "source_in": round(current_start, 4), "source_out": round(segment_end, 4), "cut_reason": gap.kind, "removed_after": round(removed, 4), }) current_start = next_start end = min(media_duration, float(words[-1]["end"]) + tail) if end > current_start + 0.05: segments.append({ "source_in": round(current_start, 4), "source_out": round(end, 4), "cut_reason": "tail", "removed_after": 0.0, }) return segments def quantise_to_frames(segments: list[dict], fps: int) -> list[dict]: """Make every segment last a whole number of output frames. Audio trimming is sample-exact but video is snapped to the frame grid, so a segment whose length is not a multiple of the frame period ends up with picture and sound of slightly different duration. `concat` then aligns the streams at each join and the mismatch shows up as lip sync creeping out over a long recut — seventeen segments at 30 fps can accumulate a third of a second. Rounding the *output* length to whole frames and deriving the source span back from it moves each cut by at most half a frame while making the two streams exactly equal. """ for segment in segments: speed = float(segment.get("speed", 1.0)) span = segment["source_out"] - segment["source_in"] frames = max(1, round(span / speed * fps)) segment["output_frames"] = frames segment["source_out"] = round(segment["source_in"] + frames / fps * speed, 6) segment["source_seconds"] = round(segment["source_out"] - segment["source_in"], 6) segment["output_seconds"] = round(frames / fps, 6) return segments def assign_speed(segments: list[dict], words: list[dict], base_speed: float, ramp: bool, emphasis_ids: set) -> list[dict]: """Give each segment a playback speed. Content decides the ramp: a long uninterrupted explanation tolerates a nudge upward, while a segment that lands a sentence or carries a number is left alone or slowed, because that is the part the viewer is supposed to absorb. """ for segment in segments: speed = base_speed inside = [ word for word in words if word["start"] < segment["source_out"] and word["end"] > segment["source_in"] ] length = segment["source_out"] - segment["source_in"] notes = [] if ramp: has_emphasis = any(word.get("id") in emphasis_ids for word in inside) ends_sentence = bool(inside) and str(inside[-1]["word"]).rstrip()[-1:] in ".!?…" if has_emphasis: speed = min(speed, 0.97) notes.append("несёт акцент или цифру — притормаживаем") elif ends_sentence and length < 3.0: notes.append("короткая фраза с точкой — оставляем как есть") elif length > 6.0: speed = speed * 1.06 notes.append("длинный непрерывный кусок — ускоряем на 6%") elif length > 3.5: speed = speed * 1.03 notes.append("средний кусок — ускоряем на 3%") segment["speed"] = round(max(SPEED_MIN, min(SPEED_MAX, speed)), 4) segment["words"] = [word["id"] for word in inside] segment["word_count"] = len(inside) segment["source_seconds"] = round(length, 4) segment["output_seconds"] = round(length / segment["speed"], 4) segment["speed_note"] = "; ".join(notes) or "базовая скорость" return segments def clips_with_global_times(segments: list[dict], media: Path) -> list[dict]: """Segments annotated with their position on the output timeline.""" clips = [] cursor = 0.0 for index, item in enumerate(segments): length = item["output_seconds"] clips.append({ "source_index": 0, "clip_index": index, "file": str(media.resolve()), "source_in": item["source_in"], "source_out": item["source_out"], "speed": item["speed"], "global_start": round(cursor, 4), "global_end": round(cursor + length, 4), }) cursor += length return clips def remap_words(segments: list[dict], words: list[dict]) -> tuple[list[dict], list[dict]]: """Map every word onto the output timeline. Returns (remapped, lost).""" remapped: list[dict] = [] seen: set = set() cursor = 0.0 for segment in segments: speed = segment["speed"] begin = segment["source_in"] finish = segment["source_out"] for word in words: if word["end"] <= begin or word["start"] >= finish: continue if word["id"] in seen: continue clipped_start = max(word["start"], begin) clipped_end = min(word["end"], finish) remapped.append({ **word, "source_start": round(word["start"], 3), "source_end": round(word["end"], 3), "start": round(cursor + (clipped_start - begin) / speed, 3), "end": round(cursor + (clipped_end - begin) / speed, 3), "speed": speed, "clipped": clipped_start > word["start"] + 0.005 or clipped_end < word["end"] - 0.005, }) seen.add(word["id"]) cursor += (finish - begin) / speed remapped.sort(key=lambda item: item["start"]) lost = [word for word in words if word["id"] not in seen] return remapped, lost def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("media") parser.add_argument("words", help="*.words.json from transcribe_words.py") parser.add_argument("--out", default="recut_plan.json") parser.add_argument("--timeline", default="timeline_recut.json", help="Where to write the retimed word timeline") parser.add_argument("--style", default=DEFAULT_STYLE, choices=list(STYLES), help="How hard to trim non-speech") parser.add_argument("--speed", type=float, default=1.0, help=f"Base playback speed ({SPEED_MIN}-{SPEED_MAX})") parser.add_argument("--ramp", action="store_true", help="Vary speed per segment by content instead of one flat factor") parser.add_argument("--emphasis", default=r"^\d+([.,]\d+)?%?$", help="Regex for words to protect from speed-up") parser.add_argument("--lead-in", type=float, default=0.14, help="Seconds kept before the first word") parser.add_argument("--tail", type=float, default=0.45, help="Seconds kept after the last word") parser.add_argument("--fps", type=int, help="Output frame rate for boundary quantisation " "(default: the source rate)") parser.add_argument("--report", help="Write the full analysis here") args = parser.parse_args() if not SPEED_MIN <= args.speed <= SPEED_MAX: raise SystemExit(f"--speed must be between {SPEED_MIN} and {SPEED_MAX}") media = Path(args.media) info = probe(media) if not info.get("has_audio"): raise SystemExit(f"{media} has no audio track; a speech recut needs one") words = load_words(Path(args.words)) if len(words) < 2: raise SystemExit("Need at least two words to plan a recut") profile, gaps, extras = analyse(media, words, args.style) import re pattern = re.compile(args.emphasis) if args.emphasis else None emphasis_ids = { word["id"] for word in words if pattern and pattern.search(str(word["word"]).strip(".,!?:;»«")) } fps = int(round((info.get("video") or {}).get("avg_frame_rate") or (info.get("video") or {}).get("r_frame_rate") or 30)) if args.fps: fps = args.fps segments = build_segments(words, gaps, args.lead_in, args.tail, float(info.get("duration") or 0.0)) segments = assign_speed(segments, words, args.speed, args.ramp, emphasis_ids) # Quantise after the speed is known: the frame grid applies to output length. segments = quantise_to_frames(segments, fps) remapped, lost = remap_words(segments, words) source_duration = float(info.get("duration") or 0.0) kept_source = sum(item["source_seconds"] for item in segments) output_duration = sum(item["output_seconds"] for item in segments) speeds = sorted({item["speed"] for item in segments}) plan = { "version": "1.0", "source": str(media.resolve()), "source_duration": round(source_duration, 3), "style": args.style, "base_speed": args.speed, "ramp": args.ramp, "fps": fps, "frame_quantised": True, "segments": segments, "expected_duration": round(output_duration, 3), "audio_retime": "rubberband", "contract": ( "Каждый сегмент воспроизводится со своей скоростью; видео и звук " "используют один множитель, поэтому липсинк сохраняется. Смена " "скорости происходит только на склейке." ), } Path(args.out).parent.mkdir(parents=True, exist_ok=True) Path(args.out).write_text(json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") timeline = { "version": "3.0", "origin": "plan_speech_recut", "source": str(media.resolve()), "recut_plan": str(Path(args.out).resolve()), # `global_start`/`global_end` are where each segment lands on the *output* # timeline. Everything downstream (checkpoints, b-roll anchors, transitions) # works in output time, so a timeline without them forces every consumer to # re-derive the same cumulative sum. "clips": clips_with_global_times(segments, media), "words": remapped, "expected_duration": round(output_duration, 3), "first_word_start": remapped[0]["start"] if remapped else None, "last_word_end": remapped[-1]["end"] if remapped else None, "tail_after_speech": round(output_duration - remapped[-1]["end"], 3) if remapped else None, } Path(args.timeline).parent.mkdir(parents=True, exist_ok=True) Path(args.timeline).write_text(json.dumps(timeline, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") summary = { "status": "ok" if not lost else "words_lost", "plan": str(Path(args.out).resolve()), "timeline": str(Path(args.timeline).resolve()), "style": args.style, "source_duration": round(source_duration, 2), "kept_source": round(kept_source, 2), "trimmed_from_gaps": round(kept_source and source_duration - kept_source, 2), "expected_duration": round(output_duration, 2), "saved_seconds": round(source_duration - output_duration, 2), "saved_percent": round(100 * (source_duration - output_duration) / source_duration, 1) if source_duration else 0.0, "segments": len(segments), "fps": fps, "speeds_used": speeds, "words_in": len(words), "words_out": len(remapped), "words_lost": [word["word"] for word in lost], "words_clipped": sum(1 for word in remapped if word["clipped"]), "gap_summary": extras["gap_summary"], "speech_profile": profile.as_dict(), "kept_whole_because_probably_speech": sum( 1 for gap in gaps if gap.kind == "speech_bleed" ), "uncertain_gaps": [gap.as_dict() for gap in gaps if gap.kind == "uncertain"], "hesitations_found": extras["hesitations"], "filler_candidates_for_review": [item["word"] for item in extras["filler_candidates"]], "next": ( f"python scripts/render_speech_recut.py {args.out} OUT.mp4 " f"# затем субтитры строй по {args.timeline}" ), } if args.report: Path(args.report).parent.mkdir(parents=True, exist_ok=True) Path(args.report).write_text( json.dumps({**summary, "gaps": [gap.as_dict() for gap in gaps]}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) emit(summary) if lost: print(f"error: {len(lost)} слов не попали в план — это баг, не сдавай такой монтаж", file=sys.stderr) raise SystemExit(2) if __name__ == "__main__": main()