Search: direct scrape as primary tier, Piped as fallback

Discovery and engagement scoring are now separate concerns, because they fail
independently.

Discovery is tiered, first tier with results wins:
direct scrape -> Piped -> Data API search.list -> yt-dlp ytsearch.

The scrape runs in-process, so it depends on no third party, costs no API quota,
honours YT_PROXY (residential egress), and yields view count + duration straight
from YouTube. Piped keeps its instance failover and remains the fallback.

Engagement metrics (likes/dislikes/subscribers/comments) are attached from Piped
to whichever tier produced the candidates, so scrape-first does not lose the
normalized weighted ranking. When Piped is unreachable, search still succeeds and
ranking falls through to the sentiment stage as before.

Enrichment is deliberately impatient: 3 instances max on a 6s timeout, abandoned
on the first unreachable candidate, plus a 15-minute TTL cache on the instance
list. Without those bounds a dead Piped cost 52.8s per search; now 4.1s cold,
2.6s warm.

Also drops sub-minute Shorts before a (quota-capped) download is spent on one,
with a starve guard and treating unknown duration as keep. No caption filter:
transcripts come from faster-whisper ASR on the downloaded video, so videos
without a caption track are perfectly usable.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

README.md

@@ -20,7 +20,14 @@ single best YouTube video on that topic. **No video download** — acquisition i
 
 ## Pipeline
 
-1. **Search** — top 5 videos via the [`adarshajay/youtube-search`](https://huggingface.co/spaces/adarshajay/youtube-search) Space.
+1. **Search** — tiered discovery, first tier with results wins:
+   **direct scrape** of YouTube's results page (no third party, no API quota, honours
+   `YT_PROXY`) → **Piped API** (with instance failover) → **Data API `search.list`** →
+   **yt-dlp `ytsearch`**. Sub-minute Shorts are dropped so a download request is never
+   wasted on one. Engagement metrics (likes / dislikes / subscribers / comment count) are
+   then attached from Piped when the winning tier didn't already supply them, and
+   candidates are ranked by a normalized weighted score. If Piped is unreachable the
+   search still succeeds — ranking simply falls to the sentiment stage.
 2. **Sentiment rank** — fetch each video's comments via the **YouTube Data API v3** and
    score them with the BERT classifier
    [`OmarMedhat7/youtube-sentiment-analysis-model`](https://huggingface.co/OmarMedhat7/youtube-sentiment-analysis-model);

app.py

@@ -134,7 +134,8 @@ def check_access():
         f"- **Comment sentiment** (`YOUTUBE_API_KEY`): "
         f"{'set ✅' if os.environ.get('YOUTUBE_API_KEY') else 'not set ⚪ (sentiment skipped)'}",
         "- **Transcript**: faster-whisper, local (no key/proxy needed) ✅",
-        "- **Search**: Piped API (no key needed) ✅",
+        f"- **Search**: direct scrape → Piped → Data API → yt-dlp (no key needed) ✅"
+        f"{' · scrape routed via `YT_PROXY`' if _resolve_proxy() else ''}",
     ]
 
     api_ok = None
@@ -230,11 +231,14 @@ def run_pipeline(topic, hf_token, yt_api_key, llm_model, vlm_model,
         # 1. Search ------------------------------------------------------------------
         progress(0.03, desc="Searching")
         yield status(f"🔍 Searching top videos for **{topic}** (engagement-ranked)…"), gr.update(), gr.update(), gr.update()
-        videos = search_mod.search_top5(topic, api_key=api_key)
+        videos = search_mod.search_top5(topic, api_key=api_key, proxy=_resolve_proxy())
         eng_note = ""
         if videos and "engagement" in videos[0]:
             eng_note = " • ranked by likes+comments+subscribers−dislikes (normalized)"
-        yield status(f"Found {len(videos)} candidate videos{eng_note}."), gr.update(), gr.update(), gr.update()
+        found = f"Found {len(videos)} candidate videos via {videos[0].get('tier', 'search')}{eng_note}."
+        if videos and videos[0].get("filter_note"):
+            found += f" ({videos[0]['filter_note']})"
+        yield status(found), gr.update(), gr.update(), gr.update()
 
         # 2. Sentiment ranking (YouTube Data API comments) ---------------------------
         if api_key:

pipeline/search.py

@@ -1,25 +1,41 @@
 """Stage 1: find the top videos for a topic and pick candidates by an engagement signal.
 
-Primary source is the **Piped API** (a privacy frontend for YouTube). Piped's
-``/streams/{id}`` exposes likes, dislikes and the uploader's subscriber count, and
-``/comments/{id}`` exposes the comment count — everything needed for a per-video
-engagement score. Public Piped instances are ephemeral, so we discover a live instance
-list and **fail over across instances** on any error.
+Discovery and engagement scoring are separate concerns here, because they fail
+independently.
 
-Engagement (each metric min-max normalized across the candidate pool, then weighted):
+**Discovery** is tiered — the first tier that returns candidates wins:
+
+1. **Direct scrape** of ``youtube.com/results``. Runs in this process, so it depends on
+   no third party, costs no API quota, and honours ``YT_PROXY`` — meaning it can exit
+   from a residential IP rather than the Space's datacenter one. It also yields view
+   count and duration straight from YouTube.
+2. **Piped API** (a privacy frontend for YouTube). Public instances are ephemeral, so we
+   discover a live instance list and **fail over across instances** on any error.
+3. **YouTube Data API** ``search.list`` — deterministic, but 100 quota units a call.
+4. **yt-dlp** ``ytsearch`` — last resort.
+
+**Engagement** is a best-effort layer applied to whichever tier produced the candidates.
+Piped's ``/streams/{id}`` exposes likes, dislikes and the uploader's subscriber count, and
+``/comments/{id}`` the comment count. Each metric is min-max normalized across the
+candidate pool, then weighted:
 
     score = w_like*likes + w_comment*comments + w_sub*subscribers - w_dislike*dislikes
 
-Engagement picks the top ``max_results`` candidates; the downstream sentiment stage then
-decides the final winner. If Piped is entirely unreachable we fall back to the YouTube
-Data API (``search.list``) or yt-dlp ``ytsearch`` — without engagement metadata, letting
-sentiment alone rank.
+Tier 2 gets these for free while searching; the other tiers have them fetched separately.
+If Piped is unreachable entirely, candidates are returned in discovery order and the
+downstream sentiment stage alone decides the winner — exactly as before.
+
+Shorts are dropped before the caller spends a (quota-capped) video download on them.
+Note there is deliberately **no caption filter**: transcripts come from faster-whisper ASR
+on the downloaded video, so a video without a caption track works just as well.
 """
 from __future__ import annotations
 
 import html
 import json
 import os
+import re
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -45,6 +61,37 @@ W_LIKE, W_COMMENT, W_SUB, W_DISLIKE = 1.0, 1.0, 0.5, 1.0
 SEARCH_API = "https://www.googleapis.com/youtube/v3/search"
 _UA = {"User-Agent": "TutorialMaker/1.0"}
 
+# Piped instance list is cached process-wide; public instances churn, so not for long.
+INSTANCE_TTL = 900
+_INSTANCE_CACHE: tuple[float, list[str]] | None = None
+
+# Engagement enrichment is optional, so it must fail *fast* — when Piped is down we'd
+# otherwise pay a full rotation across every instance before giving up on a search that
+# already has its candidates. Discovery (tier 2) keeps the patient full-rotation budget.
+ENRICH_MAX_INSTANCES = 3
+ENRICH_TIMEOUT = 6
+
+# --- direct scrape ---------------------------------------------------------------
+RESULTS_URL = "https://www.youtube.com/results"
+# YouTube's "Type: Video" result filter — keeps channels and playlists out.
+_SP_VIDEOS_ONLY = "EgIQAQ%3D%3D"
+_BROWSER_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+               "(KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36")
+
+# Anything this short is a Short (or a trailer) — never a tutorial worth downloading.
+SHORTS_MAX_SECONDS = 60
+
+# Raw ids in the results page JSON, used only when the ytInitialData walk comes up empty.
+_BARE_ID_RE = re.compile(r'"videoId"\s*:\s*"([A-Za-z0-9_-]{11})"')
+# Strip credentials from any proxy URL an exception might echo, so a configured
+# http://user:pass@host proxy never leaks into the UI or logs.
+_CRED_RE = re.compile(r"(https?://)[^/@\s]+@")
+_VIEWS_RE = re.compile(r"([\d.,]+)\s*([KMB])?", re.I)
+
+
+def _redact(text) -> str:
+    return _CRED_RE.sub(r"\1", str(text))
+
 
 def _get_json(url: str, timeout: int = 15):
     req = urllib.request.Request(url, headers=_UA)
@@ -53,7 +100,17 @@ def _get_json(url: str, timeout: int = 15):
 
 
 def _instances() -> list[str]:
-    """Live Piped instances (dynamic list first, then seeds), deduped in order."""
+    """Live Piped instances (dynamic list first, then seeds), deduped in order.
+
+    Cached for ``INSTANCE_TTL`` so a Space serving many searches doesn't re-pay the
+    instance-list fetch every time — but still short enough to pick up churn, since
+    public instances come and go.
+    """
+    global _INSTANCE_CACHE
+    now = time.time()
+    if _INSTANCE_CACHE and now - _INSTANCE_CACHE[0] < INSTANCE_TTL:
+        return _INSTANCE_CACHE[1]
+
     insts: list[str] = []
     try:
         data = _get_json(PIPED_INSTANCE_LIST, timeout=10)
@@ -67,6 +124,7 @@ def _instances() -> list[str]:
         s = s.rstrip("/")
         if s not in insts:
             insts.append(s)
+    _INSTANCE_CACHE = (now, insts)
     return insts
 
 
@@ -77,9 +135,11 @@ class _Piped:
         self.instances = _instances()
         self.current = None
 
-    def get(self, path: str, timeout: int = 15):
+    def get(self, path: str, timeout: int = 15, max_instances: int | None = None):
         order = ([self.current] if self.current else [])
         order += [i for i in self.instances if i != self.current]
+        if max_instances:
+            order = order[:max_instances]
         last = None
         for inst in order:
             try:
@@ -108,6 +168,153 @@ def _nn(x) -> int:
     return v if v > 0 else 0
 
 
+# ---------------------------------------------------------------- tier 1: direct scrape
+def _initial_data(page: str) -> dict | None:
+    """Pull the ``ytInitialData`` JSON blob out of a results page.
+
+    Brace-matched rather than regex'd: the blob contains plenty of nested braces and
+    escaped quotes inside string literals.
+    """
+    for marker in ('var ytInitialData = ', 'window["ytInitialData"] = ', 'ytInitialData = '):
+        i = page.find(marker)
+        if i == -1:
+            continue
+        start = page.find("{", i)
+        if start == -1:
+            continue
+        depth, in_str, esc = 0, False, False
+        for j in range(start, len(page)):
+            ch = page[j]
+            if in_str:
+                if esc:
+                    esc = False
+                elif ch == "\\":
+                    esc = True
+                elif ch == '"':
+                    in_str = False
+                continue
+            if ch == '"':
+                in_str = True
+            elif ch == "{":
+                depth += 1
+            elif ch == "}":
+                depth -= 1
+                if depth == 0:
+                    try:
+                        return json.loads(page[start:j + 1])
+                    except json.JSONDecodeError:
+                        break  # try the next marker
+    return None
+
+
+def _walk_renderers(node, out: list) -> None:
+    """Collect every ``videoRenderer`` dict anywhere in the response tree."""
+    if isinstance(node, dict):
+        vr = node.get("videoRenderer")
+        if isinstance(vr, dict) and vr.get("videoId"):
+            out.append(vr)
+        for value in node.values():
+            _walk_renderers(value, out)
+    elif isinstance(node, list):
+        for value in node:
+            _walk_renderers(value, out)
+
+
+def _renderer_text(field) -> str:
+    """YouTube renders text as either ``simpleText`` or a list of ``runs``."""
+    if not isinstance(field, dict):
+        return ""
+    if field.get("simpleText"):
+        return str(field["simpleText"]).strip()
+    return "".join(r.get("text", "") for r in field.get("runs") or []).strip()
+
+
+def _hms_to_seconds(text: str | None) -> int | None:
+    """Parse a ``lengthText`` like ``12:34`` or ``1:02:03`` into seconds."""
+    text = (text or "").strip()
+    if not text:
+        return None
+    try:
+        nums = [int(p) for p in text.split(":")]
+    except ValueError:
+        return None  # "LIVE", "SHORTS", etc.
+    total = 0
+    for n in nums:
+        total = total * 60 + n
+    return total
+
+
+def _parse_views(text: str | None) -> int:
+    """Parse a ``viewCountText`` like ``1,234 views`` or ``1.2M views`` into an int."""
+    text = (text or "").strip()
+    if not text:
+        return 0
+    m = _VIEWS_RE.match(text)
+    if not m:
+        return 0
+    try:
+        value = float(m.group(1).replace(",", ""))
+    except ValueError:
+        return 0
+    return int(value * {"k": 1e3, "m": 1e6, "b": 1e9}.get((m.group(2) or "").lower(), 1))
+
+
+def _parse_results_page(page: str) -> list[dict]:
+    data = _initial_data(page)
+    videos: list[dict] = []
+    seen: set[str] = set()
+
+    if data:
+        renderers: list[dict] = []
+        _walk_renderers(data, renderers)
+        for vr in renderers:
+            vid = vr.get("videoId")
+            if not vid or vid in seen:
+                continue
+            seen.add(vid)
+            videos.append({
+                "video_id": vid,
+                "url": f"https://www.youtube.com/watch?v={vid}",
+                "title": html.unescape(_renderer_text(vr.get("title")) or vid),
+                "channel": _renderer_text(vr.get("ownerText")),
+                "duration_s": _hms_to_seconds(_renderer_text(vr.get("lengthText"))),
+                "views": _parse_views(_renderer_text(vr.get("viewCountText"))),
+            })
+
+    if not videos:
+        # Parser drift (YouTube reshuffles this tree periodically): fall back to raw ids
+        # in page order. Less precise — may catch a shelf or promo — but still usable.
+        for vid in dict.fromkeys(_BARE_ID_RE.findall(page)):
+            videos.append({
+                "video_id": vid,
+                "url": f"https://www.youtube.com/watch?v={vid}",
+                "title": vid,
+                "channel": "",
+                "duration_s": None,
+                "views": 0,
+            })
+    return videos
+
+
+def _search_scrape(topic: str, pool: int, proxy: str | None, timeout: int = 30) -> list[dict]:
+    """Fetch and parse YouTube's results page ourselves, through ``proxy`` when set."""
+    import requests
+
+    url = (f"{RESULTS_URL}?{urllib.parse.urlencode({'search_query': topic})}"
+           f"&sp={_SP_VIDEOS_ONLY}")
+    headers = {
+        "User-Agent": _BROWSER_UA,
+        "Accept-Language": "en-US,en;q=0.9",
+        # Skip the EU consent interstitial, which otherwise replaces the results page.
+        "Cookie": "CONSENT=YES+1; SOCS=CAI",
+    }
+    proxies = {"http": proxy, "https": proxy} if proxy else None
+    resp = requests.get(url, headers=headers, proxies=proxies, timeout=timeout)
+    resp.raise_for_status()
+    return _parse_results_page(resp.text)[:pool]
+
+
+# ---------------------------------------------------------------- tier 2: Piped
 def _search_piped(topic: str, pool: int) -> list[dict]:
     """Piped search + per-candidate engagement metrics. Returns unscored candidates."""
     p = _Piped()
@@ -127,10 +334,13 @@ def _search_piped(topic: str, pool: int) -> list[dict]:
             comments = _nn(p.get(f"/comments/{vid}").get("commentCount"))
         except Exception:
             comments = 0  # best-effort; don't drop the candidate
+        duration = _nn(it.get("duration")) or None
         cands.append({
             "video_id": vid,
             "url": f"https://www.youtube.com/watch?v={vid}",
             "title": html.unescape(it.get("title") or st.get("title") or vid),
+            "channel": (it.get("uploaderName") or "").strip(),
+            "duration_s": duration,
             "views": _nn(st.get("views")),
             "likes": _nn(st.get("likes")),
             "dislikes": _nn(st.get("dislikes")),
@@ -140,6 +350,47 @@ def _search_piped(topic: str, pool: int) -> list[dict]:
     return cands
 
 
+def _enrich_engagement(cands: list[dict]) -> bool:
+    """Best-effort: attach Piped engagement metrics to candidates that lack them.
+
+    Lets tiers 1/3/4 be ranked by the same signal tier 2 gets for free. Returns whether
+    any candidate was enriched.
+
+    Deliberately impatient: each call tries at most ``ENRICH_MAX_INSTANCES`` instances on
+    a short timeout, and the whole pass is abandoned the first time a candidate can't be
+    reached. Ranking is a nice-to-have — a dead Piped must cost seconds, not a minute,
+    since the caller already has its candidates and degrades to sentiment-only ranking.
+    """
+    missing = [c for c in cands if "likes" not in c]
+    if not missing:
+        return False
+
+    p = _Piped()
+    enriched = 0
+    for c in missing:
+        vid = c["video_id"]
+        try:
+            st = p.get(f"/streams/{vid}", timeout=ENRICH_TIMEOUT,
+                       max_instances=ENRICH_MAX_INSTANCES)
+        except Exception:
+            break  # Piped unreachable — stop trying
+        try:
+            c["comments"] = _nn(p.get(f"/comments/{vid}", timeout=ENRICH_TIMEOUT,
+                                      max_instances=ENRICH_MAX_INSTANCES)
+                                .get("commentCount"))
+        except Exception:
+            c["comments"] = 0
+        c["likes"] = _nn(st.get("likes"))
+        c["dislikes"] = _nn(st.get("dislikes"))
+        c["subscribers"] = _nn(st.get("uploaderSubscriberCount"))
+        if not c.get("views"):
+            c["views"] = _nn(st.get("views"))
+        if not c.get("duration_s"):
+            c["duration_s"] = _nn(st.get("duration")) or None
+        enriched += 1
+    return enriched > 0
+
+
 def _rank_by_engagement(cands: list[dict]) -> list[dict]:
     """Attach a normalized weighted ``engagement`` score and sort desc.
 
@@ -164,7 +415,7 @@ def _rank_by_engagement(cands: list[dict]) -> list[dict]:
     return sorted(cands, key=lambda c: -c["engagement"])
 
 
-# ---------------------------------------------------------------- fallbacks (no engagement)
+# ---------------------------------------------------------------- tiers 3 & 4
 def _search_data_api(topic: str, api_key: str, max_results: int) -> list[dict]:
     params = {"part": "snippet", "q": topic, "type": "video",
               "maxResults": str(max(1, min(max_results, 50))),
@@ -198,45 +449,78 @@ def _search_ytdlp(topic: str, max_results: int, proxy: str | None) -> list[dict]
         if e.get("id"):
             out.append({"video_id": e["id"],
                         "url": e.get("url") or f"https://www.youtube.com/watch?v={e['id']}",
-                        "title": e.get("title") or e["id"]})
+                        "title": e.get("title") or e["id"],
+                        "duration_s": _nn(e.get("duration")) or None})
     return out
 
 
+# ---------------------------------------------------------------- public
+def _drop_shorts(cands: list[dict]) -> tuple[list[dict], str | None]:
+    """Remove sub-minute videos, which make poor tutorials and waste a download request.
+
+    Never starves the pipeline: if every candidate looks like a Short (usually a bad
+    duration read rather than a page of Shorts), keep them all and say so.
+    """
+    kept = [c for c in cands
+            if c.get("duration_s") is None or c["duration_s"] >= SHORTS_MAX_SECONDS]
+    if len(kept) == len(cands):
+        return cands, None
+    if not kept:
+        return cands, "every candidate looked like a Short — kept them all"
+    return kept, f"dropped {len(cands) - len(kept)} Short(s) under {SHORTS_MAX_SECONDS}s"
+
+
 def search_top5(topic: str, api_key: str | None = None, proxy: str | None = None,
                 max_results: int = 5, pool: int = 8) -> list[dict]:
-    """Return up to ``max_results`` videos for ``topic``, chosen by engagement.
+    """Return up to ``max_results`` videos for ``topic``, best-effort engagement-ranked.
 
-    Piped supplies candidates + engagement metrics; the top ``max_results`` by the
-    normalized weighted score are returned (each item carries ``views/likes/dislikes/
-    subscribers/comments/engagement`` for transparency). On total Piped failure, falls
-    back to the Data API or yt-dlp (candidates only, no engagement).
+    Discovery falls through direct scrape -> Piped -> Data API -> yt-dlp; the first tier
+    with results wins. Engagement metrics are then attached from Piped when the winning
+    tier didn't already supply them, and candidates carrying metrics are sorted by the
+    normalized weighted score. Without metrics they stay in discovery order and the
+    sentiment stage ranks them.
+
+    Each item carries at least ``video_id/url/title``, plus ``tier`` and whichever of
+    ``views/likes/dislikes/subscribers/comments/duration_s/engagement`` were available.
     """
     topic = (topic or "").strip()
     if not topic:
         raise ValueError("Please enter a topic to search for.")
 
-    errors: list[str] = []
-    try:
-        cands = _search_piped(topic, max(pool, max_results))
-        if cands:
-            return _rank_by_engagement(cands)[:max_results]
-        errors.append("Piped: no scorable candidates")
-    except Exception as exc:
-        errors.append(f"Piped: {exc}")
-
-    # Fallbacks — search only, sentiment stage will do the ranking.
+    want = max(pool, max_results)
+    tiers = [
+        ("direct scrape" + (" via proxy" if proxy else ""),
+         lambda: _search_scrape(topic, want, proxy)),
+        ("Piped", lambda: _search_piped(topic, want)),
+    ]
     if api_key:
+        tiers.append(("Data API", lambda: _search_data_api(topic, api_key, max_results)))
+    tiers.append(("yt-dlp", lambda: _search_ytdlp(topic, max_results, proxy)))
+
+    cands: list[dict] = []
+    tier_label = ""
+    errors: list[str] = []
+    for label, fetch in tiers:
         try:
-            res = _search_data_api(topic, api_key, max_results)
-            if res:
-                return res[:max_results]
-        except Exception as exc:
-            errors.append(f"Data API: {exc}")
-    try:
-        res = _search_ytdlp(topic, max_results, proxy)
-        if res:
-            return res[:max_results]
-    except Exception as exc:
-        errors.append(f"yt-dlp: {exc}")
+            found = fetch()
+        except Exception as exc:  # noqa: BLE001 - any tier may fail; try the next
+            errors.append(f"{label}: {_redact(exc)[:160]}")
+            continue
+        if found:
+            cands, tier_label = found, label
+            break
+        errors.append(f"{label}: no candidates")
 
-    raise RuntimeError("Video search failed. " + " | ".join(errors)[:400])
+    if not cands:
+        raise RuntimeError("Video search failed. " + " | ".join(errors)[:400])
+
+    cands, shorts_note = _drop_shorts(cands)
+    _enrich_engagement(cands)
+    if any("likes" in c for c in cands):
+        cands = _rank_by_engagement(cands)
+
+    for c in cands:
+        c["tier"] = tier_label
+        if shorts_note:
+            c["filter_note"] = shorts_note
+    return cands[:max_results]