1035 lines
34 KiB
Python
1035 lines
34 KiB
Python
"""Application-level configuration for cmshopee.
|
|
|
|
This module owns `data/config.json` plus `data/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
|
|
`data/config/ai_models.json`.
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
DATA_DIR_NAME = "data"
|
|
|
|
|
|
def app_base_dir() -> str:
|
|
"""Return the application directory used as the parent of local data."""
|
|
|
|
if getattr(sys, "frozen", False):
|
|
return os.path.dirname(os.path.abspath(sys.executable))
|
|
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
def default_data_dir(base_dir=None) -> str:
|
|
"""Return the default local user data directory."""
|
|
|
|
return os.path.abspath(os.path.join(base_dir or app_base_dir(), DATA_DIR_NAME))
|
|
|
|
|
|
CONFIG_PATH = os.path.join(default_data_dir(), "config.json")
|
|
AI_MODELS_PATH = os.path.join(default_data_dir(), "config", "ai_models.json")
|
|
CMHUB_CONFIG_PATH = os.path.join(default_data_dir(), "config", "cmhub.json")
|
|
CATEGORIES = {"text", "image"}
|
|
API_TYPES = {"chat", "images_edits", "auto"}
|
|
AI_BACKENDS = {"direct", "cmhub"}
|
|
AI_GENERATE_MODES = {"title", "cover", "title_cover"}
|
|
SHOPEE_UPDATE_MODES = {"title", "cover", "title_cover"}
|
|
AI_CONCURRENCY_MIN = 1
|
|
AI_CONCURRENCY_MAX = 5
|
|
AI_RETRY_MIN = 0
|
|
AI_RETRY_MAX = 10
|
|
SHOPEE_PARALLEL_ACCOUNTS_MIN = 1
|
|
SHOPEE_PARALLEL_ACCOUNTS_MAX = 5
|
|
CMHUB_CONNECT_TIMEOUT_DEFAULT = 66
|
|
CMHUB_CONNECT_TIMEOUT_OLD_DEFAULT = 10
|
|
RUNTIME_CONFIG_KEYS = {
|
|
"config_path",
|
|
"ai_models_path",
|
|
"cmhub_config_path",
|
|
"data_dir",
|
|
}
|
|
LEGACY_USER_DATA_PATHS = (
|
|
"config.json",
|
|
"config",
|
|
"cmshopee.db",
|
|
"cmshopee.db-wal",
|
|
"cmshopee.db-shm",
|
|
"db.sqlite",
|
|
"chrome_user_data_dir",
|
|
"images",
|
|
"logs",
|
|
"prompts",
|
|
"title_prompt.txt",
|
|
)
|
|
|
|
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",
|
|
"generate_cover": False,
|
|
"generate_mode": "",
|
|
"backend": "cmhub",
|
|
"cmhub": {
|
|
"base_url": "",
|
|
"title_alias": "",
|
|
"image_alias": "",
|
|
"connect_timeout": CMHUB_CONNECT_TIMEOUT_DEFAULT,
|
|
"use_system_proxy": False,
|
|
"download_with_curl": "auto",
|
|
"check_balance_before_batch": False,
|
|
},
|
|
"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",
|
|
"update_mode": "",
|
|
"max_items_per_run": 1,
|
|
"dry_run": False,
|
|
"max_parallel_accounts": 1,
|
|
},
|
|
}
|
|
|
|
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": {},
|
|
},
|
|
]
|
|
}
|
|
|
|
|
|
DEFAULT_CMHUB_CONFIG = {
|
|
"api_key": "",
|
|
}
|
|
SECRET_FIELD_NAMES = {"api_key", "apikey", "key", "token", "password"}
|
|
|
|
|
|
class ConfigError(RuntimeError):
|
|
"""Raised when app configuration is missing or malformed."""
|
|
|
|
|
|
class DataMigrationConflictError(ConfigError):
|
|
"""Raised when old-layout data cannot be moved into data/ safely."""
|
|
|
|
|
|
class DataDirectoryWriteError(ConfigError):
|
|
"""Raised when the local data directory is not writable."""
|
|
|
|
|
|
def _data_dir_for_config_path(path) -> str:
|
|
config_path = os.path.abspath(path or CONFIG_PATH)
|
|
if config_path == os.path.abspath(CONFIG_PATH):
|
|
return default_data_dir()
|
|
return os.path.dirname(config_path)
|
|
|
|
|
|
def data_dir(config=None) -> str:
|
|
"""Return the absolute local user data directory."""
|
|
|
|
if isinstance(config, dict):
|
|
if config.get("data_dir"):
|
|
return os.path.abspath(str(config["data_dir"]))
|
|
if config.get("config_path"):
|
|
return _data_dir_for_config_path(config["config_path"])
|
|
return default_data_dir()
|
|
|
|
|
|
def _strip_runtime_config_keys(config):
|
|
if not isinstance(config, dict):
|
|
return config
|
|
return {
|
|
key: copy.deepcopy(value)
|
|
for key, value in config.items()
|
|
if key not in RUNTIME_CONFIG_KEYS
|
|
}
|
|
|
|
|
|
def _runtime_paths(config_path, data_dir_path=None):
|
|
root = os.path.abspath(data_dir_path or _data_dir_for_config_path(config_path))
|
|
return {
|
|
"config_path": os.path.abspath(config_path or os.path.join(root, "config.json")),
|
|
"data_dir": root,
|
|
"ai_models_path": os.path.join(root, "config", "ai_models.json"),
|
|
"cmhub_config_path": os.path.join(root, "config", "cmhub.json"),
|
|
}
|
|
|
|
|
|
def _with_runtime_paths(config, path):
|
|
result = copy.deepcopy(config)
|
|
result.update(_runtime_paths(path))
|
|
return result
|
|
|
|
|
|
def _data_parent_dir(config=None) -> str:
|
|
return os.path.dirname(data_dir(config))
|
|
|
|
|
|
def resolve_data_path(path, config=None) -> str:
|
|
"""Resolve a user-data path under data_dir unless it is already absolute."""
|
|
|
|
text = str(path or "").strip()
|
|
if not text:
|
|
return ""
|
|
if os.path.isabs(text):
|
|
return os.path.abspath(text)
|
|
normalized = os.path.normpath(text)
|
|
first_part = normalized.split(os.sep, 1)[0]
|
|
if first_part == DATA_DIR_NAME:
|
|
return os.path.abspath(os.path.join(_data_parent_dir(config), normalized))
|
|
return os.path.abspath(os.path.join(data_dir(config), normalized))
|
|
|
|
|
|
def data_path(*parts, config=None) -> str:
|
|
return os.path.abspath(os.path.join(data_dir(config), *[str(part) for part in parts]))
|
|
|
|
|
|
def ai_models_config_path(config=None) -> str:
|
|
if isinstance(config, dict) and config.get("ai_models_path"):
|
|
return os.path.abspath(str(config["ai_models_path"]))
|
|
return data_path("config", "ai_models.json", config=config)
|
|
|
|
|
|
def cmhub_config_file_path(config=None) -> str:
|
|
if isinstance(config, dict) and config.get("cmhub_config_path"):
|
|
return os.path.abspath(str(config["cmhub_config_path"]))
|
|
return data_path("config", "cmhub.json", config=config)
|
|
|
|
|
|
def title_prompt_path(config=None) -> str:
|
|
return data_path("title_prompt.txt", config=config)
|
|
|
|
|
|
def cover_prompts_dir(config=None) -> str:
|
|
return data_path("prompts", "cover", config=config)
|
|
|
|
|
|
def diagnostic_log_dir(config=None) -> str:
|
|
return data_path("logs", config=config)
|
|
|
|
|
|
def _is_within(path, parent) -> bool:
|
|
try:
|
|
return os.path.commonpath([os.path.abspath(path), os.path.abspath(parent)]) == os.path.abspath(parent)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def migrate_legacy_user_data(base_dir=None, data_dir_path=None):
|
|
"""Move old exe-top-level local data into data_dir without overwriting."""
|
|
|
|
base = os.path.abspath(base_dir or app_base_dir())
|
|
target_root = os.path.abspath(data_dir_path or default_data_dir(base))
|
|
if base == target_root:
|
|
return []
|
|
os.makedirs(target_root, exist_ok=True)
|
|
|
|
moves = []
|
|
seen_sources = set()
|
|
for relative in LEGACY_USER_DATA_PATHS:
|
|
source = os.path.abspath(os.path.join(base, relative))
|
|
if source in seen_sources or not os.path.exists(source):
|
|
continue
|
|
seen_sources.add(source)
|
|
if _is_within(source, target_root):
|
|
continue
|
|
if not _is_within(source, base):
|
|
raise ConfigError(f"旧数据路径不在程序目录内: {source}")
|
|
target = os.path.abspath(os.path.join(target_root, relative))
|
|
if not _is_within(target, target_root):
|
|
raise ConfigError(f"迁移目标路径不在 data 目录内: {target}")
|
|
moves.append((relative, source, target))
|
|
|
|
conflicts = [relative for relative, _source, target in moves if os.path.exists(target)]
|
|
if conflicts:
|
|
raise DataMigrationConflictError(
|
|
"检测到旧布局数据和 data 目录内数据同时存在，无法自动迁移。"
|
|
"请先手动合并或备份后再启动。冲突项: " + "、".join(conflicts)
|
|
)
|
|
|
|
moved = []
|
|
for relative, source, target in moves:
|
|
os.makedirs(os.path.dirname(target), exist_ok=True)
|
|
shutil.move(source, target)
|
|
moved.append(relative)
|
|
return moved
|
|
|
|
|
|
def ensure_writable_data_dir(path=None) -> str:
|
|
root = os.path.abspath(path or default_data_dir())
|
|
try:
|
|
os.makedirs(root, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
prefix=".cmshopee_write_test_",
|
|
dir=root,
|
|
delete=False,
|
|
) as fh:
|
|
marker = fh.name
|
|
fh.write("ok")
|
|
os.remove(marker)
|
|
except Exception as exc:
|
|
raise DataDirectoryWriteError(
|
|
f"数据目录不可写：{root}。请把程序放到可写目录，勿放 Program Files。"
|
|
) from exc
|
|
return root
|
|
|
|
|
|
def prepare_data_dir(base_dir=None, data_dir_path=None, migrate=True) -> str:
|
|
root = os.path.abspath(data_dir_path or default_data_dir(base_dir))
|
|
if migrate:
|
|
migrate_legacy_user_data(base_dir=base_dir, data_dir_path=root)
|
|
return ensure_writable_data_dir(root)
|
|
|
|
|
|
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 _normalize_config_values(config, migrate_old_cmhub_connect_timeout=False):
|
|
if not isinstance(config, dict):
|
|
return config
|
|
ai = config.get("ai")
|
|
if isinstance(ai, dict):
|
|
ai["generate_mode"] = normalize_generate_mode(
|
|
ai.get("generate_mode"),
|
|
generate_cover=ai.get("generate_cover", False),
|
|
)
|
|
ai["generate_cover"] = generate_mode_includes_cover(ai["generate_mode"])
|
|
ai["title_concurrency"] = _clamp_int(
|
|
ai.get("title_concurrency"),
|
|
AI_CONCURRENCY_MIN,
|
|
AI_CONCURRENCY_MAX,
|
|
DEFAULT_CONFIG["ai"]["title_concurrency"],
|
|
)
|
|
ai["image_concurrency"] = _clamp_int(
|
|
ai.get("image_concurrency"),
|
|
AI_CONCURRENCY_MIN,
|
|
AI_CONCURRENCY_MAX,
|
|
DEFAULT_CONFIG["ai"]["image_concurrency"],
|
|
)
|
|
ai["retry"] = _clamp_int(
|
|
ai.get("retry"),
|
|
AI_RETRY_MIN,
|
|
AI_RETRY_MAX,
|
|
DEFAULT_CONFIG["ai"]["retry"],
|
|
)
|
|
cmhub = ai.get("cmhub")
|
|
if isinstance(cmhub, dict):
|
|
cmhub["base_url"] = normalize_cmhub_base_url(cmhub.get("base_url", ""))
|
|
cmhub["connect_timeout"] = _normalize_cmhub_connect_timeout(
|
|
cmhub.get("connect_timeout"),
|
|
migrate_old_default=migrate_old_cmhub_connect_timeout,
|
|
)
|
|
cmhub["download_with_curl"] = _normalize_cmhub_download_with_curl(
|
|
cmhub.get("download_with_curl", "auto")
|
|
)
|
|
update = config.get("shopee_update")
|
|
if isinstance(update, dict):
|
|
old_parallel_accounts = update.pop("parallel_accounts", None)
|
|
old_allow_cover_update = update.get("allow_cover_update", False)
|
|
update.pop("allow_real_submit", None)
|
|
update.pop("allow_cover_update", None)
|
|
update.pop("close_success_tab", None)
|
|
update["update_mode"] = normalize_update_mode(
|
|
update.get("update_mode"),
|
|
allow_cover_update=old_allow_cover_update,
|
|
)
|
|
if old_parallel_accounts is False:
|
|
update["max_parallel_accounts"] = SHOPEE_PARALLEL_ACCOUNTS_MIN
|
|
else:
|
|
update["max_parallel_accounts"] = _clamp_int(
|
|
update.get("max_parallel_accounts"),
|
|
SHOPEE_PARALLEL_ACCOUNTS_MIN,
|
|
SHOPEE_PARALLEL_ACCOUNTS_MAX,
|
|
DEFAULT_CONFIG["shopee_update"]["max_parallel_accounts"],
|
|
)
|
|
update["max_items_per_run"] = _clamp_int(
|
|
update.get("max_items_per_run"),
|
|
1,
|
|
9999,
|
|
DEFAULT_CONFIG["shopee_update"]["max_items_per_run"],
|
|
)
|
|
return config
|
|
|
|
|
|
def _clamp_int(value, minimum, maximum, default):
|
|
try:
|
|
number = int(value)
|
|
except (TypeError, ValueError):
|
|
number = int(default)
|
|
return min(int(maximum), max(int(minimum), number))
|
|
|
|
|
|
def _normalize_cmhub_connect_timeout(value, migrate_old_default=False):
|
|
try:
|
|
number = int(value)
|
|
except (TypeError, ValueError):
|
|
number = CMHUB_CONNECT_TIMEOUT_DEFAULT
|
|
if number <= 0:
|
|
number = CMHUB_CONNECT_TIMEOUT_DEFAULT
|
|
if migrate_old_default and number == CMHUB_CONNECT_TIMEOUT_OLD_DEFAULT:
|
|
return CMHUB_CONNECT_TIMEOUT_DEFAULT
|
|
return number
|
|
|
|
|
|
def _normalize_cmhub_download_with_curl(value):
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
text = str(value or "auto").strip().lower()
|
|
if text in {"auto", "true", "false"}:
|
|
return text
|
|
return "auto"
|
|
|
|
|
|
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 mask_email(email) -> str:
|
|
"""Return a display-safe email string with the local part masked."""
|
|
|
|
text = str(email or "").strip()
|
|
if not text or "@" not in text:
|
|
return text
|
|
local, domain = text.split("@", 1)
|
|
if not local or not domain:
|
|
return mask_secret(text)
|
|
if len(local) == 1:
|
|
masked_local = "***"
|
|
elif len(local) == 2:
|
|
masked_local = f"{local[0]}***"
|
|
else:
|
|
masked_local = f"{local[0]}***{local[-1]}"
|
|
return f"{masked_local}@{domain}"
|
|
|
|
|
|
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():
|
|
lowered = str(key).lower()
|
|
if _is_secret_field_name(key):
|
|
sanitized[key] = mask_secret(child)
|
|
elif lowered == "email" or lowered.endswith("_email"):
|
|
sanitized[key] = mask_email(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 default_cmhub_config() -> dict:
|
|
"""Return a new copy of the default cmhub key config."""
|
|
|
|
return copy.deepcopy(DEFAULT_CMHUB_CONFIG)
|
|
|
|
|
|
def _normalize_cmhub_config(config):
|
|
if config is None:
|
|
config = {}
|
|
if not isinstance(config, dict):
|
|
raise ConfigError("cmhub 配置必须是对象")
|
|
return {"api_key": str(config.get("api_key", "") or "")}
|
|
|
|
|
|
def load_cmhub_config(path=CMHUB_CONFIG_PATH) -> dict:
|
|
"""Load cmhub API key config. Missing file means key is not configured."""
|
|
|
|
if not os.path.exists(path):
|
|
return default_cmhub_config()
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
try:
|
|
loaded = json.load(fh)
|
|
except json.JSONDecodeError as exc:
|
|
raise ConfigError(f"cmhub 配置不是有效 JSON: {path}") from exc
|
|
return _normalize_cmhub_config(loaded)
|
|
|
|
|
|
def save_cmhub_config(config, path=CMHUB_CONFIG_PATH) -> dict:
|
|
"""Persist cmhub API key config, including the local plaintext key."""
|
|
|
|
normalized = _normalize_cmhub_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 get_cmhub_api_key(path=CMHUB_CONFIG_PATH, masked=False) -> str:
|
|
key = load_cmhub_config(path).get("api_key", "")
|
|
return mask_secret(key) if masked else key
|
|
|
|
|
|
def save_config(config, path=CONFIG_PATH) -> dict:
|
|
"""Persist config to JSON and return the normalized config."""
|
|
|
|
normalized = _deep_merge(DEFAULT_CONFIG, _strip_runtime_config_keys(config))
|
|
_normalize_config_values(normalized)
|
|
_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 _with_runtime_paths(normalized, path)
|
|
|
|
|
|
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)
|
|
_normalize_config_values(normalized, migrate_old_cmhub_connect_timeout=True)
|
|
_assert_no_secrets(normalized)
|
|
return _with_runtime_paths(normalized, path)
|
|
|
|
|
|
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:
|
|
cfg = _config_or_load(config)
|
|
return resolve_data_path(cfg.get("user_data_root", "chrome_user_data_dir"), cfg)
|
|
|
|
|
|
def image_dir(config=None) -> str:
|
|
cfg = _config_or_load(config)
|
|
return resolve_data_path(cfg.get("image_dir", "images"), cfg)
|
|
|
|
|
|
def db_path(config=None) -> str:
|
|
cfg = _config_or_load(config)
|
|
return resolve_data_path(cfg.get("db_path", "cmshopee.db"), cfg)
|
|
|
|
|
|
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 normalize_generate_mode(value=None, generate_cover=None) -> str:
|
|
text = str(value or "").strip().lower()
|
|
if text in AI_GENERATE_MODES:
|
|
return text
|
|
return "title_cover" if bool(generate_cover) else "title"
|
|
|
|
|
|
def generate_mode_includes_title(mode) -> bool:
|
|
return normalize_generate_mode(mode) in {"title", "title_cover"}
|
|
|
|
|
|
def generate_mode_includes_cover(mode) -> bool:
|
|
return normalize_generate_mode(mode) in {"cover", "title_cover"}
|
|
|
|
|
|
def ai_generate_mode(config=None) -> str:
|
|
ai = ai_config(config)
|
|
return normalize_generate_mode(ai.get("generate_mode"), generate_cover=ai.get("generate_cover", False))
|
|
|
|
|
|
def normalize_update_mode(value=None, allow_cover_update=None) -> str:
|
|
text = str(value or "").strip().lower()
|
|
if text in SHOPEE_UPDATE_MODES:
|
|
return text
|
|
return "title_cover" if bool(allow_cover_update) else "title"
|
|
|
|
|
|
def update_mode_includes_title(mode) -> bool:
|
|
return normalize_update_mode(mode) in {"title", "title_cover"}
|
|
|
|
|
|
def update_mode_includes_cover(mode) -> bool:
|
|
return normalize_update_mode(mode) in {"cover", "title_cover"}
|
|
|
|
|
|
def shopee_update_config(config=None) -> dict:
|
|
cfg = _config_or_load(config)
|
|
loaded = cfg.get("shopee_update", {})
|
|
if not isinstance(loaded, dict):
|
|
loaded = {}
|
|
merged = _deep_merge(DEFAULT_CONFIG["shopee_update"], loaded)
|
|
merged["update_mode"] = normalize_update_mode(
|
|
merged.get("update_mode"),
|
|
allow_cover_update=merged.get("allow_cover_update", False),
|
|
)
|
|
if merged.get("parallel_accounts") is False:
|
|
merged["max_parallel_accounts"] = SHOPEE_PARALLEL_ACCOUNTS_MIN
|
|
else:
|
|
merged["max_parallel_accounts"] = _clamp_int(
|
|
merged.get("max_parallel_accounts"),
|
|
SHOPEE_PARALLEL_ACCOUNTS_MIN,
|
|
SHOPEE_PARALLEL_ACCOUNTS_MAX,
|
|
DEFAULT_CONFIG["shopee_update"]["max_parallel_accounts"],
|
|
)
|
|
for removed_key in (
|
|
"allow_real_submit",
|
|
"allow_cover_update",
|
|
"close_success_tab",
|
|
"parallel_accounts",
|
|
):
|
|
merged.pop(removed_key, None)
|
|
return merged
|
|
|
|
|
|
|
|
def ai_backend(config=None) -> str:
|
|
ai = ai_config(config)
|
|
default_backend = DEFAULT_CONFIG["ai"]["backend"]
|
|
backend = str(ai.get("backend", default_backend) or default_backend).strip().lower()
|
|
if backend not in AI_BACKENDS:
|
|
raise ConfigError("AI backend 必须是 direct 或 cmhub")
|
|
return backend
|
|
|
|
|
|
def cmhub_config(config=None) -> dict:
|
|
ai = ai_config(config)
|
|
value = ai.get("cmhub", {})
|
|
if not isinstance(value, dict):
|
|
raise ConfigError("ai.cmhub 必须是对象")
|
|
merged = _deep_merge(DEFAULT_CONFIG["ai"]["cmhub"], value)
|
|
merged["base_url"] = normalize_cmhub_base_url(merged.get("base_url", ""))
|
|
merged["title_alias"] = str(merged.get("title_alias", "") or "").strip()
|
|
merged["image_alias"] = str(merged.get("image_alias", "") or "").strip()
|
|
merged["connect_timeout"] = _normalize_cmhub_connect_timeout(
|
|
merged.get("connect_timeout"),
|
|
migrate_old_default=False,
|
|
)
|
|
merged["download_with_curl"] = _normalize_cmhub_download_with_curl(
|
|
merged.get("download_with_curl", "auto")
|
|
)
|
|
merged["check_balance_before_batch"] = bool(merged.get("check_balance_before_batch", False))
|
|
if merged["connect_timeout"] <= 0:
|
|
raise ConfigError("ai.cmhub.connect_timeout 必须大于 0")
|
|
return merged
|
|
|
|
|
|
def normalize_cmhub_base_url(base_url) -> str:
|
|
"""Return the cmhub gateway root URL without path/query/fragment."""
|
|
|
|
text = str(base_url or "").strip()
|
|
if not text:
|
|
return ""
|
|
parts = urllib.parse.urlsplit(text)
|
|
if parts.scheme and parts.netloc:
|
|
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, "", "", ""))
|
|
if parts.netloc:
|
|
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, "", "", ""))
|
|
without_query = text.split("?", 1)[0].split("#", 1)[0].strip().strip("/")
|
|
if "://" not in without_query:
|
|
return without_query.split("/", 1)[0]
|
|
return text.rstrip("/")
|
|
|
|
|
|
def cmhub_request_url(base_url, endpoint) -> str:
|
|
base = normalize_cmhub_base_url(base_url)
|
|
path = "/" + str(endpoint or "").strip().lstrip("/")
|
|
if not base:
|
|
return path
|
|
return base + path
|
|
|
|
|
|
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("AI 模型清单必须包含 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)}
|