import logging
from datetime import UTC, datetime
from html import escape
import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import settings
from app.models.car import AdminNotification
logger = logging.getLogger(__name__)
ADMIN_EVENT_FLAGS = {
"user_registered": "admin_notify_new_users",
"vehicle_created": "admin_notify_new_users",
"first_record_created": "admin_notify_new_users",
"sto_application_created": "admin_notify_sto_applications",
"sto_application_updated": "admin_notify_sto_applications",
"sto_approved": "admin_notify_sto_applications",
"sto_suspended": "admin_notify_sto_applications",
"security_event": "admin_notify_security_events",
"rate_limit_exceeded": "admin_notify_security_events",
"upload_blocked": "admin_notify_security_events",
"system_error": "admin_notify_system_errors",
"ocr_failed": "admin_notify_system_errors",
}
def admin_event_enabled(event_type: str) -> bool:
flag = ADMIN_EVENT_FLAGS.get(event_type)
return bool(getattr(settings, flag, True)) if flag else True
def admin_recipients() -> list[str]:
recipients: list[str] = []
if settings.admin_notification_chat_id:
recipients.append(settings.admin_notification_chat_id)
recipients.extend(str(item) for item in settings.admin_telegram_id_list)
return list(dict.fromkeys(recipients))
def admin_notification_url(entity_type: str | None = None, entity_id: str | int | None = None) -> str:
base = settings.effective_webapp_url
if entity_type == "service_center" and entity_id:
return f"{base}/admin.html?section=sto-applications&entity_id={entity_id}"
if entity_type == "user" and entity_id:
return f"{base}/admin.html?section=users&entity_id={entity_id}"
return f"{base}/admin.html"
async def create_admin_notification(
session: AsyncSession,
*,
event_type: str,
title: str,
body: str | None = None,
entity_type: str | None = None,
entity_id: int | str | None = None,
severity: str = "info",
idempotency_key: str | None = None,
metadata: dict | None = None,
send_telegram: bool = True,
) -> AdminNotification:
key = idempotency_key or f"{event_type}:{entity_type or 'system'}:{entity_id or title}"
existing = (
await session.execute(select(AdminNotification).where(AdminNotification.idempotency_key == key))
).scalar_one_or_none()
if existing:
return existing
notification = AdminNotification(
event_type=event_type,
title=title,
body=body,
entity_type=entity_type,
entity_id=str(entity_id) if entity_id is not None else None,
severity=severity,
idempotency_key=key,
metadata_json=metadata,
telegram_status="pending" if send_telegram else "skipped",
)
session.add(notification)
await session.flush()
if send_telegram and admin_event_enabled(event_type):
await send_admin_telegram_notification(notification)
elif send_telegram:
notification.telegram_status = "skipped"
return notification
async def send_admin_telegram_notification(notification: AdminNotification) -> None:
recipients = admin_recipients()
if not recipients or not settings.bot_token:
notification.telegram_status = "skipped"
return
link = admin_notification_url(notification.entity_type, notification.entity_id)
text = "\n".join(
item
for item in [
f"{escape(notification.title)}",
escape(notification.body or ""),
f"Событие: {escape(notification.event_type)}",
f"Открыть: {escape(link)}",
]
if item
)
errors: list[str] = []
async with httpx.AsyncClient(timeout=8) as client:
for chat_id in recipients:
try:
response = await client.post(
f"https://api.telegram.org/bot{settings.bot_token}/sendMessage",
json={
"chat_id": chat_id,
"text": text,
"parse_mode": "HTML",
"disable_web_page_preview": True,
},
)
response.raise_for_status()
except Exception as error: # noqa: BLE001 - notification delivery is best-effort
logger.warning("Admin Telegram notification failed: %s", error)
errors.append(str(error))
if errors:
notification.telegram_status = "failed"
notification.telegram_error = "; ".join(errors)[:2000]
else:
notification.telegram_status = "sent"
async def mark_admin_notification_read(
session: AsyncSession, notification: AdminNotification
) -> AdminNotification:
notification.status = "read"
notification.read_at = datetime.now(UTC)
await session.flush()
return notification
async def dismiss_admin_notification(
session: AsyncSession, notification: AdminNotification
) -> AdminNotification:
notification.status = "dismissed"
notification.dismissed_at = datetime.now(UTC)
await session.flush()
return notification