playground/rg_download.py

#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.12"
# dependencies = ["httpx"]
# ///
"""
RapidGator (rapidgator.net) downloader — reverse-engineered from JDownloader's RapidGatorNet plugin.

Usage:
  ./rg_download.py <url> [--user EMAIL] [--pass PASSWORD] [--out DIR] [--2captcha KEY]

Premium accounts use the clean v2 API path (no captcha, full speed).
Free/anonymous downloads scrape the website; reCAPTCHA v2 or Cloudflare Turnstile is
required — a 2captcha API key is strongly recommended (manual solving isn't feasible for
these captcha types).
"""

import re
import sys
import os
import time
import argparse
import getpass
from pathlib import Path
from urllib.parse import urlparse, unquote

import httpx

# ── constants ────────────────────────────────────────────────────────────────

API_BASE = "https://rapidgator.net/api/v2/"
SITE_BASE = "https://rapidgator.net"

SUPPORTED_DOMAINS = {"rapidgator.net", "rapidgator.asia", "rg.to"}

# JD plugin regex: (?i)/file/([a-z0-9]{32}|\d+)
FILE_ID_RE = re.compile(r"(?i)/file/([a-z0-9]{32}|\d+)")

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Accept-Language": "en-US,en;q=0.8",
}


# ── helpers ──────────────────────────────────────────────────────────────────

def extract_file_id(url: str) -> str | None:
    m = FILE_ID_RE.search(url)
    return m.group(1) if m else None


def check_domain(url: str):
    host = urlparse(url).hostname or ""
    host = host.removeprefix("www.")
    if host not in SUPPORTED_DOMAINS:
        sys.exit(f"Unsupported domain '{host}'. Supported: {', '.join(sorted(SUPPORTED_DOMAINS))}")


def _api_get(session: httpx.Client, endpoint: str, params: dict) -> dict:
    """GET request to the v2 API; raises on HTTP error, returns parsed JSON."""
    r = session.get(f"{API_BASE}{endpoint}", params=params, headers=HEADERS, timeout=30)
    try:
        data = r.json()
    except Exception:
        r.raise_for_status()
        raise
    return data


def _api_check(data: dict, context: str = "") -> dict:
    """Raise a RuntimeError if the API returned an error status."""
    status = data.get("status", "")
    if status != "success":
        err = data.get("details") or data.get("error") or data
        raise RuntimeError(f"API error{' (' + context + ')' if context else ''}: {err}")
    return data.get("details", {})


# ── API calls (mirroring JD's loginAPI / requestFileInformationAPI / handlePremium_api) ──

def api_login(session: httpx.Client, email: str, password: str) -> str:
    """GET /user/login → returns session_id (used as ?token= in all other calls)."""
    data = _api_get(session, "user/login", {"login": email, "password": password})
    details = _api_check(data, "login")
    # JD stores the field named "session_id" (PROPERTY_sessionid constant)
    sid = details.get("session_id") or details.get("token")
    if not sid:
        raise RuntimeError(f"Login succeeded but no session_id in response: {details}")
    return sid


def api_user_info(session: httpx.Client, token: str) -> dict:
    """GET /user/info?token= → account details (is_premium, premium_end_time, …)."""
    data = _api_get(session, "user/info", {"token": token})
    return _api_check(data, "user/info")


def api_file_info(session: httpx.Client, token: str, file_id: str) -> dict:
    """GET /file/info?token=&file_id= → file metadata (name, size, hash, …)."""
    data = _api_get(session, "file/info", {"token": token, "file_id": file_id})
    details = _api_check(data, "file/info")
    # Nested under "file" key
    return details.get("file") or details


def api_file_download(session: httpx.Client, token: str, file_id: str) -> str:
    """GET /file/download?token=&file_id= → direct download_url."""
    data = _api_get(session, "file/download", {"token": token, "file_id": file_id})
    details = _api_check(data, "file/download")
    url = details.get("download_url")
    if not url:
        raise RuntimeError(f"No download_url in response: {details}")
    return url


# ── 2captcha ─────────────────────────────────────────────────────────────────

class TwoCaptcha:
    """
    Thin wrapper around the 2captcha API v2.
    Supports reCAPTCHA v2 (RecaptchaV2TaskProxyless) and
    Cloudflare Turnstile (TurnstileTaskProxyless) — the two captcha types
    used by RapidGator's free download page.
    """

    API = "https://api.2captcha.com"
    POLL_INTERVAL = 5
    POLL_TIMEOUT = 180

    def __init__(self, api_key: str, client: httpx.Client):
        self.api_key = api_key
        self.client = client

    def _call(self, endpoint: str, payload: dict) -> dict:
        r = self.client.post(
            f"{self.API}{endpoint}",
            json={"clientKey": self.api_key, **payload},
            headers={"Content-Type": "application/json"},
            timeout=30,
        )
        r.raise_for_status()
        data = r.json()
        err_id = data.get("errorId") or 0
        if err_id and err_id != 0:
            raise RuntimeError(
                f"2captcha error: {data.get('errorCode') or data.get('errorDescription') or data}"
            )
        return data

    def _poll(self, task_id: int) -> str:
        """Poll until ready, return solution token."""
        print(f"  2captcha task {task_id} submitted, polling…", flush=True)
        deadline = time.monotonic() + self.POLL_TIMEOUT
        while time.monotonic() < deadline:
            time.sleep(self.POLL_INTERVAL)
            result = self._call("/getTaskResult", {"taskId": task_id})
            if result.get("status") == "ready":
                token = result["solution"]["token"]
                print(f"  2captcha solved (token length: {len(token)})")
                return token
            print("  …waiting for solution")
        raise RuntimeError(f"2captcha timed out after {self.POLL_TIMEOUT}s")

    def solve_recaptcha_v2(self, site_key: str, page_url: str) -> str:
        """Submit a RecaptchaV2TaskProxyless and return the g-recaptcha-response token."""
        task_id = self._call("/createTask", {
            "task": {
                "type": "RecaptchaV2TaskProxyless",
                "websiteURL": page_url,
                "websiteKey": site_key,
            }
        })["taskId"]
        return self._poll(task_id)

    def solve_turnstile(self, site_key: str, page_url: str) -> str:
        """Submit a TurnstileTaskProxyless and return the cf-turnstile-response token."""
        task_id = self._call("/createTask", {
            "task": {
                "type": "TurnstileTaskProxyless",
                "websiteURL": page_url,
                "websiteKey": site_key,
            }
        })["taskId"]
        return self._poll(task_id)


# ── free download (website scrape) ────────────────────────────────────────────

_RECAPTCHA_RE = re.compile(
    r'class=["\']g-recaptcha["\'][^>]*data-sitekey=["\']([^"\']+)["\']'
    r'|data-sitekey=["\']([^"\']+)["\'][^>]*class=["\']g-recaptcha["\']'
)
_TURNSTILE_RE = re.compile(
    r'class=["\']cf-turnstile["\'][^>]*data-sitekey=["\']([^"\']+)["\']'
    r'|data-sitekey=["\']([^"\']+)["\'][^>]*class=["\']cf-turnstile["\']'
)
_DL_URL_RE = re.compile(
    r"'(https?://[A-Za-z0-9\-_]+\.[^/]+//\?r=download/index&session_id=[A-Za-z0-9]+)'"
    r"|\"(https?://[^/]+/download/[^<>\"']+)\""
)

# X-Requested-With is required by the AJAX endpoints — they return empty without it
_XHR_HEADERS = {"X-Requested-With": "XMLHttpRequest"}


def _first_group(m: re.Match | None) -> str | None:
    if not m:
        return None
    return next((g for g in m.groups() if g), None)


def free_download(
    session: httpx.Client,
    file_id: str,
    file_name: str,
    solver: TwoCaptcha | None,
    out_dir: Path,
):
    """
    Website-based free download flow (mirrors JD handleDownloadWebsite / free path).

    Actual page flow discovered by inspection:
      1. GET /file/<hex_id>  → server sets PHPSESSID + file_id cookies;
                               page JS has: var fid = <numeric>; var secs = 180;
                               var startTimerUrl = '/download/AjaxStartTimer';
      2. GET /download/AjaxStartTimer?fid=<numeric>  (XHR)
                             → {"state":"started", "sid":"<token>"}
      3. Wait `secs` seconds (free-tier countdown)
      4. GET /download/AjaxGetDownloadLink?sid=<token>  (XHR)
                             → poll until {"state":"done"}
      5. GET /download/captcha  → HTML with reCAPTCHA v2 or Turnstile widget
      6. POST /download/captcha with solved token
         → redirect to the actual download URL  OR
           response body contains it
    """
    file_url = f"{SITE_BASE}/file/{file_id}"

    # ── step 1: load file page ────────────────────────────────────────────────
    print(f"Loading file page: {file_url}")
    r = session.get(file_url, headers=HEADERS, timeout=30)
    r.raise_for_status()
    page_html = r.text

    # Check for premium-only direct link already embedded (logged-in premium path)
    pm = re.search(r"var\s+premium_download_link\s*=\s*'(https?://[^']+)'", page_html)
    if pm and pm.group(1):
        stream_download(session, pm.group(1), out_dir, file_name)
        return

    # Extract numeric fid (set by server in page JS and cookie, distinct from hex URL ID)
    fid_m = re.search(r"var\s+fid\s*=\s*(\d+)", page_html)
    if not fid_m:
        sys.exit("Could not find numeric fid in page JS — page layout may have changed.")
    numeric_fid = fid_m.group(1)

    # Extract client-side wait timer (default 180 for free users)
    secs_m = re.search(r"var\s+secs\s*=\s*(\d+)", page_html)
    wait_secs = int(secs_m.group(1)) if secs_m else 180

    print(f"  Numeric fid={numeric_fid}, wait={wait_secs}s")

    if wait_secs > 600:
        sys.exit(f"Server wait is {wait_secs}s — IP appears rate-limited. Try again later.")

    # ── step 2: start the server-side timer ──────────────────────────────────
    print("Starting download timer…")
    r2 = session.get(
        f"{SITE_BASE}/download/AjaxStartTimer",
        params={"fid": numeric_fid},
        headers={**HEADERS, **_XHR_HEADERS, "Referer": file_url},
        timeout=30,
    )
    r2.raise_for_status()
    try:
        timer_data = r2.json()
    except Exception:
        sys.exit(f"AjaxStartTimer returned non-JSON: {r2.text[:300]!r}")

    if timer_data.get("state") == "error":
        sys.exit(f"AjaxStartTimer error: {timer_data.get('code') or timer_data}")
    if timer_data.get("state") != "started":
        sys.exit(f"Unexpected AjaxStartTimer state: {timer_data}")

    sid = timer_data.get("sid")
    if not sid:
        sys.exit(f"AjaxStartTimer response missing sid: {timer_data}")
    print(f"  sid={sid}")

    # ── step 3: wait for the server countdown ────────────────────────────────
    print(f"  Waiting {wait_secs}s…", flush=True)
    for remaining in range(wait_secs, 0, -5):
        time.sleep(min(5, remaining))
        print(f"\r  {remaining - min(5, remaining)}s remaining…  ", end="", flush=True)
    print()

    # ── step 4: confirm server is ready ──────────────────────────────────────
    print("Confirming download ready…")
    for attempt in range(10):
        r3 = session.get(
            f"{SITE_BASE}/download/AjaxGetDownloadLink",
            params={"sid": sid},
            headers={**HEADERS, **_XHR_HEADERS, "Referer": file_url},
            timeout=30,
        )
        r3.raise_for_status()
        try:
            link_data = r3.json()
        except Exception:
            sys.exit(f"AjaxGetDownloadLink returned non-JSON: {r3.text[:300]!r}")

        dl_state = link_data.get("state")
        if dl_state == "done":
            break
        if dl_state == "error":
            code = link_data.get("code", "")
            if "wait" in str(code).lower():
                print(f"  Server not ready yet, retrying in 5s… ({code})")
                time.sleep(5)
                continue
            sys.exit(f"AjaxGetDownloadLink error: {link_data}")
        print(f"  Unexpected state '{dl_state}', retrying…")
        time.sleep(5)
    else:
        sys.exit(f"AjaxGetDownloadLink never returned done. Last response: {link_data}")

    # ── step 5: load captcha page ─────────────────────────────────────────────
    captcha_url = f"{SITE_BASE}/download/captcha"
    print("Loading captcha page…")
    r4 = session.get(captcha_url, headers={**HEADERS, "Referer": file_url}, timeout=30)
    r4.raise_for_status()
    captcha_html = r4.text

    if not captcha_html.strip():
        sys.exit("Captcha page returned empty — session may have expired. Try again.")

    # Detect captcha type and extract site key
    recaptcha_key = _first_group(_RECAPTCHA_RE.search(captcha_html))
    turnstile_key = _first_group(_TURNSTILE_RE.search(captcha_html))

    # Also extract CSRF token from <meta name="__token"> if present
    csrf_m = re.search(r'<meta\s+name=["\']__token["\']\s+content=["\']([^"\']+)["\']', captcha_html)
    csrf_token = csrf_m.group(1) if csrf_m else None

    # Extract any hidden form fields
    hidden_fields: dict[str, str] = {}
    for hm in re.finditer(r'<input[^>]+type=["\']hidden["\'][^>]*name=["\']([^"\']+)["\'][^>]*value=["\']([^"\']*)["\']', captcha_html, re.I):
        hidden_fields[hm.group(1)] = hm.group(2)
    for hm in re.finditer(r'<input[^>]+name=["\']([^"\']+)["\'][^>]+type=["\']hidden["\'][^>]*value=["\']([^"\']*)["\']', captcha_html, re.I):
        hidden_fields[hm.group(1)] = hm.group(2)

    if recaptcha_key:
        captcha_type = "reCAPTCHA v2"
        captcha_field = "g-recaptcha-response"
        site_key = recaptcha_key
    elif turnstile_key:
        captcha_type = "Cloudflare Turnstile"
        captcha_field = "cf-turnstile-response"
        site_key = turnstile_key
    else:
        sys.exit(
            "No reCAPTCHA v2 or Turnstile found on captcha page.\n"
            f"Page excerpt: {captcha_html[:500]}"
        )

    print(f"  Detected {captcha_type} (sitekey: {site_key[:24]}…)")

    if not solver:
        sys.exit(
            f"{captcha_type} detected — a --2captcha key is required for free downloads.\n"
            "Alternatively use a premium account (--user / --pass)."
        )

    if recaptcha_key:
        captcha_token = solver.solve_recaptcha_v2(site_key, file_url)
    else:
        captcha_token = solver.solve_turnstile(site_key, file_url)

    # ── step 6: submit captcha form ───────────────────────────────────────────
    print("Submitting captcha…")
    form_data: dict = {
        **hidden_fields,
        captcha_field: captcha_token,
        "DownloadCaptchaForm[verifyCode]": captcha_token,
    }
    if csrf_token:
        form_data["_csrf"] = csrf_token

    r5 = session.post(
        captcha_url,
        data=form_data,
        headers={**HEADERS, "Referer": captcha_url},
        timeout=30,
    )
    r5.raise_for_status()

    # The final URL may come from a redirect the client followed, or be in the body
    final_url = str(r5.url)
    if "/download/" in final_url or final_url.startswith("http") and "rapidgator" not in final_url:
        stream_download(session, final_url, out_dir, file_name)
        return

    # Search for download URL in response body
    dl_m = _DL_URL_RE.search(r5.text)
    if dl_m:
        stream_download(session, _first_group(dl_m), out_dir, file_name)
        return

    # Last-resort: look for any https link pointing outside rapidgator (CDN URL)
    cdn_m = re.search(r"https?://[a-z0-9\-]+\.rapidgator\.net/[^\s\"'<>]+", r5.text)
    if cdn_m:
        stream_download(session, cdn_m.group(0), out_dir, file_name)
        return

    sys.exit(
        f"Could not extract download URL after captcha submission.\n"
        f"Response URL: {final_url}\n"
        f"Body excerpt: {r5.text[:500]}"
    )


# ── download ──────────────────────────────────────────────────────────────────

def _filename_from_response(r: httpx.Response, url: str, fallback: str) -> str:
    cd = r.headers.get("content-disposition", "")
    if cd:
        m = re.search(r"filename\*=UTF-8''([^;\r\n]+)", cd, re.I)
        if m:
            return unquote(m.group(1).strip().strip('"'))
        m = re.search(r'filename="?([^";\r\n]+)"?', cd, re.I)
        if m:
            return m.group(1).strip()
    path = urlparse(url).path
    name = Path(path).name
    return unquote(name) if name else (fallback or "download")


def stream_download(session: httpx.Client, url: str, dest: Path, fallback_name: str = ""):
    # Enforce HTTPS (JD plugin does this explicitly)
    if url.startswith("http://"):
        url = "https://" + url[7:]

    dest.mkdir(parents=True, exist_ok=True)
    with session.stream("GET", url, headers=HEADERS, timeout=None, follow_redirects=True) as r:
        r.raise_for_status()
        filename = _filename_from_response(r, url, fallback_name)
        out_path = dest / filename
        print(f"Downloading → {out_path}")
        total = int(r.headers.get("content-length", 0))
        done = 0
        with open(out_path, "wb") as f:
            for chunk in r.iter_bytes(chunk_size=65536):
                f.write(chunk)
                done += len(chunk)
                if total:
                    pct = done * 100 // total
                    print(f"\r  {pct}%  {done // 1024} / {total // 1024} KB", end="", flush=True)
    print(f"\nDone: {out_path}")


# ── main flow ─────────────────────────────────────────────────────────────────

def download(
    url: str,
    username: str | None,
    password: str | None,
    out_dir: Path,
    twocaptcha_key: str | None = None,
):
    check_domain(url)

    file_id = extract_file_id(url)
    if not file_id:
        sys.exit(f"Could not extract file ID from: {url}")

    session = httpx.Client(follow_redirects=True)
    token: str | None = None
    solver = TwoCaptcha(twocaptcha_key, session) if twocaptcha_key else None

    # ── 1. authenticate (premium path) ────────────────────────────────────────
    if username and password:
        print(f"Logging in as {username} …")
        try:
            token = api_login(session, username, password)
            print("Logged in OK.")
        except RuntimeError as e:
            msg = str(e).lower()
            if "wrong" in msg or "password" in msg or "login" in msg:
                sys.exit(f"Login failed: {e}")
            raise

    # ── 2. check file ─────────────────────────────────────────────────────────
    if token:
        print(f"Checking file {file_id} via API…")
        try:
            info = api_file_info(session, token, file_id)
        except RuntimeError as e:
            sys.exit(f"File info failed: {e}")

        name = info.get("name") or file_id
        size = info.get("size") or 0
        hash_ = info.get("hash") or info.get("md5") or ""
        print(f"  Name : {name}")
        print(f"  Size : {size // 1024 // 1024} MB" if size else "  Size : unknown")
        if hash_:
            print(f"  MD5  : {hash_}")

        # ── 3. get download URL (premium API) ─────────────────────────────────
        print("Requesting download URL…")
        try:
            direct_url = api_file_download(session, token, file_id)
        except RuntimeError as e:
            msg = str(e).lower()
            if "daily" in msg or "limit" in msg:
                sys.exit(f"Download limit reached: {e}")
            sys.exit(f"Could not get download URL: {e}")

        # ── 4. download ───────────────────────────────────────────────────────
        stream_download(session, direct_url, out_dir, name)

    else:
        # ── free / anonymous path — website scrape + captcha ──────────────────
        print("No credentials provided — attempting free download (captcha required).")
        # Get file name from page if possible (best effort)
        name = file_id
        try:
            r0 = session.get(f"{SITE_BASE}/file/{file_id}", headers=HEADERS, timeout=15)
            title_m = re.search(r"<title>\s*Download file\s*([^<>\"]+)</title>", r0.text, re.I)
            if title_m:
                name = title_m.group(1).strip()
        except Exception:
            pass

        free_download(session, file_id, name, solver, out_dir)


# ── CLI ───────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="RapidGator downloader")
    parser.add_argument("url", help="rapidgator.net / rg.to file URL")
    parser.add_argument("--user", "-u", default=os.environ.get("RG_USER"), help="Account email")
    parser.add_argument("--pass", "-p", dest="password",
                        default=os.environ.get("RG_PASS"), help="Account password")
    parser.add_argument("--out", "-o", default=None, help="Output directory (default: current directory)")
    parser.add_argument("--2captcha", dest="twocaptcha_key",
                        default=os.environ.get("TWOCAPTCHA_API_KEY"),
                        help="2captcha API key (or set TWOCAPTCHA_API_KEY env var)")
    args = parser.parse_args()

    password = args.password
    if args.user and not password:
        password = getpass.getpass(f"Password for {args.user}: ")

    download(
        url=args.url,
        username=args.user,
        password=password,
        out_dir=Path(args.out) if args.out else Path.cwd(),
        twocaptcha_key=args.twocaptcha_key,
    )


if __name__ == "__main__":
    main()