"""Daily retry for fetched copies that never got English subtitles (2026-10-18).
An ingest asks for subtitles once, when the copy lands. On 10-18 wyzie
answered 503 for every file all evening, and 43 fetched copies had none --
fixed by a one-off sweep, but nothing would have retried them. So once a day
the service runs the ingest's own subtitle step (_fetch_subtitles_for: wyzie,
sync, OpenSubtitles, sync) for every fetched copy that has never had an
English track.
"Never had" is deliberate: a copy whose English was measured and withdrawn
(a different cut) already had both sources tried, and fetching the same bad
track again every day would only refill the originals store. A copy being
written is left to its own ingest.
Run: .venv/bin/python3 -m pytest test_subtitle_retry.py -q
"""
import json
import os
import pytest
import addon
@pytest.fixture
def lib(tmp_path, monkeypatch):
root = tmp_path / "lib"
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(root))
monkeypatch.setattr(addon, "_mp4_jobs", {})
copies = {}
def hold(name, key, subs=()):
path = root / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"video")
for code in subs:
(root / (os.path.splitext(name)[0] + f".{code}.srt")).write_text("x")
copies[os.path.abspath(path)] = [key]
return str(path)
monkeypatch.setattr(addon, "_fetched_dests", lambda: dict(copies))
monkeypatch.setattr(addon, "_library_meta",
lambda kind, ident, s=0, e=0: {"tmdb_id": ident.split(":")[1],
"title": "T", "year": "2020",
"episode_title": ""})
calls = []
def fetch(dest, meta, season, episode, kind="movie"):
calls.append((os.path.basename(dest), kind, meta["tmdb_id"], season, episode))
monkeypatch.setattr(addon, "_fetch_subtitles_for", fetch)
return hold, calls
def test_a_copy_with_no_subtitles_is_retried_with_its_own_ids(lib):
hold, calls = lib
hold("Movies/Film (2025).mp4", "movie:tmdb:2:0:0:1080")
hold("Shows/Show - S02E06.mp4", "series:tmdb:1:2:6:1080")
addon._subtitle_retry_pass()
assert sorted(calls) == [("Film (2025).mp4", "movie", "2", 0, 0),
("Show - S02E06.mp4", "series", "1", 2, 6)]
def test_a_copy_with_english_is_left_alone(lib):
hold, calls = lib
hold("Movies/Film (2025).mp4", "movie:tmdb:2:0:0:1080", subs=("en",))
addon._subtitle_retry_pass()
assert calls == []
def test_other_languages_alone_still_count_as_no_english(lib):
hold, calls = lib
hold("Movies/Film (2025).mp4", "movie:tmdb:2:0:0:1080", subs=("fr", "de"))
addon._subtitle_retry_pass()
assert len(calls) == 1
def test_english_that_was_withdrawn_is_not_fetched_again(lib):
hold, calls = lib
dest = hold("Movies/Film (2025).mp4", "movie:tmdb:2:0:0:1080")
en = os.path.splitext(dest)[0] + ".en.srt"
with open(addon.SUBTITLE_SYNC_FILE, "w") as f:
json.dump({os.path.abspath(en): {"status": "unfixable", "ref": 2,
"offset": 30.0, "scale": 1.0}}, f)
addon._subtitle_retry_pass()
assert calls == []
def test_a_copy_being_written_is_skipped(lib, monkeypatch):
hold, calls = lib
hold("Movies/Film (2025).mp4", "movie:tmdb:2:0:0:1080")
monkeypatch.setattr(addon, "_mp4_jobs", {"movie:tmdb:2:0:0:1080": "running"})
addon._subtitle_retry_pass()
assert calls == []
def test_a_copy_gone_from_disk_is_skipped(lib):
hold, calls = lib
dest = hold("Movies/Film (2025).mp4", "movie:tmdb:2:0:0:1080")
os.remove(dest)
addon._subtitle_retry_pass()
assert calls == []
def test_the_pass_reports_what_it_did(lib, monkeypatch):
hold, calls = lib
hold("Movies/A (2025).mp4", "movie:tmdb:2:0:0:1080")
hold("Movies/B (2025).mp4", "movie:tmdb:3:0:0:1080", subs=("en",))
hold("Movies/C (2025).mp4", "movie:tmdb:4:0:0:1080")
def fetch(dest, meta, season, episode, kind="movie"):
if "A (2025)" in dest:
open(os.path.splitext(dest)[0] + ".en.srt", "w").close()
monkeypatch.setattr(addon, "_fetch_subtitles_for", fetch)
assert addon._subtitle_retry_pass() == {"checked": 3, "retried": 2, "gained": 1}
def test_one_copy_failing_does_not_stop_the_pass(lib, monkeypatch):
hold, calls = lib
hold("Movies/A (2025).mp4", "movie:tmdb:2:0:0:1080")
hold("Movies/B (2025).mp4", "movie:tmdb:3:0:0:1080")
seen = []
def fetch(dest, meta, season, episode, kind="movie"):
seen.append(os.path.basename(dest))
if "A (2025)" in dest:
raise RuntimeError("boom")
monkeypatch.setattr(addon, "_fetch_subtitles_for", fetch)
addon._subtitle_retry_pass()
assert sorted(seen) == ["A (2025).mp4", "B (2025).mp4"]
def test_an_unreadable_sync_record_touches_nothing(lib):
hold, calls = lib
hold("Movies/Film (2025).mp4", "movie:tmdb:2:0:0:1080")
with open(addon.SUBTITLE_SYNC_FILE, "w") as f:
f.write("{not json")
addon._subtitle_retry_pass()
assert calls == []