"""Telegram webhook handler for bot commands.""" from __future__ import annotations import logging from typing import Any import aiohttp from fastapi import APIRouter, Depends, Header, HTTPException, Request from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession from notify_bridge_core.notifications.telegram.media import TELEGRAM_API_BASE_URL from ..database.engine import get_session from ..database.models import TelegramBot from ..services.telegram import save_chat_from_webhook from .handler import handle_command, send_media_group, send_reply _LOGGER = logging.getLogger(__name__) router = APIRouter(prefix="/api/telegram", tags=["telegram-webhook"]) # Webhook secret — set via NOTIFY_BRIDGE_TELEGRAM_WEBHOOK_SECRET env var _webhook_secret: str | None = None def set_webhook_secret(secret: str | None) -> None: global _webhook_secret _webhook_secret = secret @router.post("/webhook/{webhook_id}") async def telegram_webhook( webhook_id: str, request: Request, x_telegram_bot_api_secret_token: str | None = Header(default=None), session: AsyncSession = Depends(get_session), ): """Handle incoming Telegram messages — route commands to handlers.""" # Validate webhook secret if configured if _webhook_secret: if x_telegram_bot_api_secret_token != _webhook_secret: raise HTTPException(status_code=403, detail="Invalid webhook secret") # Find bot by opaque webhook path ID (not by token — token must not appear in URLs) bot_result = await session.exec( select(TelegramBot).where(TelegramBot.webhook_path_id == webhook_id) ) bot = bot_result.first() if not bot: raise HTTPException(status_code=403, detail="Unknown webhook") try: update = await request.json() except Exception: return {"ok": True, "error": "invalid_json"} message = update.get("message") if not message: return {"ok": True, "skipped": "no_message"} chat_info = message.get("chat", {}) chat_id = str(chat_info.get("id", "")) text = message.get("text", "") if not chat_id or not text: return {"ok": True, "skipped": "empty"} # Auto-persist chat from incoming message try: await save_chat_from_webhook(session, bot.id, chat_info) await session.commit() except Exception: _LOGGER.warning("Failed to auto-save chat %s", chat_id, exc_info=True) # Handle commands if text.startswith("/"): language_code = message.get("from", {}).get("language_code", "") cmd_response = await handle_command(bot, chat_id, text, language_code=language_code) if cmd_response is not None: if isinstance(cmd_response, list): await send_media_group(bot.token, chat_id, cmd_response) else: await send_reply(bot.token, chat_id, cmd_response) return {"ok": True} return {"ok": True, "skipped": "not_a_command"} async def register_webhook(bot_token: str, webhook_url: str, secret: str | None = None) -> dict: """Register webhook URL with Telegram Bot API.""" async with aiohttp.ClientSession() as http: url = f"{TELEGRAM_API_BASE_URL}{bot_token}/setWebhook" payload: dict[str, Any] = {"url": webhook_url} if secret: payload["secret_token"] = secret try: async with http.post(url, json=payload) as resp: result = await resp.json() if result.get("ok"): return {"success": True} return {"success": False, "error": result.get("description")} except aiohttp.ClientError as err: return {"success": False, "error": str(err)} async def unregister_webhook(bot_token: str) -> dict: """Remove webhook from Telegram Bot API.""" async with aiohttp.ClientSession() as http: url = f"{TELEGRAM_API_BASE_URL}{bot_token}/deleteWebhook" try: async with http.post(url) as resp: result = await resp.json() return {"success": result.get("ok", False)} except aiohttp.ClientError as err: return {"success": False, "error": str(err)}