Spaces:
Running
Running
File size: 4,759 Bytes
8dd998f 516b9df 8dd998f 6805cfc 8dd998f 516b9df 8dd998f 516b9df 64931bf 516b9df eb4d318 37f6fb1 eb4d318 64931bf 8dd998f 516b9df 8dd998f 516b9df 8dd998f 516b9df 8dd998f 516b9df 8dd998f 516b9df 8dd998f |
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 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
"""
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.
"""
import io
import os
import requests
from pathlib import Path
from typing import Optional
from fastapi import APIRouter, Response, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles # NEW
# ─── Locate templates folder absolute path ────────────────────────────────────
BASE_DIR = Path(__file__).resolve().parent
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
def include_in_app(app):
"""
Helper so the main application can pull in both router *and* static mount.
"""
app.include_router(router)
app.mount(
"/static",
StaticFiles(directory=str(BASE_DIR / "static")),
name="static",
)
# ─── Environment ───────────────────────────────────────────────────────────────
TOKEN = os.getenv("BOT_TOKEN")
FALLBACK_IMG = "https://telegram.org/img/t_logo.png"
if not TOKEN:
raise RuntimeError("BOT_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
def _get_bot_identity() -> dict:
"""Return {'first_name': ..., 'username': ...} from getMe"""
try:
r = requests.get(f"{API_ROOT}/getMe", timeout=5).json()
return {
"first_name": r["result"].get("first_name", "Telegram Bot"),
"username": r["result"].get("username", "unknown_bot"),
}
except Exception:
return {"first_name": "Telegram Bot", "username": "unknown_bot"}
# ─── Routes ────────────────────────────────────────────────────────────────────
@router.get("/", include_in_schema=False, response_class=HTMLResponse)
def bot_preview(request: Request):
description = _get_description()
identity = _get_bot_identity()
return templates.TemplateResponse(
"preview.html",
{
"request": request,
"title": f"@{identity['username']}",
"username": identity["username"],
"first_name": identity["first_name"],
"description": description,
},
)
@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")
|