Feature/telegram bot

brightIDfaucet/settings.py

@@ -71,6 +71,9 @@ def str2bool(v):
 MEMCACHED_USERNAME = os.environ.get("MEMCACHEDCLOUD_USERNAME")
 MEMCACHED_PASSWORD = os.environ.get("MEMCACHEDCLOUD_PASSWORD")
 DEPLOYMENT_ENV = os.environ.get("DEPLOYMENT_ENV")
+TELEGRAM_CHANNEL_ID = os.environ.get("TELEGRAM_CHANNEL_ID")
+TELEGRAM_API_TOKEN = os.environ.get("TELEGRAM_API_TOKEN")
+
 
 assert DEPLOYMENT_ENV in ["dev", "main"]
 
@@ -138,6 +141,9 @@ def before_send(event, hint):
     "django.contrib.auth.middleware.AuthenticationMiddleware",
     "django.contrib.messages.middleware.MessageMiddleware",
     "django.middleware.clickjacking.XFrameOptionsMiddleware",
+    "core.telegram.LogMiddleware"
+
+
 ]
 
 ROOT_URLCONF = "brightIDfaucet.urls"
@@ -259,3 +265,5 @@ def before_send(event, hint):
     ),
 }
 CELERY_BROKER_URL = REDIS_URL
+
+TELEGRAM_MIN_LOG_INTERVAL = 10 # seconds

core/telegram.py

@@ -0,0 +1,49 @@
+import requests
+import time
+from collections import defaultdict
+from django.utils.deprecation import MiddlewareMixin
+from django.conf import settings
+
+API_TOKEN = settings.TELEGRAM_API_TOKEN
+CHANNEL_ID = settings.TELEGRAM_CHANNEL_ID
+MIN_INTERVAL = settings.TELEGRAM_MIN_LOG_INTERVAL
+
+
+BASE_URL = f"https://api.telegram.org/bot{API_TOKEN}"
+
+
+def send_telegram_log(text):
+    url = f"{BASE_URL}/sendMessage"
+    payload = {
+        "chat_id": CHANNEL_ID,
+        "text": text,
+        "parse_mode": "MarkdownV2",
+        "disable_web_page_preview": True,
+    }
+    try:
+        response = requests.post(url, data=payload)
+        return {"ok":True}
+    except Exception as e:
+        return {"ok": False}
+
+
+log_cache = defaultdict(int)
+
+
+
+class LogMiddleware(MiddlewareMixin):
+    def log_message(self, message):
+        log_hash = hash(message)
+        current_time = time.time()
+
+        # Check if the log has been sent before
+        if current_time - log_cache[log_hash] > MIN_INTERVAL:
+            log_cache[log_hash] = current_time
+            return self.send_to_telegram(message)
+
+        return {"ok": False}
+
+    def send_to_telegram(self, message):
+        return send_telegram_log(message)
+
+

core/tests.py

@@ -1,3 +1,4 @@
+import time
 from unittest.mock import PropertyMock, patch
 
 from django.contrib.auth.models import User
@@ -6,6 +7,12 @@
 from authentication.models import GitcoinPassportConnection, UserProfile, Wallet
 from core.models import Chain, NetworkTypes, WalletAccount
 
+from django.test import TestCase
+from core.telegram import LogMiddleware
+from django.conf import settings
+
+TELEGRAM_MIN_LOG_INTERVAL = settings.TELEGRAM_MIN_LOG_INTERVAL
+
 from .constraints import (
     Attest,
     BeAttestedBy,
@@ -363,3 +370,28 @@ def test_gitcoin_passport_connection_constraint_fail(self):
         constraint = HasGitcoinPassportProfile(self.not_connected_user_profile)
 
         self.assertEqual(constraint.is_observed(), False)
+
+
+
+
+
+class LogMiddlewareTests(TestCase):
+    def setUp(self):
+        self.middleware = LogMiddleware(get_response=lambda request: None)
+
+
+    def test_log_message_sent_to_telegram(self):
+
+        res = self.middleware.log_message("Test log message")
+        self.assertEqual(res['ok'],  True)
+        res = self.middleware.log_message("Test log message")
+        self.assertEqual(res['ok'],  False)
+        # delay
+        time.sleep(TELEGRAM_MIN_LOG_INTERVAL)
+        res = self.middleware.log_message("Test log message")
+        self.assertEqual(res['ok'],  True)
+        res = self.middleware.log_message("Test log message")
+        self.assertEqual(res['ok'],  False)
+        res = self.middleware.log_message("Test log message2")
+        self.assertEqual(res['ok'],  True)
+

sample.env

@@ -7,3 +7,5 @@ BRIGHT_PRIVATE_KEY=""
 DEBUG="True"
 SENTRY_DSN="DEBUG-DSN"
 DEPLOYMENT_ENV="env"
+TELEGRAM_API_TOKEN=""
+TELEGRAM_CHANNEL_ID=""