Tom FluxandClaude Opus 5
18bb2e420b
A DVR for YouTube subscriptions, laid out so Jellyfin presents each channel as a show and each video as an episode. Cron-driven, idempotent, with a public admin UI for a non-operator. Verified end to end on susan against three real channels: PO tokens, h264 downloads, Jellyfin resolution from local NFOs with all providers disabled, retention and tombstones. Corrections to the original design handover (specs.md documents each with the evidence, and specs.handover-original.md preserves the original): - The format sort selected 360p. Ranking acodec above res makes `bv*` prefer the combined 360p stream, which carries AAC, over the 720p video-only stream whose acodec is none. vcodec now leads, so a video without h264 at 720p yields h264 lower down rather than VP9 this hardware cannot transcode. - yt-dlp now requires a JS runtime and the yt-dlp-ejs solver scripts, which only ship with the [default] extra. Without them the n challenge fails and the mweb formats disappear entirely. - --flat-playlist carries no upload dates, so the specced client-side date filter for backfill was impossible. Backfill is RSS-first. - skipped_old was terminal, so raising a channel's retention appeared to do nothing. Added an explicit rescan. - is_upcoming premieres now defer and retry instead of being skipped forever. - TubeArchivist is gone, so the media root and the tube.jihakuz.xyz vhost were both reclaimed; the latter still pointed at its dead port. 240 offline tests, no network and no real yt-dlp invocation. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
260 lines
9.7 KiB
Python
260 lines
9.7 KiB
Python
"""Discovery: feed parsing, the fallback path, and the two repair mechanisms."""
|
|
|
|
from datetime import date, timedelta
|
|
|
|
import pytest
|
|
|
|
from conftest import add_video, feed_bytes
|
|
from youtube_automate import discovery, util, videos
|
|
|
|
|
|
class TestParseEntries:
|
|
def test_parses_all_entries(self):
|
|
entries = discovery.parse_entries(feed_bytes("uulf_feed.xml"))
|
|
assert len(entries) == 3
|
|
assert entries[0]["video_id"] == "08Ajr5fP52I"
|
|
assert entries[0]["published"] == date(2026, 8, 7)
|
|
|
|
def test_unescapes_description_entities(self):
|
|
entries = discovery.parse_entries(feed_bytes("uulf_feed.xml"))
|
|
assert "&" in entries[0]["description"]
|
|
assert "<of a series>" in entries[0]["description"]
|
|
|
|
def test_keeps_characters_the_filename_would_strip(self):
|
|
entries = discovery.parse_entries(feed_bytes("uulf_feed.xml"))
|
|
assert entries[1]["title"] == "Trying to use a Nortel PBX: part two"
|
|
|
|
def test_empty_feed_yields_nothing(self):
|
|
empty = b'<?xml version="1.0"?><feed xmlns="http://www.w3.org/2005/Atom"/>'
|
|
assert discovery.parse_entries(empty) == []
|
|
|
|
def test_unparseable_feed_raises(self):
|
|
with pytest.raises(discovery.FeedUnavailable):
|
|
discovery.parse_entries(b"<not xml")
|
|
|
|
def test_entry_without_video_id_is_skipped(self):
|
|
payload = (
|
|
b'<?xml version="1.0"?><feed xmlns="http://www.w3.org/2005/Atom">'
|
|
b"<entry><title>no id</title></entry></feed>"
|
|
)
|
|
assert discovery.parse_entries(payload) == []
|
|
|
|
|
|
class TestFeedUrls:
|
|
def test_uulf_strips_the_uc_prefix(self):
|
|
url = discovery.uulf_feed_url("UCW7jUEpYT_t0Gsf632d6_wQ")
|
|
assert "playlist_id=UULFW7jUEpYT_t0Gsf632d6_wQ" in url
|
|
|
|
def test_uc_feed_uses_channel_id(self):
|
|
url = discovery.uc_feed_url("UCW7jUEpYT_t0Gsf632d6_wQ")
|
|
assert "channel_id=UCW7jUEpYT_t0Gsf632d6_wQ" in url
|
|
|
|
|
|
class TestPollChannel:
|
|
def test_queues_recent_and_skips_old(self, conn, settings, channel, monkeypatch):
|
|
recent = util.today() - timedelta(days=2)
|
|
stale = util.today() - timedelta(days=400)
|
|
monkeypatch.setattr(
|
|
discovery,
|
|
"fetch_feed",
|
|
lambda url, timeout=30.0: b"ignored",
|
|
)
|
|
monkeypatch.setattr(
|
|
discovery,
|
|
"parse_entries",
|
|
lambda payload: [
|
|
{"video_id": "new1", "title": "new", "published": recent, "description": ""},
|
|
{"video_id": "old1", "title": "old", "published": stale, "description": ""},
|
|
],
|
|
)
|
|
stats = discovery.poll_channel(conn, settings, channel)
|
|
assert stats["queued"] == 1
|
|
assert stats["old"] == 1
|
|
assert videos.get(conn, "new1")["state"] == videos.PENDING
|
|
assert videos.get(conn, "old1")["state"] == videos.SKIPPED_OLD
|
|
|
|
def test_falls_back_to_channel_feed_when_uulf_is_empty(
|
|
self, conn, settings, channel, monkeypatch
|
|
):
|
|
seen_urls = []
|
|
|
|
def fake_fetch(url, timeout=30.0):
|
|
seen_urls.append(url)
|
|
return None if "playlist_id" in url else b"feed"
|
|
|
|
monkeypatch.setattr(discovery, "fetch_feed", fake_fetch)
|
|
monkeypatch.setattr(
|
|
discovery,
|
|
"parse_entries",
|
|
lambda payload: [
|
|
{
|
|
"video_id": "fb1",
|
|
"title": "fallback",
|
|
"published": util.today(),
|
|
"description": "",
|
|
}
|
|
],
|
|
)
|
|
stats = discovery.poll_channel(conn, settings, channel)
|
|
|
|
assert stats["source"] == videos.SOURCE_UC
|
|
assert any("playlist_id" in url for url in seen_urls)
|
|
assert any("channel_id" in url for url in seen_urls)
|
|
assert videos.get(conn, "fb1")["discovery_source"] == videos.SOURCE_UC
|
|
|
|
def test_feed_failure_increments_the_counter(self, conn, settings, channel, monkeypatch):
|
|
def boom(url, timeout=30.0):
|
|
raise discovery.FeedUnavailable("HTTP 503")
|
|
|
|
monkeypatch.setattr(discovery, "fetch_feed", boom)
|
|
stats = discovery.poll_channel(conn, settings, channel)
|
|
|
|
assert "error" in stats
|
|
row = conn.execute(
|
|
"SELECT * FROM channel WHERE id = ?", (channel["id"],)
|
|
).fetchone()
|
|
assert row["consecutive_poll_failures"] == 1
|
|
assert row["last_poll_ok"] == 0
|
|
|
|
def test_success_resets_the_failure_counter(self, conn, settings, channel, monkeypatch):
|
|
with conn:
|
|
conn.execute(
|
|
"UPDATE channel SET consecutive_poll_failures = 4 WHERE id = ?",
|
|
(channel["id"],),
|
|
)
|
|
monkeypatch.setattr(discovery, "fetch_feed", lambda url, timeout=30.0: b"x")
|
|
monkeypatch.setattr(discovery, "parse_entries", lambda payload: [])
|
|
|
|
discovery.poll_channel(conn, settings, channel)
|
|
row = conn.execute(
|
|
"SELECT * FROM channel WHERE id = ?", (channel["id"],)
|
|
).fetchone()
|
|
assert row["consecutive_poll_failures"] == 0
|
|
assert row["last_poll_ok"] == 1
|
|
|
|
|
|
class TestSkippedShortRepair:
|
|
"""A fallback-discovered video wrongly rejected as a Short must come back
|
|
once the authoritative UULF feed lists it."""
|
|
|
|
def _poll_with(self, monkeypatch, entry):
|
|
monkeypatch.setattr(discovery, "fetch_feed", lambda url, timeout=30.0: b"x")
|
|
monkeypatch.setattr(discovery, "parse_entries", lambda payload: [entry])
|
|
|
|
def test_repairs_a_uc_discovered_skipped_short(
|
|
self, conn, settings, channel, monkeypatch
|
|
):
|
|
add_video(
|
|
conn,
|
|
channel["id"],
|
|
"short1",
|
|
state=videos.SKIPPED_SHORT,
|
|
discovery_source=videos.SOURCE_UC,
|
|
upload_date=util.today().isoformat(),
|
|
)
|
|
self._poll_with(
|
|
monkeypatch,
|
|
{
|
|
"video_id": "short1",
|
|
"title": "t",
|
|
"published": util.today(),
|
|
"description": "",
|
|
},
|
|
)
|
|
stats = discovery.poll_channel(conn, settings, channel)
|
|
|
|
assert stats["repaired"] == 1
|
|
row = videos.get(conn, "short1")
|
|
assert row["state"] == videos.PENDING
|
|
assert row["discovery_source"] == videos.SOURCE_UULF
|
|
|
|
def test_does_not_repair_one_discovered_via_uulf(
|
|
self, conn, settings, channel, monkeypatch
|
|
):
|
|
add_video(
|
|
conn,
|
|
channel["id"],
|
|
"short2",
|
|
state=videos.SKIPPED_SHORT,
|
|
discovery_source=videos.SOURCE_UULF,
|
|
)
|
|
self._poll_with(
|
|
monkeypatch,
|
|
{"video_id": "short2", "title": "t", "published": util.today(), "description": ""},
|
|
)
|
|
discovery.poll_channel(conn, settings, channel)
|
|
assert videos.get(conn, "short2")["state"] == videos.SKIPPED_SHORT
|
|
|
|
def test_never_resurrects_a_deleted_tombstone(
|
|
self, conn, settings, channel, monkeypatch
|
|
):
|
|
add_video(
|
|
conn,
|
|
channel["id"],
|
|
"gone1",
|
|
state=videos.DELETED,
|
|
discovery_source=videos.SOURCE_UC,
|
|
)
|
|
self._poll_with(
|
|
monkeypatch,
|
|
{"video_id": "gone1", "title": "t", "published": util.today(), "description": ""},
|
|
)
|
|
discovery.poll_channel(conn, settings, channel)
|
|
assert videos.get(conn, "gone1")["state"] == videos.DELETED
|
|
|
|
|
|
class TestRescan:
|
|
def test_requeues_skipped_old_inside_the_window(self, conn, settings, channel):
|
|
inside = (util.today() - timedelta(days=5)).isoformat()
|
|
add_video(conn, channel["id"], "v1", state=videos.SKIPPED_OLD, upload_date=inside)
|
|
|
|
assert discovery.rescan_channel(conn, settings, channel) == 1
|
|
assert videos.get(conn, "v1")["state"] == videos.PENDING
|
|
|
|
def test_leaves_videos_outside_the_window_alone(self, conn, settings, channel):
|
|
outside = (util.today() - timedelta(days=200)).isoformat()
|
|
add_video(conn, channel["id"], "v2", state=videos.SKIPPED_OLD, upload_date=outside)
|
|
|
|
assert discovery.rescan_channel(conn, settings, channel) == 0
|
|
assert videos.get(conn, "v2")["state"] == videos.SKIPPED_OLD
|
|
|
|
def test_honours_a_per_channel_override(self, conn, settings, channel):
|
|
age = (util.today() - timedelta(days=30)).isoformat()
|
|
add_video(conn, channel["id"], "v3", state=videos.SKIPPED_OLD, upload_date=age)
|
|
|
|
# Default retention is 14 days, so nothing moves.
|
|
assert discovery.rescan_channel(conn, settings, channel) == 0
|
|
|
|
with conn:
|
|
conn.execute(
|
|
"UPDATE channel SET retention_days = 60 WHERE id = ?", (channel["id"],)
|
|
)
|
|
widened = conn.execute(
|
|
"SELECT * FROM channel WHERE id = ?", (channel["id"],)
|
|
).fetchone()
|
|
assert discovery.rescan_channel(conn, settings, widened) == 1
|
|
|
|
def test_never_resurrects_a_tombstone(self, conn, settings, channel):
|
|
add_video(
|
|
conn,
|
|
channel["id"],
|
|
"v4",
|
|
state=videos.DELETED,
|
|
upload_date=util.today().isoformat(),
|
|
)
|
|
assert discovery.rescan_channel(conn, settings, channel) == 0
|
|
assert videos.get(conn, "v4")["state"] == videos.DELETED
|
|
|
|
|
|
class TestEffectiveRetention:
|
|
def test_override_wins(self, settings, channel, conn):
|
|
with conn:
|
|
conn.execute(
|
|
"UPDATE channel SET retention_days = 60 WHERE id = ?", (channel["id"],)
|
|
)
|
|
row = conn.execute("SELECT * FROM channel WHERE id = ?", (channel["id"],)).fetchone()
|
|
assert discovery.effective_retention_days(settings, row) == 60
|
|
|
|
def test_falls_back_to_the_global_default(self, settings, channel):
|
|
assert discovery.effective_retention_days(settings, channel) == 14
|