mirror of
https://github.com/artemium428/tvsignals-to-tg.git
synced 2026-09-15 17:16:21 +00:00
59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
from app.config import Settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TELEGRAM_API = "https://api.telegram.org"
|
|
|
|
|
|
class TelegramError(Exception):
|
|
pass
|
|
|
|
|
|
async def send_photo(
|
|
settings: Settings,
|
|
*,
|
|
photo: bytes,
|
|
caption: str,
|
|
filename: str = "setup.png",
|
|
) -> dict:
|
|
url = f"{TELEGRAM_API}/bot{settings.telegram_bot_token}/sendPhoto"
|
|
data = {
|
|
"chat_id": settings.telegram_chat_id,
|
|
"message_thread_id": str(settings.telegram_message_thread_id),
|
|
"caption": caption,
|
|
"parse_mode": "HTML",
|
|
}
|
|
files = {"photo": (filename, photo, "image/png")}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(url, data=data, files=files)
|
|
payload = response.json()
|
|
if response.status_code >= 400 or not payload.get("ok"):
|
|
raise TelegramError(f"sendPhoto failed: {payload}")
|
|
return payload
|
|
|
|
|
|
async def send_message(
|
|
settings: Settings,
|
|
*,
|
|
text: str,
|
|
) -> dict:
|
|
url = f"{TELEGRAM_API}/bot{settings.telegram_bot_token}/sendMessage"
|
|
data = {
|
|
"chat_id": settings.telegram_chat_id,
|
|
"message_thread_id": str(settings.telegram_message_thread_id),
|
|
"text": text,
|
|
"parse_mode": "HTML",
|
|
"disable_web_page_preview": True,
|
|
}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(url, data=data)
|
|
payload = response.json()
|
|
if response.status_code >= 400 or not payload.get("ok"):
|
|
raise TelegramError(f"sendMessage failed: {payload}")
|
|
return payload
|