HeXi/hexi/web_hub/bot_settings.py

# -*- coding: utf-8 -*-
"""Bot 全局配置的读取 / 写入（SUPERUSERS / NICKNAME / DEBUG / LOG_LEVEL）。

- 读取：优先取 NoneBot driver.config 的当前生效值，退化到 .env / os.environ。
- 写入：落盘到 .env（保留 JSON 列表格式），写 os.environ（即时可读），
  并在可能范围内热应用到 driver.config 与 loguru（LOG_LEVEL）。
- 注意：个别插件在 import 时把 SUPERUSERS/NICKNAME 拷贝成了常量，
  想让所有插件完全生效建议保存后重启 bot（前端会给出提示）。
"""

from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any

from nonebot import get_driver, logger

_REPO_ROOT = Path(__file__).resolve().parents[2]
# 仓库根目录的 .env（NoneBot 启动时自动加载)
_ENV_PATH = _REPO_ROOT / ".env"

# 表单键 -> 环境变量名
_KEY_ENV: dict[str, str] = {
    "superusers": "SUPERUSERS",
    "nickname": "NICKNAME",
    "debug": "DEBUG",
    "log_level": "LOG_LEVEL",
}

# LOG_LEVEL 可选项（写入时建议大写）
LOG_LEVEL_OPTIONS = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

# 日志清理大小（MB）默认值（未在 .env 配置时）
_LOG_MAX_MB_DEFAULT = 20


def _env_get(name: str, default: str = "") -> str:
    """os.environ > .env 文件 > default。"""
    val = os.getenv(name)
    if val:
        return val
    if _ENV_PATH.exists():
        for line in _ENV_PATH.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if line.startswith(name + "=") or line.startswith(name + " ="):
                return line.split("=", 1)[1].strip().split(" #", maxsplit=1)[0].strip()
    return default


def _parse_list(value: str) -> list[str]:
    """把 env 里可能是 JSON 数组或逗号/空白分隔的列表解析成 list[str]。"""
    value = (value or "").strip()
    if not value:
        return []
    try:
        parsed = json.loads(value)
        if isinstance(parsed, list):
            return [str(x) for x in parsed]
        return [str(parsed)]
    except Exception:
        # 兼容逗号/空格分隔、带引号的形式
        return [
            part.strip().strip("\"'").strip()
            for part in value.replace(" ", ",").split(",")
            if part.strip()
        ]


def _serialize_list(items: list[str]) -> str:
    """序列化成 .env 使用的 JSON 数组字符串。"""
    return json.dumps([str(x) for x in items], ensure_ascii=False)


def _normalize_log_level(value: Any) -> str:
    """把 log_level 统一成大写字符串。"""
    if isinstance(value, int):
        # loguru 数字等级：20=DEBUG 30=INFO 40=WARNING 50=ERROR
        mapping = {10: "TRACE", 20: "DEBUG", 30: "INFO", 40: "WARNING", 50: "ERROR", 60: "CRITICAL"}
        return mapping.get(value, "INFO")
    return str(value).upper()


def _current_log_max_mb() -> int:
    """当前日志清理大小（MB）。优先级：HEXI_LOG_MAX_MB > HEXI_LOG_MAX_BYTES(换算) > 默认。"""
    raw = _env_get("HEXI_LOG_MAX_MB").strip()
    if raw:
        try:
            val = float(raw)
            if val > 0:
                return int(val)
        except ValueError:
            pass
    raw_bytes = _env_get("HEXI_LOG_MAX_BYTES").strip()
    if raw_bytes:
        try:
            b = int(raw_bytes)
            if b > 0:
                return max(1, b // (1024 * 1024))
        except ValueError:
            pass
    return _LOG_MAX_MB_DEFAULT


def current_values() -> dict[str, Any]:
    """返回当前生效值（读取优先级：driver.config > .env）。"""
    try:
        cfg = get_driver().config
    except Exception:
        cfg = None

    if cfg is not None:
        superusers = sorted(str(x) for x in getattr(cfg, "superusers", []) or [])
        nickname = sorted(str(x) for x in getattr(cfg, "nickname", []) or [])
        debug = bool(getattr(cfg, "debug", False))
        log_level = _normalize_log_level(getattr(cfg, "log_level", "INFO"))
    else:
        superusers = _parse_list(_env_get("SUPERUSERS"))
        nickname = _parse_list(_env_get("NICKNAME"))
        debug = _env_get("DEBUG", "").strip().lower() in ("1", "true", "yes", "on")
        log_level = _normalize_log_level(_env_get("LOG_LEVEL", "INFO"))

    return {
        "superusers": superusers,
        "nickname": nickname,
        "debug": bool(debug),
        "log_level": log_level,
        "log_level_options": LOG_LEVEL_OPTIONS,
        "log_max_mb": _current_log_max_mb(),
        "log_max_mb_default": _LOG_MAX_MB_DEFAULT,
        "source": "config" if cfg is not None else "env",
    }


def _set_env_line(name: str, value: str) -> None:
    """替换 .env 中已有的 name= 行（保留其它行），不存在则追加。"""
    try:
        if _ENV_PATH.exists():
            lines = _ENV_PATH.read_text(encoding="utf-8").splitlines()
        else:
            lines = []
        replaced = False
        out: list[str] = []
        for line in lines:
            stripped = line.strip()
            if stripped.startswith(name + "=") or stripped.startswith(name + " ="):
                if not replaced:
                    out.append(f"{name}={value}")
                    replaced = True
                # 重复行跳过
            else:
                out.append(line)
        if not replaced:
            out.append(f"{name}={value}")
        out_text = "\n".join(out).rstrip("\n") + "\n"
        _ENV_PATH.write_text(out_text, encoding="utf-8")
    except Exception as e:  # noqa: BLE001
        logger.warning(f"写入 .env({name}) 失败: {type(e).__name__}: {e}")


def _apply_log_level(log_level: str) -> None:
    """把 LOG_LEVEL 热应用到 loguru 的默认过滤配置。"""
    try:
        from nonebot.log import logger as nb_logger

        nb_logger.configure(extra={"nonebot_log_level": log_level})
    except Exception as e:  # noqa: BLE001
        logger.warning(f"热应用 LOG_LEVEL 失败: {type(e).__name__}: {e}")


def save_bot_settings(values: dict[str, Any]) -> dict[str, Any]:
    """保存并尽量热应用。返回（新的当前值 + 提示）。"""
    # 规范化输入
    superusers = [str(x) for x in (values.get("superusers") or []) if str(x).strip()]
    nickname = [str(x) for x in (values.get("nickname") or []) if str(x).strip()]
    debug = bool(values.get("debug", False))
    log_level = _normalize_log_level(values.get("log_level", "INFO"))
    if log_level not in LOG_LEVEL_OPTIONS:
        raise ValueError(f"LOG_LEVEL 必须是 {'/'.join(LOG_LEVEL_OPTIONS)} 之一")

    # 日志清理大小（MB，写入 .env 的 HEXI_LOG_MAX_MB）
    log_max_mb: int | None = None
    _log_max_mb_raw = values.get("log_max_mb")
    if _log_max_mb_raw is not None and str(_log_max_mb_raw).strip() != "":
        try:
            log_max_mb = int(float(_log_max_mb_raw))
        except (TypeError, ValueError):
            raise ValueError("日志清理大小必须是正整数（MB）")
        if log_max_mb <= 0:
            raise ValueError("日志清理大小必须是正整数（MB）")

    # 1. 写 os.environ（即时可读，进程内 getenv 立刻生效）
    os.environ["SUPERUSERS"] = _serialize_list(superusers)
    os.environ["NICKNAME"] = _serialize_list(nickname)
    os.environ["DEBUG"] = "true" if debug else "false"
    os.environ["LOG_LEVEL"] = log_level
    if log_max_mb is not None:
        os.environ["HEXI_LOG_MAX_MB"] = str(log_max_mb)

    # 2. 落盘 .env（重启后依然生效）
    _set_env_line("SUPERUSERS", _serialize_list(superusers))
    _set_env_line("NICKNAME", _serialize_list(nickname))
    _set_env_line("DEBUG", "true" if debug else "false")
    _set_env_line("LOG_LEVEL", log_level)
    if log_max_mb is not None:
        _set_env_line("HEXI_LOG_MAX_MB", str(log_max_mb))

    # 3. 热应用到 driver.config
    applied: list[str] = []
    try:
        cfg = get_driver().config
        if hasattr(cfg, "superusers"):
            cfg.superusers = set(superusers)
            applied.append("SUPERUSERS")
        if hasattr(cfg, "nickname"):
            cfg.nickname = set(nickname)
            applied.append("NICKNAME")
        # DEBUG 在 NoneBot Config 中属于 extra 字段，直接透传
        try:
            setattr(cfg, "debug", debug)
            applied.append("DEBUG")
        except Exception:
            pass
        if hasattr(cfg, "log_level"):
            cfg.log_level = log_level
            applied.append("LOG_LEVEL")
    except Exception as e:  # noqa: BLE001
        logger.warning(f"热应用 driver.config 失败: {type(e).__name__}: {e}")

    # 4. LOG_LEVEL 应用到 loguru
    _apply_log_level(log_level)
    if log_max_mb is not None:
        applied.append("HEXI_LOG_MAX_MB")

    return {
        "ok": True,
        **current_values(),
        "applied": applied,
        "restart_required": True,
        "notice": "已保存到 .env 并尽量热应用；部分插件在 import 时缓存了 SUPERUSERS/NICKNAME，"
        "如需完全生效，请在下方执行「重启 bot」。",
    }