Spaces:
Running
Running
File size: 2,233 Bytes
a96145c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | import logging
from urllib.parse import urlparse
from .models import RedditPost
logger = logging.getLogger(__name__)
class RedditScraper:
def __init__(self, config):
self.config = config
self._praw = None
self._init_praw()
def _init_praw(self):
cid = self.config.reddit_client_id
secret = self.config.reddit_client_secret
if not cid or not secret:
raise ValueError(
"Reddit API credentials not found.\n"
" Set REDDIT_CLIENT_ID and REDDIT_CLIENT_SECRET in .env, or\n"
" use python main.py --demo to run with sample data."
)
try:
import praw
self._praw = praw.Reddit(
client_id=cid,
client_secret=secret,
user_agent=self.config.reddit_user_agent,
)
logger.info("Reddit API initialised via PRAW")
except ImportError:
raise ImportError("praw is required. Install with: pip install praw")
def fetch_posts(self) -> list[RedditPost]:
seen = set()
posts = []
for sub in self.config.news_subreddits:
logger.info("Fetching r/%s ...", sub)
try:
batch = self._fetch_subreddit(sub)
for p in batch:
if p.id not in seen:
seen.add(p.id)
posts.append(p)
except Exception as exc:
logger.error("Failed to fetch r/%s: %s", sub, exc)
return posts
def _fetch_subreddit(self, subreddit: str) -> list[RedditPost]:
results = []
sub = self._praw.subreddit(subreddit)
for submission in sub.hot(limit=self.config.posts_per_subreddit):
if submission.is_self:
continue
results.append(RedditPost(
id=submission.id,
title=submission.title,
url=submission.url,
subreddit=subreddit.lower(),
score=submission.score,
num_comments=submission.num_comments,
source_domain=urlparse(submission.url).netloc,
))
return results
|