fix: preserve existing subtitles instead of overwriting on move

Subtitles already placed next to a video are now preserved (marked _tested)
and the extracted subs/ archive is kept rather than cleaned up. This was
always the intended behaviour; moving a new best subtitle should not silently
overwrite or discard a previously placed one.

ㅤ

Author: vagabondHustler

src/subsearch/core/pipeline.py

@@ -226,7 +226,9 @@ def subtitle_rename(self) -> None:
 
     @run_if_conditions_met
     def subtitle_move_best(self, target: Path) -> None:
-        file_system.move_and_replace(self.bootstrap.autoload_src, target)
+        file_system.move_best_next_to_video(
+            self.bootstrap.autoload_src, target, SEARCH_SUBJECT.search_term, WORKSPACE.extraction_directory
+        )
         capture("Best subtitle moved")
 
     @run_if_conditions_met
@@ -288,16 +290,12 @@ def _elapsed(self) -> float:
 
     @run_if_conditions_met
     def clean_up(self) -> None:
+        # subs/ is the kept extraction archive; only tmp_subsearch and its downloads are removed.
         file_system.del_directory_content(APP_PATHS.tmp_dir)
         tracker = get_file_tracker()
         if WORKSPACE.download_directory != Path(""):
-            tracker.delete_tracked_within(WORKSPACE.extraction_directory, "*.nfo")
             tracker.delete_tracked_within(WORKSPACE.download_directory)
             tracker.delete_if_tracked(WORKSPACE.download_directory)
-            if WORKSPACE.extraction_directory.is_dir() and file_system.directory_is_empty(
-                WORKSPACE.extraction_directory
-            ):
-                tracker.delete_if_tracked(WORKSPACE.extraction_directory)
         capture("Cleanup completed")
 
     def _wait_for_terminal_input(self) -> None:

src/subsearch/io/file_system.py

@@ -1,3 +1,4 @@
+import hashlib
 import json
 import re
 import shutil
@@ -60,6 +61,9 @@ def is_zip_payload(chunk: bytes) -> bool:
 
 _HASH_MATCH_PREFIX = "hashmatch__"
 
+# marks a subtitle that was already placed next to the video and later displaced, so it is not re-selected
+_TESTED_MARKER = "_tested"
+
 
 def _cloudflare_block_reason(response: "CurlResponse") -> str:
     headers = response.headers
@@ -158,6 +162,24 @@ def _archive_listing(archive: zipfile.ZipFile) -> str:
     return "\n".join(f"  {info.filename} ({info.file_size} bytes)" for info in archive.infolist())
 
 
+def content_hash(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def file_content_hash(path: Path) -> str:
+    return content_hash(path.read_bytes())
+
+
+def _resolve_extraction_target(destination: Path, filename: str, payload: bytes) -> Path | None:
+    target = destination / Path(filename).name
+    if not target.exists():
+        return target
+    if file_content_hash(target) == content_hash(payload):
+        capture(f"Skipping {target.name} (identical copy already extracted)", level=LogLevel.DEBUG)
+        return None
+    return _next_available_path(destination, target.stem, target.suffix)
+
+
 def _safe_extract_archive(archive: zipfile.ZipFile, dst: Path, hash_match: bool = False) -> int:
     members = archive.infolist()
     total_uncompressed = sum(info.file_size for info in members)
@@ -180,12 +202,15 @@ def _safe_extract_archive(archive: zipfile.ZipFile, dst: Path, hash_match: bool
         if member_path.suffix.lower() not in _SUBTITLE_EXTENSIONS:
             capture(f"Ignoring {member.filename} (not a subtitle file)", level=LogLevel.DEBUG)
             continue
-        extracted_path = Path(archive.extract(member, dst))
-        if hash_match and not extracted_path.name.startswith(_HASH_MATCH_PREFIX):
-            renamed_path = extracted_path.with_name(f"{_HASH_MATCH_PREFIX}{extracted_path.name}")
-            extracted_path.rename(renamed_path)
-            extracted_path = renamed_path
-        _track(extracted_path)
+        payload = archive.read(member)
+        member_name = Path(member.filename).name
+        if hash_match and not member_name.startswith(_HASH_MATCH_PREFIX):
+            member_name = f"{_HASH_MATCH_PREFIX}{member_name}"
+        target = _resolve_extraction_target(dst, member_name, payload)
+        if target is None:
+            continue
+        target.write_bytes(payload)
+        _track(target)
         extracted_count += 1
     return extracted_count
 
@@ -215,11 +240,17 @@ def extract_subtitle_by_id(subtitle_id: str, src: Path, dst: Path, extension: st
     return extracted_count
 
 
+def _is_tested_subtitle(stem: str) -> bool:
+    return stem.endswith(_TESTED_MARKER) or f"{_TESTED_MARKER}_v" in stem
+
+
 def subtitle_files_in(directory: Path) -> list[Path]:
     if not directory.is_dir():
         return []
     return sorted(
-        file for file in directory.iterdir() if file.is_file() and file.suffix.lower() in _SUBTITLE_EXTENSIONS
+        file
+        for file in directory.iterdir()
+        if file.is_file() and file.suffix.lower() in _SUBTITLE_EXTENSIONS and not _is_tested_subtitle(file.stem)
     )
 
 
@@ -265,9 +296,21 @@ def move_all(src: Path, dst: Path) -> int:
     return moved_count
 
 
-def move_and_replace(source_file: Path, destination_directory: Path) -> None:
-    source_file.replace(destination_directory / source_file.name)
-    capture(f"Moving file: {source_file} -> {destination_directory}")
+def move_best_next_to_video(
+    source_file: Path, destination_directory: Path, video_stem: str, extraction_directory: Path
+) -> None:
+    if _is_tested_subtitle(source_file.stem):
+        capture(f"Refusing to move already-tested subtitle: {source_file}", level=LogLevel.DEBUG)
+        return
+    target = destination_directory / f"{video_stem}{source_file.suffix}"
+    if target.exists():
+        preserved = _next_available_path(extraction_directory, f"{video_stem}{_TESTED_MARKER}", target.suffix)
+        capture(f"Preserving existing subtitle: {target} -> {preserved}")
+        target.replace(preserved)
+        _track(preserved)
+    capture(f"Moving file: {source_file} -> {target}")
+    source_file.replace(target)
+    _track(target)
 
 
 def del_file_type(cwd: Path, extension: str) -> None:

src/subsearch/ui/services/post_processing.py

@@ -46,9 +46,9 @@ def _count_already_placed(self) -> int:
     def _rename_and_place(self) -> int:
         target_path = self._resolve_target_path()
         renamed = subtitle_selection.autoload_rename(SEARCH_SUBJECT.search_term, self._extraction_dir)
-        file_system.move_and_replace(renamed, target_path)
-        placed = (target_path / renamed.name).exists()
-        if not placed:
+        placed_name = f"{SEARCH_SUBJECT.search_term}{renamed.suffix}"
+        file_system.move_best_next_to_video(renamed, target_path, SEARCH_SUBJECT.search_term, self._extraction_dir)
+        if not (target_path / placed_name).exists():
             self._log_no_files(0)
             return 0
         capture(f"Moved 1 subtitles to {target_path}")

tests/test_file_system.py

@@ -116,6 +116,74 @@ def test_extract_files_in_dir_returns_extracted_subtitle_count(tmp_path) -> None
     assert sorted(path.name for path in destination.iterdir()) == ["movie.ass", "movie.srt"]
 
 
+def test_extract_skips_identical_content_already_in_destination(tmp_path) -> None:
+    body = b"1\n00:00:00,000 --> 00:00:01,000\nfoo\n"
+    destination = tmp_path / "subs"
+    destination.mkdir()
+    (destination / "movie.srt").write_bytes(body)
+    with zipfile.ZipFile(tmp_path / "download.zip", "w") as archive:
+        archive.writestr("movie.srt", body)
+
+    extracted_count = file_system.extract_files_in_dir(tmp_path, destination)
+
+    assert extracted_count == 0
+    assert [path.name for path in destination.iterdir()] == ["movie.srt"]
+
+
+def test_extract_versions_same_name_different_content(tmp_path) -> None:
+    destination = tmp_path / "subs"
+    destination.mkdir()
+    (destination / "movie.srt").write_bytes(b"existing content\n")
+    with zipfile.ZipFile(tmp_path / "download.zip", "w") as archive:
+        archive.writestr("movie.srt", b"different content\n")
+
+    extracted_count = file_system.extract_files_in_dir(tmp_path, destination)
+
+    assert extracted_count == 1
+    assert sorted(path.name for path in destination.iterdir()) == ["movie.srt", "movie_v1.srt"]
+
+
+def test_move_best_uses_video_name_and_preserves_existing_into_subs(tmp_path) -> None:
+    video_directory = tmp_path / "video"
+    subs_directory = tmp_path / "subs"
+    video_directory.mkdir()
+    subs_directory.mkdir()
+    video_stem = "the.foo.bar.2021.1080p.web.h264-foobar"
+    (video_directory / f"{video_stem}.srt").write_text("old subtitle\n", encoding="utf-8")
+    source = _make_srt(subs_directory, "downloaded.srt")
+
+    file_system.move_best_next_to_video(source, video_directory, video_stem, subs_directory)
+
+    assert [path.name for path in video_directory.glob("*.srt")] == [f"{video_stem}.srt"]
+    assert (video_directory / f"{video_stem}.srt").read_text(encoding="utf-8").startswith("1\n")
+    assert (subs_directory / f"{video_stem}_tested.srt").read_text(encoding="utf-8") == "old subtitle\n"
+
+
+def test_move_best_refuses_to_move_a_tested_subtitle(tmp_path) -> None:
+    video_directory = tmp_path / "video"
+    subs_directory = tmp_path / "subs"
+    video_directory.mkdir()
+    subs_directory.mkdir()
+    video_stem = "the.foo.bar.2021.1080p.web.h264-foobar"
+    tested_source = _make_srt(subs_directory, f"{video_stem}_tested_v1.srt")
+
+    file_system.move_best_next_to_video(tested_source, video_directory, video_stem, subs_directory)
+
+    assert list(video_directory.glob("*.srt")) == []
+    assert tested_source.exists()
+
+
+def test_tested_subtitles_are_excluded_from_candidates(tmp_path) -> None:
+    video_stem = "the.foo.bar.2021.1080p.web.h264-foobar"
+    _make_srt(tmp_path, f"{video_stem}.srt")
+    _make_srt(tmp_path, f"{video_stem}_tested.srt")
+    _make_srt(tmp_path, f"{video_stem}_tested_v1.srt")
+
+    candidates = [path.name for path in file_system.subtitle_files_in(tmp_path)]
+
+    assert candidates == [f"{video_stem}.srt"]
+
+
 def test_extract_subtitle_by_id_unpacks_only_the_matching_archive(tmp_path) -> None:
     downloads = tmp_path / "downloads"
     downloads.mkdir()

tests/test_release_validation_body.py

@@ -16,35 +16,35 @@ def validation() -> ReleaseValidation:
 
 
 def test_block_inserted_into_empty_body(validation: ReleaseValidation) -> None:
-    block = validation._body_block("abc123", "passed", "999")
+    block = validation._body_block("passed", "999")
     body = validation._replace_body_block("", block)
     assert body.startswith(validation.BODY_BLOCK_START)
     assert "passed" in body
 
 
 def test_block_prepended_above_existing_body(validation: ReleaseValidation) -> None:
     existing = "#### Features:\n- thing"
-    body = validation._replace_body_block(existing, validation._body_block("abc123", "passed", "999"))
+    body = validation._replace_body_block(existing, validation._body_block("passed", "999"))
     assert body.startswith(validation.BODY_BLOCK_START)
     assert body.rstrip().endswith("- thing")
 
 
 def test_block_replaced_in_place_without_duplication(validation: ReleaseValidation) -> None:
-    first = validation._replace_body_block("- thing", validation._body_block("abc123", "passed", "999"))
-    second = validation._replace_body_block(first, validation._body_block("def456", "failed", "1000"))
+    first = validation._replace_body_block("- thing", validation._body_block("passed", "999"))
+    second = validation._replace_body_block(first, validation._body_block("failed", "1000"))
     assert second.count(validation.BODY_BLOCK_START) == 1
-    assert "def456" in second and "failed" in second
-    assert "abc123" not in second
+    assert "1000" in second and "failed" in second
+    assert "999" not in second
     assert second.rstrip().endswith("- thing")
 
 
 def test_jobs_preserves_validation_block_across_regeneration(validation: ReleaseValidation) -> None:
     pull_request = OpenMainPullRequest()
-    existing = validation._replace_body_block("old changelog", validation._body_block("abc123", "passed", "999"))
+    existing = validation._replace_body_block("old changelog", validation._body_block("passed", "999"))
     regenerated = pull_request._pr_body("#### NEW CHANGELOG", existing)
     assert regenerated.startswith(pull_request.VALIDATION_BLOCK_START)
     assert regenerated.count(pull_request.VALIDATION_BLOCK_START) == 1
-    assert "abc123" in regenerated
+    assert "999" in regenerated
     assert "#### NEW CHANGELOG" in regenerated
 
 

tests/test_ui_post_processing.py

@@ -33,7 +33,9 @@ def _run_unpack_worker(monkeypatch, rename: bool, extracted_count: int):
     monkeypatch.setattr(
         post_processing.subtitle_selection, "autoload_rename", lambda search_term, src: src / "renamed.srt"
     )
-    monkeypatch.setattr(post_processing.file_system, "move_and_replace", lambda src, dst: None)
+    monkeypatch.setattr(
+        post_processing.file_system, "move_best_next_to_video", lambda src, dst, video_stem, subs_dir: None
+    )
     monkeypatch.setattr(post_processing.file_system, "create_path_from_string", lambda *args, **kwargs: Path("target"))
     worker = post_processing._PostProcessWorker(rename=rename, store=SettingsStore(), subtitle=_make_subtitle())
     return worker.execute()
@@ -79,16 +81,16 @@ def test_rename_and_place_delivers_to_target(monkeypatch, tmp_path, post_process
     monkeypatch.setattr(
         post_processing.subtitle_selection, "autoload_rename", lambda search_term, src: src / "renamed.srt"
     )
-    placed: list[tuple[Path, Path]] = []
+    placed: list[tuple[Path, Path, str]] = []
 
-    def fake_move(source_file, destination_directory):
-        placed.append((source_file, destination_directory))
-        (destination_directory / source_file.name).write_text("subtitle")
+    def fake_move(source_file, destination_directory, video_stem, subs_dir):
+        placed.append((source_file, destination_directory, video_stem))
+        (destination_directory / f"{video_stem}{source_file.suffix}").write_text("subtitle")
 
-    monkeypatch.setattr(post_processing.file_system, "move_and_replace", fake_move)
+    monkeypatch.setattr(post_processing.file_system, "move_best_next_to_video", fake_move)
     monkeypatch.setattr(post_processing.file_system, "create_path_from_string", lambda *args, **kwargs: target_dir)
     worker = post_processing._PostProcessWorker(rename=True, store=SettingsStore(), subtitle=_make_subtitle())
     delivered = worker.execute()
 
     assert delivered == 1
-    assert placed == [(Path("subs") / "renamed.srt", target_dir)]
+    assert placed == [(Path("subs") / "renamed.srt", target_dir, post_processing.SEARCH_SUBJECT.search_term)]