Video search via Piped API with normalized engagement ranking + instance failover; Data API/yt-dlp fallback

pipeline/search.py

@@ -1,77 +1,242 @@
-"""Stage 1: search the top videos for a topic via the adarshajay/youtube-search Space.
+"""Stage 1: find the top videos for a topic and pick candidates by an engagement signal.
 
-The upstream Space exposes ``api_name="/youtube_search"``. It takes a single query
-string and returns a single plain-text blob listing the top 5 results (title, URL and
-video id only). We parse the ids/titles out of that text.
+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.
+
+Engagement (each metric 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.
 """
 from __future__ import annotations
 
-import re
+import html
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
 
-from gradio_client import Client
+# Live instance list (best-effort) + a seed list to fall back on.
+PIPED_INSTANCE_LIST = "https://piped-instances.kavin.rocks/"
+SEED_INSTANCES = [
+    "https://api.piped.private.coffee",
+    "https://pipedapi.kavin.rocks",
+    "https://pipedapi.adminforge.de",
+    "https://pipedapi.drgns.space",
+    "https://pipedapi.ducks.party",
+    "https://pipedapi.reallyaweso.me",
+    "https://piped-api.lunar.icu",
+    "https://pipedapi.r4fo.com",
+    "https://pipedapi.phoenixthrush.com",
+    "https://api.piped.yt",
+]
 
-SEARCH_SPACE = "adarshajay/youtube-search"
-SEARCH_API = "/youtube_search"
+# Engagement weights (subscribers down-weighted: channel-level, not video-level).
+W_LIKE, W_COMMENT, W_SUB, W_DISLIKE = 1.0, 1.0, 0.5, 1.0
 
-# 11-char YouTube ids as they appear in watch?v=... or youtu.be/... URLs.
-_ID_RE = re.compile(r"(?:v=|youtu\.be/|/watch/|/embed/)([A-Za-z0-9_-]{11})")
+SEARCH_API = "https://www.googleapis.com/youtube/v3/search"
+_UA = {"User-Agent": "TutorialMaker/1.0"}
 
 
-def _parse_titles(text: str, ids: list[str]) -> dict[str, str]:
-    """Best-effort: map each id to a nearby title line in the result blob.
+def _get_json(url: str, timeout: int = 15):
+    req = urllib.request.Request(url, headers=_UA)
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return json.load(resp)
 
-    The upstream format is not guaranteed, so this is intentionally forgiving: for each
-    id we take the longest non-URL line that appears on or just before the line holding
-    the id. Anything we can't resolve falls back to the id itself in ``search_top5``.
-    """
-    lines = text.splitlines()
-    titles: dict[str, str] = {}
-    for vid in ids:
-        idx = next((i for i, ln in enumerate(lines) if vid in ln), None)
-        if idx is None:
+
+def _instances() -> list[str]:
+    """Live Piped instances (dynamic list first, then seeds), deduped in order."""
+    insts: list[str] = []
+    try:
+        data = _get_json(PIPED_INSTANCE_LIST, timeout=10)
+        for entry in data if isinstance(data, list) else []:
+            api = (entry or {}).get("api_url")
+            if api:
+                insts.append(api.rstrip("/"))
+    except Exception:
+        pass
+    for s in SEED_INSTANCES:
+        s = s.rstrip("/")
+        if s not in insts:
+            insts.append(s)
+    return insts
+
+
+class _Piped:
+    """Fetches Piped API paths, sticking to a working instance and rotating on failure."""
+
+    def __init__(self):
+        self.instances = _instances()
+        self.current = None
+
+    def get(self, path: str, timeout: int = 15):
+        order = ([self.current] if self.current else [])
+        order += [i for i in self.instances if i != self.current]
+        last = None
+        for inst in order:
+            try:
+                data = _get_json(inst + path, timeout=timeout)
+                self.current = inst  # remember the one that worked
+                return data
+            except Exception as exc:
+                last = exc
+                continue
+        raise RuntimeError(f"all Piped instances failed for {path}: {last}")
+
+
+def _vid_from_watch(url: str) -> str | None:
+    if not url:
+        return None
+    q = urllib.parse.urlparse(url).query
+    return urllib.parse.parse_qs(q).get("v", [None])[0]
+
+
+def _nn(x) -> int:
+    """Non-negative int (Piped returns -1 when a metric is unavailable)."""
+    try:
+        v = int(x)
+    except (TypeError, ValueError):
+        return 0
+    return v if v > 0 else 0
+
+
+def _search_piped(topic: str, pool: int) -> list[dict]:
+    """Piped search + per-candidate engagement metrics. Returns unscored candidates."""
+    p = _Piped()
+    data = p.get("/search?" + urllib.parse.urlencode({"q": topic, "filter": "videos"}))
+    items = [it for it in (data.get("items") or [])
+             if str(it.get("url", "")).startswith("/watch")][:pool]
+    cands: list[dict] = []
+    for it in items:
+        vid = _vid_from_watch(it.get("url", ""))
+        if not vid:
             continue
-        window = lines[max(0, idx - 1): idx + 1]
-        cands = [ln.strip(" -*\t") for ln in window if "http" not in ln and vid not in ln]
-        cands = [c for c in cands if c]
-        if cands:
-            titles[vid] = max(cands, key=len)
-    return titles
+        try:
+            st = p.get(f"/streams/{vid}")
+        except Exception:
+            continue  # can't score this one; skip
+        try:
+            comments = _nn(p.get(f"/comments/{vid}").get("commentCount"))
+        except Exception:
+            comments = 0  # best-effort; don't drop the candidate
+        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),
+            "views": _nn(st.get("views")),
+            "likes": _nn(st.get("likes")),
+            "dislikes": _nn(st.get("dislikes")),
+            "subscribers": _nn(st.get("uploaderSubscriberCount")),
+            "comments": comments,
+        })
+    return cands
+
+
+def _rank_by_engagement(cands: list[dict]) -> list[dict]:
+    """Attach a normalized weighted ``engagement`` score and sort desc.
+
+    Each metric is min-max normalized across the pool so wildly different scales
+    (subscribers in millions vs comments in thousands) contribute comparably.
+    """
+    if not cands:
+        return cands
+
+    def norm(key: str) -> list[float]:
+        vals = [c.get(key, 0) or 0 for c in cands]
+        lo, hi = min(vals), max(vals)
+        if hi == lo:
+            return [0.5] * len(vals)  # neutral when all equal
+        return [(v - lo) / (hi - lo) for v in vals]
+
+    nl, nc, ns, nd = (norm("likes"), norm("comments"),
+                      norm("subscribers"), norm("dislikes"))
+    for i, c in enumerate(cands):
+        c["engagement"] = round(
+            W_LIKE * nl[i] + W_COMMENT * nc[i] + W_SUB * ns[i] - W_DISLIKE * nd[i], 4)
+    return sorted(cands, key=lambda c: -c["engagement"])
+
+
+# ---------------------------------------------------------------- fallbacks (no engagement)
+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))),
+              "order": "relevance", "key": api_key}
+    try:
+        data = _get_json(SEARCH_API + "?" + urllib.parse.urlencode(params), timeout=30)
+    except urllib.error.HTTPError as exc:
+        body = exc.read().decode("utf-8", "ignore")
+        raise RuntimeError(f"Data API search HTTP {exc.code}: {body[:160]}") from exc
+    out = []
+    for item in data.get("items", []):
+        vid = item.get("id", {}).get("videoId")
+        if vid:
+            out.append({"video_id": vid,
+                        "url": f"https://www.youtube.com/watch?v={vid}",
+                        "title": html.unescape((item.get("snippet") or {}).get("title", "") or vid)})
+    return out
+
+
+def _search_ytdlp(topic: str, max_results: int, proxy: str | None) -> list[dict]:
+    from yt_dlp import YoutubeDL
+    opts = {"quiet": True, "no_warnings": True, "skip_download": True, "extract_flat": True}
+    if proxy:
+        opts["proxy"] = proxy
+        if os.environ.get("SSL_CERT_FILE"):
+            opts["compat_opts"] = ["no-certifi"]
+    with YoutubeDL(opts) as ydl:
+        info = ydl.extract_info(f"ytsearch{max_results}:{topic}", download=False)
+    out = []
+    for e in (info.get("entries") or [])[:max_results]:
+        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"]})
+    return out
 
 
-def search_top5(topic: str, max_results: int = 5) -> list[dict]:
-    """Return up to ``max_results`` unique videos for ``topic``.
+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.
 
-    Each item: ``{"video_id", "url", "title"}``. Raises RuntimeError if the upstream
-    Space is unreachable or returns no parseable video ids.
+    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).
     """
     topic = (topic or "").strip()
     if not topic:
         raise ValueError("Please enter a topic to search for.")
 
+    errors: list[str] = []
     try:
-        client = Client(SEARCH_SPACE)
-        raw = client.predict(topic, api_name=SEARCH_API)
-    except Exception as exc:  # network / Space down / API renamed
-        raise RuntimeError(
-            f"Could not reach the search Space '{SEARCH_SPACE}'. It may be sleeping or "
-            f"its API changed. Details: {exc}"
-        ) from exc
-
-    text = raw if isinstance(raw, str) else str(raw)
-    ids = list(dict.fromkeys(_ID_RE.findall(text)))  # preserve order, dedupe
-    if not ids:
-        raise RuntimeError(
-            "The search Space returned no recognizable YouTube video ids. Raw output:\n"
-            + text[:500]
-        )
-
-    ids = ids[:max_results]
-    titles = _parse_titles(text, ids)
-    return [
-        {
-            "video_id": vid,
-            "url": f"https://www.youtube.com/watch?v={vid}",
-            "title": titles.get(vid, vid),
-        }
-        for vid in ids
-    ]
+        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.
+    if api_key:
+        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}")
+
+    raise RuntimeError("Video search failed. " + " | ".join(errors)[:400])