Spaces:
Sleeping
Sleeping
| import time | |
| import requests | |
| import logging | |
| from typing import List, Dict | |
| from config import REDDIT_RPS | |
| logger = logging.getLogger(__name__) | |
| HEADERS = {"User-Agent": "FounderIntelligenceEngine/4.0 (by /u/yourusername)"} | |
| class RateLimiter: | |
| def __init__(self, requests_per_second: float): | |
| self.interval = 1.0 / requests_per_second if requests_per_second > 0 else 0 | |
| self.last_request = 0.0 | |
| def wait(self): | |
| if self.interval <= 0: return | |
| now = time.time() | |
| elapsed = now - self.last_request | |
| if elapsed < self.interval: | |
| time.sleep(self.interval - elapsed) | |
| self.last_request = time.time() | |
| class RedditClient: | |
| BASE_URL = "https://www.reddit.com" | |
| def __init__(self): | |
| self.rate_limiter = RateLimiter(REDDIT_RPS) | |
| def _request(self, url: str, params: dict = None) -> dict: | |
| self.rate_limiter.wait() | |
| try: | |
| resp = requests.get(url, headers=HEADERS, params=params, timeout=15) | |
| resp.raise_for_status() | |
| return resp.json() | |
| except Exception as e: | |
| logger.error(f"Reddit request failed: {e}") | |
| return {} | |
| def get_top_posts(self, subreddit: str, limit: int = 30, time_filter="day") -> List[dict]: | |
| all_posts, after = [], None | |
| remaining = limit | |
| per_page = min(limit, 100) | |
| while remaining > 0: | |
| params = {"limit": min(remaining, per_page), "t": time_filter} | |
| if after: params["after"] = after | |
| url = f"{self.BASE_URL}/r/{subreddit}/top.json" | |
| data = self._request(url, params) | |
| if not data: break | |
| children = data.get("data", {}).get("children", []) | |
| if not children: break | |
| for child in children: | |
| all_posts.append(child["data"]) | |
| remaining -= 1 | |
| if remaining == 0: break | |
| after = data.get("data", {}).get("after") | |
| if not after: break | |
| return all_posts | |
| def get_post_comments(self, post_id: str, limit: int = 5) -> List[dict]: | |
| url = f"{self.BASE_URL}/comments/{post_id}.json" | |
| data = self._request(url) | |
| if not data or not isinstance(data, list) or len(data) < 2: | |
| return [] | |
| comments_data = data[1].get("data", {}).get("children", []) | |
| return [c["data"] for c in comments_data if c["kind"] == "t1"][:limit] |