"""任务执行器：单 worker 后台线程，从 SQLite tasks 表 FIFO 消费任务。 支持两种模式： - full : 走完整的注册→付款→上传 CPA 流程（cfg.account_count 强制为 1） - pay_only : 传入已有 ChatGPT session JSON，跳过注册直接付款 任务整体失败会重试，最多 attempts = max_attempts 次（默认 3）。 """ from __future__ import annotations import threading import time import traceback import uuid from typing import Callable, Optional from chatgpt_flow import run_full, run_pay_only from config import AppConfig from storage import ( get_next_queued_task, get_task, init_db, update_task, ) class TaskRunner: def __init__(self, *, log: Callable[[str], None] = print): self.log = log self._thread: threading.Thread | None = None self._stopping = threading.Event() self._current_task_id: str | None = None self._cancel_flags: dict[str, bool] = {} self._lock = threading.Lock() # ----- public API ----- def start(self): if self._thread and self._thread.is_alive(): return self._stopping.clear() self._thread = threading.Thread(target=self._loop, daemon=True, name="task-runner") self._thread.start() self.log("[runner] worker 已启动") def shutdown(self, timeout: float = 5.0): self._stopping.set() if self._thread: self._thread.join(timeout=timeout) self.log("[runner] worker 已停止") def cancel(self, task_id: str) -> bool: with self._lock: self._cancel_flags[task_id] = True # 若是当前正在跑的 → 让 stop_check 抛出 return True def current_task_id(self) -> str | None: return self._current_task_id # ----- internals ----- def _loop(self): init_db() while not self._stopping.is_set(): try: task = get_next_queued_task() except Exception as exc: self.log(f"[runner] 取任务异常: {exc!r}") task = None if not task: # 没活干，sleep 1s 再轮询 self._stopping.wait(1.0) continue try: self._run_task(task) except Exception as exc: self.log(f"[runner] 执行任务 {task.get('task_id')} 顶层异常: {exc!r}") self.log(traceback.format_exc()) def _run_task(self, task: dict): task_id = task["task_id"] mode = task.get("mode") or "full" max_attempts = int(task.get("max_attempts") or 3) params = task.get("params") or {} self._current_task_id = task_id update_task(task_id, {"status": "running", "started_at": int(time.time() * 1000), "stage": "queued→running"}) last_error = "" for attempt in range(1, max_attempts + 1): if self._is_cancelled(task_id): update_task(task_id, {"status": "cancelled", "stage": "cancelled", "finished_at": int(time.time() * 1000)}) self._current_task_id = None return update_task(task_id, {"attempts": attempt, "stage": f"attempt {attempt}/{max_attempts}"}) self.log(f"[runner] task={task_id} mode={mode} 第 {attempt}/{max_attempts} 次尝试") try: if mode == "full": record = self._do_full(task_id, params) elif mode == "pay_only": record = self._do_pay_only(task_id, params) else: raise RuntimeError(f"未知 mode: {mode}") except _TaskCancelled: update_task(task_id, { "status": "cancelled", "stage": "cancelled", "finished_at": int(time.time() * 1000), }) self._current_task_id = None return except Exception as exc: last_error = repr(exc) self.log(f"[runner] task={task_id} 第 {attempt} 次执行异常: {last_error}") self.log(traceback.format_exc()) update_task(task_id, { "last_error": last_error, "stage": f"attempt {attempt} failed", }) if attempt >= max_attempts: break # sleep 5s 让外部资源喘口气 time.sleep(5) continue # 看 record["stage"] 决定是成功还是失败 stage = (record or {}).get("stage", "") if stage in ("cpa_uploaded", "cpa_skipped"): update_task(task_id, { "status": "success", "stage": stage, "result": record, "email": record.get("email") or task.get("email"), "plan_type": record.get("planType"), "cpa_file_name": (record.get("cpa") or {}).get("fileName") if record.get("cpa") else None, "finished_at": int(time.time() * 1000), "last_error": "", }) self.log(f"[runner] task={task_id} 成功 stage={stage}") self._current_task_id = None return else: # plus_check_failed / error 之类视为失败，进入重试 last_error = (record or {}).get("error") or f"stage={stage}" self.log(f"[runner] task={task_id} 第 {attempt} 次完成但未成功: {last_error}") update_task(task_id, {"last_error": last_error, "stage": f"attempt {attempt}: {stage}"}) if attempt >= max_attempts: break time.sleep(5) # 走到这说明全部 attempt 都失败 update_task(task_id, { "status": "failed", "stage": "exhausted", "last_error": last_error or "all attempts failed", "finished_at": int(time.time() * 1000), }) self.log(f"[runner] task={task_id} 失败（{max_attempts}/{max_attempts} 次都失败）") self._current_task_id = None def _do_full(self, task_id: str, params: dict) -> dict: cfg = AppConfig.load() cfg.account_count = 1 # 单任务只跑一个账号 # params 可覆盖 cfg for k in ("headless", "use_promo", "phone_e164", "sms_api_url", "cpa_url", "cpa_management_key", "proxy_url", "paypal_only_proxy", "mail_helper_url", "mail_domain"): if k in (params or {}): setattr(cfg, k, params[k]) def stop_check(): if self._is_cancelled(task_id): raise _TaskCancelled() def task_log(msg: str): self.log(f"[task:{task_id[:8]}] {msg}") def task_stage(stage: str): update_task(task_id, {"stage": stage[:200]}) # run_full 是为多账号写的，这里复用但只跑 1 个 # 把 stop hook 注入：run_full 内部用 full_ctx.check_stop，我们没法直接钩， # 但 Stop API 通过 cancel 设标志位 → 在每次 update_task 时也 stop_check # 简化：包一层 thread 跑，10s 内查一次 cancel stopping_holder = {"stop": False} result_holder: dict = {} def run(): try: full_ctx = run_full(cfg, log=task_log, on_stage=task_stage) # full_ctx.accounts[0] 就是结果 if full_ctx.accounts: result_holder["record"] = full_ctx.accounts[0] else: result_holder["record"] = {"stage": "error", "error": "no account record"} except Exception as exc: result_holder["error"] = repr(exc) # 因为 run_full 内部在阻塞调用 sync_playwright，cancel 只能等当前 attempt 跑完 run() if "error" in result_holder: raise RuntimeError(result_holder["error"]) if self._is_cancelled(task_id): raise _TaskCancelled() return result_holder.get("record") or {"stage": "error", "error": "empty record"} def _do_pay_only(self, task_id: str, params: dict) -> dict: cfg = AppConfig.load() cfg.account_count = 1 for k in ("headless", "use_promo", "phone_e164", "sms_api_url", "cpa_url", "cpa_management_key", "proxy_url", "paypal_only_proxy"): if k in (params or {}): setattr(cfg, k, params[k]) session = params.get("session") if not isinstance(session, dict) or not session.get("accessToken"): raise RuntimeError("pay_only 需要 params.session 是 JSON 且包含 accessToken") def stop_check(): if self._is_cancelled(task_id): raise _TaskCancelled() def task_log(msg: str): self.log(f"[task:{task_id[:8]}] {msg}") def task_stage(stage: str): update_task(task_id, {"stage": stage[:200]}) return run_pay_only( cfg, session=session, log=task_log, on_stage=task_stage, stop_check=stop_check, ) def _is_cancelled(self, task_id: str) -> bool: with self._lock: return self._cancel_flags.get(task_id, False) class _TaskCancelled(Exception): pass # ------------------ public helpers ------------------ def make_task_id() -> str: return f"t-{int(time.time())}-{uuid.uuid4().hex[:8]}" _GLOBAL_RUNNER: TaskRunner | None = None _GLOBAL_RUNNER_LOCK = threading.Lock() def get_runner(log: Callable[[str], None] = print) -> TaskRunner: global _GLOBAL_RUNNER with _GLOBAL_RUNNER_LOCK: if _GLOBAL_RUNNER is None: _GLOBAL_RUNNER = TaskRunner(log=log) _GLOBAL_RUNNER.start() return _GLOBAL_RUNNER