feat: store packaged user data under data directory

This commit is contained in:
chengma
2026-07-07 14:12:32 +08:00
parent caa04a9aa7
commit 148c4a73eb
33 changed files with 602 additions and 196 deletions
+231 -14
View File
@@ -1,24 +1,64 @@
"""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`.
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
CONFIG_PATH = "config.json"
AI_MODELS_PATH = os.path.join("config", "ai_models.json")
CMHUB_CONFIG_PATH = os.path.join("config", "cmhub.json")
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"}
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",
@@ -104,6 +144,180 @@ 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."""
@@ -241,7 +455,7 @@ def _normalize_cmhub_config(config):
if config is None:
config = {}
if not isinstance(config, dict):
raise ConfigError("config/cmhub.json 必须是对象")
raise ConfigError("cmhub 配置必须是对象")
return {"api_key": str(config.get("api_key", "") or "")}
@@ -279,7 +493,7 @@ def get_cmhub_api_key(path=CMHUB_CONFIG_PATH, masked=False) -> str:
def save_config(config, path=CONFIG_PATH) -> dict:
"""Persist config to JSON and return the normalized config."""
normalized = _deep_merge(DEFAULT_CONFIG, 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))
@@ -288,7 +502,7 @@ def save_config(config, path=CONFIG_PATH) -> dict:
with open(path, "w", encoding="utf-8") as fh:
json.dump(normalized, fh, ensure_ascii=False, indent=2)
fh.write("\n")
return normalized
return _with_runtime_paths(normalized, path)
def load_config(path=CONFIG_PATH) -> dict:
@@ -304,7 +518,7 @@ def load_config(path=CONFIG_PATH) -> dict:
normalized = _deep_merge(DEFAULT_CONFIG, loaded)
_normalize_config_values(normalized)
_assert_no_secrets(normalized)
return normalized
return _with_runtime_paths(normalized, path)
def update_config(updates, path=CONFIG_PATH) -> dict:
@@ -323,15 +537,18 @@ def chrome_path(config=None) -> str:
def user_data_root(config=None) -> str:
return _config_or_load(config).get("user_data_root", "chrome_user_data_dir")
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:
return _config_or_load(config).get("image_dir", "images")
cfg = _config_or_load(config)
return resolve_data_path(cfg.get("image_dir", "images"), cfg)
def db_path(config=None) -> str:
return _config_or_load(config).get("db_path", "cmshopee.db")
cfg = _config_or_load(config)
return resolve_data_path(cfg.get("db_path", "cmshopee.db"), cfg)
def default_debug_port(config=None) -> int:
@@ -446,7 +663,7 @@ def _normalize_ai_model(model):
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 列表")
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"])
+2 -2
View File
@@ -10,7 +10,7 @@ import traceback
from . import appconfig
DEFAULT_LOG_DIR = "logs"
DEFAULT_LOG_DIR = appconfig.diagnostic_log_dir()
DEFAULT_LOG_FILE = "cmshopee.log"
DEFAULT_MAX_BYTES = 2 * 1024 * 1024
DEFAULT_BACKUPS = 3
@@ -115,4 +115,4 @@ def _rotate_if_needed(path, max_bytes=DEFAULT_MAX_BYTES, backups=DEFAULT_BACKUPS
target = f"{path}.{index + 1}"
if os.path.exists(source):
os.replace(source, target)
os.replace(path, f"{path}.1")
os.replace(path, f"{path}.1")
+7 -2
View File
@@ -5,7 +5,7 @@ import os
import time
from urllib.parse import urlparse
from . import image_paths
from . import appconfig, image_paths
from .cdp import CDP, close_tab, create_tab, create_tab_info, find_product_tab, http_get
@@ -834,7 +834,12 @@ def collect(account, task, on_step=None) -> dict:
old_cover_src = read_cover_src(cdp)
out_path = _get(task, "old_cover_path")
if not out_path:
out_path = image_paths.task_image_path("images", task, account, "old")
out_path = image_paths.task_image_path(
appconfig.image_dir(),
task,
account,
"old",
)
_notify_collect_step(on_step, "download_cover")
old_cover_path = download_cover(old_cover_src, out_path)
return {
+12
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import os
import sys
from .. import appconfig
from . import widgets as _widgets
from .widgets import *
@@ -46,6 +47,17 @@ def main() -> int:
return 1
_ensure_offscreen_for_headless_tests()
app = QApplication.instance() or QApplication(sys.argv)
try:
appconfig.prepare_data_dir()
except appconfig.DataMigrationConflictError as exc:
QMessageBox.critical(None, "数据迁移冲突", str(exc))
return 1
except appconfig.DataDirectoryWriteError as exc:
QMessageBox.critical(None, "数据目录不可写", str(exc))
return 1
except appconfig.ConfigError as exc:
QMessageBox.critical(None, "启动配置错误", str(exc))
return 1
window = MainWindow()
window.show()
return app.exec()
+3 -1
View File
@@ -84,7 +84,7 @@ class MainWindow(QMainWindow):
self.ai_models_path = (
ai_models_path
or self.config.get("ai_models_path")
or appconfig.AI_MODELS_PATH
or appconfig.ai_models_config_path(self.config)
)
self.setWindowTitle("蝦皮圈優化助手")
_fit_and_center_window(self)
@@ -123,6 +123,8 @@ class MainWindow(QMainWindow):
config=self.config,
config_path=self.config_path,
status_callback=self.statusBar().showMessage,
title_prompt_path=appconfig.title_prompt_path(self.config),
cover_prompts_dir=appconfig.cover_prompts_dir(self.config),
open_accounts_callback=lambda: self.open_accounts_tab(),
)
if title == "③ 更新shopee":
+4 -4
View File
@@ -48,8 +48,8 @@ class GenerateTab(QWidget):
self.db_path = _database_path(db_path, self.config)
self.status_callback = status_callback
self.open_accounts_callback = open_accounts_callback
self.title_prompt_path = title_prompt_path or prompts.TITLE_PROMPT_PATH
self.cover_prompts_dir = cover_prompts_dir or prompts.COVER_PROMPTS_DIR
self.title_prompt_path = title_prompt_path or appconfig.title_prompt_path(self.config)
self.cover_prompts_dir = cover_prompts_dir or appconfig.cover_prompts_dir(self.config)
self.current_cover_template = None
self.generate_worker = None
self.generate_thread = None
@@ -288,7 +288,7 @@ class GenerateTab(QWidget):
payload = {
key: value
for key, value in self.config.items()
if key not in {"config_path", "ai_models_path", "cmhub_config_path"}
if key not in {"config_path", "ai_models_path", "cmhub_config_path", "data_dir"}
}
payload["ai"] = ai_settings
try:
@@ -299,7 +299,7 @@ class GenerateTab(QWidget):
internal = {
key: value
for key, value in self.config.items()
if key in {"config_path", "ai_models_path", "cmhub_config_path"}
if key in {"config_path", "ai_models_path", "cmhub_config_path", "data_dir"}
}
self.config.clear()
self.config.update(saved)
+12 -19
View File
@@ -10,7 +10,7 @@ from ..workers import CMHubSettingsWorker as _RealCMHubSettingsWorker
PLAINTEXT_CMHUB_API_KEY_WARNING = (
"cmhub API Key 会以本地明文保存到 config/cmhub.json,仅供本机调用 cmhub 网关使用。"
"cmhub API Key 会以本地明文保存到 data/config/cmhub.json,仅供本机调用 cmhub 网关使用。"
"该文件已 gitignore,UI 打码显示,日志/导出不记录明文。"
)
@@ -23,7 +23,7 @@ def CMHubSettingsWorker(*args, **kwargs):
return _call_package_attr("CMHubSettingsWorker", _RealCMHubSettingsWorker, *args, **kwargs)
class SettingsTab(QWidget):
"""Tab 5: AI model definitions stored in config/ai_models.json."""
"""Tab 5: AI model definitions stored in data/config/ai_models.json."""
BACKEND_ITEMS = [("直连模型", "direct"), ("cmhub 网关", "cmhub")]
CATEGORY_ITEMS = [("文本", "text"), ("图像", "image")]
@@ -48,11 +48,11 @@ class SettingsTab(QWidget):
self.ai_models_path = (
ai_models_path
or self.config.get("ai_models_path")
or appconfig.AI_MODELS_PATH
or appconfig.ai_models_config_path(self.config)
)
self.cmhub_config_path = (
self.config.get("cmhub_config_path")
or self._default_cmhub_config_path(self.config_path)
or appconfig.cmhub_config_file_path(self.config)
)
self.status_callback = status_callback
self.models = []
@@ -546,15 +546,6 @@ class SettingsTab(QWidget):
self._cmhub_auto_refresh_done = True
self.refresh_cmhub_models()
def _default_cmhub_config_path(self, config_path):
if config_path and config_path != appconfig.CONFIG_PATH:
return os.path.join(
os.path.dirname(os.path.abspath(config_path)),
"config",
"cmhub.json",
)
return appconfig.CMHUB_CONFIG_PATH
def _on_backend_changed(self, index=None):
self.backend_combo.setVisible(False)
self.model_picker_panel.setVisible(False)
@@ -752,7 +743,7 @@ class SettingsTab(QWidget):
settings = {
key: value
for key, value in self.config.items()
if key not in {"config_path", "ai_models_path", "cmhub_config_path"}
if key not in {"config_path", "ai_models_path", "cmhub_config_path", "data_dir"}
}
settings.update(
{
@@ -799,6 +790,8 @@ class SettingsTab(QWidget):
internal["ai_models_path"] = self.ai_models_path
if self.cmhub_config_path != appconfig.CMHUB_CONFIG_PATH:
internal["cmhub_config_path"] = self.cmhub_config_path
if self.config.get("data_dir"):
internal["data_dir"] = self.config.get("data_dir")
self.config.clear()
self.config.update(saved)
self.config.update(internal)
@@ -847,9 +840,11 @@ class SettingsTab(QWidget):
)
self.jpg_quality_spin.setValue(int(ai_cfg.get("jpg_quality", 90) or 90))
self.chrome_path_edit.setText(appconfig.chrome_path(self.config))
self.user_data_root_edit.setText(appconfig.user_data_root(self.config))
self.image_dir_edit.setText(appconfig.image_dir(self.config))
self.db_path_edit.setText(appconfig.db_path(self.config))
self.user_data_root_edit.setText(
str(self.config.get("user_data_root", "chrome_user_data_dir") or "")
)
self.image_dir_edit.setText(str(self.config.get("image_dir", "images") or ""))
self.db_path_edit.setText(str(self.config.get("db_path", "cmshopee.db") or ""))
self.default_debug_port_spin.setValue(
int(appconfig.default_debug_port(self.config))
)
@@ -1318,5 +1313,3 @@ class SettingsTab(QWidget):
def _category_label(self, category):
return {"text": "文本", "image": "图像"}.get(category, category)
+1 -1
View File
@@ -170,7 +170,7 @@ def run_worker(*args, **kwargs):
PLAINTEXT_SECRET_TITLE = "本地明文保存提示"
PLAINTEXT_API_KEY_WARNING = (
"API Key 会以本地明文保存到 config/ai_models.json,仅供本机调用 AI 使用。"
"API Key 会以本地明文保存到 data/config/ai_models.json,仅供本机调用 AI 使用。"
"该文件已 gitignore,UI 打码显示,日志/导出不记录明文。"
)
PLAINTEXT_PASSWORD_WARNING = (
+3 -2
View File
@@ -4,9 +4,10 @@ from __future__ import annotations
import os
from . import appconfig
TITLE_PROMPT_PATH = "title_prompt.txt"
COVER_PROMPTS_DIR = os.path.join("prompts", "cover")
TITLE_PROMPT_PATH = appconfig.title_prompt_path()
COVER_PROMPTS_DIR = appconfig.cover_prompts_dir()
TEMPLATE_EXT = ".txt"
INVALID_NAME_CHARS = set('\\/:*?"<>|')