mirror of
https://github.com/i701/sarlink-portal-api.git
synced 2025-06-06 17:36:20 +00:00
i701
4db9d7fabd
All checks were successful
Build and Push Docker Images / Build and Push Docker Images (push) Successful in 4m18s
96 lines
3.0 KiB
Python
96 lines
3.0 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=""))
|
|
|
|
|
|
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 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"}
|
|
|
|
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)
|