mirror of
https://gitverse.ru/kpa39l/telepost.git
synced 2026-09-29 09:55:09 +00:00
127 lines
4.6 KiB
Python
127 lines
4.6 KiB
Python
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)}") |