reddit_scraper/monitor.py
Rama master · réplica de lectura del repositorio original.
"""RedditMonitor: evaluates a subreddit (or BST thread) against keyword/filter rules."""
import logging
import os
import pickle
import time
from . import config, credentials, notifications, sources
class RedditMonitor:
max_file_size = 5 * 1024 * 1024 # 5 MB
# Kept for backwards-compat with tests; the active source is now driven by the
# source chain in `sources`, not this flag.
_using_json_fallback = False
_thread_cache = {} # subreddit+pattern -> {thread_id, cached_at}
THREAD_CACHE_TTL = 3600 # refresh cached thread ID every hour
def __init__(
self,
reddit,
subreddit,
keywords,
min_upvotes=None,
exclude_keywords=None,
domain_contains=None,
domain_excludes=None,
flair_contains=None,
author_includes=None,
author_excludes=None,
**kwargs,
):
self.reddit = reddit
self.subreddit = subreddit
self.keywords = keywords
self.min_upvotes = min_upvotes
self.exclude_keywords = exclude_keywords or []
self.domain_contains = domain_contains or []
self.domain_excludes = domain_excludes or []
self.flair_contains = flair_contains or []
self.author_includes = author_includes or []
self.author_excludes = author_excludes or []
self.monitor_type = kwargs.get('monitor_type', 'posts')
self.thread_title_pattern = kwargs.get('thread_title_pattern', 'Buy/Sell/Trade')
self.keyword_logic = kwargs.get('keyword_logic', 'any')
self.load_processed_submissions()
def run(self):
"""Dispatch to the correct monitor method based on monitor_type."""
if self.monitor_type == 'thread_comments':
self.search_thread_comments()
else:
self.search_reddit_for_keywords()
def send_push_notification(self, message, title=None):
"""Send notification via Apprise to all configured services."""
urls = credentials.CREDENTIALS.get('notification_urls', []) if credentials.CREDENTIALS else []
if not urls:
logging.debug("No notification services configured, skipping notification")
return
logging.info(f"Sending notification to {len(urls)} service(s)...")
result = notifications.dispatch(message, title or f"Reddit Alert: r/{self.subreddit}")
if result:
logging.info("✅ Notification sent successfully")
else:
logging.warning("⚠️ Some notifications may have failed")
def send_error_notification(self, error_message):
"""Send error notification via Apprise to all configured services."""
notifications.notify_error(error_message)
@property
def processed_submissions_file(self):
return config.get_processed_submissions_path()
def load_processed_submissions(self):
try:
with open(self.processed_submissions_file, 'rb') as file:
self.processed_submissions = pickle.load(file)
except FileNotFoundError:
self.processed_submissions = set()
def save_processed_submissions(self):
path = self.processed_submissions_file
if os.path.exists(path) and os.path.getsize(path) > self.max_file_size:
logging.info("Processed submissions file exceeded max size. Deleting and creating a new one.")
os.remove(path)
self.processed_submissions = set()
with open(path, 'wb') as file:
pickle.dump(self.processed_submissions, file)
def find_current_thread(self):
"""Find the thread ID of the current weekly megathread, with 1-hour caching."""
cache_key = f"{self.subreddit}-{self.thread_title_pattern}"
cached = RedditMonitor._thread_cache.get(cache_key)
if cached and (time.time() - cached['cached_at']) < RedditMonitor.THREAD_CACHE_TTL:
return cached['thread_id']
thread_id = None
pattern = self.thread_title_pattern.lower()
# Stickied posts are the most reliable signal, but only PRAW exposes them.
if self.reddit and sources._source_available('oauth'):
try:
sub = self.reddit.subreddit(self.subreddit)
for slot in [1, 2]:
try:
sticky = sub.sticky(number=slot)
if pattern in sticky.title.lower():
thread_id = sticky.id
break
except Exception:
pass
except Exception as e:
logging.warning(f"PRAW error finding sticky BST thread: {e}")
# Otherwise scan recent posts via the configured source chain.
if not thread_id:
posts, _ = sources.fetch_posts(self.subreddit, 25, self.reddit)
for post in posts or []:
if pattern in (post['title'] or '').lower():
if post.get('id'):
thread_id = post['id']
else:
parts = post['permalink'].strip('/').split('/')
if 'comments' in parts:
thread_id = parts[parts.index('comments') + 1]
break
if thread_id:
RedditMonitor._thread_cache[cache_key] = {'thread_id': thread_id, 'cached_at': time.time()}
logging.info(f"Thread located: {thread_id}")
else:
logging.warning(f"Could not find thread matching '{self.thread_title_pattern}' in r/{self.subreddit}")
return thread_id
def search_thread_comments(self):
"""Scan a pinned/recurring thread's comments for keyword matches."""
thread_id = self.find_current_thread()
if not thread_id:
return
logging.info(f"Scanning thread {thread_id} comments in r/{self.subreddit}...")
comments = sources.fetch_thread_comments(self.subreddit, thread_id, self.reddit)
if comments is None:
return
for comment in comments:
self._process_comment(
comment_id=comment['id'],
body=comment['body'],
author=comment['author'],
permalink=comment['permalink'],
)
logging.info(f"Finished scanning thread comments in r/{self.subreddit}.")
def _process_comment(self, comment_id, body, author, permalink):
"""Check a BST comment against keyword filters and notify on match."""
submission_id = f"{self.subreddit}-comment-{comment_id}"
if submission_id in self.processed_submissions:
return False
body_lower = body.lower()
if self.keyword_logic == 'any':
has_keywords = any(kw.lower() in body_lower for kw in self.keywords)
else:
has_keywords = all(kw.lower() in body_lower for kw in self.keywords)
has_excluded = any(kw.lower() in body_lower for kw in self.exclude_keywords)
author_lower = author.lower()
meets_author_includes = not self.author_includes or author_lower in [a.lower() for a in self.author_includes]
meets_author_excludes = author_lower not in [a.lower() for a in self.author_excludes]
if has_keywords and not has_excluded and meets_author_includes and meets_author_excludes:
excerpt = body[:300] + '...' if len(body) > 300 else body
message = (
f"BST listing match in r/{self.subreddit}!\n"
f"u/{author}:\n{excerpt}\n"
f"Link: https://www.reddit.com{permalink}"
)
self.send_push_notification(message, title="FMF BST Match")
logging.info(f"BST match: u/{author} | {body[:80]}...")
self.processed_submissions.add(submission_id)
self.save_processed_submissions()
return True
return False
def search_reddit_for_keywords(self):
"""Search a subreddit for keywords, fetching posts through the configured source chain."""
logging.info(f"Searching '{self.subreddit}' subreddit for keywords...")
posts, source = sources.fetch_posts(self.subreddit, 10, self.reddit)
if posts is None:
logging.error(f"Failed to fetch posts for '{self.subreddit}' from all sources")
return
for post in posts:
self._process_post(
post_id=post['id'],
title=post['title'],
url=post['url'],
score=post['score'],
permalink=post['permalink'],
domain=post['domain'],
flair=post['link_flair_text'] or '',
author=post['author'],
)
logging.info(f"Finished searching '{self.subreddit}' subreddit (source: {source}).")
def _process_post(self, post_id, title, url, score, permalink, domain, flair, author):
"""Process a single post and send notification if it matches filters."""
submission_id = f"{self.subreddit}-{post_id}"
if submission_id in self.processed_submissions:
logging.debug(f"Skipping duplicate post: {title}")
return False
message = (
f"Match found in '{self.subreddit}' subreddit:\n"
f"Title: {title}\n"
f"URL: {url}\n"
f"Upvotes: {score}\n"
f"Permalink: https://www.reddit.com{permalink}\n"
)
# Check filters
title_lower = title.lower()
has_all_keywords = all(keyword in title_lower for keyword in self.keywords)
has_excluded = any(keyword in title_lower for keyword in self.exclude_keywords)
meets_upvotes = self.min_upvotes is None or score >= self.min_upvotes
# Domain filters
submission_domain = domain.lower()
meets_domain_contains = not self.domain_contains or any(d in submission_domain for d in self.domain_contains)
meets_domain_excludes = not any(d in submission_domain for d in self.domain_excludes)
# Flair filter
submission_flair = flair.lower()
meets_flair = not self.flair_contains or any(f.lower() in submission_flair for f in self.flair_contains)
# Author filters
author_name = author.lower()
meets_author_includes = not self.author_includes or author_name in [a.lower() for a in self.author_includes]
meets_author_excludes = author_name not in [a.lower() for a in self.author_excludes]
if (
has_all_keywords
and not has_excluded
and meets_upvotes
and meets_domain_contains
and meets_domain_excludes
and meets_flair
and meets_author_includes
and meets_author_excludes
):
logging.info(message)
self.send_push_notification(message)
logging.info('-' * 40)
self.processed_submissions.add(submission_id)
self.save_processed_submissions()
return True
return False