Aaron LiangandGitHub
4b030214d9
Rewrite the module-level descriptions for all 28 Python files and remove the obsolete integration note from grok_register_ttk.py. Verified on Python 3.9 and Python 3.12.
238 lines
8.3 KiB
Python
238 lines
8.3 KiB
Python
"""负责应用配置的默认值、加载保存、规范化和运行前校验。"""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import urllib.parse
|
|
|
|
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
|
|
|
|
DEFAULT_CONFIG = {
|
|
"duckmail_api_key": "",
|
|
"cloudflare_api_base": "",
|
|
"cloudflare_api_key": "",
|
|
"cloudflare_auth_mode": "none",
|
|
"cloudflare_path_domains": "/api/domains",
|
|
"cloudflare_path_accounts": "/api/new_address",
|
|
"cloudflare_path_token": "/api/token",
|
|
"cloudflare_path_messages": "/api/mails",
|
|
"cloudmail_api_base": "",
|
|
"cloudmail_public_token": "",
|
|
"cloudmail_domains": "",
|
|
"cloudmail_path_messages": "/api/public/emailList",
|
|
"proxy": "",
|
|
"enable_nsfw": True,
|
|
"register_count": 1,
|
|
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36",
|
|
"grok2api_auto_add_local": False,
|
|
"grok2api_local_token_file": "",
|
|
"grok2api_pool_name": "ssoBasic",
|
|
"grok2api_auto_add_remote": False,
|
|
"grok2api_remote_base": "",
|
|
"grok2api_remote_app_key": "",
|
|
"api_reverse_tools": "",
|
|
"cpa_export_enabled": True,
|
|
"cpa_auth_dir": "./cpa_auths",
|
|
"cpa_copy_to_hotload": False,
|
|
"cpa_hotload_dir": "",
|
|
"cpa_base_url": "https://cli-chat-proxy.grok.com/v1",
|
|
"cpa_proxy": "",
|
|
"cpa_headless": False,
|
|
"cpa_force_standalone": True,
|
|
"cpa_mint_timeout_sec": 300,
|
|
"cpa_mint_cookie_inject": True,
|
|
"cpa_oidc_request_timeout_sec": 15,
|
|
"cpa_oidc_poll_timeout_sec": 15,
|
|
"grok2api_allow_legacy_full_save": False,
|
|
"email_provider": "duckmail",
|
|
"yyds_api_key": "",
|
|
"yyds_jwt": "",
|
|
"defaultDomains": "",
|
|
}
|
|
|
|
|
|
config = DEFAULT_CONFIG.copy()
|
|
|
|
class ConfigError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _require_bool(cfg, key):
|
|
value = cfg.get(key)
|
|
if type(value) is not bool:
|
|
raise ConfigError(f"配置项 {key} 必须是布尔值 true/false")
|
|
return value
|
|
|
|
|
|
def _require_int(cfg, key, minimum, maximum):
|
|
value = cfg.get(key)
|
|
if type(value) is not int:
|
|
raise ConfigError(f"配置项 {key} 必须是整数")
|
|
if not minimum <= value <= maximum:
|
|
raise ConfigError(f"配置项 {key} 必须在 {minimum} 到 {maximum} 之间")
|
|
return value
|
|
|
|
|
|
def _require_string(cfg, key, path=False):
|
|
value = cfg.get(key)
|
|
if not isinstance(value, str):
|
|
raise ConfigError(f"配置项 {key} 必须是字符串")
|
|
value = value.strip() if key not in ("user_agent",) else value
|
|
if "\x00" in value:
|
|
raise ConfigError(f"配置项 {key} 包含非法空字符")
|
|
if path and value:
|
|
os.path.expanduser(value)
|
|
return value
|
|
|
|
|
|
def validate_config_structure(raw):
|
|
if not isinstance(raw, dict):
|
|
raise ConfigError("config root must be a JSON object")
|
|
cfg = {**DEFAULT_CONFIG, **raw}
|
|
bool_keys = (
|
|
"enable_nsfw", "grok2api_auto_add_local", "grok2api_auto_add_remote",
|
|
"grok2api_allow_legacy_full_save", "cpa_export_enabled",
|
|
"cpa_copy_to_hotload", "cpa_headless", "cpa_force_standalone",
|
|
"cpa_mint_cookie_inject",
|
|
)
|
|
for key in bool_keys:
|
|
cfg[key] = _require_bool(cfg, key)
|
|
cfg["register_count"] = _require_int(cfg, "register_count", 1, 2500)
|
|
cfg["cpa_mint_timeout_sec"] = _require_int(cfg, "cpa_mint_timeout_sec", 30, 1800)
|
|
cfg["cpa_oidc_request_timeout_sec"] = _require_int(cfg, "cpa_oidc_request_timeout_sec", 3, 120)
|
|
cfg["cpa_oidc_poll_timeout_sec"] = _require_int(cfg, "cpa_oidc_poll_timeout_sec", 3, 120)
|
|
string_keys = tuple(key for key, value in DEFAULT_CONFIG.items() if isinstance(value, str))
|
|
path_keys = {"grok2api_local_token_file", "api_reverse_tools", "cpa_auth_dir", "cpa_hotload_dir"}
|
|
for key in string_keys:
|
|
cfg[key] = _require_string(cfg, key, path=key in path_keys)
|
|
enums = {
|
|
"email_provider": {"duckmail", "yyds", "cloudflare", "cloudmail"},
|
|
"cloudflare_auth_mode": {"query-key", "bearer", "x-api-key", "x-admin-auth", "none"},
|
|
"grok2api_pool_name": {"ssoBasic", "ssoSuper"},
|
|
}
|
|
for key, allowed in enums.items():
|
|
value = cfg.get(key, DEFAULT_CONFIG.get(key, ""))
|
|
if value not in allowed:
|
|
raise ConfigError(f"配置项 {key} 的值无效: {value!r}; 允许值: {sorted(allowed)}")
|
|
cfg[key] = value
|
|
|
|
api_path_keys = {
|
|
"cloudflare_path_domains", "cloudflare_path_accounts",
|
|
"cloudflare_path_token", "cloudflare_path_messages",
|
|
"cloudmail_path_messages",
|
|
}
|
|
for key in api_path_keys:
|
|
value = cfg[key]
|
|
if value and not value.startswith("/"):
|
|
value = "/" + value
|
|
cfg[key] = value
|
|
|
|
url_keys = {
|
|
"cloudflare_api_base", "cloudmail_api_base",
|
|
"grok2api_remote_base", "cpa_base_url",
|
|
}
|
|
for key in url_keys:
|
|
value = cfg[key]
|
|
if not value:
|
|
continue
|
|
parsed = urllib.parse.urlsplit(value)
|
|
if parsed.scheme not in ("http", "https") or not parsed.netloc:
|
|
raise ConfigError(f"配置项 {key} 必须是有效的 http/https URL")
|
|
|
|
for key in path_keys:
|
|
value = cfg[key]
|
|
if value.startswith("~"):
|
|
cfg[key] = os.path.expanduser(value)
|
|
return cfg
|
|
|
|
|
|
def validate_run_requirements(cfg):
|
|
cfg = validate_config_structure(cfg)
|
|
provider = cfg["email_provider"]
|
|
if provider == "cloudflare" and not cfg["cloudflare_api_base"]:
|
|
raise ConfigError("Cloudflare 模式需要配置 cloudflare_api_base")
|
|
if provider == "cloudmail":
|
|
missing = [
|
|
key for key in ("cloudmail_api_base", "cloudmail_public_token", "cloudmail_domains")
|
|
if not cfg[key]
|
|
]
|
|
if missing:
|
|
raise ConfigError("Cloud Mail 模式缺少必需配置: " + ", ".join(missing))
|
|
if provider == "yyds" and not (cfg["yyds_api_key"] or cfg["yyds_jwt"]):
|
|
raise ConfigError("YYDS 模式需要至少配置 yyds_api_key 或 yyds_jwt")
|
|
if cfg["grok2api_auto_add_remote"]:
|
|
missing = [
|
|
key for key in ("grok2api_remote_base", "grok2api_remote_app_key")
|
|
if not cfg[key]
|
|
]
|
|
if missing:
|
|
raise ConfigError("远端 token 入池缺少必需配置: " + ", ".join(missing))
|
|
if cfg["cpa_export_enabled"] and cfg["cpa_copy_to_hotload"] and not cfg["cpa_hotload_dir"]:
|
|
raise ConfigError("启用 CPA 热加载复制时必须配置 cpa_hotload_dir")
|
|
return cfg
|
|
|
|
|
|
def validate_config(raw):
|
|
"""Backward-compatible full validation used before a run or save."""
|
|
return validate_run_requirements(raw)
|
|
|
|
|
|
|
|
def _replace_config(value):
|
|
config.clear()
|
|
config.update(value)
|
|
return config
|
|
|
|
|
|
def load_config():
|
|
if os.path.exists(CONFIG_FILE):
|
|
try:
|
|
with open(CONFIG_FILE, "r", encoding="utf-8") as handle:
|
|
loaded = json.load(handle)
|
|
return _replace_config(validate_config_structure(loaded))
|
|
except ConfigError:
|
|
raise
|
|
except Exception as exc:
|
|
raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc
|
|
return _replace_config(validate_config_structure(DEFAULT_CONFIG.copy()))
|
|
|
|
|
|
def save_config():
|
|
normalized = validate_config_structure(config)
|
|
_replace_config(normalized)
|
|
config_dir = os.path.dirname(os.path.abspath(CONFIG_FILE))
|
|
os.makedirs(config_dir, exist_ok=True)
|
|
fd = None
|
|
temp_path = None
|
|
try:
|
|
fd, temp_path = tempfile.mkstemp(prefix=".config-", suffix=".json.tmp", dir=config_dir)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
fd = None
|
|
json.dump(config, handle, indent=4, ensure_ascii=False)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
try:
|
|
os.chmod(temp_path, 0o600)
|
|
except Exception:
|
|
pass
|
|
os.replace(temp_path, CONFIG_FILE)
|
|
temp_path = None
|
|
try:
|
|
os.chmod(CONFIG_FILE, 0o600)
|
|
except Exception:
|
|
pass
|
|
except Exception as exc:
|
|
raise ConfigError(f"保存配置失败: {exc}") from exc
|
|
finally:
|
|
if fd is not None:
|
|
try:
|
|
os.close(fd)
|
|
except Exception:
|
|
pass
|
|
if temp_path and os.path.exists(temp_path):
|
|
try:
|
|
os.unlink(temp_path)
|
|
except Exception:
|
|
pass
|
|
return config
|