import requests from typing import Dict, Any, Optional from app.providers.base import BaseProvider from app.core.config import settings import logging logger = logging.getLogger(__name__) class TelegramProvider(BaseProvider): """ Telegram Bot API provider for testing. Setup: 1. Talk to @BotFather on Telegram 2. Create new bot with /newbot 3. Copy the bot token 4. Set TELEGRAM_BOT_TOKEN in .env 5. Start a chat with your bot 6. Use your Telegram user ID as the phone number in contacts To get your Telegram user ID: - Message @userinfobot on Telegram - OR use https://api.telegram.org/bot/getUpdates after messaging your bot """ def __init__(self): self.bot_token = settings.TELEGRAM_BOT_TOKEN self.base_url = f"https://api.telegram.org/bot{self.bot_token}" def send_message( self, to: str, template_name: Optional[str], template_body: str, variables: Dict[str, str], language: str = "en" ) -> str: """ Send a message via Telegram Bot API. Args: to: Telegram user ID or chat ID (stored in phone_e164 field) template_name: Ignored for Telegram (no template system) template_body: Message text with variables like {{first_name}} variables: Dict of variables to replace in template_body language: Ignored for Telegram Returns: message_id: Telegram message ID as string """ # Replace variables in template message_text = template_body for key, value in variables.items(): message_text = message_text.replace(f"{{{{{key}}}}}", value) # Prepare request url = f"{self.base_url}/sendMessage" payload = { "chat_id": to, "text": message_text, "parse_mode": "HTML" # Support basic formatting } logger.info(f"Sending Telegram message to chat_id: {to}") try: response = requests.post(url, json=payload, timeout=10) # Get response body for debugging result = response.json() if not response.ok or not result.get("ok"): error_desc = result.get("description", "Unknown error") error_code = result.get("error_code", 0) logger.error(f"Telegram API error {error_code}: {error_desc}") logger.error(f"Request payload: {payload}") raise Exception(f"Telegram API error {error_code}: {error_desc}") message_id = str(result["result"]["message_id"]) logger.info(f"Telegram message sent successfully. Message ID: {message_id}") return message_id except requests.exceptions.RequestException as e: logger.error(f"Telegram API request failed: {str(e)}") raise Exception(f"Failed to send Telegram message: {str(e)}") def get_provider_name(self) -> str: return "telegram"