"""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. 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 html import json import os import urllib.error import urllib.parse import urllib.request # 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", ] # 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 SEARCH_API = "https://www.googleapis.com/youtube/v3/search" _UA = {"User-Agent": "TutorialMaker/1.0"} 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) 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 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, 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. 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: 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])