"""Application-level configuration for cmshopee. This module owns `config.json` plus `config/ai_models.json`. `config.json` stores local app settings and AI role/generation parameters. AI provider definitions and local plaintext API keys live in ignored `config/ai_models.json`. """ import copy import json import os import urllib.error import urllib.request CONFIG_PATH = "config.json" AI_MODELS_PATH = os.path.join("config", "ai_models.json") CATEGORIES = {"text", "image"} API_TYPES = {"chat", "images_edits", "auto"} 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, }, }, "shopee_update": { "test_item_id": "51100639510", "allow_real_submit": False, "allow_cover_update": False, "max_items_per_run": 1, "close_success_tab": False, "dry_run": False, "parallel_accounts": False, "max_parallel_accounts": 2, }, } DEFAULT_AI_MODELS_CONFIG = { "models": [ { "name": "GPT-5.5 文本", "category": "text", "enabled": True, "url": "", "model": "", "api_key": "", "api_type": "chat", "connect_timeout_seconds": 30, "timeout_seconds": 0, "extra_body": {}, }, { "name": "Nano Banana 2", "category": "image", "enabled": True, "url": "https://api.vectorengine.ai/v1/chat/completions", "model": "gemini-3.1-flash-image-preview", "api_key": "", "api_type": "auto", "connect_timeout_seconds": 30, "timeout_seconds": 0, "extra_body": {}, }, ] } 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 default_ai_models_config() -> dict: """Return a new copy of the default AI model list.""" return copy.deepcopy(DEFAULT_AI_MODELS_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 _is_secret_field_name(lowered): 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 _is_secret_field_name(name) -> bool: lowered = str(name).lower() return lowered in SECRET_FIELD_NAMES or lowered.endswith( ("_key", "_token", "_password") ) def mask_secret(secret) -> str: """Return a display-safe representation of a secret value.""" if secret is None: return "" if isinstance(secret, (dict, list, tuple, set)): return "***" if secret else "" text = str(secret or "") if not text: return "" if len(text) <= 8: return "***" return f"{text[:4]}***{text[-4:]}" def sanitize_for_log(value): """Return a copy of value with secret fields masked for logs/status/export.""" if isinstance(value, dict): sanitized = {} for key, child in value.items(): if _is_secret_field_name(key): sanitized[key] = mask_secret(child) else: sanitized[key] = sanitize_for_log(child) return sanitized if isinstance(value, list): return [sanitize_for_log(item) for item in value] if isinstance(value, tuple): return tuple(sanitize_for_log(item) for item in value) return value def redact_secrets(text, secret_values=None) -> str: """Replace known secret values inside free-form text.""" redacted = str(text) for secret in secret_values or []: secret_text = str(secret or "") if secret_text: redacted = redacted.replace(secret_text, "***") return redacted 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]) def _normalize_ai_model(model): if not isinstance(model, dict): raise ConfigError("AI 模型定义必须是对象") normalized = { "name": str(model.get("name", "")).strip(), "category": str(model.get("category", "")).strip(), "enabled": bool(model.get("enabled", True)), "url": str(model.get("url", "")).strip(), "model": str(model.get("model", "")).strip(), "api_key": str(model.get("api_key", "")), "api_type": str(model.get("api_type", "auto")).strip() or "auto", "connect_timeout_seconds": int(model.get("connect_timeout_seconds", 30) or 30), "timeout_seconds": int(model.get("timeout_seconds", 0) or 0), "extra_body": copy.deepcopy(model.get("extra_body", {})), } if not normalized["name"]: raise ConfigError("AI 模型 name 不能为空") if normalized["category"] not in CATEGORIES: raise ConfigError("AI 模型 category 必须是 text 或 image") if normalized["api_type"] not in API_TYPES: raise ConfigError("AI 模型 api_type 必须是 chat、images_edits 或 auto") if normalized["connect_timeout_seconds"] <= 0: raise ConfigError("AI 模型 connect_timeout_seconds 必须大于 0") if normalized["timeout_seconds"] < 0: raise ConfigError("AI 模型 timeout_seconds 不能小于 0") if not isinstance(normalized["extra_body"], dict): raise ConfigError("AI 模型 extra_body 必须是对象") return normalized def _normalize_ai_models_config(config): models = config.get("models") if isinstance(config, dict) else None if not isinstance(models, list): raise ConfigError("config/ai_models.json 必须包含 models 列表") normalized = {"models": [_normalize_ai_model(model) for model in models]} _assert_unique_model_names(normalized["models"]) _assert_required_categories(normalized["models"]) return normalized def _assert_unique_model_names(models): names = [model["name"] for model in models] duplicates = sorted({name for name in names if names.count(name) > 1}) if duplicates: raise ConfigError(f"AI 模型 name 重复: {', '.join(duplicates)}") def _assert_required_categories(models): enabled_categories = { model["category"] for model in models if model.get("enabled", True) } missing = sorted(CATEGORIES - enabled_categories) if missing: raise ConfigError( "AI 模型清单至少需要启用一个 text 和一个 image 模型，缺少: " + ", ".join(missing) ) def _model_index(models, name): for index, model in enumerate(models): if model["name"] == name: return index raise ConfigError(f"AI 模型不存在: {name}") def _mask_api_key(api_key): return mask_secret(api_key) def _public_model(model): public = copy.deepcopy(model) public["api_key"] = _mask_api_key(public.get("api_key", "")) public["api_key_set"] = bool(model.get("api_key")) return public def save_ai_models_config(config, path=AI_MODELS_PATH) -> dict: """Persist AI model definitions, including local plaintext API keys.""" normalized = _normalize_ai_models_config(config) 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_ai_models_config(path=AI_MODELS_PATH) -> dict: """Load AI model definitions, writing defaults first if missing.""" if not os.path.exists(path): return save_ai_models_config(default_ai_models_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"AI 模型清单不是有效 JSON: {path}") from exc return _normalize_ai_models_config(loaded) def list_ai_models(category=None, path=AI_MODELS_PATH, reveal_api_key=False): """Return AI models, optionally filtered by category.""" if category is not None and category not in CATEGORIES: raise ConfigError("category 必须是 text 或 image") models = load_ai_models_config(path)["models"] filtered = [ copy.deepcopy(model) for model in models if category is None or model["category"] == category ] if reveal_api_key: return filtered return [_public_model(model) for model in filtered] def add_ai_model(model, path=AI_MODELS_PATH) -> None: config = load_ai_models_config(path) normalized = _normalize_ai_model(model) if any(item["name"] == normalized["name"] for item in config["models"]): raise ConfigError(f"AI 模型 name 已存在: {normalized['name']}") config["models"].append(normalized) save_ai_models_config(config, path=path) def update_ai_model(model_name, path=AI_MODELS_PATH, **fields) -> None: if not fields: return config = load_ai_models_config(path) index = _model_index(config["models"], model_name) updated = copy.deepcopy(config["models"][index]) updated.update(fields) normalized = _normalize_ai_model(updated) if normalized["name"] != model_name and any( model["name"] == normalized["name"] for model in config["models"] ): raise ConfigError(f"AI 模型 name 已存在: {normalized['name']}") config["models"][index] = normalized save_ai_models_config(config, path=path) def delete_ai_model(name, path=AI_MODELS_PATH) -> None: config = load_ai_models_config(path) index = _model_index(config["models"], name) remaining = config["models"][:index] + config["models"][index + 1 :] _assert_required_categories(remaining) save_ai_models_config({"models": remaining}, path=path) def get_model(name, path=AI_MODELS_PATH) -> dict: """Return the model definition including api_key. Callers must not log it.""" models = load_ai_models_config(path)["models"] return copy.deepcopy(models[_model_index(models, name)]) def _test_request_payload(model): if model["api_type"] == "images_edits": payload = {"model": model["model"], "prompt": "ping"} payload.update(model.get("extra_body", {})) return payload payload = { "model": model["model"], "messages": [{"role": "user", "content": "ping"}], } payload.update(model.get("extra_body", {})) return payload def test_ai_model(name, path=AI_MODELS_PATH) -> dict: """Send a minimal request to the configured model endpoint.""" model = None try: model = get_model(name, path=path) missing = [ field for field in ("url", "model", "api_key") if not str(model.get(field, "")).strip() ] if missing: return {"ok": False, "error": "模型缺少字段: " + ", ".join(missing)} data = json.dumps(_test_request_payload(model), ensure_ascii=False).encode( "utf-8" ) request = urllib.request.Request( model["url"], data=data, headers={ "Authorization": "Bearer " + model["api_key"], "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen( request, timeout=int(model["connect_timeout_seconds"]) ) as response: response.read(1024) return {"ok": 200 <= response.status < 300, "status": response.status} except urllib.error.HTTPError as exc: return {"ok": False, "status": exc.code, "error": f"HTTP {exc.code}"} except urllib.error.URLError as exc: secret_values = [model.get("api_key")] if model else [] return {"ok": False, "error": redact_secrets(str(exc.reason), secret_values)} except Exception as exc: secret_values = [model.get("api_key")] if model else [] return {"ok": False, "error": redact_secrets(str(exc), secret_values)}