autoChatGPTPlus-Paypal/task_runner.py

"""任务执行器：单 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