Feat/fix everything

app/celery/userDataComputation/pii.py → app/celery/compute/pii.py

@@ -5,17 +5,13 @@
 
 
 def remove_pii(text):
-    # Initialize Presidio engines for PII detection and anonymization
     analyzer = AnalyzerEngine()
     anonymizer = AnonymizerEngine()
 
-    # Step 1: Analyze the text for PII entities
     results = analyzer.analyze(text=text, entities=[], language="en")
 
-    # Anonymize the text based on detected PII and the anonymization configuration
     anonymized_result = anonymizer.anonymize(text=text, analyzer_results=results)
 
-    # Extract the anonymized text from the EngineResult object
     anonymized_text = anonymized_result.text
 
     pii_pattern = r"<(.*?)>"

app/celery/tasks.py

@@ -1,12 +1,9 @@
-from app.celery.userDataComputation.activityClassification import (
+from app.celery.compute.classifier import (
     get_most_relevant_activity,
     get_most_relevant_activity_for_batch,
 )
-from ..core.controllers.history import *
 from ..core.models.history import *
 from ..core.models.user import *
-from ..core.models.celery_tasks import *
-from ..core.models.visits import *
 from ..core.modules.upload import upload_to_arweave, prepare_history_json
 import redis
 from celery import shared_task
@@ -50,46 +47,6 @@ def update_user_graph_cache(self, userAddress):
 
 
 
-def send_telegram_message(slug, body):
-    tg_token_api = os.environ.get("TELEGRAM_API_TOKEN")
-    channel_id = "-1002178791722"  # The channel ID you provided
-
-    subject = f"Activity for Slug: {slug}"
-    message = f"```{json.dumps(body, indent=2)}```"
-
-    # Send the message
-    telegram_api_url = f"https://api.telegram.org/bot{tg_token_api}/sendMessage"
-
-    payload = {
-        "chat_id": channel_id,
-        "text": f"{subject}\n\n{message}",
-        "parse_mode": "Markdown",
-    }
-
-    try:
-        response = requests.post(telegram_api_url, json=payload)
-        if response.status_code == 200:
-            print(f"Telegram message sent successfully for slug: {slug}")
-        else:
-            print(
-                f"Failed to send Telegram message for slug: {slug}. Status code: {response.status_code}"
-            )
-    except Exception as e:
-        print(f"Failed to send Telegram message for slug: {slug}. Error: {str(e)}")
-
-
-@shared_task(
-    bind=True,
-    base=AbortableTask,
-    ack_later=True,
-    default_retry_delay=20,
-    max_retries=2,
-    queue="send-email",
-)
-def send_telegram_notification(self, slug, response):
-    send_telegram_message(slug, response)
-
-
 @shared_task(
     bind=True,
     base=AbortableTask,
@@ -110,9 +67,7 @@ def contextual_activity_classification(self, item, address):
     if not user:
         print(f"User with address {address} not found")
         return
-
-
-    print(item)
+
     activity_json = get_activity_json(address)
     history_entry = History(
         address=address,
@@ -192,16 +147,13 @@ def contextual_activity_classification_for_batch(self, history_batch, address):
             print(f"User with address {address} not found")
             continue
 
-        # Create a new History entry
         history_entry = History(
             address=address,
             url=item["url"],
             title=item.get("title", "No Title Available"),
             visitTime=float(item.get("lastVisitTime", datetime.now().timestamp())),
             category=activity,
         )
-
-        # Save the history entry to the database
         history_entry.save()
 
         print(