优化了部分支付宝界面

2026-08-12 19:12:22 +08:00
parent 1a20c367db
commit 5074faf9f6
13 changed files with 980 additions and 117 deletions
@@ -55,10 +55,17 @@ def _safe_log(job: dict, line: str) -> None:
        job["logs"] = (job.get("logs", []) + [clean.strip()])[-100:]


-def _run_process(job_id: str, command: list[str], phase: str) -> None:
+def _run_process(job_id: str, command: list[str], phase: str, mark_success: bool = True) -> int:
+    """Run one stage process and return its exit code.
+
+    When ``mark_success`` is False the caller owns the post-success state
+    transition (used by the staged payment flow).
+    """
    job = _jobs[job_id]
    environment = os.environ.copy()
    environment.pop("NODE_OPTIONS", None)
+    # 子进程输出经管道实时转存日志；关闭块缓冲避免日志滞后。
+    environment["PYTHONUNBUFFERED"] = "1"
    with _lock:
        job["phase"] = phase
        job["status"] = "running"
@@ -75,21 +82,27 @@ def _run_process(job_id: str, command: list[str], phase: str) -> None:
        with _lock:
            job["process_pid"] = None
            if code != 0:
-                job["status"] = "failed"
-                job["phase"] = phase
-                job["message"] = f"{phase}失败（退出码 {code}）"
+                if job.get("status") == "stopped":
+                    # 停止请求已 kill 本进程：保留停止语义，不被阶段失败覆盖。
+                    job["message"] = "任务已停止"
+                else:
+                    job["status"] = "failed"
+                    job["phase"] = phase
+                    job["message"] = f"{phase}失败（退出码 {code}）"
            elif phase == "login":
                job["status"] = "ready"
                job["phase"] = "selection"
                job["message"] = "登录成功，请选择平台、点券、区服和角色"
-            else:
+            elif mark_success:
                job["status"] = "success"
                job["phase"] = "completed"
                job["message"] = "付款流程已完成"
+        return code
    except Exception as exc:  # noqa: BLE001
        with _lock:
            job["status"] = "failed"
            job["message"] = str(exc)
+        return 1


 def _start_login(job_id: str, provider: str, timeout: int) -> None:
@@ -131,47 +144,158 @@ def _selection_options(job_id: str, platform: str, points: int | None, zone_id:
            "default_product": product, "default_zone": selected_zone}


-def _start_payment(job_id: str, selection: dict) -> None:
+def _payment_stage(job_id: str, command: list[str], phase: str, running_message: str,
+                   failed_message: str) -> bool:
+    """Run one payment stage; return True on success without touching final state."""
+    job = _jobs[job_id]
+    with _lock:
+        if job.get("status") == "stopped":
+            return False
+        job["status"] = "running"
+        job["phase"] = phase
+        job["message"] = running_message
+    code = _run_process(job_id, command, phase, mark_success=False)
+    if code != 0:
+        with _lock:
+            if job.get("status") == "stopped":
+                return False
+            job["status"] = "failed"
+            job["phase"] = phase
+            job["message"] = f"{failed_message}（退出码 {code}）"
+        return False
+    return True
+
+
+PAYMENT_CHECK_INTERVAL = 3
+PAYMENT_CHECK_TIMEOUT = 300
+
+
+def _check_payment_once(job_id: str) -> int:
+    """Read-only completion check; returns 0=confirmed, 1=not yet, 2=check failed."""
+    job = _jobs[job_id]
+    directory = _job_dir(job_id)
+    command = [sys.executable, "scripts/jsdom-pay.py", "--check-only",
+               "--session", str(directory / "mall-session.json"),
+               "--out-dir", str(directory / "jsdom-order")]
+    environment = os.environ.copy()
+    environment.pop("NODE_OPTIONS", None)
+    try:
+        result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True,
+                                env=environment, timeout=90)
+    except subprocess.TimeoutExpired:
+        with _lock:
+            job["payment_last_checked_at"] = int(time.time())
+        return 2
+    for line in (result.stdout or "").splitlines():
+        _safe_log(job, line)
+    for line in (result.stderr or "").splitlines():
+        _safe_log(job, line)
+    with _lock:
+        job["payment_last_checked_at"] = int(time.time())
+    return result.returncode
+
+
+def _monitor_payment(job_id: str) -> None:
+    """Background loop that re-checks completion until timeout or confirmation."""
+    job = _jobs[job_id]
+    deadline = time.monotonic() + PAYMENT_CHECK_TIMEOUT
+    while time.monotonic() < deadline:
+        time.sleep(PAYMENT_CHECK_INTERVAL)
+        with _lock:
+            if job.get("status") in {"stopped", "success"}:
+                return
+        code = _check_payment_once(job_id)
+        if code == 0:
+            with _lock:
+                job["status"] = "success"
+                job["phase"] = "completed"
+                job["message"] = "已确认到账，充值完成"
+            return
+        if code == 2:
+            with _lock:
+                job["message"] = "到账检测暂时失败，将继续重试"
+    with _lock:
+        job["status"] = "payment_timeout"
+        job["phase"] = "payment"
+        job["message"] = "付款码已生成，但未在时限内确认到账，请人工核对"
+
+
+def _payment_flow(job_id: str, selection: dict) -> None:
+    """Staged payment: create mall order -> create payment QR -> monitor arrival.
+
+    The QR is generated before any waiting, so the UI switches to
+    waiting_payment immediately; only a timeout moves to payment_timeout.
+    """
+    job = _jobs[job_id]
    directory = _job_dir(job_id)
    session = directory / "mall-session.json"
    response = directory / "mall-order-response.json"
    output = directory / "jsdom-order"
-    command = [sys.executable, "main.py", "mall", "auto", "--session", str(session),
-               "--order-template", str(ROOT / "config" / "mall-order-template.json"),
-               "--quantity", "1", "--product-id", str(selection["product_id"]),
-               "--offer-id", str(selection["offer_id"]),
-               "--role-id", str(selection["role_id"]), "--role-name", str(selection["role_name"]),
-               "--zone-id", str(selection["zone_id"]), "--zone-name", str(selection["zone_name"]),
-               "--output", str(response)]
+    with _lock:
+        job["status"] = "ordering"
+        job["phase"] = "payment"
+        job["message"] = "正在创建商城订单"
+    order_cmd = [sys.executable, "main.py", "mall", "auto", "--session", str(session),
+                 "--order-template", str(ROOT / "config" / "mall-order-template.json"),
+                 "--quantity", "1", "--product-id", str(selection["product_id"]),
+                 "--offer-id", str(selection["offer_id"]),
+                 "--role-id", str(selection["role_id"]), "--role-name", str(selection["role_name"]),
+                 "--zone-id", str(selection["zone_id"]), "--zone-name", str(selection["zone_name"]),
+                 "--output", str(response)]
    if selection.get("order_pf"):
-        command.extend(["--pf", str(selection["order_pf"])])
-    def run() -> None:
-        _run_process(job_id, command, "order")
-        job = _jobs[job_id]
-        if job.get("status") != "success" or not response.exists():
+        order_cmd.extend(["--pf", str(selection["order_pf"])])
+    if not _payment_stage(job_id, order_cmd, "order", "正在创建商城订单", "创建商城订单失败"):
+        return
+    pay_cmd = [sys.executable, "scripts/jsdom-pay.py", "--session", str(session),
+               "--mall-response", str(response), "--out-dir", str(output),
+               "--zone-id", str(selection["zone_id"]),
+               "--pf", str(selection.get("order_pf", "")),
+               "--amount-fen", str(selection["price_fen"]),
+               "--skip-payment-check"]
+    if not _payment_stage(job_id, pay_cmd, "payment", "正在生成微信付款码", "生成付款码失败"):
+        return
+    meta_path = output / "payment-meta.json"
+    try:
+        meta = json.loads(meta_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        meta = {}
+    with _lock:
+        job["status"] = "waiting_payment"
+        job["phase"] = "payment"
+        job["message"] = "付款码已生成，请扫码付款"
+        job["payment_qr_created_at"] = meta.get("qr_created_at", int(time.time()))
+    _monitor_payment(job_id)
+
+
+def _start_payment(job_id: str, selection: dict) -> None:
+    with _lock:
+        if _jobs[job_id].get("status") == "stopped":
            return
-        pay_command = [sys.executable, "scripts/jsdom-pay.py", "--session", str(session),
-                       "--mall-response", str(response), "--out-dir", str(output),
-                       "--zone-id", str(selection["zone_id"]),
-                       "--pf", str(selection.get("order_pf", "")),
-                       "--amount-fen", str(selection["price_fen"])]
-        _run_process(job_id, pay_command, "payment")
-    threading.Thread(target=run, daemon=True).start()
+        _jobs[job_id]["status"] = "ordering"
+        _jobs[job_id]["phase"] = "payment"
+        _jobs[job_id]["message"] = "正在创建商城订单"
+    threading.Thread(target=_payment_flow, args=(job_id, selection), daemon=True).start()


 def _stop_job(job_id: str) -> None:
    if job_id not in _jobs:
        raise ValueError("任务不存在")
-    pid = _jobs[job_id].get("process_pid")
+    job = _jobs[job_id]
+    pid = job.get("process_pid")
    if pid:
        try:
            os.kill(int(pid), 15)
        except ProcessLookupError:
            pass
    with _lock:
-        _jobs[job_id]["status"] = "failed"
-        _jobs[job_id]["phase"] = "stopped"
-        _jobs[job_id]["message"] = "任务已停止"
+        if job.get("status") == "waiting_payment":
+            job["status"] = "stopped"
+            job["phase"] = "payment"
+            job["message"] = "已放弃到账追踪；商城订单仍可能完成，请人工核对"
+        else:
+            job["status"] = "stopped"
+            job["phase"] = "stopped"
+            job["message"] = "任务已停止"


 def _public_job(job_id: str) -> dict:
@@ -206,6 +330,9 @@ def _public_job(job_id: str) -> dict:
                }
            except (OSError, json.JSONDecodeError):
                pass
+    for key in ("payment_qr_created_at", "payment_last_checked_at"):
+        if job.get(key) is not None:
+            result[key] = job[key]
    return result


@@ -289,13 +416,44 @@ class Handler(BaseHTTPRequestHandler):
                    "order_pf": selector.PLATFORMS[body["platform"]]["order_pf"],
                }
                _jobs[job_id]["phase"] = "payment"
-                return self._json(200, _public_job(job_id))
+                result = _public_job(job_id)
+                result["selection"] = _jobs[job_id]["selection"]
+                return self._json(200, result)
            if len(path) == 4 and path[:2] == ["v1", "jobs"] and path[3] == "payment":
                job_id = path[2]
-                if job_id not in _jobs or not _jobs[job_id].get("selection"):
+                if job_id not in _jobs:
+                    return self._json(400, {"detail": "任务不存在"})
+                job = _jobs[job_id]
+                with _lock:
+                    if job.get("status") == "stopped":
+                        return self._json(400, {"detail": "任务已停止，不能生成付款码"})
+                    if job.get("status") in {"ordering", "waiting_payment", "payment_timeout"} \
+                            or job.get("process_pid"):
+                        return self._json(400, {"detail": "已有进行中的支付流程，请勿重复操作"})
+                if not job.get("selection"):
                    return self._json(400, {"detail": "请先完成角色选择"})
-                _start_payment(job_id, _jobs[job_id]["selection"])
+                _start_payment(job_id, job["selection"])
                return self._json(202, _public_job(job_id))
+            if len(path) == 5 and path[:2] == ["v1", "jobs"] and path[3:5] == ["payment", "check"]:
+                job_id = path[2]
+                if job_id not in _jobs:
+                    return self._json(400, {"detail": "任务不存在"})
+                job = _jobs[job_id]
+                with _lock:
+                    if job.get("status") not in {"waiting_payment", "payment_timeout"}:
+                        return self._json(400, {"detail": "当前任务状态不支持检测到账"})
+                code = _check_payment_once(job_id)
+                with _lock:
+                    if code == 0:
+                        job["status"] = "success"
+                        job["phase"] = "completed"
+                        job["message"] = "已确认到账，充值完成"
+                    elif code == 1:
+                        # 保持 payment_timeout：检测由人工反复触发，避免无人监控的 waiting_payment。
+                        job["message"] = "仍未确认到账，请继续人工核对"
+                    else:
+                        job["message"] = "到账检测暂时失败，请稍后重试"
+                return self._json(200, _public_job(job_id))
            if len(path) == 4 and path[:2] == ["v1", "jobs"] and path[3] == "stop":
                _stop_job(path[2])
                return self._json(200, _public_job(path[2]))
@@ -317,6 +475,66 @@ class Handler(BaseHTTPRequestHandler):
        return


+def _restore_jobs(data_dir: Path) -> int:
+    """Rebuild in-memory jobs from task directories after a restart.
+
+    Only payment-phase jobs with a persisted QR baseline are restored so that
+    the payment QR stays visible and the read-only check can be re-run.
+    """
+    restored = 0
+    for directory in data_dir.iterdir():
+        if not directory.is_dir():
+            continue
+        job_id = directory.name
+        if job_id in _jobs:
+            continue
+        session = directory / "mall-session.json"
+        output = directory / "jsdom-order"
+        meta_path = output / "payment-meta.json"
+        if not session.exists():
+            continue
+        if not meta_path.exists():
+            _jobs[job_id] = {
+                "job_id": job_id, "status": "ready", "phase": "selection",
+                "logs": ["服务重启，已从任务目录恢复登录会话"],
+                "directory": str(directory),
+                "created_at": int(time.time()),
+                "message": "已恢复：登录会话有效，请选择充值信息",
+            }
+            restored += 1
+            continue
+        try:
+            meta = json.loads(meta_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        status = "waiting_payment"
+        phase = "payment"
+        message = "已恢复：付款码已生成，可重新检测到账"
+        last_checked = None
+        status_path = output / "payment-status.json"
+        if status_path.exists():
+            try:
+                record = json.loads(status_path.read_text(encoding="utf-8"))
+                last_checked = record.get("checked_at")
+                if record.get("matched_completion"):
+                    status = "success"
+                    phase = "completed"
+                    message = "已恢复：已确认到账"
+            except (OSError, json.JSONDecodeError):
+                pass
+        _jobs[job_id] = {
+            "job_id": job_id, "status": status, "phase": phase,
+            "logs": ["服务重启，已从任务目录恢复本任务"],
+            "directory": str(directory),
+            "created_at": int(time.time()),
+            "message": message,
+            "payment_qr_created_at": meta.get("qr_created_at"),
+            "payment_last_checked_at": last_checked,
+        }
+        restored += 1
+    return restored
+
+
 def main() -> int:
    global DEFAULT_DATA, WORKER_KEY
    parser = argparse.ArgumentParser(description="YYB admin worker")
@@ -325,12 +543,16 @@ def main() -> int:
    parser.add_argument("--data-dir", type=Path, default=DEFAULT_DATA)
    parser.add_argument("--key", default=None, help="HTTP Bearer 鉴权密钥；默认读取 YYB_WORKER_KEY")
    args = parser.parse_args()
-    DEFAULT_DATA = args.data_dir
+    # 强制绝对路径：子进程以 ROOT 为 cwd，相对路径会让任务文件写到错误位置。
+    DEFAULT_DATA = Path(args.data_dir).resolve()
    if args.key is not None:
        WORKER_KEY = args.key
    if args.host not in {"127.0.0.1", "localhost", "::1"} and not WORKER_KEY:
        parser.error("监听非本机地址时必须设置 --key 或 YYB_WORKER_KEY")
    DEFAULT_DATA.mkdir(parents=True, exist_ok=True)
+    restored = _restore_jobs(DEFAULT_DATA)
+    if restored:
+        print(f"已从任务目录恢复 {restored} 个支付任务", flush=True)
    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"YYB worker listening on {args.host}:{args.port}", flush=True)
    server.serve_forever()