"""RedGifs, oficjalne publiczne API v2. Główne źródło klipów do feedu Swipe. Dodane 2026-09-14. **Dlaczego RedGifs, a nie nsfwswipe.** Sprawdzone tego samego dnia: nsfwswipe to lustro wideo natywnych Reddita (v.redd.it) bez audio, z ostatnim klipem z 19.08 (stoi od czterech tygodni) i z losowym, nieoficjalnym API oddającym HTML. RedGifs daje JSON, klipy z tej minuty, 1080x1920, audio w ok. 1/3 klipów, twórcę z odznaką weryfikacji, lajki i widoki. Pliki `media.redgifs.com/.mp4` są publiczne, bez podpisu, grają bez Referera z VPS-a i z telefonu (206), więc do bazy wchodzi gotowy adres, bez ekstraktora. **Token.** `GET /v2/auth/temporary` oddaje token na kilka godzin, związany z IP i User-Agentem wołającego: kolejne zapytania MUSZĄ iść z tymi samymi nagłówkami, inaczej 401 `WrongSender` (zjadło pół godziny przy pierwszym podejściu, bo curl i node miały inne UA). Na 401 odnawiamy token raz i powtarzamy. **Listing.** `GET /v2/gifs/search?order=latest&count=100&page=N` bez `search_text` to globalne „najnowsze" (98-100 na stronę). `order` ∈ latest|top|top7|top28|trending|score, NIE `new` (400 BadOrder). Każde zapytanie widzi maks. 10 000 wyników (`pages` ≤ 100, strona 101 → 500), więc pełny katalog chodzi się po niszach (`/v2/niches`, 1812 społeczności, `/v2/niches//gifs`) i po twórcach (`/v2/users//search`). Brak nagłówków rate-limit; trzymamy ~3 zapytania/s i ustępujemy przy 429. **Rekord gifa** (pola, które bierzemy): `id`, `type` (1 = wideo, 2 = obraz, pomijamy), `urls{hd,sd,poster,thumbnail}`, `duration`, `width/height`, `hasAudio`, `hls`, `tags[]`, `niches[]`, `contentType`, `sexuality[]`, `likes`, `views`, `createDate` (unix s), `userName`, `verified`, `cta`/`promoted` (reklama, informacyjnie). """ from __future__ import annotations import logging import re import time from dataclasses import dataclass, field from datetime import UTC, datetime import httpx log = logging.getLogger(__name__) _BASE = "https://api.redgifs.com/v2" #: Stałe UA i Origin: token jest do nich przywiązany (patrz docstring). _HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/152.0.0.0 Safari/537.36" ), "Accept": "application/json", "Origin": "https://www.redgifs.com", "Referer": "https://www.redgifs.com/", } _PAGE_MAX = 100 _MIN_INTERVAL_S = 0.35 _TAG_SLUG_RE = re.compile(r"[^a-z0-9]+") @dataclass class RawClip: source: str source_id: str title: str | None creator: str | None creator_verified: bool duration_sec: float | None width: int | None height: int | None has_audio: bool content_type: str | None niches: list[str] = field(default_factory=list) tags: list[str] = field(default_factory=list) hd_url: str | None = None sd_url: str | None = None poster_url: str | None = None thumb_url: str | None = None hls: bool = False likes: int = 0 views: int = 0 published_at: datetime | None = None def tag_slug(name: str) -> str: """"Big Tits" → "big-tits". Ta sama reguła dla RedGifs i (później) Pornhuba, żeby afiniczność liczyła się na wspólnych slugach.""" return _TAG_SLUG_RE.sub("-", (name or "").strip().lower()).strip("-")[:64] def map_gif(g: dict) -> RawClip | None: """Rekord API → RawClip. None dla obrazów, rekordów bez pliku i bez id.""" if not isinstance(g, dict) or g.get("type") not in (1, None): return None gid = str(g.get("id") or "").strip() urls = g.get("urls") or {} hd, sd = urls.get("hd"), urls.get("sd") if not gid or not (hd or sd): return None created = g.get("createDate") published = ( datetime.fromtimestamp(int(created), tz=UTC) if isinstance(created, (int, float)) and created > 0 else None ) tags: list[str] = [] seen: set[str] = set() for t in g.get("tags") or []: s = tag_slug(str(t)) if s and s not in seen: seen.add(s) tags.append(s) niches = [str(n)[:64] for n in (g.get("niches") or []) if n] dur = g.get("duration") return RawClip( source="redgifs", source_id=gid, title=(g.get("description") or None), creator=(g.get("userName") or None), creator_verified=bool(g.get("verified")), duration_sec=float(dur) if isinstance(dur, (int, float)) and dur > 0 else None, width=g.get("width") if isinstance(g.get("width"), int) else None, height=g.get("height") if isinstance(g.get("height"), int) else None, has_audio=bool(g.get("hasAudio")), content_type=(str(g["contentType"])[:32] if g.get("contentType") else None), niches=niches, tags=tags, hd_url=hd, sd_url=sd, poster_url=urls.get("poster"), thumb_url=urls.get("thumbnail"), hls=bool(g.get("hls")), likes=int(g.get("likes") or 0), views=int(g.get("views") or 0), published_at=published, ) class RedgifsClient: """Cienki klient z tokenem tymczasowym i odnowieniem na 401.""" def __init__(self, *, timeout: float = 30.0) -> None: self._http = httpx.Client(headers=_HEADERS, timeout=timeout) self._token: str | None = None self._last_call = 0.0 def close(self) -> None: self._http.close() def _auth(self) -> str: r = self._http.get(f"{_BASE}/auth/temporary") r.raise_for_status() tok = (r.json() or {}).get("token") if not tok: raise RuntimeError("redgifs: brak tokenu w /auth/temporary") self._token = tok return tok def _get(self, path: str, params: dict | None = None) -> dict: wait = _MIN_INTERVAL_S - (time.monotonic() - self._last_call) if wait > 0: time.sleep(wait) tok = self._token or self._auth() for attempt in range(3): r = self._http.get(f"{_BASE}{path}", params=params, headers={"Authorization": f"Bearer {tok}"}) self._last_call = time.monotonic() if r.status_code == 401 and attempt == 0: tok = self._auth() continue if r.status_code == 429: time.sleep(5 * (attempt + 1)) continue r.raise_for_status() return r.json() raise RuntimeError(f"redgifs: {path} nie odpowiedzial poprawnie po 3 probach") # ---- listingi --------------------------------------------------------------------- def latest(self, page: int = 1, count: int = _PAGE_MAX) -> list[dict]: """Globalne „najnowsze" (bez search_text).""" return self._get("/gifs/search", {"order": "latest", "count": count, "page": page}).get("gifs") or [] def search(self, text: str, *, order: str = "latest", page: int = 1, count: int = _PAGE_MAX) -> list[dict]: return self._get( "/gifs/search", {"search_text": text, "order": order, "count": count, "page": page} ).get("gifs") or [] def niches(self, page: int = 1, count: int = _PAGE_MAX) -> dict: """`{"page","pages","total","niches":[{"id","name","gifs","subscribers","tags":[...]}]}`""" return self._get("/niches", {"count": count, "page": page}) def niche_gifs(self, niche_id: str, *, order: str = "latest", page: int = 1, count: int = _PAGE_MAX) -> list[dict]: return self._get(f"/niches/{niche_id}/gifs", {"order": order, "count": count, "page": page}).get("gifs") or [] def user_gifs(self, user: str, *, order: str = "latest", page: int = 1, count: int = _PAGE_MAX) -> list[dict]: return self._get(f"/users/{user}/search", {"order": order, "count": count, "page": page}).get("gifs") or [] def gif(self, gif_id: str) -> dict | None: return (self._get(f"/gifs/{gif_id}") or {}).get("gif")