Background date enricher for followed YouTube channels (no key)
Follow a channel → a background job fetches its FULL upload-date catalog so the channel page's year-seasons populate fully, cached permanently (one-time per channel, instant after). - core/video/youtube.py: proxy_channel_dates() — no-key BULK dates via Piped/ Invidious instances (tries several, paginates); parse_proxy_dates handles both shapes. The clean path the user wanted. - core/video/youtube_enrichment.py: YoutubeDateEnricher daemon — enqueue(channel) → proxy bulk → cache; per-video yt-dlp only as a throttled fallback (cap 60, 0.4s) for wished videos when every proxy is down. Skips channels enriched in the last 24h. - db: youtube_channel_enrichment table + mark/check + wishlisted_video_ids_for_ channel. Enqueued from /youtube/follow and on opening a followed channel. Scoped to FOLLOWED channels only (per request). 174 tests green; enricher imports nothing from music.
This commit is contained in:
BoulderBadgeDad
parent
ec5af17de7
commit
30dc587ebf
8 changed files with 384 additions and 0 deletions
|
|
@ -81,6 +81,12 @@ def register_routes(bp):
|
|||
|
||||
followed = db.add_channel_to_watchlist(channel)
|
||||
added = db.add_videos_to_wishlist(channel, channel.get("videos") or [], server_source=_server())
|
||||
if followed: # followed channels get their full upload-date catalog in the background
|
||||
try:
|
||||
from core.video.youtube_enrichment import get_youtube_date_enricher
|
||||
get_youtube_date_enricher().enqueue(channel.get("youtube_id"))
|
||||
except Exception:
|
||||
pass
|
||||
return jsonify({"success": followed, "following": followed, "added_videos": added,
|
||||
"channel": {k: channel.get(k) for k in ("youtube_id", "title", "avatar_url")},
|
||||
"counts": db.youtube_wishlist_counts()})
|
||||
|
|
@ -147,6 +153,12 @@ def register_routes(bp):
|
|||
return jsonify({"success": False, "error": "Channel not found"}), 404
|
||||
cid = channel["youtube_id"]
|
||||
following = bool(db.channel_watch_state([cid]))
|
||||
if following: # opening a followed channel → ensure its dates get enriched
|
||||
try:
|
||||
from core.video.youtube_enrichment import get_youtube_date_enricher
|
||||
get_youtube_date_enricher().enqueue(cid)
|
||||
except Exception:
|
||||
pass
|
||||
# backfill the real avatar onto wished rows (flat listing often omits it)
|
||||
if channel.get("avatar_url"):
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -250,6 +250,108 @@ def resolve_channel(raw, limit=30, ydl_factory=None, db=None):
|
|||
return shaped if shaped.get("youtube_id") else None
|
||||
|
||||
|
||||
# No-key bulk date sources: community Piped/Invidious instances. Flaky (they go
|
||||
# up/down), so we try several and fall back to per-video yt-dlp if none answer.
|
||||
_PROXY_INSTANCES = [
|
||||
("piped", "https://pipedapi.kavin.rocks"),
|
||||
("piped", "https://pipedapi.adminforge.de"),
|
||||
("piped", "https://api.piped.private.coffee"),
|
||||
("invidious", "https://invidious.nerdvpn.de"),
|
||||
("invidious", "https://inv.nadeko.net"),
|
||||
]
|
||||
|
||||
|
||||
def _epoch_to_date(v):
|
||||
"""epoch seconds OR milliseconds → 'YYYY-MM-DD', or None."""
|
||||
try:
|
||||
n = float(v)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
if n <= 0:
|
||||
return None
|
||||
if n > 1e12: # milliseconds
|
||||
n /= 1000.0
|
||||
try:
|
||||
return datetime.fromtimestamp(n, tz=timezone.utc).date().isoformat()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _vid_from_url(u):
|
||||
m = re.search(r"[?&]v=([\w-]+)", u or "")
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def parse_proxy_dates(obj):
|
||||
"""{video_id: 'YYYY-MM-DD'} from a Piped (relatedStreams) or Invidious (videos)
|
||||
channel JSON response — handles both shapes."""
|
||||
out = {}
|
||||
if not isinstance(obj, dict):
|
||||
return out
|
||||
for s in (obj.get("relatedStreams") or []): # Piped
|
||||
if not isinstance(s, dict):
|
||||
continue
|
||||
vid = _vid_from_url(s.get("url")) or s.get("videoId")
|
||||
d = _epoch_to_date(s.get("uploaded"))
|
||||
if vid and d:
|
||||
out[vid] = d
|
||||
for v in (obj.get("videos") or []): # Invidious
|
||||
if not isinstance(v, dict):
|
||||
continue
|
||||
vid = v.get("videoId")
|
||||
d = _epoch_to_date(v.get("published"))
|
||||
if vid and d:
|
||||
out[vid] = d
|
||||
return out
|
||||
|
||||
|
||||
def _proxy_get(url, fetch):
|
||||
if fetch is not None:
|
||||
return fetch(url)
|
||||
import requests
|
||||
r = requests.get(url, timeout=8, headers={"User-Agent": _UA, "Accept": "application/json"})
|
||||
return r.json() if r.status_code == 200 else None
|
||||
|
||||
|
||||
def _harvest(kind, base, cid, pages, fetch):
|
||||
"""Walk one instance's channel pages, accumulating {video_id: date}."""
|
||||
from urllib.parse import quote
|
||||
out, token = {}, None
|
||||
for i in range(max(1, pages)):
|
||||
if kind == "piped":
|
||||
url = (base + "/channel/" + cid) if token is None \
|
||||
else (base + "/nextpage/channel/" + cid + "?nextpage=" + quote(token, safe=""))
|
||||
else: # invidious
|
||||
url = base + "/api/v1/channels/" + cid + "/videos" + \
|
||||
(("?continuation=" + quote(token, safe="")) if token else "")
|
||||
obj = _proxy_get(url, fetch)
|
||||
page = parse_proxy_dates(obj if isinstance(obj, dict) else {})
|
||||
if not page and i == 0:
|
||||
return {} # instance gave nothing → caller tries next
|
||||
out.update(page)
|
||||
token = (obj.get("nextpage") if kind == "piped" else obj.get("continuation")) if isinstance(obj, dict) else None
|
||||
if not token:
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def proxy_channel_dates(channel_id, pages=6, fetch=None, instances=None):
|
||||
"""Bulk upload dates for a whole channel via a no-key proxy (Piped/Invidious),
|
||||
paginated. Tries instances until one answers. {video_id: 'YYYY-MM-DD'} (empty
|
||||
if all are down → caller falls back to per-video yt-dlp)."""
|
||||
cid = str(channel_id or "").strip()
|
||||
if not cid:
|
||||
return {}
|
||||
for kind, base in (instances or _PROXY_INSTANCES):
|
||||
try:
|
||||
dates = _harvest(kind, base, cid, pages, fetch)
|
||||
if dates:
|
||||
return dates
|
||||
except Exception:
|
||||
continue
|
||||
return {}
|
||||
|
||||
|
||||
def parse_rss_dates(xml_text):
|
||||
"""{video_id: 'YYYY-MM-DD'} from a YouTube channel RSS feed (pure, testable)."""
|
||||
import xml.etree.ElementTree as ET
|
||||
|
|
|
|||
119
core/video/youtube_enrichment.py
Normal file
119
core/video/youtube_enrichment.py
Normal file
|
|
@ -0,0 +1,119 @@
|
|||
"""Background YouTube date-enricher (video side, isolated).
|
||||
|
||||
Followed channels get their full upload-date catalog fetched in the background so
|
||||
the channel page's year-seasons populate fully (the fast flat listing has no
|
||||
dates). Cheap no-key bulk source first (Piped/Invidious proxy via
|
||||
``proxy_channel_dates``); per-video yt-dlp only as a throttled fallback for the
|
||||
channel's wished videos when every proxy is down. Everything is cached in
|
||||
``youtube_video_dates`` so it's a one-time cost per channel and instant after.
|
||||
|
||||
A single daemon thread drains an enqueue() queue. Enqueue a channel when it's
|
||||
followed (or its page opened while followed). Reads/writes only video_library.db.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
logger = get_logger("video.youtube_enrichment")
|
||||
|
||||
# Throttle the per-video fallback so we don't burst YouTube into rate-limiting.
|
||||
_FALLBACK_CAP = 60
|
||||
_FALLBACK_DELAY = 0.4
|
||||
|
||||
|
||||
class YoutubeDateEnricher:
|
||||
def __init__(self, db_factory=None):
|
||||
self._db_factory = db_factory or self._default_db
|
||||
self._q: "queue.Queue[str]" = queue.Queue()
|
||||
self._inflight = set()
|
||||
self._thread = None
|
||||
self._lock = threading.Lock()
|
||||
|
||||
@staticmethod
|
||||
def _default_db():
|
||||
from database.video_database import VideoDatabase
|
||||
return VideoDatabase()
|
||||
|
||||
def enqueue(self, channel_id):
|
||||
"""Queue a followed channel for full date enrichment (deduped; starts the
|
||||
worker thread on first use)."""
|
||||
cid = str(channel_id or "").strip()
|
||||
if not cid:
|
||||
return
|
||||
with self._lock:
|
||||
if cid in self._inflight:
|
||||
return
|
||||
self._inflight.add(cid)
|
||||
self._q.put(cid)
|
||||
if self._thread is None or not self._thread.is_alive():
|
||||
self._thread = threading.Thread(target=self._run, name="yt-date-enricher", daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def _run(self):
|
||||
while True:
|
||||
try:
|
||||
cid = self._q.get(timeout=45)
|
||||
except queue.Empty:
|
||||
return # idle → let the thread die; re-spawned on next enqueue
|
||||
try:
|
||||
self._enrich(cid)
|
||||
except Exception:
|
||||
logger.exception("YouTube date enrichment failed for %s", cid)
|
||||
finally:
|
||||
with self._lock:
|
||||
self._inflight.discard(cid)
|
||||
self._q.task_done()
|
||||
|
||||
def _enrich(self, channel_id):
|
||||
"""Fetch + cache a channel's upload dates. Proxy in bulk; per-video fallback."""
|
||||
from core.video import youtube as yt
|
||||
db = self._db_factory()
|
||||
cid = str(channel_id or "").strip()
|
||||
if not cid or db.channel_dates_enriched_recently(cid):
|
||||
return
|
||||
|
||||
dates = {}
|
||||
try:
|
||||
dates = yt.proxy_channel_dates(cid) or {}
|
||||
except Exception:
|
||||
logger.info("proxy date fetch failed for %s", cid, exc_info=True)
|
||||
if dates:
|
||||
db.cache_video_dates([{"youtube_id": k, "published_at": v} for k, v in dates.items()])
|
||||
logger.info("YouTube dates: %d cached for %s (proxy)", len(dates), cid)
|
||||
|
||||
# Fallback: per-video for the channel's wished videos still lacking a date.
|
||||
ids = db.wishlisted_video_ids_for_channel(cid)
|
||||
have = db.get_video_dates(ids)
|
||||
missing = [i for i in ids if i not in have and i not in dates]
|
||||
filled = 0
|
||||
for vid in missing[:_FALLBACK_CAP]:
|
||||
try:
|
||||
v = yt.video_detail(vid)
|
||||
except Exception:
|
||||
v = None
|
||||
if v and v.get("published_at"):
|
||||
db.cache_video_dates([{"youtube_id": vid, "published_at": v["published_at"]}])
|
||||
filled += 1
|
||||
time.sleep(_FALLBACK_DELAY)
|
||||
if filled:
|
||||
logger.info("YouTube dates: %d cached for %s (per-video fallback)", filled, cid)
|
||||
|
||||
db.mark_channel_dates_enriched(cid, len(dates) + filled)
|
||||
|
||||
|
||||
_enricher = None
|
||||
_enricher_lock = threading.Lock()
|
||||
|
||||
|
||||
def get_youtube_date_enricher():
|
||||
global _enricher
|
||||
if _enricher is None:
|
||||
with _enricher_lock:
|
||||
if _enricher is None:
|
||||
_enricher = YoutubeDateEnricher()
|
||||
return _enricher
|
||||
|
|
@ -2027,6 +2027,52 @@ class VideoDatabase:
|
|||
finally:
|
||||
conn.close()
|
||||
|
||||
def wishlisted_video_ids_for_channel(self, channel_id) -> list:
|
||||
"""The youtube video ids wished under a channel (the per-video date fallback set)."""
|
||||
if not channel_id:
|
||||
return []
|
||||
conn = self._get_connection()
|
||||
try:
|
||||
return [r["source_id"] for r in conn.execute(
|
||||
"SELECT source_id FROM video_wishlist WHERE kind='video' AND parent_source_id=?",
|
||||
(str(channel_id),))]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def mark_channel_dates_enriched(self, channel_id, date_count=0) -> None:
|
||||
"""Record that the background enricher swept this channel (skip re-sweeps)."""
|
||||
if not channel_id:
|
||||
return
|
||||
conn = self._get_connection()
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO youtube_channel_enrichment (channel_id, enriched_at, date_count) "
|
||||
"VALUES (?, CURRENT_TIMESTAMP, ?) ON CONFLICT(channel_id) DO UPDATE SET "
|
||||
"enriched_at=CURRENT_TIMESTAMP, date_count=excluded.date_count",
|
||||
(str(channel_id), int(date_count or 0)))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def channel_dates_enriched_recently(self, channel_id, within_hours=24) -> bool:
|
||||
"""True if the channel was date-enriched within the window (don't re-sweep)."""
|
||||
if not channel_id:
|
||||
return False
|
||||
conn = self._get_connection()
|
||||
try:
|
||||
row = conn.execute("SELECT enriched_at FROM youtube_channel_enrichment WHERE channel_id=?",
|
||||
(str(channel_id),)).fetchone()
|
||||
if not row or not row["enriched_at"]:
|
||||
return False
|
||||
try:
|
||||
when = datetime.strptime(row["enriched_at"], "%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
now = datetime.now(timezone.utc).replace(tzinfo=None) # naive UTC, matches CURRENT_TIMESTAMP
|
||||
return (now - when) < timedelta(hours=within_hours)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def youtube_wishlist_counts(self) -> dict:
|
||||
"""{'channel': n distinct channels, 'video': n videos} in the wishlist."""
|
||||
conn = self._get_connection()
|
||||
|
|
|
|||
|
|
@ -271,6 +271,14 @@ CREATE TABLE IF NOT EXISTS youtube_video_dates (
|
|||
cached_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
-- Tracks which followed channels have had their full upload dates fetched (by the
|
||||
-- background enricher) so we don't re-sweep them constantly.
|
||||
CREATE TABLE IF NOT EXISTS youtube_channel_enrichment (
|
||||
channel_id TEXT PRIMARY KEY,
|
||||
enriched_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
date_count INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
-- ── Owned media files (the Library = content that has a file) ────────────────
|
||||
-- Exactly one owner FK is set (no polymorphic id). 1 row per physical file;
|
||||
-- usually 1:1 with its content, but the table allows history/extras.
|
||||
|
|
|
|||
|
|
@ -1167,3 +1167,13 @@ def test_video_date_cache_roundtrip(db):
|
|||
# upsert refreshes
|
||||
db.cache_video_dates([{"youtube_id": "a", "published_at": "2025-12-31"}])
|
||||
assert db.get_video_dates(["a"]) == {"a": "2025-12-31"}
|
||||
|
||||
|
||||
def test_channel_enrichment_tracking(db):
|
||||
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"},
|
||||
[{"youtube_id": "a", "title": "A"}, {"youtube_id": "b", "title": "B"}])
|
||||
assert set(db.wishlisted_video_ids_for_channel("UCx")) == {"a", "b"}
|
||||
assert db.channel_dates_enriched_recently("UCx") is False
|
||||
db.mark_channel_dates_enriched("UCx", date_count=2)
|
||||
assert db.channel_dates_enriched_recently("UCx") is True
|
||||
assert db.channel_dates_enriched_recently("UCx", within_hours=0) is False # window respected
|
||||
|
|
|
|||
|
|
@ -338,3 +338,39 @@ def test_channel_recent_dates_via_injected_fetch():
|
|||
out = yt.channel_recent_dates("UCx", fetch=lambda url: _RSS)
|
||||
assert out["v1"] == "2024-06-01"
|
||||
assert yt.channel_recent_dates("", fetch=lambda url: _RSS) == {}
|
||||
|
||||
|
||||
# ── proxy bulk dates (Piped / Invidious, no key) ─────────────────────────────
|
||||
|
||||
def test_parse_proxy_dates_piped_and_invidious():
|
||||
piped = {"relatedStreams": [
|
||||
{"url": "/watch?v=v1", "uploaded": 1700000000000}, # ms
|
||||
{"url": "https://youtube.com/watch?v=v2", "uploaded": 1690000000000},
|
||||
{"url": "/watch?v=v3"}, # no date → skipped
|
||||
]}
|
||||
assert yt.parse_proxy_dates(piped) == {"v1": "2023-11-14", "v2": "2023-07-22"}
|
||||
inv = {"videos": [
|
||||
{"videoId": "a", "published": 1700000000}, # seconds
|
||||
{"videoId": "b", "published": 0}, # bad → skipped
|
||||
]}
|
||||
assert yt.parse_proxy_dates(inv) == {"a": "2023-11-14"}
|
||||
assert yt.parse_proxy_dates("nope") == {}
|
||||
|
||||
|
||||
def test_proxy_channel_dates_paginates_and_falls_through_instances():
|
||||
# first instance returns nothing → try next; piped paginates via nextpage
|
||||
pages = {
|
||||
"https://up/channel/UCx": {"relatedStreams": [{"url": "/watch?v=p1", "uploaded": 1700000000000}],
|
||||
"nextpage": "TOK"},
|
||||
"https://up/nextpage/channel/UCx?nextpage=TOK": {"relatedStreams": [{"url": "/watch?v=p2", "uploaded": 1690000000000}]},
|
||||
}
|
||||
|
||||
def fetch(url):
|
||||
if url.startswith("https://down"):
|
||||
raise RuntimeError("instance down")
|
||||
return pages.get(url)
|
||||
|
||||
insts = [("piped", "https://down"), ("piped", "https://up")]
|
||||
out = yt.proxy_channel_dates("UCx", pages=5, fetch=fetch, instances=insts)
|
||||
assert out == {"p1": "2023-11-14", "p2": "2023-07-22"} # both pages, second instance
|
||||
assert yt.proxy_channel_dates("", fetch=fetch, instances=insts) == {}
|
||||
|
|
|
|||
51
tests/test_video_youtube_enrichment.py
Normal file
51
tests/test_video_youtube_enrichment.py
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
"""Seam tests for the background YouTube date enricher."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from core.video.youtube_enrichment import YoutubeDateEnricher
|
||||
from database.video_database import VideoDatabase
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def db(tmp_path):
|
||||
return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
|
||||
|
||||
|
||||
def test_enrich_caches_proxy_dates_and_marks_done(db, monkeypatch):
|
||||
import core.video.youtube as yt
|
||||
monkeypatch.setattr(yt, "proxy_channel_dates",
|
||||
lambda cid, *a, **k: {"v1": "2024-06-01", "v2": "2023-02-02"})
|
||||
# if proxy covers everything, the per-video fallback must NOT run
|
||||
monkeypatch.setattr(yt, "video_detail", lambda vid: (_ for _ in ()).throw(AssertionError("no fallback")))
|
||||
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"}, [{"youtube_id": "v1", "title": "A"}])
|
||||
|
||||
e = YoutubeDateEnricher(db_factory=lambda: db)
|
||||
e._enrich("UCx")
|
||||
assert db.get_video_dates(["v1", "v2"]) == {"v1": "2024-06-01", "v2": "2023-02-02"}
|
||||
assert db.channel_dates_enriched_recently("UCx") is True
|
||||
# already enriched → second pass is a no-op (no proxy call)
|
||||
monkeypatch.setattr(yt, "proxy_channel_dates", lambda *a, **k: (_ for _ in ()).throw(AssertionError("re-swept")))
|
||||
e._enrich("UCx")
|
||||
|
||||
|
||||
def test_enrich_falls_back_to_per_video_when_proxy_empty(db, monkeypatch):
|
||||
import core.video.youtube as yt
|
||||
monkeypatch.setattr(yt, "proxy_channel_dates", lambda cid, *a, **k: {}) # all proxies down
|
||||
monkeypatch.setattr(yt, "video_detail",
|
||||
lambda vid: {"youtube_id": vid, "published_at": "2022-03-03"} if vid == "w1" else None)
|
||||
import core.video.youtube_enrichment as mod
|
||||
monkeypatch.setattr(mod.time, "sleep", lambda s: None) # no real throttle delay in tests
|
||||
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"},
|
||||
[{"youtube_id": "w1", "title": "A"}, {"youtube_id": "w2", "title": "B"}])
|
||||
|
||||
e = YoutubeDateEnricher(db_factory=lambda: db)
|
||||
e._enrich("UCx")
|
||||
assert db.get_video_dates(["w1", "w2"]) == {"w1": "2022-03-03"} # only w1 had a date
|
||||
assert db.channel_dates_enriched_recently("UCx") is True
|
||||
|
||||
|
||||
def test_enricher_imports_nothing_from_music():
|
||||
src = Path("core/video/youtube_enrichment.py").read_text(encoding="utf-8")
|
||||
assert "database.music_database" not in src and "from database import" not in src
|
||||
Loading…
Reference in a new issue