fix: violation detection fix extraction query!

TogetherCrew · Aug 8, 2024 · 4f30968 · 4f30968
1 parent b196840
commit 4f30968
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 2 deletions.
diff --git a/dags/violation_detection_etl.py b/dags/violation_detection_etl.py
@@ -49,6 +49,8 @@ def process_platforms(platform: dict[str, Any]):
         recompute = platform["recompute"]
         discord_users = platform["selected_discord_users"]
 
+        logging.info(f"Processing PLATFORM_ID: {platform_id}!")
+
         # EXTRACT
 
         # TODO: Get resource_identifier from platform analyzer config
@@ -72,6 +74,7 @@ def process_platforms(platform: dict[str, Any]):
 
         # Load
         if len(transformed_data) != 0:
+            logging.info(f"Loading {len(transformed_data)} documents into db!")
             loader = LoadPlatformLabeledData()
             loader.load(platform_id=platform, transformed_data=transformed_data)
 
@@ -84,7 +87,9 @@ def process_platforms(platform: dict[str, Any]):
                     reporter = SendReportDiscordUser(platform_id=platform_id)
                     reporter.send(discord_user_id=discord_id, message=report)
         else:
-            logging.warning("No documents were transformed!")
+            logging.warning(
+                f"PLATFORM_ID: {platform_id}, No documents were transformed!"
+            )
 
     platforms = get_violation_modules()
     process_platforms.expand(platform=platforms)
diff --git a/dags/violation_detection_helpers/extract.py b/dags/violation_detection_helpers/extract.py
@@ -102,7 +102,7 @@ def extract(
             {
                 **date_query,
                 "text": {"$ne": None},
-                "source_id": {"$in": resources},
+                self.resource_name: {"$in": resources},
             }
         )
         return cursor, override_recompute