PTSRaiAgent/app/integrations/notify.py

"""Doctor notifications (SPEC §9): in-memory feed for the dashboard + optional Telegram doctor chat.

The alert itself is already committed to the clinical DB before `send()` is called (the dashboard reads
alerts from there via the API), so every channel here is best effort: failures are logged, never raised.
Messages carry only the level, the rule reason/code and a short pseudonymous patient id — no PII
(guardrail #8); the reason texts come from `config/rules.yaml`, not from patient messages.
"""

import asyncio
import itertools
import logging
from collections import deque
from collections.abc import Awaitable, Callable
from datetime import datetime
from typing import Any

from pydantic import BaseModel

from app.config import get_settings
from app.core import clock
from app.integrations.base import AlertLevelName

log = logging.getLogger(__name__)

FEED_SIZE = 500
SHORT_ID_LEN = 6
MAX_REASON_CHARS = 200
TELEGRAM_TIMEOUT_S = 5.0
# Doctor-facing labels (not patient texts); acute alerts are visually distinct in the chat.
LEVEL_LABELS: dict[str, str] = {
    "acute": "КРИТИЧНО",
    "high": "Высокий",
    "moderate": "Умеренный",
    "info": "Инфо",
}

ChatSender = Callable[[int, str], Awaitable[Any]]


def short_id(token: str | None) -> str:
    return token[:SHORT_ID_LEN] if token else "—"


def format_message(level: str, text: str, token: str | None) -> str:
    reason = " ".join(text.split())
    if len(reason) > MAX_REASON_CHARS:
        reason = reason[: MAX_REASON_CHARS - 1] + "…"
    return f"[{LEVEL_LABELS.get(level, level)}] {reason}\nПациент #{short_id(token)}"


class Notification(BaseModel):
    id: int
    ts: datetime
    doctor_id: str
    level: str
    text: str
    token: str | None = None


class NotificationFeed:
    """Bounded in-memory feed; ids increase monotonically so a poller asks only for `recent(since_id)`."""

    def __init__(self, size: int = FEED_SIZE) -> None:
        self._items: deque[Notification] = deque(maxlen=size)
        self._ids = itertools.count(1)

    def add(self, doctor_id: str, level: str, text: str, token: str | None, ts: datetime) -> Notification:
        item = Notification(id=next(self._ids), ts=ts, doctor_id=doctor_id, level=level, text=text, token=token)
        self._items.append(item)
        return item

    def recent(self, since_id: int = 0, doctor_id: str | None = None) -> list[Notification]:
        return [n for n in self._items if n.id > since_id and (doctor_id is None or n.doctor_id == doctor_id)]


class TelegramDoctorChat:
    """Sends to the doctor chat `DOCTOR_NOTIFY_CHAT_ID` through the bot `TG_BOT_TOKEN`.

    The aiogram `Bot` is created on the first message only; with either setting empty the channel is off.
    """

    def __init__(self, chat_id: int | None = None, bot_token: str | None = None, sender: ChatSender | None = None):
        settings = get_settings()
        self.chat_id = chat_id if chat_id is not None else settings.doctor_notify_chat_id
        self._bot_token = bot_token if bot_token is not None else settings.tg_bot_token
        self._sender = sender
        self._bot: Any = None

    @property
    def enabled(self) -> bool:
        return self.chat_id is not None and (self._sender is not None or bool(self._bot_token))

    def _get_sender(self) -> ChatSender:
        if self._sender is None:
            from app.channels.telegram import make_bot

            self._bot = make_bot(self._bot_token)
            self._sender = self._bot.send_message
        return self._sender

    async def send(self, level: str, text: str, token: str | None) -> None:
        if not self.enabled or self.chat_id is None:
            return
        try:
            await asyncio.wait_for(
                self._get_sender()(self.chat_id, format_message(level, text, token)), timeout=TELEGRAM_TIMEOUT_S
            )
        except Exception as exc:  # noqa: BLE001 — any channel error; the alert is already in the DB
            err = type(exc).__name__
            log.warning("doctor chat notify failed level=%s tok=%s error=%s", level, short_id(token), err)

    async def aclose(self) -> None:
        if self._bot is not None:
            await self._bot.session.close()
            self._bot = None
            self._sender = None


class CompositeNotifier:
    """`Notifier`: dashboard feed + Telegram doctor chat (when configured)."""

    def __init__(self, feed: NotificationFeed | None = None, chat: TelegramDoctorChat | None = None) -> None:
        self.feed = feed or NotificationFeed()
        self._chat = chat

    @property
    def chat(self) -> TelegramDoctorChat:
        # Lazy: settings are read on the first notification, not when the package is imported.
        if self._chat is None:
            self._chat = TelegramDoctorChat()
        return self._chat

    def recent(self, since_id: int = 0, doctor_id: str | None = None) -> list[Notification]:
        return self.feed.recent(since_id, doctor_id)

    async def send(self, doctor_id: str, level: AlertLevelName, text: str, token: str | None = None) -> None:
        try:
            self.feed.add(doctor_id, level, text, token, clock.now())
        except Exception as exc:  # noqa: BLE001 — never break the caller
            log.warning("notify feed failed error=%s", type(exc).__name__)
        await self.chat.send(level, text, token)