feat(enrichers): email_to_username + fix sherlock

flowsint-enrichers/src/flowsint_enrichers/email/to_username.py

@@ -0,0 +1,61 @@
+from typing import List
+
+from flowsint_core.core.enricher_base import Enricher
+from flowsint_types.email import Email
+from flowsint_types.username import Username
+
+from flowsint_enrichers.registry import flowsint_enricher
+
+
+@flowsint_enricher
+class EmailToUsernameEnricher(Enricher):
+    """From email to username."""
+
+    InputType = Email
+    OutputType = Username
+
+    @classmethod
+    def name(cls) -> str:
+        return "email_to_username"
+
+    @classmethod
+    def category(cls) -> str:
+        return "Email"
+
+    @classmethod
+    def key(cls) -> str:
+        return "email"
+
+    async def scan(self, data: List[InputType]) -> List[OutputType]:
+        results: List[OutputType] = []
+
+        for email in data:
+            # chamyanis@gmail.com
+            splitted = email.email.split("@")
+            username = splitted[0]
+            results.append(Username(value=username))
+
+        return results
+
+    def postprocess(
+        self, results: List[OutputType], original_input: List[InputType]
+    ) -> List[OutputType]:
+        for email_obj, username_obj in zip(original_input, results):
+            if not self.neo4j_conn:
+                continue
+            # Create email node
+            self.create_node(email_obj)
+            self.create_node(username_obj)
+            # Create relationship between email and gravatar
+            self.create_relationship(email_obj, username_obj, "HAS_USERNAME")
+
+            self.log_graph_message(
+                f"Exctracted username for {email_obj.email} -> username: {username_obj.value}"
+            )
+
+        return results
+
+
+# Make types available at module level for easy access
+InputType = EmailToUsernameEnricher.InputType
+OutputType = EmailToUsernameEnricher.OutputType

flowsint-enrichers/src/flowsint_enrichers/social/to_sherlock.py

@@ -1,11 +1,12 @@
 import subprocess
 from pathlib import Path
-from typing import List, Union
-from flowsint_core.utils import is_valid_username
-from flowsint_types import SocialAccount, Username
+from typing import List
+
 from flowsint_core.core.enricher_base import Enricher
-from flowsint_enrichers.registry import flowsint_enricher
 from flowsint_core.core.logger import Logger
+from flowsint_types import SocialAccount, Username
+
+from flowsint_enrichers.registry import flowsint_enricher
 
 
 @flowsint_enricher
@@ -72,7 +73,9 @@ class SherlockEnricher(Enricher):
                 # Create Social objects for each found account
                 for platform, url in found_accounts.items():
                     results.append(
-                        SocialAccount(username=username, platform=platform, profile_url=url)
+                        SocialAccount(
+                            username=username, platform=platform, profile_url=url
+                        )
                     )
 
             except subprocess.TimeoutExpired:
@@ -90,16 +93,34 @@ class SherlockEnricher(Enricher):
 
         return results
 
-    def postprocess(self, results: List[OutputType], original_input: List[InputType]) -> List[OutputType]:
+    def postprocess(
+        self, results: List[OutputType], original_input: List[InputType]
+    ) -> List[OutputType]:
         """Create Neo4j relationships for found social accounts."""
         if not self.neo4j_conn:
             return results
 
         for social_account in results:
-            self.create_node(social_account)
-            self.log_graph_message(
-                f"Found social account: {social_account.username.value} on {social_account.platform}"
-            )
+            try:
+                # Create username node
+                self.create_node(social_account.username)
+                # Create social account node
+                self.create_node(social_account)
+                # Create relationship
+                self.create_relationship(
+                    social_account.username, social_account, "HAS_SOCIAL_ACCOUNT"
+                )
+                self.log_graph_message(
+                    f"{social_account.username.value} -> account found on {social_account.platform}"
+                )
+            except Exception as e:
+                Logger.error(
+                    self.sketch_id,
+                    {
+                        "message": f"Failed to create graph nodes for {social_account.username.value} on {social_account.platform}: {e}"
+                    },
+                )
+                continue
 
         return results