config_util.py
6.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
"""
config.json 安全加载器。
公开接口:
- DEFAULT_CONFIG: 默认配置常量
- load_config_safe(path): 区分处理各类 IO / JSON 错误,返回 (config, error_message)
- get_config_dir(): 跨平台返回配置目录
- get_config_path(): 返回 config.json 绝对路径
"""
from __future__ import annotations
import json
import logging
import os
import shutil
import sys
import platform
from datetime import datetime
from pathlib import Path
from typing import Optional, Tuple
logger = logging.getLogger(__name__)
DEFAULT_CONFIG: dict = {
"api_key": "",
"saved_prompts": [],
"db_config": None,
"last_user": "",
"saved_password_hash": "",
"logging_config": {
"enabled": True,
"level": "INFO",
"log_to_console": True,
},
"history_config": {
"max_history_count": 100,
},
}
def get_config_dir() -> Path:
"""跨平台返回用户级配置目录(与 image_generator.py 中 get_config_dir 一致)。"""
if getattr(sys, "frozen", False):
system = platform.system()
if system == "Darwin":
d = Path.home() / "Library" / "Application Support" / "ZB100ImageGenerator"
elif system == "Windows":
d = Path(os.getenv("APPDATA", str(Path.home()))) / "ZB100ImageGenerator"
else:
d = Path.home() / ".config" / "zb100imagegenerator"
else:
d = Path(".").resolve()
d.mkdir(parents=True, exist_ok=True)
return d
def get_config_path() -> Path:
return get_config_dir() / "config.json"
def load_config_safe(config_path: Path) -> Tuple[dict, str]:
"""
安全加载 config.json。
返回 (config, error):
- 成功:(合并后的 config dict, "")
- 失败但可恢复:(DEFAULT_CONFIG 副本, 错误描述);preflight 会根据返回
的 error 和 config 内容共同决定是否拦截启动
- 不抛异常
行为:
- 文件不存在 → (DEFAULT_CONFIG 副本, "") # 让 preflight 判断是否允许
- 空文件 / JSON 损坏 → 备份为 .bak.<timestamp> + 返回默认值 + error
- PermissionError / OSError → 返回默认值 + error(不备份,读都读不到)
- 顶层不是 object → 返回默认值 + error
- 正常 → defaults.update(loaded) 合并后返回(保留未知字段以兼容)
"""
config_path = Path(config_path)
if not config_path.exists():
return dict(DEFAULT_CONFIG), ""
try:
content = config_path.read_text(encoding="utf-8")
except PermissionError as e:
return dict(DEFAULT_CONFIG), f"permission denied: {e}"
except OSError as e:
return dict(DEFAULT_CONFIG), f"IO error: {e}"
if not content.strip():
_backup(config_path, reason="empty")
return dict(DEFAULT_CONFIG), "config.json was empty, using defaults"
try:
loaded = json.loads(content)
except json.JSONDecodeError as e:
_backup(config_path, reason="parse-error")
return dict(DEFAULT_CONFIG), f"JSON parse error, backed up to .bak: {e}"
if not isinstance(loaded, dict):
return dict(DEFAULT_CONFIG), "config.json top-level is not an object"
merged = dict(DEFAULT_CONFIG)
merged.update(loaded)
return merged, ""
def find_bundled_config() -> Optional[Path]:
"""
定位打包进二进制的 config.json (只在 frozen 态有意义)。
查找顺序: macOS app bundle Resources/ → exe 同目录 → PyInstaller _MEIPASS/
都找不到返回 None。
"""
if not getattr(sys, "frozen", False):
return None
candidates = []
if platform.system() == "Darwin":
candidates.append(Path(sys.executable).parent.parent / "Resources" / "config.json")
else:
candidates.append(Path(sys.executable).parent / "config.json")
if hasattr(sys, "_MEIPASS"):
candidates.append(Path(sys._MEIPASS) / "config.json")
for p in candidates:
if p.exists():
return p
return None
def sync_bundled_api_key(user_config_path: Path) -> None:
"""
如果打包 config.json 里的 api_key 和用户目录的不一致,用打包的值覆盖那一个字段。
其他字段 (saved_prompts / last_user / saved_password_hash / db_config ...) 全部保留。
只在 frozen (打包态) 下生效 —— 开发态不干涉本地 config.json。
保护策略 (避免把用户 api_key 清零):
- 找不到打包 config → 不动
- 打包 config 解析失败 → 不动
- 打包 api_key 为空 → 不动
- 用户 config 不存在 → 不动 (外层的 "从 bundled 拷贝整个文件" 逻辑会处理)
- 用户 config 解析失败 → 不动 (load_config_safe 会走损坏分支备份)
写入失败不抛异常,记错误日志。
"""
if not getattr(sys, "frozen", False):
return
bundled = find_bundled_config()
if bundled is None:
return
try:
bundled_data = json.loads(bundled.read_text(encoding="utf-8"))
except Exception as e:
logger.warning(f"打包 config 解析失败, 跳过 api_key 同步: {e}")
return
bundled_key = str(bundled_data.get("api_key", "")).strip()
if not bundled_key:
logger.warning("打包 config 的 api_key 为空, 跳过同步 (避免清零用户 key)")
return
if not user_config_path.exists():
# 用户 config 还不存在 → 外层的整文件拷贝逻辑会处理
return
try:
user_data = json.loads(user_config_path.read_text(encoding="utf-8"))
except Exception as e:
logger.warning(f"用户 config 解析失败, 跳过 api_key 同步 (load_config_safe 会处理): {e}")
return
if not isinstance(user_data, dict):
return
if user_data.get("api_key") == bundled_key:
return # 一致,无需动作
# 只覆盖 api_key 一个字段, 其他原样保留
user_data["api_key"] = bundled_key
try:
user_config_path.write_text(
json.dumps(user_data, ensure_ascii=False, indent=2),
encoding="utf-8",
)
logger.info(f"[BOOT] api_key 已从 bundled 同步到用户 config: {user_config_path}")
except Exception as e:
logger.error(f"[BOOT] 写回用户 config 失败, api_key 未同步: {e}")
def _backup(src: Path, reason: str) -> None:
"""把损坏 / 空的 config 文件备份,不抛异常。"""
try:
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
dst = src.with_suffix(f".json.bak.{reason}.{ts}")
shutil.copy2(src, dst)
except Exception:
pass