506 lines
17 KiB
Python
506 lines
17 KiB
Python
"""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 model_request_url(model) -> str:
|
|
"""Return the HTTP endpoint used for a configured AI model."""
|
|
|
|
raw_url = str(model.get("url", "") or "").strip()
|
|
api_type = str(model.get("api_type", "auto") or "auto").strip()
|
|
if api_type == "images_edits":
|
|
return _append_default_endpoint(raw_url, "images/edits")
|
|
return _append_default_endpoint(raw_url, "chat/completions")
|
|
|
|
|
|
def _append_default_endpoint(raw_url, endpoint):
|
|
if not raw_url:
|
|
return raw_url
|
|
parts = urllib.parse.urlsplit(raw_url)
|
|
path = parts.path.rstrip("/")
|
|
lowered = path.lower()
|
|
endpoint_path = "/" + endpoint.strip("/")
|
|
if lowered.endswith(endpoint_path):
|
|
return raw_url
|
|
base_markers = ("", "/v1", "/v1beta", "/api/v1", "/api/v1beta")
|
|
if lowered in base_markers or lowered.endswith(base_markers[1:]):
|
|
path = path + endpoint_path
|
|
return urllib.parse.urlunsplit(
|
|
(parts.scheme, parts.netloc, path, parts.query, parts.fragment)
|
|
)
|
|
return raw_url
|
|
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_request_url(model),
|
|
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)}
|