scraper.py

"""
scraper.py — Tenor.com HTML scraper (post Tenor-API-shutdown workaround)

Tenor's public API shut down on 2026-06-30. Tenor.com itself is still
online and still works for humans, so this module scrapes the rendered
site instead of calling the (now-dead) api.tenor.com endpoints.

Tenor's frontend is a React/Next.js app. On first page load it embeds a
JSON payload in the HTML (typically inside a <script id="__NEXT_DATA__">
tag, or a similar `window.__PRELOADED_STATE__`-style blob) that contains
the actual GIF objects used to hydrate the page. We try, in order:

  1. Parse embedded JSON state blobs (__NEXT_DATA__ / __PRELOADED_STATE__ /
     any <script type="application/json"> blob that looks GIF-shaped).
  2. Regex-scan the raw HTML for tenor.com/media / c.tenor.com URLs
     (.gif / .mp4 / .webp) as a structural fallback if the JSON blob
     approach fails or Tenor changes their bundler/markup.

Because this depends on Tenor's page markup, IT WILL BREAK if Tenor
changes their frontend. Both strategies are kept independent so a
change to one doesn't necessarily kill the other. If Tenor moves to a
fully client-side-only render with no SSR JSON, only the regex fallback
will keep working, and if they also start lazy-loading images via
IntersectionObserver with no URLs in the initial HTML at all, this
approach stops working entirely and would need a headless browser
(Playwright/Selenium) — not included by default here to keep the HF
Space lightweight, but see `render_with_browser()` stub at the bottom.
"""

import re
import json
import html
import logging
import random
from urllib.parse import quote

import requests

logger = logging.getLogger("tenor_scraper")

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 "
    "(KHTML, like Gecko) Version/17.4 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/124.0.0.0 Safari/537.36",
]

BASE_URL = "https://tenor.com"

# Matches direct tenor media CDN links for gif/webp/mp4 assets.
MEDIA_URL_RE = re.compile(
    r'https?://(?:media|c)\.tenor\.com/[A-Za-z0-9_\-./]+\.(?:gif|mp4|webp)',
    re.IGNORECASE,
)

# Matches /view/<slug>-gif-<id> style permalinks, used to dedupe / get ids.
VIEW_URL_RE = re.compile(r'/view/[a-z0-9\-]+-gif-(\d+)', re.IGNORECASE)

NEXT_DATA_RE = re.compile(
    r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', re.DOTALL
)

GENERIC_JSON_SCRIPT_RE = re.compile(
    r'<script type="application/json"[^>]*>(.*?)</script>', re.DOTALL
)


class TenorScraperError(Exception):
    pass


def _headers():
    return {
        "User-Agent": random.choice(USER_AGENTS),
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9",
    }


def _fetch(url, timeout=15):
    resp = requests.get(url, headers=_headers(), timeout=timeout)
    resp.raise_for_status()
    return resp.text


def _search_url(query, pos_page=1):
    slug = quote(query.strip().lower().replace(" ", "-"))
    if not slug.endswith("-gifs"):
        slug = f"{slug}-gifs"
    url = f"{BASE_URL}/search/{slug}"
    if pos_page > 1:
        url += f"?page={pos_page}"
    return url


def _walk_json_for_gif_objects(obj, found):
    """
    Recursively walk a parsed JSON structure looking for dict entries
    that look like Tenor GIF result objects, i.e. anything with a
    media/gif URL and an id/title. Tenor's exact schema may vary by
    build, so this is intentionally loose/duck-typed rather than
    matching one fixed schema.
    """
    if isinstance(obj, dict):
        # Heuristic: object has some url-ish field pointing at tenor media
        url_fields = {}
        for key in ("url", "gif_url", "mp4", "webp", "src"):
            val = obj.get(key)
            if isinstance(val, str) and MEDIA_URL_RE.search(val):
                url_fields[key] = val
        # Nested media dict pattern: {"media": {"gif": {"url": ...}}}
        media = obj.get("media") or obj.get("media_formats") or obj.get("formats")
        if isinstance(media, dict):
            for fmt_name, fmt_val in media.items():
                if isinstance(fmt_val, dict):
                    u = fmt_val.get("url")
                    if isinstance(u, str) and MEDIA_URL_RE.search(u):
                        url_fields[fmt_name] = u
                elif isinstance(fmt_val, str) and MEDIA_URL_RE.search(fmt_val):
                    url_fields[fmt_name] = fmt_val

        if url_fields:
            gif_id = obj.get("id") or obj.get("gif_id")
            title = obj.get("title") or obj.get("h1_title") or obj.get("name") or ""
            best_gif = url_fields.get("gif") or url_fields.get("url") or next(iter(url_fields.values()))
            best_mp4 = url_fields.get("mp4")
            found.append({
                "id": str(gif_id) if gif_id else None,
                "title": html.unescape(str(title)),
                "gif_url": best_gif,
                "mp4_url": best_mp4,
                "raw_fields": list(url_fields.keys()),
            })
            # Don't recurse into the "media"/"formats" dict we already
            # consumed above -- walking into it produces spurious
            # duplicate/partial "results" for each format entry (e.g. a
            # standalone {"id": None, "gif_url": "...mp4"} from the mp4
            # sub-dict). Recurse into everything else on this object instead.
            skip_keys = {"media", "media_formats", "formats"}
            for k, v in obj.items():
                if k in skip_keys:
                    continue
                _walk_json_for_gif_objects(v, found)
        else:
            for v in obj.values():
                _walk_json_for_gif_objects(v, found)

    elif isinstance(obj, list):
        for item in obj:
            _walk_json_for_gif_objects(item, found)


def _extract_via_json_blob(html_text):
    results = []

    m = NEXT_DATA_RE.search(html_text)
    candidates = []
    if m:
        candidates.append(m.group(1))
    candidates.extend(GENERIC_JSON_SCRIPT_RE.findall(html_text))

    for blob in candidates:
        try:
            data = json.loads(blob)
        except (json.JSONDecodeError, ValueError):
            continue
        found = []
        _walk_json_for_gif_objects(data, found)
        results.extend(found)

    # de-dupe by gif_url
    seen = set()
    deduped = []
    for r in results:
        if r["gif_url"] in seen:
            continue
        seen.add(r["gif_url"])
        deduped.append(r)
    return deduped


def _extract_via_regex(html_text):
    """
    Fallback: just pull every distinct tenor media URL out of the raw
    HTML. No titles/ids available this way (unless we can pair with a
    nearby /view/ URL), so this produces weaker metadata than the JSON
    path but is much more resistant to markup/schema changes.
    """
    urls = MEDIA_URL_RE.findall(html_text)
    # prefer .gif entries as the canonical url, keep others as alt formats
    seen = set()
    results = []
    for u in urls:
        # normalize AA/AAAAd sizing suffixes are part of the path, keep as-is
        if u in seen:
            continue
        seen.add(u)
        results.append({
            "id": None,
            "title": "",
            "gif_url": u if u.lower().endswith((".gif", ".webp")) else None,
            "mp4_url": u if u.lower().endswith(".mp4") else None,
            "raw_fields": ["regex_fallback"],
        })

    # merge entries that only differ by extension but share the same
    # base path (common on tenor's CDN: name.gif vs name.mp4)
    merged = {}
    for r in results:
        base = None
        for ext in (".gif", ".mp4", ".webp"):
            src = r["gif_url"] or r["mp4_url"]
            if src and src.lower().endswith(ext):
                base = src.rsplit(".", 1)[0]
                break
        if base is None:
            continue
        entry = merged.setdefault(base, {"id": None, "title": "", "gif_url": None, "mp4_url": None, "raw_fields": ["regex_fallback"]})
        if r["gif_url"]:
            entry["gif_url"] = r["gif_url"]
        if r["mp4_url"]:
            entry["mp4_url"] = r["mp4_url"]

    return [v for v in merged.values() if v["gif_url"] or v["mp4_url"]]


def scrape_search(query, limit=20, page=1):
    """
    Scrape Tenor search results for `query`. Returns a list of dicts:
      { id, title, gif_url, mp4_url, raw_fields }
    Tries the JSON-blob strategy first, falls back to regex scraping.
    Raises TenorScraperError if both strategies find nothing (which
    usually means Tenor changed their markup or is blocking the
    request — check status codes / consider adding a browser fallback).
    """
    url = _search_url(query, pos_page=page)
    try:
        html_text = _fetch(url)
    except requests.RequestException as e:
        raise TenorScraperError(f"Failed to fetch Tenor search page: {e}") from e

    results = _extract_via_json_blob(html_text)
    strategy = "json_blob"

    if not results:
        results = _extract_via_regex(html_text)
        strategy = "regex_fallback"

    if not results:
        raise TenorScraperError(
            "No GIF results extracted — Tenor's page markup may have changed, "
            "or the request was blocked/rate-limited. Try again or inspect "
            "the raw HTML."
        )

    for r in results:
        r["source_strategy"] = strategy

    return results[:limit]


def scrape_random(query, limit=1):
    """Convenience wrapper: scrape then shuffle-pick `limit` results."""
    results = scrape_search(query, limit=50)
    random.shuffle(results)
    return results[:limit]


def render_with_browser(query):  # pragma: no cover
    """
    Optional heavy fallback stub. If Tenor ever moves to a render path
    where NO gif URLs appear anywhere in the initial HTML (fully
    client-fetched after JS execution with no SSR at all), the
    strategies above will stop finding anything and you'd need a real
    headless browser to load the page, wait for network idle, and read
    the DOM/network requests for media URLs.

    Not implemented by default because:
      - Playwright/Selenium + a real browser binary is heavy for a
        free-tier HF Space (slow cold starts, larger image, more RAM
        pressure under concurrent requests).
      - It's meaningfully slower per-request than the HTML strategies.

    If you need it: `pip install playwright && playwright install chromium`,
    launch headless, page.goto(search_url, wait_until="networkidle"),
    then either page.content() into _extract_via_json_blob/_extract_via_regex,
    or intercept network responses matching MEDIA_URL_RE directly.
    """
    raise NotImplementedError(
        "Browser rendering fallback not wired up. See docstring for how to "
        "add Playwright if the HTML-based strategies stop working."
    )