kadence/backend/app/services/notify.py

"""Outbound messages: Telegram first, Slack next, the log as a last resort.

The chunking, the HTML parse mode and the "Telegram > Slack > console" order are
jira-helper's, because they are the right answers. What is added here is quiet
hours and a record of what was sent, so the nudges cannot turn into spam.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import date

import httpx
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import EventLog
from app.settings_schema import AppSettings
from app.timeutil import in_window, local_date, local_time

log = logging.getLogger(__name__)

# Telegram rejects messages longer than 4096 characters; leave headroom.
_TELEGRAM_MAX_CHARS = 4000
_TIMEOUT = 20


class NotifyError(RuntimeError):
    pass


@dataclass(slots=True)
class Delivery:
    channel: str
    ok: bool
    detail: str = ""

    def as_dict(self) -> dict[str, object]:
        return {"channel": self.channel, "ok": self.ok, "detail": self.detail}


def _chunk(text: str, size: int = _TELEGRAM_MAX_CHARS) -> list[str]:
    """Split on line boundaries so a message never exceeds Telegram's limit."""
    chunks: list[str] = []
    current = ""
    for line in text.split("\n"):
        candidate = f"{current}\n{line}" if current else line
        if len(candidate) > size and current:
            chunks.append(current)
            current = line
        else:
            current = candidate
    if current:
        chunks.append(current)
    return chunks or [""]


def send_telegram(
    settings: AppSettings, text: str, *, reply_markup: dict | None = None
) -> Delivery:
    telegram = settings.notifications.telegram
    if not (telegram.enabled and telegram.bot_token and telegram.chat_id):
        return Delivery("telegram", False, "Telegram is not configured.")

    url = f"https://api.telegram.org/bot{telegram.bot_token}/sendMessage"
    parts = _chunk(text)
    try:
        with httpx.Client(timeout=_TIMEOUT) as client:
            for index, part in enumerate(parts):
                payload = {
                    "chat_id": telegram.chat_id,
                    "text": part,
                    "parse_mode": "HTML",
                    "disable_web_page_preview": True,
                }
                # Buttons belong on the last chunk, where the message ends.
                if reply_markup and index == len(parts) - 1:
                    payload["reply_markup"] = reply_markup
                resp = client.post(url, json=payload)
                if not resp.is_success:
                    return Delivery(
                        "telegram", False, f"{resp.status_code}: {resp.text[:200]}"
                    )
    except httpx.HTTPError as exc:
        return Delivery("telegram", False, str(exc))
    return Delivery("telegram", True, f"{len(parts)} message(s)")


def send_slack(settings: AppSettings, text: str) -> Delivery:
    slack = settings.notifications.slack
    if not (slack.enabled and slack.webhook_url):
        return Delivery("slack", False, "Slack is not configured.")
    try:
        with httpx.Client(timeout=_TIMEOUT) as client:
            resp = client.post(slack.webhook_url, json={"text": _strip_html(text)})
        if not resp.is_success:
            return Delivery("slack", False, f"{resp.status_code}: {resp.text[:200]}")
    except httpx.HTTPError as exc:
        return Delivery("slack", False, str(exc))
    return Delivery("slack", True, "ok")


def _strip_html(text: str) -> str:
    import re

    return re.sub(r"<[^>]+>", "", text)


def in_quiet_hours(settings: AppSettings) -> bool:
    quiet = settings.notifications.telegram.quiet_hours
    if not quiet.enabled:
        return False
    return in_window(local_time(settings.timezone), quiet.start, quiet.end)


def notify(
    settings: AppSettings,
    text: str,
    *,
    urgent: bool = False,
    reply_markup: dict | None = None,
) -> list[Delivery]:
    """Send to every configured channel. `urgent` ignores quiet hours."""
    if not urgent and in_quiet_hours(settings):
        return [Delivery("quiet-hours", False, "suppressed by quiet hours")]

    deliveries: list[Delivery] = []
    telegram = send_telegram(settings, text, reply_markup=reply_markup)
    if telegram.ok or telegram.detail != "Telegram is not configured.":
        deliveries.append(telegram)
    slack = send_slack(settings, text)
    if slack.ok or slack.detail != "Slack is not configured.":
        deliveries.append(slack)
    if not deliveries:
        log.info("notification (no channel configured):\n%s", _strip_html(text))
        deliveries.append(Delivery("log", True, "written to the server log"))
    return deliveries


# --------------------------------------------------------------------------
# "Did we already send this today?"
# --------------------------------------------------------------------------
def already_sent(session: Session, kind: str, day: date) -> bool:
    return (
        session.scalar(
            select(EventLog.id).where(EventLog.kind == kind, EventLog.local_date == day)
        )
        is not None
    )


def count_sent(session: Session, kind: str, day: date) -> int:
    return len(
        list(
            session.scalars(
                select(EventLog.id).where(EventLog.kind == kind, EventLog.local_date == day)
            )
        )
    )


def record_sent(session: Session, kind: str, day: date, payload: dict | None = None) -> None:
    session.add(EventLog(kind=kind, local_date=day, payload=payload or {}))
    session.commit()


def today(settings: AppSettings) -> date:
    return local_date(settings.timezone)