"""Alarms about the outside services' use (AV-052), to the owner:

* a level reached -- YELLOW, RED, STOP (mountain_twin.budgets);
* a series of refusals -- SERIES_429_COUNT 429s within SERIES_429_SECONDS;
* a jump -- this hour's weight over SPIKE_FACTOR x the mean hour of the
  past week (and at least SPIKE_MIN_WEIGHT, so a quiet week's first
  request is no alarm).

At most one alarm per service, kind and UTC day (usage.UsageStore.mark_alarm):
never a flood. The channel (ALARM_CHANNEL): ``macos`` -- the log and a macOS
notification (the development Mac's default), ``log`` -- the log only,
``ntfy`` -- the log and a push to the owner's phone through ntfy (the
production server's channel, the owner's decision of 05.10.2026:
docs/BETA_CAPACITY.md section 5). ntfy needs NTFY_TOPIC -- a secret topic
name, kept only in the server's environment; NTFY_SERVER changes the
instance (default https://ntfy.sh, later possibly the own server). Without a
topic the ntfy channel logs only and says so once.
"""

from __future__ import annotations

import base64
import os
import subprocess
import sys
import urllib.request

from mountain_twin import budgets
from mountain_twin.brand import brand

SERIES_429_COUNT = 3
SERIES_429_SECONDS = 600.0
SPIKE_FACTOR = 3.0
SPIKE_MIN_WEIGHT = 50.0
CHANNEL_ENVIRONMENT = "ALARM_CHANNEL"
CHANNELS = ("macos", "log", "ntfy")
NTFY_TOPIC_ENVIRONMENT = "NTFY_TOPIC"
NTFY_SERVER_ENVIRONMENT = "NTFY_SERVER"
NTFY_DEFAULT_SERVER = "https://ntfy.sh"
LEVEL_TEXT = {
    "YELLOW": "żółty (50 % dziennego limitu): prefetch wyłączony, porównanie modeli tylko z cache",
    "RED": "czerwony (80 %): tylko cache, źródła zapasowe i działania użytkownika",
    "STOP": "stop (95 %): żadnych zapytań do odnowienia limitu",
}


def macos_notify(title: str, message: str) -> None:
    script = f"display notification {message!r} with title {title!r}".replace("'", '"')
    subprocess.run(["osascript", "-e", script], check=False, timeout=10, capture_output=True)


def _header(text: str) -> str:
    """A header value as RFC 2047 (ntfy decodes it): http.client sends
    headers as Latin-1, and the alarm's title has a dash and Polish letters."""
    return "=?UTF-8?B?" + base64.b64encode(text.encode("utf-8")).decode("ascii") + "?="


def ntfy_notifier(environ=os.environ, *, opener=None):
    """``notify(title, message)`` posting to ntfy, or None without NTFY_TOPIC.
    The request goes through mountain_twin.offline.urlopen (counted as the
    ``ntfy`` service, never sent in offline mode)."""
    topic = (environ.get(NTFY_TOPIC_ENVIRONMENT) or "").strip()
    if not topic:
        return None
    server = (environ.get(NTFY_SERVER_ENVIRONMENT) or NTFY_DEFAULT_SERVER).strip().rstrip("/")
    if opener is None:
        from mountain_twin.offline import urlopen as opener

    def notify(title: str, message: str) -> None:
        request = urllib.request.Request(
            f"{server}/{topic}",
            data=message.encode("utf-8"),
            method="POST",
            headers={"Title": _header(title), "Priority": "high", "Tags": "warning"},
        )
        with opener(request, timeout=10) as response:
            response.read()

    return notify


def default_channel(environ=os.environ) -> str:
    from mountain_twin.offline import offline_mode

    if offline_mode() is not None:
        return "log"  # tests and the simulator never post notifications
    value = (environ.get(CHANNEL_ENVIRONMENT) or "").strip().lower()
    if value in CHANNELS:
        return value
    return "macos" if sys.platform == "darwin" else "log"


class UsageAlarms:
    """A listener of usage.UsageStore (``store.listeners.append(alarms)``).
    ``notify(title, message)`` and ``log(line)`` are injectable."""

    def __init__(self, store, *, notify=None, log=None, channel: str | None = None):
        self.store = store
        self.channel = channel or default_channel()
        self._log = log or (lambda line: print(line, file=sys.stderr, flush=True))
        if notify is None and self.channel == "macos":
            notify = macos_notify
        elif notify is None and self.channel == "ntfy":
            notify = ntfy_notifier()
            if notify is None:
                self._log(f"ALARM channel ntfy without {NTFY_TOPIC_ENVIRONMENT}: the log only")
        self._notify = notify
        self.sent: list[tuple[str, str, str]] = []  # (service, kind, message): observability

    def __call__(self, service: str, feature: str, outcome: str) -> None:
        self.check(service, outcome)

    def check(self, service: str, outcome: str = "OK") -> None:
        level = budgets.level(service)
        if level != "GREEN":
            self._alarm(service, f"LEVEL_{level}", f"{service}: poziom {LEVEL_TEXT[level]}")
        if outcome == "RATE_LIMITED":
            refused = self.store.errors_since(service, "RATE_LIMITED", SERIES_429_SECONDS)
            if refused >= SERIES_429_COUNT:
                self._alarm(
                    service,
                    "SERIES_429",
                    f"{service}: {refused} odmów (429) w ciągu {SERIES_429_SECONDS / 60:.0f} min",
                )
        hour = self.store.this_hour(service)["weight"]
        mean = self.store.mean_hourly_weight(service)
        if hour >= SPIKE_MIN_WEIGHT and mean > 0 and hour > SPIKE_FACTOR * mean:
            self._alarm(
                service,
                "SPIKE",
                f"{service}: skok zużycia — {hour:.0f} w tej godzinie, średnio {mean:.1f}/h w tygodniu",
            )

    def _alarm(self, service: str, kind: str, message: str) -> None:
        if not self.store.mark_alarm(service, kind):
            return
        self.sent.append((service, kind, message))
        self._log(f"ALARM usage {kind}: {message}")
        if self._notify is not None:
            try:
                self._notify(f"{brand()['name']} — zużycie usług", message)
            except Exception as error:  # an alarm channel never breaks a request
                self._log(f"ALARM channel failed: {type(error).__name__}: {error}")