access v0.1.0 [core]

"""access.engine — единый владелец ADMIN_IDS и операций над правами.

Единственное место, где живёт список TG-администраторов: все модули
(admin-движок, сканер сети, notify, веб-панель) обращаются сюда через
ctx.api, а не к своим копиям ADMIN_IDS. Файл прав принадлежит ядру
(core.rights): чтение — живое (без TTL-кешей), запись — диффом через
сервис (атомарно, с mtime-перечитыванием у остальных).
"""
import logging
import re

logger = logging.getLogger(__name__)

_rights = None  # RightsService (ставит module.setup)
_admin_ids = []  # [int] TG-администраторы
_env_path = None  # modules/access/.env (для add/remove через веб/TG)

_KEY_RE = re.compile(r"^(telegram|vk|web):\d+$")


def configure(rights, admin_ids, env_path=None) -> None:
    global _rights, _admin_ids, _env_path
    _rights = rights
    _admin_ids = [int(x) for x in (admin_ids or [])]
    _env_path = env_path


def _norm(uid) -> str:
    uid = str(uid)
    if ":" in uid:
        return uid
    return f"telegram:{uid}"


def admin_ids():
    """Список TG id администраторов (копия)."""
    return list(_admin_ids)


def is_admin(uid) -> bool:
    """Входит ли uid в ADMIN_IDS (семантика старого AdminMenu.is_admin)."""
    try:
        num = int(str(uid).split(":")[-1])
    except (ValueError, TypeError):
        return False
    return num in _admin_ids


def load_rights() -> dict:
    """Все записи прав {qid: entry} (живьём из ядра)."""
    try:
        return {qid: _rights.get_entry(qid) for qid in _rights.known_users()}
    except Exception:  # noqa: BLE001
        return {}


def guest_defaults() -> dict:
    """Гостевые дефолты новой записи (динамика из реестра ядра)."""
    return _defaults()


def get_entry(key) -> dict:
    try:
        return _rights.get_entry(key) or {}
    except Exception:  # noqa: BLE001
        return {}


def get_right(uid, key) -> bool:
    """Право пользователя (telegram-ns по умолчанию; живьём)."""
    try:
        return bool(_rights.can(_norm(uid), key))
    except Exception:  # noqa: BLE001
        return False


def _defaults() -> dict:
    """Гостевые дефолты из реестра прав ядра (динамика вместо статики)."""
    try:
        if _rights is not None:
            return _rights.guest_defaults()
    except Exception:  # noqa: BLE001
        pass
    return {"all": False}


def toggle_right(key, right) -> bool:
    """Переключить право; вернуть новое значение (семантика toggle_admin_right)."""
    if not _KEY_RE.match(key or ""):
        raise ValueError("Некорректный ключ пользователя")
    entry = get_entry(key)
    if not entry:
        entry = _defaults()
    entry[right] = not entry.get(right, False)
    _rights.set_entry(key, entry, actor="system:legacy")
    return bool(entry[right])


def save_rights(rights: dict) -> None:
    """Полная запись словаря прав (для вендорного движка): дифф через сервис."""
    current = load_rights()
    for key in list(current):
        if key not in rights:
            try:
                _rights.delete_entry(key, actor="system:seed")
            except Exception:  # noqa: BLE001
                pass
    for key, entry in (rights or {}).items():
        try:
            _rights.set_entry(key, dict(entry), actor="system:seed")
        except Exception as e:  # noqa: BLE001
            logger.warning("access save %s: %s", key, e)


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


def seed() -> None:
    """Синхронизировать права со списком ADMIN_IDS (бывший check_and_update_*).

    - каждому админу из списка — запись с недостающими ключами из
      гостевых дефолтов ядра (динамика);
    - telegram:* вне списка — удалить (web:/vk: не трогаем);
    - pve_manage_hosts дотягивает pve-модуль (свои хосты знает только он).
    """
    rights = load_rights()
    modified = False
    for admin in _admin_ids:
        key = f"telegram:{admin}"
        entry = rights.get(key)
        if entry is None:
            rights[key] = _defaults()
            modified = True
        else:
            for k, default_value in _defaults().items():
                if k not in entry:
                    entry[k] = default_value
                    modified = True
    valid = {f"telegram:{a}" for a in _admin_ids}
    for qid in list(rights):
        if qid.startswith("telegram:") and qid not in valid:
            del rights[qid]
            modified = True
    if modified:
        save_rights(rights)
        logger.info("access: права синхронизированы с ADMIN_IDS.")


def _write_env(ids) -> None:
    """Перезаписать строку ADMIN_IDS в .env модуля (сохраняя остальное)."""
    import os as _os
    if not _env_path:
        raise RuntimeError("access .env недоступен")
    with open(_env_path, "r", encoding="utf-8") as f:
        lines = f.read().splitlines()
    new_lines, done = [], False
    for ln in lines:
        if re.match(r"^\s*ADMIN_IDS\s*=", ln):
            new_lines.append("ADMIN_IDS=" + ",".join(str(x) for x in ids))
            done = True
        else:
            new_lines.append(ln)
    if not done:
        new_lines.append("ADMIN_IDS=" + ",".join(str(x) for x in ids))
    tmp = _env_path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write("\n".join(new_lines) + "\n")
    _os.replace(tmp, _env_path)


def add_admin_id(uid) -> tuple:
    """Добавить TG-админа: .env + seed. Вернуть (changed, msg)."""
    global _admin_ids
    try:
        num = int(str(uid).strip())
    except (ValueError, TypeError):
        raise ValueError("ID должен быть числом")
    if num in _admin_ids:
        return False, f"Администратор {num} уже в списке"
    _admin_ids.append(num)
    _write_env(_admin_ids)
    seed()
    return True, (f"Администратор {num} добавлен (вступит после перезапуска; "
                   "права выданы сразу)")


def remove_admin_id(uid) -> tuple:
    """Убрать TG-админа: .env + права. Вернуть (cfg_changed, rights_changed)."""
    global _admin_ids
    try:
        num = int(str(uid).strip())
    except (ValueError, TypeError):
        raise ValueError("ID должен быть числом")
    cfg_changed = False
    if num in _admin_ids:
        _admin_ids = [x for x in _admin_ids if x != num]
        _write_env(_admin_ids)
        cfg_changed = True
    rights_changed = False
    try:
        if _rights.delete_entry(f"telegram:{num}", actor="system:seed"):
            rights_changed = True
    except Exception:  # noqa: BLE001
        pass
    if not cfg_changed and not rights_changed:
        raise LookupError(f"Администратор {num} не найден")
    return cfg_changed, rights_changed

"""access.Module — единый владелец ADMIN_IDS и операций над правами.

Файл прав принадлежит ядру; этот модуль — API-слой: admin_ids/is_admin,
get_right/toggle_right, notify_targets, seed(). Сидинг — в start():
реестр прав полон только после всех setup'ов.
"""
import logging
import os

from core.base_module import BaseModule

from . import engine as access

logger = logging.getLogger(__name__)

_DEFAULT_ADMIN_IDS = ""  # пусто: админы только из access.env (дефолтов нет)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("access", mod_dir)
        self.ctx = ctx
        raw = (self.config.get("ADMIN_IDS", "") or "").strip() or _DEFAULT_ADMIN_IDS
        ids = [int(x) for x in raw.split(",") if x.strip().isdigit()]
        try:
            env_path = os.path.join(ctx.data_dir, "env", "access.env")
        except Exception:  # noqa: BLE001
            env_path = os.path.join(mod_dir, ".env")
        access.configure(ctx.rights, ids, env_path)
        ctx.api.register("access", "admin_ids", access.admin_ids)
        ctx.api.register("access", "is_admin", access.is_admin)
        ctx.api.register("access", "get_right", access.get_right)
        ctx.api.register("access", "toggle_right", access.toggle_right)
        ctx.api.register("access", "notify_targets", access.notify_targets)
        ctx.api.register("access", "seed", access.seed)
        ctx.api.register("access", "add_admin", access.add_admin_id)
        ctx.api.register("access", "remove_admin", access.remove_admin_id)

    def start(self) -> None:
        # Реестр прав собирается по ходу setup'ов — сидим здесь,
        # когда все манифесты уже зарегистрированы.
        try:
            access.seed()
        except Exception as e:  # noqa: BLE001
            logger.warning("access seed: %s", e)

    def health(self) -> dict:
        return {"ok": True, "module": self.name, "admins": len(access.admin_ids())}