pipeline/downloader.py

"""Download a YouTube video via the ``youtube-video-fast-downloader-24-7`` RapidAPI.

The API returns a link on the provider's OWN server (e.g. ``s5-audio.12388101.xyz``), not
a ``googlevideo.com`` URL — so the Space downloads the file directly, bypassing both the
egress DPI and the datacenter-IP block, with no IP-lock. The link is prepared
asynchronously: it 404s for ~20-300s while the provider fetches it, then is live for about
10 minutes. We poll until it's ready, then stream it to disk.

Requires the ``RAPIDAPI_KEY`` Space secret. Quality ``18`` is 360p muxed (video+audio) —
small, and its audio track is enough for faster-whisper.
"""
from __future__ import annotations

import os
import time

import requests

RAPIDAPI_HOST = "youtube-video-fast-downloader-24-7.p.rapidapi.com"
DEFAULT_QUALITY = "18"  # itag 18 = 360p muxed (has audio, ~10-30 MB for typical videos)

# RapidAPI tracks monthly usage against the plan limit and returns it in every response
# header. We mirror the latest values here and enforce the cap — no persistence needed,
# and it's Space-wide + monthly (the plan's billing cycle) by construction.
_quota = {"limit": None, "remaining": None, "reset": None}  # reset = seconds until reset


class DownloadError(RuntimeError):
    """Raised when the video can't be obtained from the download API."""


def _update_quota(headers) -> None:
    def _int(name):
        try:
            return int(headers.get(name))
        except (TypeError, ValueError):
            return None
    for key, hdr in (("limit", "X-RateLimit-Requests-Limit"),
                     ("remaining", "X-RateLimit-Requests-Remaining"),
                     ("reset", "X-RateLimit-Requests-Reset")):
        val = _int(hdr)
        if val is not None:
            _quota[key] = val


def quota_status() -> dict:
    """Latest known monthly quota: ``{limit, remaining, reset(sec), used, self_cap}``."""
    q = dict(_quota)
    q["used"] = (q["limit"] - q["remaining"]) if (q["limit"] and q["remaining"] is not None) else None
    q["self_cap"] = _self_cap()
    return q


def _reset_days() -> str:
    r = _quota.get("reset")
    return f"~{max(1, r // 86400)} day(s)" if r else "the next cycle"


def _self_cap() -> int | None:
    """Optional lower monthly cap (env ``MONTHLY_DOWNLOAD_LIMIT``); None = use plan limit."""
    v = os.environ.get("MONTHLY_DOWNLOAD_LIMIT", "").strip()
    try:
        return int(v) if v else None
    except ValueError:
        return None


def _enforce_quota() -> None:
    """Refuse before spending a request if the monthly cap is already reached."""
    q = _quota
    if q["remaining"] is None:
        return  # unknown yet (fresh start) — let the call itself surface a 429
    if q["limit"] is not None:
        used = q["limit"] - q["remaining"]
        cap = _self_cap()
        if cap is not None and used >= cap:
            raise DownloadError(
                f"Self-imposed monthly cap reached: {used} of {cap} used. Resets in {_reset_days()}.")
    if q["remaining"] <= 0:
        raise DownloadError(
            f"Monthly request limit reached (plan: {q['limit']}). Resets in {_reset_days()}.")


def _headers() -> dict:
    key = os.environ.get("RAPIDAPI_KEY", "").strip()
    if not key:
        raise DownloadError("RAPIDAPI_KEY is not set (required for the video download API).")
    return {"X-RapidAPI-Key": key, "X-RapidAPI-Host": RAPIDAPI_HOST}


def _request_urls(video_id: str, quality: str, timeout: int = 60) -> tuple[list[str], dict]:
    """Ask the API for a download link; return candidate URLs (primary + reserved)."""
    url = f"https://{RAPIDAPI_HOST}/download_video/{video_id}"
    try:
        r = requests.get(url, params={"quality": quality}, headers=_headers(), timeout=timeout)
    except requests.RequestException as exc:
        raise DownloadError(f"download API unreachable: {exc}") from exc
    _update_quota(r.headers)
    if r.status_code == 429:
        raise DownloadError(f"Monthly request limit reached (plan: {_quota.get('limit')}). "
                            f"Resets in {_reset_days()}.")
    if r.status_code in (401, 403):
        raise DownloadError(f"download API auth failed (HTTP {r.status_code}); "
                            "check RAPIDAPI_KEY / that you're subscribed.")
    if r.status_code != 200:
        raise DownloadError(f"download API HTTP {r.status_code}: {r.text[:160]}")
    data = r.json()
    urls = [u for u in (data.get("file"), data.get("reserved_file")) if u]
    if not urls:
        raise DownloadError(f"no download URL in API response: {str(data)[:200]}")
    return urls, data


def download_video(video_id: str, dest: str, quality: str = DEFAULT_QUALITY,
                   poll_timeout: int = 330, chunk: int = 1 << 20,
                   progress=None) -> str:
    """Download ``video_id`` to ``dest`` and return the path.

    Polls the (async) provider link until ready (404 -> wait), then streams it to ``dest``.
    Raises DownloadError if it never becomes ready within ``poll_timeout`` seconds, or if
    the monthly request cap is already reached.
    """
    _enforce_quota()  # refuse before spending a request if the monthly cap is reached
    urls, _ = _request_urls(video_id, quality)
    deadline = time.time() + poll_timeout
    last = "not ready"
    while time.time() < deadline:
        for url in urls:
            try:
                with requests.get(url, stream=True, timeout=90) as resp:
                    if resp.status_code == 404:
                        last = "404 (server still preparing)"
                        continue
                    resp.raise_for_status()
                    with open(dest, "wb") as fh:
                        for c in resp.iter_content(chunk):
                            if c:
                                fh.write(c)
                if os.path.getsize(dest) > 0:
                    return dest
            except requests.RequestException as exc:
                last = f"{type(exc).__name__}: {exc}"
        if progress:
            progress(0.0, desc="Preparing video (server-side, up to ~5 min)…")
        time.sleep(8)
    raise DownloadError(f"video not ready after {poll_timeout}s ({last}).")