Ingest fixes and event cleanup

redditrepostsleuth/core/celery/task_logic/monitored_sub_task_logic.py

@@ -29,7 +29,7 @@ def process_monitored_subreddit_submission(post_id: str, monitored_sub_svc: Moni
 
     if monitored_sub.adult_promoter_remove_post or monitored_sub.adult_promoter_ban_user or monitored_sub.adult_promoter_notify_mod_mail:
         try:
-            check_user_for_only_fans(uow, post.author)
+            check_user_for_only_fans(uow, post.author, monitored_sub_svc.reddit)
         except (UtilApiException, ConnectionError, TooManyRequests) as e:
             log.warning('Failed to do onlyfans check for user %s', post.author)
 

redditrepostsleuth/core/celery/tasks/monitored_sub_tasks.py

@@ -44,7 +44,8 @@ def __init__(self):
         response_handler = ResponseHandler(self.reddit, self.uowm, event_logger, source='submonitor', live_response=self.config.live_responses)
         dup_image_svc = DuplicateImageService(self.uowm, event_logger, self.reddit, config=self.config)
         response_builder = ResponseBuilder(self.uowm)
-        self.monitored_sub_svc = MonitoredSubService(dup_image_svc, self.uowm, self.reddit, response_builder, response_handler, event_logger=event_logger, config=self.config)
+        self.monitored_sub_svc = MonitoredSubService(dup_image_svc, self.uowm, self.reddit, response_builder, event_logger=event_logger, config=self.config)
+
 
 
 @celery.task(
@@ -55,11 +56,15 @@ def __init__(self):
     retry_kwards={'max_retries': 3}
 )
 def sub_monitor_check_post(self, post_id: str, monitored_sub: MonitoredSub):
-    update_log_context_data(log, {'trace_id': str(randint(100000, 999999)), 'post_id': post_id,
-                                  'subreddit': monitored_sub.name, 'service': 'Subreddit_Monitor'})
-
-    with self.uowm.start() as uow:
-        process_monitored_subreddit_submission(post_id, self.monitored_sub_svc, uow)
+    try:
+        update_log_context_data(log, {'trace_id': str(randint(100000, 999999)), 'post_id': post_id,
+                                      'subreddit': monitored_sub.name, 'service': 'Subreddit_Monitor'})
+
+        with self.uowm.start() as uow:
+            process_monitored_subreddit_submission(post_id, self.monitored_sub_svc, uow)
+    except Exception as e:
+        log.exception('General failure')
+        pass
 
 
 @celery.task(

redditrepostsleuth/core/celery/tasks/reddit_action_tasks.py

@@ -8,6 +8,7 @@
 from redditrepostsleuth.core.db.db_utils import get_db_engine
 from redditrepostsleuth.core.db.uow.unitofworkmanager import UnitOfWorkManager
 from redditrepostsleuth.core.logging import get_configured_logger
+from redditrepostsleuth.core.model.events.RedditAdminActionEvent import RedditAdminActionEvent
 from redditrepostsleuth.core.notification.notification_service import NotificationService
 from redditrepostsleuth.core.services.eventlogging import EventLogging
 from redditrepostsleuth.core.services.response_handler import ResponseHandler
@@ -38,6 +39,12 @@ def remove_submission_task(self, submission: Submission, removal_reason: str, mo
         removal_reason_id = get_removal_reason_id(removal_reason, submission.subreddit)
         log.info('Attempting to remove post https://redd.it/%s with removal ID %s', submission.id, removal_reason_id)
         submission.mod.remove(reason_id=removal_reason_id, mod_note=mod_note)
+        self.event_logger.save_event(
+            RedditAdminActionEvent(
+                submission.subreddit.display_name,
+                'remove_submission'
+            )
+        )
     except Forbidden:
         log.error('Failed to remove post https://redd.it/%s, no permission', submission.id)
         send_modmail_task.apply_async(
@@ -66,6 +73,12 @@ def ban_user_task(self, username: str, subreddit_name: str, ban_reason: str, not
     try:
         subreddit = self.reddit.subreddit(subreddit_name)
         subreddit.banned.add(username, ban_reason=ban_reason, note=note)
+        self.event_logger.save_event(
+            RedditAdminActionEvent(
+                subreddit_name,
+                'ban_user'
+            )
+        )
     except TooManyRequests as e:
         log.warning('Too many requests when banning user')
         raise e
@@ -109,6 +122,12 @@ def lock_submission_task(self, submission: Submission) -> None:
     log.info('Locking submission https://redd.it/%s', submission.id)
     try:
         submission.mod.lock()
+        self.event_logger.save_event(
+            RedditAdminActionEvent(
+                submission.subreddit.display_name,
+                'submission_lock'
+            )
+        )
     except TooManyRequests as e:
         log.warning('Too many requests when locking submission')
         raise e
@@ -129,6 +148,12 @@ def lock_comment_task(self, comment: Comment) -> None:
     log.info('Locking comment https://reddit.com%s', comment.permalink)
     try:
         comment.mod.lock()
+        self.event_logger.save_event(
+            RedditAdminActionEvent(
+                comment.subreddit.display_name,
+                'comment_lock'
+            )
+        )
     except TooManyRequests as e:
         log.warning('Too many requests when locking comment')
         raise e
@@ -149,6 +174,12 @@ def sticky_comment_task(self, comment: Comment) -> None:
     log.info('Make comment sticky: https://reddit.com%s ', comment.permalink)
     try:
         comment.mod.distinguish(sticky=True)
+        self.event_logger.save_event(
+            RedditAdminActionEvent(
+                comment.subreddit.display_name,
+                'comment_sticky'
+            )
+        )
     except TooManyRequests as e:
         log.warning('Too many requests when sticky comment')
         raise e
@@ -169,6 +200,12 @@ def mark_as_oc_task(self, submission: Submission) -> None:
     log.info('Marking submission %s as OC', submission.id)
     try:
         submission.mod.set_original_content()
+        self.event_logger.save_event(
+            RedditAdminActionEvent(
+                submission.subreddit.display_name,
+                'submission_mark_oc'
+            )
+        )
     except TooManyRequests as e:
         log.warning('Too many requests when marking submission OC')
         raise e
@@ -196,6 +233,12 @@ def report_submission_task(self, submission: Submission, report_msg: str) -> Non
     log.info('Reporting submission https://redd.it/%s', submission.id)
     try:
         submission.report(report_msg[:99])  # TODO: Until database column length is fixed
+        self.event_logger.save_event(
+            RedditAdminActionEvent(
+                submission.subreddit.display_name,
+                'submission_report'
+            )
+        )
     except TooManyRequests as e:
         log.warning('Too many requests when reporting submission')
         raise e

redditrepostsleuth/core/exception.py

@@ -75,3 +75,7 @@ def __init__(self, message):
 class RedGifsTokenException(RepostSleuthException):
     def __init__(self, message):
         super(RedGifsTokenException, self).__init__(message)
+
+class RedditTokenExpiredException(RepostSleuthException):
+    def __init__(self, message):
+        super(RedditTokenExpiredException, self).__init__(message)

redditrepostsleuth/core/model/events/RedditAdminActionEvent.py

@@ -0,0 +1,16 @@
+from redditrepostsleuth.core.model.events.influxevent import InfluxEvent
+
+
+class RedditAdminActionEvent(InfluxEvent):
+    def __init__(self, subreddit: str, action: str, event_type:str = None):
+        super(RedditAdminActionEvent, self).__init__(event_type=event_type)
+        self.subreddit = subreddit
+        self.count = 1
+        self.action = action
+
+    def get_influx_event(self):
+        event = super().get_influx_event()
+        #event[0]['fields']['count'] = self.count
+        event[0]['tags']['subreddit'] = self.subreddit
+        event[0]['tags']['action'] = self.action
+        return event

redditrepostsleuth/core/services/duplicateimageservice.py

@@ -9,7 +9,6 @@
 from requests.exceptions import ConnectionError
 from sqlalchemy.exc import IntegrityError
 
-from redditrepostsleuth.core.celery.admin_tasks import delete_post_task
 from redditrepostsleuth.core.config import Config
 from redditrepostsleuth.core.db.databasemodels import Post, MemeTemplate, MemeHash
 from redditrepostsleuth.core.db.uow.unitofworkmanager import UnitOfWorkManager
@@ -395,7 +394,7 @@ def _final_meme_filter(self,
                     match_hash = meme_hashes['dhash_h']
                 except ImageConversionException:
                     log.warning('Failed to get meme hash for %s.  Sending to delete queue', match.post.post_id)
-                    delete_post_task.apply_async((match.post.post_id,))
+                    #delete_post_task.apply_async((match.post.post_id,))
                     continue
                 except Exception:
                     log.exception('Failed to get meme hash for %s', match.post.url, exc_info=True)

redditrepostsleuth/core/services/eventlogging.py

@@ -66,7 +66,7 @@ def _flush_unsaved(self) -> NoReturn:
     def _write_to_influx(self, event: InfluxEvent) -> bool:
         try:
             self._influx_client.write(bucket=self._config.influx_bucket, record=event.get_influx_event())
-            #log.debug('Wrote to Influx: %s', event.get_influx_event())
+            log.debug('Wrote to Influx: %s', event.get_influx_event())
             self._successive_failures = 0
             return True
         except Exception as e:

redditrepostsleuth/core/util/onlyfans_handling.py

@@ -8,7 +8,7 @@
 
 import requests
 from praw import Reddit
-from prawcore import TooManyRequests
+from prawcore import TooManyRequests, NotFound
 from requests import Response
 from requests.exceptions import ConnectionError
 from sqlalchemy import func
@@ -115,6 +115,20 @@ def fetch_from_util_api(url: str) -> Response:
 
     return response
 
+def check_bio_for_promoter_links(username: str, reddit: Reddit) -> Optional[str]:
+    try:
+        redditor = reddit.redditor(username)
+        bio = redditor.subreddit.public_description
+    except (NotFound, AttributeError):
+        log.warning('Failed to get Redditor bio for username %s', username)
+        return
+
+    log.debug('Checking for of %s: %s', username, redditor.subreddit.public_description)
+
+    for domain in flagged_words:
+        if domain in bio:
+            return domain
+
 def get_profile_links(username: str) -> list[str]:
     url = f'{config.util_api}/profile?username={username}'
     response = fetch_from_util_api(url)
@@ -133,7 +147,12 @@ def get_profile_links(username: str) -> list[str]:
         raise UtilApiException(f'Unexpected status {response.status_code} from util API')
 
 
-def check_user_for_promoter_links(username: str) -> Optional[LinkCheckResult]:
+def check_user_for_promoter_links(username: str, reddit: Reddit) -> Optional[LinkCheckResult]:
+
+    flagged_bio_domain = check_bio_for_promoter_links(username, reddit)
+
+    if flagged_bio_domain:
+        return LinkCheckResult(source='Bio', url=flagged_bio_domain)
 
     profile_links = get_profile_links(username)
 
@@ -226,7 +245,7 @@ def get_links_from_comments_praw(username: str, reddit: Reddit) -> list[str]:
 
     return list(set(all_urls))
 
-def check_user_for_only_fans(uow: UnitOfWork, username: str) -> Optional[UserReview]:
+def check_user_for_only_fans(uow: UnitOfWork, username: str, reddit: Reddit) -> Optional[UserReview]:
     skip_names = ['[deleted]', 'AutoModerator']
 
     if username in skip_names:
@@ -238,7 +257,7 @@ def check_user_for_only_fans(uow: UnitOfWork, username: str) -> Optional[UserRev
 
         if user:
             delta = datetime.utcnow() - user.last_checked
-            if delta.days < 30:
+            if delta.days < 7:
                 log.info('Skipping existing user %s, last check was %s days ago', username, delta.days)
                 return
             user.content_links_found = False
@@ -249,7 +268,7 @@ def check_user_for_only_fans(uow: UnitOfWork, username: str) -> Optional[UserRev
         if not user:
             user = UserReview(username=username)
         try:
-            result = check_user_for_promoter_links(username)
+            result = check_user_for_promoter_links(username, reddit)
         except UserNotFound as e:
             log.warning(e)
             return

redditrepostsleuth/core/util/reddithelpers.py

@@ -2,7 +2,6 @@
 from typing import Text, Optional, List
 
 import requests
-from asyncpraw import Reddit as AsyncReddit
 from praw import Reddit
 from praw.exceptions import APIException
 from praw.models import Subreddit