| """注册机 CLI 入口。 |
| |
| 用法： |
| uv run python -m registrar -n 5 -w 2 --proxy http://host:port |
| uv run python -m registrar --count 0 --workers 3 # 无限运行直到 Ctrl+C |
| |
| 并发模型：ThreadPoolExecutor 维持 -w 个任务在飞；单账号失败不致命，仅记日志。 |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import sys |
| import traceback |
| from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait |
|
|
| from registrar.http_client import HttpClient |
| from registrar.models import RegistrarConfig, load_registrar_config |
| from registrar.pipeline import register_one |
|
|
|
|
| def _build_parser() -> argparse.ArgumentParser: |
| ap = argparse.ArgumentParser(prog="registrar", description="Anuma2api 注册机（mailboxtemp 临邮 + hCaptcha）") |
| ap.add_argument("-n", "--count", type=int, default=1, |
| help="本次注册数量上限（0=无限循环直到 Ctrl+C）") |
| ap.add_argument("-w", "--workers", type=int, default=1, help="并发线程数") |
| ap.add_argument( |
| "--proxy", default=None, |
| help="覆盖 [proxy]：HTTP/HTTPS/SOCKS 代理 URL（注册请求与 captcha 共用；" |
| "未传则用 config 的 registrar_url → url）", |
| ) |
| ap.add_argument("--config", default=None, help="config.toml 路径（默认 $TWOAPI_CONFIG 或 config.toml）") |
| ap.add_argument("--captcha-method", default=None, |
| help="覆盖 config [captcha].method（cdp/semi/api）") |
| return ap |
|
|
|
|
| def _resolve_cli_proxy(cfg: RegistrarConfig, cli_proxy: str | None) -> str | None: |
| """CLI ``--proxy`` 优先，否则用配置解析结果（registrar → default → 直连）。""" |
| if cli_proxy is not None and str(cli_proxy).strip(): |
| return str(cli_proxy).strip() |
| return cfg.effective_proxy() |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| args = _build_parser().parse_args(argv) |
| cfg = load_registrar_config(args.config) |
| proxy = _resolve_cli_proxy(cfg, args.proxy) |
| |
| if proxy: |
| cfg.captcha.proxy_url = proxy |
| http = HttpClient(proxy=proxy) |
|
|
| infinite = args.count <= 0 |
| target = args.count |
| workers = max(args.workers, 1) |
| success = failed = submitted = 0 |
|
|
| try: |
| with ThreadPoolExecutor(max_workers=workers) as pool: |
| def fill(futures: set) -> None: |
| nonlocal submitted |
| while len(futures) < workers and (infinite or submitted < target): |
| futures.add(pool.submit(_safe_register, cfg, http, args, proxy)) |
| submitted += 1 |
|
|
| futures: set = set() |
| fill(futures) |
| while futures: |
| done, futures = wait(futures, return_when=FIRST_COMPLETED) |
| for fut in done: |
| ok, msg = fut.result() |
| if ok: |
| success += 1 |
| print(f"[OK] {msg}", flush=True) |
| else: |
| failed += 1 |
| print(f"[FAIL] {msg}", flush=True, file=sys.stderr) |
| fill(futures) |
| except KeyboardInterrupt: |
| print("\n[!] 收到中断，等待在飞任务结束后退出...", file=sys.stderr) |
|
|
| print(f"\n==== 完成：成功 {success}，失败 {failed} ====", flush=True) |
| return 0 |
|
|
|
|
| def _safe_register( |
| cfg: RegistrarConfig, |
| http: HttpClient, |
| args: argparse.Namespace, |
| proxy: str | None, |
| ) -> tuple[bool, str]: |
| """单账号注册的异常包装：失败返回 (False, 错误摘要)。""" |
| try: |
| acc = register_one(cfg, http, proxy=proxy, captcha_method=args.captcha_method) |
| return True, f"{acc.get('source_email', '?')} -> account/{acc.get('name', '?')}.json" |
| except Exception as exc: |
| tail = traceback.format_exc().splitlines()[-1] |
| return False, f"{exc} | {tail}" |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
