last30days-skill/scripts/lib/reddit_public.py

"""Standalone Reddit public JSON search module.

Searches Reddit using the free public JSON endpoints (no API key required).
Promoted from last-resort fallback to robust primary free path.

Endpoints:
- 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 json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from typing import Any, Dict, List, Optional


USER_AGENT = "last30days/3.0 (research tool)"

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

# How many top posts to enrich with comments, by depth
ENRICH_LIMITS = {
    "quick": 3,
    "default": 5,
    "deep": 8,
}

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",
    }
    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

                body = resp.read().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.lstrip("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 _enrich_post(item: Dict[str, Any], timeout: int = 10) -> Dict[str, Any]:
    """Enrich a single post with top comments. Never raises."""
    try:
        from . import reddit_enrich
        thread_data = reddit_enrich.fetch_thread_data(item["url"], timeout=timeout)
        if not thread_data:
            return item
        parsed = reddit_enrich.parse_thread_data(thread_data)
        comments = parsed.get("comments", [])
        top = reddit_enrich.get_top_comments(comments)
        item["top_comments"] = [
            {
                "score": c.get("score", 0),
                "excerpt": (c.get("body") or "")[:200],
                "author": c.get("author", ""),
            }
            for c in top[:10]
        ]
    except Exception:
        # Never discard — keep post with empty metadata
        pass
    return item


def _enrich_posts(posts: List[Dict[str, Any]], depth: str = "default") -> List[Dict[str, Any]]:
    """Enrich top N posts with comment data using threads. Total budget 45s."""
    limit = ENRICH_LIMITS.get(depth, ENRICH_LIMITS["default"])
    to_enrich = posts[:limit]
    rest = posts[limit:]

    if not to_enrich:
        return posts

    enriched = []
    try:
        with ThreadPoolExecutor(max_workers=min(limit, 4)) as executor:
            futures = {
                executor.submit(_enrich_post, post, 10): i
                for i, post in enumerate(to_enrich)
            }
            # Collect results with 45s total budget
            import concurrent.futures
            done, not_done = concurrent.futures.wait(futures, timeout=45)
            # Build result list preserving order
            result_map: Dict[int, Dict[str, Any]] = {}
            for future in done:
                idx = futures[future]
                try:
                    result_map[idx] = future.result(timeout=0)
                except Exception:
                    result_map[idx] = to_enrich[idx]
            # Any not-done futures: keep original post
            for future in not_done:
                idx = futures[future]
                result_map[idx] = to_enrich[idx]
                future.cancel()
            enriched = [result_map[i] for i in range(len(to_enrich))]
    except Exception:
        enriched = to_enrich

    return enriched + rest


def _search_subreddit(sub: str, topic: str, depth: str, timeout: int = 15) -> List[Dict[str, Any]]:
    """Search a single subreddit. Never raises."""
    try:
        return search(topic, depth=depth, subreddit=sub, timeout=timeout)
    except Exception as e:
        _log(f"Subreddit search failed for r/{sub}: {e}")
        return []


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 Reddit public search matching the openai_reddit interface.

    When subreddits are provided (from agent planning), searches each targeted
    sub first, then does global search, and deduplicates across both. This
    mirrors the SC search_and_enrich() flow where pre-resolved subreddits get
    priority.

    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.
    """
    all_posts: List[Dict[str, Any]] = []

    # Phase 1: Search targeted subreddits in parallel (if provided)
    if subreddits:
        _log(f"Searching {len(subreddits)} targeted subreddits: {subreddits}")
        workers = min(4, len(subreddits))
        with ThreadPoolExecutor(max_workers=workers) as executor:
            futures = {
                executor.submit(_search_subreddit, sub, topic, depth): sub
                for sub in subreddits
            }
            for future in futures:
                sub = futures[future]
                try:
                    sub_posts = future.result(timeout=30)
                    _log(f"  -> {len(sub_posts)} results from r/{sub}")
                    all_posts.extend(sub_posts)
                except (Exception, FuturesTimeoutError) as e:
                    _log(f"  -> r/{sub} failed: {e}")

    # Phase 2: Global search
    global_posts = search(topic, depth=depth)
    all_posts.extend(global_posts)

    # Deduplicate by URL (targeted results keep priority since they come first)
    seen_urls: set = set()
    results: List[Dict[str, Any]] = []
    for post in all_posts:
        if post["url"] not in seen_urls:
            seen_urls.add(post["url"])
            results.append(post)

    # Date filter: keep posts in range or with unknown dates
    filtered = []
    for item in results:
        d = item.get("date")
        if d is None or (from_date <= d <= to_date):
            filtered.append(item)

    # Sort by engagement (score desc)
    filtered.sort(
        key=lambda x: x.get("engagement", {}).get("score", 0),
        reverse=True,
    )

    # Enrich top posts with comments
    filtered = _enrich_posts(filtered, depth=depth)

    # Re-index IDs
    for i, item in enumerate(filtered):
        item["id"] = f"R{i + 1}"

    return filtered