reddit_api.py

"""Reddit public JSON endpoints. No auth needed.

Used for two things:
  1) Subreddit discovery via /subreddits/search.json
  2) Live thread fetches (post metadata + delete-state cross-check)

Reddit's anonymous JSON has aggressive rate limits — keep delay >= 2s.
"""
from __future__ import annotations

import logging
import re
import time
from typing import Optional

import requests

REDDIT_BASE = "https://www.reddit.com"
DEFAULT_TIMEOUT = 30
USER_AGENT = "pakka-corpus/0.1 (research)"

log = logging.getLogger(__name__)


class RedditError(Exception):
    pass


class RedditClient:
    def __init__(self, delay_seconds: float = 2.0):
        self.delay = delay_seconds
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": USER_AGENT})

    def _get(self, path: str, params: Optional[dict] = None) -> dict:
        url = f"{REDDIT_BASE}{path}"
        r = self.session.get(url, params=params or {}, timeout=DEFAULT_TIMEOUT)
        if r.status_code == 429:
            raise RedditError("429 from Reddit; back off and retry later")
        if r.status_code != 200:
            raise RedditError(f"HTTP {r.status_code} on {path}: {r.text[:200]}")
        try:
            return r.json()
        except ValueError as e:
            raise RedditError(f"Bad JSON from {path}: {e}") from e

    def search_subreddits(self, query: str, limit: int = 25) -> list[dict]:
        data = self._get("/subreddits/search.json", {"q": query, "limit": limit})
        out = []
        for c in data.get("data", {}).get("children", []):
            d = c.get("data", {})
            out.append({
                "name": d.get("display_name", ""),
                "subscribers": int(d.get("subscribers") or 0),
                "description": (d.get("public_description") or "").strip(),
                "lang": d.get("lang", "") or "",
                "over_18": int(bool(d.get("over18", False))),
            })
        time.sleep(self.delay)
        return out

    def fetch_posts_by_ids(self, ids: list[str]) -> list[dict]:
        """Fetch up to 100 posts by ID via /api/info.json. Prepends t3_ prefix if absent."""
        fullnames = ",".join(
            f"t3_{pid}" if not pid.startswith("t3_") else pid for pid in ids
        )
        data = self._get("/api/info.json", {"id": fullnames})
        children = data.get("data", {}).get("children", [])
        time.sleep(self.delay)
        return [c.get("data", {}) for c in children if c.get("data")]

    def get_thread(self, post_id: str) -> Optional[list[dict]]:
        try:
            data = self._get(f"/comments/{post_id}.json")
            time.sleep(self.delay)
            return data
        except RedditError as e:
            log.warning("Reddit get_thread(%s) failed: %s", post_id, e)
            return None

    def paginate_keyword_search(
        self,
        q: str,
        max_items: int = 1000,
        sort: str = "new",
    ) -> Iterator[dict]:
        """Search Reddit for posts matching *q* across all subreddits.

        Yields raw post dicts in the same shape upsert_post expects.
        Pagination uses Reddit's listing `after` cursor (fullname).
        """
        from typing import Iterator  # local to avoid circular at module level
        after: Optional[str] = None
        yielded = 0
        seen: set[str] = set()

        while yielded < max_items:
            params: dict = {
                "q": q,
                "type": "link",
                "limit": 100,
                "sort": sort,
                "include_over_18": "on",
            }
            if after:
                params["after"] = after

            try:
                data = self._get("/search.json", params)
            except RedditError as e:
                log.warning("Keyword search for %r failed: %s", q, e)
                raise  # propagate so _keyword_worker stores it as a run error

            listing = data.get("data", {})
            children = listing.get("children", [])
            if not children:
                return

            for child in children:
                d = child.get("data", {})
                pid = d.get("id", "")
                if not pid or pid in seen:
                    continue
                seen.add(pid)
                yield d
                yielded += 1
                if yielded >= max_items:
                    return

            after = listing.get("after")
            if not after:
                return
            time.sleep(self.delay)


REDDIT_URL_RE = re.compile(
    r"reddit\.com/r/(?P<sub>[^/]+)/comments/(?P<post_id>[a-z0-9]+)",
    re.IGNORECASE,
)


def parse_thread_url(url: str) -> Optional[tuple[str, str]]:
    m = REDDIT_URL_RE.search(url)
    if not m:
        return None
    return m.group("sub"), m.group("post_id").lower()