import requests from typing import Dict, Optional from fastapi import HTTPException import logging logger = logging.getLogger(__name__) class TelegramBot: def __init__(self, bot_token: str): self.bot_token = bot_token self.base_url = f"https://api.telegram.org/bot{bot_token}" def send_message(self, chat_id: str, text: str, parse_mode: str = "HTML") -> Dict: """ Отправить текстовое сообщение в Telegram Args: chat_id: ID чата или канала text: Текст сообщения parse_mode: Режим парсинга ('HTML' или 'MarkdownV2') Returns: Ответ от Telegram API """ url = f"{self.base_url}/sendMessage" payload = { "chat_id": chat_id, "text": text, "parse_mode": parse_mode, "disable_web_page_preview": True } try: response = requests.post(url, json=payload) response.raise_for_status() result = response.json() if not result.get("ok"): raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}") return result except requests.exceptions.RequestException as e: logger.error(f"Error sending message to Telegram: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to send message to Telegram: {str(e)}") def send_photo(self, chat_id: str, photo_url: str, caption: str = "", parse_mode: str = "HTML") -> Dict: """ Отправить фото в Telegram Args: chat_id: ID чата или канала photo_url: URL фото caption: Подпись к фото parse_mode: Режим парсинга ('HTML' или 'MarkdownV2') Returns: Ответ от Telegram API """ url = f"{self.base_url}/sendPhoto" payload = { "chat_id": chat_id, "photo": photo_url, "caption": caption, "parse_mode": parse_mode } try: response = requests.post(url, json=payload) response.raise_for_status() result = response.json() if not result.get("ok"): raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}") return result except requests.exceptions.RequestException as e: logger.error(f"Error sending photo to Telegram: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to send photo to Telegram: {str(e)}") def get_me(self) -> Dict: """ Получить информацию о боте Returns: Информация о боте от Telegram API """ url = f"{self.base_url}/getMe" try: response = requests.get(url) response.raise_for_status() result = response.json() if not result.get("ok"): raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}") return result except requests.exceptions.RequestException as e: logger.error(f"Error getting bot info from Telegram: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to get bot info from Telegram: {str(e)}") def get_chat(self, chat_id: str) -> Dict: """ Получить информацию о чате Args: chat_id: ID чата или канала Returns: Информация о чате от Telegram API """ url = f"{self.base_url}/getChat" payload = {"chat_id": chat_id} try: response = requests.post(url, json=payload) response.raise_for_status() result = response.json() if not result.get("ok"): raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}") return result except requests.exceptions.RequestException as e: logger.error(f"Error getting chat info from Telegram: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to get chat info from Telegram: {str(e)}")