pipeline/search.py

"""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])