import asyncio import html import sys from pathlib import Path from typing import Optional _ROOT = Path(__file__).resolve().parent.parent if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) 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.filters import CommandStart from aiogram.types import CopyTextButton, InlineKeyboardButton, InlineKeyboardMarkup, Message from server.config import BOT_TOKEN, PANEL_BOT_TOKEN, SSL_CERTFILE, SSL_KEYFILE from api.db_api import get_device, resolve_sms_destination, set_sms_chat_id from utils.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 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=(PANEL_BOT_TOKEN or BOT_TOKEN).strip(), default=DefaultBotProperties(parse_mode=ParseMode.HTML), ) dp = Dispatcher() @dp.message(CommandStart()) async def cmd_start(message: Message) -> None: user_id = message.from_user.id if message.from_user else 0 await message.answer( "OZ Panel будет присылать сюда коды и SMS с ваших личных кабинетов.\n" f"Ваш Telegram ID: {user_id}\n\n" "Чтобы слать SMS в другой чат, укажите его ID в настройках панели и добавьте туда этого бота." ) @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 account = get_device(device_name) or {} owner_id = account.get("tg_id") try: owner_id = int(owner_id) if owner_id not in (None, "", 0, "0") else None except (TypeError, ValueError): owner_id = None if owner_id is None: raise HTTPException( status_code=404, detail=f"У устройства '{device_name}' нет владельца — SMS некуда отправить", ) chat_id, custom = resolve_sms_destination(owner_id) try: message_text = payload.notification_text except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc parsed = parse_message(message_text) formatted_text, reply_markup = format_notification_message( message_text, account_name=account.get("name"), balance=account.get("balance"), ) try: await bot.send_message( chat_id=chat_id, text=f"💻 {html.escape(device_name)}\n\n{formatted_text}", reply_markup=reply_markup, ) except Exception as exc: if custom: set_sms_chat_id(owner_id, None) try: await bot.send_message( chat_id=owner_id, text=f"💻 {html.escape(device_name)}\n\n{formatted_text}", reply_markup=reply_markup, ) chat_id = owner_id except Exception: raise HTTPException( status_code=502, detail="Чат отсутствует или бот в него не добавлен", ) from exc else: raise HTTPException( status_code=502, detail="Чат отсутствует или бот в него не добавлен", ) from exc 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("server.notify_server:app", **uvicorn_kwargs)