220 lines
8.1 KiB
Python
220 lines
8.1 KiB
Python
"""Minimal GoFile API client - one guest account per run.
|
|
|
|
The API interaction mirrors Lysagxra/GoFileDownloader (same headers, website
|
|
token scheme and endpoints), but with two deliberate differences:
|
|
|
|
1. A single guest account is created per run (GoFileDownloader creates one per
|
|
URL, which trips Gofile's rate limit after ~6 requests).
|
|
2. Failures raise exceptions instead of sys.exit(), so the orchestrator can
|
|
track them per file/URL and keep going.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
API = "https://api.gofile.io"
|
|
USER_AGENT = "Mozilla/5.0"
|
|
LOCALE = "en-US"
|
|
TOKEN_SECRET = "12af056dacea0b" # public website-token seed, see GoFileDownloader # noqa: S105
|
|
TOKEN_WINDOW_SEC = 14400
|
|
CHUNK_SIZE = 1024 * 1024
|
|
|
|
COMMON_HEADERS = {
|
|
"User-Agent": USER_AGENT,
|
|
"Accept-Encoding": "gzip",
|
|
"Accept": "*/*",
|
|
"Connection": "keep-alive",
|
|
}
|
|
EXTENDED_HEADERS = {
|
|
**COMMON_HEADERS,
|
|
"Sec-Fetch-Dest": "empty",
|
|
"Sec-Fetch-Mode": "cors",
|
|
"Sec-Fetch-Site": "same-site",
|
|
"Pragma": "no-cache",
|
|
"Cache-Control": "no-cache",
|
|
}
|
|
|
|
|
|
class GoFileError(Exception):
|
|
"""Base error for GoFile operations."""
|
|
|
|
|
|
class GoFilePasswordRequired(GoFileError):
|
|
"""The content is password protected and none (or a wrong one) was given."""
|
|
|
|
|
|
class GoFileAuthError(GoFileError):
|
|
"""The account token was rejected; the client can re-auth once."""
|
|
|
|
|
|
def content_id_from_url(url: str) -> str:
|
|
"""Extract the content id from a gofile.io/d/<id> URL."""
|
|
parts = [p for p in urlparse(url).path.split("/") if p]
|
|
if len(parts) >= 2 and parts[-2] == "d":
|
|
return parts[-1]
|
|
raise GoFileError(f"not a gofile content URL (expected gofile.io/d/<id>): {url}")
|
|
|
|
|
|
class GoFileClient:
|
|
"""GoFile API client that reuses one guest account token for the whole run."""
|
|
|
|
def __init__(self, password: str | None = None):
|
|
self._password_hash = (
|
|
hashlib.sha256(password.encode()).hexdigest() if password else None
|
|
)
|
|
self._token: str | None = None
|
|
self.session = requests.Session()
|
|
|
|
# -- authentication -----------------------------------------------------
|
|
|
|
def ensure_token(self, max_attempts: int = 8, wait_seconds: int = 20) -> str:
|
|
"""Create a guest account; backs off and retries on rate limiting."""
|
|
if self._token:
|
|
return self._token
|
|
|
|
last_status = "unknown"
|
|
for attempt in range(1, max_attempts + 1):
|
|
try:
|
|
resp = self.session.post(
|
|
f"{API}/accounts", headers=COMMON_HEADERS, timeout=15
|
|
)
|
|
data = resp.json()
|
|
except (ValueError, requests.RequestException) as exc:
|
|
last_status = f"request error: {exc}"
|
|
log.warning(
|
|
"GoFile account creation attempt %d/%d failed: %s",
|
|
attempt, max_attempts, last_status,
|
|
)
|
|
else:
|
|
if isinstance(data, dict) and data.get("status") == "ok":
|
|
self._token = data["data"]["token"]
|
|
log.info("GoFile guest account created (token reused for whole run)")
|
|
return self._token
|
|
last_status = str(data.get("status", data)[:200] if data else data)
|
|
log.warning(
|
|
"GoFile account creation attempt %d/%d failed: %s",
|
|
attempt, max_attempts, last_status,
|
|
)
|
|
if attempt < max_attempts:
|
|
time.sleep(wait_seconds * attempt)
|
|
raise GoFileError(
|
|
f"GoFile account creation failed after {max_attempts} attempts "
|
|
f"(last: {last_status})"
|
|
)
|
|
|
|
def reset(self) -> None:
|
|
"""Drop the current token so the next ensure_token() creates a new one."""
|
|
self._token = None
|
|
|
|
def _website_token(self) -> str:
|
|
window = int(time.time()) // TOKEN_WINDOW_SEC
|
|
seed = f"{USER_AGENT}::{LOCALE}::{self._token}::{window}::{TOKEN_SECRET}"
|
|
return hashlib.sha256(seed.encode()).hexdigest()
|
|
|
|
def _auth_headers(self) -> dict:
|
|
return {
|
|
**EXTENDED_HEADERS,
|
|
"Authorization": f"Bearer {self._token}",
|
|
"X-Website-Token": self._website_token(),
|
|
"X-BL": LOCALE,
|
|
}
|
|
|
|
# -- listing ------------------------------------------------------------
|
|
|
|
def list_files(self, url: str) -> list[tuple[str, str, str]]:
|
|
"""Return (relpath, filename, download_link) for every file under a URL."""
|
|
cid = content_id_from_url(url)
|
|
results: list[tuple[str, str, str]] = []
|
|
self._walk(cid, [], results)
|
|
# Direct files at the content root get their own folder named after the id
|
|
normalized = [
|
|
(relpath if relpath else cid, name, link) for relpath, name, link in results
|
|
]
|
|
return normalized
|
|
|
|
def _walk(self, cid: str, prefix: list[str], results: list[tuple[str, str, str]]) -> None:
|
|
query = "?cache=true&sortField=createTime&sortDirection=1"
|
|
if self._password_hash:
|
|
query += f"&password={self._password_hash}"
|
|
try:
|
|
resp = self.session.get(
|
|
f"{API}/contents/{cid}{query}", headers=self._auth_headers(), timeout=30
|
|
)
|
|
except requests.RequestException as exc:
|
|
raise GoFileError(f"content request for {cid} failed: {exc}") from exc
|
|
|
|
if resp.status_code in (401, 403):
|
|
raise GoFileAuthError(f"content request denied ({resp.status_code}) for {cid}")
|
|
try:
|
|
data = resp.json()
|
|
except ValueError as exc:
|
|
raise GoFileError(f"non-JSON response for {cid}: {resp.text[:200]}") from exc
|
|
|
|
if data.get("status") != "ok":
|
|
raise GoFileError(f"content request failed for {cid}: {data.get('status')}")
|
|
|
|
d = data["data"]
|
|
if d.get("passwordStatus") not in (None, "passwordOk"):
|
|
raise GoFilePasswordRequired(
|
|
f"content {cid} is password protected (use --password)"
|
|
)
|
|
|
|
if d.get("type") == "folder":
|
|
here = prefix + [d["name"]]
|
|
for child in (d.get("children") or {}).values():
|
|
if child.get("type") == "folder":
|
|
self._walk(child["id"], here, results)
|
|
else:
|
|
results.append(("/".join(here), child["name"], child["link"]))
|
|
else:
|
|
results.append(("/".join(prefix), d["name"], d["link"]))
|
|
|
|
# -- downloading ----------------------------------------------------------
|
|
|
|
def download_file(self, link: str, dest: Path) -> int:
|
|
"""Download one file; returns the file size. Skips existing files."""
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if dest.exists() and dest.stat().st_size > 0:
|
|
log.info("Skip existing: %s", dest)
|
|
return dest.stat().st_size
|
|
|
|
tmp = dest.with_name(dest.name + ".part")
|
|
headers = {
|
|
**EXTENDED_HEADERS,
|
|
"Cookie": f"accountToken={self._token}",
|
|
"Referer": link,
|
|
"Origin": link,
|
|
}
|
|
try:
|
|
with self.session.get(
|
|
link, headers=headers, stream=True, timeout=(10, 60), allow_redirects=True
|
|
) as resp:
|
|
if resp.status_code != 200:
|
|
raise GoFileError(f"HTTP {resp.status_code} while downloading {dest.name}")
|
|
total = int(resp.headers.get("Content-Length") or 0)
|
|
written = 0
|
|
with tmp.open("wb") as fh:
|
|
for chunk in resp.iter_content(chunk_size=CHUNK_SIZE):
|
|
fh.write(chunk)
|
|
written += len(chunk)
|
|
if written == 0:
|
|
raise GoFileError(f"empty body for {dest.name}")
|
|
os.replace(tmp, dest)
|
|
return written
|
|
finally:
|
|
if tmp.exists() and not dest.exists():
|
|
try:
|
|
tmp.unlink()
|
|
except OSError:
|
|
pass
|