"""Reddit subreddit scraper - PRIMARY collector for trending questions. Scrapes "explain it to me" subreddits where people ask about trending topics. Uses direct HTTP calls to old.reddit.com JSON endpoints - NO PRAW, NO auth. Supports incremental collection via reddit_seen table. """ import sqlite3 import time import requests from pathlib import Path from typing import List, Optional, Set, Tuple from datetime import datetime, timezone from .base import BaseCollector, TrendItem from config import ( REDDIT_SUBREDDITS, REDDIT_POSTS_PER_SUB, REDDIT_REQUEST_DELAY, REDDIT_USER_AGENT, ) # Database path (same as main.py) DB_PATH = Path(__file__).parent.parent / "output" / "trends_history.db" class RedditScraperCollector(BaseCollector): """Scrapes Reddit subreddits for trending questions/topics with incremental collection.""" def __init__(self, subreddits: List[str] = None): """ Initialize the Reddit scraper. Args: subreddits: Optional list of subreddit names to scrape. If None, uses the default REDDIT_SUBREDDITS from config.py. When a channel is specified, pass the channel's configured subreddits. """ super().__init__("reddit") self.headers = {"User-Agent": REDDIT_USER_AGENT} self.backoff_times = [30, 60, 120, 240, 300] # Exponential backoff self.stats = {"new": 0, "seen": 0, "by_subreddit": {}} self.subreddits = subreddits if subreddits else REDDIT_SUBREDDITS def _get_seen_post_ids(self) -> Set[str]: """Load all previously seen post IDs from database.""" try: conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute("SELECT post_id FROM reddit_seen") seen = {row[0] for row in cursor.fetchall()} conn.close() return seen except Exception as e: self.errors.append(f"Error loading seen posts: {e}") return set() def _mark_posts_seen(self, posts: List[Tuple[str, str]]): """Mark post IDs as seen in database. Args: posts: List of (post_id, subreddit) tuples """ if not posts: return try: conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() now = datetime.now(timezone.utc).isoformat() cursor.executemany( "INSERT OR IGNORE INTO reddit_seen (post_id, subreddit, first_seen_at) VALUES (?, ?, ?)", [(p[0], p[1], now) for p in posts] ) conn.commit() conn.close() except Exception as e: self.errors.append(f"Error marking posts seen: {e}") def _fetch_json(self, url: str, retries: int = 3) -> Optional[dict]: """ Fetch JSON from Reddit with exponential backoff on 429s. Args: url: The Reddit JSON endpoint retries: Number of retry attempts Returns: Parsed JSON dict or None on failure """ for attempt in range(retries): try: response = requests.get( url, headers=self.headers, timeout=15 ) if response.status_code == 429: # Rate limited - exponential backoff backoff = self.backoff_times[min(attempt, len(self.backoff_times) - 1)] print(f" Rate limited, waiting {backoff}s...") time.sleep(backoff) continue response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: self.errors.append(f"Fetch error {url}: {str(e)}") if attempt < retries - 1: time.sleep(5) return None def _scrape_subreddit( self, subreddit: str, sort: str, params: dict = None, seen_ids: Set[str] = None ) -> Tuple[List[TrendItem], List[Tuple[str, str]]]: """ Scrape posts from a subreddit, filtering out already-seen posts. Args: subreddit: Subreddit name (without r/) sort: Sort type (hot, top, rising) params: Additional URL params (e.g., t=month for top) seen_ids: Set of post IDs to skip (already collected) Returns: Tuple of (List of TrendItem objects, List of (post_id, subreddit) for new posts) """ if seen_ids is None: seen_ids = set() base_url = f"https://old.reddit.com/r/{subreddit}/{sort}.json" url_params = {"limit": REDDIT_POSTS_PER_SUB} if params: url_params.update(params) param_str = "&".join(f"{k}={v}" for k, v in url_params.items()) url = f"{base_url}?{param_str}" data = self._fetch_json(url) if not data: return [], [] items = [] new_post_ids = [] posts = data.get("data", {}).get("children", []) for rank, post in enumerate(posts, start=1): post_data = post.get("data", {}) post_id = post_data.get("id", "") title = post_data.get("title", "") if not title or not post_id: continue # Skip already-seen posts if post_id in seen_ids: self.stats["seen"] += 1 continue # Get selftext (first 500 chars) selftext = post_data.get("selftext", "")[:500] if post_data.get("selftext") else "" items.append(TrendItem( topic_name=title, source="reddit", rank=rank, region="global", # Reddit is global, not region-specific timestamp=datetime.now(timezone.utc).isoformat(), url=f"https://reddit.com{post_data.get('permalink', '')}", metadata={ "subreddit": subreddit, "post_id": post_id, "score": post_data.get("score", 0), "num_comments": post_data.get("num_comments", 0), "created_utc": post_data.get("created_utc"), "selftext": selftext, "sort": sort, "upvote_ratio": post_data.get("upvote_ratio"), "author": post_data.get("author"), } )) new_post_ids.append((post_id, subreddit)) self.stats["new"] += 1 return items, new_post_ids def collect(self, region: str = "global") -> List[TrendItem]: """ Collect trending questions from all target subreddits (incremental). Fetches 'hot', 'top/month', and 'rising' for each subreddit. Skips posts already seen in previous runs. Args: region: Ignored for Reddit (always global) Returns: List of TrendItem objects (only NEW posts) """ # Reset stats for this run self.stats = {"new": 0, "seen": 0, "by_subreddit": {}} # Load previously seen post IDs seen_ids = self._get_seen_post_ids() print(f" Loaded {len(seen_ids)} previously seen post IDs") all_items = [] all_new_post_ids = [] for subreddit in self.subreddits: print(f" Scraping r/{subreddit}...", end=" ") sub_items = [] sub_new_ids = [] # Fetch HOT posts (currently trending) hot_items, hot_ids = self._scrape_subreddit(subreddit, "hot", seen_ids=seen_ids) sub_items.extend(hot_items) sub_new_ids.extend(hot_ids) # Add to seen_ids so we don't duplicate within same run seen_ids.update(id for id, _ in hot_ids) time.sleep(REDDIT_REQUEST_DELAY) # Fetch TOP posts from last month (proven engagement) top_items, top_ids = self._scrape_subreddit(subreddit, "top", {"t": "month"}, seen_ids=seen_ids) sub_items.extend(top_items) sub_new_ids.extend(top_ids) seen_ids.update(id for id, _ in top_ids) time.sleep(REDDIT_REQUEST_DELAY) # Fetch RISING posts (early signal - catching momentum) rising_items, rising_ids = self._scrape_subreddit( subreddit, "rising", {"limit": 50}, seen_ids=seen_ids ) sub_items.extend(rising_items) sub_new_ids.extend(rising_ids) seen_ids.update(id for id, _ in rising_ids) time.sleep(REDDIT_REQUEST_DELAY) self.stats["by_subreddit"][subreddit] = len(sub_items) print(f"{len(sub_items)} new posts") all_items.extend(sub_items) all_new_post_ids.extend(sub_new_ids) # Mark all new posts as seen for future runs self._mark_posts_seen(all_new_post_ids) print(f" Reddit collection: {self.stats['new']} new, {self.stats['seen']} skipped (already seen)") return all_items