shihaam
6cdf042f49
Build and Push Docker Images / Build and Push Docker Images (push) Failing after 8s
108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
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=""))
|
|
# Optional forum topic to post into. If empty, messages go to the group's General topic.
|
|
topic_id = str(config("TG_TOPIC_ID", cast=str, default=""))
|
|
|
|
|
|
def format_mobile(mobile: str) -> str:
|
|
"""Normalize a Maldives mobile number to E.164 (+960XXXXXXX)."""
|
|
number = str(mobile).strip().replace(" ", "")
|
|
if number.startswith("+"):
|
|
return number
|
|
if number.startswith("960"):
|
|
return "+" + number
|
|
return "+960" + number
|
|
|
|
|
|
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",
|
|
"X-API-Key": api_key,
|
|
}
|
|
data = {
|
|
"to": format_mobile(mobile),
|
|
"text": message,
|
|
}
|
|
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",
|
|
"X-API-Key": api_key,
|
|
}
|
|
data = {
|
|
"to": format_mobile(mobile),
|
|
"text": message,
|
|
}
|
|
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 normalize_whitespace(text: str) -> str:
|
|
return "\n".join(line.strip() for line in text.strip().splitlines())
|
|
|
|
|
|
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"}
|
|
if topic_id:
|
|
payload["message_thread_id"] = topic_id
|
|
|
|
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)}
|
|
|
|
|
|
def send_clean_telegram_markdown(message: str):
|
|
text = normalize_whitespace(message)
|
|
escaped = escape_markdown_v2(text)
|
|
return send_telegram_markdown(escaped)
|