cmshoppe/app/appconfig.py

"""Application-level configuration for cmshopee.

This module owns `config.json`, which stores local app settings and AI role/
generation parameters. AI provider definitions and API keys are intentionally
kept out of this file; T-005 will own `config/ai_models.json`.
"""

import copy
import json
import os


CONFIG_PATH = "config.json"

DEFAULT_CONFIG = {
    "chrome_path": r"C:\Program Files\Google\Chrome\Application\chrome.exe",
    "user_data_root": "chrome_user_data_dir",
    "image_dir": "images",
    "db_path": "cmshopee.db",
    "default_debug_port": 9222,
    "debug_port_range": [9222, 9260],
    "cdp_ready_timeout": 60,
    "ai": {
        "default_text_model": "GPT-5.5 文本",
        "default_image_model": "Nano Banana 2",
        "title_concurrency": 4,
        "image_concurrency": 4,
        "retry": 2,
        "jpg_quality": 90,
        "resolution": "1k",
        "resolution_timeouts": {
            "512": 180,
            "1k": 240,
            "2k": 360,
            "4k": 600,
        },
    },
}

SECRET_FIELD_NAMES = {"api_key", "apikey", "key", "token", "password"}


class ConfigError(RuntimeError):
    """Raised when app configuration is missing or malformed."""


def default_config() -> dict:
    """Return a new copy of the default config."""

    return copy.deepcopy(DEFAULT_CONFIG)


def _deep_merge(defaults, loaded):
    if not isinstance(defaults, dict):
        return copy.deepcopy(loaded) if loaded is not None else copy.deepcopy(defaults)
    merged = copy.deepcopy(defaults)
    if not isinstance(loaded, dict):
        return merged
    for key, value in loaded.items():
        if isinstance(merged.get(key), dict) and isinstance(value, dict):
            merged[key] = _deep_merge(merged[key], value)
        else:
            merged[key] = copy.deepcopy(value)
    return merged


def _assert_no_secrets(config):
    def visit(value, path):
        if isinstance(value, dict):
            for key, child in value.items():
                lowered = str(key).lower()
                if lowered in SECRET_FIELD_NAMES or lowered.endswith("_key"):
                    raise ConfigError(
                        f"config.json 不允许保存敏感字段: {'.'.join(path + [str(key)])}"
                    )
                visit(child, path + [str(key)])
        elif isinstance(value, list):
            for index, child in enumerate(value):
                visit(child, path + [str(index)])

    visit(config, [])


def save_config(config, path=CONFIG_PATH) -> dict:
    """Persist config to JSON and return the normalized config."""

    normalized = _deep_merge(DEFAULT_CONFIG, config)
    _assert_no_secrets(normalized)
    directory = os.path.dirname(os.path.abspath(path))
    if directory:
        os.makedirs(directory, exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(normalized, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    return normalized


def load_config(path=CONFIG_PATH) -> dict:
    """Load config, writing defaults first if the file does not exist."""

    if not os.path.exists(path):
        return save_config(default_config(), path=path)
    with open(path, "r", encoding="utf-8") as fh:
        try:
            loaded = json.load(fh)
        except json.JSONDecodeError as exc:
            raise ConfigError(f"配置文件不是有效 JSON: {path}") from exc
    normalized = _deep_merge(DEFAULT_CONFIG, loaded)
    _assert_no_secrets(normalized)
    return normalized


def update_config(updates, path=CONFIG_PATH) -> dict:
    """Merge updates into the persisted config."""

    config = load_config(path)
    return save_config(_deep_merge(config, updates), path=path)


def _config_or_load(config):
    return load_config() if config is None else config


def chrome_path(config=None) -> str:
    return _config_or_load(config).get("chrome_path", "")


def user_data_root(config=None) -> str:
    return _config_or_load(config).get("user_data_root", "chrome_user_data_dir")


def image_dir(config=None) -> str:
    return _config_or_load(config).get("image_dir", "images")


def db_path(config=None) -> str:
    return _config_or_load(config).get("db_path", "cmshopee.db")


def default_debug_port(config=None) -> int:
    return int(_config_or_load(config).get("default_debug_port", 9222))


def debug_port_range(config=None) -> tuple:
    values = _config_or_load(config).get("debug_port_range", [9222, 9260])
    if not isinstance(values, list) or len(values) != 2:
        raise ConfigError("debug_port_range 必须是 [start, end]")
    return int(values[0]), int(values[1])


def cdp_ready_timeout(config=None) -> int:
    return int(_config_or_load(config).get("cdp_ready_timeout", 60))


def ai_config(config=None) -> dict:
    return copy.deepcopy(_config_or_load(config).get("ai", DEFAULT_CONFIG["ai"]))


def response_timeout(config=None) -> int:
    ai = ai_config(config)
    resolution = str(ai.get("resolution", DEFAULT_CONFIG["ai"]["resolution"]))
    timeouts = ai.get("resolution_timeouts", {})
    if resolution not in timeouts:
        raise ConfigError(f"未配置分辨率 {resolution} 的返回超时")
    return int(timeouts[resolution])