pullpush.py

"""PullPush API client. Thin, polite, paginating.

PullPush is a free third-party service that resurrected Pushshift's Reddit
archive. Endpoints are public and CORS-open. Be polite: their rate limits
are undocumented and they're one person's project.
"""
from __future__ import annotations

import logging
import time
from typing import Iterator, Optional

import requests

PULLPUSH_BASE = "https://api.pullpush.io/reddit"
DEFAULT_TIMEOUT = 30
USER_AGENT = "pakka-corpus/0.1 (research)"

log = logging.getLogger(__name__)


class PullPushError(Exception):
    pass


class PullPushClient:
    def __init__(
        self,
        delay_seconds: float = 1.2,
        max_retries: int = 3,
        backoff_base: int = 2,
    ):
        self.delay = delay_seconds
        self.max_retries = max_retries
        self.backoff_base = backoff_base
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": USER_AGENT})

    def _get(self, path: str, params: dict) -> dict:
        url = f"{PULLPUSH_BASE}{path}"
        last_err = None
        for attempt in range(self.max_retries):
            try:
                r = self.session.get(url, params=params, timeout=DEFAULT_TIMEOUT)
                if r.status_code == 200:
                    try:
                        return r.json()
                    except ValueError as e:
                        raise PullPushError(f"Bad JSON: {e}") from e
                if r.status_code == 429 or r.status_code >= 500:
                    # 429 needs a long wait; 5xx gets standard exponential backoff
                    wait = 60 if r.status_code == 429 else self.backoff_base ** attempt
                    log.warning(
                        "PullPush %s on %s; backoff %ss (attempt %d)",
                        r.status_code, path, wait, attempt + 1,
                    )
                    time.sleep(wait)
                    last_err = f"HTTP {r.status_code}"
                    continue
                raise PullPushError(f"HTTP {r.status_code}: {r.text[:200]}")
            except requests.RequestException as e:
                wait = self.backoff_base ** attempt
                log.warning("PullPush request error on %s: %s; backoff %ss", path, e, wait)
                time.sleep(wait)
                last_err = str(e)
        raise PullPushError(f"Exhausted retries on {path}: {last_err}")

    def search_comments(
        self,
        subreddit: Optional[str] = None,
        q: Optional[str] = None,
        link_id: Optional[str] = None,
        author: Optional[str] = None,
        after: Optional[int] = None,
        before: Optional[int] = None,
        size: int = 100,
    ) -> list[dict]:
        params: dict = {"size": size, "sort": "desc", "sort_type": "created_utc"}
        if subreddit: params["subreddit"] = subreddit
        if q: params["q"] = q
        if link_id: params["link_id"] = link_id
        if author: params["author"] = author
        if after: params["after"] = after
        if before: params["before"] = before
        data = self._get("/search/comment/", params)
        return data.get("data", []) or []

    def search_submissions(
        self,
        subreddit: Optional[str] = None,
        q: Optional[str] = None,
        author: Optional[str] = None,
        after: Optional[int] = None,
        before: Optional[int] = None,
        size: int = 100,
    ) -> list[dict]:
        params: dict = {"size": size, "sort": "desc", "sort_type": "created_utc"}
        if subreddit: params["subreddit"] = subreddit
        if q: params["q"] = q
        if author: params["author"] = author
        if after: params["after"] = after
        if before: params["before"] = before
        data = self._get("/search/submission/", params)
        return data.get("data", []) or []

    def paginate_submissions(
        self,
        subreddit: str,
        since_ts: Optional[int] = None,
        max_items: int = 5000,
    ) -> Iterator[dict]:
        before: Optional[int] = None
        yielded = 0
        while yielded < max_items:
            batch = self.search_submissions(subreddit=subreddit, before=before, size=100)
            if not batch:
                return
            for item in batch:
                created = int(item.get("created_utc", 0))
                if since_ts is not None and created < since_ts:
                    return
                yield item
                yielded += 1
                if yielded >= max_items:
                    return
            before = int(batch[-1]["created_utc"])
            time.sleep(self.delay)

    def paginate_comments_for_post(
        self,
        post_id: str,
        max_items: int = 20000,
    ) -> Iterator[dict]:
        before: Optional[int] = None
        yielded = 0
        while yielded < max_items:
            batch = self.search_comments(link_id=post_id, before=before, size=100)
            if not batch:
                return
            for item in batch:
                yield item
                yielded += 1
                if yielded >= max_items:
                    return
            before = int(batch[-1]["created_utc"])
            time.sleep(self.delay)

    def paginate_comments_for_subreddit(
        self,
        subreddit: str,
        since_ts: Optional[int] = None,
        max_items: int = 10000,
    ) -> Iterator[dict]:
        before: Optional[int] = None
        yielded = 0
        while yielded < max_items:
            batch = self.search_comments(subreddit=subreddit, before=before, size=100)
            if not batch:
                return
            for item in batch:
                created = int(item.get("created_utc", 0))
                if since_ts is not None and created < since_ts:
                    return
                yield item
                yielded += 1
                if yielded >= max_items:
                    return
            before = int(batch[-1]["created_utc"])
            time.sleep(self.delay)