pipeline/search.py

"""Stage 1: find the top videos for a topic and pick candidates by an engagement signal.

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

**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

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

# 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"}

# 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)
    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.

    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)
        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)
    _INSTANCE_CACHE = (now, insts)
    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, 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:
                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


# ---------------------------------------------------------------- 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()
    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
        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")),
            "subscribers": _nn(st.get("uploaderSubscriberCount")),
            "comments": comments,
        })
    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.

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


# ---------------------------------------------------------------- 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))),
              "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"],
                        "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``, best-effort engagement-ranked.

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

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

    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]