config_util.py 6.45 KB
"""
config.json 安全加载器。

公开接口:
- DEFAULT_CONFIG: 默认配置常量
- load_config_safe(path): 区分处理各类 IO / JSON 错误,返回 (config, error_message)
- get_config_dir(): 跨平台返回配置目录
- get_config_path(): 返回 config.json 绝对路径
"""
from __future__ import annotations

import json
import logging
import os
import shutil
import sys
import platform
from datetime import datetime
from pathlib import Path
from typing import Optional, Tuple


logger = logging.getLogger(__name__)


DEFAULT_CONFIG: dict = {
    "api_key": "",
    "saved_prompts": [],
    "db_config": None,
    "last_user": "",
    "saved_password_hash": "",
    "logging_config": {
        "enabled": True,
        "level": "INFO",
        "log_to_console": True,
    },
    "history_config": {
        "max_history_count": 100,
    },
}


def get_config_dir() -> Path:
    """跨平台返回用户级配置目录(与 image_generator.py 中 get_config_dir 一致)。"""
    if getattr(sys, "frozen", False):
        system = platform.system()
        if system == "Darwin":
            d = Path.home() / "Library" / "Application Support" / "ZB100ImageGenerator"
        elif system == "Windows":
            d = Path(os.getenv("APPDATA", str(Path.home()))) / "ZB100ImageGenerator"
        else:
            d = Path.home() / ".config" / "zb100imagegenerator"
    else:
        d = Path(".").resolve()
    d.mkdir(parents=True, exist_ok=True)
    return d


def get_config_path() -> Path:
    return get_config_dir() / "config.json"


def load_config_safe(config_path: Path) -> Tuple[dict, str]:
    """
    安全加载 config.json。

    返回 (config, error):
    - 成功:(合并后的 config dict, "")
    - 失败但可恢复:(DEFAULT_CONFIG 副本, 错误描述);preflight 会根据返回
      的 error 和 config 内容共同决定是否拦截启动
    - 不抛异常

    行为:
    - 文件不存在 → (DEFAULT_CONFIG 副本, "")  # 让 preflight 判断是否允许
    - 空文件 / JSON 损坏 → 备份为 .bak.<timestamp> + 返回默认值 + error
    - PermissionError / OSError → 返回默认值 + error(不备份,读都读不到)
    - 顶层不是 object → 返回默认值 + error
    - 正常 → defaults.update(loaded) 合并后返回(保留未知字段以兼容)
    """
    config_path = Path(config_path)

    if not config_path.exists():
        return dict(DEFAULT_CONFIG), ""

    try:
        content = config_path.read_text(encoding="utf-8")
    except PermissionError as e:
        return dict(DEFAULT_CONFIG), f"permission denied: {e}"
    except OSError as e:
        return dict(DEFAULT_CONFIG), f"IO error: {e}"

    if not content.strip():
        _backup(config_path, reason="empty")
        return dict(DEFAULT_CONFIG), "config.json was empty, using defaults"

    try:
        loaded = json.loads(content)
    except json.JSONDecodeError as e:
        _backup(config_path, reason="parse-error")
        return dict(DEFAULT_CONFIG), f"JSON parse error, backed up to .bak: {e}"

    if not isinstance(loaded, dict):
        return dict(DEFAULT_CONFIG), "config.json top-level is not an object"

    merged = dict(DEFAULT_CONFIG)
    merged.update(loaded)
    return merged, ""


def find_bundled_config() -> Optional[Path]:
    """
    定位打包进二进制的 config.json (只在 frozen 态有意义)。
    查找顺序: macOS app bundle Resources/ → exe 同目录 → PyInstaller _MEIPASS/
    都找不到返回 None。
    """
    if not getattr(sys, "frozen", False):
        return None
    candidates = []
    if platform.system() == "Darwin":
        candidates.append(Path(sys.executable).parent.parent / "Resources" / "config.json")
    else:
        candidates.append(Path(sys.executable).parent / "config.json")
    if hasattr(sys, "_MEIPASS"):
        candidates.append(Path(sys._MEIPASS) / "config.json")
    for p in candidates:
        if p.exists():
            return p
    return None


def sync_bundled_api_key(user_config_path: Path) -> None:
    """
    如果打包 config.json 里的 api_key 和用户目录的不一致,用打包的值覆盖那一个字段。
    其他字段 (saved_prompts / last_user / saved_password_hash / db_config ...) 全部保留。

    只在 frozen (打包态) 下生效 —— 开发态不干涉本地 config.json。

    保护策略 (避免把用户 api_key 清零):
    - 找不到打包 config → 不动
    - 打包 config 解析失败 → 不动
    - 打包 api_key 为空 → 不动
    - 用户 config 不存在 → 不动 (外层的 "从 bundled 拷贝整个文件" 逻辑会处理)
    - 用户 config 解析失败 → 不动 (load_config_safe 会走损坏分支备份)

    写入失败不抛异常,记错误日志。
    """
    if not getattr(sys, "frozen", False):
        return

    bundled = find_bundled_config()
    if bundled is None:
        return

    try:
        bundled_data = json.loads(bundled.read_text(encoding="utf-8"))
    except Exception as e:
        logger.warning(f"打包 config 解析失败, 跳过 api_key 同步: {e}")
        return

    bundled_key = str(bundled_data.get("api_key", "")).strip()
    if not bundled_key:
        logger.warning("打包 config 的 api_key 为空, 跳过同步 (避免清零用户 key)")
        return

    if not user_config_path.exists():
        # 用户 config 还不存在 → 外层的整文件拷贝逻辑会处理
        return

    try:
        user_data = json.loads(user_config_path.read_text(encoding="utf-8"))
    except Exception as e:
        logger.warning(f"用户 config 解析失败, 跳过 api_key 同步 (load_config_safe 会处理): {e}")
        return

    if not isinstance(user_data, dict):
        return

    if user_data.get("api_key") == bundled_key:
        return  # 一致,无需动作

    # 只覆盖 api_key 一个字段, 其他原样保留
    user_data["api_key"] = bundled_key
    try:
        user_config_path.write_text(
            json.dumps(user_data, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
        logger.info(f"[BOOT] api_key 已从 bundled 同步到用户 config: {user_config_path}")
    except Exception as e:
        logger.error(f"[BOOT] 写回用户 config 失败, api_key 未同步: {e}")


def _backup(src: Path, reason: str) -> None:
    """把损坏 / 空的 config 文件备份,不抛异常。"""
    try:
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        dst = src.with_suffix(f".json.bak.{reason}.{ts}")
        shutil.copy2(src, dst)
    except Exception:
        pass