Files
gnommo/gnommo/cli.py
T

5030 lines
184 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""CLI entry point for GnommoEditor."""
import argparse
import json
from logging import config
import re
import time
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from gnommo.parser import _read_json
from . import __version__
from .errors import GnommoError, ParseError, ValidationError, RenderError
from .cache import get_cache_info, resolve_with_cache
from typing import Optional, Union
class NotImplementedException(GnommoError):
"""Feature not yet implemented."""
pass
def main() -> int:
"""Main entry point."""
parser = argparse.ArgumentParser(
prog="gnommo",
description="GnommoEditor - A code-first video editing pipeline",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
gnommo -p video1 render Render the full project
gnommo -p video1 render --slides S1:S10 Render only slides S1-S10
gnommo -p video1 render --slides S10: Render from S10 to end
gnommo -p video1 validate Validate only
gnommo -p video1 import Generate slides.json from images
gnommo -p video1 pre Preprocess videos (chroma key, etc.)
gnommo -p video1 clear Delete preprocessed outputs so preprocess re-runs them
gnommo -p video1 prune Remove unused entries from videos.json/audio.json/narration.json
gnommo -p video1 prune --dry-run Preview which manifest entries would be removed
gnommo -p video1 trim Auto-detect silence and set skip/take in narration.json
gnommo -p video1 trim --force Redo trim even for segments that already have skip/take
gnommo -p video1 trim --threshold -25 Raise threshold to ignore clothing/room noise
gnommo -p video1 trim -v Show detected silence periods for debugging
gnommo -p video1 transcode Transcode narration folder to H.265 (1st pass, before preprocess)
gnommo -p video1 transcode --replace Delete originals after successful transcode
gnommo -p video1 transcode --crf 28 Lower quality / smaller files (default CRF: 23)
gnommo -p video1 transcode --processed Compress _processed.mov files to HEVC+alpha (2nd pass, after preprocess)
gnommo -p video1 transcode --processed --alpha-quality 0.5 More aggressive alpha compression
gnommo -p video1 transcode --processed --dry-run Preview what would be compressed
gnommo -p video1 transcode --force Re-transcode even if output already exists
gnommo -p video0 new Create a new project with standard folder structure
gnommo -p video1 all Full pipeline: import → preprocess → trim → render → push → handoff → up
gnommo -p video1 render --dry-run Show FFmpeg command without running
gnommo -p video1 grade Sample a few seconds of a raw_mov clip through the talkinghead filters for grading
gnommo -p video1 grade --ss 12 --dur 4 Seek 12s in, produce a 4s preview
gnommo -p video1 grade --file media/narration/raw_mov/clipA.mov Grade a specific raw clip
gnommo -p video1 description Generate YouTube description file
gnommo -p video1 archive Copy project to connected external drive
gnommo -p video1 load Copy project from external drive to local
gnommo -p video1 commit -m "msg" Record a commit to commits.log (required before up)
gnommo -p video1 up Push manifest files to rendering server
gnommo -p video1 up --dry-run Preview which files would be pushed
gnommo -p video1 down Pull files from rendering server to local
gnommo -p video1 push Push project metadata to the local gnommoweb server
gnommo -p video1 push --prod Push project metadata to production gnommoweb (glitch.university)
gnommo -p video1 push --force Force push, overwriting the server copy
gnommo -p video1 pull Pull (fetch) project metadata from the local server
gnommo -p video1 pull --prod Pull project metadata from production
gnommo -p video1 pull --force Force pull, overwriting the local copy
gnommo -p video1 handoff --prod Upload the rendered video for online review (glitch.university/review/<id>)
gnommo -p video1 handoff Upload the rendered video to the local server
gnommo -p video1 handoff --file X Upload a specific video file instead of out/<output_video>
Note: 'push' sends metadata (script/slides/etc); 'handoff' uploads the actual video file.
""",
)
parser.add_argument(
"--version",
action="version",
version=f"%(prog)s {__version__}",
)
# Required arguments
parser.add_argument(
"-p",
"--project",
type=str,
default=None,
help="Project directory (required for all actions except 'pexels --search')",
)
parser.add_argument(
"action",
type=str,
nargs="?",
default="render",
choices=[
"validate",
"preprocess",
"pre",
"trim",
"render",
"grade",
"all",
"align",
"import",
"description",
"archive",
"load",
"commit",
"up",
"down",
"push",
"pull",
"handoff",
"transcode",
"pexels",
"clear",
"prune",
"new",
],
help="Action to perform (default: render)",
)
# Optional arguments
parser.add_argument(
"-v",
"--verbose",
action="store_true",
help="Verbose output",
)
parser.add_argument(
"-f",
"--force",
action="store_true",
help="Force overwrite existing files",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would be done without executing",
)
parser.add_argument(
"--slides",
type=str,
help="Render only a range of slides (e.g., S1:S10, S5:, S10:S20)",
)
parser.add_argument(
"--chunk-slides",
type=int,
default=0,
dest="chunk_slides",
help="Split render into chunks of N slides each and concatenate (overrides render_chunk_slides in .gnommo.conf)",
)
parser.add_argument(
"--res",
type=str,
choices=["full", "low", "tiny"],
default="full",
help="Resolution: 'full' (project res), 'low' (490x270), 'tiny' (320x180 ultrafast)",
)
parser.add_argument(
"-w",
"--workers",
type=int,
default=1,
help="Number of parallel workers for preprocessing (default: 1)",
)
parser.add_argument(
"--file",
default=None,
help="For handoff: path to video file (overrides output_video in project.json)",
)
parser.add_argument(
"--prod",
action="store_true",
help="Target production server (GNOMMOWEB_PROD_URL / GNOMMOWEB_PROD_API_KEY)",
)
parser.add_argument(
"--threshold",
type=float,
default=-40.0,
help="For trim: silence threshold in dB (default: -40). Raise (e.g. -25) to ignore clothing/room noise.",
)
parser.add_argument(
"--model",
type=str,
default="base",
help="For trim: Whisper model for transcription (tiny/base/small/medium/large, default: base)",
)
parser.add_argument(
"--crf",
type=int,
default=23,
help="For transcode: H.265 quality (CRF, default: 23; lower=better quality, larger file)",
)
parser.add_argument(
"--replace",
action="store_true",
help="For transcode: delete original files after successful transcode",
)
parser.add_argument(
"--processed",
action="store_true",
help="For transcode: compress _processed.mov files (with alpha) using HEVC+alpha instead of narration files",
)
parser.add_argument(
"--alpha-quality",
type=float,
default=1.0,
dest="alpha_quality",
help="For transcode --processed: HEVC alpha quality 0.0-1.0 (default: 0.75; lower=smaller file)",
)
parser.add_argument(
"-m",
"--message",
type=str,
default=None,
help="For commit: commit message",
)
parser.add_argument(
"--search",
type=str,
default=None,
metavar="QUERY",
help="For pexels: search Pexels for QUERY and download all results to the assets disk",
)
parser.add_argument(
"--max",
type=int,
default=200,
dest="search_max",
help="For pexels --search: maximum number of videos to download (default: 200)",
)
parser.add_argument(
"--ss",
type=float,
default=None,
dest="grade_ss",
help="For grade: seconds to seek into the raw clip before sampling (default: 5)",
)
parser.add_argument(
"--dur",
type=float,
default=3.0,
dest="grade_dur",
help="For grade: duration in seconds of the preview clip (default: 3)",
)
parser.add_argument(
"--ffmpeg-log",
type=str,
default=None,
dest="ffmpeg_log",
choices=["quiet", "error", "warning", "info", "verbose", "debug", "trace"],
help="Stream FFmpeg output at this -loglevel (e.g. 'verbose', 'debug') instead of the quiet progress bar",
)
args = parser.parse_args()
# FFmpeg log verbosity for every ffmpeg run: an explicit --ffmpeg-log level
# wins; otherwise -v turns on verbose streaming.
from .preprocessor import set_ffmpeg_verbose, set_ffmpeg_loglevel
if args.ffmpeg_log:
set_ffmpeg_loglevel(args.ffmpeg_log)
elif args.verbose:
set_ffmpeg_verbose(True)
# Resolve project path (optional for 'pexels --search')
action = args.action
if args.project is None:
if action == "pexels" and args.search:
project_path = Path.cwd() # placeholder; cmd_pexels won't use it in search mode
else:
parser.error("argument -p/--project is required")
return 1
else:
project_path = Path(args.project)
if not project_path.is_absolute():
project_path = Path.cwd() / project_path
try:
# Handle actions
if action == "import":
return cmd_import(project_path, args.force, args.verbose)
elif action == "validate":
return cmd_validate(project_path, args.verbose)
elif action == "new":
return cmd_new(project_path, args.verbose)
elif action == "clear":
return cmd_clear(project_path, args.verbose)
elif action == "prune":
return cmd_prune(project_path, args.verbose, args.dry_run)
elif action in ("preprocess", "pre"):
return cmd_preprocess(
project_path,
args.verbose,
args.dry_run,
args.force,
args.workers,
args.res,
)
elif action == "trim":
return cmd_trim(
project_path, args.verbose, args.force, args.threshold, args.res, args.model
)
elif action == "transcode":
return cmd_transcode(
project_path,
args.verbose,
args.dry_run,
args.replace,
args.crf,
args.force,
args.processed,
args.alpha_quality,
)
elif action == "render":
return cmd_render(
project_path,
args.verbose,
args.dry_run,
args.slides,
args.res,
args.force,
chunk_slides=args.chunk_slides,
)
elif action == "grade":
return cmd_grade(
project_path,
args.verbose,
file=args.file,
ss=args.grade_ss,
dur=args.grade_dur,
)
elif action == "align":
return cmd_align(project_path, args.verbose)
elif action == "all":
return cmd_all(
project_path, args.verbose, args.dry_run, args.res, args.force
)
elif action == "description":
return cmd_description(project_path, args.verbose)
elif action == "archive":
return cmd_archive(project_path, args.verbose, args.dry_run)
elif action == "load":
return cmd_load(project_path, args.verbose, args.dry_run)
elif action == "commit":
from .transfer import cmd_commit
if not args.message:
print("Error: -m 'message' is required for commit.")
return 1
return cmd_commit(project_path, args.message)
elif action == "up":
from .transfer import cmd_up
return cmd_up(project_path, args.verbose, args.dry_run)
elif action == "down":
from .transfer import cmd_down
return cmd_down(project_path, args.verbose, args.dry_run)
elif action == "push":
from .push import cmd_push
return cmd_push(project_path, args.verbose, args.force, args.prod)
elif action == "pull":
from .pull import cmd_pull
return cmd_pull(project_path, args.verbose, args.force, args.prod)
elif action == "handoff":
from .handoff import cmd_handoff
return cmd_handoff(
project_path, args.verbose, args.file, args.prod, args.res
)
elif action == "pexels":
return cmd_pexels(project_path, args.verbose, args.search, args.search_max)
except GnommoError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
except KeyboardInterrupt:
print("\nAborted.", file=sys.stderr)
return 130
return 0
# =============================================================================
# Import Command
# =============================================================================
def _prompter_break_line(text: str, max_words: int = 10) -> list[str]:
"""Break a long line into short breath-sized chunks for teleprompter display.
Priority order for break points:
1. Sentence end (. ? !) — always break here
2. Em/en-dash ( — / - ) — break if ≥ 4 words accumulated
3. Semicolon (;) — break if ≥ 4 words accumulated
4. Comma (,) — break if ≥ 5 words accumulated
5. Word ceiling (max_words) — force break at word boundary
Each output line is stripped and non-empty.
"""
import re
words = text.split()
if len(words) <= max_words:
return [text.strip()] if text.strip() else []
lines: list[str] = []
current: list[str] = []
# Dash detection: standalone dash token, or word that is entirely dashes/hyphens
_DASH_RE = re.compile(r"^[-–—]+$")
_SENTENCE_END_RE = re.compile(r'[.?!]["\']?$')
_SOFT_END_COMMA = re.compile(r",$")
_SOFT_END_SEMI = re.compile(r";$")
def flush(force: bool = False) -> None:
"""Flush current words to a line.
If the accumulated chunk is very short (≤ 2 words) and this isn't a
sentence-end forced flush, hold it — it will merge into the next line.
This prevents orphaned 1-2-word lines from soft breaks.
"""
if not force and len(current) <= 2:
return # too short to stand alone — let more words join
chunk = " ".join(current).strip()
if chunk:
lines.append(chunk)
current.clear()
i = 0
while i < len(words):
word = words[i]
n = len(current)
# Peek ahead: is the next token a standalone dash?
next_is_dash = (i + 1 < len(words)) and _DASH_RE.match(words[i + 1])
if _DASH_RE.match(word):
# Standalone dash: attach to current line then break.
current.append(word)
if n > 0:
flush()
elif _SENTENCE_END_RE.search(word):
current.append(word)
flush(force=True) # sentence end always flushes regardless of length
elif _SOFT_END_SEMI.search(word) and n >= 4:
current.append(word)
flush()
elif _SOFT_END_COMMA.search(word) and n >= 5:
current.append(word)
flush()
elif next_is_dash and n >= 5:
flush()
current.append(word)
elif n >= max_words:
flush(force=True) # word ceiling: must break even if line is short
current.append(word)
else:
current.append(word)
i += 1
flush()
return lines
def _export_prompter_manuscript(
manuscript_path: Path,
verbose: bool,
prompter_wpm: int = 130,
max_words: int = 10,
) -> None:
"""Generate manuscript_prompter.txt from manuscript.txt.
Strips all bracketed markers except [S*] and [cue:*], auto-breaks long
lines into breath-sized chunks, and converts [pause:Xs] markers into blank
lines calibrated to the teleprompter scroll speed.
Calibration: the scroll speed is assumed to match reading speed (prompter_wpm).
At that speed each line takes (60 * max_words / prompter_wpm) seconds, so a
[pause:Xs] marker becomes round(X * prompter_wpm / (60 * max_words)) blank lines,
preceded by a visible label line ("· · · 8s · · ·").
"""
import re
output_path = manuscript_path.parent / "manuscript_prompter.txt"
text = manuscript_path.read_text(encoding="utf-8")
_MARKER_RE = re.compile(r"\[[^\]]+\]")
_KEEP_RE = re.compile(r"^\[S[^\]]*\]$|^\[cue:[^\]]*\]$")
_SEGMENT_RE = re.compile(r"^\[segment:[^\]]*\]$")
_PAUSE_RE = re.compile(r"^\[pause:(\d+(?:\.\d+)?)s?\]$")
_BLOCK_BREAK = "\x00BLOCK\x00"
def _pause_lines(seconds: float) -> list[str]:
"""Convert a pause duration into a label + blank lines for the teleprompter."""
n_blanks = max(1, round(seconds * prompter_wpm / (60 * max_words)))
label = f"· · · {seconds:g}s · · ·"
return [label] + [""] * n_blanks
# Step 1: strip non-display markers, collect cleaned lines
cleaned_lines: list[str] = []
for line in text.splitlines():
stripped = line.strip()
if _MARKER_RE.fullmatch(stripped):
if _SEGMENT_RE.match(stripped):
cleaned_lines.append(_BLOCK_BREAK)
elif _KEEP_RE.match(stripped):
cleaned_lines.append(stripped)
else:
m = _PAUSE_RE.match(stripped)
if m:
cleaned_lines.extend(_pause_lines(float(m.group(1))))
# else: drop marker-only lines entirely
else:
def _replace(m: re.Match) -> str:
# Handle inline [pause:Xs] — convert to label inline
if _PAUSE_RE.match(m.group(0)):
sec = float(_PAUSE_RE.match(m.group(0)).group(1))
return f"· · · {sec:g}s · · ·"
return m.group(0) if _KEEP_RE.match(m.group(0)) else ""
cleaned = _MARKER_RE.sub(_replace, line).strip()
if cleaned:
cleaned_lines.append(cleaned)
else:
cleaned_lines.append("") # preserve intentional blank lines
# Step 2: auto-break long text lines; pass markers, blanks, and pause labels through
_PAUSE_LABEL_RE = re.compile(r"^· · · .+ · · ·$")
broken: list[str] = []
for line in cleaned_lines:
stripped = line.strip()
if not stripped or stripped == _BLOCK_BREAK or _KEEP_RE.match(stripped) or _PAUSE_LABEL_RE.match(stripped):
broken.append(line)
else:
broken.extend(_prompter_break_line(stripped))
# Step 3: merge each slide marker onto the first word of its slide,
# then lay out the final lines.
# - [S*] markers are prepended inline to the next text line
# - Block-break sentinels expand to a double blank (Elgato block boundary)
# - Ordinary consecutive blank lines are collapsed to one
merged: list[str] = []
pending_marker: str = ""
pending_breaks: int = 0 # block-breaks that arrived while a marker was pending
for line in broken:
stripped = line.strip()
if stripped and _KEEP_RE.match(stripped):
pending_marker = stripped # hold until the next text line
elif pending_marker:
if stripped and stripped != _BLOCK_BREAK:
# Flush any buffered block-breaks first, then attach marker to text
for _ in range(pending_breaks):
merged.append(_BLOCK_BREAK)
pending_breaks = 0
merged.append(f"{pending_marker} {stripped}")
pending_marker = ""
elif stripped == _BLOCK_BREAK:
# Buffer the block-break — emit it after the marker attaches
pending_breaks += 1
else:
# Ordinary blank: carry forward (swallow into the marker's gap)
pass
else:
merged.append(line)
if pending_marker:
for _ in range(pending_breaks):
merged.append(_BLOCK_BREAK)
merged.append(pending_marker)
result: list[str] = []
prev_blank_count = 0
for line in merged:
if line == _BLOCK_BREAK:
needed = 2 - prev_blank_count
result.extend([""] * max(needed, 0))
prev_blank_count = 2
continue
is_blank = not line.strip()
if is_blank:
if prev_blank_count >= 1:
continue # collapse ordinary consecutive blanks
result.append("")
prev_blank_count += 1
else:
result.append(line)
prev_blank_count = 0
output_path.write_text("\n".join(result).strip() + "\n", encoding="utf-8")
print(f" Prompter manuscript → {output_path.name}")
def cmd_import(project_path: Path, force: bool, verbose: bool) -> int:
"""Import assets and generate metadata JSON files."""
from .parser import parse_project_config, _read_json
print(f"Importing assets for: {project_path.name}")
if not project_path.exists():
print(f"Error: Project directory not found: {project_path}", file=sys.stderr)
return 1
# Load project config if it exists (for videos_path and default_filters)
config = None
if (project_path / "project.json").exists():
config = parse_project_config(project_path)
# Import videos from media/videos directory
if config and config.videos_path:
videos_json_path = project_path / config.videos_path
videos_dir = videos_json_path.parent
else:
videos_dir = project_path / "media" / "videos"
if videos_dir.exists():
_import_videos(videos_dir, config, verbose)
# Import narration segments from media/narration directory
narration_dir = project_path / "media" / "narration"
if narration_dir.exists():
_import_narration_segments(narration_dir, config, verbose)
# Import presenter notes from Keynote file (also exports slide PNGs)
keynote_files = list(project_path.glob("*.key"))
if keynote_files:
keynote_file = keynote_files[0] # Use first .key file found
if len(keynote_files) > 1:
print(f" Warning: Multiple .key files found, using {keynote_file.name}")
_import_presenter_notes(project_path, keynote_file, verbose, config)
# Generate slides.json for each slide directory (after Keynote export)
slides_base = project_path / "media" / "slides"
slides_dirs = (
[d for d in slides_base.glob("*/") if d.is_dir()]
if slides_base.exists()
else []
)
for slides_dir in slides_dirs:
_generate_slides_json(slides_dir, verbose)
else:
if verbose:
print(" No .key file found, skipping presenter notes import")
# Import shared assets (pexels, etc.) from shared_assets directory
# Look for shared_assets relative to project or in parent directories
shared_assets_dir = _find_shared_assets(project_path)
if shared_assets_dir:
_import_shared_assets(shared_assets_dir, verbose)
_import_shared_audio(shared_assets_dir, project_path, config, verbose)
_sync_shared_videos_to_local(project_path, config, shared_assets_dir, verbose)
# Probe and cache audio file durations into audio.json
_probe_audio_durations(project_path, config, force, verbose, shared_assets_dir)
# Probe and cache video metadata (duration, has_audio) into videos.json
_probe_video_metadata(project_path, config, shared_assets_dir, force, verbose)
# ETL: if a manuscript exists, project shorthand marker semantics (cutout/layer)
# into videos.json so the render stage is always data-driven from the manuscript.
# Run AFTER sync so newly-added shared videos are already present when we write
# their cutout/layer. Also warn about any referenced video that is still missing.
manuscript_path = project_path / "manuscript.txt"
if manuscript_path.exists() and config:
from .parser import parse_manuscript
from .transformer import _SHORTHAND_PREFIXES
_, markers, _, _ = parse_manuscript(project_path)
if markers:
_project_markers_to_videos(
markers,
project_path / config.videos_path,
config,
project_path,
)
# Warn about shorthand-referenced videos still absent from videos.json
videos_json_path = project_path / config.videos_path
local_vids: dict = (
_read_json(videos_json_path) if videos_json_path.exists() else {}
)
seen_missing: set[str] = set()
for marker in markers:
for prefix in _SHORTHAND_PREFIXES:
if marker.startswith(prefix):
vid_id = marker[len(prefix):].lower()
if vid_id not in local_vids and vid_id not in seen_missing:
hint = (
"run 'gnommo pexels' to download"
if vid_id.startswith("pexels/")
else "add it manually"
)
print(
f" ⚠ [{marker}] video '{vid_id}' not found in "
f"videos.json or shared_assets — {hint}"
)
seen_missing.add(vid_id)
break
# Generate teleprompter version of manuscript
if manuscript_path.exists():
import configparser as _cp
_pcfg = _cp.ConfigParser()
_pcfg.read(Path.home() / ".gnommo.conf")
_wpm = int(_pcfg.get("prompter", "wpm", fallback="130"))
_mpw = int(_pcfg.get("prompter", "max_words_per_line", fallback="10"))
_export_prompter_manuscript(manuscript_path, verbose, prompter_wpm=_wpm, max_words=_mpw)
print("Import complete.")
return 0
def _import_shared_audio(
shared_assets_dir: Path,
project_path: Path,
config,
verbose: bool,
) -> None:
"""Import audio files from shared_assets/media/audio into the project's audio.json."""
audio_extensions = {".mp3", ".wav", ".aac", ".m4a", ".ogg", ".flac"}
shared_audio_dir = shared_assets_dir / "media" / "audio"
if not shared_audio_dir.exists():
if verbose:
print(f" No shared audio dir found at {shared_audio_dir}")
return
audio_files = sorted(
f
for f in shared_audio_dir.iterdir()
if f.is_file()
and f.suffix.lower() in audio_extensions
and not f.name.startswith(".")
)
if not audio_files:
if verbose:
print(f" No audio files found in {shared_audio_dir}")
return
# Resolve project audio.json path
if config and config.audio_path:
audio_json_path = project_path / config.audio_path
else:
audio_json_path = project_path / "media" / "audio" / "audio.json"
audio_json_path.parent.mkdir(parents=True, exist_ok=True)
existing: dict = _read_json(audio_json_path) if audio_json_path.exists() else {}
added = 0
for f in audio_files:
audio_id = f.stem.lower()
if audio_id in existing:
if verbose:
print(f" Skipping {audio_id} (already in audio.json)")
continue
existing[audio_id] = {
"file": f.name,
"is_shared": True,
"volume": 1.0,
}
added += 1
if verbose:
print(f" Added shared audio: {audio_id}")
if added > 0:
with open(audio_json_path, "w", encoding="utf-8") as fh:
json.dump(existing, fh, indent=2)
print(
f" Updated {audio_json_path.relative_to(project_path)} (+{added} shared audio files)"
)
else:
if verbose:
print(f" No new shared audio files to add")
def _probe_audio_durations(
project_path: Path,
config,
force: bool,
verbose: bool,
shared_assets_dir: Optional[Path] = None,
) -> None:
"""Probe and cache audio file durations into audio.json.
Runs once at import time so the render stage never needs to scan audio files.
Skips entries that already have a duration unless --force is set.
"""
from .renderer import _get_audio_duration
if config and config.audio_path:
audio_json_path = project_path / config.audio_path
else:
audio_json_path = project_path / "audio.json"
if not audio_json_path.exists():
return
audio_dir = audio_json_path.parent
data = _read_json(audio_json_path)
updated = False
for audio_id, audio_data in data.items():
if "file" not in audio_data:
continue
if "duration" in audio_data and not force:
if verbose:
print(f" Audio '{audio_id}': cached ({audio_data['duration']:.1f}s)")
continue
if audio_data.get("is_shared") and shared_assets_dir:
audio_path = shared_assets_dir / "media" / "audio" / audio_data["file"]
else:
audio_path = audio_dir / audio_data["file"]
if not audio_path.exists():
if verbose:
print(f" Audio '{audio_id}': file not found, skipping")
continue
print(
f" Probing audio '{audio_id}' ({audio_path.name})...", end=" ", flush=True
)
try:
duration = _get_audio_duration(audio_path)
data[audio_id]["duration"] = round(duration, 3)
updated = True
print(f"{duration:.1f}s")
except Exception as e:
print(f"failed ({e})")
if updated:
with open(audio_json_path, "w") as f:
json.dump(data, f, indent=4)
print(f" Saved durations to {audio_json_path.name}")
def _probe_video_metadata(
project_path: Path,
config,
shared_assets_dir: Optional[Path],
force: bool,
verbose: bool,
) -> None:
"""Probe and cache video file duration and audio presence into videos.json.
Runs once at import time so the render stage never needs to probe video files.
Shared entries are written back to shared_assets/videos.json (canonical source).
Local entries are written to the project's videos.json.
Skips entries that already have both fields unless --force is set.
"""
from .preprocessor import get_video_duration
from .renderer import _has_audio_stream
if config and config.videos_path:
videos_json_path = project_path / config.videos_path
else:
videos_json_path = project_path / "media" / "videos" / "videos.json"
if not videos_json_path.exists():
return
videos_dir = videos_json_path.parent
local_data = _read_json(videos_json_path)
# Load shared_assets/videos.json separately — shared probes write there
shared_json_path = shared_assets_dir / "videos.json" if shared_assets_dir else None
shared_data = (
_read_json(shared_json_path)
if shared_json_path and shared_json_path.exists()
else {}
)
local_updated = False
shared_updated = False
for video_id, video_data in local_data.items():
if "source_file" not in video_data:
continue
is_shared = video_data.get("is_shared", False)
# For shared entries, check the shared_assets/videos.json for cached values
if is_shared and video_id in shared_data:
canonical = shared_data[video_id]
else:
canonical = video_data
if not force and "duration" in canonical and "has_audio" in canonical:
if verbose:
print(
f" Video '{video_id}': cached ({canonical['duration']:.1f}s, audio={canonical['has_audio']})"
)
continue
base_dir = (
shared_assets_dir if (is_shared and shared_assets_dir) else videos_dir
)
# Mirror renderer._resolve_video_path: try output_file first, then source_file
# Use resolve_with_cache so files on external disks (LaCie, GnommoDisk) are found.
from .cache import resolve_with_cache
video_path = None
output_file = video_data.get("output_file")
if output_file:
for candidate_dir in [base_dir, base_dir.parent]:
candidate = candidate_dir / output_file
candidate, _ = resolve_with_cache(candidate, project_path)
if candidate.exists():
video_path = candidate
break
mov_candidate = candidate.with_suffix(".mov")
mov_candidate, _ = resolve_with_cache(mov_candidate, project_path)
if mov_candidate.exists():
video_path = mov_candidate
break
if video_path is None:
source_candidate = base_dir / video_data["source_file"]
source_candidate, _ = resolve_with_cache(source_candidate, project_path)
if source_candidate.exists():
video_path = source_candidate
if video_path is None:
if verbose:
print(f" Video '{video_id}': file not found, skipping")
continue
print(
f" Probing video '{video_id}' ({video_path.name})...", end=" ", flush=True
)
try:
duration = get_video_duration(video_path)
has_audio = _has_audio_stream(video_path)
result = {"duration": round(duration, 3), "has_audio": has_audio}
print(f"{duration:.1f}s, audio={has_audio}")
if is_shared and video_id in shared_data:
# Write back to shared_assets/videos.json — canonical source for shared assets
shared_data[video_id].update(result)
shared_updated = True
else:
local_data[video_id].update(result)
local_updated = True
except Exception as e:
print(f"failed ({e})")
if local_updated:
with open(videos_json_path, "w") as f:
json.dump(local_data, f, indent=4)
print(f" Saved metadata to {videos_json_path.name}")
if shared_updated and shared_json_path:
with open(shared_json_path, "w") as f:
json.dump(shared_data, f, indent=4)
print(f" Saved shared metadata to {shared_json_path.name}")
def _sync_shared_videos_to_local(
project_path: Path, config, shared_assets_dir: Path, verbose: bool
) -> None:
"""Append entries from shared_assets/videos.json into the project's local videos.json.
Each new entry gets is_shared=true so the renderer looks in shared_assets_dir.
Existing local entries are never overwritten (preserves cutout, layer, filters, etc.).
"""
shared_videos_json = shared_assets_dir / "videos.json"
if not shared_videos_json.exists():
return
shared_videos = _read_json(shared_videos_json)
if not shared_videos:
return
if config and config.videos_path:
local_json_path = project_path / config.videos_path
else:
local_json_path = project_path / "media" / "videos" / "videos.json"
local_videos: dict = {}
if local_json_path.exists():
local_videos = _read_json(local_json_path)
_METADATA_FIELDS = ("duration", "has_audio")
added = []
metadata_updated = []
for video_id, shared_entry in shared_videos.items():
if video_id in local_videos:
# Propagate any metadata fields that were probed into shared_assets/videos.json
changed = False
for field in _METADATA_FIELDS:
if (
field in shared_entry
and local_videos[video_id].get(field) != shared_entry[field]
):
local_videos[video_id][field] = shared_entry[field]
changed = True
if changed:
metadata_updated.append(video_id)
elif verbose:
print(f" Shared '{video_id}': already in local videos.json, skipping")
continue
# New entry — copy base metadata from shared and mark it as shared.
# Exclude presentation fields (cutout, layer) so _project_markers_to_videos
# can set them from the manuscript without stale shared values interfering.
_PRESENTATION_FIELDS = {"cutout", "layer", "pause_narration"}
local_entry = {
k: v for k, v in shared_entry.items() if k not in _PRESENTATION_FIELDS
}
local_entry["is_shared"] = True
local_videos[video_id] = local_entry
added.append(video_id)
if added or metadata_updated:
local_json_path.parent.mkdir(parents=True, exist_ok=True)
with open(local_json_path, "w", encoding="utf-8") as f:
json.dump(local_videos, f, indent=4)
if added:
print(
f" Synced {len(added)} shared asset(s) to local videos.json: {', '.join(added)}"
)
if metadata_updated:
print(
f" Updated metadata for {len(metadata_updated)} shared asset(s): {', '.join(metadata_updated)}"
)
elif verbose:
print(" No new shared assets to sync to local videos.json")
def _find_shared_assets(project_path: Path) -> Optional[Path]:
"""Find the shared_assets directory.
Looks in:
1. project_path/shared_assets
2. project_path/../shared_assets (sibling to project)
"""
# Check if shared_assets is inside project
if (project_path / "shared_assets").exists():
return project_path / "shared_assets"
# Check if shared_assets is sibling to project
if (project_path.parent / "shared_assets").exists():
return project_path.parent / "shared_assets"
return None
def _import_shared_assets(shared_assets_dir: Path, verbose: bool) -> None:
"""Import video files from shared_assets directory into videos.json.
Scans the root level and all subdirectories for video files and creates
a unified videos.json in shared_assets/.
Video IDs use the filename for root-level files (e.g., "Logo") or
are prefixed with the subfolder name for subdirectory files (e.g., "pexels/filename").
"""
video_extensions = {".mov", ".mp4", ".webm", ".avi", ".mkv", ".m4v"}
# Find all video files in shared_assets (root level and subdirectories).
# Also scan external disk mirrors so files placed there are registered.
from .cache import load_assets_config, load_cache_config
scan_roots: list[Path] = [shared_assets_dir]
for external_base in (load_cache_config(), load_assets_config()):
if external_base:
ext_shared = external_base / "shared_assets"
if ext_shared.exists() and ext_shared != shared_assets_dir:
scan_roots.append(ext_shared)
video_files: list[tuple[Path, Path]] = [] # (relative_path, absolute_path)
seen_rel: set[str] = set() # deduplicate by relative path
for scan_root in scan_roots:
for item in scan_root.iterdir():
if item.name.startswith("."):
continue
if item.is_file():
if (
item.suffix.lower() in video_extensions
and not item.name.endswith("_processed.mov")
and not item.name.endswith("_processed.webm")
):
rel_path = item.relative_to(scan_root)
if str(rel_path) not in seen_rel:
seen_rel.add(str(rel_path))
video_files.append((rel_path, item))
elif item.is_dir():
for video_file in item.rglob("*"):
if (
video_file.is_file()
and video_file.suffix.lower() in video_extensions
and not video_file.name.endswith("_processed.mov")
and not video_file.name.endswith("_processed.webm")
):
rel_path = video_file.relative_to(scan_root)
if str(rel_path) not in seen_rel:
seen_rel.add(str(rel_path))
video_files.append((rel_path, video_file))
if not video_files:
if verbose:
print(f" No video files found in {shared_assets_dir}")
return
# Load existing videos.json if it exists
videos_json_path = shared_assets_dir / "videos.json"
existing_videos: dict = {}
if videos_json_path.exists():
existing_videos = _read_json(videos_json_path)
# Remove entries whose source file no longer exists on any scan root
removed_count = 0
stale_keys = [
vid_id
for vid_id, vid_data in existing_videos.items()
if vid_data.get("source_file")
and not any((root / vid_data["source_file"]).exists() for root in scan_roots)
]
for vid_id in stale_keys:
del existing_videos[vid_id]
removed_count += 1
if verbose:
print(f" Removed stale shared entry: {vid_id}")
if removed_count > 0:
print(f" Removed {removed_count} stale shared asset(s) from videos.json")
# Add new videos (don't overwrite existing)
added_count = 0
for rel_path, abs_path in sorted(video_files):
# Use path relative to shared_assets without extension as video_id (lowercase)
# e.g., "logo" for root files, "pexels/6759604-hd" for subdirectory files
video_id = str(rel_path.with_suffix("")).lower()
if video_id in existing_videos:
if verbose:
print(f" Skipping {video_id} (already exists)")
continue
existing_videos[video_id] = {
"source_file": str(rel_path),
}
added_count += 1
if verbose:
print(f" Added: {video_id}")
if added_count > 0 or removed_count > 0:
# Write updated videos.json
with open(videos_json_path, "w", encoding="utf-8") as f:
json.dump(existing_videos, f, indent=2)
if added_count > 0:
print(f" Updated {videos_json_path} (+{added_count} shared assets)")
else:
print(f" No new shared assets to add")
def _generate_slides_json(directory: Path, verbose: bool) -> None:
"""Generate slides.json from Keynote export folder."""
extensions = {".png", ".gif", ".pdf", ".jpg", ".jpeg"}
files = [f for f in directory.iterdir() if f.suffix.lower() in extensions]
if not files:
print(f" Warning: No image files in {directory}")
return
# Extract numeric suffix from filenames like "Video1.001.png"
pattern = re.compile(r"\.(\d+)\.[^.]+$")
slides = {}
for file in files:
match = pattern.search(file.name)
if match:
num = int(match.group(1))
slide_id = f"S{num}"
slides[slide_id] = {
"image": file.name,
"type": "fullscreen",
}
if not slides:
print(f" Warning: No valid slide files in {directory}")
return
# Sort by slide number
sorted_slides = dict(sorted(slides.items(), key=lambda x: int(x[0][1:])))
# Write slides.json only if content changed
output_path = directory / "slides.json"
new_content = json.dumps(sorted_slides, indent=2)
existing_content = (
output_path.read_text(encoding="utf-8") if output_path.exists() else None
)
if new_content != existing_content:
with open(output_path, "w", encoding="utf-8") as f:
f.write(new_content)
print(f" Generated {output_path} ({len(sorted_slides)} slides)")
if verbose:
for slide_id in sorted_slides:
print(f" [{slide_id}]")
def _import_videos(videos_dir: Path, config, verbose: bool) -> None:
"""Import video files into videos.json.
Scans the videos directory for video files and adds them to videos.json.
Uses the filename (without extension) as the video_id.
Does not overwrite existing entries - only adds new ones.
If the video filename matches a pattern like 'talkinghead*' and a 'talkinghead'
filter preset exists in default_filters, it will be applied automatically.
"""
video_extensions = {".mov", ".mp4", ".webm", ".avi", ".mkv", ".m4v"}
# Find all video files (exclude processed outputs, proxies, and intermediate files)
video_files = [
f
for f in videos_dir.iterdir()
if f.is_file()
and f.suffix.lower() in video_extensions
and "_processed" not in f.stem # Exclude any _processed files
and "_fixed" not in f.stem # Exclude any _fixed files
]
# Also exclude files in subdirectories (proxy/, intermediate/, etc.)
video_files = [f for f in video_files if f.parent == videos_dir]
# Ensure videos.json exists even if there are no video files yet
videos_json_path = videos_dir / "videos.json"
if not videos_json_path.exists():
videos_dir.mkdir(parents=True, exist_ok=True)
with open(videos_json_path, "w", encoding="utf-8") as f:
json.dump({}, f, indent=2)
print(
f" Created empty {videos_json_path.relative_to(videos_dir.parent.parent)}"
)
if not video_files:
if verbose:
print(f" No new video files found in {videos_dir}")
return
# Load existing videos.json
existing_videos: dict = {}
if videos_json_path.exists():
existing_videos = _read_json(videos_json_path)
# Remove entries whose source file no longer exists on disk.
# Skip is_shared entries — their source files live in shared_assets, not videos_dir.
removed_count = 0
stale_keys = [
vid_id
for vid_id, vid_data in existing_videos.items()
if not vid_data.get("is_shared")
and vid_data.get("source_file")
and not (videos_dir / vid_data["source_file"]).exists()
]
for vid_id in stale_keys:
del existing_videos[vid_id]
removed_count += 1
if verbose:
print(f" Removed stale entry: {vid_id}")
if removed_count > 0:
print(f" Removed {removed_count} stale video(s) from {videos_json_path.name}")
# Get available filter presets from config
default_filters = config.default_filters if config else {}
# Add new videos (don't overwrite existing)
added_count = 0
for video_file in sorted(video_files):
# Use filename without extension as video_id (lowercase for new entries)
video_id = video_file.stem.lower()
if video_id in existing_videos:
if verbose:
print(f" Skipping {video_id} (already exists)")
continue
# Build the video entry
video_entry = {
"source_file": video_file.name,
"output_file": video_file.name,
"cutout": "square",
"filter": [],
}
if verbose:
print(f" Added: {video_id}")
existing_videos[video_id] = video_entry
added_count += 1
if added_count > 0 or removed_count > 0:
# Write updated videos.json
with open(videos_json_path, "w", encoding="utf-8") as f:
json.dump(existing_videos, f, indent=2)
if added_count > 0:
print(f" Updated {videos_json_path.name} (+{added_count} videos)")
else:
print(f" No new videos to add")
def _import_narration_segments(narration_dir: Path, config, verbose: bool) -> None:
"""Import narration video files into narration.json.
Folder structure:
media/narration/raw_mov/ ← raw recordings from iPhone/QuickTime
media/narration/processed/ ← chroma-keyed output (preprocess)
media/narration/narration.json
Scans processed/ for ready-to-render files and raw/ for any new raw
recordings not yet represented in narration.json.
Priority: processed/ files define the segment catalogue.
Raw files discovered in raw/ add new entries pointing at raw/ with
output_file preset to processed/<stem>_processed.mov.
"""
video_extensions = {".mov", ".mp4", ".webm", ".avi", ".mkv", ".m4v"}
processed_dir = narration_dir / "processed"
raw_dir = narration_dir / "raw_mov"
processed_dir.mkdir(parents=True, exist_ok=True)
raw_dir.mkdir(parents=True, exist_ok=True)
# Load / create narration.json
narration_json_path = narration_dir / "narration.json"
existing_narration: dict = {}
if narration_json_path.exists():
existing_narration = _read_json(narration_json_path)
# Remove stale entries: no raw_mov file and source_file also gone
_raw_video_exts_set = {".mov", ".mp4", ".avi", ".mkv", ".m4v"}
raw_stems: set[str] = set()
if raw_dir.exists():
for _f in raw_dir.iterdir():
if _f.is_file() and _f.suffix.lower() in _raw_video_exts_set:
raw_stems.add(_f.stem.lower())
stale_keys = [
seg_id
for seg_id, seg_data in existing_narration.items()
if seg_id.lower() not in raw_stems
and not (narration_dir / seg_data.get("source_file", "")).exists()
]
for seg_id in stale_keys:
del existing_narration[seg_id]
print(f" Removed stale narration segment: {seg_id}")
# Normalise keys to lowercase, merging case-duplicates by keeping the entry
# with more data (skip/take values are the most important thing to preserve).
normalised: dict = {}
merged_count = 0
for seg_id, seg_data in existing_narration.items():
lower_id = seg_id.lower()
if lower_id in normalised:
existing_has_trim = "skip" in normalised[lower_id] or "take" in normalised[lower_id]
incoming_has_trim = "skip" in seg_data or "take" in seg_data
if incoming_has_trim and not existing_has_trim:
normalised[lower_id] = seg_data
merged_count += 1
print(f" Merged duplicate narration segment: '{seg_id}' → '{lower_id}'")
else:
normalised[lower_id] = seg_data
existing_narration = normalised
default_filters = config.default_filters if config else {}
added_count = 0
def _scan(directory: Path) -> list[Path]:
if not directory.exists():
return []
return sorted(
f
for f in directory.iterdir()
if f.is_file()
and f.suffix.lower() in video_extensions
and not f.name.startswith(".")
)
# 1. Scan processed/ — only add entries when NO raw_mov equivalent exists.
# If raw_mov has the source, step 2 will create the entry pointing there
# (with the filter chain), which is better for re-processing later.
_raw_video_exts = {".mov", ".mp4", ".avi", ".mkv", ".m4v"}
for video_file in _scan(processed_dir):
segment_id = video_file.stem
# Strip _processed suffix for cleaner segment IDs if present
if segment_id.endswith("_processed"):
segment_id = segment_id[:-10]
segment_id = segment_id.lower()
if segment_id in existing_narration:
if verbose:
print(f" Skipping {segment_id} (already exists)")
continue
# If a raw_mov equivalent exists, skip — step 2 will handle it
raw_mov_has_file = raw_dir.exists() and any(
(raw_dir / f"{segment_id}{ext}").exists() for ext in _raw_video_exts
)
if raw_mov_has_file:
continue
narration_entry = {
"source_file": f"processed/{video_file.name}",
}
narration_entry["use_audio_channels"] = "auto"
# Loudnorm is applied per-segment during preprocess (not deferred to
# preprocess), so the processed files are already normalized and ready to be
# concatenated directly at render time.
narration_entry["defer_loudnorm"] = False
existing_narration[segment_id] = narration_entry
added_count += 1
print(f" Added narration segment: {segment_id} (from processed/)")
# 2. Scan raw/ — add entries for raw files not yet in narration.json
for video_file in _scan(raw_dir):
segment_id = video_file.stem.lower()
if segment_id in existing_narration:
if verbose:
print(f" Skipping {segment_id} (already exists)")
continue
narration_entry = {
"source_file": f"raw_mov/{video_file.name}",
"output_file": f"processed/{video_file.stem}_processed.mov",
}
if "talkinghead" in default_filters:
narration_entry["cutout"] = "talkinghead"
narration_entry["filter"] = "talkinghead"
narration_entry["use_audio_channels"] = "auto"
# Loudnorm is applied per-segment during preprocess (not deferred to
# preprocess), so the processed files are already normalized and ready to be
# concatenated directly at render time.
narration_entry["defer_loudnorm"] = False
existing_narration[segment_id] = narration_entry
added_count += 1
print(f" Added narration segment: {segment_id} (from raw_mov)")
removed_count = len(stale_keys)
if added_count > 0 or removed_count > 0 or merged_count > 0 or not narration_json_path.exists():
with open(narration_json_path, "w", encoding="utf-8") as f:
json.dump(existing_narration, f, indent=2)
if added_count > 0 or removed_count > 0 or merged_count > 0:
parts = []
if added_count:
parts.append(f"+{added_count}")
if removed_count:
parts.append(f"-{removed_count}")
if merged_count:
parts.append(f"merged {merged_count} duplicate(s)")
print(f" Updated narration.json ({', '.join(parts)} segments)")
else:
if not existing_narration:
print(f" narration.json created (empty — add files to processed/ or raw/)")
else:
print(f" No new narration segments to add")
def _write_youtube_meta(
project_path: Path, config, citations: list[str]
) -> None:
"""Write youtube_meta.txt with project description and collected citations."""
meta_path = project_path / "youtube_meta.txt"
lines: list[str] = []
if config and config.description:
lines.append("== Description ==")
lines.append(config.description)
lines.append("")
if citations:
lines.append("== References ==")
for i, cite in enumerate(citations, 1):
lines.append(f"{i}. {cite}")
lines.append("")
meta_path.write_text("\n".join(lines), encoding="utf-8")
print(f" Wrote {meta_path.name} ({len(citations)} reference(s))")
def _import_presenter_notes(
project_path: Path, keynote_file: Path, verbose: bool, config=None
) -> None:
"""Extract presenter notes from Keynote and write to manuscript.txt.
[cite:...] markers are stripped from the manuscript and collected into
youtube_meta.txt alongside the project description.
Uses the JXA script (extract_keynote_notes.js) to extract notes via osascript.
Also exports slides as PNG images to media/slides/{project_name}/.
Backs up existing manuscript.txt before overwriting.
"""
# osascript is macOS-only; skip gracefully on WSL/Linux/Windows
if shutil.which("osascript") is None:
print(
f" Warning: osascript not available (not macOS) — skipping Keynote import for {keynote_file.name}.",
file=sys.stderr,
)
return
print(f" Extracting presenter notes from {keynote_file.name}...")
# Find the JXA script (in the same directory as this module)
script_dir = Path(__file__).parent
jxa_script = script_dir / "extract_keynote_notes.js"
if not jxa_script.exists():
print(f" Error: JXA script not found at {jxa_script}", file=sys.stderr)
return
# Backup existing manuscript.txt if it exists
manuscript_path = project_path / "manuscript.txt"
if manuscript_path.exists():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = project_path / f"manuscript.txt.{timestamp}.bak"
shutil.copy2(manuscript_path, backup_path)
if verbose:
print(f" Backed up manuscript.txt to {backup_path.name}")
# Slides export directory: {project}/media/slides/{project_name}/
# Use lowercase so the path is consistent on case-sensitive filesystems (WSL/Linux).
slides_dir = project_path / "media" / "slides" / project_path.name.lower()
print(f" Exporting slides to {slides_dir}...")
# Run JXA extractor via osascript (also exports slides)
proc = subprocess.run(
[
"osascript",
"-l",
"JavaScript",
str(jxa_script),
str(keynote_file.resolve()),
str(slides_dir.resolve()),
],
capture_output=True,
text=True,
)
if proc.returncode != 0:
print(f" Error extracting presenter notes:", file=sys.stderr)
print(f" {proc.stderr}", file=sys.stderr)
return
# Parse JSON output from JXA script
try:
notes_data = json.loads(proc.stdout) if proc.stdout.strip() else []
except json.JSONDecodeError as e:
print(f" Error parsing notes JSON: {e}", file=sys.stderr)
return
# Convert to manuscript.txt format, stripping [cite:...] markers
_CITE_RE = re.compile(r"\[cite:([^\]]+)\]")
lines = []
citations: list[str] = []
seen_citations: set[str] = set()
for item in notes_data:
idx = item.get("slide_index")
notes = (item.get("notes") or "").rstrip()
lines.append(f"[S{idx}]")
if notes:
clean_note_lines = []
for note_line in notes.splitlines():
for m in _CITE_RE.finditer(note_line):
cite_text = m.group(1).strip()
if cite_text not in seen_citations:
citations.append(cite_text)
seen_citations.add(cite_text)
cleaned = _CITE_RE.sub("", note_line).strip()
if cleaned:
clean_note_lines.append(cleaned)
if clean_note_lines:
lines.append("\n".join(clean_note_lines))
lines.append("") # blank line between slides
# Write manuscript.txt with Unix line endings (Keynote notes may contain \r\n or \r)
content = "\n".join(lines).rstrip() + "\n"
content = content.replace("\r\n", "\n").replace("\r", "\n")
manuscript_path.write_text(content, encoding="utf-8")
print(f" Wrote {manuscript_path} ({len(notes_data)} slides)")
# Write youtube_meta.txt with description + collected citations
_write_youtube_meta(project_path, config, citations)
if citations and verbose:
for i, cite in enumerate(citations, 1):
print(f" {i}. {cite}")
if verbose:
non_empty = sum(1 for item in notes_data if item.get("notes"))
print(f" {non_empty} slides have presenter notes")
# =============================================================================
# Tasks File
# =============================================================================
_TASKS_VIDEO_PREFIXES = {
"video:": 6,
"vft:": 4,
"vfb:": 4,
"vf2t:": 5,
"vf2b:": 5,
"vst:": 4,
"vsb:": 4,
"vftp:": 5,
"vfbp:": 5,
"vf2tp:": 6,
"vf2bp:": 6,
"vstp:": 5,
"vsbp:": 5,
"narration:": 10,
}
def _collect_missing_video_markers(
markers: list[str], videos: dict
) -> list[tuple[str, str]]:
"""Return (marker_text, video_id) for video markers not defined in videos.json."""
missing = []
seen = set()
for marker in markers:
matched = next((p for p in _TASKS_VIDEO_PREFIXES if marker.startswith(p)), None)
if matched is None:
continue
video_id = marker[_TASKS_VIDEO_PREFIXES[matched] :]
if video_id not in videos and video_id not in seen:
seen.add(video_id)
missing.append((marker, video_id))
return missing
def _write_tasks_file(
project_path: Path,
missing_videos: list[tuple[str, str]],
alignment_issues: list[tuple[str, str]],
) -> None:
"""Write tasks.md to project_path with missing assets and alignment issues."""
tasks_path = project_path / "tasks.md"
today = datetime.now().strftime("%Y-%m-%d")
lines = [
f"# Tasks: {project_path.name}",
f"_Generated: {today}_",
"",
]
if missing_videos:
lines += [
f"## Missing Video Assets ({len(missing_videos)})",
"Referenced in manuscript.txt but not defined in videos.json.",
"",
]
for marker, video_id in missing_videos:
lines.append(f"- [ ] `{video_id}` — referenced as `[{marker}]`")
lines.append("")
if alignment_issues:
lines += [
f"## Slide Alignment Issues ({len(alignment_issues)})",
"Slide markers that could not be matched to the spoken narration (likely adlibbed).",
"",
]
for marker_id, context in alignment_issues:
lines.append(f'- [ ] `{marker_id}` — _"{context}"_')
lines.append("")
if not missing_videos and not alignment_issues:
lines += ["_No outstanding tasks._", ""]
tasks_path.write_text("\n".join(lines), encoding="utf-8")
print(
f" Tasks written → tasks.md"
+ (f" ({len(missing_videos)} missing videos)" if missing_videos else "")
+ (f" ({len(alignment_issues)} alignment issues)" if alignment_issues else "")
)
# =============================================================================
# Pexels Download Command
# =============================================================================
def cmd_pexels(
project_path: Path,
verbose: bool,
search_query: Optional[str] = None,
search_max: int = 200,
) -> int:
"""Download missing Pexels videos and enrich metadata for existing ones."""
from .parser import parse_manuscript, parse_project_config, parse_videos
from .pexels import (
get_pexels_api_key,
find_missing_pexels_videos,
download_video,
update_videos_json,
enrich_missing_descriptions,
search_and_download,
)
api_key = get_pexels_api_key()
if not api_key:
print(
"Error: Pexels API key not configured.\n"
"Add to ~/.gnommo.conf:\n"
" [pexels]\n"
" api_key = YOUR_KEY_HERE\n"
"Get a free key at https://www.pexels.com/api/",
file=sys.stderr,
)
return 1
# --- Search mode: download all results for a query to the assets disk ---
if search_query:
from .cache import load_assets_config
# Find shared_assets: prefer assets disk (LaCie), fall back to local
assets_root = load_assets_config()
if assets_root and assets_root.exists():
pexels_dir = assets_root / "shared_assets" / "pexels"
# Use the assets disk's shared_assets/videos.json as the registry
shared_videos_json = assets_root / "shared_assets" / "videos.json"
print(f"Downloading to: {pexels_dir}")
else:
# No assets disk — find local shared_assets from project_path or cwd
local_shared = _find_shared_assets(project_path)
if not local_shared:
print("Error: shared_assets directory not found.", file=sys.stderr)
return 1
pexels_dir = local_shared / "pexels"
shared_videos_json = local_shared / "videos.json"
print(f"Assets disk not found — downloading to local: {pexels_dir}")
dl, sk = search_and_download(
search_query, pexels_dir, shared_videos_json, api_key, max_results=search_max
)
print(f"\nDone — {dl} downloaded, {sk} already present.")
return 0
shared_assets_dir = _find_shared_assets(project_path)
if not shared_assets_dir:
print("Error: shared_assets directory not found.", file=sys.stderr)
return 1
shared_videos_json = shared_assets_dir / "videos.json"
# --- Normal mode: download missing files referenced in the manuscript ---
config = parse_project_config(project_path)
_, markers, _, _ = parse_manuscript(project_path)
videos, _ = parse_videos(project_path, config)
local_videos_json = project_path / config.videos_path
# 1. Collect missing files: manuscript markers + background
missing = find_missing_pexels_videos(markers, videos, shared_assets_dir)
# Also check background video from project.json
if config and config.background and config.background.startswith("pexels/"):
from .cache import resolve_with_cache
bg_id = config.background
shared_vids_raw = _read_json(shared_videos_json) if shared_videos_json.exists() else {}
bg_entry = shared_vids_raw.get(bg_id)
if bg_entry:
bg_sf = bg_entry.get("source_file", "")
bg_path = shared_assets_dir / bg_sf
resolved, _ = resolve_with_cache(bg_path, shared_assets_dir)
if not resolved.exists():
# Only add if not already in the manuscript-marker list
if not any(vid_id == bg_id for vid_id, _ in missing):
missing.append((bg_id, bg_sf))
failed = 0
if missing:
print(f"Downloading {len(missing)} missing Pexels video(s)...")
for video_id, source_file in missing:
meta = download_video(source_file, shared_assets_dir, api_key)
if meta is None:
failed += 1
continue
for json_path in (local_videos_json, shared_videos_json):
update_videos_json(json_path, video_id, meta)
if failed:
print(f"\n {failed}/{len(missing)} download(s) failed.")
else:
print(f"\n {len(missing)} video(s) downloaded.")
else:
print("No missing Pexels videos.")
# 2. Enrich descriptions for existing files that have none
enrich_missing_descriptions(shared_assets_dir, api_key)
return 1 if failed else 0
# =============================================================================
# Validate Command
# =============================================================================
def cmd_validate(project_path: Path, verbose: bool) -> int:
"""Validate project configuration."""
from .parser import (
parse_audio,
parse_manuscript,
parse_project_config,
parse_slides,
parse_videos,
)
from .validator import validate_project
print(f"Validating: {project_path.name}")
if not (project_path / "project.json").exists():
print(f"Error: project.json not found in {project_path}", file=sys.stderr)
return 1
# Parse all files
_, markers, malformed, _ = parse_manuscript(project_path)
config = parse_project_config(project_path)
slides = parse_slides(project_path, config)
videos, videos_dir = parse_videos(project_path, config)
audio, _ = parse_audio(project_path, config)
if verbose:
print(f" - Markers in manuscript: {len(markers)}")
print(f" - Slides defined: {len(slides)}")
print(f" - Videos defined: {len(videos)}")
print(f" - Audio cues defined: {len(audio)}")
# Validate
warnings = validate_project(
project_path, markers, config, slides, videos, videos_dir, malformed, audio
)
for w in warnings:
print(f" Warning: {w}")
# Write tasks file (missing assets only — no alignment data at validate time)
missing_videos = _collect_missing_video_markers(markers, videos)
_write_tasks_file(project_path, missing_videos, alignment_issues=[])
print("Validation passed.")
return 0
# =============================================================================
# Preprocess Command
# =============================================================================
def _resolve_process_cache(project_path: Path, config) -> Optional[Path]:
"""Return per-project cache dir on an external disk, or None if none is accessible.
Resolution order:
1. config.process_cache (project.json) — e.g. /Volumes/GnommoDisk/gnommocache
Must already exist (disk must be mounted and dir pre-created).
2. [assets] process cache (~/.gnommo.conf) — derived as <assets_path>cache
e.g. /Volumes/LaCie Jens/Projects/gnommocache.
Only requires the parent to exist (volume mounted); the cache dir
itself is created on first use by the caller.
"""
from .cache import load_assets_process_cache
# Primary: explicitly configured in project.json
if config and config.process_cache:
p = Path(config.process_cache)
if not p.is_absolute():
p = (project_path / p).resolve()
if p.exists():
return p / project_path.name
# Fallback: [assets] disk (need only the parent volume/dir to be present)
assets_cache = load_assets_process_cache()
if assets_cache and assets_cache.parent.exists():
return assets_cache / project_path.name
return None
def cmd_new(project_path: Path, verbose: bool) -> int:
"""Create a new gnommo project with standard folder structure and a project.json template."""
project_name = project_path.name
project_id = project_name
if project_path.exists() and list(project_path.iterdir()):
print(f"Initialising project: {project_path} (folder exists, filling in missing structure)")
else:
print(f"Creating new project: {project_path}")
# ------------------------------------------------------------------ #
# Directories #
# ------------------------------------------------------------------ #
dirs = [
project_path,
project_path / "media" / "videos",
project_path / "media" / "audio",
project_path / "media" / "narration" / "raw_mov",
project_path / "media" / "slides",
project_path / "out",
]
for d in dirs:
d.mkdir(parents=True, exist_ok=True)
if verbose:
print(f" mkdir {d.relative_to(project_path.parent)}")
# ------------------------------------------------------------------ #
# Copy talkinghead filter from the nearest sibling project #
# ------------------------------------------------------------------ #
talkinghead_filter = None
for sibling in sorted(project_path.parent.iterdir()):
if sibling == project_path or not sibling.is_dir():
continue
sib_json = sibling / "project.json"
if sib_json.exists():
try:
sib_cfg = json.loads(sib_json.read_text(encoding="utf-8"))
talkinghead_filter = (sib_cfg.get("default_filters") or {}).get("talkinghead")
if talkinghead_filter:
print(f" Copied talkinghead filter from: {sibling.name}/project.json")
break
except (json.JSONDecodeError, OSError):
pass
if not talkinghead_filter:
# Sensible placeholder — user should tweak gnommokey values for their camera
talkinghead_filter = [
{
"type": "audio_normalize",
"compress": False,
"normalize": True,
"target_lufs": -14,
"target_lra": 11,
"target_tp": -1.5,
},
{
"type": "gnommokey",
"screen_color": [81, 137, 65],
"screen_gain": 175,
"screen_balance": 58,
"despill_bias": [217, 240, 255],
"despill_strength": 5.0,
"edge_erode": 1.0,
"clip_black": 0,
"clip_white": 100,
},
{
"type": "color_grade",
"saturation": 0.95,
"contrast": 1.06,
"rm": -0.05,
"gm": 0.02,
"bm": -0.04,
"curves_master": "0/0.02 0.5/0.5 1/0.97",
},
{
"type": "mask",
"left": 0.05,
"right": 0.1,
"top": 0.1,
"bottom": 0.0,
},
]
print(" Using default talkinghead filter (adjust gnommokey values for your camera)")
# ------------------------------------------------------------------ #
# project.json #
# ------------------------------------------------------------------ #
project_json_path = project_path / "project.json"
if not project_json_path.exists():
project_data = {
"id": project_id,
"name": "",
"description": "",
"platform_targets": ["youtube"],
"status": "scripted",
"resolution": [1920, 1080],
"fps": 30,
"manuscript": "manuscript.txt",
"videos": "media/videos/videos.json",
"narration": "media/narration/narration.json",
"slides": f"media/slides/{project_id}/slides.json",
"audio": "media/audio/audio.json",
"output_video": f"{project_id}.mp4",
"default_filters": {"talkinghead": talkinghead_filter},
"cutouts": {
"talkinghead": {"x": "-10%", "y": "40%", "height": "80%"},
"square": {"x": "46.5%", "y": "4.5%", "width": "50%", "height": "90%"},
"fullscreen": {"x": "0%", "y": "0%", "height": "100%"},
"fullscreen2": {"x": "10%", "y": "7%", "height": "80%"},
},
}
project_json_path.write_text(
json.dumps(project_data, indent=2, ensure_ascii=False), encoding="utf-8"
)
print(" Created: project.json")
else:
print(" Skipped: project.json (already exists)")
# ------------------------------------------------------------------ #
# Stub JSON files #
# ------------------------------------------------------------------ #
stubs: dict[str, object] = {
"media/videos/videos.json": {},
"media/audio/audio.json": {},
"media/narration/narration.json": {},
}
for rel, content in stubs.items():
p = project_path / rel
if not p.exists():
p.write_text(json.dumps(content, indent=2), encoding="utf-8")
print(f" Created: {rel}")
# ------------------------------------------------------------------ #
# Manuscript template #
# ------------------------------------------------------------------ #
manuscript_path = project_path / "manuscript.txt"
if not manuscript_path.exists():
manuscript_path.write_text(
"[S1]\nYour narration for slide 1 goes here.\n\n[S2]\n\n",
encoding="utf-8",
)
print(" Created: manuscript.txt")
# ------------------------------------------------------------------ #
# Instructions #
# ------------------------------------------------------------------ #
print(f"""
Done. Here is what to do next:
1. Place your Keynote presentation (.key) in the project folder:
{project_path}/
2. Record your talking head segments using a teleprompter.
Name each recording after the slide range it covers:
S1-10.mov covers slides 1 10
S11-32.mov covers slides 11 32
S33-end.mov covers slides 33 to the end
Place the recordings in:
{project_path}/media/narration/raw_mov/
3. Edit manuscript.txt so the spoken words appear under the right [SN] marker.
Add [vfb:Logo6sec] or other video markers where needed.
4. Run the pipeline step by step:
gnommo -p {project_name} import # extract slides from Keynote
gnommo -p {project_name} pre # chroma key + audio normalise
gnommo -p {project_name} trim # auto-detect skip/take per segment
gnommo -p {project_name} render # produce the final video
Or run everything in one go:
gnommo -p {project_name} all
Output: {project_path}/out/{project_id}.mp4
""")
return 0
def cmd_clear(project_path: Path, verbose: bool) -> int:
"""Delete preprocessed outputs so that 'preprocess' re-runs them from scratch.
Removes *_processed.mov files from the processed/ directory (or the
process cache on the external disk if one is configured). narration.json
skip/take values and raw source files are NOT touched.
"""
from .parser import parse_project_config
print(f"Clearing preprocessed outputs: {project_path.name}")
config = parse_project_config(project_path)
narration_dir = project_path / "media" / "narration"
cache_root = _resolve_process_cache(project_path, config)
if cache_root:
processed_dir = cache_root / "media" / "narration" / "processed"
print(f" Cache: {processed_dir}")
else:
processed_dir = narration_dir / "processed"
print(f" Local: {processed_dir}")
if not processed_dir.exists():
print(" Nothing to clear — processed/ directory does not exist.")
return 0
candidates = sorted(
f for f in processed_dir.iterdir()
if f.is_file() and "_processed" in f.stem
)
if not candidates:
print(" Nothing to clear — no *_processed.* files found.")
return 0
total_bytes = 0
for f in candidates:
size = f.stat().st_size
total_bytes += size
print(f" Deleting {f.name} ({size / 1e9:.2f} GB)")
f.unlink()
print(f"\n Cleared {len(candidates)} file(s), freed {total_bytes / 1e9:.2f} GB.")
print(" Run 'gnommo preprocess' (without --force) to reprocess selectively.")
return 0
# =============================================================================
# Prune Command
# =============================================================================
# Marker prefixes that reference a videos.json entry, mapped to the length of the
# prefix (so marker[length:] is the video ID). Kept in sync with validator.py.
_PRUNE_VIDEO_PREFIXES = {
"video:": 6,
"vft:": 4, "vfb:": 4, "vfm:": 4,
"vf2t:": 5, "vf2b:": 5, "vf2m:": 5,
"vst:": 4, "vsb:": 4, "vsm:": 4,
"vftp:": 5, "vfbp:": 5, "vfmp:": 5,
"vf2tp:": 6, "vf2bp:": 6, "vf2mp:": 6,
"vstp:": 5, "vsbp:": 5, "vsmp:": 5,
}
def _detect_json_indent(path: Path, default: int = 2) -> int:
"""Return the indentation width of the first indented line in a JSON file."""
try:
for line in path.read_text(encoding="utf-8").splitlines():
stripped = line.lstrip(" ")
if stripped and stripped != line:
return len(line) - len(stripped)
except OSError:
pass
return default
def _write_json_preserve(path: Path, data: dict) -> None:
"""Rewrite a JSON manifest, preserving its existing indentation style."""
indent = _detect_json_indent(path)
text = json.dumps(data, indent=indent, ensure_ascii=False)
try:
trailing_nl = path.read_text(encoding="utf-8").endswith("\n")
except OSError:
trailing_nl = True
path.write_text(text + ("\n" if trailing_nl else ""), encoding="utf-8")
def _prune_manifest(
path: Path,
keep_ids: set[str],
label: str,
lowercase_keys: bool,
verbose: bool,
dry_run: bool,
) -> int:
"""Remove entries from a videos.json/audio.json manifest whose ID isn't in keep_ids.
Returns the number of entries removed (or that would be removed, in dry-run mode).
"""
if not path.exists():
return 0
try:
data = _read_json(path)
except json.JSONDecodeError as e:
print(f" Skipping {path.name}: invalid JSON ({e})")
return 0
if not isinstance(data, dict):
return 0
to_remove = [
key
for key in data
if (key.lower() if lowercase_keys else key) not in keep_ids
]
# Case-collision duplicates: when keys are case-insensitive (videos), several
# keys can share the same lowercase form (e.g. "KnightRotating" and
# "knightrotating"). Markers resolve by lowercased id, so only the exact
# lowercase key is ever reachable — the other case-variants are dead weight.
# Collapse each surviving collision group to its lowercase canonical.
dup_remove: list[tuple[str, str]] = [] # (removed_key, kept_canonical_key)
if lowercase_keys:
remove_set = set(to_remove)
groups: dict[str, list[str]] = {}
for key in data:
if key in remove_set:
continue
groups.setdefault(key.lower(), []).append(key)
for low, keys in groups.items():
if len(keys) < 2:
continue
# Keep the already-lowercase key (what the renderer resolves); else
# the first-seen, so at least one entry survives.
canonical = next((k for k in keys if k == low), keys[0])
for key in keys:
if key != canonical:
dup_remove.append((key, canonical))
if not to_remove and not dup_remove:
if verbose:
print(f" {path.name}: all {len(data)} {label} entries in use.")
return 0
if to_remove:
print(f" {path.name}: removing {len(to_remove)} unused {label} entr"
f"{'y' if len(to_remove) == 1 else 'ies'}:")
for key in to_remove:
print(f" - {key}")
if dup_remove:
print(f" {path.name}: removing {len(dup_remove)} case-duplicate {label} entr"
f"{'y' if len(dup_remove) == 1 else 'ies'}:")
for key, canonical in dup_remove:
print(f" - {key} (duplicate of '{canonical}')")
if not dry_run:
for key in to_remove:
del data[key]
for key, _ in dup_remove:
del data[key]
_write_json_preserve(path, data)
return len(to_remove) + len(dup_remove)
def _prune_narration(project_path: Path, verbose: bool, dry_run: bool) -> int:
"""Remove narration.json entries whose raw source file is missing from raw_mov/.
Returns the number of entries removed (or that would be removed, in dry-run mode).
"""
narration_dir = project_path / "media" / "narration"
narration_json = narration_dir / "narration.json"
if not narration_json.exists():
return 0
try:
data = _read_json(narration_json)
except json.JSONDecodeError as e:
print(f" Skipping narration.json: invalid JSON ({e})")
return 0
if not isinstance(data, dict):
return 0
raw_dir = narration_dir / "raw_mov"
to_remove = []
for seg_id, entry in data.items():
source_file = entry.get("source_file") if isinstance(entry, dict) else None
if source_file:
exists = (narration_dir / source_file).exists()
else:
# No source_file recorded — look for a matching file in raw_mov/
exists = bool(list(raw_dir.glob(f"{seg_id}.*"))) if raw_dir.exists() else False
if not exists:
to_remove.append(seg_id)
if not to_remove:
if verbose:
print(f" narration.json: all {len(data)} entries have a raw_mov/ source.")
return 0
print(f" narration.json: removing {len(to_remove)} entr"
f"{'y' if len(to_remove) == 1 else 'ies'} with no raw_mov/ source:")
for seg_id in to_remove:
print(f" - {seg_id}")
if not dry_run:
for seg_id in to_remove:
del data[seg_id]
_write_json_preserve(narration_json, data)
return len(to_remove)
def cmd_prune(project_path: Path, verbose: bool, dry_run: bool) -> int:
"""Remove unused entries from videos.json, audio.json, and narration.json.
- videos.json / audio.json: drop entries whose IDs are never referenced by a
marker in manuscript.txt. Videos referenced by project.json (outro sequence,
main_video) are kept even when no marker triggers them.
- narration.json: drop entries whose raw source file is missing from
media/narration/raw_mov/.
Only JSON manifest entries are removed — media files on disk are never touched.
Use --dry-run to preview.
"""
from .parser import parse_manuscript, parse_project_config
suffix = " (dry run)" if dry_run else ""
print(f"Pruning manifests: {project_path.name}{suffix}")
config = parse_project_config(project_path)
# --- Collect IDs referenced by the script (manuscript markers) ---
_, markers, _, _ = parse_manuscript(project_path)
referenced_videos: set[str] = set()
referenced_audio: set[str] = set()
for marker in markers:
prefix = next((p for p in _PRUNE_VIDEO_PREFIXES if marker.startswith(p)), None)
if prefix is not None:
referenced_videos.add(marker[_PRUNE_VIDEO_PREFIXES[prefix]:].lower())
elif marker.startswith("narration:"):
referenced_videos.add(marker[10:].lower())
elif marker.startswith("audio:"):
referenced_audio.add(marker[6:])
elif marker.startswith("A") and len(marker) > 1 and marker[1:].isalnum():
referenced_audio.add(marker[1:])
# Videos referenced by project.json (outro sequence, main video, background)
# are kept even though no manuscript marker triggers them.
def _add_video(value) -> None:
if isinstance(value, str) and value:
referenced_videos.add(value.lower())
elif isinstance(value, list):
for item in value:
_add_video(item)
_add_video(config.outro)
_add_video(config.main_video)
_add_video(config.background)
_add_video(config.background_video)
total_removed = 0
total_removed += _prune_manifest(
project_path / config.videos_path,
referenced_videos,
"video",
lowercase_keys=True,
verbose=verbose,
dry_run=dry_run,
)
total_removed += _prune_manifest(
project_path / config.audio_path,
referenced_audio,
"audio",
lowercase_keys=False,
verbose=verbose,
dry_run=dry_run,
)
total_removed += _prune_narration(project_path, verbose, dry_run)
if total_removed == 0:
print(" Nothing to prune — all manifest entries are in use.")
elif dry_run:
print(f"\n Would remove {total_removed} entr"
f"{'y' if total_removed == 1 else 'ies'}. "
f"Re-run without --dry-run to apply.")
else:
print(f"\n Removed {total_removed} entr"
f"{'y' if total_removed == 1 else 'ies'}.")
return 0
def cmd_preprocess(
project_path: Path,
verbose: bool,
dry_run: bool,
force: bool = False,
workers: int = 1,
res: str = "full",
) -> int:
"""Run preprocessing pipeline on narration segments and videos.
Discovers source files directly from raw_mov/ (preferred) or raw_mp4/
(fallback when raw_mov/ is empty). Does NOT require narration.json to
exist — it writes/updates narration.json after processing.
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
from .parser import parse_project_config, parse_videos
from .preprocessor import preprocess_video, RES_CONFIGS
from .models import VideoSource as _VideoSource
mode_str = f" ({res.upper()})" if res != "full" else ""
print(f"Preprocessing narration: {project_path.name}{mode_str}")
config = parse_project_config(project_path)
# Narration directory — source files always in project media/narration/
narration_dir = project_path / "media" / "narration"
narration_dir.mkdir(parents=True, exist_ok=True)
raw_dir = narration_dir / "raw_mov"
compressed_dir = narration_dir / "raw_mp4"
# process_cache: write processed outputs to external disk to save laptop space
cache_root = _resolve_process_cache(project_path, config)
if cache_root:
# Mirror the project's media/ structure so GnommoCache (resolve_with_cache)
# finds these files transparently during render.
cache_narration_dir = cache_root / "media" / "narration"
cache_narration_dir.mkdir(parents=True, exist_ok=True)
(cache_narration_dir / "processed").mkdir(parents=True, exist_ok=True)
print(f" Using process cache: {cache_root}")
else:
cache_narration_dir = None
processed_dir = (cache_narration_dir or narration_dir) / "processed"
processed_dir.mkdir(parents=True, exist_ok=True)
# Remove any .tmp files left by a previously interrupted preprocess run.
# These are partial outputs that would block the segment from being reprocessed.
for stale_tmp in processed_dir.glob("*.tmp"):
print(f" Removing incomplete output from previous run: {stale_tmp.name}")
stale_tmp.unlink()
# Resolve intermediate directory
gnommo_scratch = None
if config.gnommo_scratch:
gnommo_scratch = Path(config.gnommo_scratch)
if not gnommo_scratch.is_absolute():
gnommo_scratch = project_path / gnommo_scratch
print(f" Using intermediate dir: {gnommo_scratch}")
# --- Filter pipeline ---
talkinghead_filter = (config.default_filters or {}).get("talkinghead", [])
if not talkinghead_filter:
print(
" ERROR: No 'talkinghead' filter defined in project.json default_filters."
)
print(" Add a 'talkinghead' entry under 'default_filters' in project.json.")
return 1
# --- Source discovery ---
_video_exts = {".mov", ".mp4", ".avi", ".mkv", ".m4v"}
def _scan_dir(d: Path) -> list[Path]:
if not d.exists():
return []
return sorted(
f
for f in d.iterdir()
if f.is_file()
and f.suffix.lower() in _video_exts
and not f.name.startswith(".")
)
raw_mov_files = _scan_dir(raw_dir)
raw_mp4_files = _scan_dir(compressed_dir)
if raw_mov_files:
source_files = raw_mov_files
using_compressed = False
elif raw_mp4_files:
source_files = raw_mp4_files
using_compressed = True
print(
" WARNING: raw_mov/ is empty — using compressed files from raw_mp4/ instead. Quality may be reduced."
)
else:
print(f" No source files found in raw_mov/ or raw_mp4/.")
print(f" Place .mov recordings in {raw_dir}")
return 1
# --- Load existing narration.json to preserve per-segment settings ---
narration_json_path = narration_dir / "narration.json"
existing_narration: dict = {}
if narration_json_path.exists():
existing_narration = _read_json(narration_json_path)
# --- Build segments list ---
# Per-segment staleness: reprocess a segment when its raw source's fingerprint
# differs from the one recorded the last time we produced its output. An
# existing output with no recorded fingerprint is adopted (recorded, not
# reprocessed) so introducing state never triggers a needless re-encode.
from . import state as _state
_stage_key = f"preprocess:{res}"
_prev_fps = _state.get_items(project_path, _stage_key)
_seg_source_fp: dict[str, str] = {} # segment_id -> current source fingerprint
_adopted_fps: dict[str, str] = {} # up-to-date segments to (re)record
segments_to_process: list[tuple[str, _VideoSource]] = []
skipped_count = 0
for source_file in source_files:
segment_id = source_file.stem
# Strip _compressed suffix (raw_mp4 naming convention)
if using_compressed and segment_id.endswith("_compressed"):
segment_id = segment_id[: -len("_compressed")]
# For non-full res, write processed outputs into the res subdir.
# (narration.json still records the plain "processed/..." path.)
_res_cfg = RES_CONFIGS.get(res) if res != "full" else None
if _res_cfg:
_, _, _subdir = _res_cfg
output_file = f"{_subdir}/processed/{segment_id}_processed.mov"
else:
output_file = f"processed/{segment_id}_processed.mov"
# When process_cache is set, output goes to the cache dir; narration.json
# still records the relative path so render (also using cache) can find it.
output_base = cache_narration_dir or narration_dir
output_path = output_base / output_file
current_fp = _state.fingerprint_path(source_file, _state.META)
_seg_source_fp[segment_id] = current_fp
if output_path.exists() and not force:
recorded = _prev_fps.get(segment_id)
if recorded is None or recorded == current_fp:
# Up to date (or adopting a pre-existing output into state).
print(f" {segment_id}: output exists, skipping (use --force to reprocess)")
skipped_count += 1
_adopted_fps[segment_id] = current_fp
continue
print(f" {segment_id}: raw source changed since last run — reprocessing")
# Filter: from existing narration.json entry (if explicitly set), else talkinghead
existing_entry = existing_narration.get(segment_id, {})
raw_filter = existing_entry.get("filter")
if raw_filter:
if isinstance(raw_filter, str):
filter_list = (config.default_filters or {}).get(
raw_filter, talkinghead_filter
)
else:
filter_list = raw_filter
else:
filter_list = talkinghead_filter
video_source = _VideoSource(
source_file=source_file,
filter=filter_list,
output_file=output_file,
use_audio_channels=existing_entry.get("use_audio_channels", "auto"),
# Default False: apply loudnorm now (in preprocess), not deferred.
defer_loudnorm=existing_entry.get("defer_loudnorm", False),
)
segments_to_process.append((segment_id, video_source))
if not segments_to_process:
if skipped_count:
print(
f"\n All {skipped_count} segment(s) already preprocessed. Use --force to reprocess."
)
else:
print("\n No segments to preprocess.")
return 0
if dry_run:
for segment_id, segment_source in segments_to_process:
print(f"\n Would preprocess: {segment_id}")
print(f" Source: {segment_source.source_file}")
print(f" Output: {segment_source.output_file}")
print(f" Filters: {len(segment_source.filter)} step(s)")
return 0
# --- Process segments ---
successfully_processed: list[tuple[str, _VideoSource]] = []
if workers > 1 and len(segments_to_process) > 1:
num_workers = min(workers, len(segments_to_process))
print(
f"\n Processing {len(segments_to_process)} segments in parallel ({num_workers} workers)"
)
def process_segment_task(task):
seg_id, seg_source = task
preprocess_video(
cache_narration_dir or narration_dir,
seg_id,
seg_source,
verbose=False,
force=force,
custom_gnommo_scratch=gnommo_scratch,
res=res,
)
return task
completed = 0
with ThreadPoolExecutor(max_workers=num_workers) as executor:
futures = {
executor.submit(process_segment_task, t): t for t in segments_to_process
}
for future in as_completed(futures):
seg_id, seg_source = future.result()
completed += 1
print(f" Completed: {seg_id} ({completed}/{len(segments_to_process)})")
output_path = (
cache_narration_dir or narration_dir
) / seg_source.output_file
if output_path.exists():
successfully_processed.append((seg_id, seg_source))
else:
for segment_id, segment_source in segments_to_process:
_out_full = (
cache_narration_dir or narration_dir
) / segment_source.output_file
print(f"\n Processing: {segment_id}")
print(f" Source: {segment_source.source_file}")
print(f" Output: {_out_full}")
print(f" Filters: {len(segment_source.filter)} step(s)")
preprocess_video(
cache_narration_dir or narration_dir,
segment_id,
segment_source,
verbose,
force,
gnommo_scratch,
res=res,
)
output_path = (
cache_narration_dir or narration_dir
) / segment_source.output_file
if output_path.exists():
successfully_processed.append((segment_id, segment_source))
# --- Update narration.json ---
# Write processed segments; preserve any existing per-segment settings (skip/take/etc.)
_PRESERVE_KEYS = (
"skip",
"take",
"begin",
"end",
"cutout",
"use_audio_channels",
"defer_loudnorm",
"volume",
"zoom",
)
for segment_id, segment_source in successfully_processed:
existing_entry = existing_narration.get(segment_id, {})
entry: dict = {}
# Preserve settings the user may have set (trim points, cutout, etc.)
for key in _PRESERVE_KEYS:
if key in existing_entry:
entry[key] = existing_entry[key]
# Always record the plain path; the res subdir shift happens at render for low/tiny.
entry["source_file"] = f"processed/{segment_id}_processed.mov"
entry.setdefault("use_audio_channels", "auto")
entry.setdefault("defer_loudnorm", False)
existing_narration[segment_id] = entry
with open(narration_json_path, "w", encoding="utf-8") as f:
json.dump(existing_narration, f, indent=2)
# Record source fingerprints for every segment whose output is now present
# (freshly processed + adopted skips) so future runs can detect raw changes.
_record_fps = dict(_adopted_fps)
for _seg_id, _ in successfully_processed:
_fp = _seg_source_fp.get(_seg_id)
if _fp is not None:
_record_fps[_seg_id] = _fp
if _record_fps:
_state.record_items(project_path, _stage_key, _record_fps)
if successfully_processed:
print(f"\n Updated narration.json ({len(successfully_processed)} segment(s))")
print(
f"\n Narration segments are concatenated automatically at render time — run 'gnommo -p <project> render'."
)
# Also preprocess videos from videos.json (e.g. chroma key, color grade)
videos, videos_dir = parse_videos(project_path, config)
videos_to_process = [
(vid_id, vid_src)
for vid_id, vid_src in videos.items()
if vid_src.filter and not vid_src.is_shared
]
if videos_to_process:
print(f"\n Processing {len(videos_to_process)} video(s) from videos.json:")
for video_id, video_source in videos_to_process:
if video_source.output_file:
output_path = videos_dir / video_source.output_file
if output_path.exists() and not force:
print(
f" {video_id}: output exists, skipping (use --force to reprocess)"
)
continue
if dry_run:
print(
f" Would preprocess: {video_id} ({len(video_source.filter)} filter(s))"
)
continue
print(f" Processing: {video_id}")
preprocess_video(
videos_dir,
video_id,
video_source,
verbose,
force,
gnommo_scratch,
res=res,
)
print("\nPreprocessing complete.")
return 0
# =============================================================================
# Trim Command — transcript-based trimming for slide-range segments
# =============================================================================
# Words so common they're useless for matching slide boundaries.
_TRIM_STOP_WORDS = frozenset({
"a", "an", "the", "and", "or", "but", "in", "on", "at", "to", "of",
"for", "is", "it", "its", "we", "our", "this", "that", "with", "be",
"are", "was", "has", "not", "so", "as", "by", "do", "if", "up",
"you", "i", "he", "she", "they", "them", "just", "now", "can",
"than", "then", "from", "about", "into", "out", "what", "there",
"when", "how", "who", "all", "very", "also", "more", "get", "have",
})
# Transcript-based auto-trim padding: keep this much lead-in before the start
# slide's first word, and this much tail after the end slide's last word.
_TRIM_LEAD_IN = 0.5
_TRIM_TAIL_OUT = 2.0
def _user_begin_skip(existing: dict) -> "float | None":
"""Return the skip (seconds) implied by a user-pinned begin/start, or None."""
from .parser import parse_timestamp
if existing.get("begin"):
return parse_timestamp(existing["begin"])
if existing.get("start"):
return parse_timestamp(existing["start"])
return None
def _parse_segment_slide_range(seg_id: str) -> "tuple[int, int | None] | None":
"""Parse 'S11-24' → (11, 24), 'S1-end' → (1, None), else None."""
m = re.match(r"^s(\d+)-(?:s?(\d+)|(end))$", seg_id.lower())
if not m:
return None
start = int(m.group(1))
end = int(m.group(2)) if m.group(2) else None
return (start, end)
def _extract_slide_texts(manuscript_path: Path) -> "dict[int, str]":
"""Return {slide_num: text} for each [SN] marker in the manuscript."""
text = manuscript_path.read_text(encoding="utf-8")
parts = re.split(r"\[S(\d+)\]", text)
result: dict[int, str] = {}
i = 1
while i + 1 < len(parts):
slide_num = int(parts[i])
content = re.sub(r"\[[^\]]+\]", " ", parts[i + 1])
result[slide_num] = content.strip()
i += 2
return result
def _trim_content_words(text: str) -> "list[str]":
"""Lowercase words stripped of punctuation, excluding stop words."""
words = re.findall(r"[a-zA-Z0-9']+", text.lower())
return [w for w in words if w not in _TRIM_STOP_WORDS and len(w) > 2]
def _find_slide_end_in_transcript(
transcript_words: list,
slide_text: str,
verbose: bool = False,
) -> "float | None":
"""
Locate where slide_text ends in the transcript and return that word's
end timestamp. Searches the tail of the transcript for the last few
content words from slide_text using a sequential fuzzy match.
Returns None if no confident match is found.
"""
target = _trim_content_words(slide_text)[-12:]
if not target:
return None
# Normalise transcript words (strip punctuation, lowercase)
norm = [re.sub(r"[^a-z0-9']", "", w.word.lower()) for w in transcript_words]
# Index into transcript of content words only
content_idxs = [
i for i, w in enumerate(norm)
if w and w not in _TRIM_STOP_WORDS and len(w) > 2
]
if not content_idxs:
return None
n = len(target)
threshold = max(1, int(n * 0.55))
# Scan backwards: find the latest window of n content words that matches,
# and return the end timestamp of the LAST matched word — not the window's
# last word, which may be trailing filler after the slide's actual last word.
for end_ci in range(len(content_idxs) - 1, n - 2, -1):
window_idxs = content_idxs[max(0, end_ci - n + 1): end_ci + 1]
window = [norm[i] for i in window_idxs]
# Sequential match: iterate target left-to-right, consume window matches
score = 0
wi = 0
last_match_wi = None
for t_word in target:
while wi < len(window):
w = window[wi]
matched = w == t_word or (len(w) >= 4 and len(t_word) >= 4 and w[:4] == t_word[:4])
wi += 1
if matched:
last_match_wi = wi - 1
score += 1
break
if score >= threshold and last_match_wi is not None:
last_tw = transcript_words[window_idxs[last_match_wi]]
if verbose:
print(f"\n → matched slide end: '{last_tw.word}' at {last_tw.end:.2f}s")
return last_tw.end
if verbose:
print(f"\n → could not match slide end (target tail: {target[-5:]})")
return None
def _find_slide_start_in_transcript(
transcript_words: list,
slide_text: str,
verbose: bool = False,
) -> "float | None":
"""
Locate where slide_text begins in the transcript and return that word's
start timestamp. Searches the head of the transcript for the first few
content words from slide_text using a sequential fuzzy match (the mirror of
_find_slide_end_in_transcript). Returns None if no confident match is found.
"""
target = _trim_content_words(slide_text)[:12]
if not target:
return None
norm = [re.sub(r"[^a-z0-9']", "", w.word.lower()) for w in transcript_words]
content_idxs = [
i for i, w in enumerate(norm)
if w and w not in _TRIM_STOP_WORDS and len(w) > 2
]
if not content_idxs:
return None
n = len(target)
threshold = max(1, int(n * 0.55))
# Scan forwards: find the earliest window of n content words that matches,
# and return the timestamp of the FIRST matched word (not the window start —
# the target sequence may begin partway into the window, after filler).
for start_ci in range(len(content_idxs)):
window_idxs = content_idxs[start_ci: start_ci + n]
if len(window_idxs) < threshold:
break
window = [norm[i] for i in window_idxs]
score = 0
wi = 0
first_match_wi = None
for t_word in target:
while wi < len(window):
w = window[wi]
matched = w == t_word or (len(w) >= 4 and len(t_word) >= 4 and w[:4] == t_word[:4])
wi += 1
if matched:
if first_match_wi is None:
first_match_wi = wi - 1
score += 1
break
if score >= threshold and first_match_wi is not None:
first_tw = transcript_words[window_idxs[first_match_wi]]
if verbose:
print(f"\n → matched slide start: '{first_tw.word}' at {first_tw.start:.2f}s")
return first_tw.start
if verbose:
print(f"\n → could not match slide start (target head: {target[:5]})")
return None
def cmd_trim(
project_path: Path,
verbose: bool,
force: bool = False,
threshold_db: float = -40.0,
res: str = "full",
whisper_model: str = "base",
) -> int:
"""
Auto-detect skip/take for narration segments and write them into narration.json.
Trim only fills in the side(s) the user hasn't pinned. A user-set `begin`/
`start` pins the beginning; a user-set `end` pins the end — those are always
respected (even under --force). Only the un-pinned side is auto-detected:
For segments named S{N}-{M}.mov or S{N}-end.mov (transcript-based):
- begin = start of slide N's first word 0.5s (falls back to first word)
- end = end of slide M's last word + 2.0s (S{N}-end: last spoken word + 2.0s)
- Transcripts are cached in narration/transcripts/{seg_id}.json
For other segments: falls back to silence detection (also honouring pins).
"""
from .parser import parse_project_config, parse_narration, parse_timestamp
from .preprocessor import detect_silence_bounds, get_video_duration
print(f"Trimming narration: {project_path.name}")
config = parse_project_config(project_path)
narration, narration_dir = parse_narration(project_path, config)
if not narration:
print(" No narration segments found in narration.json")
print(" Run 'gnommo -p <project> import' first.")
return 1
# Load slide texts from manuscript for transcript-based trimming
manuscript_path = project_path / "manuscript.txt"
slide_texts: dict[int, str] = {}
if manuscript_path.exists():
slide_texts = _extract_slide_texts(manuscript_path)
_video_exts = {".mov", ".mp4", ".avi", ".mkv", ".m4v"}
raw_dir = narration_dir / "raw_mov"
compressed_dir = narration_dir / "raw_mp4"
transcripts_dir = narration_dir / "transcripts"
# Build lookup of raw files keyed by lowercased stem
raw_lookup: dict[str, Path] = {}
for search_dir in (raw_dir, compressed_dir):
if search_dir.exists():
for f in search_dir.iterdir():
if (
f.is_file()
and f.suffix.lower() in _video_exts
and not f.name.startswith(".")
):
stem = f.stem
if stem.endswith("_compressed"):
stem = stem[: -len("_compressed")]
raw_lookup[stem.lower()] = f
narration_json_path = narration_dir / "narration.json"
raw_data: dict = _read_json(narration_json_path)
# Per-segment staleness: re-trim (and re-transcribe) a segment when its raw
# source's fingerprint differs from the one recorded last time it was trimmed.
from . import state as _state
_prev_trim_fps = _state.get_items(project_path, "trim")
_trim_fps: dict[str, str] = {} # segments to (re)record after the loop
updated = 0
for seg_id in sorted(narration.keys()):
seg = narration[seg_id]
# Prefer raw file; fall back to source_file from narration.json
source_path = raw_lookup.get(seg_id)
if source_path is None:
source_path = narration_dir / seg.source_file
if not source_path.exists():
print(f" {seg_id}: source file not found, skipping")
continue
# Has the raw source changed since we last trimmed this segment?
current_fp = _state.fingerprint_path(source_path, _state.META)
recorded_fp = _prev_trim_fps.get(seg_id)
source_changed = recorded_fp is not None and recorded_fp != current_fp
seg_force = force or source_changed
existing = raw_data.get(seg_id, {})
# Respect the user's manual trim points independently per side, and only
# auto-detect the side they haven't pinned:
# - begin/start pin the beginning; end pins the end (user vocabulary).
# - skip/take are what trim itself writes (or raw manual overrides).
# A side counts as "resolved" if pinned by the user OR already written by
# trim; --force re-detects auto (skip/take) sides but never overwrites a
# user-pinned begin/end.
begin_user = bool(existing.get("begin") or existing.get("start"))
end_user = bool(existing.get("end"))
begin_resolved = begin_user or "skip" in existing
end_resolved = end_user or "take" in existing
if begin_resolved and end_resolved and not seg_force:
print(f" {seg_id}: begin & end already set, skipping (use --force to redo)")
_trim_fps[seg_id] = current_fp # adopt/keep current fingerprint
continue
if source_changed:
print(f" {seg_id}: raw source changed since last trim — re-trimming")
slide_range = _parse_segment_slide_range(seg_id) if slide_texts else None
if slide_range is not None:
# --- Transcript-based trimming ---
from .transcriber import transcribe_video, save_transcript, load_transcript, TranscriptionError
start_slide, end_slide = slide_range
transcripts_dir.mkdir(parents=True, exist_ok=True)
transcript_path = transcripts_dir / f"{seg_id}.json"
try:
if transcript_path.exists() and not seg_force:
words = load_transcript(transcript_path)
print(f" {seg_id}: loaded cached transcript ({len(words)} words)")
else:
print(f" {seg_id}: transcribing {source_path.parent.name}/{source_path.name} (model={whisper_model})...", end="", flush=True)
words = transcribe_video(source_path, model=whisper_model)
save_transcript(words, transcript_path)
print(f" {len(words)} words")
if not words:
print(f" no words found — falling back to silence detection")
slide_range = None
else:
total_dur = get_video_duration(source_path)
# ---- Beginning: keep user's begin, else start of start-slide ----
if begin_user:
skip = _user_begin_skip(existing) or 0.0
begin_note = f"begin kept (user-set → {skip:.2f}s)"
else:
start_text = slide_texts.get(start_slide, "")
start_ts = _find_slide_start_in_transcript(words, start_text, verbose) if start_text else None
src = f"S{start_slide} first word" if start_ts is not None else "first word"
if start_ts is None:
start_ts = words[0].start
skip = max(0.0, round(start_ts - _TRIM_LEAD_IN, 3))
begin_note = f"begin auto: {src} {start_ts:.2f}s {_TRIM_LEAD_IN:g}s → {skip:.2f}s"
# ---- End: keep user's end, else end of end-slide + tail ----
if end_user:
end_abs = parse_timestamp(existing["end"])
end_note = f"end kept (user-set → {end_abs:.2f}s)"
elif end_slide is None:
# S{N}-end: end a couple of seconds after the last spoken word.
end_abs = min(words[-1].end + _TRIM_TAIL_OUT, total_dur)
end_note = f"end auto: last word {words[-1].end:.2f}s +{_TRIM_TAIL_OUT:g}s → {end_abs:.2f}s"
else:
end_text = slide_texts.get(end_slide, "")
end_ts = _find_slide_end_in_transcript(words, end_text, verbose) if end_text else None
if end_ts is None:
end_abs = total_dur
end_note = f"end auto: S{end_slide} not found → to end {total_dur:.2f}s"
else:
end_abs = min(end_ts + _TRIM_TAIL_OUT, total_dur)
end_note = f"end auto: S{end_slide} last word {end_ts:.2f}s +{_TRIM_TAIL_OUT:g}s → {end_abs:.2f}s"
# Write only the side(s) not pinned by the user, in skip/take terms.
if not begin_user:
raw_data[seg_id]["skip"] = skip
if not end_user:
raw_data[seg_id]["take"] = round(max(0.0, end_abs - skip), 3)
print(f" {begin_note} · {end_note}")
_trim_fps[seg_id] = current_fp
updated += 1
continue
except Exception as exc:
from .transcriber import TranscriptionError
label = "Whisper not installed" if "openai-whisper" in str(exc) else str(exc)
print(f"\n ⚠ transcription failed ({label}), falling back to silence detection")
slide_range = None # fall through
# --- Silence-based fallback (respects user-pinned begin/end too) ---
print(
f" {seg_id}: analysing {source_path.parent.name}/{source_path.name}...",
end="",
flush=True,
)
first_sound, last_sound = detect_silence_bounds(
source_path, noise_threshold_db=threshold_db, verbose=verbose
)
total_dur = get_video_duration(source_path)
if begin_user:
skip = _user_begin_skip(existing) or 0.0
begin_note = f"begin kept (user-set → {skip:.2f}s)"
else:
skip = max(0.0, round(first_sound - _TRIM_LEAD_IN, 3))
raw_data[seg_id]["skip"] = skip
begin_note = f"begin auto: first sound {first_sound:.2f}s {_TRIM_LEAD_IN:g}s → {skip:.2f}s"
if end_user:
end_abs = parse_timestamp(existing["end"])
end_note = f"end kept (user-set → {end_abs:.2f}s)"
else:
end_abs = min(total_dur, last_sound + 3.0)
raw_data[seg_id]["take"] = round(max(0.0, end_abs - skip), 3)
end_note = f"end auto: last sound {last_sound:.2f}s +3.0s → {end_abs:.2f}s"
print(f" {begin_note} · {end_note}")
_trim_fps[seg_id] = current_fp
updated += 1
if updated > 0:
with open(narration_json_path, "w", encoding="utf-8") as f:
json.dump(raw_data, f, indent=2)
print(f"\n Updated {updated} segment(s) in narration.json")
else:
print(f"\n No segments updated")
# Persist per-segment source fingerprints (freshly trimmed + adopted skips)
# so a later standalone or 'all' run can tell whether a raw was re-recorded.
if _trim_fps:
_state.record_items(project_path, "trim", _trim_fps)
return 0
# =============================================================================
# Transcode Command — compress narration folder to H.265
# =============================================================================
def _get_video_codec(path: Path) -> str:
"""Return the codec name of the first video stream (e.g. 'hevc', 'prores', 'h264')."""
result = subprocess.run(
[
"ffprobe",
"-v",
"error",
"-select_streams",
"v:0",
"-show_entries",
"stream=codec_name",
"-of",
"default=noprint_wrappers=1:nokey=1",
str(path),
],
capture_output=True,
text=True,
)
return result.stdout.strip().lower()
def _transcode_processed_files(
project_path: Path,
verbose: bool,
dry_run: bool,
replace: bool,
force: bool,
alpha_quality: float,
) -> int:
"""
Compress _processed.mov files (ProRes 4444 + alpha) to HEVC+alpha via
Apple VideoToolbox.
For each _processed.mov:
1. Transcode to a temp file using hevc_videotoolbox with alpha.
2. Move the ProRes original into a prores/ subdirectory (never deleted).
3. Rename the compressed file to the original _processed.mov name
so render finds it unchanged.
The prores/ subdirectory is never scanned — only top-level files are candidates.
If prores/<filename> already exists the file has already been compressed —
skip unless --force.
"""
from .parser import parse_project_config, parse_narration
print(f"Transcoding processed files (HEVC+alpha): {project_path.name}")
config = parse_project_config(project_path)
# Resolve narration_dir and videos_dir — processed files live in both
_narration, narration_dir = parse_narration(project_path, config)
videos_json_path = project_path / config.videos_path
videos_dir = videos_json_path.parent
# Glob both directories for *_processed.mov; skip any _prores.mov archives
search_dirs = [d for d in [narration_dir, videos_dir] if d.exists()]
candidates: list[Path] = []
seen: set[Path] = set()
for d in search_dirs:
for p in d.glob("*_processed.mov"):
if p not in seen and "_prores" not in p.stem:
seen.add(p)
candidates.append(p)
if not candidates:
print(" No _processed.mov files found.")
return 0
# Smallest first
candidates = [c for c in candidates if c.exists()]
candidates.sort(key=lambda f: f.stat().st_size)
total_original = 0
total_compressed = 0
transcoded = 0
skipped = 0
for src in candidates:
# Archive goes into prores/ subdirectory alongside the source file
prores_dir = src.parent / "prores"
archive = prores_dir / src.name
# Always skip files already encoded as HEVC — regardless of --replace or --force
if _get_video_codec(src) == "hevc":
print(f" {src.name}: already HEVC, skipping")
skipped += 1
continue
# Without --replace, skip if the archive already exists in prores/
if not replace and archive.exists() and not force:
size_mb = src.stat().st_size / 1_048_576
print(
f" {src.name}: already compressed ({size_mb:.1f} MB), skipping (use --force to redo)"
)
skipped += 1
continue
src_mb = src.stat().st_size / 1_048_576
print(f" {src.name} ({src_mb:.1f} MB) → HEVC+alpha", end="")
if dry_run:
print(" [dry-run]")
continue
print(" ...", end="", flush=True)
tmp_out = src.with_name(src.stem + "_hevc_tmp.mov")
cmd = [
"ffmpeg",
"-i",
str(src),
"-c:v",
"hevc_videotoolbox",
"-allow_sw",
"1",
"-alpha_quality",
str(alpha_quality),
"-tag:v",
"hvc1",
"-c:a",
"copy",
"-y",
str(tmp_out),
]
if verbose:
print()
print(" " + " ".join(cmd))
result = subprocess.run(
cmd,
capture_output=not verbose,
text=True,
)
if result.returncode != 0:
print(f"\n ERROR transcoding {src.name}")
if tmp_out.exists():
tmp_out.unlink()
if not verbose and result.stderr:
last_lines = result.stderr.strip().splitlines()[-5:]
for line in last_lines:
print(f" {line}", file=sys.stderr)
continue
out_mb = tmp_out.stat().st_size / 1_048_576
ratio = (1.0 - tmp_out.stat().st_size / src.stat().st_size) * 100
if replace:
# Delete ProRes original, move compressed into its place
src.unlink()
tmp_out.rename(src)
print(
f"\r {src.name} ({src_mb:.1f} MB) → HEVC+alpha"
f" ({out_mb:.1f} MB, -{ratio:.0f}%)"
)
else:
# Move ProRes original into prores/ subdirectory, compressed takes its place
prores_dir.mkdir(exist_ok=True)
src.rename(archive)
tmp_out.rename(src)
print(
f"\r {src.name} ({src_mb:.1f} MB) → HEVC+alpha"
f" ({out_mb:.1f} MB, -{ratio:.0f}%)"
f" [ProRes → prores/{archive.name}]"
)
total_original += int(src_mb * 1_048_576)
total_compressed += int(out_mb * 1_048_576)
transcoded += 1
print()
if dry_run:
print(f" [dry-run] Would compress {len(candidates) - skipped} file(s)")
return 0
if transcoded > 0:
orig_mb = total_original / 1_048_576
comp_mb = total_compressed / 1_048_576
saved_mb = orig_mb - comp_mb
ratio = (saved_mb / orig_mb * 100) if orig_mb else 0
print(
f" Compressed {transcoded} file(s): {orig_mb:.1f} MB → {comp_mb:.1f} MB"
f" (saved {saved_mb:.1f} MB, -{ratio:.0f}%)"
)
if skipped:
print(f" Skipped {skipped} already-compressed file(s)")
return 0
def cmd_transcode(
project_path: Path,
verbose: bool,
dry_run: bool = False,
replace: bool = False,
crf: int = 23,
force: bool = False,
processed: bool = False,
alpha_quality: float = 0.75,
) -> int:
"""
Transcode project video files to save disk space.
Default (1st pass, before preprocess):
Compress raw narration recordings to H.265. Output: {stem}_compressed.mp4.
Skips files with '_compressed.' or '_processed.' in the name.
Use --replace to delete originals after success.
With --processed (2nd pass, after preprocess):
Compress _processed.mov files (ProRes 4444 + alpha) to HEVC+alpha.
Archives the ProRes original as _prores.mov (never deleted).
The compressed file takes the original _processed.mov name so the
rest of the pipeline (render) finds it unchanged.
Uses Apple VideoToolbox (hevc_videotoolbox) with --alpha-quality.
"""
if processed:
return _transcode_processed_files(
project_path, verbose, dry_run, replace, force, alpha_quality
)
from .parser import parse_project_config, parse_narration
print(f"Transcoding narration: {project_path.name}")
config = parse_project_config(project_path)
_narration, narration_dir = parse_narration(project_path, config)
raw_dir = narration_dir / "raw_mov"
compressed_dir = narration_dir / "raw_mp4"
if not raw_dir.exists():
print(f" raw/ directory not found: {raw_dir}", file=sys.stderr)
print(f" Place raw recordings in {raw_dir} and run 'import' first.")
return 1
compressed_dir.mkdir(parents=True, exist_ok=True)
# Collect eligible video files from raw/ only
video_extensions = {".mp4", ".mov", ".avi", ".mkv", ".m4v", ".mts", ".webm"}
candidates = [
f
for f in raw_dir.iterdir()
if f.is_file()
and f.suffix.lower() in video_extensions
and not f.name.startswith(".")
]
if not candidates:
print(f" No video files found in {raw_dir}.")
return 0
# Process smallest files first
candidates.sort(key=lambda f: f.stat().st_size)
total_original = 0
total_compressed = 0
transcoded = 0
skipped = 0
for src in candidates:
# Output: compressed/<stem>.mp4 (clean name, no _compressed suffix)
output = compressed_dir / f"{src.stem}.mp4"
if output.exists() and not force:
size_mb = output.stat().st_size / 1_048_576
print(
f" {src.name}: already transcoded ({size_mb:.1f} MB), skipping (use --force to redo)"
)
skipped += 1
continue
src_mb = src.stat().st_size / 1_048_576
print(
f" raw/{src.name} ({src_mb:.1f} MB) → compressed/{output.name}", end=""
)
if dry_run:
print(" [dry-run]")
continue
print(" ...", end="", flush=True)
cmd = [
"ffmpeg",
"-i",
str(src),
"-vf",
"scale=-2:1080",
"-c:v",
"libx265",
"-crf",
str(crf),
"-preset",
"medium",
"-c:a",
"aac",
"-b:a",
"128k",
"-tag:v",
"hvc1",
"-y",
str(output),
]
if verbose:
print()
print(" " + " ".join(cmd))
result = subprocess.run(
cmd,
capture_output=not verbose,
text=True,
)
if result.returncode != 0:
print(f"\n ERROR transcoding {src.name}")
if not verbose and result.stderr:
# Print last few lines of ffmpeg stderr for diagnosis
last_lines = result.stderr.strip().splitlines()[-5:]
for line in last_lines:
print(f" {line}", file=sys.stderr)
continue
out_mb = output.stat().st_size / 1_048_576
ratio = (1.0 - output.stat().st_size / src.stat().st_size) * 100
print(
f"\r raw/{src.name} ({src_mb:.1f} MB) → compressed/{output.name} ({out_mb:.1f} MB, -{ratio:.0f}%)"
)
total_original += src.stat().st_size
total_compressed += output.stat().st_size
transcoded += 1
print()
if dry_run:
print(f" [dry-run] Would transcode {len(candidates) - skipped} file(s)")
return 0
if transcoded > 0:
orig_mb = total_original / 1_048_576
comp_mb = total_compressed / 1_048_576
saved_mb = orig_mb - comp_mb
ratio = (saved_mb / orig_mb * 100) if orig_mb else 0
print(
f" Transcoded {transcoded} file(s): {orig_mb:.1f} MB → {comp_mb:.1f} MB (saved {saved_mb:.1f} MB, -{ratio:.0f}%)"
)
if replace:
print(f" Originals deleted.")
if skipped:
print(f" Skipped {skipped} already-transcoded file(s)")
return 0
# =============================================================================
# Render Command
# =============================================================================
def _format_time(seconds: float) -> str:
"""Format seconds as MM:SS.ms"""
if seconds < 0:
return "??:??.??"
mins = int(seconds // 60)
secs = seconds % 60
return f"{mins:02d}:{secs:05.2f}"
def _print_render_plan_details(plan, marker_timings, slides: dict) -> None:
"""
Print a detailed render plan showing each marker with its aligned time.
Uses marker_timings from the transformer which contains alignment info.
"""
from .models import CAMERA_PRESETS
print("\n RENDER PLAN:")
print(" " + "-" * 76)
# Build lookup for video events by video_id
video_events_by_id = {}
for event in plan.video_events:
video_events_by_id[event.video_id] = event
audio_events_by_time = {}
for event in plan.audio_events:
t = round(event.start_time, 1)
if t not in audio_events_by_time:
audio_events_by_time[t] = []
audio_events_by_time[t].append(event)
camera_events_by_time = {}
for event in plan.camera_events:
t = round(event.time, 1)
if t not in camera_events_by_time:
camera_events_by_time[t] = []
camera_events_by_time[t].append(event)
# Detect slide markers that share a timestamp with the adjacent slide marker.
# Two slides at the same time means alignment is ambiguous — treat as an error.
slide_timings = [
t for t in marker_timings if t.marker_id in slides and t.timestamp >= 0
]
collision_ids: set[str] = set()
for a, b in zip(slide_timings, slide_timings[1:]):
if abs(a.timestamp - b.timestamp) < 0.1:
collision_ids.add(a.marker_id)
collision_ids.add(b.marker_id)
# Print each marker timing
aligned_count = 0
unaligned_count = 0
collision_count = 0
for timing in marker_timings:
marker_id = timing.marker_id
context = timing.context
if len(context) > 50:
context = context[:47] + "..."
if timing.timestamp >= 0:
time_str = _format_time(timing.timestamp)
# Show confidence if fuzzy match
conf_str = ""
if timing.confidence < 1.0:
conf_str = f" ({timing.confidence:.0%})"
# Determine marker type for display
if marker_id in slides:
if marker_id in collision_ids:
collision_count += 1
print(
f' {marker_id:6} {time_str}{conf_str} COLLISION - same time as adjacent slide - "{context}"'
)
else:
aligned_count += 1
print(f' {marker_id:6} {time_str}{conf_str} "{context}"')
elif any(
marker_id.startswith(p)
for p in (
"video:",
"vft:",
"vfb:",
"vf2t:",
"vf2b:",
"vst:",
"vsb:",
"vftp:",
"vfbp:",
"vf2tp:",
"vf2bp:",
"vstp:",
"vsbp:",
)
):
aligned_count += 1
pfx_len = next(
len(p)
for p in (
"video:",
"vft:",
"vfb:",
"vf2t:",
"vf2b:",
"vst:",
"vsb:",
"vftp:",
"vfbp:",
"vf2tp:",
"vf2bp:",
"vstp:",
"vsbp:",
)
if marker_id.startswith(p)
)
video_id = marker_id[pfx_len:]
# Find corresponding event by video_id
event = video_events_by_id.get(video_id)
if event:
cutout_name = event.cutout_name
end_on = event.video_source.end_on or "next_slide"
layer_tag = f" [{event.layer}]"
else:
cutout_name = "?"
end_on = "next_slide"
layer_tag = ""
cache_ind = " 📁" if video_id in plan.cached_files else ""
print(
f" {marker_id:20} {time_str} in '{cutout_name}' [{end_on}]{layer_tag}{cache_ind}"
)
elif marker_id.startswith("narration:"):
aligned_count += 1
video_id = marker_id[10:]
cache_ind = " 📁" if video_id in plan.cached_files else ""
print(f" {marker_id:20} {time_str} (continuous){cache_ind}")
elif marker_id in CAMERA_PRESETS:
aligned_count += 1
print(f" {time_str} [{marker_id}]")
elif marker_id.startswith("audio:"):
aligned_count += 1
print(f" {time_str} [{marker_id}]")
else:
aligned_count += 1
print(f' {marker_id:6} {time_str} "{context}"')
else:
unaligned_count += 1
# Check if this is a slide that was interpolated into the plan
if marker_id in slides:
interp_event = next(
(e for e in plan.slide_events if e.slide_id == marker_id), None
)
if interp_event:
interp_str = _format_time(interp_event.start_time)
print(f' {marker_id:6} ~{interp_str} INTERPOLATED - "{context}"')
else:
print(f' {marker_id:6} ??:??.?? NOT ALIGNED - "{context}"')
else:
print(f' {marker_id:6} ??:??.?? NOT ALIGNED - "{context}"')
print(" " + "-" * 76)
# Summary
total_markers = len(marker_timings)
slide_markers = [t for t in marker_timings if t.marker_id in slides]
good_slides = len(
[
t
for t in slide_markers
if t.timestamp >= 0 and t.marker_id not in collision_ids
]
)
total_slides = len(slide_markers)
issues = []
if unaligned_count:
issues.append(f"{unaligned_count} UNALIGNED")
if collision_count:
issues.append(f"{collision_count} COLLISION")
status = "OK" if not issues else ", ".join(issues)
print(f" Markers: {aligned_count}/{total_markers} aligned ({status})")
print(f" Slides: {good_slides}/{total_slides}")
print(
f" Videos: {len(plan.video_events)} triggered, {len(plan.narration_videos)} always-visible"
)
if plan.outro_events:
print(f" Outro: {len(plan.outro_events)} video(s)")
for event in plan.outro_events:
print(
f" - {event.video_id}: {_format_time(event.start_time)} - {_format_time(event.end_time)}"
)
print(f" Duration: {_format_time(plan.total_duration)}")
def _parse_slide_range(slides_arg: str) -> tuple[str, Optional[str]]:
"""Parse slide range argument like 'S1:S10' or 'S5:' into a tuple."""
if ":" not in slides_arg:
raise ValueError(
f"Invalid slide range '{slides_arg}'. Expected format: S1:S10 or S5:"
)
parts = slides_arg.split(":", 1)
start_slide = parts[0].strip()
end_slide = parts[1].strip() if parts[1].strip() else None
if not start_slide:
raise ValueError(
f"Invalid slide range '{slides_arg}'. Start slide is required."
)
return start_slide, end_slide
def _project_markers_to_videos(
markers: list[str], videos_json_path: Path, config, project_path: Path = None
) -> None:
"""ETL: project shorthand marker semantics into videos.json.
Scans the manuscript marker list for shorthand prefixes (vft:, vfb:, vst:,
vsb:, vf2t:, vf2b: and their pause variants) and writes the implied cutout
and layer values directly into videos.json. This runs before parse_videos
so the render pass reads already-projected data and needs no shorthand logic.
Videos may live in the project's local videos.json or in shared_assets/videos.json.
Both files are updated so the render pass always finds the projected values.
The manuscript is the authoritative source: the LAST shorthand reference to
a given video_id wins, matching what a human editor would expect when they
change a marker near the end of the script.
"""
if not videos_json_path.exists():
return
from .transformer import _SHORTHAND_PREFIXES # (cutout, layer) lookup table
_PAUSE_PREFIXES = {
"vftp:", "vfbp:", "vfmp:",
"vf2tp:", "vf2bp:", "vf2mp:",
"vstp:", "vsbp:", "vsmp:",
}
# Build projection: video_id → {cutout, layer, auto_pause_narration}
# auto_pause_narration=True means: write pause_narration=duration if not already set.
projection: dict[str, dict] = {}
for marker in markers:
for prefix, implied in _SHORTHAND_PREFIXES.items():
if marker.startswith(prefix):
video_id = marker[len(prefix):].lower()
cutout, layer = implied[0], implied[1]
projection[video_id] = {
"cutout": cutout,
"layer": layer,
"_auto_pause": prefix in _PAUSE_PREFIXES,
}
break
if not projection:
return
# Build a case-insensitive index of shared_assets pause_narration values.
# When a video is marked is_shared but its local entry is missing pause_narration,
# we pull the value from the shared canonical entry so it's never lost when
# the ETL writes back cutout/layer under a lowercase key.
_shared_pause: dict[str, float] = {}
for _shared_candidate in [
project_path / "shared_assets" / "videos.json",
project_path.parent / "shared_assets" / "videos.json",
]:
if _shared_candidate and _shared_candidate.exists():
try:
with open(_shared_candidate, "r", encoding="utf-8") as _f:
_shared_raw = json.load(_f)
for _k, _v in _shared_raw.items():
pn = _v.get("pause_narration")
if pn:
_shared_pause[_k.lower()] = float(pn)
except (json.JSONDecodeError, OSError):
pass
break
def _apply_projection(json_path: Path) -> list[str]:
"""Apply projection to one videos.json file; return list of updated IDs."""
if not json_path.exists():
return []
with open(json_path, "r", encoding="utf-8") as f:
raw = json.load(f)
changed = False
updated = []
for video_id, fields in projection.items():
if video_id not in raw:
continue
entry = raw[video_id]
video_changed = False
for field, value in fields.items():
if field == "_auto_pause":
# Write pause_narration = duration only when:
# - marker is a pause-prefix (value is True)
# - pause_narration not already set (preserve manual overrides)
# - duration is known (probed by import)
if value and not entry.get("pause_narration") and entry.get("duration"):
entry["pause_narration"] = entry["duration"]
changed = True
video_changed = True
elif entry.get(field) != value:
entry[field] = value
changed = True
video_changed = True
# For is_shared entries: inherit pause_narration from shared_assets if
# not already set locally (handles case where explicit pause_narration
# lives on a different-case key in the shared library).
if entry.get("is_shared") and not entry.get("pause_narration"):
shared_pn = _shared_pause.get(video_id.lower())
if shared_pn:
entry["pause_narration"] = shared_pn
changed = True
video_changed = True
if video_changed:
updated.append(video_id)
if changed:
with open(json_path, "w", encoding="utf-8") as f:
json.dump(raw, f, indent=2, ensure_ascii=False)
return updated
updated_local = _apply_projection(videos_json_path)
if updated_local:
print(f" Projected marker semantics → videos.json: {', '.join(updated_local)}")
def _writeback_video_metadata(plan, project_path, config) -> None:
"""Write back cutout/layer derived from shorthand markers to videos.json.
When a shorthand like [vfb:FARTSection1] is used and FARTSection1 has no
'cutout' set in videos.json, this persists the resolved cutout (and layer if
the shorthand implies a non-default layer) back to the file. Once written,
subsequent renders read the value directly and no further write-back occurs.
"""
import json
videos_json_path = project_path / config.videos_path
if not videos_json_path.exists():
return
# Collect field updates per video_id
writebacks: dict[str, dict] = {}
for event in plan.video_events:
video_id = event.video_id
source = event.video_source
if source.is_shared:
continue # shared videos live in their own file
updates = {}
if source.cutout is None and event.cutout_name:
updates["cutout"] = event.cutout_name
if event.layer != source.layer:
updates["layer"] = event.layer
if updates:
writebacks.setdefault(video_id, {}).update(updates)
if not writebacks:
return
with open(videos_json_path, "r", encoding="utf-8") as f:
raw = json.load(f)
changed = False
for video_id, updates in writebacks.items():
if video_id not in raw:
continue
for field, value in updates.items():
if raw[video_id].get(field) != value:
raw[video_id][field] = value
changed = True
if changed:
with open(videos_json_path, "w", encoding="utf-8") as f:
json.dump(raw, f, indent=2, ensure_ascii=False)
written = ", ".join(
f"{vid}({', '.join(upd)})" for vid, upd in writebacks.items()
)
print(f" Updated videos.json: {written}")
def _chunked_render(
project_path: Path,
verbose: bool,
dry_run: bool,
res: str,
force: bool,
chunk_size: int,
slide_ids: list[str],
out_dir: Path,
final_output: Path,
) -> int:
"""Render in slide-based chunks then concatenate — avoids filter graph OOM."""
import math
# Split slide IDs into groups of chunk_size
groups = [
slide_ids[i : i + chunk_size] for i in range(0, len(slide_ids), chunk_size)
]
print(
f"\n Auto-chunking: {len(slide_ids)} slides → {len(groups)} chunks of ≤{chunk_size}"
)
chunks_dir = out_dir / "chunks"
chunks_dir.mkdir(parents=True, exist_ok=True)
chunk_paths: list[Path] = []
for i, group in enumerate(groups):
start = group[0]
end = groups[i + 1][0] if i + 1 < len(groups) else None
slides_arg = f"{start}:{end}" if end else f"{start}:"
chunk_path = chunks_dir / f"chunk_{i+1:03d}_{start}-{end or 'end'}.mp4"
print(f"\n {'='*56}")
print(f" Chunk {i+1}/{len(groups)}: {slides_arg}{chunk_path.name}")
print(f" {'='*56}")
result = cmd_render(
project_path,
verbose,
dry_run,
slides_arg=slides_arg,
res=res,
force=force,
_output_path_override=chunk_path,
)
if result != 0:
print(f"\n Chunk {i+1} failed — aborting.", file=sys.stderr)
return result
chunk_paths.append(chunk_path)
if dry_run:
print(
f"\n [dry-run] Would concatenate {len(chunk_paths)} chunks → {final_output}"
)
return 0
# Concatenate chunks
print(f"\n Concatenating {len(chunk_paths)} chunks → {final_output.name}...")
concat_list = chunks_dir / "concat.txt"
with open(concat_list, "w") as f:
for p in chunk_paths:
f.write(f"file '{p.resolve()}'\n")
concat_cmd = [
"ffmpeg",
"-y",
"-f",
"concat",
"-safe",
"0",
"-i",
str(concat_list),
"-c",
"copy",
str(final_output),
]
result = subprocess.run(concat_cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f" Concatenation failed:\n{result.stderr}", file=sys.stderr)
return 1
# Clean up chunk files
for p in chunk_paths:
p.unlink(missing_ok=True)
concat_list.unlink(missing_ok=True)
try:
chunks_dir.rmdir()
except OSError:
pass
print(f" Output: {final_output}")
return 0
def _build_merged_transcription(project_path: Path, config, verbose: bool = False):
"""Deterministic merged transcript for slide alignment.
Builds a single word-level transcript from the per-segment transcripts +
the current narration.json skip/take, re-timed into the combined timeline
(see narration.build_narration_schedule). This keeps alignment in sync with
narration.json and avoids re-transcribing the combined file.
Returns None (caller falls back to the on-disk transcript) when there are no
narration segments or any segment is missing its per-segment transcript.
"""
from .parser import parse_narration, get_video_duration
from .narration import build_narration_schedule
try:
narration, narration_dir = parse_narration(project_path, config)
except GnommoError:
return None
if not narration:
return None
transcripts_dir = narration_dir / "transcripts"
missing = [sid for sid in narration if not (transcripts_dir / f"{sid}.json").exists()]
if missing:
if verbose:
print(f" Merged transcript unavailable (no per-segment transcript for: "
f"{', '.join(missing)}) — using on-disk transcript.")
return None
_segments, merged = build_narration_schedule(
narration, narration_dir, get_video_duration,
transcripts_dir=transcripts_dir, verbose=verbose,
)
return merged or None
def cmd_render(
project_path: Path,
verbose: bool,
dry_run: bool,
slides_arg: str = None,
res: str = "full",
force: bool = False,
chunk_slides: int = 0,
_output_path_override: Path = None,
) -> int:
"""Render final video."""
from .parser import (
parse_audio,
parse_manuscript,
parse_project_config,
parse_slides,
parse_videos,
save_citations,
)
from .transcriber import load_transcript
from .validator import validate_project
from .transformer import build_render_plan
from .renderer import render, generate_ffmpeg_command_string
from .preprocessor import RES_CONFIGS, ensure_downscaled_files_exist
# Parse slide range if provided
slide_range = None
if slides_arg:
slide_range = _parse_slide_range(slides_arg)
print(f"Rendering: {project_path.name} (slides {slides_arg})")
else:
print(f"Rendering: {project_path.name}")
# Show resolution mode
if res != "full":
cfg = RES_CONFIGS[res]
print(f" Resolution: {res.upper()} ({cfg[0]}x{cfg[1]})")
# Show cache status
cache_info = get_cache_info()
if cache_info:
print(f" Cache: {cache_info}")
# Stage 1: Parse
print("\n[1/4] Parsing...")
manuscript_text, markers, malformed, citations = parse_manuscript(project_path)
# Save citations for later use (e.g., description generation)
if citations:
citations_path = project_path / "citations.json"
save_citations(citations, citations_path)
config = parse_project_config(project_path)
# ETL: project shorthand marker semantics (cutout/layer) into videos.json
# before parse_videos reads it, so the render pass is purely data-driven.
_project_markers_to_videos(markers, project_path / config.videos_path, config, project_path)
# Override resolution for preview modes
if res != "full":
cfg = RES_CONFIGS[res]
config.resolution = (cfg[0], cfg[1])
slides = parse_slides(project_path, config)
videos, videos_dir = parse_videos(project_path, config)
source_videos_dir = videos_dir # keep original for validation (pre-downscale)
# Non-full res: use downscaled video directory, create on-the-fly if needed
if res != "full":
# Skip downscaling sources that have a preprocessed output_file — the
# renderer will use the full-res processed version instead, saving disk space.
sources_with_output = {v.source_file for v in videos.values() if v.output_file}
videos_dir = ensure_downscaled_files_exist(
videos_dir,
res,
force=False,
verbose=verbose,
skip_sources=sources_with_output,
)
if verbose:
print(f" Using {res} dir: {videos_dir}")
audio, audio_dir = parse_audio(project_path, config)
# --- Narration: render-time concat of the processed segments ---
# narration.json is the single source of truth; the processed segments are
# concatenated directly in the render graph.
from .narration import build_narration_schedule
from .parser import parse_narration as _parse_narr, get_video_duration
narration_map, narration_seg_dir = _parse_narr(project_path, config)
narration_schedule: list = []
narration_source = None
transcript_path = None
if narration_map:
narration_schedule, _ = build_narration_schedule(
narration_map, narration_seg_dir, get_video_duration
)
missing = [s.seg_id for s in narration_schedule if not s.source_path.exists()]
if missing:
print(
f"Error: processed narration segment(s) not found: {', '.join(missing)}",
file=sys.stderr,
)
print(f"Run 'gnommo -p {project_path.name} preprocess' first.", file=sys.stderr)
return 1
# Talking-head cutout/zoom/audio settings come from the first segment.
narration_source = narration_map[narration_schedule[0].seg_id]
# --- Transcript for slide alignment ---
# Prefer the deterministic merged transcript (per-segment transcripts re-timed
# into the concatenated timeline). Otherwise fall back to an on-disk transcript.
transcription = _build_merged_transcription(project_path, config, verbose)
if transcription is not None:
if verbose:
print(f" Using merged per-segment transcript ({len(transcription)} words)")
else:
if config.transcript_path and (project_path / config.transcript_path).exists():
transcript_path = project_path / config.transcript_path
elif narration_map:
# Narration project with no per-segment transcripts to merge.
print(
"Error: No per-segment transcripts found for the narration segments.",
file=sys.stderr,
)
print(
f"Run 'gnommo -p {project_path.name} trim' first (it transcribes each segment).",
file=sys.stderr,
)
return 1
else:
result = _find_narration_video(config, videos)
if result:
_vid, _src = result
config.main_video = _vid
transcript_path = (videos_dir / _src.source_file).with_suffix(".transcript.json")
else:
transcript_path = project_path / "transcript.json"
transcript_path, _ = resolve_with_cache(transcript_path, project_path)
if not transcript_path.exists():
print(f"Error: Transcription not found: {transcript_path}", file=sys.stderr)
print(f"Run 'gnommo -p {project_path.name} trim' first (it produces per-segment transcripts).", file=sys.stderr)
return 1
transcription = load_transcript(transcript_path, project_path)
if verbose:
print(f" - Markers in manuscript: {len(markers)}")
print(f" - Slides defined: {len(slides)}")
print(f" - Audio clips: {len(audio)}")
print(f" - Transcription words: {len(transcription)}")
# Stage 2: Validate
print("\n[2/4] Validating...")
warnings = validate_project(
project_path, markers, config, slides, videos, source_videos_dir, malformed
)
for w in warnings:
print(f" Warning: {w}")
print(" Passed.")
# Stage 3: Transform (includes on-the-fly alignment)
print("\n[3/4] Building render plan...")
plan, marker_timings = build_render_plan(
project_path,
config,
slides,
videos,
videos_dir,
manuscript_text,
transcription,
audio,
audio_dir,
slide_range=slide_range,
narration_schedule=narration_schedule,
narration_source=narration_source,
)
if plan.time_offset > 0:
print(f" Time offset: {plan.time_offset:.1f}s (partial render)")
if plan.narration_segments:
print(f" Narration concat: {len(plan.narration_segments)} segment(s) at render time")
# Print detailed render plan with alignment info
_print_render_plan_details(plan, marker_timings, slides)
if plan.audio_events:
print(f"\n Audio effects:")
for event in plan.audio_events:
loop_str = " (loop)" if event.audio_def.loop else ""
pause_str = " [ignores pauses]" if event.audio_def.ignore_pauses else ""
print(
f" - {event.audio_id}: '{event.audio_def.file}' @ {_format_time(event.start_time)}{loop_str}{pause_str}"
)
# Show always-visible videos
if plan.narration_videos:
print(f"\n Always-visible videos:")
for video_id, video_source, cutout in plan.narration_videos:
skip_str = (
f" (skip: {video_source.skip:.1f}s)" if video_source.skip > 0 else ""
)
cache_ind = " 📁" if video_id in plan.cached_files else ""
print(f" - {video_id} in '{video_source.cutout}'{skip_str}{cache_ind}")
# Show narration pauses
if plan.narration_pauses:
print(f"\n Narration pauses:")
for pause in plan.narration_pauses:
print(
f" - {pause.video_id} at {_format_time(pause.output_time)} "
f"for {pause.duration:.1f}s (narration freezes at {_format_time(pause.narration_time)})"
)
# Write tasks file with both missing assets and alignment issues
missing_videos = _collect_missing_video_markers(markers, videos)
slide_timings_for_collision = [
t for t in marker_timings if t.marker_id in slides and t.timestamp >= 0
]
collision_ids_render = set()
for _a, _b in zip(slide_timings_for_collision, slide_timings_for_collision[1:]):
if abs(_a.timestamp - _b.timestamp) < 0.1:
collision_ids_render.add(_a.marker_id)
collision_ids_render.add(_b.marker_id)
alignment_issues = [
(t.marker_id, t.context)
for t in marker_timings
if t.marker_id in slides
and (t.timestamp < 0 or t.marker_id in collision_ids_render)
]
_write_tasks_file(project_path, missing_videos, alignment_issues)
# Check for unaligned markers
unaligned = [t for t in marker_timings if t.timestamp < 0]
if unaligned:
print(f"\n WARNING: {len(unaligned)} marker(s) could not be aligned!")
for t in unaligned:
print(f' [{t.marker_id}] - "{t.context}"')
if not force:
print(f"\n Run with -f/--force to render anyway.")
return 1
else:
print(f"\n Continuing anyway due to --force flag...")
# Stage 4: Render
# Determine output filename and directory
if _output_path_override:
output_path = _output_path_override
out_dir = output_path.parent
out_filename = output_path.name
elif config.output_video:
out_filename = config.output_video
out_dir = project_path / "out" / res if res != "full" else project_path / "out"
output_path = out_dir / out_filename
elif slide_range:
start, end = slide_range
range_suffix = f"_{start}-{end}" if end else f"_{start}-end"
out_filename = f"final{range_suffix}.mp4"
out_dir = project_path / "out" / res if res != "full" else project_path / "out"
output_path = out_dir / out_filename
else:
out_filename = f"{config.co}.mp4"
out_dir = project_path / "out" / res if res != "full" else project_path / "out"
output_path = out_dir / out_filename
# Check if chunked rendering is needed (avoids filter graph OOM on long videos)
from .cache import get_render_chunk_size
_chunk_size = chunk_slides or get_render_chunk_size() or 0
_slide_ids = [e.slide_id for e in plan.slide_events]
if _chunk_size > 0 and not slide_range and len(_slide_ids) > _chunk_size:
return _chunked_render(
project_path,
verbose,
dry_run,
res,
force,
_chunk_size,
_slide_ids,
out_dir,
output_path,
)
plan.output_path = output_path
if dry_run:
print("\n[4/4] FFmpeg command (dry run):")
print(generate_ffmpeg_command_string(plan, output_path))
return 0
# Stage-level staleness gate. Only whole-project renders are gated — partial
# (--slides) renders and internal chunk sub-renders always run. The render is
# skipped when the combined narration, manifests, manuscript, transcript and
# slide images are all unchanged since the last successful render of this
# resolution and the output file still exists.
from . import state as _state
_render_gateable = (
not force and slide_range is None and _output_path_override is None
)
def _render_input_specs() -> list:
_slides_json = project_path / config.slides_path.lower()
_slides_dir = _slides_json.parent
specs = [
("videos.json", project_path / config.videos_path, _state.HASH),
("audio.json", project_path / config.audio_path, _state.HASH),
("manuscript.txt", project_path / "manuscript.txt", _state.HASH),
("project.json", project_path / "project.json", _state.HASH),
("slides.json", _slides_json, _state.HASH),
# narration.json (skip/take) drives the concat timeline + alignment.
("narration.json", project_path / "media" / "narration" / "narration.json", _state.HASH),
]
if transcript_path:
specs.append(("transcript", transcript_path, _state.HASH))
# The concatenated narration segments (the render's main video input).
for _seg in plan.narration_segments:
specs.append((f"narr:{_seg.seg_id}", _seg.source_path, _state.META))
# Per-segment transcripts feed the merged transcript for alignment.
_tdir = project_path / "media" / "narration" / "transcripts"
if _tdir.is_dir():
for _tj in sorted(_tdir.glob("*.json")):
specs.append((f"transcript:{_tj.stem}", _tj, _state.HASH))
for _sid, _sdef in slides.items():
specs.append((f"slide:{_sid}", _slides_dir / _sdef.image, _state.META))
return specs
_render_key = f"render:{res}"
_render_current = _render_gateable and _state.is_current(
project_path, _render_key, _state.compute(_render_input_specs()), [output_path]
)
# Timestamp guard (make-style): even if the fingerprint matches, re-render
# when a key upstream artifact — the concatenated narration or its transcript —
# is newer than the rendered output. This catches cases the fingerprint
# can't: the state file isn't transferred to the render rig (it's a dotfile),
# and the combined may live on an external cache disk. If the previous stage
# produced a newer file, the render is stale regardless of recorded state.
if _render_current and output_path.exists():
try:
_out_mtime = output_path.stat().st_mtime
_deps = [s.source_path for s in plan.narration_segments]
if transcript_path:
_deps.append(transcript_path)
for _dep in _deps:
if _dep and Path(_dep).exists() and Path(_dep).stat().st_mtime > _out_mtime:
print(f" {Path(_dep).name} is newer than the render — regenerating.")
_render_current = False
break
except OSError:
_render_current = False
if _render_current:
print(f"\n[4/4] Output up to date: {output_path}")
print(" (inputs unchanged since last render — use --force to re-render)")
print("\nDone.")
return 0
print("\n[4/4] Rendering...")
render(plan, output_path, verbose=verbose)
print(f" Output: {output_path}")
if _render_gateable:
_state.record(project_path, _render_key, _state.compute(_render_input_specs()))
print("\nDone.")
return 0
# =============================================================================
# Transcribe Command
# =============================================================================
def _find_narration_video(config, videos: dict) -> Optional[tuple[str, "VideoSource"]]:
"""
Find the video to use for transcription/narration.
Priority:
1. config.audio_source if set
2. First video with always_visible=True
3. First video in dict
"""
from .models import VideoSource
# 1. Check audio_source config
if config.audio_source and config.audio_source in videos:
return config.audio_source, videos[config.audio_source]
# 2. Find always_visible video (main talking head)
for video_id, video_source in videos.items():
if video_source.always_visible:
return video_id, video_source
# 3. Fall back to first video
if videos:
video_id = next(iter(videos.keys()))
return video_id, videos[video_id]
return None
# =============================================================================
# Grade Command
# =============================================================================
def cmd_grade(
project_path: Path,
verbose: bool,
file: Optional[str] = None,
ss: Optional[float] = None,
dur: float = 3.0,
) -> int:
"""Sample a few seconds of a raw narration clip through the talkinghead
filter chain so you can iterate on gnommokey / color_grade settings without
running a full preprocess.
Writes two files to the project root:
grade_preview.mov — the exact keyed ProRes 4444 output (alpha over black)
grade_preview.mp4 — the same result flattened over mid-gray, easy to view
in any player (best for judging spill and skin tone)
"""
from .parser import parse_project_config
from .preprocessor import _process_chunk_to_prores4444, get_video_duration
config = parse_project_config(project_path)
talkinghead_filter = (config.default_filters or {}).get("talkinghead", [])
if not talkinghead_filter:
print(
" ERROR: No 'talkinghead' filter defined in project.json default_filters."
)
return 1
# --- Resolve source clip ---
raw_dir = project_path / "media" / "narration" / "raw_mov"
_video_exts = {".mov", ".mp4", ".avi", ".mkv", ".m4v"}
if file:
source = Path(file)
if not source.is_absolute():
source = project_path / file
if not source.exists():
alt = raw_dir / Path(file).name
if alt.exists():
source = alt
if not source.exists():
print(f" ERROR: source file not found: {file}")
return 1
else:
candidates = (
sorted(
f
for f in raw_dir.iterdir()
if f.is_file()
and f.suffix.lower() in _video_exts
and not f.name.startswith(".")
)
if raw_dir.exists()
else []
)
if not candidates:
print(f" ERROR: no raw clips found in {raw_dir}")
print(" Pass --file to point at a specific clip.")
return 1
source = candidates[0]
# --- Resolve seek / duration, clamped to the clip length ---
clip_len = get_video_duration(source)
if ss is None:
# Default: 5s in, or centred if the clip is short.
ss = 5.0 if clip_len > 8 else max(0.0, clip_len / 2 - dur / 2)
if ss >= clip_len:
ss = max(0.0, clip_len - dur)
take = min(dur, max(0.1, clip_len - ss))
print(f"Grading preview: {project_path.name}")
print(f" Source: {source}")
print(f" Sample: {take:.1f}s starting at {ss:.1f}s (clip is {clip_len:.1f}s)")
print(f" Filters: {len(talkinghead_filter)} step(s)")
mov_out = project_path / "grade_preview.mov"
_process_chunk_to_prores4444(
source,
mov_out,
talkinghead_filter,
start_time=ss,
chunk_duration=take,
verbose=verbose,
take=take,
keep_audio=False,
)
print(f"\n Done. Keyed preview (ProRes 4444, alpha): {mov_out}")
return 0
# =============================================================================
# Align Command
# =============================================================================
def cmd_align(project_path: Path, verbose: bool) -> int:
"""Preview manuscript marker alignment (no files written)."""
from .transcriber import load_transcript
from .transformer import align_markers_to_transcription
from .parser import (
parse_project_config,
parse_videos,
parse_slides,
parse_audio,
parse_manuscript,
save_citations,
)
print(f"Alignment preview: {project_path.name}")
print(" (This is a preview - alignment happens automatically during render)")
# Load manuscript (cites are stripped at parse time)
manuscript_text, _, _, citations = parse_manuscript(project_path)
# Save citations for later use (e.g., description generation)
if citations:
citations_path = project_path / "citations.json"
save_citations(citations, citations_path)
# Load project config and resources
config = parse_project_config(project_path)
slides = parse_slides(project_path, config)
videos, videos_dir = parse_videos(project_path, config)
audio, _ = parse_audio(project_path, config)
# Find transcription (from narration video)
result = _find_narration_video(config, videos)
if not result:
print("Error: No suitable video found for transcription", file=sys.stderr)
return 1
video_id, video_source = result
video_path = videos_dir / video_source.source_file
transcript_path = video_path.with_suffix(".transcript.json")
# Try cache fallback for transcript
transcript_path, _ = resolve_with_cache(transcript_path, project_path)
if not transcript_path.exists():
print(f"Error: Transcription not found: {transcript_path}", file=sys.stderr)
print(f"Run 'gnommo -p {project_path.name} trim' first (it produces per-segment transcripts).", file=sys.stderr)
return 1
print(f" Loading: {transcript_path.name}")
transcription = load_transcript(transcript_path, project_path)
print(f" - {len(transcription)} words")
# Align (cite markers already stripped at parse time)
print("\n Aligning markers to transcription...")
timings = align_markers_to_transcription(
manuscript_text, transcription, slides=slides, videos=videos, audio=audio
)
# Report alignment results
unmatched = 0
fuzzy_matched = 0
exact_matched = 0
for t in timings:
if t.timestamp >= 0:
if t.confidence >= 1.0:
exact_matched += 1
if verbose:
print(f" [{t.marker_id}] @ {_format_time(t.timestamp)}")
else:
fuzzy_matched += 1
# Always show fuzzy matches so user can verify
print(
f" [{t.marker_id}] @ {_format_time(t.timestamp)} (fuzzy {t.confidence:.0%})"
)
else:
print(f' [{t.marker_id}] NOT FOUND - "{t.context}"')
unmatched += 1
# Summary
total = len(timings)
print(f"\n Alignment summary:")
print(f" - Exact matches: {exact_matched}/{total}")
if fuzzy_matched > 0:
print(f" - Fuzzy matches (60%+ words): {fuzzy_matched}/{total}")
if unmatched > 0:
print(f" - NOT FOUND: {unmatched}/{total}")
print(
f"\n Some markers could not be aligned. Check manuscript.txt matches the spoken audio."
)
return 0
# =============================================================================
# All Command (Full Pipeline)
# =============================================================================
def _files_modified_since(root: Path, since: float, pattern: str) -> bool:
"""Return True if any file matching pattern under root has mtime > since."""
try:
for p in root.rglob(pattern):
if p.is_file() and p.stat().st_mtime > since:
return True
except (OSError, PermissionError):
pass
return False
def _trim_outputs_current(project_path: Path) -> bool:
"""Return True if the trim stage can be safely skipped for every segment.
A segment is considered resolved when narration.json already records an
explicit skip/take for it, or a cached Whisper transcript exists at
narration/transcripts/{seg_id}.json (from which trim would just recompute
the same skip/take). Used by the 'all' pipeline to avoid re-running the
expensive transcription stage when nothing upstream changed.
Returns False (i.e. "run trim") if narration can't be read or any segment
is still unresolved.
"""
from .parser import parse_project_config, parse_narration
try:
config = parse_project_config(project_path)
narration, narration_dir = parse_narration(project_path, config)
except GnommoError:
return False
if not narration:
return False
transcripts_dir = narration_dir / "transcripts"
try:
raw_data = _read_json(narration_dir / "narration.json")
except (OSError, json.JSONDecodeError):
raw_data = {}
for seg_id in narration:
entry = raw_data.get(seg_id, {})
if "skip" in entry or "take" in entry:
continue # already trimmed
if (transcripts_dir / f"{seg_id}.json").exists():
continue # transcript cached — trim would just reuse it
return False # unresolved segment: trim still has work to do
return True
def cmd_all(
project_path: Path,
verbose: bool,
dry_run: bool,
res: str = "full",
force: bool = False,
) -> int:
"""Run full pipeline: import → prune → preprocess → trim → render → push → handoff → up.
Cascade rule: if any stage produces output, all subsequent stages are forced
to re-run (cascade_force=True), regardless of whether --force was passed.
This ensures downstream caches are always consistent with upstream changes.
"""
from .handoff import cmd_handoff
from .push import cmd_push
print(f"=== Full Pipeline: {project_path.name} ===\n")
# cascade_force starts at --force. Once any stage does real work it flips to
# True so all downstream stages re-run unconditionally.
cascade_force = force
print(">>> Step 1/8: Import\n")
t0 = time.time()
result = cmd_import(project_path, cascade_force, verbose)
if result != 0:
return result
if _files_modified_since(project_path, t0, "slides.json") or _files_modified_since(
project_path, t0, "narration.json"
):
cascade_force = True
print("\n>>> Step 2/8: Prune\n")
# Drop manifest entries left over from edits (e.g. a video split into two
# projects). Only removes unused entries, so it never forces downstream re-runs.
result = cmd_prune(project_path, verbose, dry_run)
if result != 0:
return result
print("\n>>> Step 3/8: Preprocess\n")
t0 = time.time()
result = cmd_preprocess(
project_path, verbose, dry_run, cascade_force, workers=1, res=res
)
if result != 0:
return result
if _files_modified_since(
project_path, t0, "*_processed.mov"
) or _files_modified_since(project_path, t0, "*_processed.webm"):
cascade_force = True
print("\n>>> Step 4/8: Trim\n")
# Skip the (Whisper-heavy) trim stage when nothing upstream changed
# (cache intact) and every segment is already resolved — i.e. it has a
# cached transcript or explicit skip/take. A cascade_force from preprocess
# (a raw video was reprocessed) always forces trim to re-run.
if not cascade_force and _trim_outputs_current(project_path):
print(" Cache intact and transcripts present for all segments — skipping trim.")
else:
t0 = time.time()
result = cmd_trim(project_path, verbose, force=cascade_force, threshold_db=-40.0)
if result != 0:
return result
# Trim modifies narration.json skip/take values; any change invalidates the render
if _files_modified_since(project_path, t0, "narration.json"):
cascade_force = True
print("\n>>> Step 5/8: Render\n")
result = cmd_render(project_path, verbose, dry_run, res=res, force=cascade_force)
if result != 0:
return result
print("\n>>> Step 6/8: Push\n")
result = cmd_push(project_path, verbose, force=False, prod=True)
if result != 0:
return result
print("\n>>> Step 7/8: Handoff\n")
result = cmd_handoff(project_path, verbose, file_override=None, prod=True, res=res)
if result != 0:
return result
print("\n>>> Step 8/8: Upload\n")
from .transfer import cmd_up
return cmd_up(project_path, verbose, dry_run)
# =============================================================================
# Description Command
# =============================================================================
def cmd_description(project_path: Path, verbose: bool) -> int:
"""Generate YouTube description file with chapters, citations, and attributions."""
from .parser import (
parse_audio,
parse_manuscript,
parse_project_config,
parse_slides,
parse_videos,
load_citations,
)
from .transcriber import load_transcript
from .transformer import align_markers_to_transcription
from .description import write_description_file
print(f"Generating description: {project_path.name}")
# Parse all project files
manuscript_text, markers, _, _ = parse_manuscript(project_path)
# Load citations from file (saved during parse/render/align stages)
citations_path = project_path / "citations.json"
citations = load_citations(citations_path)
config = parse_project_config(project_path)
slides = parse_slides(project_path, config)
videos, videos_dir = parse_videos(project_path, config)
audio, _ = parse_audio(project_path, config)
# Load transcription for alignment (optional but recommended)
transcription = None
result = _find_narration_video(config, videos)
if result:
_, narration_source = result
video_path = videos_dir / narration_source.source_file
transcript_path = video_path.with_suffix(".transcript.json")
# Try cache fallback for transcript
transcript_path, _ = resolve_with_cache(transcript_path, project_path)
if transcript_path.exists():
transcription = load_transcript(transcript_path, project_path)
if verbose:
print(f" Loaded transcription: {len(transcription)} words")
else:
print(f" Warning: No transcription found at {transcript_path}")
print(
f" Run 'gnommo -p {project_path.name} trim' for better timestamps."
)
# Align markers to get timings
print(" Aligning markers...")
marker_timings = align_markers_to_transcription(
manuscript_text,
transcription or [],
slides=slides,
videos=videos,
audio=audio,
)
if verbose:
aligned = sum(1 for t in marker_timings if t.timestamp >= 0)
print(f" Aligned {aligned}/{len(marker_timings)} markers")
# Generate description
output_path = project_path / "out" / "description_youtube.txt"
description = write_description_file(
output_path=output_path,
config=config,
manuscript_text=manuscript_text,
slides=slides,
videos=videos,
marker_timings=marker_timings,
transcription=transcription,
citations=citations,
)
# Print summary
lines = description.split("\n")
print(f"\n Output: {output_path}")
print(f" Length: {len(description)} characters, {len(lines)} lines")
# Show sections found
sections = []
if config.description:
sections.append("description")
if "CHAPTERS" in description:
sections.append("chapters")
if "REFERENCES" in description:
sections.append("references")
if "STOCK FOOTAGE" in description:
sections.append("attributions")
if config.footer:
sections.append("footer")
print(f" Sections: {', '.join(sections)}")
if verbose:
print("\n --- Preview ---")
preview_lines = lines[:20]
for line in preview_lines:
print(f" {line}")
if len(lines) > 20:
print(f" ... ({len(lines) - 20} more lines)")
print("\nDone.")
return 0
# Files and directories excluded from all sync/archive/load operations.
# Covers intermediate processing artifacts, chunk scratch dirs, venv, and
# common OS/editor noise.
_RSYNC_EXCLUDES = [
# Intermediate processing files
"media/narration/intermediate/",
"media/narration/intermediate/**",
"media/videos/intermediate/",
"media/videos/intermediate/**",
"media/narration/processed/",
"media/narration/processed/**",
# Low-res preview files (generated locally, not synced)
"media/narration/low/",
"media/narration/low/**",
"media/videos/low/",
"media/videos/low/**",
# Chunk scratch directories
"**/chunks/",
"**/chunks/**",
# Python
"*.py",
"__pycache__/",
"venv/",
# Version control / OS noise
".git/",
".DS_Store",
"*.tmp",
]
def cmd_archive(project_path: Path, verbose: bool, dry_run: bool) -> int:
"""Archive project files to external cache storage."""
from .cache import load_cache_config, load_assets_config
print(f"Archiving: {project_path.name}")
# Prefer [cache] disk; fall back to [assets] disk if cache isn't mounted.
cache_base = None
for label, candidate in (("cache", load_cache_config()), ("assets", load_assets_config())):
if candidate is not None and candidate.exists():
cache_base = candidate
print(f" Using {label} disk: {candidate}")
break
if cache_base is None:
# Give a useful error: list which paths were tried
tried = []
for label, fn in (("cache", load_cache_config), ("assets", load_assets_config)):
p = fn()
if p is not None:
tried.append(f" [{label}] {p}")
if tried:
print("Error: No configured external drive is mounted. Tried:")
for t in tried:
print(t)
else:
print("Error: No external drive configured. Create ~/.gnommo.conf with:")
print(" [cache]")
print(" path = /Volumes/YourDisk/gnommo")
return 1
# Build destination path
dest_path = cache_base / project_path.name
print(f" Source: {project_path}")
print(f" Destination: {dest_path}")
# Create destination if needed
if not dry_run:
dest_path.mkdir(parents=True, exist_ok=True)
rsync_cmd = [
"rsync",
"-av",
"--progress",
*[f"--exclude={p}" for p in _RSYNC_EXCLUDES],
f"{project_path}/",
f"{dest_path}/",
]
if dry_run:
rsync_cmd.insert(1, "--dry-run")
print("\n [DRY RUN] Would execute:")
print(f" {' '.join(rsync_cmd)}")
else:
print("\n Syncing files...")
if verbose:
print(f" Command: {' '.join(rsync_cmd)}")
result = subprocess.run(rsync_cmd)
if result.returncode != 0:
print(f"Error: rsync failed with code {result.returncode}")
return 1
# Update project.json with synced_time
if not dry_run:
project_json_path = project_path / "project.json"
if project_json_path.exists():
try:
data = _read_json(project_json_path)
data["synced_time"] = datetime.now().isoformat()
project_json_path.write_text(
json.dumps(data, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
print(
f"\n Updated project.json with synced_time: {data['synced_time']}"
)
except (json.JSONDecodeError, IOError) as e:
print(f"Warning: Could not update project.json: {e}")
print("\nDone.")
return 0
def cmd_load(project_path: Path, verbose: bool, dry_run: bool) -> int:
"""Load project files from external cache storage onto the local drive."""
from .cache import load_cache_config, load_assets_config
print(f"Loading: {project_path.name}")
# Prefer [cache] disk; fall back to [assets] disk if cache isn't mounted.
cache_base = None
for label, candidate in (("cache", load_cache_config()), ("assets", load_assets_config())):
if candidate is not None and candidate.exists():
proj_candidate = candidate / project_path.name
if proj_candidate.exists():
cache_base = candidate
print(f" Using {label} disk: {candidate}")
break
if cache_base is None:
tried = []
for label, fn in (("cache", load_cache_config), ("assets", load_assets_config)):
p = fn()
if p is not None:
tried.append(f" [{label}] {p}")
if tried:
print(f"Error: Project '{project_path.name}' not found on any mounted external drive. Tried:")
for t in tried:
print(t)
else:
print("Error: No external drive configured. Create ~/.gnommo.conf with:")
print(" [cache]")
print(" path = /Volumes/YourDisk/gnommo")
return 1
# Build source path on the external drive
src_path = cache_base / project_path.name
if not src_path.exists():
print(f"Error: Project not found on external drive: {src_path}")
return 1
print(f" Source: {src_path}")
print(f" Destination: {project_path}")
# Create destination if needed
if not dry_run:
project_path.mkdir(parents=True, exist_ok=True)
rsync_cmd = [
"rsync",
"-av",
"--progress",
*[f"--exclude={p}" for p in _RSYNC_EXCLUDES],
f"{src_path}/",
f"{project_path}/",
]
if dry_run:
rsync_cmd.insert(1, "--dry-run")
print("\n [DRY RUN] Would execute:")
print(f" {' '.join(rsync_cmd)}")
else:
print("\n Copying files...")
if verbose:
print(f" Command: {' '.join(rsync_cmd)}")
result = subprocess.run(rsync_cmd)
if result.returncode != 0:
print(f"Error: rsync failed with code {result.returncode}")
return 1
print("\nDone.")
return 0
if __name__ == "__main__":
sys.exit(main())