Add SMS and Telegram notification functions to handle user verification messages

api/notifications.py

@@ -0,0 +1,85 @@
+from decouple import config
+import requests
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+api_url = str(config("SMS_API_URL", cast=str, default=""))
+api_key = str(config("SMS_API_KEY", cast=str, default=""))
+bot_token = str(config("TG_BOT_TOKEN", cast=str, default=""))
+chat_id = str(config("TG_CHAT_ID", cast=str, default=""))
+
+
+def send_otp(mobile: str, message: str):
+    if not api_url or not api_key:
+        logger.debug("Failed to send SMS. Missing SMS_API_URL or SMS_API_KEY.")
+        return False
+
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}",
+    }
+    data = {
+        "number": mobile,
+        "message": message,
+        "check_delivery": False,
+    }
+    response = requests.post(api_url, headers=headers, data=json.dumps(data))
+    if response.status_code == 200:
+        return True
+    else:
+        logger.debug(f"Failed to send SMS. Status code: {response.status_code}")
+        return False
+
+
+def send_sms(mobile: str, message: str):
+    logger.info(f"Sending SMS to {mobile}")
+    try:
+        if not api_url or not api_key:
+            logger.debug("Failed to send SMS. Missing SMS_API_URL or SMS_API_KEY.")
+            return False
+
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {api_key}",
+        }
+        data = {
+            "number": mobile,
+            "message": message,
+            "check_delivery": False,
+        }
+        response = requests.post(api_url, headers=headers, data=json.dumps(data))
+        if response.status_code == 200:
+            return True
+        else:
+            logger.debug(f"Failed to send SMS. Status code: {response.status_code}")
+            return False
+    except requests.exceptions.RequestException as e:
+        logger.debug(f"Failed to send SMS. Error: {e}")
+        return False
+
+
+# def escape_markdown_v2(text: str) -> str:
+#     special_chars = r"\_*[]()~`>#+-=|{}.!"
+#     return "".join(["\\" + c if c in special_chars else c for c in text])
+
+
+def send_telegram_markdown(message: str):
+    """
+    Sends a MarkdownV2-formatted message to a Telegram chat.
+
+    Parameters:
+        bot_token (str): Your Telegram bot token
+        chat_id (str): Target chat ID (e.g., "-1001234567890" for channels/groups)
+        message (str): The message content, already escaped for MarkdownV2
+    """
+    try:
+        url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
+        payload = {"chat_id": chat_id, "text": message, "parse_mode": "MarkdownV2"}
+
+        response = requests.post(url, data=payload)
+        response.raise_for_status()
+        return response.json()
+    except requests.RequestException as e:
+        logger.error(f"Error sending Telegram message: {e}")
+        return {"error": str(e)}