core/app.py

"""Kick VOD streamer surface.

Exactly three endpoints plus a health check, all Kick-only:

    * ``GET /``
    * ``GET /health``
    * ``GET /formats?url=...``                  -> public quality list
    * ``GET /formats?report=1&url=...``         -> per-solution listing evidence
    * ``GET /formats?report=1&url=...&probe=1&format_id=best``
                                                -> per-solution byte probe
    * ``GET /stream?url=...&format_id=best``

Every one of them takes an optional ``&solution=<name>`` that restricts the run
to a single rung, so each step can be watched on its own instead of guessing
which rung inside the ladder produced a result:

    * ``GET /formats?url=...&solution=yt-dlp``           -> what that rung lists
    * ``GET /formats?report=1&url=...&solution=yt-dlp``  -> its evidence alone
    * ``GET /formats?...&probe=1&solution=yt-dlp``       -> does it stream bytes
    * ``GET /stream?url=...&solution=yt-dlp``            -> stream from it only

Nothing about Twitch or YouTube lives in this file. The ``/debug/*`` endpoints
are deleted: the debugging phase is closed, and only the formats list proves that
each of the four solutions works on this image.

For debugging, the ``sample_stream`` harness inside ``solutions.base`` is the
right place to drop a candidate and watch what it does; there is no HTTP-exposed
debug endpoint, because the user asked to close that phase.
"""

from __future__ import annotations

import os
import shutil
import time
import asyncio
from collections import deque
from concurrent.futures import ThreadPoolExecutor, as_completed

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse

from core.logging import log, track
from core.naming import name_from_url
from core.pipeline import (
    FFMPEG,
    FFMPEG_FIRST_BYTE,
    Relay,
    iter_body,
    kill_proc,
    pump_stderr,
    start_ffmpeg,
    tail,
)
from solutions.base import (
    FIRST_BYTE,
    open_stream,
    sample_stream,
    versions,
)
from solutions import SOLUTIONS
from cdn.routes import router as cdn_router

app = FastAPI(
    title="Kick VOD Downloader",
    version="4.0",
    description=(
        "Two ways to get at a Kick VOD.\n\n"
        "GET /cdn/formats resolves a VOD into its renditions and real sizes, "
        "and hands back Kick's CDN coordinates so a *browser* downloads the "
        "video. No video bytes pass through this service, which is what makes "
        "it cheap and infinitely scalable. This is the recommended path.\n\n"
        "GET /stream remuxes and proxies the video through this service. It "
        "works, but it is CPU-bound: a 1080p60 VOD remuxes at single-digit KB/s "
        "on this hardware. Kept for compatibility; prefer /cdn/formats.\n\n"
        "GET /formats reports which solutions can *see* a url. Diagnostic."
    ),
    openapi_tags=[
        {"name": "client-side", "description": "Resolution only. The browser fetches the video from Kick's CDN."},
        {"name": "proxy", "description": "Video is remuxed and streamed through this service. CPU-bound."},
        {"name": "diagnostics", "description": "Liveness and what this image actually contains."},
    ],
)
app.include_router(cdn_router)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["Content-Disposition", "X-Upstream"],
)


def require_kick(url: str) -> None:
    if "kick.com" not in (url or "").lower():
        raise HTTPException(
            400,
            {"error": "URL must point to kick.com (https://kick.com/.../videos/...)"},
        )


def select_solutions(name: str | None) -> list:
    """All solutions, or exactly the one named by ``?solution=``.

    This is the per-step switch: with no name every rung runs, with a name only
    that rung runs, so one solution can be watched on its own without the others
    hiding its result. Spelling is forgiving - ``yt-dlp``, ``ytdlp`` and
    ``yt_dlp`` all name the same rung - and anything unknown is a 400 that lists
    the valid names instead of silently falling back to all of them.
    """
    if not name:
        return list(SOLUTIONS)
    wanted = name.strip().lower()
    for solution in SOLUTIONS:
        canonical = solution.NAME.lower()
        if wanted in {canonical, canonical.replace("-", ""), canonical.replace("-", "_")}:
            return [solution]
    raise HTTPException(
        400,
        {
            "error": f"unknown solution {name!r}",
            "solutions": [s.NAME for s in SOLUTIONS],
        },
    )


@app.get("/", tags=["diagnostics"], summary="Service index")
def index():
    return {
        "service": "kick-vod-downloader",
        "recommended": "/cdn/formats?url=<kick-vod-url>",
        "endpoints": [
            "/cdn/formats?url=<kick-vod-url>",
            "/formats?url=<kick-vod-url>",
            "/formats?url=<kick-vod-url>&solution=<name>",
            "/formats?report=1&url=<kick-vod-url>&format_id=best",
            "/formats?report=1&probe=1&url=<kick-vod-url>&format_id=best",
            "/stream?url=<kick-vod-url>&format_id=best",
            "/stream?url=<kick-vod-url>&format_id=best&solution=<name>",
            "/health",
        ],
        "solutions": [s.NAME for s in SOLUTIONS],
        "note": "add solution=<name> to run one rung on its own; /cdn/formats returns real sizes, /formats does not",
    }


def browser_bin() -> str | None:
    """Where the browser in this image is, or ``None`` when there is not one.

    The image ships lightpanda (the ``lightpanda`` wheel bundles the binary);
    this asks the same helper the rungs use, so ``/health`` cannot disagree
    with what would actually run. Never raises: a broken package must not take
    the health surface down with it.
    """
    try:
        from solutions.kickapi import browser

        return browser()
    except Exception:  # noqa: BLE001 - reporting only
        return shutil.which("lightpanda") or None


@app.get("/health", tags=["diagnostics"], summary="What this image actually contains")
def health():
    try:
        version_info = versions()
    except Exception as exc:  # noqa: BLE001
        version_info = {"error": f"{type(exc).__name__}: {exc}"[:120]}
    return {
        "ok": True,
        "tools": {
            "ffmpeg": shutil.which(FFMPEG) or None,
            "streamlink": shutil.which("streamlink") or None,
            "curl_cffi": _curl_cffi_version(),
            "lightpanda": shutil.which("lightpanda") or None,
        },
        "browser": {
            "engine": "lightpanda",
            "path": browser_bin(),
            "binary": "https://github.com/lightpanda-io/browser",
            "installed_by": "pip (the lightpanda wheel bundles the browser binary)",
            "needed_by": ["streamlink"],
        },
        "versions": version_info,
        "cookies": os.path.isfile("cookies.txt") or os.path.isfile("/data/cookies.txt"),
        "solutions": [s.NAME for s in SOLUTIONS],
        "endpoints": {
            "client_side": ["/cdn/formats"],
            "proxy": ["/stream"],
            "diagnostic": ["/formats", "/health"],
        },
        "removed_rungs": {
            "yt-dlp": "no working Kick VOD extractor; resolves as [kick:live] and answers 'channel is not currently live'",
            "streamlink": "rung needed a browser that can clear Kick's JS challenge; this image ships Lightpanda, a different engine",
            "any-dl": "needs a real Chrome on its command line; this image deliberately ships no Chromium",
            "note": "each was removed after being measured to fail on every real url while kick-api returned all five renditions",
        },
    }


def _curl_cffi_version() -> str | None:
    try:
        import curl_cffi

        return getattr(curl_cffi, "__version__", "unknown")
    except Exception:  # noqa: BLE001 - reporting only
        return None


@app.get("/formats", tags=["diagnostics"], summary="Which solutions can see this url (no sizes)")
async def formats(
    url: str = Query(
        ...,
        description="kick.com VOD url (e.g. https://kick.com/<channel>/videos/<slug>)",
    ),
    report: int | bool = False,
    format_id: str = Query("best", description="an id from /formats, used only with report+probe"),
    probe: bool = False,
    solution: str | None = Query(
        None,
        description="run only this solution (kick-api)",
    ),
):
    require_kick(url)
    chosen = select_solutions(solution)

    if report:
        return await report_formats(url, format_id, probe, chosen)

    started = time.monotonic()
    try:
        pairs = await _list_all(url, chosen)
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        track("formats", url=url, status="error", detail=str(exc)[:200])
        raise HTTPException(502, str(exc)[:600]) from exc

    variant_rows = []
    for p in pairs:
        for v in p.variants:
            variant_rows.append(
                {
                    "solution": p.solution.NAME,
                    "tool": p.solution.tool(),
                    "format_id": v.quality_name(),
                    "name": v.name,
                }
                | v.public()
            )
    # With ?solution= only one rung runs, so "ok" has to follow that rung's
    # outcome: claiming ok with an empty row list hid a failing solution.
    ok = any(p.entry.get("ok") for p in pairs)
    track(
        "formats",
        url=url,
        status="ok" if ok else "no_solution",
        detail=f"{len(variant_rows)} rows from {[s.NAME for s in chosen]} "
        f"in {int((time.monotonic() - started) * 1000)}ms",
    )
    return {
        "url": url,
        "ok": ok,
        "rows": variant_rows,
        "solutions": [s.NAME for s in chosen],
        "failed": [
            {"solution": p.solution.NAME, "detail": p.entry.get("detail")}
            for p in pairs
            if not p.entry.get("ok")
        ],
    }


@app.get("/stream", tags=["proxy"], summary="Remux and stream the video through this service (CPU-bound)")
async def stream(
    url: str = Query(
        ...,
        description="kick.com VOD url (e.g. https://kick.com/<channel>/videos/<slug>)",
    ),
    format_id: str = Query("best", description="an id from /formats"),
    solution: str | None = Query(
        None,
        description="use only this solution (kick-api)",
    ),
):
    require_kick(url)
    chosen = select_solutions(solution)
    started = time.monotonic()

    try:
        upstream = open_stream(url, format_id, chosen)
    except Exception as exc:  # noqa: BLE001 - StreamOpenError or any solution failure
        track("stream", url=url, status="extract_failed", detail=str(exc)[:200])
        raise HTTPException(
            502,
            {
                "error": "no solution produced data",
                "format_id": format_id,
                "solutions": [s.NAME for s in chosen],
                "tried": str(exc)[:900],
            },
        ) from exc

    ff = start_ffmpeg(upstream)
    ff_sink: deque = deque()
    pump_stderr(ff, ff_sink)
    ff_relay = Relay(ff)

    if not ff_relay.wait_first(FFMPEG_FIRST_BYTE):
        kill_proc(ff)
        kill_proc(upstream.proc)
        ff_relay.close()
        upstream.relay.close()
        detail = {
            "error": "remux produced no output",
            "upstream": upstream.name,
            "upstream_stderr": tail(upstream.sink),
            "ffmpeg_stderr": tail(ff_sink),
        }
        track("stream", url=url, status="remux_failed", detail=str(detail)[:200])
        raise HTTPException(502, detail)

    filename = f"{name_from_url(url)}.mp4"
    log.info(
        "stream ok upstream=%s format=%s in %dms",
        upstream.name,
        format_id,
        int((time.monotonic() - started) * 1000),
    )
    track("stream", url=url, status="ok", detail=upstream.name)
    return StreamingResponse(
        iter_body(upstream, ff, ff_relay, ff_sink),
        media_type="video/mp4",
        headers={
            "Content-Disposition": f'attachment; filename="{filename}"',
            "X-Upstream": upstream.name,
        },
    )


# --------------------------------------------------------------------------- #
# per-solution report helpers
# --------------------------------------------------------------------------- #


class ListPair:
    """Bundle one solution's listing evidence with the solution itself."""

    def __init__(self, solution, entry: dict, variants: list):
        self.solution = solution
        self.entry = entry
        self.variants = variants


async def _list_all(url: str, solutions: list) -> list[ListPair]:
    """Schedule each requested solution's listing concurrently, in ladder order."""
    with ThreadPoolExecutor(max_workers=len(solutions), thread_name_prefix="list") as pool:
        future_by_solution = {pool.submit(safe_list, s, url): s for s in solutions}
        results: list[ListPair] = []
        for future in as_completed(future_by_solution):
            solution = future_by_solution[future]
            try:
                entry, variants = future.result(timeout=300)
            except Exception as exc:  # noqa: BLE001
                entry, variants = _failure_entry(solution, exc), ()
            results.append(ListPair(solution, entry, variants))

    by_name = {p.solution.NAME: p for p in results}
    return [by_name[s.NAME] for s in solutions]


def _failure_entry(solution, exc: Exception) -> dict:
    return {
        "solution": solution.NAME,
        "tool": solution.tool(),
        "ok": False,
        "ms": 0,
        "detail": f"{type(exc).__name__}: {exc}"[:400],
        "variants": [],
    }


async def report_formats(url: str, format_id: str, probe: bool, solutions: list) -> dict:
    """Run each requested solution against this url and return evidence.

    When ``probe=1`` each solution also pulls a handful of bytes through its
    own pipe, so the report answers the question that matters: can this
    solution actually stream video here?
    """
    started = time.monotonic()
    pairs = await _list_all(url, solutions)

    # merged public variants - only from solutions whose listing succeeded
    by_quality: dict[str, dict] = {}
    for p in pairs:
        if not p.entry.get("ok"):
            continue
        for v in p.variants:
            key = v.quality_name()
            row = by_quality.get(key)
            if row is None:
                by_quality[key] = v.public() | {"solutions": []}
                row = by_quality[key]
            if p.solution.NAME not in row["solutions"]:
                row["solutions"].append(p.solution.NAME)
            if v.fps and not row.get("fps"):
                row["fps"] = v.fps
            if v.size_mb and not row.get("size_mb"):
                row["size_mb"] = v.size_mb

    public_rows = _build_public_rows(by_quality)

    # per-solution probe rows
    probe_rows: list[tuple[str, dict]] | None = None
    probe_by_name: dict[str, dict] | None = None
    if probe:
        probe_rows = []
        probe_by_name = {}
        for p in pairs:
            try:
                row = sample_stream(p.solution, url, format_id)
            except Exception as exc:  # noqa: BLE001
                row = {
                    "solution": p.solution.NAME,
                    "ok": False,
                    "bytes": 0,
                    "ms": 0,
                    "detail": f"probe failed: {exc}"[:300],
                    "first_bytes": None,
                    "stderr": "",
                    "format_id": format_id,
                }
            probe_rows.append((p.solution.NAME, row))
            probe_by_name[p.solution.NAME] = row
        probe_rows.sort(key=lambda pair: [s.NAME for s in solutions].index(pair[0]))

    body = {
        "url": url,
        "ok": any(p.entry.get("ok") for p in pairs),
        "elapsed_ms": int((time.monotonic() - started) * 1000),
        "solutions": [s.NAME for s in solutions],
        "formats": [
            _format_report_entry(p, probe_by_name) if probe else _format_report_entry(p, None)
            for p in pairs
        ],
        "public_variants": public_rows,
        "probe": [p[1] for p in (probe_rows or [])] if probe else [],
    }
    track(
        "formats:report",
        url=url,
        status="ok",
        detail=f"elapsed {body['elapsed_ms']}ms; "
        f"{sum(1 for p in pairs if p.entry['ok'])}/{len(solutions)} listing solutions ok",
    )
    return body


def _build_public_rows(by_quality: dict[str, dict]) -> list[dict]:
    """Consolidate every successful listing into a sorted, deduplicated list."""
    rows = sorted(
        by_quality.values(),
        key=lambda r: r.get("height") or 0,
        reverse=True,
    )
    for row in rows:
        row["id"] = quality_id(row.get("height"), row.get("fps"), row.get("quality", ""))
    return rows


def quality_id(height: int | None, fps: int | None, fallback: str = "") -> str:
    """Stable public id for a row: ``1080p60``, ``720p``, ``best``, or fallback."""
    if height and fps and fps > 30:
        return f"{height}p{fps}"
    if height:
        return f"{height}p"
    return fallback or ""


def _format_report_entry(p: ListPair, probe_rows: dict[str, dict] | None) -> dict:
    """One entry per solution, with the probe for that solution if requested."""
    entry = {
        "solution": p.solution.NAME,
        "tool": p.solution.tool(),
        "ok": p.entry.get("ok"),
        "ms": p.entry.get("ms"),
        "detail": p.entry.get("detail"),
        "variants": p.entry.get("variants", []),
    }
    if probe_rows and p.solution.NAME in probe_rows:
        entry["probe"] = probe_rows[p.solution.NAME]
    return entry


def safe_list(solution, url: str):
    """Thin wrapper so _list_all can call solutions.base.safe_list without a circular import."""
    import solutions.base as _base

    return _base.safe_list(solution, url)