"""端到端自动化：拿 ChatGPT 长链 → Stripe 选 PayPal → PayPal 注册绑卡 → 等短信。""" from __future__ import annotations import json import os import re import time import traceback from dataclasses import dataclass, field from typing import Callable, Optional from providers import fetch_us_address, fetch_visa_card, fetch_sms_code try: from curl_cffi import requests as curl_requests except Exception: curl_requests = None CHECKOUT_URL = "https://chatgpt.com/backend-api/payments/checkout" PAYURL_CHECKOUT_URL = "https://payurl.ark2.cn/api/checkout" PROXY_FOR_LONGLINK = "http://127.0.0.1:7890" PHONE_E164 = "+15822201173" PHONE_NUMBER = PHONE_E164.removeprefix("+1") PHONE_COUNTRY = "US" POST_PAYMENT_WAIT_TIMEOUT = 600 POST_PAYMENT_WAIT_INTERVAL = 2 POST_PAYMENT_MANUAL_INTERVAL = 5 POST_SMS_PAYPAL_ACTION_TEXTS = ( "Agree & Create Account", "Agree and Create Account", "Agree & Continue", "Agree and Continue", "同意并继续", ) POST_SMS_STRIPE_ACTION_TEXTS = ("Subscribe", "Pay", "Continue", "订阅", "訂閱") LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs") _PAYMENT_COMPLETE_RE = re.compile( r"(payment\s+(complete|successful)|purchase\s+complete|thanks?\s+for\s+(your\s+)?(payment|purchase|subscribing)|" r"thank\s+you|you['’]?re\s+all\s+set|subscription\s+(active|started|complete)|welcome\s+to\s+chatgpt\s+plus|" r"支付成功|付款成功|订阅成功|已完成)", re.I, ) _PAYMENT_BLOCKING_RE = re.compile( r"(validation|error|declined|failed|couldn['’]?t|cannot|try\s+again|invalid|" r"验证|错误|失败|无效|无法|重试)", re.I, ) _PAYMENT_FAILED_URL_RE = re.compile(r"redirect_status=failed|payment[_-]?status=failed", re.I) _PAYMENT_FAILED_TEXT_RE = re.compile( r"an\s+error\s+occurred\s+while\s+processing\s+your\s+payment|" r"payment\s+failed|" r"try\s+again\s+later\s+or\s+with\s+a\s+different\s+payment\s+method|" r"card\s+(?:was\s+)?declined|" r"payment\s+method\s+(?:is\s+)?not\s+supported|" r"we\s+weren['’]?t\s+able\s+to\s+add\s+this\s+card|" r"unable\s+to\s+add\s+this\s+card|" r"check\s+all\s+the\s+details\s+are\s+correct\s+and\s+try\s+again|" r"try\s+a\s+different\s+card|" r"sorry,?\s+something\s+went\s+wrong\.?\s*please\s+try\s+again|" r"we\s+couldn['’]?t\s+process\s+(?:your\s+)?(?:payment|request)|" r"we\s+can'?t\s+continue\s+with\s+this\s+payment|" r"this\s+card\s+has\s+already\s+been\s+added\s+to\s+another\s+paypal\s+account|" r"create_card_account_candidate_validation_error|" r"unmapped[_\s-]?oas[_\s-]?error|" r"oas[_\s-]?error|" r"cc[_\s-]?linked[_\s-]?to[_\s-]?full[_\s-]?account|" r"instrument[_\s-]?sharing[_\s-]?limit[_\s-]?exceeded|" r"支付失败|付款失败|银行卡(?:已)?(?:被)?拒绝|请稍后重试或使用其他支付方式|" r"无法添加(?:此|这张)?(?:银行)?卡|请检查(?:卡)?信息|请尝试其他(?:银行)?卡", re.I, ) class PayPalPaymentFailed(Exception): """PayPal/Stripe 明确返回"支付失败"，触发清缓存+重开长链重试。""" MAX_PAYPAL_RETRIES = 3 def _detect_payment_failure(page) -> str: """检测明确的"支付失败"信号。返回非空字符串表示已失败。""" url = getattr(page, "url", "") or "" if _PAYMENT_FAILED_URL_RE.search(url): return f"URL 标记 redirect_status=failed: {url}" snapshot = _payment_page_snapshot(page) text = snapshot.get("text", "") or "" alerts = snapshot.get("alerts", "") or "" blob = f"{text}\n{alerts}" m = _PAYMENT_FAILED_TEXT_RE.search(blob) if m: # 截上下文方便日志 start = max(0, (m.start() if hasattr(m, "start") else 0) - 60) end = min(len(blob), (m.end() if hasattr(m, "end") else 0) + 200) return f"页面失败提示：{blob[start:end].strip()[:400]}" return "" def _replace_browser_context(ctx, page): """彻底重新生成一个浏览器 context（PerimeterX 等基于 fingerprint 的反爬最有效的对策）。 注意：调用方持有的 page 引用不会被替换；这里只是把当前 context 关掉重开+加载空白页， 并复用同一个 browser。如果调用方传入的 page 已经死，会自然在下次 page.goto 时报错。 """ ctx.log("[playwright] 重建 browser context（清 PerimeterX 等指纹）") try: old_ctx = page.context browser = old_ctx.browser except Exception as exc: ctx.log(f"[playwright] 取 browser 失败: {exc!r}，回退普通清缓存") _clear_browser_state(ctx, page) return try: for p in list(old_ctx.pages): try: p.close() except Exception: pass old_ctx.close() ctx.log("[playwright] 旧 context 已关闭") except Exception as exc: ctx.log(f"[playwright] 关旧 context 失败: {exc!r}") # 上层调用者拿不到新 page；为不破坏接口，这里就让它在下次 goto 时报错走兜底 # 兜底：用 _clear_browser_state 当作降级 try: new_ctx = browser.new_context( locale="en-US", timezone_id="America/New_York", viewport={"width": 1280, "height": 900}, ) new_page = new_ctx.new_page() new_page.on("console", lambda m: ctx.log(f"[browser-console:{m.type}] {m.text[:300]}")) new_page.on("pageerror", lambda e: ctx.log(f"[browser-pageerror] {e}")) new_page.on("framenavigated", lambda f: ctx.log(f"[nav] {f.url}") if f == new_page.main_frame else None) new_page.on("requestfailed", lambda r: ctx.log(f"[req-failed] {r.method} {r.url} -> {r.failure}")) new_page.goto("about:blank", wait_until="domcontentloaded", timeout=10000) # 替换 ctx 上的 page 引用（chatgpt_flow 调 run_paypal_flow 时也是从 ctx._next_page 取） ctx._next_page = new_page # type: ignore ctx.log("[playwright] 新 context + 新 page 已就绪 (ctx._next_page)") except Exception as exc: ctx.log(f"[playwright] 新 context 创建失败: {exc!r}") def _clear_browser_state(ctx, page): """清掉当前 context 的 cookies / localStorage / sessionStorage / IndexedDB。""" ctx.log("[playwright] 清理浏览器缓存与 cookies（支付失败重试用）") try: bctx = page.context bctx.clear_cookies() ctx.log("[playwright] context.clear_cookies() 完成") except Exception as exc: ctx.log(f"[playwright] clear_cookies 失败: {exc!r}") try: bctx = page.context # 在 paypal/stripe/openai/chatgpt 各域都跑一遍清理 page.evaluate(r"""() => { try { localStorage.clear(); } catch (_) {} try { sessionStorage.clear(); } catch (_) {} try { if (window.indexedDB && indexedDB.databases) { indexedDB.databases().then((dbs) => { (dbs || []).forEach((db) => { if (db && db.name) { try { indexedDB.deleteDatabase(db.name); } catch (_) {} } }); }); } } catch (_) {} try { if (window.caches && caches.keys) { caches.keys().then((keys) => keys.forEach((k) => caches.delete(k))); } } catch (_) {} }""") ctx.log("[playwright] localStorage/sessionStorage/IndexedDB/caches 清理完成（当前页）") except Exception as exc: ctx.log(f"[playwright] 当前页 storage 清理失败: {exc!r}") # 关掉所有 PayPal/Stripe 旧标签，最后回到一个空白页 try: for p in list(page.context.pages): try: if p is page: continue p.close() except Exception: pass page.goto("about:blank", wait_until="domcontentloaded", timeout=15000) except Exception as exc: ctx.log(f"[playwright] 关旧 tab/转空白页 失败: {exc!r}") @dataclass class RunContext: token: str plan: str = "plus" country: str = "US" currency: str = "USD" use_promo: bool = True headless: bool = False log: Callable[[str], None] = print state: str = "running" stage: str = "" on_stage: Optional[Callable[[str], None]] = None long_link: str = "" email: str = "" # ChatGPT 注册邮箱（@edu.a4sky.com） paypal_email: str = "" # PayPal 字段用的独立邮箱（@gmail.com，每次重试都换） password: str = "" card: dict = field(default_factory=dict) address: dict = field(default_factory=dict) run_id: str = field(default_factory=lambda: time.strftime("%Y%m%d-%H%M%S")) @property def artifact_dir(self) -> str: d = os.path.join(LOG_DIR, self.run_id) os.makedirs(d, exist_ok=True) return d def set_stage(self, name: str): self.stage = name self.log(f"[stage] {name}") if self.on_stage: try: self.on_stage(name) except Exception: pass def _request_headers(token: str) -> dict: return { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", "Origin": "https://chatgpt.com", "Referer": "https://chatgpt.com/", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/136.0.0.0 Safari/537.36" ), } def _checkout_payload(plan: str, country: str, currency: str, use_promo: bool) -> dict: payload = { "plan_name": "chatgptplusplan" if plan == "plus" else "chatgptteamplan", "billing_details": {"country": country.upper(), "currency": currency.upper()}, "checkout_ui_mode": "hosted", "cancel_url": "https://chatgpt.com/#pricing", } if use_promo and plan == "plus": payload["promo_campaign"] = { "promo_campaign_id": "plus-1-month-free", "is_coupon_from_query_param": True, } return payload def generate_long_link(ctx: RunContext) -> str: ctx.log(f"[longlink] === 开始生成长链 ===") ctx.log(f"[longlink] plan={ctx.plan} country={ctx.country} currency={ctx.currency} use_promo={ctx.use_promo}") ctx.log(f"[longlink] 代理={PROXY_FOR_LONGLINK}") payload = _checkout_payload(ctx.plan, ctx.country, ctx.currency, ctx.use_promo) ctx.log(f"[longlink] payload={json.dumps(payload, ensure_ascii=False)}") if curl_requests is None: raise RuntimeError("缺少 curl_cffi，请先 pip install curl_cffi") proxies = {"http": PROXY_FOR_LONGLINK, "https": PROXY_FOR_LONGLINK} started = time.time() try: response = curl_requests.post( CHECKOUT_URL, json=payload, headers=_request_headers(ctx.token), impersonate="chrome136", proxies=proxies, timeout=30, ) except Exception as exc: ctx.log(f"[longlink] 请求异常（{int((time.time()-started)*1000)}ms）: {exc!r}") raise text = response.text ctx.log(f"[longlink] HTTP {response.status_code} 耗时 {int((time.time()-started)*1000)}ms 返回长度={len(text)}") ctx.log(f"[longlink] 响应预览: {text[:500]}") if response.status_code >= 400: raise RuntimeError(f"创建 checkout 失败: HTTP {response.status_code} {text[:300]}") try: data = json.loads(text or "{}") except json.JSONDecodeError as exc: raise RuntimeError(f"长链响应不是 JSON: {exc!r} 原文={text[:300]}") link = data.get("url") or data.get("stripe_hosted_url") or data.get("checkout_url") session_id = data.get("checkout_session_id") processor = data.get("processor_entity") ctx.log(f"[longlink] checkout_session_id={session_id} processor={processor}") if not link: raise RuntimeError(f"未在响应中解析到长链: {text[:300]}") ctx.log(f"[longlink] 成功取到 long_link={link}") return link def generate_long_link_payurl(ctx: RunContext) -> str: """走 payurl.ark2.cn（与 Chrome 扩展 get-plus-link.js 一致），不走本地代理。""" ctx.log("[longlink] === 通过 payurl.ark2.cn 获取 Plus 长链 ===") ctx.log(f"[longlink] plan={ctx.plan} country={ctx.country} currency={ctx.currency} use_promo={ctx.use_promo}") payload = { "token": ctx.token, "plan": ctx.plan or "plus", "checkout_ui_mode": "hosted", "ui_language": "en", "country": (ctx.country or "US").upper(), "currency": (ctx.currency or "USD").upper(), "proxy": "", "use_promo": bool(ctx.use_promo), "promo_code": "STRIPEATLASGPT4BIZ050126", "workspace_name": "linux-do", "seat_quantity": 2, } headers = { "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.9", "Content-Type": "application/json", "DNT": "1", "Origin": "https://payurl.ark2.cn", "Referer": "https://payurl.ark2.cn/", } masked = dict(payload) masked["token"] = (ctx.token or "")[:24] + "..." ctx.log(f"[longlink] payload(masked)={json.dumps(masked, ensure_ascii=False)}") max_attempts = 5 last_err = "" for attempt in range(1, max_attempts + 1): ctx.log(f"[longlink] 第 {attempt}/{max_attempts} 次请求 {PAYURL_CHECKOUT_URL}") started = time.time() try: if curl_requests is not None: r = curl_requests.post( PAYURL_CHECKOUT_URL, json=payload, headers=headers, impersonate="chrome136", timeout=30, ) text = r.text status = r.status_code else: import urllib.request data = json.dumps(payload).encode("utf-8") req = urllib.request.Request(PAYURL_CHECKOUT_URL, data=data, method="POST") for k, v in headers.items(): req.add_header(k, v) with urllib.request.urlopen(req, timeout=30) as resp: text = resp.read().decode("utf-8", errors="replace") status = resp.status except Exception as exc: last_err = repr(exc) ctx.log(f"[longlink] 第 {attempt} 次异常 ({int((time.time()-started)*1000)}ms): {last_err}") time.sleep(1.5) continue ctx.log(f"[longlink] HTTP {status} 耗时 {int((time.time()-started)*1000)}ms 长度={len(text)} 预览={text[:300]}") if status >= 400: last_err = f"HTTP {status}: {text[:300]}" time.sleep(1.5) continue try: data = json.loads(text or "{}") except Exception as exc: last_err = f"非 JSON: {exc!r}" time.sleep(1.5) continue link = data.get("url") or data.get("openai_payurl") or data.get("chatgpt_checkout_url") if link: ctx.log(f"[longlink] 成功取到 long_link={link} sessionId={data.get('checkout_session_id', '?')}") return link last_err = f"响应缺 url 字段: {text[:300]}" time.sleep(1.5) raise RuntimeError(f"payurl.ark2.cn 长链获取连续失败 {max_attempts} 次：{last_err}") def _rand_email() -> str: import random import string name = "".join(random.choice(string.ascii_lowercase + string.digits) for _ in range(16)) return f"{name}@gmail.com" def _rand_paypal_email() -> str: """PayPal 注册用的独立邮箱（与 ChatGPT 注册邮箱解耦）。 用 gmail 域，一来 PayPal 对 gmail 的接受度高，二来不与 a4sky 邮件助手挂钩。 """ import random import string # 8-12 位随机字母数字 + 点号风格更像真人 body_len = random.randint(8, 12) body = "".join(random.choice(string.ascii_lowercase + string.digits) for _ in range(body_len)) # 加一个随机数字尾巴 tail = "".join(random.choice(string.digits) for _ in range(random.randint(2, 4))) return f"{body}{tail}@gmail.com" def _rand_password() -> str: import random import string pools = [ random.choice(string.ascii_uppercase), random.choice(string.ascii_lowercase), random.choice(string.digits), random.choice("!@#$%^"), ] pools += [random.choice(string.ascii_letters + string.digits + "!@#$%^") for _ in range(10)] random.shuffle(pools) return "".join(pools) def _check_stop(ctx: RunContext): if ctx.state == "stopped": raise RuntimeError("STOPPED_BY_USER") def _dump_page(ctx: RunContext, page, tag: str): """失败/检查点时把当前页 URL/截图/HTML 都落盘。""" try: url = page.url except Exception: url = "" ctx.log(f"[dump:{tag}] URL={url}") try: title = page.title() ctx.log(f"[dump:{tag}] title={title!r}") except Exception as exc: ctx.log(f"[dump:{tag}] 取标题失败: {exc!r}") base = os.path.join(ctx.artifact_dir, f"{int(time.time()*1000)}-{tag}") try: page.screenshot(path=base + ".png", full_page=True) ctx.log(f"[dump:{tag}] 截图 -> {base}.png") except Exception as exc: ctx.log(f"[dump:{tag}] 截图失败: {exc!r}") try: html = page.content() with open(base + ".html", "w", encoding="utf-8") as f: f.write(html) ctx.log(f"[dump:{tag}] HTML -> {base}.html ({len(html)} bytes)") except Exception as exc: ctx.log(f"[dump:{tag}] HTML 落盘失败: {exc!r}") def run_paypal_flow(ctx: RunContext, page=None): """跑 Stripe → PayPal。这一段不走代理。 page 不为空时复用外部浏览器（避免嵌套 sync_playwright），否则自己开。 支付失败会清缓存+重开长链最多重试 MAX_PAYPAL_RETRIES 次。 """ if page is None: return _run_paypal_flow_self_browser(ctx) ctx.log("[playwright] === 准备外部资源（地址/卡/邮箱/密码） ===") ctx.set_stage("准备地址/卡/账号") ctx.address = fetch_us_address(log=ctx.log) ctx.card = fetch_visa_card(log=ctx.log) reuse = bool(getattr(ctx, "_reuse_account_for_paypal", False)) if reuse and ctx.email and ctx.password: ctx.log(f"[playwright] 复用已有账号 email={ctx.email}（来自注册阶段）") else: ctx.email = _rand_email() ctx.password = _rand_password() ctx.log(f"[playwright] email={ctx.email}") ctx.log(f"[playwright] password={ctx.password}") ctx.log(f"[playwright] card={{number=***{ctx.card['number'][-4:]}, expiry={ctx.card['expiry']}, cvv={ctx.card['cvv']}}}") ctx.log(f"[playwright] address={ctx.address}") ctx.log(f"[playwright] phone={PHONE_E164}") ctx.log("[playwright] 复用外部浏览器 page 跑 Stripe/PayPal") last_err: Exception | None = None for attempt in range(1, MAX_PAYPAL_RETRIES + 1): # PayPal 字段用独立的 gmail 邮箱，每次重试都换（避开 PayPal 把上次失败的邮箱标黑） ctx.paypal_email = _rand_paypal_email() ctx.log(f"[playwright] PayPal 用邮箱={ctx.paypal_email}（与 ChatGPT 注册邮箱 {ctx.email} 解耦）") try: ctx.set_stage(f"支付尝试 {attempt}/{MAX_PAYPAL_RETRIES}") ctx.log(f"[playwright] 第 {attempt}/{MAX_PAYPAL_RETRIES} 次尝试，打开 Stripe 长链: {ctx.long_link}") page.goto(ctx.long_link, wait_until="domcontentloaded", timeout=60000) _dump_page(ctx, page, f"01-stripe-loaded-attempt{attempt}") _stripe_select_paypal_and_submit(ctx, page) _paypal_signup_and_pay(ctx, page) ctx.log("[playwright] 流程完成") _dump_page(ctx, page, "99-final") return except PayPalPaymentFailed as exc: last_err = exc ctx.log(f"[playwright] 第 {attempt} 次支付失败: {exc}") if attempt >= MAX_PAYPAL_RETRIES: ctx.log(f"[playwright] 已达最大重试次数 {MAX_PAYPAL_RETRIES}，放弃") raise ctx.set_stage(f"支付失败，准备第 {attempt + 1} 次重试（清缓存+换 context）") try: _replace_browser_context(ctx, page) # _replace_browser_context 返回新的 page，但因为这里的 page 是参数无法重新赋值 # 改为修改 ctx 上的引用，让上层重新拿 # 但上层 chatgpt_flow 是直接传 page 进来的，没法接到新 page # 简单粗暴：在原 context 上彻底清干净 except Exception as exc2: ctx.log(f"[playwright] 换 context 异常，回退到清 cookie: {exc2!r}") try: _clear_browser_state(ctx, page) except Exception as exc3: ctx.log(f"[playwright] 清缓存异常（继续）: {exc3!r}") try: ctx.address = fetch_us_address(log=ctx.log) ctx.card = fetch_visa_card(log=ctx.log) ctx.log(f"[playwright] 重试用新地址={ctx.address} 新卡尾号=***{ctx.card['number'][-4:]}") except Exception as exc2: ctx.log(f"[playwright] 重新拉取地址/卡片异常（继续）: {exc2!r}") # 让 PayPal 反爬规则稍微衰减 ctx.log("[playwright] 重试前等待 30s（让 PayPal/PerimeterX 指纹/速率衰减）") time.sleep(30) except Exception as exc: ctx.log(f"[playwright] 流程异常: {exc!r}") ctx.log(traceback.format_exc()) try: _dump_page(ctx, page, f"99-error-attempt{attempt}") except Exception: pass raise if last_err is not None: raise last_err def _run_paypal_flow_self_browser(ctx: RunContext): """旧入口：自己开浏览器（保留兼容性）。""" from playwright.sync_api import sync_playwright ctx.log("[playwright] === 准备外部资源（地址/卡/邮箱/密码） ===") ctx.set_stage("准备地址/卡/账号") ctx.address = fetch_us_address(log=ctx.log) ctx.card = fetch_visa_card(log=ctx.log) reuse = bool(getattr(ctx, "_reuse_account_for_paypal", False)) if reuse and ctx.email and ctx.password: ctx.log(f"[playwright] 复用已有账号 email={ctx.email}（来自注册阶段）") else: ctx.email = _rand_email() ctx.password = _rand_password() ctx.log(f"[playwright] email={ctx.email}") ctx.log(f"[playwright] password={ctx.password}") ctx.log(f"[playwright] card={{number=***{ctx.card['number'][-4:]}, expiry={ctx.card['expiry']}, cvv={ctx.card['cvv']}}}") ctx.log(f"[playwright] address={ctx.address}") ctx.log(f"[playwright] phone={PHONE_E164}") with sync_playwright() as p: ctx.log(f"[playwright] 启动 Chromium headless={ctx.headless}") browser = p.chromium.launch( headless=ctx.headless, args=["--disable-blink-features=AutomationControlled"], ) context = browser.new_context( locale="en-US", timezone_id="America/New_York", viewport={"width": 1280, "height": 900}, ) new_page = context.new_page() new_page.on("console", lambda m: ctx.log(f"[browser-console:{m.type}] {m.text[:300]}")) new_page.on("pageerror", lambda e: ctx.log(f"[browser-pageerror] {e}")) new_page.on("framenavigated", lambda f: ctx.log(f"[nav] {f.url}") if f == new_page.main_frame else None) new_page.on("requestfailed", lambda r: ctx.log(f"[req-failed] {r.method} {r.url} -> {r.failure}")) try: for attempt in range(1, MAX_PAYPAL_RETRIES + 1): ctx.paypal_email = _rand_paypal_email() ctx.log(f"[playwright] PayPal 用邮箱={ctx.paypal_email}") try: ctx.log(f"[playwright] 第 {attempt}/{MAX_PAYPAL_RETRIES} 次尝试，打开 Stripe 长链: {ctx.long_link}") new_page.goto(ctx.long_link, wait_until="domcontentloaded", timeout=60000) _dump_page(ctx, new_page, f"01-stripe-loaded-attempt{attempt}") _stripe_select_paypal_and_submit(ctx, new_page) _paypal_signup_and_pay(ctx, new_page) ctx.log("[playwright] 流程完成") _dump_page(ctx, new_page, "99-final") return except PayPalPaymentFailed as exc: ctx.log(f"[playwright] 第 {attempt} 次支付失败: {exc}") if attempt >= MAX_PAYPAL_RETRIES: raise _clear_browser_state(ctx, new_page) try: ctx.address = fetch_us_address(log=ctx.log) ctx.card = fetch_visa_card(log=ctx.log) except Exception as exc2: ctx.log(f"[playwright] 重抽地址/卡异常（继续）: {exc2!r}") ctx.log("[playwright] 重试前等待 30s") time.sleep(30) except Exception as exc: ctx.log(f"[playwright] 流程异常: {exc!r}") ctx.log(traceback.format_exc()) try: _dump_page(ctx, new_page, "99-error") except Exception: pass raise finally: try: browser.close() except Exception: pass def _stripe_select_paypal(ctx, page) -> bool: """多策略尝试选中 PayPal 支付方式。命中返回 True。""" # 1) 先等待至少一种 PayPal 候选出现，避免 timing 问题 try: page.wait_for_function( r"""() => { const sels = [ '#payment-method-accordion-item-title-paypal', 'input[type="radio"][value="paypal"]', 'input[name="paymentMethod"][value="paypal"]', '[data-testid*="paypal" i]', 'label[for*="paypal" i]', 'button[id*="paypal" i]', ]; for (const s of sels) { if (document.querySelector(s)) return true; } const labels = Array.from(document.querySelectorAll('label, button, [role="radio"], [role="button"]')); return labels.some((el) => /paypal/i.test((el.innerText || el.getAttribute('aria-label') || ''))); }""", timeout=15000, ) except Exception as exc: ctx.log(f"[stripe] 等待 PayPal 候选出现超时: {exc!r}") # 2) 候选 selector 列表，按可靠度排序 candidates = [ '#payment-method-accordion-item-title-paypal', 'input[type="radio"][value="paypal"]', 'input[name="paymentMethod"][value="paypal"]', '[data-testid="payment-method-paypal"]', '[data-testid*="paypal" i]', 'label[for*="paypal" i]', 'button[id*="paypal" i]', '[role="radio"][aria-label*="paypal" i]', ] for sel in candidates: loc = page.locator(sel).first if loc.count() == 0: continue ctx.log(f"[stripe] PayPal 候选命中 {sel}") # 多种点法兜底 for action in ("check", "click", "force_click", "js_click"): try: if action == "check": loc.check(force=True, timeout=4000) elif action == "click": loc.click(timeout=4000) elif action == "force_click": loc.click(timeout=4000, force=True) else: loc.evaluate("el => { el.click(); }") ctx.log(f"[stripe] {action}({sel}) 成功") page.wait_for_timeout(800) if _stripe_paypal_is_selected(page): return True except Exception as exc: ctx.log(f"[stripe] {action}({sel}) 失败: {exc!r}") continue # 没确认选中也继续下一种 if _stripe_paypal_is_selected(page): return True # 3) 文本扫描：找包含 "PayPal" 文本的可点击元素，逐个尝试 ctx.log("[stripe] 进入文本扫描兜底") try: n = page.evaluate( r"""() => { const all = Array.from(document.querySelectorAll( 'label, button, [role="radio"], [role="button"], [role="tab"], div[tabindex], a' )); const matches = all.filter((el) => { const t = (el.innerText || el.getAttribute('aria-label') || '').trim(); return /paypal/i.test(t); }); window.__paypalCandidates = matches; return matches.length; }""" ) ctx.log(f"[stripe] 文本扫描候选数={n}") for i in range(int(n or 0)): try: page.evaluate( r"""(i) => { const el = (window.__paypalCandidates || [])[i]; if (el) { el.scrollIntoView({block:'center'}); el.click(); } }""", i, ) page.wait_for_timeout(700) if _stripe_paypal_is_selected(page): ctx.log(f"[stripe] 文本扫描候选 idx={i} 命中") return True except Exception as exc: ctx.log(f"[stripe] 文本扫描候选 idx={i} 失败: {exc!r}") except Exception as exc: ctx.log(f"[stripe] 文本扫描兜底异常: {exc!r}") return False def _stripe_paypal_is_selected(page) -> bool: """通过多种特征判断 PayPal 已被选中。""" try: return bool(page.evaluate( r"""() => { // 1) 单选框 checked const r = document.querySelector('input[type="radio"][value="paypal"]:checked') || document.querySelector('input[name="paymentMethod"][value="paypal"]:checked'); if (r) return true; // 2) accordion item with aria-selected/expanded=true 含 paypal const items = Array.from(document.querySelectorAll('[role="radio"], [role="tab"], button, label, div')); const sel = items.find((el) => { const t = (el.innerText || el.getAttribute('aria-label') || '').trim(); if (!/paypal/i.test(t)) return false; const checked = el.getAttribute('aria-checked') === 'true' || el.getAttribute('aria-selected') === 'true' || el.getAttribute('aria-expanded') === 'true' || /selected|active|checked/i.test(el.className || ''); return checked; }); if (sel) return true; // 3) 提交按钮文本包含 PayPal（Stripe 选 PayPal 后按钮会变 "Pay with PayPal"） const btn = document.querySelector('button[data-testid="hosted-payment-submit-button"]'); if (btn && /paypal/i.test(btn.innerText || '')) return true; return false; }""" )) except Exception: return False def _stripe_select_paypal_and_submit(ctx, page): ctx.log("[stripe] === 在 Stripe 选择 PayPal 并提交 ===") page.wait_for_timeout(2000) _check_stop(ctx) selected = _stripe_select_paypal(ctx, page) if not selected: _dump_page(ctx, page, "02-stripe-paypal-not-found") raise RuntimeError("无法在 Stripe 上选中 PayPal 支付方式") ctx.log("[stripe] PayPal 已选中") page.wait_for_timeout(1500) ctx.log("[stripe] 切换 billingCountry=US") _select_by_text(ctx, page, '#billingCountry', 'US', "billingCountry") page.wait_for_timeout(500) ctx.log("[stripe] 填写账单地址") addr = ctx.address _safe_fill(ctx, page, '#billingAddressLine1', addr["street"], "billingAddressLine1") page.keyboard.press('Escape') page.wait_for_timeout(200) _safe_fill(ctx, page, '#billingLocality', addr["city"], "billingLocality") _safe_fill(ctx, page, '#billingPostalCode', addr["zip"], "billingPostalCode") _select_by_text(ctx, page, '#billingAdministrativeArea', addr["state"], "billingAdministrativeArea") if page.locator('#phoneNumber').count() > 0: _safe_fill(ctx, page, '#phoneNumber', PHONE_NUMBER, "phoneNumber") page.wait_for_timeout(400) cb = page.locator('#termsOfServiceConsentCheckbox') if cb.count() > 0: try: checked = cb.is_checked() ctx.log(f"[stripe] 协议复选框 checked={checked}") if not checked: cb.check(force=True, timeout=2000) ctx.log("[stripe] 已勾选协议") except Exception as exc: ctx.log(f"[stripe] 协议复选框处理失败: {exc!r}") else: ctx.log("[stripe] 未发现协议复选框") _dump_page(ctx, page, "03-stripe-filled") submit = page.locator('button[data-testid="hosted-payment-submit-button"]').first if submit.count() == 0: ctx.log("[stripe] 找不到 hosted-payment-submit-button") _dump_page(ctx, page, "04-stripe-no-submit") raise RuntimeError("hosted-payment-submit-button 不存在") klass = submit.get_attribute("class") or "" ctx.log(f"[stripe] 提交按钮 class={klass}") if "incomplete" in klass: ctx.log("[stripe] 警告：按钮仍处于 incomplete 状态，将再等 2s 重试") page.wait_for_timeout(2000) klass = submit.get_attribute("class") or "" ctx.log(f"[stripe] 再次检查 class={klass}") submit.click(timeout=8000) ctx.log("[stripe] 已点击提交，等待跳转 PayPal") try: page.wait_for_url(re.compile(r"paypal\.com"), timeout=60000) except Exception as exc: ctx.log(f"[stripe] 等待跳转到 PayPal 超时: {exc!r}") _dump_page(ctx, page, "04-stripe-no-redirect") raise ctx.log(f"[stripe] 已跳转: {page.url}") _dump_page(ctx, page, "05-paypal-arrived") def _paypal_signup_and_pay(ctx, page): ctx.log("[paypal] === 进入 PayPal 流程 ===") page.wait_for_load_state("domcontentloaded", timeout=30000) page.wait_for_timeout(1500) _check_stop(ctx) _paypal_clear_session(ctx, page) # 强制走"创建账号"路径，避免点上方 Next 被识别为已存在账号要求输密码 from paypal_flow import ensure_checkoutweb on_stage = getattr(ctx, "on_stage", None) path_taken = ensure_checkoutweb( page, fallback_email=ctx.paypal_email or ctx.email, log=ctx.log, on_stage=on_stage, ) ctx.log(f"[paypal] checkoutweb 进入路径: {path_taken}") _dump_page(ctx, page, "06-paypal-checkoutweb-entered") ctx.log("[paypal] 等待关键字段挂载（cardNumber/phone/billingLine1）") for sel in ('#cardNumber', '#phone', '#billingLine1'): try: page.locator(sel).first.wait_for(state="visible", timeout=20000) ctx.log(f"[paypal] {sel} 已可见") except Exception as exc: ctx.log(f"[paypal] 等待 {sel} 超时: {exc!r}") _dump_page(ctx, page, "07-paypal-checkoutweb") ctx.log("[paypal] 检查国家选择器") country = page.locator('#country') if country.count() > 0: try: current = country.input_value() ctx.log(f"[paypal] 当前国家={current}") if current != "US": ctx.log("[paypal] 切换国家为 US") country.select_option("US") page.wait_for_timeout(2500) ctx.log(f"[paypal] 切换后国家={country.input_value()}, URL={page.url}") _dump_page(ctx, page, "08-paypal-country-switched") except Exception as exc: ctx.log(f"[paypal] 切换国家失败: {exc!r}") else: ctx.log("[paypal] 未发现 #country 选择器") addr = ctx.address card = ctx.card ctx.log("[paypal] 第一轮填表") _paypal_fill_form(ctx, page, addr, card) page.wait_for_timeout(800) invalid = _paypal_collect_invalid(page) if invalid: ctx.log(f"[paypal] 第一轮后仍有 aria-invalid='true' 的字段: {invalid}，再补一次") _paypal_fill_form(ctx, page, addr, card, retry_only=invalid) page.wait_for_timeout(600) _dump_page(ctx, page, "09-paypal-filled") ctx.log("[paypal] 提交付款") if not _click_next(ctx, page, "PayPal 提交"): ctx.log("[paypal] 警告：未能找到可点击的提交按钮") _dump_page(ctx, page, "10-paypal-no-submit") page.wait_for_timeout(2500) invalid_after = _paypal_collect_invalid(page) if invalid_after: ctx.log(f"[paypal] 提交后页面报错的字段: {invalid_after}，再补再提交一次") _paypal_fill_form(ctx, page, addr, card, retry_only=invalid_after) page.wait_for_timeout(500) _click_next(ctx, page, "PayPal 提交-2") _paypal_handle_sms(ctx, page) _PAYPAL_FIELDS = [ ("email", "value"), ("phone", "value"), ("cardNumber", "card-number"), ("cardExpiry", "card-expiry"), ("cardCvv", "card-cvv"), ("firstName", "value"), ("lastName", "value"), ("billingLine1", "value"), ("billingCity", "value"), ("billingPostalCode", "value"), ("password", "value"), ] def _paypal_fill_form(ctx, page, addr, card, retry_only=None): targets = retry_only or [k for k, _ in _PAYPAL_FIELDS] if "email" in targets: _safe_fill(ctx, page, '#email', ctx.paypal_email or ctx.email, "email") if "phone" in targets: _paypal_type_into(ctx, page, '#phone', PHONE_NUMBER, "phone") if "cardNumber" in targets: _paypal_type_into(ctx, page, '#cardNumber', card["number"], "cardNumber", mask=True) if "cardExpiry" in targets: _paypal_type_into(ctx, page, '#cardExpiry', card["expiry"], "cardExpiry") if "cardCvv" in targets: _paypal_type_into(ctx, page, '#cardCvv', card["cvv"], "cardCvv") if "firstName" in targets: _safe_fill(ctx, page, '#firstName', "James", "firstName") if "lastName" in targets: _safe_fill(ctx, page, '#lastName', "Smith", "lastName") if "billingLine1" in targets: _safe_fill(ctx, page, '#billingLine1', addr["street"], "billingLine1") try: page.keyboard.press('Escape') except Exception: pass if "billingCity" in targets: _safe_fill(ctx, page, '#billingCity', addr["city"], "billingCity") if "billingPostalCode" in targets: _safe_fill(ctx, page, '#billingPostalCode', addr["zip"], "billingPostalCode") if any(k in targets for k in ("billingLine1", "billingCity", "billingPostalCode")): _select_by_text(ctx, page, '#billingState', addr["state"], "billingState") if "password" in targets: _safe_fill(ctx, page, '#password', ctx.password, "password", mask=True) def _paypal_type_into(ctx, page, selector: str, value: str, label: str = "", mask: bool = False): label = label or selector shown = "***" if mask else value try: loc = page.locator(selector).first if loc.count() == 0: ctx.log(f"[type] {label}({selector}) 不存在，跳过") return loc.wait_for(state="visible", timeout=8000) loc.click() loc.fill("") loc.type(value, delay=20) ctx.log(f"[type] {label}({selector}) <- {shown}") except Exception as exc: ctx.log(f"[type] {label}({selector}) 失败: {exc!r}") def _paypal_collect_invalid(page): try: return page.evaluate("""() => { const out = []; document.querySelectorAll('[aria-invalid="true"]').forEach(el => { if (el.id) out.push(el.id); else if (el.name) out.push(el.name); }); return out; }""") except Exception: return [] def _paypal_clear_session(ctx, page): ctx.log("[paypal] 清理 cookie/storage") try: before = page.evaluate("() => document.cookie.split(';').filter(Boolean).length") page.evaluate("""() => { try { localStorage.clear(); } catch (e) {} try { sessionStorage.clear(); } catch (e) {} const host = location.hostname; const parts = host.split('.'); const domains = [host, '.' + host]; for (let i = 1; i < parts.length - 1; i++) domains.push('.' + parts.slice(i).join('.')); const cookies = document.cookie ? document.cookie.split(';') : []; cookies.forEach(c => { const name = c.split('=')[0].trim(); if (!name) return; ['/', location.pathname].forEach(p => { domains.forEach(d => { document.cookie = name + '=; expires=Thu, 01 Jan 1970 00:00:00 GMT; path=' + p + '; domain=' + d; }); document.cookie = name + '=; expires=Thu, 01 Jan 1970 00:00:00 GMT; path=' + p; }); }); }""") after = page.evaluate("() => document.cookie.split(';').filter(Boolean).length") ctx.log(f"[paypal] cookie 数量 {before} -> {after}（httpOnly 项 JS 删不掉）") except Exception as exc: ctx.log(f"[paypal] 清理会话异常: {exc!r}") def _paypal_remove_captcha_overlay(ctx, page): try: removed = page.evaluate("""() => { const ids = ['captchaComponent', 'recaptcha', 'recaptcha-overlay', 'reCAPTCHAOverlay']; let n = 0; for (const id of ids) { const el = document.getElementById(id); if (el) { el.remove(); n++; } } document.querySelectorAll( 'iframe[name="recaptcha"], iframe[title*="recaptcha" i], div[class*="captcha" i][class*="overlay" i]' ).forEach(el => { el.remove(); n++; }); return n; }""") if removed: ctx.log(f"[paypal] 已删除 captcha 相关蒙层节点 {removed} 个") except Exception as exc: ctx.log(f"[paypal] 删除 captcha 蒙层失败: {exc!r}") def _paypal_handle_sms(ctx, page): ctx.log("[paypal] === 检查是否需要短信验证码 ===") page.wait_for_timeout(3000) _check_stop(ctx) _paypal_remove_captcha_overlay(ctx, page) digit_inputs = page.locator('input[id^="ci-ciBasic-"]') digit_count = 0 try: digit_count = digit_inputs.count() except Exception as exc: ctx.log(f"[paypal] 探测分立 OTP 输入框失败: {exc!r}") ctx.log(f"[paypal] 分立 OTP 输入框 ci-ciBasic-* count={digit_count}") single_input = None if digit_count == 0: sms_inputs = [ 'input[name="otp"]', 'input[autocomplete="one-time-code"]', 'input#otp', 'input[name="smsOtp"]', 'input[id*="otp" i]', ] for sel in sms_inputs: loc = page.locator(sel).first try: count = loc.count() ctx.log(f"[paypal] 探测 SMS 输入框 selector={sel} count={count}") if count == 0: continue loc.wait_for(state="visible", timeout=4000) single_input = (loc, sel) ctx.log(f"[paypal] 命中验证码输入框: {sel}") break except Exception as exc: ctx.log(f"[paypal] selector={sel} 等待可见失败: {exc!r}") continue if digit_count == 0 and single_input is None: ctx.log("[paypal] 未检测到验证码输入框，跳过 SMS 步骤") _dump_page(ctx, page, "11-paypal-no-sms") _paypal_wait_for_payment_completion(ctx, page) return _dump_page(ctx, page, "12-paypal-sms-prompt") ctx.log("[paypal] 开始等待短信验证码（最长 180s）") code = fetch_sms_code(timeout=180, interval=5, log=ctx.log) if digit_count > 0: ctx.log(f"[paypal] 逐位填入 OTP，长度={len(code)} 输入框={digit_count}") if len(code) < digit_count: ctx.log(f"[paypal] 警告：验证码位数 {len(code)} < 输入框数 {digit_count}") try: first = digit_inputs.nth(0) first.wait_for(state="visible", timeout=4000) first.click() except Exception as exc: ctx.log(f"[paypal] 聚焦第 1 格失败: {exc!r}") for idx in range(min(digit_count, len(code))): try: box = digit_inputs.nth(idx) box.fill("") box.type(code[idx], delay=30) ctx.log(f"[paypal] OTP[{idx}] <- {code[idx]}") except Exception as exc: ctx.log(f"[paypal] OTP[{idx}] 填入失败: {exc!r}") else: loc, sel = single_input loc.fill(code) ctx.log(f"[paypal] 已填入验证码到 {sel}") page.wait_for_timeout(800) _click_next(ctx, page, "SMS 提交") ctx.log("[paypal] 已提交验证码") _dump_page(ctx, page, "13-paypal-sms-submitted") _paypal_wait_for_payment_completion(ctx, page) def _paypal_wait_for_payment_completion( ctx, page, timeout: int = POST_PAYMENT_WAIT_TIMEOUT, interval: int = POST_PAYMENT_WAIT_INTERVAL, ) -> bool: ctx.log(f"[paypal] 等待支付完成，自动等待 {timeout}s，间隔 {interval}s") deadline = time.time() + timeout last_status = "" manual_mode = False clicked_actions = set() while True: _check_stop(ctx) if _page_is_closed(page): ctx.log("[paypal] 页面已被关闭，停止等待支付完成") return False completion_reason = _detect_payment_completion(page) if completion_reason: ctx.log(f"[paypal] 支付完成确认: {completion_reason}") _dump_page(ctx, page, "14-paypal-payment-complete") return True failure_reason = _detect_payment_failure(page) if failure_reason: ctx.log(f"[paypal] 支付失败信号: {failure_reason}") _dump_page(ctx, page, "14-paypal-payment-failed") raise PayPalPaymentFailed(failure_reason) if _click_post_sms_action_if_available(ctx, page, clicked_actions): last_status = "" page.wait_for_timeout(2500) continue status = _summarize_payment_wait_status(page) if status and status != last_status: ctx.log(f"[paypal] 仍在等待支付完成: {status}") last_status = status if time.time() >= deadline: if ctx.headless: _dump_page(ctx, page, "14-paypal-payment-wait-timeout") raise TimeoutError(f"等待支付完成超时 {timeout}s") if not manual_mode: ctx.log("[paypal] 自动等待已超时，当前为可视浏览器，保持窗口打开；完成后会继续检测，或点击停止结束任务") _dump_page(ctx, page, "14-paypal-payment-wait-timeout") manual_mode = True interval = POST_PAYMENT_MANUAL_INTERVAL page.wait_for_timeout(interval * 1000) def _click_post_sms_action_if_available(ctx, page, clicked_actions: set) -> bool: url = getattr(page, "url", "") or "" candidates = [] if "paypal.com" in url: candidates.extend((f'button:has-text("{text}")', text) for text in POST_SMS_PAYPAL_ACTION_TEXTS) if "pay.openai.com" in url or "checkout.stripe.com" in url: candidates.append(('button[data-testid="hosted-payment-submit-button"]', "Stripe hosted submit")) candidates.extend((f'button:has-text("{text}")', text) for text in POST_SMS_STRIPE_ACTION_TEXTS) for selector, label in candidates: key = (url.split("?", 1)[0], label) if key in clicked_actions: continue try: loc = page.locator(selector).first if loc.count() == 0: continue enabled = loc.is_enabled() ctx.log(f"[paypal] 后续确认候选 {label} selector={selector} enabled={enabled}") if not enabled: continue loc.click(timeout=4000) clicked_actions.add(key) ctx.log(f"[paypal] 已点击后续确认按钮: {label}") return True except Exception as exc: ctx.log(f"[paypal] 后续确认按钮 {label} 处理失败: {exc!r}") continue return False def _page_is_closed(page) -> bool: try: is_closed = getattr(page, "is_closed", None) return bool(is_closed and is_closed()) except Exception: return False def _detect_payment_completion(page) -> str: url = getattr(page, "url", "") or "" snapshot = _payment_page_snapshot(page) text = snapshot.get("text", "") if "chatgpt.com" in url: return f"已跳回 ChatGPT: {url}" if _PAYMENT_COMPLETE_RE.search(text): return "页面出现完成提示" if ( ("pay.openai.com" in url or "checkout.stripe.com" in url) and re.search(r"(complete|success|return|receipt)", url, re.I) ): return f"支付页进入完成 URL: {url}" return "" def _summarize_payment_wait_status(page) -> str: url = getattr(page, "url", "") or "" snapshot = _payment_page_snapshot(page) alerts = snapshot.get("alerts", "") text = snapshot.get("text", "") invalid_count = snapshot.get("invalid_count", 0) parts = [f"URL={url}"] if invalid_count: parts.append(f"invalid_fields={invalid_count}") if alerts: parts.append(f"alert={alerts[:300]}") else: blocking = _PAYMENT_BLOCKING_RE.search(text) if blocking: start = max(0, blocking.start() - 80) end = min(len(text), blocking.end() + 160) parts.append(f"page_hint={text[start:end].strip()[:300]}") return " | ".join(parts) def _payment_page_snapshot(page) -> dict: try: data = page.evaluate(r"""() => { const bodyText = (document.body && document.body.innerText || '').replace(/\s+/g, ' ').trim(); const alerts = Array.from(document.querySelectorAll( '[role="alert"], [data-testid*="error" i], [class*="error" i], [aria-invalid="true"]' )).map(el => (el.innerText || el.value || el.getAttribute('aria-label') || '').replace(/\s+/g, ' ').trim()) .filter(Boolean) .slice(0, 5) .join(' | '); return { title: document.title || '', text: bodyText.slice(0, 5000), alerts, invalid_count: document.querySelectorAll('[aria-invalid="true"]').length }; }""") return data if isinstance(data, dict) else {} except Exception: return {} def _safe_fill(ctx, page, selector: str, value: str, label: str = "", mask: bool = False): label = label or selector shown = "***" if mask else value try: loc = page.locator(selector).first count = loc.count() if count == 0: ctx.log(f"[fill] {label}({selector}) 不存在，跳过") return loc.wait_for(state="visible", timeout=8000) loc.fill(value) ctx.log(f"[fill] {label}({selector}) <- {shown}") except Exception as exc: ctx.log(f"[fill] {label}({selector}) 失败: {exc!r}") def _select_by_text(ctx, page, selector: str, text: str, label: str = ""): label = label or selector try: loc = page.locator(selector).first if loc.count() == 0: ctx.log(f"[select] {label}({selector}) 不存在，跳过") return loc.wait_for(state="visible", timeout=4000) for kind in ("value", "label", "raw"): try: if kind == "value": loc.select_option(value=text) elif kind == "label": loc.select_option(label=text) else: loc.select_option(text) ctx.log(f"[select] {label}({selector}) <- {text!r} 命中方式={kind}") return except Exception as exc: ctx.log(f"[select] {label} 尝试 {kind} 失败: {exc!r}") ctx.log(f"[select] {label}({selector}) 三种方式都失败") except Exception as exc: ctx.log(f"[select] {label}({selector}) 异常: {exc!r}") def _click_next(ctx, page, tag: str = ""): candidates = [ 'button[data-testid="submit-button"]', 'button[data-testid="hosted-payment-submit-button"]', 'button[data-atomic-wait-intent="Submit_Email"]', 'button.SubmitButton--complete', ] for sel in candidates: loc = page.locator(sel).first if loc.count() == 0: continue try: enabled = loc.is_enabled() ctx.log(f"[click:{tag}] 候选 {sel} count={loc.count()} enabled={enabled}") if not enabled: continue try: loc.click(timeout=4000) ctx.log(f"[click:{tag}] 已点击 {sel}") return True except Exception as exc: msg = str(exc) if "intercepts pointer events" in msg or "captchaComponent" in msg: ctx.log(f"[click:{tag}] {sel} 被蒙层挡住，尝试删 captcha 后重试") _paypal_remove_captcha_overlay(ctx, page) try: loc.click(timeout=4000, force=True) ctx.log(f"[click:{tag}] 重试已点击 {sel}") return True except Exception as exc2: ctx.log(f"[click:{tag}] {sel} 重试仍失败: {exc2!r}") continue ctx.log(f"[click:{tag}] {sel} 点击失败: {exc!r}") except Exception as exc: ctx.log(f"[click:{tag}] {sel} 处理失败: {exc!r}") continue for text in ["Next", "Subscribe", "Pay", "Continue", "Agree", "下一步", "下一页", "訂閱"]: loc = page.locator(f'button:has-text("{text}")').first if loc.count() == 0: continue try: enabled = loc.is_enabled() ctx.log(f"[click:{tag}] 文本候选 has-text={text!r} enabled={enabled}") if not enabled: continue try: loc.click(timeout=4000) ctx.log(f"[click:{tag}] 已点击 has-text={text!r}") return True except Exception as exc: msg = str(exc) if "intercepts pointer events" in msg or "captchaComponent" in msg: ctx.log(f"[click:{tag}] has-text={text!r} 被蒙层挡住，删 captcha 重试") _paypal_remove_captcha_overlay(ctx, page) try: loc.click(timeout=4000, force=True) ctx.log(f"[click:{tag}] 重试已点击 has-text={text!r}") return True except Exception as exc2: ctx.log(f"[click:{tag}] has-text={text!r} 重试仍失败: {exc2!r}") continue ctx.log(f"[click:{tag}] has-text={text!r} 点击失败: {exc!r}") except Exception as exc: ctx.log(f"[click:{tag}] has-text={text!r} 处理失败: {exc!r}") continue ctx.log(f"[click:{tag}] 未找到任何可点击的下一步按钮") return False def run(ctx: RunContext): ctx.log(f"[run] === 任务开始 run_id={ctx.run_id} artifact_dir={ctx.artifact_dir} ===") try: ctx.long_link = generate_long_link(ctx) run_paypal_flow(ctx) ctx.state = "done" ctx.log("[run] === 全部步骤已尝试完成 ===") except Exception as exc: if str(exc) == "STOPPED_BY_USER": ctx.log("[run] === 用户中止 ===") ctx.state = "stopped" else: ctx.log(f"[run] === 异常退出: {exc!r} ===") ctx.log(traceback.format_exc()) ctx.state = "error"