last30days-skill/skills/last30days/scripts/lib/reddit_public.py

"""Reddit public ``.json`` search module (demoted to keyless Tier 0).

Reddit's public ``.json`` endpoints now return HTTP 403 from most contexts
(shreddit anti-bot), so this is no longer the primary free path. The keyless
pipeline (see reddit_keyless.py) still calls ``search`` as a cheap one-shot
Tier 0 attempt — a residential machine may occasionally get a 200 — before
falling through to RSS discovery (reddit_rss.py) and shreddit comment
enrichment (reddit_shreddit.py).

``search_reddit_public`` is retained as a compatibility shim that delegates to
the keyless pipeline, so existing callers (pipeline.py) need no change.

Endpoints (Tier 0):
- Global: https://www.reddit.com/search.json?q={query}&sort=relevance&t=month&limit={limit}
- Subreddit: https://www.reddit.com/r/{sub}/search.json?q={query}&restrict_sr=on&sort=relevance&t=month

Handles 429 rate limits with exponential backoff, HTML anti-bot responses,
network timeouts, and missing subreddits.
"""

import gzip
import json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional


USER_AGENT = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/124.0.0.0 Safari/537.36"
)

# Depth-aware limits for thread counts
DEPTH_LIMITS = {
    "quick": 10,
    "default": 25,
    "deep": 50,
}

MAX_RETRIES = 3
BASE_BACKOFF = 2.0  # seconds


def _log(msg: str):
    """Log to stderr."""
    sys.stderr.write(f"[RedditPublic] {msg}\n")
    sys.stderr.flush()


def _url_encode(text: str) -> str:
    """URL-encode a query string."""
    return urllib.parse.quote_plus(text)


def _fetch_json(url: str, timeout: int = 15) -> Optional[Dict[str, Any]]:
    """Fetch JSON from a URL with retry on 429 and error handling.

    Returns parsed JSON dict, or None on unrecoverable failure.
    """
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/json",
        "Accept-Language": "en-US,en;q=0.9",
        "Accept-Encoding": "gzip, deflate",
        "Connection": "keep-alive",
    }
    req = urllib.request.Request(url, headers=headers)

    for attempt in range(MAX_RETRIES):
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                content_type = resp.headers.get("Content-Type", "")
                if "json" not in content_type and "text/html" in content_type:
                    _log(f"Anti-bot HTML response (Content-Type: {content_type})")
                    return None

                raw = resp.read()
                if resp.headers.get("Content-Encoding", "").lower() == "gzip":
                    raw = gzip.decompress(raw)
                body = raw.decode("utf-8")
                return json.loads(body)

        except urllib.error.HTTPError as e:
            if e.code == 429:
                delay = BASE_BACKOFF * (2 ** attempt)
                retry_after = None
                if hasattr(e, "headers"):
                    retry_after = e.headers.get("Retry-After")
                if retry_after:
                    try:
                        delay = float(retry_after)
                    except ValueError:
                        pass
                _log(f"429 rate limited, retry {attempt + 1}/{MAX_RETRIES} after {delay:.1f}s")
                if attempt < MAX_RETRIES - 1:
                    time.sleep(delay)
                    continue
                # Last attempt exhausted
                _log("429 retries exhausted")
                return None
            elif e.code == 404:
                _log(f"404 not found: {url}")
                return None
            elif e.code == 403:
                _log(f"403 forbidden: {url}")
                return None
            else:
                _log(f"HTTP {e.code}: {e.reason}")
                return None

        except (urllib.error.URLError, OSError, TimeoutError) as e:
            _log(f"Network error: {e}")
            return None

        except json.JSONDecodeError as e:
            _log(f"JSON decode error: {e}")
            return None

    return None


def _parse_posts(data: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Parse Reddit listing JSON into normalized post dicts."""
    if not data:
        return []

    children = data.get("data", {}).get("children", [])
    posts = []

    for child in children:
        if child.get("kind") != "t3":
            continue
        post = child.get("data", {})
        permalink = str(post.get("permalink", "")).strip()
        if not permalink or "/comments/" not in permalink:
            continue

        score = int(post.get("score", 0) or 0)
        num_comments = int(post.get("num_comments", 0) or 0)
        selftext = str(post.get("selftext", ""))
        author = str(post.get("author", "[deleted]"))
        created_utc = post.get("created_utc")

        # Parse date
        date_str = None
        if created_utc:
            try:
                from datetime import datetime, timezone
                dt = datetime.fromtimestamp(float(created_utc), tz=timezone.utc)
                date_str = dt.strftime("%Y-%m-%d")
            except (ValueError, TypeError, OSError):
                pass

        posts.append({
            "id": "",  # Will be assigned after dedup
            "title": str(post.get("title", "")).strip(),
            "url": f"https://www.reddit.com{permalink}",
            "score": score,
            "num_comments": num_comments,
            "subreddit": str(post.get("subreddit", "")).strip(),
            "created_utc": float(created_utc) if created_utc else None,
            "author": author if author not in ("[deleted]", "[removed]") else "[deleted]",
            "selftext": selftext[:500] if selftext else "",
            # Normalized fields matching ScrapeCreators output
            "date": date_str,
            "engagement": {
                "score": score,
                "num_comments": num_comments,
                "upvote_ratio": post.get("upvote_ratio"),
            },
            "relevance": _compute_relevance(score, num_comments),
            "why_relevant": "Reddit public search",
            "metadata": {},
        })

    return posts


def _compute_relevance(score: int, num_comments: int) -> float:
    """Estimate relevance from engagement signals."""
    score_component = min(1.0, max(0.0, score / 500.0))
    comments_component = min(1.0, max(0.0, num_comments / 200.0))
    return round((score_component * 0.6) + (comments_component * 0.4), 3)


def search(
    query: str,
    depth: str = "default",
    subreddit: Optional[str] = None,
    timeout: int = 15,
) -> List[Dict[str, Any]]:
    """Search Reddit via the public JSON endpoint.

    Args:
        query: Search query string
        depth: 'quick', 'default', or 'deep' — controls result limit
        subreddit: Optional subreddit name (without r/) for scoped search
        timeout: HTTP timeout in seconds

    Returns:
        List of normalized post dicts. Empty list on any failure.
    """
    limit = DEPTH_LIMITS.get(depth, DEPTH_LIMITS["default"])
    encoded_query = _url_encode(query)

    if subreddit:
        sub = subreddit.removeprefix("r/").strip()
        url = (
            f"https://www.reddit.com/r/{sub}/search.json"
            f"?q={encoded_query}&restrict_sr=on&sort=relevance&t=month&limit={limit}&raw_json=1"
        )
    else:
        url = (
            f"https://www.reddit.com/search.json"
            f"?q={encoded_query}&sort=relevance&t=month&limit={limit}&raw_json=1"
        )

    data = _fetch_json(url, timeout=timeout)
    posts = _parse_posts(data)

    # Dedupe by URL and assign IDs
    seen_urls = set()
    unique = []
    for post in posts:
        if post["url"] not in seen_urls:
            seen_urls.add(post["url"])
            unique.append(post)

    for i, post in enumerate(unique):
        post["id"] = f"R{i + 1}"

    return unique[:limit]


def search_reddit_public(
    topic: str,
    from_date: str,
    to_date: str,
    depth: str = "default",
    subreddits: Optional[List[str]] = None,
) -> List[Dict[str, Any]]:
    """High-level free Reddit search + enrichment (keyless).

    Thin compatibility shim over the tiered keyless pipeline: the legacy
    ``.json`` search/enrichment endpoints now return HTTP 403, so this delegates
    to ``reddit_keyless.search_and_enrich`` (Tier 0 one-shot ``.json`` →
    Tier 1 RSS discovery → Tier 2 shreddit comment enrichment). The name and
    signature are preserved so ``pipeline.py`` and other callers need no change
    and the ScrapeCreators backup still engages when this returns empty.

    The module-level ``search`` / ``_parse_posts`` helpers remain in use as the
    keyless pipeline's demoted Tier 0 ``.json`` attempt.

    Args:
        topic: Search topic
        from_date: Start date (YYYY-MM-DD)
        to_date: End date (YYYY-MM-DD)
        depth: 'quick', 'default', or 'deep'
        subreddits: Optional list of subreddit names (without r/) for targeted search

    Returns:
        List of normalized item dicts matching ScrapeCreators output format.
        Empty list on total failure (so SC backup can engage).
    """
    from . import reddit_keyless
    return reddit_keyless.search_and_enrich(
        topic, from_date, to_date, depth=depth, subreddits=subreddits
    )