notify v0.7.1 [base]

"""notify.engine — рассылки администраторам (чистая реализация + ctx).

Порт modules/notify.py: стартовые/стоп/транспорт-уведомления уходят
telegram-пользователям из прав, у которых выставлено соответствующее
receive_notifications* право (старый should_notify: право + флаг
receive_notifications_no_sound). Отправка — через
ctx.api transport_telegram.sender (активный транспорт).
"""
import logging

logger = logging.getLogger(__name__)


def _has_right(ctx, qid, right) -> bool:
    """Право через effective: покрывает all (админ без явного права —
    тоже получатель; старый get_entry его пропускал)."""
    try:
        eff = ctx.rights.effective(qid)
    except Exception:  # noqa: BLE001
        return False
    return bool((eff or {}).get(right, False))


def _targets(ctx, right):
    """[(chat_id, no_sound)] telegram-получателей с правом right."""
    out = []
    try:
        users = ctx.rights.known_users()
    except Exception:  # noqa: BLE001
        return out
    for qid in users:
        try:
            kind, _, num = (qid or "").partition(":")
            if kind != "telegram" or not num.lstrip("-").isdigit():
                continue
            if not _has_right(ctx, qid, right):
                continue
            try:
                entry = ctx.rights.get_entry(qid) or {}
            except Exception:  # noqa: BLE001
                entry = {}
            out.append((int(num), bool(entry.get(
                "receive_notifications_no_sound", False))))
        except Exception:  # noqa: BLE001
            continue
    return out


def _web_targets(ctx, right):
    """[uid] web-получателей с правом right (кабинет, web:<uid>)."""
    out = []
    try:
        users = ctx.rights.known_users()
    except Exception:  # noqa: BLE001
        return out
    for qid in users:
        try:
            kind, _, num = (qid or "").partition(":")
            if kind != "web" or not num.isdigit():
                continue
            if _has_right(ctx, qid, right):
                out.append(int(num))
        except Exception:  # noqa: BLE001
            continue
    return out


def _sender(ctx):
    try:
        return ctx.api.call("transport_telegram.sender")
    except Exception as e:  # noqa: BLE001
        logger.warning("notify: нет sender: %s", e)
        return None


def check_recipients(ctx, right):
    """Проверка прав перед рассылкой: {tg: [(chat_id, no_sound)],
    web: [uid]} — только держатели right.

    Право считается через effective (покрывает all и гостевые дефолты):
    нет записи — дефолты (всё False, рассылка не идёт); all — всё кроме
    all_ignore; явное право — в лоб. Мусорный right (пусто/не строка) —
    пустые списки, без исключений.
    """
    empty = {"tg": [], "web": []}
    if not isinstance(right, str) or not right.strip():
        return empty
    right = right.strip()
    try:
        users = ctx.rights.known_users()
    except Exception:  # noqa: BLE001
        return empty
    tg, web = [], []
    for qid in users:
        try:
            kind, _, num = (qid or "").partition(":")
            if kind == "telegram":
                if not num.lstrip("-").isdigit():
                    continue
                if not _has_right(ctx, qid, right):
                    continue
                try:
                    entry = ctx.rights.get_entry(qid) or {}
                except Exception:  # noqa: BLE001
                    entry = {}
                tg.append((int(num), bool(entry.get(
                    "receive_notifications_no_sound", False))))
            elif kind == "web":
                if not num.isdigit():
                    continue
                if not _has_right(ctx, qid, right):
                    continue
                web.append(int(num))
        except Exception:  # noqa: BLE001
            continue
    return {"tg": tg, "web": web}


def fanout(ctx, right, text, kind="send", level="info",
           silent_no_sound=True, retry=1, tg_extra=None,
           sender_name="notify.fanout", tg_skip_text=False):
    """Единый распил рассылки: проверка прав → TG + веб-инбокс.

    right — право-подписка (receive_notifications_*); получатели — только
    его держатели (effective: all покрывается, посторонние отсекаются).
    tg_extra(chat_id, sender) — хук TG-ветки (кнопки/графики; исключения
    хука гасятся, доставка считается по send_message). retry — повторы
    TG-отправки (экспоненциальная пауза 1/2/4с). tg_skip_text — не слать
    цельный текст в TG (хук шлёт сам, напр. чанки с MarkdownV2; тогда
    доставка = отсутствие исключений хука). Возвращает
    {"tg": n, "web": m} + emit notify.sent (WS и Web Push подхватят).
    """
    from . import inbox as _inbox

    dest = check_recipients(ctx, right)
    tg = 0
    sender = _sender(ctx)
    if sender is not None and dest["tg"]:
        import time as _time
        for chat_id, no_sound in dest["tg"]:
            ok = False
            if tg_skip_text:
                # Текст везёт хук (чанки/разметка); доставка — по его
                # молчанию. Ретраи — внутри хука (у network свои).
                try:
                    if tg_extra is not None:
                        tg_extra(chat_id, sender)
                    ok = True
                except Exception as e:  # noqa: BLE001
                    logger.error("notify %s extra: %s", chat_id, e)
            else:
                for attempt in range(max(1, int(retry or 1))):
                    try:
                        sender.send_message(
                            chat_id=chat_id, text=text or "…",
                            disable_notification=bool(no_sound and silent_no_sound))
                        ok = True
                        break
                    except Exception as e:  # noqa: BLE001
                        logger.warning("notify %s попытка %d: %s",
                                       chat_id, attempt + 1, e)
                        if attempt + 1 < max(1, int(retry or 1)):
                            _time.sleep(2 ** attempt)
            if ok:
                tg += 1
                if tg_extra is not None and not tg_skip_text:
                    try:
                        tg_extra(chat_id, sender)
                    except Exception as e:  # noqa: BLE001
                        logger.warning("notify extra %s: %s", chat_id, e)
            else:
                logger.error("notify %s: не доставлено", chat_id)
    web = 0
    try:
        for uid in dest["web"]:
            try:
                entry = ctx.rights.get_entry("web:%d" % uid) or {}
            except Exception:  # noqa: BLE001
                entry = {}
            if _inbox.put(uid, text, right=right, kind=kind, level=level,
                           silent=bool(silent_no_sound and entry.get(
                               "receive_notifications_no_sound", False)),
                           sender=sender_name):
                web += 1
    except Exception as e:  # noqa: BLE001
        logger.warning("notify inbox: %s", e)
    try:
        ctx.events.emit("notify.sent", {"right": right, "tg": tg,
                                        "web": web})
    except Exception:  # noqa: BLE001
        pass
    return {"tg": tg, "web": web}


def send(ctx, text, right="receive_notifications", silent_no_sound=True):
    """Разослать text держателям права right: Telegram + веб-инбокс.

    Тонкий враппер над fanout (retry=1: совместимость — раньше повторов
    не было). Возвращает {"tg": n_tg, "web": n_web}.
    """
    return fanout(ctx, right, text, kind="send",
                  silent_no_sound=silent_no_sound, retry=1,
                  sender_name="notify.send")


#: Динамический реестр подписок уведомлений для меню.
#: Владелец заявляет себя сам через notify.register (в своём start):
#: никакой статики по чужим модулям здесь нет. Порядок выдачи —
#: по right (детерминирован независимо от порядка загрузки).
_SUBSCRIPTIONS: dict = {}

_RIGHT_RX = None  # ленивая компиляция (модуль без импорта re на старте)


def _right_rx():
    global _RIGHT_RX
    if _RIGHT_RX is None:
        import re as _re
        _RIGHT_RX = _re.compile(r"^receive_notifications_[a-z0-9_]{1,40}$")
    return _RIGHT_RX


def register_subscription(label, right, edit_right="", module=""):
    """Заявить подписку уведомлений. Возвращает (ok, err).

    label — подпись кнопки (1..40), right — receive_notifications_*,
    edit_right — право на переключение себе (дефолт right + "_edit"),
    module — имя модуля-владельца (для диагностики).
    Повторная регистрация того же right — перезапись (идемпотентно).
    """
    label = (label or "").strip()[:40]
    right = (right or "").strip()
    edit_right = (edit_right or "").strip() or (right + "_edit" if right else "")
    module = (module or "").strip()[:32]
    if not label:
        return False, "пустой label"
    if not _right_rx().match(right):
        return False, "right должен быть receive_notifications_*"
    if not _right_rx().match(edit_right):
        return False, "edit_right должен быть receive_notifications_*"
    _SUBSCRIPTIONS[right] = {"label": label, "right": right,
                             "edit_right": edit_right, "module": module}
    return True, ""


def unregister_subscription(right):
    """Снять подписку. Возвращает True, если была."""
    return _SUBSCRIPTIONS.pop((right or "").strip(), None) is not None


def list_subscriptions():
    """[{label, right, edit_right, module}] — по right (детерминированно)."""
    return [dict(_SUBSCRIPTIONS[k]) for k in sorted(_SUBSCRIPTIONS)]


def toggle_map():
    """{callback_data: right} — производная от реестра."""
    return {"toggle_notif:%s" % right: right
            for right in sorted(_SUBSCRIPTIONS)}


def _tg_count(result) -> int:
    """Число TG-доставок из результата send (dict) или старого int."""
    if isinstance(result, dict):
        try:
            return int(result.get("tg") or 0)
        except (TypeError, ValueError):
            return 0
    try:
        return int(result or 0)
    except (TypeError, ValueError):
        return 0


def send_to(ctx, key, text, kind="direct", level="info"):
    """Прямое уведомление конкретному пользователю (не по праву).

    key — telegram:<id> (сообщением), web:<uid> (в инбокс), иначе 0/0.
    Групповые TG-чаты (отрицательный id, нет в правах) — тоже сообщением.
    Возвращает {"tg": n, "web": m}.
    """
    from . import inbox as _inbox

    kind_s, _, num = (key or "").partition(":")
    if kind_s == "telegram" and (num or "").lstrip("-").isdigit():
        sender = _sender(ctx)
        if sender is None:
            return {"tg": 0, "web": 0}
        try:
            sender.send_message(chat_id=int(num), text=text or "…",
                                disable_notification=False)
            return {"tg": 1, "web": 0}
        except Exception as e:  # noqa: BLE001
            logger.error("notify direct %s: %s", key, e)
            return {"tg": 0, "web": 0}
    if kind_s == "web" and (num or "").isdigit():
        ok = bool(_inbox.put(num, text, kind=kind, level=level,
                             sender="notify.send_to"))
        try:
            ctx.events.emit("notify.sent", {"right": "", "tg": 0,
                                            "web": 1 if ok else 0,
                                            "to": key})
        except Exception:  # noqa: BLE001
            pass
        return {"tg": 0, "web": 1 if ok else 0}
    return {"tg": 0, "web": 0}


def startup(ctx, message="Бот успешно запущен!"):
    """Уведомление о запуске (право receive_notifications_status)."""
    return send(ctx, message, right="receive_notifications_status")


def rights_changed(ctx, key, right, value):
    """Уведомление пользователю о смене его права (как в проде).

    Шлём напрямую затронутому пользователю, а не по правам-рассылкам:
    право могли только что отобрать. Telegram — сообщением, web —
    в инбокс. Возвращает {"tg": n, "web": m} (0/0 — некому/некуда).
    """
    from . import inbox as _inbox

    kind, _, num = (key or "").partition(":")
    text = "🔑 Право %s: %s" % (right, "выдано" if value else "отозвано")
    if kind == "telegram" and (num or "").lstrip("-").isdigit():
        sender = _sender(ctx)
        if sender is None:
            return {"tg": 0, "web": 0}
        try:
            sender.send_message(chat_id=int(num), text=text,
                                disable_notification=False)
            return {"tg": 1, "web": 0}
        except Exception as e:  # noqa: BLE001
            logger.error("notify rights %s: %s", key, e)
            return {"tg": 0, "web": 0}
    if kind == "web" and (num or "").isdigit():
        ok = bool(_inbox.put(num, text, right=right, kind="rights",
                             sender="notify.rights_changed"))
        # Пуш в WS — как в send/send_to: без emit запись видна только
        # через 30-сек поллинг (а при живом WS поллинга нет вообще).
        try:
            ctx.events.emit("notify.sent", {"right": right, "tg": 0,
                                            "web": 1 if ok else 0,
                                            "to": key})
        except Exception:  # noqa: BLE001
            pass
        return {"tg": 0, "web": 1 if ok else 0}
    return {"tg": 0, "web": 0}


def shutdown(ctx, message="Бот успешно остановлен!"):
    """Уведомление об остановке (право receive_notifications_status)."""
    return send(ctx, message, right="receive_notifications_status")


def transport_switch(ctx, old, new, reason=""):
    """Уведомление о переключении TG-транспорта (всегда беззвучно)."""
    return send(ctx, "🔁 Переключение Telegram-транспорта: %s -> %s\nПричина: %s"
                % (old, new, reason or "—"),
                right="receive_notifications_status", silent_no_sound=True) or True

"""notify.inbox — серверный буфер веб-уведомлений (доставка в браузер).

Зачем: Telegram-рассылка (engine.send) — fire-and-forget; кто был офлайн,
тот пропустил. Веб-клиент забирает уведомления из буфера (WS/REST на
этапе 2) — каждое ждёт получателя, а не сгорает. Хранение —
data/notify/inbox.json (атомарная запись, как homepage.json), один файл
на всех пользователей: {uid: [записи]}.

Запись: {id, ts, right, kind, title, body, level, silent, from}.
Политика: ≤MAX_PER_USER свежих на пользователя (старые вытесняются),
TTL — протухшие чистятся при записи/чтении. Прочитано — mark_read
(по id), unread — счётчик для бейджа колокольчика.
"""
import json
import logging
import os
import threading
import time
import uuid

logger = logging.getLogger(__name__)

_lock = threading.Lock()
_file = ""

MAX_PER_USER = 200
TTL_SECONDS = 30 * 24 * 3600

LEVELS = ("info", "ok", "warn", "error")


def configure(path: str) -> None:
    global _file
    _file = path


def _now() -> int:
    return int(time.time())


def _fresh(rec, now: int) -> bool:
    try:
        return now - int(rec.get("ts") or 0) <= TTL_SECONDS
    except Exception:  # noqa: BLE001
        return False


def load_all():
    """Всё дерево {uid: [записи]} (сырое, без чистки)."""
    try:
        with open(_file, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _save(data) -> bool:
    try:
        os.makedirs(os.path.dirname(_file) or ".", exist_ok=True)
        tmp = _file + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        os.replace(tmp, _file)
        return True
    except OSError as e:
        logger.warning("notify inbox: не записался: %s", e)
        return False


def _clean_list(items, now: int):
    """Свежие (сначала новые), в лимите."""
    live = [r for r in (items or []) if isinstance(r, dict)
            and _fresh(r, now)]
    live.sort(key=lambda r: (r.get("ts") or 0, r.get("id") or ""),
              reverse=True)
    return live[:MAX_PER_USER]


def _title_body(text, kind: str):
    """Первая строка — title, остальное — body (пусто = весь текст в body)."""
    text = (text or "").strip()
    if "\n" in text:
        head, tail = text.split("\n", 1)
        head, tail = head.strip(), tail.strip()
        if head and tail:
            return head[:120], tail[:2000]
    return "", text[:2000]


def put(uid, text, right="", kind="send", level="info", silent=False,
        sender=""):
    """Положить уведомление пользователю. Вернуть id или ''."""
    try:
        uid = str(int(uid))
    except (TypeError, ValueError):
        return ""
    if level not in LEVELS:
        level = "info"
    title, body = _title_body(text, kind)
    rec = {"id": uuid.uuid4().hex[:16], "ts": _now(),
           "right": (right or "")[:80], "kind": (kind or "")[:32],
           "title": title, "body": body or "…", "level": level,
           "silent": bool(silent), "read": False,
           "from": (sender or "")[:120]}
    with _lock:
        data = load_all()
        items = data.get(uid) or []
        if not isinstance(items, list):
            items = []
        items.append(rec)
        data[uid] = _clean_list(items, _now())
        if not _save(data):
            return ""
    return rec["id"]


def list_for(uid, since=0, include_read=True):
    """Свежие записи пользователя (сначала новые). since — ts-граница."""
    try:
        uid = str(int(uid))
    except (TypeError, ValueError):
        return []
    with _lock:
        data = load_all()
        items = _clean_list(data.get(uid), _now())
    try:
        since = int(since or 0)
    except (TypeError, ValueError):
        since = 0
    out = [r for r in items if (r.get("ts") or 0) > since]
    if not include_read:
        out = [r for r in out if not r.get("read")]
    return out


def unread_count(uid) -> int:
    """Число непрочитанных свежих."""
    return len(list_for(uid, include_read=False))


def mark_read(uid, ids) -> int:
    """Пометить прочитанными. Вернуть число изменённых."""
    try:
        uid = str(int(uid))
    except (TypeError, ValueError):
        return 0
    want = {str(i) for i in (ids or []) if i}
    if not want:
        return 0
    n = 0
    with _lock:
        data = load_all()
        items = data.get(uid)
        if not isinstance(items, list):
            return 0
        for r in items:
            if isinstance(r, dict) and r.get("id") in want \
                    and not r.get("read"):
                r["read"] = True
                n += 1
        if n and not _save(data):
            return 0
    return n


def mark_all_read(uid) -> int:
    """Прочитать всё свежее. Вернуть число изменённых."""
    with _lock:
        try:
            uid = str(int(uid))
        except (TypeError, ValueError):
            return 0
        data = load_all()
        items = data.get(uid)
        if not isinstance(items, list):
            return 0
        now = _now()
        n = 0
        for r in items:
            if isinstance(r, dict) and _fresh(r, now) \
                    and not r.get("read"):
                r["read"] = True
                n += 1
        if n and not _save(data):
            return 0
    return n

"""notify.Module — рассылки администраторам (Telegram + веб-инбокс).

Без TG-команд: только ctx.api для соседей (send/startup/shutdown/...),
инбокс-буфер для веб-клиентов (inbox/unread/read — за ними WS/REST).
Web Push (закрытая вкладка) — fan-out из подписки на notify.sent.
Старт — по on_ready транспорта (sender в cmd_run ещё не активен),
стоп — из boot.py. Права-получатели — из ctx.rights
(receive_notifications*), оба неймспейса: telegram:* и web:*.
"""
import logging
import os

from core.base_module import BaseModule

from . import engine as notify
from . import inbox as notify_inbox
from . import webpush as notify_push

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        self.ctx = ctx
        self._startup_sent = False
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("notify", mod_dir)
        from core.paths import abs_data
        notify_inbox.configure(abs_data(
            ctx.data_dir, os.path.join("notify", "inbox.json"),
            os.path.join(ctx.data_dir, "notify", "inbox.json")))
        notify_push.configure(
            abs_data(ctx.data_dir, os.path.join("notify", "vapid_private.pem"),
                     os.path.join(ctx.data_dir, "notify", "vapid_private.pem")),
            abs_data(ctx.data_dir, os.path.join("notify", "push_subs.json"),
                     os.path.join(ctx.data_dir, "notify", "push_subs.json")),
            (self.config.get("VAPID_SUBJECT", "") or "").strip()
            or "mailto:bot@localhost")
        ctx.api.register("notify", "send", lambda text, right="receive_notifications":
                         notify.send(ctx, text, right=right))
        ctx.api.register("notify", "fanout",
                         lambda right, text, kind="send", level="info",
                         retry=1, tg_extra=None, tg_skip_text=False,
                         sender_name="notify.fanout": notify.fanout(
                             ctx, right, text, kind=kind, level=level,
                             retry=int(retry or 1), tg_extra=tg_extra,
                             tg_skip_text=bool(tg_skip_text),
                             sender_name=sender_name or "notify.fanout"))
        ctx.api.register("notify", "send_to",
                         lambda key, text, kind="direct", level="info":
                         notify.send_to(ctx, key, text, kind=kind,
                                        level=level))
        ctx.api.register("notify", "startup", lambda message="Бот успешно запущен!":
                         notify.startup(ctx, message))
        ctx.api.register("notify", "shutdown", lambda message="Бот успешно остановлен!":
                         notify.shutdown(ctx, message))
        ctx.api.register("notify", "rights_changed",
                         lambda key, right, value: notify.rights_changed(
                             ctx, key, right, value))
        ctx.api.register(
            "notify", "transport_switch",
            lambda old, new, reason="": notify.transport_switch(ctx, old, new, reason))
        ctx.api.register("notify", "register",
                         lambda label, right, edit_right="", module="":
                         notify.register_subscription(label, right, edit_right, module))
        ctx.api.register("notify", "unregister",
                         lambda right: notify.unregister_subscription(right))
        ctx.api.register("notify", "subscriptions",
                         lambda: notify.list_subscriptions())
        ctx.api.register("notify", "toggle_map",
                         lambda: notify.toggle_map())
        ctx.api.register("notify", "inbox",
                         lambda uid, since=0: notify_inbox.list_for(uid, since))
        ctx.api.register("notify", "unread",
                         lambda uid: notify_inbox.unread_count(uid))
        ctx.api.register("notify", "read",
                         lambda uid, ids: notify_inbox.mark_read(uid, ids))
        ctx.api.register("notify", "read_all",
                         lambda uid: notify_inbox.mark_all_read(uid))
        # Своя подписка (старт/стоп/связь) — как у всех, через тот же реестр.
        notify.register_subscription(
            "Status", "receive_notifications_status",
            "receive_notifications_status_edit", "notify")
        # Стартовое уведомление — по готовности транспорта (см. start()).
        try:
            ctx.events.subscribe("connectivity.changed", self._on_connectivity)
        except Exception as e:  # noqa: BLE001
            logger.warning("notify subscribe: %s", e)
        # Web Push fan-out: свежий инбокс — в push-сервисы браузеров.
        # Без подписок/ключей fan_out — нули (тихое отключение внутри).
        try:
            ctx.events.subscribe("notify.sent", self._on_sent_push)
        except Exception as e:  # noqa: BLE001
            logger.warning("notify push subscribe: %s", e)

    def _on_sent_push(self, payload) -> None:
        try:
            notify_push.fan_out_async(
                lambda uid: notify_inbox.list_for(uid, include_read=False))
        except Exception as e:  # noqa: BLE001
            logger.warning("notify push fan-out: %s", e)

    def start(self) -> None:
        # Подписка здесь, а не в setup: api транспорта регистрируется
        # в его setup, порядок setup'ов не гарантирован. К моменту
        # start() все setup'ы выполнены, а активация транспорта —
        # позже (фон supervisor), так что колбэк успевает.
        import time as _time
        self._start_ts = _time.time()
        try:
            self.ctx.api.call("transport_telegram.on_ready", self._on_tg_ready)
        except Exception as e:  # noqa: BLE001
            logger.warning("notify on_ready: %s", e)

    def _on_tg_ready(self, active_name):
        if self._startup_sent:
            return
        self._startup_sent = True
        try:
            from datetime import datetime as _dt
            when = _dt.fromtimestamp(getattr(self, "_start_ts", 0) or 0)
            when_s = when.strftime("%d.%m %H:%M:%S") if getattr(
                self, "_start_ts", 0) else "?"
            n = notify.startup(
                self.ctx,
                "Бот успешно запущен!\nВремя запуска: %s\nТранспорт: %s"
                % (when_s, active_name))
            logger.info("notify startup: %s доставок (транспорт %s)",
                        n, active_name)
        except Exception as e:  # noqa: BLE001
            logger.warning("notify startup: %s", e)

    def _on_connectivity(self, payload):
        """Алерты о потере/восстановлении связи (право receive_notifications_status)."""
        from datetime import datetime as _dt
        try:
            up = bool((payload or {}).get("up", True))
            at = (payload or {}).get("at")
            when = _dt.fromtimestamp(at).strftime("%Y-%m-%d %H:%M:%S") if at else "?"
            text = ("🟢 Связь с Telegram восстановлена (%s)" % when) if up else (
                "🔴 Потеряна связь с Telegram (%s)" % when)
            notify.send(self.ctx, text, right="receive_notifications_status")
        except Exception as e:  # noqa: BLE001
            logger.warning("notify connectivity: %s", e)

    def health(self) -> dict:
        try:
            tg = len(notify._targets(self.ctx, "receive_notifications_status"))
            web = len(notify._web_targets(self.ctx, "receive_notifications_status"))
        except Exception:  # noqa: BLE001
            tg, web = -1, -1
        ok = tg >= 0 and web >= 0
        out = {"ok": ok, "module": self.name, "targets": tg,
               "web_targets": web}
        if ok:
            try:
                import time as _t
                data = notify_inbox.load_all()
                now = int(_t.time())
                pending = sum(
                    1 for items in (data or {}).values()
                    if isinstance(items, list) for r in items
                    if isinstance(r, dict) and not r.get("read")
                    and now - int(r.get("ts") or 0) <= notify_inbox.TTL_SECONDS)
                out["inbox_pending"] = pending
            except Exception:  # noqa: BLE001
                pass
            # Счётчик push-подписок. Ключ VAPID health НЕ создаёт
            # (генерация — по первому запросу ключа/пуша).
            try:
                out["push_subs"] = notify_push.subs_total()
            except Exception:  # noqa: BLE001
                pass
        return out

"""notify.ui_web — REST-фолбэк инбокса (когда WS недоступен).

GET  /api/notify/inbox?since= — свежие записи (сначала новые)
GET  /api/notify/unread — {"unread": n} (бейдж колокольчика)
POST /api/notify/read {ids: [...]} — пометить прочитанными
POST /api/notify/read_all {} — прочитать всё свежее

Всё — только вошедшим (мета right None + проверка сессии: без сессии
401). Чужие uid недоступны: uid всегда из своей сессии.
"""

ROUTES = [
    ("GET", "/api/notify/inbox", "notify_inbox", {"right": None}),
    ("GET", "/api/notify/unread", "notify_unread", {"right": None}),
    ("POST", "/api/notify/read", "notify_read", {"right": None}),
    ("POST", "/api/notify/read_all", "notify_read_all", {"right": None}),
    ("GET", "/api/notify/subs", "notify_subs", {"right": None}),
    ("POST", "/api/notify/sub", "notify_sub", {"right": None}),
    ("GET", "/api/notify/push/key", "notify_push_key", {"right": None}),
    ("POST", "/api/notify/push/subscribe", "notify_push_subscribe",
     {"right": None}),
    ("POST", "/api/notify/push/unsubscribe", "notify_push_unsubscribe",
     {"right": None}),
]

PANELS = []
NAV = []


def _uid(req):
    sess = req.get("session") or {}
    try:
        return int(sess.get("uid"))
    except (TypeError, ValueError):
        return 0


def handle_api(ctx, config, method, req):
    from core.errors import UserError

    from . import inbox as _inbox

    uid = _uid(req)
    if not uid:
        return 401, {"ok": False, "error": "auth"}

    if method == "notify_inbox":
        try:
            since = int(((req.get("query") or {}).get("since")) or 0)
        except (TypeError, ValueError):
            since = 0
        return {"ok": True, "items": _inbox.list_for(uid, since=since),
                "unread": _inbox.unread_count(uid)}

    if method == "notify_unread":
        return {"ok": True, "unread": _inbox.unread_count(uid)}

    if method == "notify_read":
        body = req.get("body") or {}
        ids = body.get("ids") or []
        if not isinstance(ids, list):
            return 400, {"ok": False, "error": "ids должен быть списком"}
        marked = _inbox.mark_read(uid, [str(i) for i in ids][:500])
        return {"ok": True, "marked": marked,
                "unread": _inbox.unread_count(uid)}

    if method == "notify_read_all":
        marked = _inbox.mark_all_read(uid)
        return {"ok": True, "marked": marked,
                "unread": _inbox.unread_count(uid)}

    if method == "notify_subs":
        # Подписки с состоянием для СВОЕГО uid (чекбоксы панели инбокса).
        from . import engine as _eng

        subs = _eng.list_subscriptions()
        out = []
        for s in subs:
            right = s.get("right") or ""
            edit = s.get("edit_right") or ""
            try:
                eff = ctx.rights.effective("web:%d" % uid) or {}
            except Exception:  # noqa: BLE001
                eff = {}
            out.append({"label": s.get("label") or right, "right": right,
                        "on": bool(eff.get(right, False)),
                        "can": bool(edit and eff.get(edit, False))})
        return {"ok": True, "subs": out}

    if method == "notify_sub":
        # Самоподписка: тоггл права себе (нужен edit_right — как TG-тоглы).
        from . import engine as _eng

        body = req.get("body") or {}
        right = str(body.get("right") or "").strip()
        sub = next((s for s in _eng.list_subscriptions()
                    if s.get("right") == right), None)
        if sub is None:
            return 400, {"ok": False, "error": "нет такой подписки"}
        edit = sub.get("edit_right") or ""
        try:
            eff = ctx.rights.effective("web:%d" % uid) or {}
        except Exception:  # noqa: BLE001
            eff = {}
        if not edit or not eff.get(edit, False):
            return 403, {"ok": False, "error": "forbidden"}
        try:
            entry = ctx.rights.get_entry("web:%d" % uid) or {}
            if not entry:
                entry = ctx.rights.guest_defaults()
            entry[right] = not entry.get(right, False)
            ctx.rights.set_entry("web:%d" % uid, entry,
                                 actor="web:%d" % uid)
        except Exception as e:  # noqa: BLE001
            return 500, {"ok": False, "error": str(e)}
        return {"ok": True, "right": right,
                "value": bool(entry[right])}

    if method == "notify_push_key":
        # Публичный VAPID-ключ для PushManager.subscribe. Сам ключ —
        # публичный по дизайну; роут за сессией (не светим лишнего).
        from . import webpush as _push

        pub = _push.public_key_b64()
        if not pub:
            return 500, {"ok": False,
                         "error": "web push недоступен (нет ключей/библиотеки)"}
        return {"ok": True, "publicKey": pub}

    if method == "notify_push_subscribe":
        from . import webpush as _push

        body = req.get("body") or {}
        keys = body.get("keys") or {}
        if not isinstance(keys, dict):
            return 400, {"ok": False, "error": "keys должен быть объектом"}
        ua = str(body.get("ua") or "")[:120]
        ok, err = _push.put_subscription(
            uid, body.get("endpoint"), keys.get("p256dh"),
            keys.get("auth"), ua)
        if not ok:
            return 400, {"ok": False, "error": err}
        return {"ok": True,
                "subs": _push.subscriptions_count(uid)}

    if method == "notify_push_unsubscribe":
        from . import webpush as _push

        body = req.get("body") or {}
        removed = _push.remove_subscription(uid, body.get("endpoint"))
        return {"ok": True, "removed": removed,
                "subs": _push.subscriptions_count(uid)}

    raise UserError("неизвестный метод: %s" % method)

"""notify.ui_ws — /ws/notify (пуш веб-уведомлений в браузер).

Протокол (JSON):
    сервер -> клиент: {"type": "hello", "unread": int}
    сервер -> клиент: {"type": "notify", "items": [записи инбокса]}
    клиент -> сервер: {"op": "sync", "since": ts} — всё свежее новее ts
    клиент -> сервер: {"op": "read", "ids": [...]} — пометить прочитанными
    клиент -> сервер: {"op": "read_all"} — прочитать всё
    сервер -> клиент: {"type": "read", "ids": [...], "unread": int}

Аутентификация — базовая (cookie/ticket/Bearer), RIGHT нет: фильтром
служит само право-рассылка (инбокс уже разложен только своим).
Новые записи пушатся через событие шины notify.sent (подписка живёт
пока открыт хоть один сокет; каждому сокету — только его uid).
"""
import logging

WS_PATH = "/ws/notify"

from botmod_transport_web.ws import make_base as _make_base

BaseSocket = _make_base()

logger = logging.getLogger(__name__)

_sockets = {}  # "web:<uid>" -> set(WSSocket)
_lock = None
_subscribed = False


def _get_lock():
    global _lock
    if _lock is None:
        import threading as _th
        _lock = _th.Lock()
    return _lock


def _uid_key(uid) -> str:
    try:
        return "web:%d" % int(uid)
    except (TypeError, ValueError):
        return ""


def _ctx_of(sock):
    return getattr(sock, "mod_ctx", None)


def _push_to(uid, items) -> None:
    """Разослать записи всем сокетам пользователя (из любого потока)."""
    import json as _json

    key = _uid_key(uid)
    if not key:
        return
    with _get_lock():
        targets = [s for s in (_sockets.get(key) or set())]
    if not targets:
        return
    obj = {"type": "notify", "items": list(items or [])}
    payload = _json.dumps(obj, ensure_ascii=False)
    for sock in targets:
        try:
            loop = getattr(sock, "ws_loop", None)
            if loop is not None:
                try:
                    loop.add_callback(sock.write_message, payload)
                    continue
                except Exception:  # noqa: BLE001
                    pass
            send_json = getattr(sock, "send_json", None)
            if callable(send_json):
                send_json(obj)
            else:
                sock.write_message(payload)
        except Exception:  # noqa: BLE001
            continue


def _on_sent(payload) -> None:
    """Шина notify.sent от engine.send: дотянуть свежее каждому сокету."""
    from . import inbox as _inbox

    with _get_lock():
        keys = [k for k, v in _sockets.items() if v]
    for key in keys:
        try:
            uid = int(key.split(":", 1)[1])
        except (ValueError, IndexError):
            continue
        try:
            items = _inbox.list_for(uid, include_read=False)
        except Exception:  # noqa: BLE001
            continue
        if items:
            _push_to(uid, items)


def _ensure_subscribed(sock) -> None:
    global _subscribed
    ctx = _ctx_of(sock)
    if ctx is None or _subscribed:
        return
    try:
        ctx.events.subscribe("notify.sent", _on_sent)
        _subscribed = True
    except Exception as e:  # noqa: BLE001
        logger.warning("notify ws subscribe: %s", e)


def _forget(sock) -> None:
    key = _uid_key(getattr(sock, "ws_uid", None))
    if not key:
        return
    with _get_lock():
        box = _sockets.get(key)
        if box is not None:
            box.discard(sock)
            if not box:
                _sockets.pop(key, None)


class WSSocket(BaseSocket):
    RIGHT = None  # любой вошедший; инбокс уже отфильтрован по правам

    def on_authed(self, sess):
        import tornado.ioloop

        from . import inbox as _inbox

        try:
            self.ws_uid = int(sess.get("uid"))
        except (TypeError, ValueError):
            try:
                self.close(code=4003, reason="bad uid")
            except Exception:  # noqa: BLE001
                pass
            return
        try:
            self.ws_loop = tornado.ioloop.IOLoop.current()
        except Exception:  # noqa: BLE001
            self.ws_loop = None
        key = _uid_key(self.ws_uid)
        with _get_lock():
            _sockets.setdefault(key, set()).add(self)
        _ensure_subscribed(self)
        try:
            unread = _inbox.unread_count(self.ws_uid)
        except Exception:  # noqa: BLE001
            unread = 0
        self.send_json({"type": "hello", "unread": unread})
        try:
            fresh = _inbox.list_for(self.ws_uid, include_read=False)
        except Exception:  # noqa: BLE001
            fresh = []
        if fresh:
            self.send_json({"type": "notify", "items": fresh})
        logger.info("notify ws открыт: %s", key)

    def on_message(self, raw):
        import json as _json

        from . import inbox as _inbox

        uid = getattr(self, "ws_uid", None)
        if uid is None:
            return
        try:
            msg = _json.loads(raw) if isinstance(raw, (str, bytes)) else {}
        except Exception:  # noqa: BLE001
            return
        if not isinstance(msg, dict):
            return
        op = msg.get("op")
        try:
            if op == "sync":
                items = _inbox.list_for(uid, since=msg.get("since", 0))
                self.send_json({"type": "notify", "items": items})
            elif op == "read":
                ids = _inbox.mark_read(uid, msg.get("ids") or [])
                self.send_json({"type": "read", "ids": msg.get("ids") or [],
                                "marked": ids,
                                "unread": _inbox.unread_count(uid)})
            elif op == "read_all":
                n = _inbox.mark_all_read(uid)
                self.send_json({"type": "read", "ids": [],
                                "marked": n,
                                "unread": _inbox.unread_count(uid)})
        except Exception as e:  # noqa: BLE001
            logger.warning("notify ws op %s: %s", op, e)

    def on_close(self):
        _forget(self)
        logger.info("notify ws закрыт: web:%s",
                    getattr(self, "ws_uid", "?"))

"""notify.webpush — Web Push (пуши при закрытой вкладке браузера).

Обычный WS-пуш умирает вместе со вкладкой; здесь рассылка идёт через
push-сервис браузера (FCM/autopush): работает пока запущен браузер,
вкладка не нужна. Требуется HTTPS (у нас есть — TLS_AUTO) и VAPID-ключи.

Ключи: data/notify/vapid_private.pem (0600, автогенерация при первом
старте, как CA в tls). Подписки устройств: data/notify/push_subs.json
{uid: [{endpoint, p256dh, auth, ua, created}]}, ≤10 на пользователя —
каждое устройство подписывается само, пуш уходит всем (кейс «один
аккаунт — несколько устройств»).

fan_out вызывается из подписки на шину notify.sent (module.setup):
берёт свежие непрочитанные из инбокса (права уже учтены там) и шлёт
последнюю запись + счётчик. Ушедшие подписки (404/410) чистятся.
Отправка — в daemon-потоке, engine.send не ждёт сеть.
Без pywebpush/cryptography — тихое отключение (лог один раз).
"""
import base64
import json
import logging
import os
import threading
import time

logger = logging.getLogger(__name__)

_lock = threading.Lock()
_key_path = ""
_subs_path = ""
_subject = "mailto:bot@localhost"
_disabled_logged = False
MAX_SUBS_PER_USER = 10


def configure(key_path: str, subs_path: str, subject="") -> None:
    global _key_path, _subs_path, _subject
    _key_path = key_path or ""
    _subs_path = subs_path or ""
    if (subject or "").strip():
        _subject = subject.strip()[:120]


def _disabled_once(reason: str) -> None:
    global _disabled_logged
    if not _disabled_logged:
        _disabled_logged = True
        logger.warning("notify webpush отключён: %s", reason)


def _b64url_decode(s: str):
    try:
        s = (s or "").strip()
        return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
    except Exception:  # noqa: BLE001
        return None


def _b64url_encode(blob: bytes) -> str:
    return base64.urlsafe_b64encode(blob).decode("ascii").rstrip("=")


def ensure_key():
    """Приватный ключ VAPID (PEM-путь). Нет — генерируем (P-256, 0600)."""
    if not _key_path:
        return ""
    try:
        with open(_key_path, "rb") as f:
            pem = f.read()
        if b"PRIVATE KEY" in pem:
            return _key_path
    except OSError:
        pass
    try:
        from cryptography.hazmat.primitives import serialization
        from cryptography.hazmat.primitives.asymmetric import ec
    except ImportError:
        _disabled_once("нет cryptography")
        return ""
    try:
        key = ec.generate_private_key(ec.SECP256R1())
        pem = key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption())
        os.makedirs(os.path.dirname(_key_path) or ".", exist_ok=True)
        tmp = _key_path + ".tmp"
        with open(tmp, "wb") as f:
            f.write(pem)
        try:
            os.chmod(tmp, 0o600)
        except OSError:
            pass
        os.replace(tmp, _key_path)
        logger.info("notify webpush: создан VAPID-ключ")
        return _key_path
    except OSError as e:
        _disabled_once(str(e)[:100])
        return ""


def public_key_b64() -> str:
    """Публичный ключ (base64url, для applicationServerKey). Пусто — нет."""
    if not ensure_key():
        return ""
    try:
        from cryptography.hazmat.primitives import serialization
        from cryptography.hazmat.primitives.serialization import load_pem_private_key
        with open(_key_path, "rb") as f:
            key = load_pem_private_key(f.read(), password=None)
        raw = key.public_key().public_bytes(
            serialization.Encoding.X962,
            serialization.PublicFormat.UncompressedPoint)
        return _b64url_encode(raw)
    except Exception as e:  # noqa: BLE001
        _disabled_once(str(e)[:100])
        return ""


def _subs_load() -> dict:
    try:
        with open(_subs_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _subs_save(data) -> bool:
    try:
        os.makedirs(os.path.dirname(_subs_path) or ".", exist_ok=True)
        tmp = _subs_path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False)
        os.replace(tmp, _subs_path)
        return True
    except OSError as e:
        logger.warning("notify webpush: подписки не записались: %s", e)
        return False


def valid_subscription(endpoint, p256dh, auth) -> str:
    """Проверка подписки. Вернуть текст ошибки или ''."""
    ep = (endpoint or "").strip()
    if not (ep.startswith("https://") or
            ep.startswith("http://localhost") or
            ep.startswith("http://127.0.0.1")):
        return "endpoint должен быть https (или localhost)"
    if len(ep) > 2000:
        return "endpoint слишком длинный"
    raw_p = _b64url_decode(p256dh or "")
    if raw_p is None or len(raw_p) != 65 or raw_p[0] != 0x04:
        return "плохой p256dh (нужна P-256 точка, 65 байт)"
    raw_a = _b64url_decode(auth or "")
    if raw_a is None or len(raw_a) != 16:
        return "плохой auth (нужно 16 байт)"
    return ""


def put_subscription(uid, endpoint, p256dh, auth, ua=""):
    """Сохранить подписку устройства. Вернуть (ok, err)."""
    try:
        uid = str(int(uid))
    except (TypeError, ValueError):
        return False, "плохой uid"
    err = valid_subscription(endpoint, p256dh, auth)
    if err:
        return False, err
    with _lock:
        data = _subs_load()
        items = data.get(uid) or []
        if not isinstance(items, list):
            items = []
        items = [s for s in items
                 if isinstance(s, dict)
                 and (s.get("endpoint") or "") != endpoint.strip()]
        items.append({"endpoint": endpoint.strip(),
                      "p256dh": (p256dh or "").strip(),
                      "auth": (auth or "").strip(),
                      "ua": (ua or "")[:120],
                      "created": int(time.time())})
        data[uid] = items[-MAX_SUBS_PER_USER:]
        if not _subs_save(data):
            return False, "ошибка записи"
    return True, ""


def remove_subscription(uid, endpoint) -> bool:
    """Убрать подписку устройства. Вернуть True если была."""
    try:
        uid = str(int(uid))
    except (TypeError, ValueError):
        return False
    with _lock:
        data = _subs_load()
        items = data.get(uid)
        if not isinstance(items, list):
            return False
        new = [s for s in items
               if not isinstance(s, dict)
               or (s.get("endpoint") or "") != (endpoint or "").strip()]
        if len(new) == len(items):
            return False
        data[uid] = new
        return _subs_save(data)


def subscriptions_count(uid) -> int:
    try:
        uid = str(int(uid))
    except (TypeError, ValueError):
        return 0
    with _lock:
        items = _subs_load().get(uid) or []
    return len(items) if isinstance(items, list) else 0


def subs_total() -> int:
    """Всего push-подписок (для health; ключи не трогает)."""
    with _lock:
        data = _subs_load()
    return sum(len(v) for v in (data or {}).values()
               if isinstance(v, list))


def _default_sender(sub, payload: str, key_path: str, subject: str):
    from pywebpush import WebPushException, webpush
    try:
        webpush({"endpoint": sub["endpoint"],
                 "keys": {"p256dh": sub["p256dh"], "auth": sub["auth"]}},
                data=payload,
                vapid_private_key=key_path,
                vapid_claims={"sub": subject},
                ttl=86400)
        return None
    except WebPushException as e:
        return e
    except Exception as e:  # noqa: BLE001
        return e


def _gone(exc) -> bool:
    try:
        resp = getattr(exc, "response", None)
        code = getattr(resp, "status_code", None)
        return code in (404, 410)
    except Exception:  # noqa: BLE001
        return False


def fan_out(get_items, sender=None) -> dict:
    """Разослать свежий инбокс по подпискам. get_items(uid) -> [записи].

    Возвращает {"sent": n, "gone": m}. sender(sub, payload, key_path,
    subject) -> None|exc (для тестов). Без ключей/подписок — нули.
    """
    key_path = ensure_key()
    if not key_path:
        return {"sent": 0, "gone": 0}
    try:
        from pywebpush import WebPushException  # noqa: F401
    except ImportError:
        _disabled_once("нет pywebpush")
        return {"sent": 0, "gone": 0}
    with _lock:
        data = _subs_load()
        uids = [u for u, v in data.items()
                if isinstance(v, list) and v]
    if not uids:
        return {"sent": 0, "gone": 0}
    send = sender or _default_sender
    sent, gone_eps = 0, []
    for uid in uids:
        try:
            items = [r for r in (get_items(uid) or [])
                     if isinstance(r, dict) and not r.get("read")]
        except Exception:  # noqa: BLE001
            continue
        if not items:
            continue
        top = items[0]
        title = (top.get("title") or "Уведомление")[:120]
        body = (top.get("body") or "")[:200]
        payload = json.dumps({"id": top.get("id") or "",
                              "title": title, "body": body,
                              "tag": "botmod-nt-%s" % (top.get("id") or ""),
                              "url": "/", "unread": len(items)},
                             ensure_ascii=False)
        with _lock:
            subs = list((_subs_load().get(uid) or []))
        for sub in subs:
            if not isinstance(sub, dict) or not sub.get("endpoint"):
                continue
            try:
                exc = send(dict(sub), payload, key_path, _subject)
            except Exception as e:  # noqa: BLE001
                exc = e
            if exc is None:
                sent += 1
            elif _gone(exc):
                gone_eps.append((uid, sub.get("endpoint") or ""))
            else:
                logger.warning("notify webpush %s: %s",
                               (sub.get("endpoint") or "")[:60],
                               str(exc)[:120])
    if gone_eps:
        with _lock:
            data = _subs_load()
            for uid, ep in gone_eps:
                items = data.get(uid)
                if isinstance(items, list):
                    data[uid] = [s for s in items
                                 if not isinstance(s, dict)
                                 or s.get("endpoint") != ep]
            _subs_save(data)
    return {"sent": sent, "gone": len(gone_eps)}


def fan_out_async(get_items) -> None:
    """fan_out в фоне (не блокировать engine.send)."""
    try:
        t = threading.Thread(target=fan_out, args=(get_items,),
                             daemon=True, name="notify-webpush")
        t.start()
    except Exception as e:  # noqa: BLE001
        logger.warning("notify webpush thread: %s", e)