Imrpoved resume for preprocess

gnommo/cli.py

@@ -2561,14 +2561,14 @@ def cmd_preprocess(
             gnommo_scratch = project_path / gnommo_scratch
         print(f"  Using intermediate dir: {gnommo_scratch}")
 
-    # Clear a segment's stale intermediate/scratch dir before (re)processing it.
+    # Prepare a segment's intermediate/scratch dir before (re)processing it.
-    # A run that crashed mid-file (e.g. the laptop battery dying with the external
+    # A run that crashed mid-file (e.g. the laptop battery dying) leaves behind a
-    # drive attached) leaves partial chunks, half-written batch files, and — for
+    # mix of COMPLETE chunks and half-written ones. We want to resume from the
-    # low/tiny res — a truncated raw downscale. create_downscaled_video reuses an
+    # completed chunks, so we do NOT wipe the dir wholesale: chunked processing
-    # existing raw_<res> file as-is, so a truncated one would silently corrupt the
+    # validates each chunk and only redoes the missing/partial ones, and all
-    # output. Wiping the scratch dir forces a clean restart of that file. It only
+    # encodes now write to a *.partial sibling that's renamed only on success.
-    # touches segments this run is about to redo, so complete outputs (files that
+    # Here we just sweep those orphaned *.partial files. With --force, the user
-    # already finished) and any concurrent run's in-flight files are left alone.
+    # is asking for a clean redo, so we clear the whole dir.
     import shutil as _shutil
 
     def _clear_segment_scratch(seg_videos_dir: Path, seg_id: str) -> None:
@@ -2577,11 +2577,25 @@ def cmd_preprocess(
             if gnommo_scratch
             else seg_videos_dir / "intermediate" / seg_id
         )
-        if scratch.exists():
+        if not scratch.exists():
-            print(
+            return
-                f"  Restarting {seg_id}: clearing incomplete intermediate files from previous run"
+        if force:
-            )
+            print(f"  {seg_id}: --force — clearing all intermediate files for a fresh run")
             _shutil.rmtree(scratch, ignore_errors=True)
+            return
+        # Resume mode: drop only half-written files, keep completed chunks so
+        # processing continues where the crashed run left off.
+        removed = 0
+        for p in scratch.rglob("*.partial.*"):
+            try:
+                p.unlink()
+                removed += 1
+            except OSError:
+                pass
+        if removed:
+            print(
+                f"  {seg_id}: removed {removed} incomplete file(s); resuming from completed chunks"
+            )
 
     # --- Filter pipeline ---
     talkinghead_filter = (config.default_filters or {}).get("talkinghead", [])

gnommo/preprocessor.py

@@ -1,5 +1,6 @@
 """Preprocessing stage: apply filters to source videos."""
 
+import json
 import os
 import subprocess
 import sys
@@ -151,6 +152,10 @@ def create_downscaled_video(
     if out_path.exists() and not force:
         return out_path
 
+    # Write to a .partial sibling and atomically rename on success, so a run that
+    # crashed mid-encode (e.g. laptop battery died) never leaves a truncated file
+    # at out_path that a resuming run would silently reuse as if complete.
+    partial = out_path.with_name(out_path.stem + ".partial" + out_path.suffix)
     cmd = [
         "ffmpeg",
         "-y",
@@ -170,16 +175,18 @@ def create_downscaled_video(
         "aac",  # re-encode audio so both streams share the same PTS origin,
         "-ar",  # avoiding the lip-sync drift caused by libx264 encoder delay
         "48000",  # when audio is copied with its original timestamps
-        str(out_path),
+        str(partial),
     ]
     result = subprocess.run(cmd, capture_output=True, text=True)
     if result.returncode != 0:
+        partial.unlink(missing_ok=True)
         raise PreprocessError(
             f"Failed to downscale {source_path.name} to {width}x{height}",
             filter_type="downscale",
             command=" ".join(cmd),
             stderr=result.stderr,
         )
+    os.replace(partial, out_path)
     return out_path
 
 
@@ -1366,6 +1373,35 @@ def parse_gnommokey_config(config: dict) -> GnommoKeyConfig:
     )
 
 
+def _chunk_is_valid(chunk_path: Path, expected_dur: float) -> bool:
+    """True if a chunk file exists and looks fully written (readable, ~right length).
+
+    Used to decide which chunks a resuming run can keep. A chunk that was mid-write
+    when the process died has no moov atom / a short duration, so ffprobe fails or
+    the duration is well under expected — either way it's rejected and redone.
+    """
+    if not chunk_path.exists() or chunk_path.stat().st_size == 0:
+        return False
+    dur = get_video_duration(chunk_path)  # 0.0 if unreadable
+    return dur > 0 and abs(dur - expected_dur) <= max(1.0, expected_dur * 0.05)
+
+
+def _chunk_fingerprint(input_path: Path, filters: list[dict], take) -> str:
+    """Identity of a chunk set: filters + source + chunk length. If any of these
+    change, previously written chunks are stale and must not be resumed."""
+    import hashlib
+    import json
+
+    try:
+        st = input_path.stat()
+        src = f"{input_path.name}:{st.st_size}:{int(st.st_mtime)}"
+    except OSError:
+        src = input_path.name
+    payload = json.dumps(filters, sort_keys=True, default=str)
+    payload += f"|{src}|{CHUNK_DURATION}|{take}"
+    return hashlib.sha1(payload.encode()).hexdigest()
+
+
 def apply_combined_video_filters_chunked(
     input_path: Path,
     output_path: Path,
@@ -1399,6 +1435,22 @@ def apply_combined_video_filters_chunked(
         scratch_dir = output_path.parent / "chunks"
     scratch_dir.mkdir(parents=True, exist_ok=True)
 
+    # Resume guard: if the filters or the source changed since the leftover chunks
+    # were written, they're stale — drop them and start clean. Otherwise we keep
+    # whatever completed chunks are on disk and only redo the missing ones.
+    fingerprint = _chunk_fingerprint(input_path, filters, take)
+    manifest_path = scratch_dir / ".chunks.json"
+    prev_fp = None
+    if manifest_path.exists():
+        try:
+            prev_fp = json.loads(manifest_path.read_text()).get("fingerprint")
+        except (json.JSONDecodeError, OSError):
+            prev_fp = None
+    if prev_fp != fingerprint:
+        for old in scratch_dir.glob("chunk_*.mov"):
+            old.unlink(missing_ok=True)
+        manifest_path.write_text(json.dumps({"fingerprint": fingerprint}))
+
     num_chunks = int(duration / CHUNK_DURATION) + 1
     chunk_files: list[Path] = []
     chunk_tasks: list[tuple] = []  # (index, chunk_path, start_time, chunk_duration)
@@ -1415,9 +1467,28 @@ def apply_combined_video_filters_chunked(
         chunk_files.append(chunk_path)
         chunk_tasks.append((i, chunk_path, start_time, chunk_duration))
 
-    num_workers = min(DEFAULT_CHUNK_WORKERS, len(chunk_tasks))
+    # Resume: keep chunks a previous run already finished; only redo the rest.
+    pending_tasks: list[tuple] = []
+    for task in chunk_tasks:
+        i, chunk_path, start_time, chunk_dur = task
+        if _chunk_is_valid(chunk_path, chunk_dur):
+            continue
+        chunk_path.unlink(missing_ok=True)  # drop a half-written chunk before redo
+        pending_tasks.append(task)
+
+    resumed = len(chunk_tasks) - len(pending_tasks)
+    if resumed and pending_tasks:
         print(
-        f"          Processing {len(chunk_tasks)} chunks in parallel ({num_workers} workers)"
+            f"          Resuming: {resumed}/{len(chunk_tasks)} chunk(s) already complete, "
+            f"processing the remaining {len(pending_tasks)}"
+        )
+    elif not pending_tasks:
+        print(f"          All {len(chunk_tasks)} chunks already complete — concatenating")
+
+    num_workers = min(DEFAULT_CHUNK_WORKERS, len(pending_tasks)) if pending_tasks else 1
+    if pending_tasks and resumed == 0:
+        print(
+            f"          Processing {len(pending_tasks)} chunks in parallel ({num_workers} workers)"
         )
 
     # Process chunks in parallel
@@ -1434,10 +1505,10 @@ def apply_combined_video_filters_chunked(
         )
         return i, chunk_path
 
-    completed = 0
+    completed = resumed
     with ThreadPoolExecutor(max_workers=num_workers) as executor:
         futures = {
-            executor.submit(process_chunk_task, task): task for task in chunk_tasks
+            executor.submit(process_chunk_task, task): task for task in pending_tasks
         }
         for future in as_completed(futures):
             i, chunk_path = future.result()
@@ -1477,11 +1548,13 @@ def apply_combined_video_filters_chunked(
             stderr=concat_result.stderr,
         )
 
-    # Clean up chunk files and concat list
+    # Clean up chunk files, concat list, and the resume manifest — only now that
+    # the batch output is safely produced.
     for chunk_path in chunk_files:
         if chunk_path.exists():
             chunk_path.unlink()
     concat_list.unlink(missing_ok=True)
+    manifest_path.unlink(missing_ok=True)
 
     # Remove chunks directory if empty
     try:
@@ -1578,7 +1651,11 @@ def _process_chunk_to_prores4444(
     else:
         cmd.append("-an")
 
-    cmd.append(str(output_path))
+    # Encode to a .partial sibling and atomically rename only after validation, so
+    # a chunk that appears at its final name is always complete. A crash mid-encode
+    # leaves the .partial behind (ignored by resume), never a half-written chunk.
+    partial = output_path.with_name(output_path.stem + ".partial" + output_path.suffix)
+    cmd.append(str(partial))
 
     if verbose:
         print(f"            Filter: {video_filter}")
@@ -1587,6 +1664,7 @@ def _process_chunk_to_prores4444(
     result = run_ffmpeg_with_progress(cmd, actual_take or chunk_duration, "Encoding")
 
     if result.returncode != 0:
+        partial.unlink(missing_ok=True)
         raise PreprocessError(
             "Chunk processing failed",
             filter_type="chunk",
@@ -1606,12 +1684,13 @@ def _process_chunk_to_prores4444(
             "format=duration",
             "-of",
             "csv=p=0",
-            str(output_path),
+            str(partial),
         ],
         capture_output=True,
         text=True,
     )
     if probe.returncode != 0 or not probe.stdout.strip():
+        partial.unlink(missing_ok=True)
         raise PreprocessError(
             f"Chunk output file is unreadable or missing moov atom: {output_path.name}",
             filter_type="chunk",
@@ -1619,6 +1698,8 @@ def _process_chunk_to_prores4444(
             stderr=probe.stderr,
         )
 
+    os.replace(partial, output_path)
+
 
 def _process_chunk_to_webm(
     input_path: Path,