From c6efbaf2ed742c645ec01621d38072922135c276 Mon Sep 17 00:00:00 2001 From: jenstandstad Date: Wed, 15 Jul 2026 08:31:33 +0200 Subject: [PATCH] Adding the state system --- gnommo/cli.py | 462 +++++++++++++++++++++++++++++++++++++++++++++--- gnommo/state.py | 149 ++++++++++++++++ 2 files changed, 582 insertions(+), 29 deletions(-) create mode 100644 gnommo/state.py diff --git a/gnommo/cli.py b/gnommo/cli.py index c7dc144..cb44f99 100644 --- a/gnommo/cli.py +++ b/gnommo/cli.py @@ -39,6 +39,8 @@ Examples: 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 stitch --res tiny -f Fast stitch with new begin/end values 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 @@ -113,6 +115,7 @@ Examples: "transcode", "pexels", "clear", + "prune", "new", ], help="Action to perform (default: render)", @@ -277,6 +280,8 @@ Examples: 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, @@ -2164,6 +2169,222 @@ def cmd_clear(project_path: Path, verbose: bool) -> int: 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 + ] + + if not to_remove: + if verbose: + print(f" {path.name}: all {len(data)} {label} entries in use.") + return 0 + + 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 not dry_run: + for key in to_remove: + del data[key] + _write_json_preserve(path, data) + + return len(to_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, @@ -2271,6 +2492,17 @@ def cmd_preprocess( 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 @@ -2294,10 +2526,18 @@ def cmd_preprocess( 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: - print(f" {segment_id}: output exists, skipping (use --force to reprocess)") - skipped_count += 1 - continue + 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, {}) @@ -2427,6 +2667,16 @@ def cmd_preprocess( 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))") @@ -2634,16 +2884,17 @@ def cmd_trim( 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] - existing = raw_data.get(seg_id, {}) - has_explicit = "skip" in existing or "take" in existing - if has_explicit and not force: - print(f" {seg_id}: already trimmed, skipping (use --force to redo)") - continue - # Prefer raw file; fall back to source_file from narration.json source_path = raw_lookup.get(seg_id) if source_path is None: @@ -2652,6 +2903,21 @@ def cmd_trim( 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, {}) + has_explicit = "skip" in existing or "take" in existing + if has_explicit and not seg_force: + print(f" {seg_id}: already trimmed, 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: @@ -2663,7 +2929,7 @@ def cmd_trim( transcript_path = transcripts_dir / f"{seg_id}.json" try: - if transcript_path.exists() and not force: + if transcript_path.exists() and not seg_force: words = load_transcript(transcript_path) print(f" {seg_id}: loaded cached transcript ({len(words)} words)", end="", flush=True) else: @@ -2706,6 +2972,7 @@ def cmd_trim( raw_data[seg_id]["skip"] = new_skip raw_data[seg_id]["take"] = new_take + _trim_fps[seg_id] = current_fp updated += 1 continue @@ -2737,6 +3004,7 @@ def cmd_trim( raw_data[seg_id]["skip"] = new_skip raw_data[seg_id]["take"] = new_take + _trim_fps[seg_id] = current_fp updated += 1 if updated > 0: @@ -2746,6 +3014,11 @@ def cmd_trim( 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 @@ -3147,6 +3420,10 @@ def cmd_stitch( print(" Run 'gnommo -p import' first to populate narration.json") return 1 + # narration.json (skip/take/order) drives stitch — capture its local path for + # fingerprinting before narration_dir is redirected to a cache/res subdir. + _local_narration_json = narration_dir / "narration.json" + # Get videos_dir for output if config and config.videos_path: videos_json_path = project_path / config.videos_path @@ -3191,10 +3468,29 @@ def cmd_stitch( stitch_output = videos_dir_out / "narration_combined.mov" - if stitch_output.exists() and not force: - print(f"\n Combined narration exists: {stitch_output.name}") - print(" (use --force to regenerate)") + # Stage-level staleness: skip when narration.json and every processed segment + # are unchanged since the last successful stitch AND the output still exists. + # A changed input (re-trim, reprocessed segment) auto-triggers a regenerate. + from . import state as _state + + _stitch_key = f"stitch:{res}" + _stitch_inputs = _state.compute( + [("narration.json", _local_narration_json, _state.HASH)] + + [ + (f"seg:{sid}", narration_dir / narration[sid].source_file, _state.META) + for sid in segment_ids + ] + ) + _stitch_current = _state.is_current( + project_path, _stitch_key, _stitch_inputs, [stitch_output] + ) + + if stitch_output.exists() and not force and _stitch_current: + print(f"\n Combined narration up to date: {stitch_output.name}") + print(" (inputs unchanged since last stitch — use --force to regenerate)") else: + if stitch_output.exists() and not force and not _stitch_current: + print("\n Inputs changed since last stitch — regenerating.") # Extract loudnorm config from talkinghead filter so stitch uses # per-project settings instead of hardcoded defaults. _loudnorm_cfg = None @@ -3217,6 +3513,20 @@ def cmd_stitch( if not cache_root: _import_videos(videos_dir_out, config, verbose) + # Record fingerprints (recompute output META now that it exists) so the + # next run can detect whether inputs changed. + _state.record( + project_path, + _stitch_key, + _state.compute( + [("narration.json", _local_narration_json, _state.HASH)] + + [ + (f"seg:{sid}", narration_dir / narration[sid].source_file, _state.META) + for sid in segment_ids + ] + ), + ) + # Always update the MAIN videos.json (parent of subdir when using low/tiny res) # Downscaled dirs only affect file paths, not JSON metadata updates main_videos_dir = ( @@ -4057,10 +4367,50 @@ def cmd_render( 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), + ("transcript", transcript_path, _state.HASH), + ] + if resolved_combined: + specs.append(("narration_combined", resolved_combined, _state.META)) + for _sid, _sdef in slides.items(): + specs.append((f"slide:{_sid}", _slides_dir / _sdef.image, _state.META)) + return specs + + _render_key = f"render:{res}" + if _render_gateable and _state.is_current( + project_path, _render_key, _state.compute(_render_input_specs()), [output_path] + ): + 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 @@ -4348,6 +4698,46 @@ def _files_modified_since(root: Path, since: float, pattern: str) -> bool: 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, @@ -4355,7 +4745,7 @@ def cmd_all( res: str = "full", force: bool = False, ) -> int: - """Run full pipeline: import → preprocess → trim → stitch → render → push → handoff → up. + """Run full pipeline: import → prune → preprocess → trim → stitch → 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. @@ -4370,7 +4760,7 @@ def cmd_all( # True so all downstream stages re-run unconditionally. cascade_force = force - print(">>> Step 1/8: Import\n") + print(">>> Step 1/9: Import\n") t0 = time.time() result = cmd_import(project_path, cascade_force, verbose) if result != 0: @@ -4380,7 +4770,14 @@ def cmd_all( ): cascade_force = True - print("\n>>> Step 2/8: Preprocess\n") + print("\n>>> Step 2/9: 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/9: Preprocess\n") t0 = time.time() result = cmd_preprocess( project_path, verbose, dry_run, cascade_force, workers=1, res=res @@ -4392,16 +4789,23 @@ def cmd_all( ) or _files_modified_since(project_path, t0, "*_processed.webm"): cascade_force = True - print("\n>>> Step 3/8: Trim\n") - 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 stitch - if _files_modified_since(project_path, t0, "narration.json"): - cascade_force = True + print("\n>>> Step 4/9: 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 stitch + if _files_modified_since(project_path, t0, "narration.json"): + cascade_force = True - print("\n>>> Step 4/8: Stitch\n") + print("\n>>> Step 5/9: Stitch\n") t0 = time.time() result = cmd_stitch(project_path, verbose, cascade_force, res=res) if result != 0: @@ -4409,22 +4813,22 @@ def cmd_all( if _files_modified_since(project_path, t0, "narration_combined.mov"): cascade_force = True - print("\n>>> Step 5/8: Render\n") + print("\n>>> Step 6/9: 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") + print("\n>>> Step 7/9: Push\n") result = cmd_push(project_path, verbose, force=False, prod=True) if result != 0: return result - print("\n>>> Step 7/8: Handoff\n") + print("\n>>> Step 8/9: 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") + print("\n>>> Step 9/9: Upload\n") from .transfer import cmd_up return cmd_up(project_path, verbose, dry_run) diff --git a/gnommo/state.py b/gnommo/state.py new file mode 100644 index 0000000..acb28eb --- /dev/null +++ b/gnommo/state.py @@ -0,0 +1,149 @@ +"""Persistent per-stage completion tracking. + +Each pipeline stage (preprocess, trim, stitch, render) records a fingerprint of +its inputs in ``.gnommo_state.json`` when it completes successfully. On the next +run a stage can ask whether its inputs are unchanged (and its output still +present) and skip the work — the same staleness intelligence that ``all``'s +in-memory cascade provides, but persisted so it also applies to stages run on +their own. + +Fingerprinting is hybrid: + - small text manifests (narration.json, videos.json, manuscript.txt, + project.json, slides.json, audio.json, transcripts) are hashed (sha256) so a + ``touch`` or a git checkout that only rewrites mtimes doesn't force a + needless rerun; + - large media (processed segments, narration_combined.mov, source videos and + images) use mtime+size, which is cheap and good enough to detect real edits. + +The state file is purely an optimization: any read/parse/write failure degrades +to "not current" (rerun) and never raises, so a corrupt or missing state file +can't break a build. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Optional, Union + +STATE_FILENAME = ".gnommo_state.json" +STATE_VERSION = 1 + +# Fingerprint modes +HASH = "hash" # sha256 of file contents — for small text manifests +META = "meta" # mtime_ns + size — for large media + +# An input descriptor is a (label, path, mode) triple. +InputSpec = tuple[str, Path, str] + + +def _state_path(project_path: Path) -> Path: + return project_path / STATE_FILENAME + + +def _empty_state() -> dict: + return {"version": STATE_VERSION, "stages": {}} + + +def load_state(project_path: Path) -> dict: + """Load the state file, returning an empty skeleton on any problem.""" + path = _state_path(project_path) + if not path.exists(): + return _empty_state() + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return _empty_state() + if not isinstance(data, dict): + return _empty_state() + data.setdefault("version", STATE_VERSION) + if not isinstance(data.get("stages"), dict): + data["stages"] = {} + return data + + +def save_state(project_path: Path, state: dict) -> None: + """Write the state file. Never raises — state is best-effort.""" + try: + _state_path(project_path).write_text( + json.dumps(state, indent=2) + "\n", encoding="utf-8" + ) + except OSError: + pass + + +def fingerprint_path(path: Union[str, Path], mode: str) -> Optional[str]: + """Return a fingerprint for a single file, or None if it can't be read.""" + p = Path(path) + try: + if mode == HASH: + h = hashlib.sha256() + with open(p, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + h.update(chunk) + return f"sha256:{h.hexdigest()}" + st = p.stat() + return f"meta:{st.st_mtime_ns}:{st.st_size}" + except OSError: + return None + + +def compute(inputs: Iterable[InputSpec]) -> dict: + """Build a {label: fingerprint} map from (label, path, mode) triples. + + A missing file yields a null fingerprint, so a file appearing or disappearing + counts as a change. + """ + return {label: fingerprint_path(path, mode) for label, path, mode in inputs} + + +def get_stage(project_path: Path, stage_key: str) -> dict: + """Return the recorded record for a stage (``{}`` if none).""" + return load_state(project_path).get("stages", {}).get(stage_key, {}) + + +def get_items(project_path: Path, stage_key: str) -> dict: + """Return the per-item fingerprint map recorded for a stage (``{}`` if none).""" + items = get_stage(project_path, stage_key).get("items") + return items if isinstance(items, dict) else {} + + +def is_current( + project_path: Path, + stage_key: str, + inputs: dict, + outputs: Iterable[Union[str, Path]] = (), +) -> bool: + """True iff the recorded input fingerprint matches ``inputs`` exactly and + every output in ``outputs`` exists on disk.""" + for out in outputs: + if not Path(out).exists(): + return False + recorded = get_stage(project_path, stage_key).get("inputs") + return recorded == inputs + + +def record(project_path: Path, stage_key: str, inputs: dict) -> None: + """Persist a stage-level input fingerprint, marking the stage complete.""" + state = load_state(project_path) + state.setdefault("stages", {})[stage_key] = { + "completed_at": datetime.now(timezone.utc).isoformat(), + "inputs": inputs, + } + save_state(project_path, state) + + +def record_items(project_path: Path, stage_key: str, items: dict) -> None: + """Merge per-item fingerprints into a stage's record (for multi-segment + stages like preprocess/trim). Existing items for other keys are preserved.""" + state = load_state(project_path) + stage = state.setdefault("stages", {}).setdefault(stage_key, {}) + stage["completed_at"] = datetime.now(timezone.utc).isoformat() + merged = stage.get("items") + if not isinstance(merged, dict): + merged = {} + merged.update(items) + stage["items"] = merged + save_state(project_path, state)