YouTube dates: custom InnerTube parser as the primary bulk source (validated live)

A Beatport-style custom parser, but pointed at YouTube's own InnerTube browse API
— no key, no Java, no third-party proxies. Reads the channel 'Videos' tab's
lockupViewModel items (contentId + relative 'N ago' text → approximate date,
which is fine for year-seasons), paginating via continuation tokens.

Validated LIVE before wiring in: GMM/MrBeast/Veritasium each return 240 dated
videos in ~7s with correct year spreads (e.g. Veritasium 2017→2026), ~8 requests
per channel (light on rate limits). End-to-end enricher run cached 240 dates.

Enricher order is now: InnerTube (primary) → configured proxy (opt-in, only if
empty) → yt-dlp per-video (the fallback / basic method, for undated gaps + exact
dates). Pure parsing is fully unit-tested (relative-date conversion, lockup
filtering, continuation-token selection, pagination, guards). 182 video tests
green; additive — if InnerTube ever breaks, it returns {} and falls back, so
nothing breaks.

core/video/youtube.py

@@ -16,7 +16,8 @@ from __future__ import annotations
 
 import logging
 import re
-from datetime import datetime, timezone
+import time
+from datetime import datetime, timedelta, timezone
 from urllib.parse import urlparse
 
 try:
@@ -390,6 +391,127 @@ def channel_recent_dates(channel_id, fetch=None):
         return {}
 
 
+# ── InnerTube: YouTube's own browse API (no key/Java/proxy) ──────────────────
+# Same technique NewPipe/yt-dlp use. We call the channel "Videos" tab and read the
+# lockupViewModel items: contentId (video id) + a relative "N units ago" string.
+# Relative → approximate date (fine for YEAR grouping). One request per ~30 videos,
+# paginated via continuation tokens (light on rate limits). Parsing is pure +
+# resilient (recursive search rather than a brittle fixed path).
+_INNERTUBE_KEY = "AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8"   # public WEB key (stable for years)
+_INNERTUBE_CTX = {"client": {"clientName": "WEB", "clientVersion": "2.20240304.00.00", "hl": "en", "gl": "US"}}
+_VIDEOS_PARAMS = "EgZ2aWRlb3PyBgQKAjoA"   # selects a channel's "Videos" tab
+_INNERTUBE_PAGES = 8
+_INNERTUBE_DELAY = 0.6                     # politeness pause between pages
+_REL_UNIT_DAYS = {"second": 0, "minute": 0, "hour": 0, "day": 1, "week": 7, "month": 30.44, "year": 365.25}
+_REL_RE = re.compile(r"(\d+)\s+(second|minute|hour|day|week|month|year)s?\s+ago", re.I)
+
+
+def _json_find_all(obj, key, acc):
+    if isinstance(obj, dict):
+        if key in obj:
+            acc.append(obj[key])
+        for v in obj.values():
+            _json_find_all(v, key, acc)
+    elif isinstance(obj, list):
+        for v in obj:
+            _json_find_all(v, key, acc)
+    return acc
+
+
+def _json_content_strings(obj, acc):
+    if isinstance(obj, dict):
+        c = obj.get("content")
+        if isinstance(c, str):
+            acc.append(c)
+        for v in obj.values():
+            _json_content_strings(v, acc)
+    elif isinstance(obj, list):
+        for v in obj:
+            _json_content_strings(v, acc)
+    return acc
+
+
+def relative_to_date(text, now=None):
+    """'2 years ago' / '9 hours ago' → approximate 'YYYY-MM-DD', or None."""
+    m = _REL_RE.search(text or "")
+    if not m:
+        return None
+    if now is None:
+        now = datetime.now(timezone.utc).date()
+    days = int(m.group(1)) * _REL_UNIT_DAYS[m.group(2).lower()]
+    return (now - timedelta(days=round(days))).isoformat()
+
+
+def innertube_parse_videos(obj):
+    """[(video_id, relative_text)] for VIDEO lockups in an InnerTube browse response."""
+    out, seen = [], set()
+    for lk in _json_find_all(obj, "lockupViewModel", []):
+        if not isinstance(lk, dict) or lk.get("contentType") != "LOCKUP_CONTENT_TYPE_VIDEO":
+            continue
+        vid = lk.get("contentId")
+        if not vid or vid in seen:
+            continue
+        rel = next((t for t in _json_content_strings(lk.get("metadata"), []) if _REL_RE.search(t)), None)
+        if rel:
+            seen.add(vid)
+            out.append((vid, rel))
+    return out
+
+
+def innertube_continuation(obj):
+    """The pagination continuation token (from a continuationItemRenderer), or None."""
+    for cir in _json_find_all(obj, "continuationItemRenderer", []):
+        if isinstance(cir, dict):
+            tok = (cir.get("continuationEndpoint") or {}).get("continuationCommand", {}).get("token")
+            if tok:
+                return tok
+    return None
+
+
+def _innertube_post(payload, post):
+    if post is not None:
+        return post(payload)
+    import requests
+    r = requests.post("https://www.youtube.com/youtubei/v1/browse", params={"key": _INNERTUBE_KEY},
+                      json=payload, timeout=10,
+                      headers={"User-Agent": _UA, "Content-Type": "application/json", "Accept-Language": "en-US"})
+    return r.json() if (r.status_code == 200 and r.headers.get("content-type", "").startswith("application/json")) else None
+
+
+def innertube_channel_dates(channel_id, pages=_INNERTUBE_PAGES, now=None, post=None):
+    """{video_id: 'YYYY-MM-DD'} for a channel's videos via YouTube's own InnerTube
+    browse API — no key/Java/proxy, ~1 request per 30 videos. Dates are APPROXIMATE
+    (from relative text), which is fine for year-seasons; the exact yt-dlp path can
+    refine specific videos later. Bounded + throttled. {} on any failure (→ caller
+    falls back)."""
+    cid = str(channel_id or "").strip()
+    if not cid.startswith("UC"):
+        return {}
+    if now is None:
+        now = datetime.now(timezone.utc).date()
+    out = {}
+    payload = {"context": _INNERTUBE_CTX, "browseId": cid, "params": _VIDEOS_PARAMS}
+    for _ in range(max(1, pages)):
+        try:
+            j = _innertube_post(payload, post)
+        except Exception:
+            break
+        if not isinstance(j, dict):
+            break
+        for vid, rel in innertube_parse_videos(j):
+            if vid not in out:
+                d = relative_to_date(rel, now)
+                if d:
+                    out[vid] = d
+        token = innertube_continuation(j)
+        if not token:
+            break
+        payload = {"context": _INNERTUBE_CTX, "continuation": token}
+        if post is None:
+            time.sleep(_INNERTUBE_DELAY)   # rate-limit politeness
+    return out
+
+
 def search_channels(query, limit=6, ydl_factory=None):
     """Search YouTube for CHANNELS (the results page filtered to channels) → a few
     {youtube_id, title, handle, avatar_url, subscriber_count} for the search page.

core/video/youtube_enrichment.py

@@ -143,25 +143,29 @@ class YoutubeDateEnricher:
         self._current = self._titles.get(cid) or cid
         logger.debug("Enriching dates for %s (%s)", self._current, cid)
 
-        # Bulk no-key proxy (Piped/Invidious) is OPT-IN via a setting — the public
+        # PRIMARY: YouTube's own InnerTube browse API (no key/Java/proxy) — bulk
-        # instances are unreliable/API-disabled, so by default we skip straight to
+        # dates for the whole videos tab in a handful of requests. The path we
-        # the yt-dlp path. Power users can point youtube_proxy_instances at a live
+        # prefer; approximate dates, great for year-seasons.
-        # instance ("piped|https://…, invidious|https://…").
         dates = {}
-        instances = self._proxy_instances(db)
+        try:
-        if instances:
+            dates = yt.innertube_channel_dates(cid) or {}
-            try:
+        except Exception:
-                dates = yt.proxy_channel_dates(cid, instances=instances) or {}
+            logger.debug("innertube date fetch failed for %s", cid, exc_info=True)
-            except Exception:
+        logger.debug("innertube returned %d dates for %s", len(dates), cid)
-                logger.debug("proxy date fetch failed for %s", cid, exc_info=True)
+        # SECONDARY (opt-in): a configured Piped/Invidious proxy — only if InnerTube
-            logger.debug("proxy returned %d dates for %s", len(dates), cid)
+        # came up empty (the public instances are unreliable/API-disabled).
+        if not dates:
+            instances = self._proxy_instances(db)
+            if instances:
+                try:
+                    dates = yt.proxy_channel_dates(cid, instances=instances) or {}
+                except Exception:
+                    logger.debug("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()])
-            for k, v in dates.items():   # per-item INFO, like the other workers
-                logger.info("Dated %s %s -> %s", self._current, k, v)
 
-        # Fallback (proxy down): date the channel's RECENT UPLOADS via yt-dlp, not
+        # FALLBACK (the basic method): exact dates per-video via yt-dlp, only for
-        # just wished videos — so years populate fully even with no working proxy.
+        # the channel's videos still UNDATED — cheap when the bulk pass worked.
         ids = set(db.wishlisted_video_ids_for_channel(cid))
         if not dates:
             try:
@@ -203,7 +207,8 @@ class YoutubeDateEnricher:
         self._dates_total += len(dates) + filled
         db.mark_channel_dates_enriched(cid, len(dates) + filled)
         # Terse per-channel summary (like the worker's "Synced full episode list…").
-        logger.info("Dated %d/%d videos for %s", len(dates) + filled, len(ids), self._current)
+        logger.info("Dated %d videos for %s (%d bulk + %d per-video)",
+                    len(dates) + filled, self._current, len(dates), filled)
 
 
 _enricher = None

tests/test_video_youtube.py

@@ -374,3 +374,85 @@ def test_proxy_channel_dates_paginates_and_falls_through_instances():
     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) == {}
+
+
+# ── InnerTube channel-date parser (the no-key bulk source) ───────────────────
+
+from datetime import date
+
+
+def _lk_item(vid, rel, ctype="LOCKUP_CONTENT_TYPE_VIDEO"):
+    """A richItemRenderer wrapping a lockupViewModel, mirroring real InnerTube JSON."""
+    return {"richItemRenderer": {"content": {"lockupViewModel": {
+        "contentId": vid, "contentType": ctype,
+        "metadata": {"lockupMetadataViewModel": {
+            "title": {"content": "Title " + vid},
+            "metadata": {"contentMetadataViewModel": {"metadataRows": [
+                {"metadataParts": [{"text": {"content": "100K views"}},
+                                   {"text": {"content": rel}, "accessibilityLabel": rel}]}]}}}}}}}}
+
+
+def _cont_item(token):
+    return {"continuationItemRenderer": {"continuationEndpoint": {"continuationCommand": {"token": token}}}}
+
+
+def _page(items, token=None):
+    contents = list(items) + ([_cont_item(token)] if token else [])
+    return {"contents": {"twoColumnBrowseResultsRenderer": {"tabs": [
+        {"tabRenderer": {"content": {"richGridRenderer": {"contents": contents}}}}]}}}
+
+
+def test_relative_to_date():
+    now = date(2026, 6, 17)
+    assert yt.relative_to_date("9 hours ago", now) == "2026-06-17"     # same day
+    assert yt.relative_to_date("2 days ago", now) == "2026-06-15"
+    assert yt.relative_to_date("3 weeks ago", now) == "2026-05-27"
+    assert yt.relative_to_date("Streamed 2 years ago", now) == "2024-06-17"   # prefix tolerated
+    assert yt.relative_to_date("Premiered 5 months ago", now) == "2026-01-16"   # ~152 days (approx, fine for years)
+    assert yt.relative_to_date("no date here", now) is None
+    assert yt.relative_to_date("", now) is None
+
+
+def test_innertube_parse_videos_filters_non_videos_and_undated():
+    page = _page([
+        _lk_item("v1", "2 years ago"),
+        _lk_item("v2", "3 months ago"),
+        _lk_item("p1", "1 day ago", ctype="LOCKUP_CONTENT_TYPE_PLAYLIST"),  # not a video → skip
+        _lk_item("v3", "No views at all"),                                  # no 'ago' → skip
+    ])
+    assert yt.innertube_parse_videos(page) == [("v1", "2 years ago"), ("v2", "3 months ago")]
+
+
+def test_innertube_continuation_picks_pagination_token():
+    page = _page([_lk_item("v1", "1 day ago")], token="REAL")
+    page["decoy"] = {"some": {"continuationCommand": {"token": "DECOY-menu-token"}}}  # not a continuationItemRenderer
+    assert yt.innertube_continuation(page) == "REAL"
+    assert yt.innertube_continuation(_page([_lk_item("v1", "1 day ago")])) is None    # no token
+
+
+def test_innertube_channel_dates_paginates_and_converts():
+    now = date(2026, 6, 17)
+    pages = [
+        _page([_lk_item("v1", "1 year ago"), _lk_item("v2", "2 years ago")], token="TOK"),
+        _page([_lk_item("v3", "5 days ago")]),   # no token → stop
+    ]
+    seen = {"n": 0}
+
+    def post(payload):
+        i = seen["n"]; seen["n"] += 1
+        if i == 0:
+            assert payload.get("browseId") == "UCxxxx" and payload.get("params")
+        else:
+            assert payload.get("continuation") == "TOK"
+        return pages[i]
+
+    out = yt.innertube_channel_dates("UCxxxx", pages=5, now=now, post=post)
+    assert set(out) == {"v1", "v2", "v3"}
+    assert out["v3"] == "2026-06-12" and out["v1"] == "2025-06-17"
+    assert seen["n"] == 2   # stopped when no continuation (didn't keep hammering)
+
+
+def test_innertube_channel_dates_guards():
+    assert yt.innertube_channel_dates("not-a-uc-id", post=lambda p: {}) == {}     # only UC… channels
+    assert yt.innertube_channel_dates("UCx", post=lambda p: None) == {}           # bad/empty response → {}
+    assert yt.innertube_channel_dates("UCx", post=lambda p: (_ for _ in ()).throw(RuntimeError())) == {}

tests/test_video_youtube_enrichment.py

@@ -13,27 +13,27 @@ 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):
+def test_enrich_caches_bulk_dates_and_marks_done(db, monkeypatch):
     import core.video.youtube as yt
-    monkeypatch.setattr(yt, "proxy_channel_dates",
+    # InnerTube (primary) covers everything → per-video fallback must NOT run
+    monkeypatch.setattr(yt, "innertube_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.set_setting("youtube_proxy_instances", "piped|https://example.test")   # proxy is opt-in
     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)
+    # already enriched → second pass is a no-op (no fetch)
-    monkeypatch.setattr(yt, "proxy_channel_dates", lambda *a, **k: (_ for _ in ()).throw(AssertionError("re-swept")))
+    monkeypatch.setattr(yt, "innertube_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):
+def test_enrich_falls_back_to_per_video_when_bulk_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, "innertube_channel_dates", lambda cid, *a, **k: {})   # InnerTube empty
+    monkeypatch.setattr(yt, "proxy_channel_dates", lambda cid, *a, **k: {})       # no proxy either
     # flat resolve adds a recent upload r1 to the date-fallback set (besides wished w1/w2)
     monkeypatch.setattr(yt, "resolve_channel",
                         lambda url, **k: {"videos": [{"youtube_id": "r1", "title": "Recent"}]})