"""本地 Web 控制台：SSO 注册、账号库与任务 API。""" from __future__ import annotations import json import queue import threading import time from dataclasses import asdict from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from chatgpt_flow import FullRunContext, run_sso_batch from config import AppConfig from cpa_uploader import build_cpa_auth_payload from recheck import recheck_account from storage import ( count_accounts, create_task, get_account, get_task, init_db, list_accounts, list_events, list_tasks, ) from task_runner import get_runner, make_task_id HOST = "127.0.0.1" # main() 会读 cfg.api_host 覆盖 PORT = 7791 BASE_DIR = Path(__file__).resolve().parent UI_DIR = BASE_DIR / "ui" STATIC_TYPES = { ".css": "text/css; charset=utf-8", ".js": "application/javascript; charset=utf-8", ".html": "text/html; charset=utf-8", } class JobManager: def __init__(self): self.lock = threading.Lock() self.full_ctx: FullRunContext | None = None self.thread: threading.Thread | None = None self.log_queue: queue.Queue[str] = queue.Queue() self.history: list[str] = [] self.stage: str = "" def _log(self, msg: str): line = f"[{time.strftime('%H:%M:%S')}] {msg}" self.history.append(line) if len(self.history) > 4000: self.history = self.history[-3000:] self.log_queue.put(line) def _on_stage(self, name: str): self.stage = name # stage 也写到日志，便于复盘 self._log(f"[STAGE] {name}") def start_sso(self, account_count: int, sso_mail_domain: str, cpa_url: str, cpa_management_key: str, headless: bool, proxy_url: str) -> str: with self.lock: if self.thread and self.thread.is_alive(): return "已有任务在运行" self.history.clear() while not self.log_queue.empty(): self.log_queue.get_nowait() self.stage = "" def runner(): try: self.full_ctx = run_sso_batch( account_count=account_count, sso_mail_domain=sso_mail_domain, cpa_url=cpa_url, cpa_management_key=cpa_management_key, headless=headless, proxy_url=proxy_url, log=self._log, on_stage=self._on_stage, ) except Exception as exc: import traceback self._log(f"[server] SSO 任务异常: {exc!r}") self._log(traceback.format_exc()) self.thread = threading.Thread(target=runner, daemon=True) self.thread.start() return "" def stop(self): if self.full_ctx: self.full_ctx.state = "stopped" self._log("[user] 已请求停止") def status(self) -> dict: running = bool(self.thread and self.thread.is_alive()) ctx = self.full_ctx accounts = [] state = "idle" if ctx: state = ctx.state for a in ctx.accounts: accounts.append({ "email": a.get("email"), "stage": a.get("stage"), "planType": a.get("planType"), "error": a.get("error"), "cpaFile": (a.get("cpa") or {}).get("fileName") if a.get("cpa") else None, }) return { "running": running, "state": state, "stage": self.stage, "accounts": accounts, } JOB = JobManager() def _read_json(handler) -> dict: length = int(handler.headers.get("content-length") or "0") if length <= 0: return {} raw = handler.rfile.read(length).decode("utf-8", errors="replace") return json.loads(raw or "{}") class Handler(BaseHTTPRequestHandler): def do_GET(self): path = self.path.split("?", 1)[0] query = self.path.split("?", 1)[1] if "?" in self.path else "" if path in ("/", "/index.html"): self._send_static_file(UI_DIR / "index.html") return if path.startswith("/static/"): name = path[len("/static/"):] if "/" in name or "\\" in name or not name: self._send_json(404, {"error": "not found"}) return self._send_static_file(UI_DIR / name) return if path in ("/docs", "/docs/"): self._send(200, SWAGGER_HTML.encode("utf-8"), "text/html; charset=utf-8") return if path == "/openapi.json": self._send_json(200, _build_openapi_spec()) return if not self._check_auth(): return if path == "/api/status": self._send_json(200, JOB.status()) return if path == "/api/config": self._send_json(200, asdict(AppConfig.load())) return if path == "/api/log": self._stream_log() return # ===== 任务化 API（GET）===== if path == "/api/tasks": from urllib.parse import parse_qs q = parse_qs(query) status = (q.get("status") or [""])[0] or None limit = int((q.get("limit") or ["100"])[0]) try: tasks = list_tasks(limit=limit, status=status) self._send_json(200, {"tasks": tasks}) except Exception as exc: self._send_json(500, {"error": str(exc)}) return if path.startswith("/api/tasks/"): from urllib.parse import unquote task_id = unquote(path[len("/api/tasks/"):]) t = get_task(task_id) if not t: self._send_json(404, {"error": "task not found"}) return self._send_json(200, {"task": t}) return if path == "/api/accounts": try: from urllib.parse import parse_qs q = parse_qs(query) status = (q.get("status") or [""])[0] or None page_num = max(1, int((q.get("page") or ["1"])[0])) page_size = max(1, min(100, int((q.get("pageSize") or ["20"])[0]))) offset = (page_num - 1) * page_size total = count_accounts(status=status) accounts = list_accounts(limit=page_size, status=status, offset=offset) slim = [] for a in accounts: slim.append({k: a.get(k) for k in ( "email", "plan_type", "final_status", "cpa_file_name", "long_link", "last_error", "created_at", "updated_at", "cpa_uploaded_at", "trial_eligible", "trial_state", "is_trial_account", "can_retry_payment" )}) self._send_json(200, {"accounts": slim, "total": total, "page": page_num, "pageSize": page_size}) except Exception as exc: self._send_json(500, {"error": str(exc)}) return if path.startswith("/api/account/") and path.endswith("/cpa.json"): from urllib.parse import unquote email = unquote(path[len("/api/account/"):-len("/cpa.json")]) acc = get_account(email) if not acc: self._send_json(404, {"error": "account not found"}) return session = acc.get("plus_session") or acc.get("initial_session") if not session: self._send_json(404, {"error": "该账号没有可下载的 session"}) return try: payload = build_cpa_auth_payload(session, email_hint=email) except Exception as exc: self._send_json(500, {"error": f"构造 CPA auth JSON 失败: {exc}"}) return file_name = acc.get("cpa_file_name") or payload["fileName"] content = json.dumps(payload["authJson"], ensure_ascii=False, indent=2).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Disposition", f'attachment; filename="{file_name}"') self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) return if path.startswith("/api/account/"): from urllib.parse import unquote email = unquote(path[len("/api/account/"):]) acc = get_account(email) if not acc: self._send_json(404, {"error": "account not found"}) return events = list_events(email, limit=200) self._send_json(200, {"account": acc, "events": events}) return self._send_json(404, {"error": "not found"}) def do_POST(self): path = self.path.split("?", 1)[0] if not self._check_auth(): return if path == "/api/config": try: body = _read_json(self) cfg = AppConfig.load().update(body or {}) self._send_json(200, asdict(cfg)) except Exception as exc: self._send_json(500, {"error": str(exc)}) return # ===== 任务化 API ===== if path == "/api/tasks": try: body = _read_json(self) or {} mode = (body.get("mode") or "full").strip().lower() if mode == "full": self._send_json(410, {"error": "full 全自动注册任务已停用，请使用 SSO 注册入口"}) return if mode != "pay_only": self._send_json(400, {"error": "mode 必须是 pay_only（full 已停用）"}) return params = body.get("params") or {} if mode == "pay_only": sess = params.get("session") if not isinstance(sess, dict) or not sess.get("accessToken"): self._send_json(400, {"error": "pay_only 需要 params.session 是 JSON 且包含 accessToken"}) return max_attempts = int(body.get("max_attempts") or 3) max_attempts = max(1, min(10, max_attempts)) task_id = make_task_id() t = create_task(task_id, mode, params, max_attempts=max_attempts) # 启动 runner（幂等） get_runner(log=lambda m: JOB._log(m)) self._send_json(200, {"task_id": task_id, "task": t}) except Exception as exc: self._send_json(500, {"error": str(exc)}) return if path.startswith("/api/tasks/") and path.endswith("/cancel"): from urllib.parse import unquote task_id = unquote(path[len("/api/tasks/"):-len("/cancel")]) t = get_task(task_id) if not t: self._send_json(404, {"error": "task not found"}) return runner = get_runner(log=lambda m: JOB._log(m)) runner.cancel(task_id) self._send_json(200, {"ok": True, "task_id": task_id}) return if path == "/api/start": self._send_json(410, {"error": "ChatGPT Plus 全自动注册已停用，请使用 SSO 注册入口"}) return if path.startswith("/api/account/") and path.endswith("/recheck"): from urllib.parse import unquote email = unquote(path[len("/api/account/"):-len("/recheck")]) cfg = AppConfig.load() try: result = recheck_account( email, cpa_url=cfg.cpa_url, cpa_management_key=cfg.cpa_management_key, log=lambda msg: JOB._log(f"[acc:{email[:24]}] {msg}"), ) self._send_json(200, result) except Exception as exc: self._send_json(500, {"error": str(exc)}) return if path.startswith("/api/account/") and path.endswith("/retry_payment"): from urllib.parse import unquote email = unquote(path[len("/api/account/"):-len("/retry_payment")]) acc = get_account(email) if not acc: self._send_json(404, {"error": "账号不存在"}) return if not acc.get("can_retry_payment"): self._send_json(400, {"error": "该账号当前不支持直接重新付款"}) return session = acc.get("plus_session") or acc.get("initial_session") if not session or not isinstance(session, dict) or not session.get("accessToken"): self._send_json(400, {"error": "该账号没有可用的 session（缺少 accessToken）"}) return try: task_id = make_task_id() t = create_task(task_id, "pay_only", {"session": session, "email": email}, max_attempts=3) get_runner(log=lambda m: JOB._log(m)) self._send_json(200, {"ok": True, "task_id": task_id, "task": t}) except Exception as exc: self._send_json(500, {"error": str(exc)}) return if path == "/api/start-sso": try: body = _read_json(self) account_count = int(body.get("account_count") or 0) if account_count < 1: self._send_json(400, {"error": "account_count 必须 >= 1"}) return cfg = AppConfig.load() err = JOB.start_sso( account_count=account_count, sso_mail_domain=str(body.get("sso_mail_domain") or cfg.sso_mail_domain or "aef.claudeai.life"), cpa_url=str(body.get("cpa_url") or cfg.cpa_url or ""), cpa_management_key=str(body.get("cpa_management_key") or cfg.cpa_management_key or ""), headless=bool(body.get("headless")) if "headless" in body else cfg.headless, proxy_url=str(body.get("proxy_url") or ""), ) if err: self._send_json(409, {"error": err}) else: self._send_json(200, {"ok": True}) except Exception as exc: self._send_json(500, {"error": str(exc)}) return if path == "/api/stop": JOB.stop() self._send_json(200, {"ok": True}) return self._send_json(404, {"error": "not found"}) def _stream_log(self): self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self.end_headers() try: for line in JOB.history[-300:]: self._sse_send(line) while True: try: line = JOB.log_queue.get(timeout=15) self._sse_send(line) except queue.Empty: self.wfile.write(b": ping\n\n") self.wfile.flush() except (BrokenPipeError, ConnectionResetError): return def _sse_send(self, line: str): for piece in line.splitlines() or [""]: self.wfile.write(b"data: " + piece.encode("utf-8") + b"\n") self.wfile.write(b"\n") self.wfile.flush() def _send_json(self, status: int, payload: dict): self._send(status, json.dumps(payload, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") def _send_static_file(self, file_path: Path): try: resolved = file_path.resolve() if UI_DIR.resolve() not in resolved.parents and resolved != (UI_DIR / "index.html").resolve(): self._send_json(404, {"error": "not found"}) return content = resolved.read_bytes() except Exception: self._send_json(404, {"error": "not found"}) return content_type = STATIC_TYPES.get(resolved.suffix.lower(), "application/octet-stream") self._send(200, content, content_type) def _send(self, status: int, content: bytes, content_type: str): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(content))) # CORS（仅 /api/* 需要时由调用方决定，但统一发也无害） try: cfg = AppConfig.load() origin = (cfg.api_cors_origin or "*").strip() self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Access-Control-Allow-Credentials", "true") except Exception: self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(content) def do_OPTIONS(self): # CORS preflight self.send_response(204) try: cfg = AppConfig.load() origin = (cfg.api_cors_origin or "*").strip() except Exception: origin = "*" self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization") self.send_header("Access-Control-Max-Age", "86400") self.send_header("Access-Control-Allow-Credentials", "true") self.end_headers() def _check_auth(self) -> bool: """非空 api_token 时校验 Authorization: Bearer。返回 True 表示放行。""" try: cfg = AppConfig.load() token = (cfg.api_token or "").strip() except Exception: token = "" if not token: return True # 公开接口豁免：根页面、OpenAPI 文档、Swagger UI、static 静态 path = self.path.split("?", 1)[0] public = ("/", "/index.html", "/docs", "/docs/", "/openapi.json", "/openapi.yaml") if path in public: return True auth = self.headers.get("Authorization", "") if auth == f"Bearer {token}": return True # 也支持 ?token=xxx if "token=" in (self.path.split("?", 1)[1] if "?" in self.path else ""): from urllib.parse import parse_qs q = parse_qs(self.path.split("?", 1)[1]) if (q.get("token") or [""])[0] == token: return True self._send_json(401, {"error": "missing or invalid Bearer token"}) return False def log_message(self, fmt, *args): return def handle_one_request(self): try: return super().handle_one_request() except (ConnectionResetError, BrokenPipeError): # 浏览器主动断开 SSE / fetch 时打印栈很碍眼，直接静音 self.close_connection = True SWAGGER_HTML = r""" API 文档 · ChatGPT Plus 自动化
""" def _build_openapi_spec() -> dict: """生成 OpenAPI 3.1 规范。""" cfg = AppConfig.load() return { "openapi": "3.1.0", "info": { "title": "Auto PayPal API", "version": "1.0.0", "description": ( "SSO 注册、账号库查询与已有 session 付款任务 API。\n\n" "**可用任务模式**：\n" "- `pay_only` — 传入已有 session JSON，跳过注册直接付款→上传 CPA\n" "- `full` 全自动注册模式已停用；请使用 Web UI 的 SSO 注册入口。\n\n" "**调用流程**：\n" "1. POST `/api/tasks` 创建任务，立即拿到 `task_id`\n" "2. 轮询 GET `/api/tasks/{task_id}` 看 `status` 和 `stage`\n" "3. `status` 变 `success` 时可调 GET `/api/account/{email}/cpa.json` 下载 CPA 文件\n\n" "**重试**：每个任务整体失败会重试 `max_attempts` 次（默认 3）。" ), }, "servers": [ {"url": f"http://{cfg.api_host or '127.0.0.1'}:{cfg.api_port or 7791}", "description": "当前实例"}, ], "components": { "securitySchemes": { "BearerAuth": { "type": "http", "scheme": "bearer", "description": "如果配置了 `api_token`，所有 /api/* 请求需带 `Authorization: Bearer `。也支持 `?token=xxx` 查询参数。", } }, "schemas": { "Task": { "type": "object", "properties": { "task_id": {"type": "string", "example": "t-1779470219-65e44d55"}, "mode": {"type": "string", "enum": ["full", "pay_only"]}, "status": {"type": "string", "enum": ["queued", "running", "success", "failed", "cancelled"]}, "stage": {"type": "string", "description": "当前阶段描述"}, "attempts": {"type": "integer"}, "max_attempts": {"type": "integer"}, "params": {"type": "object", "description": "创建任务时传入的参数（脱敏后）"}, "result": {"type": "object", "nullable": True, "description": "成功时的结果（含 CPA 文件名等）"}, "last_error": {"type": "string", "nullable": True}, "email": {"type": "string", "nullable": True, "description": "注册成功的 ChatGPT 邮箱"}, "plan_type": {"type": "string", "nullable": True, "example": "plus"}, "cpa_file_name": {"type": "string", "nullable": True, "example": "codex-foo@example.com-plus.json"}, "created_at": {"type": "integer", "description": "毫秒时间戳"}, "updated_at": {"type": "integer"}, "started_at": {"type": "integer", "nullable": True}, "finished_at": {"type": "integer", "nullable": True}, }, }, "CreateTaskRequest": { "type": "object", "required": ["mode"], "properties": { "mode": {"type": "string", "enum": ["pay_only"]}, "max_attempts": {"type": "integer", "default": 3, "minimum": 1, "maximum": 10}, "params": { "type": "object", "description": "可覆盖全局配置；pay_only 模式必须包含 session 字段", "properties": { "session": { "type": "object", "description": "ChatGPT /api/auth/session 完整 JSON（仅 pay_only 模式必填）", "properties": { "accessToken": {"type": "string"}, "user": {"type": "object"}, "account": {"type": "object"}, }, }, "headless": {"type": "boolean"}, "use_promo": {"type": "boolean"}, "phone_e164": {"type": "string", "example": "+15822201173"}, "sms_api_url": {"type": "string"}, "cpa_url": {"type": "string"}, "cpa_management_key": {"type": "string"}, "proxy_url": {"type": "string"}, "paypal_only_proxy": {"type": "string"}, "mail_helper_url": {"type": "string"}, "mail_domain": {"type": "string"}, }, }, }, }, "Account": { "type": "object", "properties": { "email": {"type": "string"}, "plan_type": {"type": "string", "nullable": True}, "final_status": {"type": "string"}, "cpa_file_name": {"type": "string", "nullable": True}, "long_link": {"type": "string", "nullable": True}, "last_error": {"type": "string", "nullable": True}, "created_at": {"type": "integer"}, "updated_at": {"type": "integer"}, "cpa_uploaded_at": {"type": "integer", "nullable": True}, }, }, "Error": { "type": "object", "properties": {"error": {"type": "string"}}, }, }, }, "security": [{"BearerAuth": []}] if cfg.api_token else [], "paths": { "/api/tasks": { "post": { "tags": ["Tasks"], "summary": "创建任务", "description": "创建一个 pay_only 任务，立即返回 task_id，任务异步执行。full 全自动注册已停用。", "requestBody": { "required": True, "content": { "application/json": { "schema": {"$ref": "#/components/schemas/CreateTaskRequest"}, "examples": { "pay_only": { "summary": "传入 session 直接付款", "value": { "mode": "pay_only", "max_attempts": 3, "params": { "session": { "accessToken": "eyJxxx...", "user": {"email": "user@example.com"}, "account": {"planType": "free"}, } }, }, }, }, } }, }, "responses": { "200": { "description": "任务已创建", "content": { "application/json": { "schema": { "type": "object", "properties": { "task_id": {"type": "string"}, "task": {"$ref": "#/components/schemas/Task"}, }, } } }, }, "400": {"description": "参数错误", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}}}, "401": {"description": "Bearer token 缺失或无效"}, "410": {"description": "full 全自动注册已停用"}, }, }, "get": { "tags": ["Tasks"], "summary": "列出任务", "parameters": [ {"name": "status", "in": "query", "schema": {"type": "string", "enum": ["queued", "running", "success", "failed", "cancelled"]}}, {"name": "limit", "in": "query", "schema": {"type": "integer", "default": 100}}, ], "responses": { "200": { "content": { "application/json": { "schema": { "type": "object", "properties": {"tasks": {"type": "array", "items": {"$ref": "#/components/schemas/Task"}}}, } } } } }, }, }, "/api/tasks/{task_id}": { "get": { "tags": ["Tasks"], "summary": "查询任务进度", "description": "轮询此接口查任务实时 status 和 stage。建议 5-10 秒间隔。", "parameters": [{"name": "task_id", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": { "200": {"content": {"application/json": {"schema": {"type": "object", "properties": {"task": {"$ref": "#/components/schemas/Task"}}}}}}, "404": {"description": "任务不存在"}, }, } }, "/api/tasks/{task_id}/cancel": { "post": { "tags": ["Tasks"], "summary": "取消任务", "description": "请求取消任务。如果任务已经在跑，会在下一个 stop 检查点退出。", "parameters": [{"name": "task_id", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": {"200": {"description": "已请求取消"}, "404": {"description": "任务不存在"}}, } }, "/api/accounts": { "get": { "tags": ["Accounts"], "summary": "列出已注册账号", "parameters": [ {"name": "status", "in": "query", "schema": {"type": "string"}, "description": "如 cpa_uploaded / plus_check_failed"}, {"name": "limit", "in": "query", "schema": {"type": "integer", "default": 200}}, ], "responses": { "200": { "content": { "application/json": { "schema": { "type": "object", "properties": {"accounts": {"type": "array", "items": {"$ref": "#/components/schemas/Account"}}}, } } } } }, } }, "/api/account/{email}": { "get": { "tags": ["Accounts"], "summary": "查询账号详情（含完整 session 和事件流）", "parameters": [{"name": "email", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": {"200": {"description": "OK"}, "404": {"description": "账号不存在"}}, } }, "/api/account/{email}/cpa.json": { "get": { "tags": ["Accounts"], "summary": "下载 CPA codex auth JSON", "description": "返回该账号当时上传给 CPA 的完整 codex auth JSON 文件。带 Content-Disposition 头，浏览器会自动下载。", "parameters": [{"name": "email", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": { "200": {"description": "OK", "content": {"application/json": {}}}, "404": {"description": "账号或 session 不存在"}, }, } }, "/api/account/{email}/recheck": { "post": { "tags": ["Accounts"], "summary": "对失败账号补救", "description": "用 DB 里存的 access_token 调 backend-api/me，若已 plus 则自动重传 CPA。", "parameters": [{"name": "email", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": {"200": {"description": "OK"}, "404": {"description": "账号不存在"}}, } }, "/api/config": { "get": {"tags": ["Config"], "summary": "读取当前配置", "responses": {"200": {"description": "OK"}}}, "post": { "tags": ["Config"], "summary": "更新配置", "requestBody": {"content": {"application/json": {"schema": {"type": "object"}}}}, "responses": {"200": {"description": "OK"}}, }, }, "/api/status": { "get": {"tags": ["Misc"], "summary": "（旧）读取 UI 任务状态", "responses": {"200": {"description": "OK"}}} }, "/api/log": { "get": {"tags": ["Misc"], "summary": "实时日志（Server-Sent Events）", "responses": {"200": {"description": "text/event-stream"}}} }, }, "tags": [ {"name": "Tasks", "description": "任务化 API（推荐用法）"}, {"name": "Accounts", "description": "账号库"}, {"name": "Config", "description": "服务配置"}, {"name": "Misc", "description": "其他"}, ], } def _silence_threading_excepthook(): """ThreadingHTTPServer 在 worker 线程里仍可能抛 ConnectionResetError；接住它。""" import threading prev = threading.excepthook def hook(args): if isinstance(args.exc_value, (ConnectionResetError, BrokenPipeError)): return prev(args) threading.excepthook = hook def main(): init_db() _silence_threading_excepthook() # 启动后台任务 worker（幂等） get_runner(log=lambda m: JOB._log(m)) cfg = AppConfig.load() host = (cfg.api_host or HOST).strip() or HOST port = int(cfg.api_port or PORT) server = ThreadingHTTPServer((host, port), Handler) print(f"Auto PayPal Console:") print(f" Web UI: http://{host}:{port}/") print(f" Docs: http://{host}:{port}/docs") print(f" OpenAPI: http://{host}:{port}/openapi.json") if cfg.api_token: print(f" Auth: Bearer （已启用）") if host == "0.0.0.0": print(f" ⚠️ 当前监听所有网卡，外网可访问。建议设置 api_token。") try: server.serve_forever() except KeyboardInterrupt: print("\nStopped.") if __name__ == "__main__": main()