Spaces:
Sleeping
Sleeping
File size: 1,645 Bytes
b6154b2 9218640 b6154b2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | from __future__ import annotations
import httpx
from ..config import settings
def _api_url(path: str) -> str:
return f"https://api.telegram.org/bot{settings.telegram_bot_token}/{path}"
async def send_message(chat_id: str | int, text: str) -> None:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.post(_api_url("sendMessage"), json={"chat_id": chat_id, "text": text})
response.raise_for_status()
async def get_file_url(file_id: str) -> str:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.get(_api_url("getFile"), params={"file_id": file_id})
response.raise_for_status()
file_path = response.json()["result"]["file_path"]
return f"https://api.telegram.org/file/bot{settings.telegram_bot_token}/{file_path}"
async def download_file(file_url: str) -> bytes:
async with httpx.AsyncClient(timeout=120) as client:
response = await client.get(file_url)
response.raise_for_status()
return response.content
def guess_photo_media_type(file_path: str | None = None) -> str:
if file_path and file_path.lower().endswith(".png"):
return "image/png"
return "image/jpeg"
async def set_webhook() -> dict:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.post(
_api_url("setWebhook"),
json={
"url": f"{settings.app_base_url}/telegram/webhook",
"secret_token": settings.telegram_webhook_secret or None,
},
)
response.raise_for_status()
return response.json()
|