Preserve completed episodes from failed batches

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.52.14 - 2026-08-18
+
+- Moved each completed episode from staging into the configured downloads library as soon as the downloader reports that episode finished.
+- Preserved already-finished episodes from failed batch jobs instead of deleting them with the failed job staging folder.
+- Added regression coverage for incremental episode finalization and failed batch preservation.
+
 ## 0.52.13 - 2026-08-16
 
 - Fixed TV library finalization so staged provider files named with `SxxEyy` keep the downloaded episode number instead of falling back to episode 1.

README.md

@@ -100,7 +100,7 @@ Keep `./.kaizoku` mounted for production instances. That directory contains the
 
 ## Download Flow
 
-Kaizoku stores provider-backed show IDs as `provider:id`, for example `anikoto:some-show-slug`. Queue jobs resolve the episode source through `providers/bridge.js`, then `provider_downloader.py` downloads the media with `ffmpeg` into a staging directory. If a provider returns a master HLS playlist, Kaizoku selects the highest-bandwidth variant before starting `ffmpeg`. When a media playlist uses extensionless CDN segments, Kaizoku skips direct ffmpeg and uses a StrawVerse-style segment downloader: it fetches the media playlist, downloads and concatenates segments itself, strips short PNG wrappers when present, retries temporary HTTP failures such as `429 Too Many Requests`, then remuxes the local transport stream to MP4. Direct ffmpeg attempts also have a timeout guard so stalled HLS inputs can fall back cleanly. Each episode is written as a temporary `.mp4.part` file and moved into place only after the download succeeds, and finalization preserves episode numbers from staged `SxxEyy` or `Episode yy` filenames before applying configured season/episode offsets.
+Kaizoku stores provider-backed show IDs as `provider:id`, for example `anikoto:some-show-slug`. Queue jobs resolve the episode source through `providers/bridge.js`, then `provider_downloader.py` downloads the media with `ffmpeg` into a staging directory. If a provider returns a master HLS playlist, Kaizoku selects the highest-bandwidth variant before starting `ffmpeg`. When a media playlist uses extensionless CDN segments, Kaizoku skips direct ffmpeg and uses a StrawVerse-style segment downloader: it fetches the media playlist, downloads and concatenates segments itself, strips short PNG wrappers when present, retries temporary HTTP failures such as `429 Too Many Requests`, then remuxes the local transport stream to MP4. Direct ffmpeg attempts also have a timeout guard so stalled HLS inputs can fall back cleanly. Each episode is written as a temporary `.mp4.part` file and moved into the downloads library after that episode succeeds, so already-finished episodes from a larger batch survive if a later episode fails. Finalization preserves episode numbers from staged `SxxEyy` or `Episode yy` filenames before applying configured season/episode offsets.
 
 If the primary provider cannot list, resolve, or download a requested episode, Kaizoku searches the same title on the remaining providers and tries the matching episode there. Existing finalization code moves staged files into the configured library layout, preserving data already present in production download folders.
 

VERSION

@@ -1 +1 @@
-0.52.13
+0.52.14

app_support.py

@@ -1037,6 +1037,16 @@ def extract_episode_from_filename(path):
     return match.group(1) if match else None
 
 
+def episode_match_key(value):
+    text = str(value or "").strip().lower()
+    match = re.match(r"^0*([0-9]+)(.*)$", text)
+    if not match:
+        return text
+    number = int(match.group(1) or "0", 10)
+    suffix = str(match.group(2) or "")
+    return f"{number}{suffix}" if number > 0 else text
+
+
 def unique_destination(path):
     if not path.exists():
         return path
@@ -1048,19 +1058,44 @@ def unique_destination(path):
         counter += 1
 
 
-def finalize_library_files(job):
+def cleanup_empty_staging_dirs(staging_dir, remove_root=False):
+    for path in sorted(staging_dir.rglob("*"), key=lambda item: len(item.parts), reverse=True):
+        if path.is_dir():
+            try:
+                path.rmdir()
+            except OSError:
+                pass
+    if remove_root:
+        try:
+            staging_dir.rmdir()
+        except OSError:
+            pass
+
+
+def finalize_library_files(job, episode=None, cleanup_staging=True, allow_empty=False):
     staging_dir = job_staging_dir(job)
     target_dir = job_output_dir(job)
     library_name = sanitize_path_component(job.get("anime_name") or job.get("title"))
     media_type = normalize_media_type(job.get("media_type"))
     season = season_label(job)
     episode_offset = normalize_episode_offset(job.get("episode_offset"))
+    episode_filter = episode_match_key(episode) if episode not in (None, "") else None
     target_dir.mkdir(parents=True, exist_ok=True)
-    files = sorted(
+    files = []
-        [path for path in staging_dir.rglob("*") if path.is_file()],
+    for path in staging_dir.rglob("*"):
-        key=lambda path: (path.stat().st_mtime, str(path.relative_to(staging_dir)).lower()),
+        if not path.is_file() or str(path.name).endswith(".part"):
-    )
+            continue
+        if episode_filter is not None:
+            extracted = extract_episode_from_filename(path)
+            if episode_match_key(extracted) != episode_filter:
+                continue
+        files.append(path)
+    files = sorted(files, key=lambda path: (path.stat().st_mtime, str(path.relative_to(staging_dir)).lower()))
     if not files:
+        if allow_empty:
+            if cleanup_staging:
+                shutil.rmtree(staging_dir, ignore_errors=True)
+            return []
         raise RuntimeError("No downloaded files were found in the staging folder")
 
     moved = []
@@ -1081,7 +1116,10 @@ def finalize_library_files(job):
         shutil.move(str(source), str(destination))
         moved.append(str(destination))
 
+    if cleanup_staging:
         shutil.rmtree(staging_dir, ignore_errors=True)
+    else:
+        cleanup_empty_staging_dirs(staging_dir, remove_root=False)
     return moved
 
 

queue_jobs.py

@@ -1135,6 +1135,9 @@ class DownloadQueue:
         clean.pop("_last_persist_monotonic", None)
         clean.pop("_dirty_log_lines", None)
         clean.pop("log", None)
+        for key in list(clean):
+            if str(key).startswith("_"):
+                clean.pop(key, None)
         for key in ("id", "status", "created_at", "updated_at"):
             clean.pop(key, None)
         for key in JOB_STRUCTURED_COLUMNS:
@@ -1491,10 +1494,22 @@ class DownloadQueue:
             if not text:
                 text = f"Download progress: {progress.get('percent', 0)}%."
         self._stdout_log(job, text)
+        moved_files = []
+        finalize_error = None
+        if progress is not None:
+            try:
+                moved_files = self._finalize_progress_episode(job, progress)
+            except Exception as exc:
+                finalize_error = exc
         with self.lock:
             if progress is not None:
                 job["progress"] = progress
             job.setdefault("log", []).append(text)
+            if finalize_error is not None:
+                job["_incremental_finalize_error"] = str(finalize_error)
+                job.setdefault("log", []).append(f"Episode library move failed: {finalize_error}")
+            for path in moved_files:
+                job.setdefault("log", []).append(f"Saved: {path}")
             job["log"] = job["log"][-MAX_LOG_LINES:]
             job["updated_at"] = now_iso()
             job["_dirty_log_lines"] = int(job.get("_dirty_log_lines") or 0) + 1
@@ -1516,6 +1531,22 @@ class DownloadQueue:
         backend = str((job or {}).get("download_backend") or "download")
         print(f"[download:{job_id}:{backend}] {text}", flush=True)
 
+    def _finalize_progress_episode(self, job, progress):
+        if str((progress or {}).get("phase") or "").strip().lower() != "done":
+            return []
+        episode = (progress or {}).get("episode")
+        if episode in (None, ""):
+            return []
+        episode_key = str(episode).strip()
+        finalized = job.setdefault("_incremental_finalized_episodes", [])
+        if episode_key in finalized:
+            return []
+        moved = finalize_library_files(job, episode=episode, cleanup_staging=False, allow_empty=True)
+        if moved:
+            finalized.append(episode_key)
+            job.setdefault("_moved_files", []).extend(moved)
+        return moved
+
     def _cleanup_staging_dir(self, job):
         staging_dir = job_staging_dir(job)
         if not staging_dir.exists():
@@ -1741,7 +1772,8 @@ class DownloadQueue:
 
                 if exit_code == 0:
                     try:
-                        moved_files = finalize_library_files(job)
+                        moved_files = list(job.get("_moved_files") or [])
+                        moved_files.extend(finalize_library_files(job, allow_empty=bool(moved_files)))
                     except Exception as exc:
                         finalize_error = exc
                         no_files_downloaded = "No downloaded files were found in the staging folder" in str(exc)
@@ -1796,7 +1828,9 @@ class DownloadQueue:
                     if successful_backend and successful_backend != attempts[0]["backend"]:
                         job.setdefault("log", []).append(f"Fallback download completed with {successful_backend}.")
                     for path in moved_files:
-                        job.setdefault("log", []).append(f"Saved: {path}")
+                        saved_line = f"Saved: {path}"
+                        if saved_line not in (job.get("log") or []):
+                            job.setdefault("log", []).append(saved_line)
                     job.setdefault("log", []).append("Download completed.")
                     if watchlist_sync_error:
                         job.setdefault("log", []).append(f"Watchlist sync failed: {watchlist_sync_error}")

test_app.py

@@ -880,6 +880,36 @@ class QueueApiTests(unittest.TestCase):
             )
             self.assertTrue(Path(moved[0]).exists())
 
+    def test_tv_finalizer_can_move_one_finished_episode_without_removing_staging(self):
+        with tempfile.TemporaryDirectory() as temp_root:
+            job = APP.app_support.build_job(
+                {
+                    "query": "Batch Show",
+                    "title": "Batch Show",
+                    "anime_name": "Batch Show",
+                    "media_type": "tv",
+                    "mode": "sub",
+                    "quality": "best",
+                    "episodes": "1-2",
+                    "download_dir": temp_root,
+                    "season": "1",
+                },
+                {"mode": "sub", "quality": "best", "download_dir": temp_root},
+            )
+            staging_dir = APP.app_support.job_staging_dir(job)
+            staging_dir.mkdir(parents=True, exist_ok=True)
+            first = staging_dir / "Batch Show - S01E01.mp4"
+            second = staging_dir / "Batch Show - S01E02.mp4"
+            first.write_bytes(b"ep1")
+            second.write_bytes(b"ep2")
+
+            moved = APP.app_support.finalize_library_files(job, episode=1, cleanup_staging=False)
+
+            self.assertEqual(moved, [f"{temp_root}/tv/Batch Show/Season 01/Batch Show - S01E01.mp4"])
+            self.assertTrue(staging_dir.exists())
+            self.assertFalse(first.exists())
+            self.assertTrue(second.exists())
+
     def test_shutdown_wait_cancels_active_process_for_deterministic_teardown(self):
         queue = object.__new__(APP.DownloadQueue)
         queue.lock = threading.RLock()
@@ -1033,6 +1063,43 @@ class DownloadQueueWorkerFailureTests(unittest.TestCase):
         self.assertEqual(send_webhook.call_args.args[2]["category"], "finished")
         self.assertEqual(send_webhook.call_args.args[2]["moved_files"], ["/tmp/example/Queue Show - S01E01.mp4"])
 
+    def test_failed_batch_preserves_episode_moved_after_progress_done(self):
+        with tempfile.TemporaryDirectory() as temp_root:
+            queue = APP.DownloadQueue(
+                lambda: {
+                    "mode": "sub",
+                    "quality": "best",
+                    "download_dir": temp_root,
+                },
+                start_worker=False,
+            )
+            job = queue.add({"query": "Batch Show", "title": "Batch Show", "season": "1", "episodes": "1-2"})
+            staging_dir = queue_jobs.job_staging_dir(job)
+            staging_dir.mkdir(parents=True, exist_ok=True)
+            staged_episode = staging_dir / "Batch Show - S01E01.mp4"
+            staged_episode.write_bytes(b"ep1")
+
+            class FakeProc:
+                pid = 4321
+                stdout = [
+                    'KAIZOKU_PROGRESS {"episode":1,"episode_index":1,"episode_total":2,"message":"Episode 1 saved.","percent":100,"phase":"done"}\n',
+                    "Episode 2 failed on all providers: boom\n",
+                ]
+
+                def wait(self):
+                    return 1
+
+            with mock.patch.object(queue_jobs.subprocess, "Popen", return_value=FakeProc()):
+                queue._run_job(job)
+
+            stored = queue._find(job["id"])
+            final_episode = Path(temp_root) / "tv" / "Batch Show" / "Season 01" / "Batch Show - S01E01.mp4"
+            self.assertEqual(stored["status"], "failed")
+            self.assertTrue(final_episode.exists())
+            self.assertFalse(staged_episode.exists())
+            self.assertFalse(staging_dir.exists())
+            self.assertIn(f"Saved: {final_episode}", stored["log"])
+
     def test_successful_detached_download_logs_sync_skipped(self):
         queue = APP.DownloadQueue(
             lambda: {