logging_bot / telegram_preview.py
AstraOS's picture
Create telegram_preview.py
8dd998f verified
raw
history blame
4.85 kB
"""
Reusable Telegram‑bot preview router.
Exposes:
β€’ GET / – HTML preview card
β€’ GET /avatar.jpg – bot avatar (streamed, in‑memory)
To use: `from telegram_preview import router` and include it in your FastAPI app.
Compatible with PythonΒ 3.8+ (uses typing.Optional, no PEPβ€―604 syntax).
"""
import io
import os
import requests
from typing import Optional
from fastapi import APIRouter, Response
from fastapi.responses import HTMLResponse, StreamingResponse
# ─── Environment ───────────────────────────────────────────────────────────────
TOKEN = os.getenv("BOT_TOKEN") # required
BOT_USERNAME = os.getenv("BOT_USERNAME", "python3463_bot")
FALLBACK_IMG = "https://telegram.org/img/t_logo.png" # fallback logo URL
if not TOKEN:
raise RuntimeError("TOKEN environment variable not set")
API_ROOT = f"https://api.telegram.org/bot{TOKEN}"
router = APIRouter()
# ─── Helpers ───────────────────────────────────────────────────────────────────
def _get_description() -> str:
"""Try short β†’ full β†’ empty string."""
try:
r = requests.get(f"{API_ROOT}/getMyShortDescription", timeout=5).json()
if (desc := r["result"].get("short_description")):
return desc
except Exception:
pass
try:
r = requests.get(f"{API_ROOT}/getMyDescription", timeout=5).json()
return r["result"].get("description", "")
except Exception:
return ""
def _fetch_avatar_bytes() -> Optional[bytes]:
"""Return the newest avatar as bytes, or None if missing/error."""
try:
me = requests.get(f"{API_ROOT}/getMe", timeout=5).json()
user_id = me["result"]["id"]
photos = requests.get(
f"{API_ROOT}/getUserProfilePhotos",
params={"user_id": user_id, "limit": 1},
timeout=5,
).json()
if photos["result"]["total_count"] == 0:
return None
file_id = photos["result"]["photos"][0][-1]["file_id"]
file_obj = requests.get(
f"{API_ROOT}/getFile", params={"file_id": file_id}, timeout=5
).json()
file_path = file_obj["result"]["file_path"]
resp = requests.get(
f"https://api.telegram.org/file/bot{TOKEN}/{file_path}", timeout=10
)
if resp.status_code == 200:
return resp.content
except Exception as exc:
print("Avatar fetch error:", exc)
return None
# ─── Routes ────────────────────────────────────────────────────────────────────
@router.get("/", include_in_schema=False, response_class=HTMLResponse)
def bot_preview():
description = _get_description()
html = f"""
<html>
<head>
<title>@{BOT_USERNAME}</title>
<style>
body {{
font-family: sans-serif;
display: flex;
justify-content: center;
padding: 40px;
background: #f7f7f7;
}}
.card {{
max-width: 420px;
background: #fff;
padding: 24px;
text-align: center;
border-radius: 12px;
box-shadow: 0 4px 12px rgba(0,0,0,.1);
}}
.avatar {{
width: 120px;
height: 120px;
border-radius: 50%;
object-fit: cover;
background: #ddd;
}}
.btn {{
display: inline-block;
margin-top: 16px;
padding: 12px 24px;
background: #2AABEE;
color: #fff;
border-radius: 8px;
text-decoration: none;
font-weight: bold;
}}
</style>
</head>
<body>
<div class="card">
<img src="avatar.jpg" class="avatar" alt="avatar">
<h2>@{BOT_USERNAME}</h2>
<p>{description}</p>
<a class="btn" href="https://t.me/{BOT_USERNAME}" target="_blank">
StartΒ Bot
</a>
</div>
</body>
</html>
"""
return HTMLResponse(html)
@router.get("/avatar.jpg", include_in_schema=False)
def serve_avatar():
"""Streams avatar JPEG or Telegram logo (PNG) as fallback."""
data = _fetch_avatar_bytes()
if data:
return StreamingResponse(io.BytesIO(data), media_type="image/jpeg")
fallback = requests.get(FALLBACK_IMG, timeout=10)
return Response(content=fallback.content, media_type="image/png")