"""本地 Web 控制台：网页配置 + 一键全自动注册→付款→上传 CPA。""" from __future__ import annotations import json import queue import threading import time from dataclasses import asdict from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from chatgpt_flow import FullRunContext, run_full from config import AppConfig from cpa_uploader import build_cpa_auth_payload from storage import get_account, init_db, list_accounts, list_events HOST = "127.0.0.1" PORT = 7791 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(self, cfg: AppConfig) -> 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_full(cfg, log=self._log, on_stage=self._on_stage) except Exception as exc: import traceback self._log(f"[server] 任务异常: {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() INDEX_HTML = r""" ChatGPT Plus 全自动注册

ChatGPT Plus 全自动注册 + CPA 上传

流程：a4sky 邮箱注册 → 拿 Plus 长链 → PayPal 创建账号付款 → 校验 plan=plus → 上传 CPA。手机号统一 +15822201173。

配置

账号数量

浏览器模式

邮件助手 URL

邮箱域名

邮箱轮询间隔（秒）

邮箱轮询次数

使用 1 个月免费 promo

PayPal 短信手机号 (E164)

接码 API URL

CPA 地址

CPA 管理密钥

空闲

当前阶段

空闲

账号进度

#	邮箱	阶段	planType	CPA 文件	错误

已注册账号库

数据库：data/accounts.db

邮箱	plan	状态	CPA 文件	注册时间	更新时间	错误	动作

点击查看选中账号详情

实时日志

""" 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(200, INDEX_HTML.encode("utf-8"), "text/html; charset=utf-8") 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 if path == "/api/accounts": try: from urllib.parse import parse_qs q = parse_qs(query) status = (q.get("status") or [""])[0] or None limit = int((q.get("limit") or ["200"])[0]) accounts = list_accounts(limit=limit, status=status) # 别把 session 全文 dump 给列表，太大；列表只回主要字段 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" )}) self._send_json(200, {"accounts": slim}) 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 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 if path == "/api/start": try: cfg = AppConfig.load() err = JOB.start(cfg) 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(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))) self.end_headers() self.wfile.write(content) 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 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() server = ThreadingHTTPServer((HOST, PORT), Handler) print(f"ChatGPT Plus Auto Console: http://{HOST}:{PORT}/") try: server.serve_forever() except KeyboardInterrupt: print("\nStopped.") if __name__ == "__main__": main()