core/config.py

import json
import os
import hashlib
import secrets
import copy
from pathlib import Path

DEFAULT_UPSTREAM_URL = "https://web.tabbitbrowser.com"
CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"

# Edition presets (国际版 / 国内版)
EDITION_PRESETS = {
    "international": {
        "base_url": "https://web.tabbitbrowser.com",
        "client_id": "e7fa44387b1238ef1f6f",
    },
    "cn": {
        "base_url": "https://web.tabbitbrowser.cn",
        "client_id": "e7fa44387b1238ef1f6f",  # 国内版 client_id 可能与国际版不同，这里占位
    },
}

DEFAULT_CONFIG = {
    "server": {"host": "0.0.0.0", "port": 8800},
    "admin": {"password_hash": "", "salt": "", "jwt_secret": ""},
    "noke": {
        "edition": "international",  # 新增，用于预设
        "base_url": DEFAULT_UPSTREAM_URL,
        "client_id": "e7fa44387b1238ef1f6f",
        "verify_tls": True,
    },
    "tokens": [],
    "proxy": {"api_key": "", "system_prompt": "", "rate_limit_qps": 0, "rate_limit_burst": 0},
    "claude": {"default_model": "best", "system_prompt": ""},
    "logging": {"max_entries": 500},
}


def _deep_merge(base: dict, override: dict) -> dict:
    result = base.copy()
    for key, value in override.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = _deep_merge(result[key], value)
        else:
            result[key] = value
    return result


def hash_password(password: str, salt: str | None = None) -> tuple[str, str]:
    if salt is None:
        salt = secrets.token_hex(16)
    hashed = hashlib.sha256((password + salt).encode()).hexdigest()
    return hashed, salt


def _apply_edition(config: dict) -> dict:
    """根据 edition 字段填充 base_url/client_id（仅当用户未显式覆盖时）。"""
    edition = config.get("noke", {}).get("edition", "international")
    preset = EDITION_PRESETS.get(edition)
    if not preset:
        return config
    # 仅当当前值与默认值一致时才覆写（即用户未手动修改）
    default_noke = DEFAULT_CONFIG.get("noke", {})
    if config["noke"]["base_url"] == default_noke.get("base_url"):
        config["noke"]["base_url"] = preset["base_url"]
    if config["noke"]["client_id"] == default_noke.get("client_id"):
        config["noke"]["client_id"] = preset["client_id"]
    return config


def _apply_env_overrides(config: dict) -> dict:
    """从环境变量覆盖所有可配置项，环境变量优先级最高"""

    # 服务器
    if v := os.environ.get("SERVER_HOST"):
        config["server"]["host"] = v
    if v := os.environ.get("SERVER_PORT"):
        config["server"]["port"] = int(v)

    # 管理员认证
    if v := os.environ.get("ADMIN_PASSWORD_HASH"):
        config["admin"]["password_hash"] = v
    if v := os.environ.get("ADMIN_SALT"):
        config["admin"]["salt"] = v
    if v := os.environ.get("JWT_SECRET"):
        config["admin"]["jwt_secret"] = v

    # Edition (新增)
    if v := os.environ.get("EDITION"):
        if v in EDITION_PRESETS:
            config["noke"]["edition"] = v

    # Upstream
    if v := os.environ.get("UPSTREAM_BASE_URL"):
        config["noke"]["base_url"] = v
    if v := os.environ.get("UPSTREAM_CLIENT_ID"):
        config["noke"]["client_id"] = v
    # 兼容新名 VERIFY_TLS，同时保留旧名 UPSTREAM_VERIFY_TLS
    verify_val = os.environ.get("VERIFY_TLS") or os.environ.get("UPSTREAM_VERIFY_TLS")
    if verify_val:
        config["noke"]["verify_tls"] = verify_val.lower() in ("true", "1", "yes")

    # 代理
    if v := os.environ.get("PROXY_API_KEY"):
        config["proxy"]["api_key"] = v
    if v := os.environ.get("PROXY_SYSTEM_PROMPT"):
        config["proxy"]["system_prompt"] = v

    # Claude
    if v := os.environ.get("CLAUDE_DEFAULT_MODEL"):
        config["claude"]["default_model"] = v
    if v := os.environ.get("CLAUDE_SYSTEM_PROMPT"):
        config["claude"]["system_prompt"] = v

    # Tokens（JSON 数组格式）
    if v := os.environ.get("UPSTREAM_TOKENS"):
        try:
            config["tokens"] = json.loads(v)
        except Exception:
            pass

    # 日志
    if v := os.environ.get("LOG_MAX_ENTRIES"):
        config["logging"]["max_entries"] = int(v)

    return config


class ConfigManager:
    def __init__(self, path: str | Path | None = None):
        self.path = Path(path) if path else CONFIG_PATH
        self.config = self._load()

    def _load(self) -> dict:
        if self.path.exists():
            with open(self.path, "r", encoding="utf-8") as f:
                saved = json.load(f)
            config = _deep_merge(copy.deepcopy(DEFAULT_CONFIG), saved)
            # 应用 edition 预设（填充 base_url/client_id 默认值）
            config = _apply_edition(config)
            self._save(config)
        else:
            config = copy.deepcopy(DEFAULT_CONFIG)
            config["admin"]["jwt_secret"] = secrets.token_hex(32)
            admin_pw = os.environ.get("NOKE_ADMIN_PASSWORD") or secrets.token_urlsafe(12)
            pw_hash, salt = hash_password(admin_pw)
            config["admin"]["password_hash"] = pw_hash
            config["admin"]["salt"] = salt
            self._save(config)
            if not os.environ.get("NOKE_ADMIN_PASSWORD"):
                print(f"\n[Noke] 初始管理员密码: {admin_pw} （仅首次显示，请立即保存并登录修改）\n")

        # 环境变量最后覆盖，优先级最高
        config = _apply_env_overrides(config)
        return config

    def _save(self, config: dict | None = None):
        if config is None:
            config = self.config
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(config, f, indent=2, ensure_ascii=False)

    def save(self):
        self._save()

    def get(self, *keys, default=None):
        val = self.config
        for k in keys:
            if isinstance(val, dict):
                val = val.get(k)
            else:
                return default
            if val is None:
                return default
        return val

    def set_val(self, *keys_and_value):
        keys = keys_and_value[:-1]
        value = keys_and_value[-1]
        d = self.config
        for k in keys[:-1]:
            d = d.setdefault(k, {})
        d[keys[-1]] = value
        self.save()