drivers_bot/app/services/admin_notifications.py

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"<b>{escape(notification.title)}</b>",
            escape(notification.body or ""),
            f"Событие: <code>{escape(notification.event_type)}</code>",
            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 retry_admin_telegram_notifications(session: AsyncSession, *, limit: int = 50) -> int:
    result = await session.execute(
        select(AdminNotification)
        .where(AdminNotification.telegram_status.in_(["pending", "failed"]))
        .order_by(AdminNotification.created_at.asc())
        .limit(limit)
    )
    delivered = 0
    for notification in result.scalars():
        await send_admin_telegram_notification(notification)
        if notification.telegram_status == "sent":
            delivered += 1
    await session.commit()
    return delivered


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