import asyncio import html from typing import Optional from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel, Field from aiogram import Bot, Dispatcher from aiogram.enums import ParseMode from aiogram.client.default import DefaultBotProperties from aiogram.exceptions import TelegramAPIError, TelegramBadRequest, TelegramForbiddenError from aiogram.types import CopyTextButton, InlineKeyboardButton, InlineKeyboardMarkup from config import BOT_TOKEN, DEVICE_CHAT_MAP, NOTIFY_BOT_TOKEN, SSL_CERTFILE, SSL_KEYFILE from db_api import get_device from pars import parse_message def _format_db_balance(balance) -> str: if balance is None or balance == "": return "—" try: value = float(balance) except (TypeError, ValueError): return html.escape(str(balance)) if value.is_integer(): formatted = f"{int(value):,}".replace(",", " ") else: formatted = f"{value:,.2f}".replace(",", " ") return f"{formatted} ₽" def format_notification_message( message: str, *, account_name: Optional[str] = None, balance=None, ) -> tuple[str, Optional[InlineKeyboardMarkup]]: parsed = parse_message(message) code = parsed.get("code") amount = parsed.get("amount") service = parsed.get("service") lines = [] if code: lines.append(f"❗️ Код: {code}") lk_name = (account_name or "").strip() or "—" lines.append(f"👤 ЛК: {html.escape(lk_name)}") lines.append( f"👛 Баланс: {_format_db_balance(balance)}" ) if amount: lines.append(f"👛 Сумма: {amount} ₽") if service: lines.append(f"🌐 Сервис: {service}") lines.append("") lines.append("📄 Полный текст уведомления:") lines.append(html.escape(message)) text = "\n".join(lines) markup = None if code: markup = InlineKeyboardMarkup( inline_keyboard=[ [ InlineKeyboardButton( text="Скопировать код", copy_text=CopyTextButton(text=code), icon_custom_emoji_id="5985774024968379294", style="success", ) ] ] ) return text, markup def _telegram_error_status(exc: TelegramAPIError) -> int: text = str(exc).lower() if isinstance(exc, (TelegramBadRequest, TelegramForbiddenError)) and any( marker in text for marker in ( "chat not found", "bot was kicked", "bot is not a member", "bot was blocked", "peer_id_invalid", "have no rights to send", ) ): return 422 return 502 async def send_notification( chat_id: int, text: str, reply_markup: Optional[InlineKeyboardMarkup] = None, ) -> None: try: await bot.send_message(chat_id=chat_id, text=text, reply_markup=reply_markup) except TelegramAPIError as exc: status = _telegram_error_status(exc) print(f"Telegram send failed (chat_id={chat_id}, http={status}): {exc}") raise HTTPException( status_code=status, detail=f"Не удалось отправить уведомление в Telegram (chat_id={chat_id}): {exc}", ) from exc class NotifyRequest(BaseModel): device_name: str = Field(..., min_length=1, description="Имя устройства") text: Optional[str] = Field(None, description="Текст уведомления") message: Optional[str] = Field(None, description="Альтернативное поле для текста") @property def notification_text(self) -> str: value = (self.text or self.message or "").strip() if not value: raise ValueError("Нужно передать text или message") return value app = FastAPI(title="Notify gateway") bot = Bot( token=NOTIFY_BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML), ) dp = Dispatcher() @app.on_event("startup") async def startup_event() -> None: asyncio.create_task(dp.start_polling(bot)) @app.on_event("shutdown") async def shutdown_event() -> None: await bot.session.close() @app.get("/health") async def health() -> dict: return {"status": "ok"} @app.post("/notify") @app.post("/ozpay/notify") async def notify(request: Request) -> dict: body_data = {} try: if request.headers.get("content-type", "").startswith("application/json"): body_data = await request.json() except Exception: body_data = {} device_name = ( (body_data.get("device_name") or request.query_params.get("device_name") or "").strip() ) text = ( body_data.get("text") or body_data.get("message") or request.query_params.get("text") or request.query_params.get("message") or "" ).strip() if not device_name: raise HTTPException(status_code=400, detail="Параметр device_name обязателен") if not text: raise HTTPException(status_code=400, detail="Нужно передать text или message") try: payload = NotifyRequest(device_name=device_name, text=text) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc chat_id = DEVICE_CHAT_MAP.get(device_name) if chat_id is None: raise HTTPException(status_code=404, detail=f"Устройство '{device_name}' не найдено в карте отправки") try: message_text = payload.notification_text except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc account = get_device(device_name) or {} parsed = parse_message(message_text) formatted_text, reply_markup = format_notification_message( message_text, account_name=account.get("name"), balance=account.get("balance"), ) await send_notification( chat_id, f"💻 {html.escape(device_name)}\n\n{formatted_text}", reply_markup, ) print(f"Notification sent to device '{device_name}' (chat_id: {chat_id}): {formatted_text}") return { "status": "ok", "device_name": device_name, "chat_id": chat_id, "message": message_text, "parsed": parsed, } if __name__ == "__main__": import uvicorn uvicorn_kwargs = { "host": "0.0.0.0", "port": 3000, "reload": False, } if SSL_CERTFILE and SSL_KEYFILE: uvicorn_kwargs["ssl_certfile"] = SSL_CERTFILE uvicorn_kwargs["ssl_keyfile"] = SSL_KEYFILE uvicorn.run("notify_server:app", **uvicorn_kwargs)