saverapi_client.py

"""
SaverAPI video fetcher.

Dispatches to the right SaverAPI endpoint based on URL:
  - YouTube / youtu.be / youtube Shorts  ->  /api/youtube-api
  - Everything else (Instagram, TikTok, X, ...)  ->  /api/all-in-one-downloader-api

Stdlib only (urllib). Runs anywhere Python 3.7+ is installed, no pip needed.

Response shapes handled:
  all-in-one / IG+TT  : {"error": false, "download_url": "https://...", ...}
  all-in-one / Vimeo  : {"medias": [{"url": "https://...", "quality": "..."}, ...], ...}
  all-in-one / others : {"url": "https://...", "source": "...", ...}  (less common)
  all-in-one URL-err  : {"error": true,  "message": "..."}                    (URL-level, do NOT mark key dead)
  all-in-one / key    : HTTP 402 + {"message": "Insufficient credits"}        (KEY-level, mark dead)
  youtube success     : {"status": "tunnel", "url": "https://...", ...}
  youtube error       : (unconfirmed — passed through as best-effort)
  Cloudflare block    : HTTP 403, body contains "error code: 1010"             (key/network, mark dead)
"""
import json
import os
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request

# ── Endpoints ──────────────────────────────────────────────────────────────
ALL_IN_ONE_URL = "https://saverapi.net/api/all-in-one-downloader-api"
YOUTUBE_URL    = "https://saverapi.net/api/youtube-api"

# ── Cloudflare 1010 bypass: send a real-browser User-Agent ────────────────
BROWSER_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept":          "application/json, text/plain, */*",
    "Accept-Language": "en-US,en;q=0.9",
    "Origin":          "https://saverapi.net",
    "Referer":         "https://saverapi.net/",
}

# ── Tuning ─────────────────────────────────────────────────────────────────
LOOKUP_TIMEOUT_SEC   = 15
DOWNLOAD_TIMEOUT_SEC = 60
DEFAULT_MAX_FILESIZE_MB = 100
YOUTUBE_DEFAULT_QUALITY  = "720"   # SaverAPI's param is misspelled "farmat"


# ── Platform detection ────────────────────────────────────────────────────
def _is_youtube(url: str) -> bool:
    u = url.lower()
    return ("youtube.com" in u) or ("youtu.be" in u) or ("youtube-nocookie.com" in u)


# ── Step 1: ask SaverAPI for a direct video URL ──────────────────────────
def _saverapi_lookup(url: str, api_key: str) -> tuple[bool, str, str]:
    """Returns (ok, direct_video_url, error_message)."""
    if _is_youtube(url):
        endpoint = YOUTUBE_URL
        params   = {"url": url, "farmat": YOUTUBE_DEFAULT_QUALITY}
    else:
        endpoint = ALL_IN_ONE_URL
        params   = {"url": url}

    full_url = f"{endpoint}?{urllib.parse.urlencode(params)}"
    req = urllib.request.Request(
        full_url,
        headers={**BROWSER_HEADERS, "x-api-key": api_key, "Content-Type": "application/json"},
        method="GET",
    )

    try:
        with urllib.request.urlopen(req, timeout=LOOKUP_TIMEOUT_SEC) as r:
            status = r.status
            body   = r.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")
        if e.code == 403 and "error code" in body:
            return False, "", "key-level:cloudflare"
        if e.code == 402 and "insufficient" in body.lower():
            return False, "", "key-level:insufficient-credits"
        return False, "", f"key-level:http-{e.code}: {body[:200]}"
    except Exception as e:
        return False, "", f"key-level:network: {type(e).__name__}: {e}"

    if status != 200:
        return False, "", f"key-level:http-{status}: {body[:200]}"

    try:
        result = json.loads(body)
    except json.JSONDecodeError:
        return False, "", f"Non-JSON response: {body[:200]}"

    if not isinstance(result, dict):
        return False, "", f"Unexpected response (not an object): {str(result)[:200]}"

    # YouTube success: {"status": "tunnel", "url": "https://..."}
    if result.get("status") == "tunnel" and result.get("url"):
        return True, result["url"], ""

    # Vimeo-style success: {"medias": [{"url": "...", "quality": "..."}, ...], ...}
    medias = result.get("medias")
    if isinstance(medias, list) and medias:
        # SaverAPI lists medias best-first; pick the first with a URL.
        for m in medias:
            if isinstance(m, dict) and m.get("url"):
                return True, m["url"], ""
        return False, "", "medias[] present but no url in any item"

    # all-in-one success: {"error": false, "download_url": "https://..."}
    if result.get("error") is False:
        direct = result.get("download_url")
        if direct:
            return True, direct, ""
        return False, "", "all-in-one: error=false but no download_url"

    # all-in-one URL-level error: {"error": true, "message": "..."}
    if result.get("error") is True:
        return False, "", f"SaverAPI: {result.get('message', 'unspecified error')}"

    # YouTube non-tunnel status (unconfirmed error shape — pass through)
    if "status" in result:
        return False, "", f"YouTube API status={result.get('status')!r}"

    # Generic fallback: some platforms return a top-level "url" + "source"
    if result.get("url") and result.get("source"):
        return True, result["url"], ""

    return False, "", f"Unexpected shape: {str(result)[:200]}"


# ── Step 2: stream the direct URL to a tempfile with a size cap ──────────
def _stream_to_file(direct_url: str, max_bytes: int) -> tuple[bool, str]:
    # Strip saverapi.net-specific headers (Referer/Origin) — those are for the
    # SaverAPI lookup, not for the underlying CDN (twimg.com, fbcdn.net, etc.).
    # Keep User-Agent + Accept so the CDN sees a real browser.
    headers = {
        "User-Agent": BROWSER_HEADERS["User-Agent"],
        "Accept":     "*/*",
    }
    req = urllib.request.Request(direct_url, headers=headers, method="GET")

    tmp = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
    tmp.close()
    try:
        with urllib.request.urlopen(req, timeout=DOWNLOAD_TIMEOUT_SEC) as r:
            # Peek first 1 KB to catch HTML error pages
            first_chunk = r.read(1024)
            if first_chunk.lstrip().lower().startswith((b"<!doctype", b"<html")):
                return False, "download_url returned HTML, not a video"

            bytes_written = 0
            with open(tmp.name, "wb") as f:
                if first_chunk:
                    f.write(first_chunk)
                    bytes_written = len(first_chunk)
                if bytes_written > max_bytes:
                    return False, f"Exceeds {max_bytes // (1024 * 1024)} MB cap"
                while True:
                    chunk = r.read(64 * 1024)
                    if not chunk:
                        break
                    bytes_written += len(chunk)
                    if bytes_written > max_bytes:
                        return False, f"Exceeds {max_bytes // (1024 * 1024)} MB cap"
                    f.write(chunk)
        if bytes_written == 0:
            return False, "Empty response"
        return True, tmp.name
    except urllib.error.HTTPError as e:
        try:
            os.remove(tmp.name)
        except OSError:
            pass
        return False, f"Download HTTP {e.code}"
    except Exception as e:
        try:
            os.remove(tmp.name)
        except OSError:
            pass
        return False, f"Download error: {type(e).__name__}: {e}"


# ── Public entry point ────────────────────────────────────────────────────
def saverapi_fetch(
    url: str,
    api_key: str,
    max_filesize_mb: int = DEFAULT_MAX_FILESIZE_MB,
) -> tuple[bool, str]:
    """
    Fetch a video from a supported platform via SaverAPI.

    Args:
        url:             Public video URL (Instagram, TikTok, YouTube, ...)
        api_key:         SaverAPI key
        max_filesize_mb: Hard cap on downloaded video size (default 100)

    Returns:
        (True,  file_path)   on success — caller must `os.remove(path)` when done
        (False, err_message) on any failure
    """
    if not url.lower().startswith(("http://", "https://")):
        return False, "Unsupported URL scheme (use http:// or https://)"

    ok, direct, err = _saverapi_lookup(url, api_key)
    if not ok:
        return False, err

    return _stream_to_file(direct, max_filesize_mb * 1024 * 1024)


# ── Quick self-test ───────────────────────────────────────────────────────
if __name__ == "__main__":
    import time
    api_key = os.environ.get("SAVER_API_KEY", "").strip()
    if not api_key:
        print("Set SAVER_API_KEY env var first.", file=sys.stderr)
        sys.exit(1)

    tests = [
        "https://www.instagram.com/reel/DZTCpI2SQsq",
        "https://www.tiktok.com/@scout2015/video/6718335390845095173",
        "https://www.facebook.com/reel/763412386794044/",
        "https://x.com/traderpodcaste/status/2063797483629629691",
        "https://vimeo.com/76979871",
        "https://www.youtube.com/shorts/LXb3EKWsInQ",
    ]
    for u in tests:
        print(f"\n→ {u}")
        t0 = time.time()
        ok, result = saverapi_fetch(u, api_key)
        dt = time.time() - t0
        if ok:
            size = os.path.getsize(result) // 1024
            print(f"  ✅  {size} KB in {dt:.1f}s  →  {result}")
            os.remove(result)
        else:
            print(f"  ❌  {result}")