From 5f974a710e816cea5a4d07b7e76504f99b98d404 Mon Sep 17 00:00:00 2001 From: jenstandstad Date: Thu, 23 Jul 2026 20:20:41 +0200 Subject: [PATCH] Imrpoved resume for preprocess --- gnommo/cli.py | 38 +++++++++++----- gnommo/preprocessor.py | 101 +++++++++++++++++++++++++++++++++++++---- 2 files changed, 117 insertions(+), 22 deletions(-) diff --git a/gnommo/cli.py b/gnommo/cli.py index eb7067e..0c2e0d6 100644 --- a/gnommo/cli.py +++ b/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. - # A run that crashed mid-file (e.g. the laptop battery dying with the external - # drive attached) leaves partial chunks, half-written batch files, and — for - # low/tiny res — a truncated raw downscale. create_downscaled_video reuses an - # existing raw_ file as-is, so a truncated one would silently corrupt the - # output. Wiping the scratch dir forces a clean restart of that file. It only - # touches segments this run is about to redo, so complete outputs (files that - # already finished) and any concurrent run's in-flight files are left alone. + # Prepare a segment's intermediate/scratch dir before (re)processing it. + # A run that crashed mid-file (e.g. the laptop battery dying) leaves behind a + # mix of COMPLETE chunks and half-written ones. We want to resume from the + # completed chunks, so we do NOT wipe the dir wholesale: chunked processing + # validates each chunk and only redoes the missing/partial ones, and all + # encodes now write to a *.partial sibling that's renamed only on success. + # Here we just sweep those orphaned *.partial files. With --force, the user + # 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(): - print( - f" Restarting {seg_id}: clearing incomplete intermediate files from previous run" - ) + if not scratch.exists(): + return + 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", []) diff --git a/gnommo/preprocessor.py b/gnommo/preprocessor.py index 6611437..4972cda 100644 --- a/gnommo/preprocessor.py +++ b/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,10 +1467,29 @@ 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)) - print( - f" Processing {len(chunk_tasks)} chunks in parallel ({num_workers} workers)" - ) + # 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" 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 def process_chunk_task(task): @@ -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,