mmc/standalone-openai-pool-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add guided standalone setup flow

Browse Source
This commit is contained in:
mmc
2026-03-19 11:12:26 +08:00
parent 9169ede86b
commit e312a62b08
4 changed files with 202 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
README.md
View File

@@ -44,6 +44,14 @@ openai-pool-standalone --help
python3 /root/standalone_cli/main.py config init
```
如果你希望像向导一样逐步填写配置，直接用：
```bash
python3 /root/standalone_cli/run.py config setup
```
`config init` 只是初始化配置文件；`config setup` 才是交互式配置引导。
初始化后请编辑：
- `/root/standalone_cli/data/sync_config.json`

155
main.py
View File

@@ -33,6 +33,7 @@ try:
print_status_block,
read_token_file,
save_runtime_proxy,
save_sync_config,
save_sub2api_credentials,
sub2api_actions_description,
sync_all_tokens_to_sub2api,
@@ -54,6 +55,7 @@ except ImportError:
print_status_block,
read_token_file,
save_runtime_proxy,
save_sync_config,
save_sub2api_credentials,
sub2api_actions_description,
sync_all_tokens_to_sub2api,
@@ -81,6 +83,9 @@ def build_parser() -> argparse.ArgumentParser:
config_init = config_subparsers.add_parser("init", help="初始化配置文件到 data/")
config_init.set_defaults(handler=handle_config_init)
config_setup = config_subparsers.add_parser("setup", help="交互式配置引导")
config_setup.set_defaults(handler=handle_config_setup)
config_show = config_subparsers.add_parser("show", help="显示当前配置")
config_show.set_defaults(handler=handle_config_show)
@@ -263,6 +268,156 @@ def handle_config_init(args: argparse.Namespace) -> dict[str, Any]:
return {"ok": True, "config_file": str(path)}
def _is_interactive() -> bool:
try:
return os.isatty(sys.stdin.fileno())
except Exception:
return False
def _prompt_text(prompt: str, default: str = "") -> str:
suffix = f" [{default}]" if default else ""
try:
value = input(f"{prompt}{suffix}: ").strip()
except EOFError:
value = ""
except KeyboardInterrupt:
print("\n已取消配置。")
raise SystemExit(130)
return value or default
def _prompt_bool(prompt: str, default: bool = False) -> bool:
suffix = "(Y/n)" if default else "(y/N)"
try:
value = input(f"{prompt} {suffix}: ").strip().lower()
except EOFError:
value = ""
except KeyboardInterrupt:
print("\n已取消配置。")
raise SystemExit(130)
if not value:
return default
return value in {"y", "yes", "1", "true"}
def _prompt_int(prompt: str, default: int) -> int:
while True:
value = _prompt_text(prompt, str(default)).strip()
try:
return int(value)
except ValueError:
print("请输入整数。")
def _prompt_choice(prompt: str, options: list[str], default: str) -> str:
option_text = "/".join(options)
default_value = default if default in options else options[0]
while True:
value = _prompt_text(f"{prompt} ({option_text})", default_value).strip().lower()
if value in options:
return value
print(f"请输入以下选项之一: {option_text}")
def handle_config_setup(args: argparse.Namespace) -> dict[str, Any]:
if not _is_interactive():
raise ValueError("`config setup` 需要交互式终端，请直接编辑 data/sync_config.json 或先使用 config init")
init_config_from_example(PROJECT_ROOT)
cfg = load_sync_config()
print("开始交互式配置，直接回车表示使用当前值。\n")
cfg["proxy"] = _prompt_text("1) 注册代理地址", str(cfg.get("proxy") or "http://127.0.0.1:17891"))
cfg["auto_register"] = _prompt_bool("2) 池不足时自动注册", bool(cfg.get("auto_register", False)))
provider_options = ["mailtm", "duckmail", "moemail", "cloudflare_temp_email"]
current_provider = str((cfg.get("mail_providers") or ["mailtm"])[0]).strip().lower() or "mailtm"
provider_name = _prompt_choice("3) 邮箱提供商", provider_options, current_provider)
cfg["mail_providers"] = [provider_name]
cfg["mail_strategy"] = "round_robin"
provider_cfgs = dict(cfg.get("mail_provider_configs") or {})
provider_cfg = dict(provider_cfgs.get(provider_name) or {})
if provider_name == "mailtm":
provider_cfg["api_base"] = _prompt_text(" Mail.tm API 地址", str(provider_cfg.get("api_base") or "https://api.mail.tm"))
elif provider_name == "duckmail":
provider_cfg["api_base"] = _prompt_text(" DuckMail API 地址", str(provider_cfg.get("api_base") or "https://api.duckmail.sbs"))
provider_cfg["bearer_token"] = _prompt_text(" DuckMail Bearer Token", str(provider_cfg.get("bearer_token") or ""))
provider_cfg["domain"] = _prompt_text(" DuckMail 域名(可留空)", str(provider_cfg.get("domain") or ""))
elif provider_name == "moemail":
provider_cfg["api_base"] = _prompt_text(" MoeMail API 地址", str(provider_cfg.get("api_base") or ""))
provider_cfg["api_key"] = _prompt_text(" MoeMail API Key", str(provider_cfg.get("api_key") or ""))
elif provider_name == "cloudflare_temp_email":
provider_cfg["api_base"] = _prompt_text(" Cloudflare Worker API 地址", str(provider_cfg.get("api_base") or ""))
provider_cfg["admin_password"] = _prompt_text(" Worker 管理密码", str(provider_cfg.get("admin_password") or ""))
provider_cfg["domain"] = _prompt_text(" 邮箱域名后缀", str(provider_cfg.get("domain") or ""))
provider_cfgs[provider_name] = provider_cfg
cfg["mail_provider_configs"] = provider_cfgs
enable_sub2api = _prompt_bool("4) 是否配置 Sub2Api", bool(str(cfg.get("base_url") or "").strip()))
if enable_sub2api:
cfg["base_url"] = _prompt_text(" Sub2Api 地址", str(cfg.get("base_url") or "https://your-sub2api.example.com"))
auth_mode = _prompt_choice(
" Sub2Api 认证方式",
["token", "password"],
"token" if str(cfg.get("bearer_token") or "").strip() else "password",
)
if auth_mode == "token":
cfg["bearer_token"] = _prompt_text(" Bearer Token", str(cfg.get("bearer_token") or ""))
cfg["email"] = _prompt_text(" 管理员邮箱(可留空)", str(cfg.get("email") or ""))
cfg["password"] = ""
else:
cfg["email"] = _prompt_text(" 管理员邮箱", str(cfg.get("email") or "admin@example.com"))
cfg["password"] = _prompt_text(" 管理员密码", str(cfg.get("password") or ""))
cfg["bearer_token"] = ""
cfg["auto_sync"] = _prompt_bool(" 注册成功后自动同步到 Sub2Api", bool(cfg.get("auto_sync", False)))
cfg["sub2api_min_candidates"] = _prompt_int(" Sub2Api 候选池阈值", int(cfg.get("sub2api_min_candidates", 200)))
cfg["sub2api_auto_maintain"] = _prompt_bool(" 启用 Sub2Api 自动维护", bool(cfg.get("sub2api_auto_maintain", False)))
else:
cfg["base_url"] = ""
cfg["bearer_token"] = ""
cfg["email"] = ""
cfg["password"] = ""
cfg["auto_sync"] = False
enable_cpa = _prompt_bool("5) 是否配置 CPA 平台", bool(str(cfg.get("cpa_base_url") or "").strip()))
if enable_cpa:
cfg["cpa_base_url"] = _prompt_text(" CPA 地址", str(cfg.get("cpa_base_url") or "https://your-cpa.example.com"))
cfg["cpa_token"] = _prompt_text(" CPA Token", str(cfg.get("cpa_token") or ""))
cfg["min_candidates"] = _prompt_int(" CPA 候选池阈值", int(cfg.get("min_candidates", 1000)))
cfg["used_percent_threshold"] = _prompt_int(" CPA 已使用比例阈值", int(cfg.get("used_percent_threshold", 95)))
cfg["auto_maintain"] = _prompt_bool(" 启用 CPA 自动维护", bool(cfg.get("auto_maintain", True)))
cfg["maintain_interval_minutes"] = _prompt_int(" CPA 自动维护间隔(分钟)", int(cfg.get("maintain_interval_minutes", 30)))
else:
cfg["cpa_base_url"] = ""
cfg["cpa_token"] = ""
cfg["auto_maintain"] = False
enable_proxy_pool = _prompt_bool("6) 是否启用代理池", bool(cfg.get("proxy_pool_enabled", False)))
cfg["proxy_pool_enabled"] = enable_proxy_pool
if enable_proxy_pool:
cfg["proxy_pool_api_url"] = _prompt_text(" 代理池 API 地址", str(cfg.get("proxy_pool_api_url") or "https://zenproxy.top/api/fetch"))
cfg["proxy_pool_auth_mode"] = _prompt_choice(
" 代理池鉴权方式",
["header", "query"],
str(cfg.get("proxy_pool_auth_mode") or "header"),
)
cfg["proxy_pool_api_key"] = _prompt_text(" 代理池 API Key", str(cfg.get("proxy_pool_api_key") or ""))
cfg["proxy_pool_count"] = _prompt_int(" 每次取代理数量", int(cfg.get("proxy_pool_count", 1)))
cfg["proxy_pool_country"] = _prompt_text(" 代理国家代码", str(cfg.get("proxy_pool_country") or "US")).upper()
cfg["account_name"] = _prompt_text("7) 默认账号名称", str(cfg.get("account_name") or "AutoReg"))
cfg["upload_mode"] = _prompt_choice("8) 上传模式", ["snapshot", "decoupled"], str(cfg.get("upload_mode") or "snapshot"))
saved = save_sync_config(cfg)
print("\n配置已保存。建议下一步执行:")
print("- python3 run.py --json config show")
print("- python3 run.py register --once")
return {"ok": True, "config_file": str((PROJECT_ROOT / 'data' / 'sync_config.json')), "config": saved}
def handle_config_show(args: argparse.Namespace) -> dict[str, Any]:
return load_sync_config()

14
run.py
View File

@@ -8,6 +8,7 @@ if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from main import main
from support import has_initialized_config, init_config_from_example
def _default_args(argv: list[str]) -> list[str]:
@@ -16,5 +17,18 @@ def _default_args(argv: list[str]) -> list[str]:
return ["--help"]
def _maybe_print_first_run_hint(argv: list[str]) -> None:
if argv:
return
init_config_from_example(PROJECT_ROOT)
if has_initialized_config():
return
print("检测到当前还没有完成首次配置。")
print("建议先执行: python3 run.py config setup")
print("如果你只想先生成模板文件，也可以执行: python3 run.py config init")
print()
if __name__ == "__main__":
_maybe_print_first_run_hint(sys.argv[1:])
raise SystemExit(main(_default_args(sys.argv[1:])))

25
support.py
View File

@@ -109,6 +109,31 @@ def load_sync_config() -> Dict[str, Any]:
return normalize_config(copy.deepcopy(DEFAULT_CONFIG))
def has_initialized_config() -> bool:
if not CONFIG_FILE.exists():
return False
try:
data = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
except Exception:
return False
if not isinstance(data, dict):
return False
proxy = str(data.get("proxy") or "").strip()
providers = data.get("mail_providers") or []
cpa_base_url = str(data.get("cpa_base_url") or "").strip()
base_url = str(data.get("base_url") or "").strip()
bearer_token = str(data.get("bearer_token") or "").strip()
email = str(data.get("email") or "").strip()
password = str(data.get("password") or "").strip()
provider_ready = isinstance(providers, list) and any(str(item).strip() for item in providers)
sub2api_ready = bool(base_url and (bearer_token or (email and password)))
cpa_ready = bool(cpa_base_url and str(data.get("cpa_token") or "").strip())
return bool(proxy and provider_ready) or sub2api_ready or cpa_ready
def normalize_config(cfg: Dict[str, Any]) -> Dict[str, Any]:
cfg = copy.deepcopy(cfg or {})
legacy = str(cfg.get("mail_provider", "mailtm") or "mailtm").strip().lower()