| import os |
| import time |
| import random |
| import requests |
| import logging |
| from typing import List, Dict, Optional |
|
|
| |
| logger = logging.getLogger(__name__) |
|
|
| |
| HEADERS = {"User-Agent": "AltomicAi/3.0 (by /u/yourusername)"} |
|
|
|
|
| class RateLimiter: |
| """Simple rate limiter to enforce a given number of requests per second.""" |
| 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, requests_per_second: float = 0.3): |
| self.rate_limiter = RateLimiter(requests_per_second) |
|
|
| |
| proxy_list_str = os.environ.get("PROXY_LIST") |
| print(f"[DEBUG] PROXY_LIST env var: {'SET (length ' + str(len(proxy_list_str)) + ')' if proxy_list_str else 'NOT SET'}", flush=True) |
|
|
| self.proxy_list = self._parse_proxies(proxy_list_str) |
| print(f"[DEBUG] Loaded {len(self.proxy_list)} proxies", flush=True) |
|
|
| if self.proxy_list: |
| print(f"[DEBUG] First proxy: {self.proxy_list[0] if self.proxy_list else 'None'}", flush=True) |
| else: |
| print("[WARNING] No proxies loaded. Requests will be made directly (likely blocked).", flush=True) |
|
|
| def _parse_proxies(self, content: Optional[str]) -> List[Dict[str, str]]: |
| """Convert proxy lines (http://user:pass@ip:port) to requests proxy dicts.""" |
| proxies = [] |
| if not content: |
| return proxies |
| for line in content.strip().splitlines(): |
| line = line.strip() |
| if line and not line.startswith('#'): |
| |
| if not line.startswith('http'): |
| line = f"http://{line}" |
| proxies.append({"http": line, "https": line}) |
| return proxies |
|
|
| def _get_random_proxy(self) -> Optional[Dict[str, str]]: |
| if not self.proxy_list: |
| return None |
| return random.choice(self.proxy_list) |
|
|
| def _request(self, url: str, params: dict = None) -> dict: |
| """Make a request with rate limiting and proxy rotation.""" |
| self.rate_limiter.wait() |
| proxy = self._get_random_proxy() |
| print(f"[DEBUG] Requesting {url} with proxy: {proxy if proxy else 'DIRECT'}", flush=True) |
|
|
| try: |
| resp = requests.get( |
| url, |
| headers=HEADERS, |
| params=params, |
| proxies=proxy, |
| timeout=15 |
| ) |
| print(f"[DEBUG] Response status: {resp.status_code}", flush=True) |
| resp.raise_for_status() |
| return resp.json() |
| except requests.exceptions.RequestException as e: |
| print(f"[ERROR] Reddit request failed: {e}", flush=True) |
| if hasattr(e, 'response') and e.response is not None: |
| print(f"[ERROR] Response text: {e.response.text[:500]}", flush=True) |
| return {} |
|
|
| def get_top_posts(self, subreddit: str, limit: int = 30, time_filter: str = "day") -> List[dict]: |
| """ |
| Fetch top posts from a subreddit, handling pagination up to `limit` posts. |
| Returns a list of post dicts. |
| """ |
| print(f"[DEBUG] Fetching top posts from r/{subreddit}, limit={limit}, time_filter={time_filter}", flush=True) |
| all_posts = [] |
| after = None |
| per_page = min(limit, 100) |
| remaining = limit |
|
|
| 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: |
| print(f"[WARNING] No data returned from Reddit for {url}", flush=True) |
| break |
|
|
| children = data.get("data", {}).get("children", []) |
| if not children: |
| print(f"[INFO] No children in response for {subreddit}", flush=True) |
| 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: |
| print(f"[INFO] No 'after' token, pagination finished", flush=True) |
| break |
|
|
| print(f"[DEBUG] Fetched {len(all_posts)} posts from r/{subreddit}", flush=True) |
| return all_posts |
|
|
| def get_post_comments(self, post_id: str, limit: int = 20) -> List[dict]: |
| """Fetch top-level comments for a post.""" |
| 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", []) |
| comments = [c["data"] for c in comments_data if c["kind"] == "t1"] |
| return comments[:limit] |
|
|
| def get_subreddit_info(self, subreddit: str) -> dict: |
| """Fetch subreddit metadata.""" |
| url = f"{self.BASE_URL}/r/{subreddit}/about.json" |
| data = self._request(url) |
| return data.get("data", {}) if data else {} |
