autoChatGPTPlus-Paypal/cpa_oauth.py

"""CPA Codex OAuth 授权：获取 OAuth URL、捕获 localhost 回调、提交给 CPA。"""
from __future__ import annotations

import time
from typing import Callable
from urllib.parse import urlparse


CALLBACK_PATHS = {"/auth/callback", "/codex/callback"}
CPA_OAUTH_URL_SELECTORS = ('[class*="authUrlValue"]', '.OAuthPage-module__authUrlValue___axvUJ')
CPA_MANAGEMENT_KEY_SELECTORS = (
    '.LoginPage-module__loginCard___OgP-R input[type="password"]',
    'input[type="password"]',
    'input[placeholder*="管理"]',
    'input[placeholder*="key" i]',
    'input[placeholder*="Key"]',
    'input[aria-label*="管理密钥"]',
)
CPA_OAUTH_NAV_SELECTORS = (
    'a[href*="#/oauth"]',
    'button:has-text("OAuth")',
    '[role="link"]:has-text("OAuth")',
    '[role="button"]:has-text("OAuth")',
)
OPENAI_EMAIL_SELECTORS = (
    'input[type="email"]',
    'input[name="email"]',
    'input[name="username"]',
    'input[autocomplete="username"]',
    'input[placeholder*="email" i]',
)


def is_localhost_oauth_callback_url(raw_url: str) -> bool:
    try:
        parsed = urlparse(str(raw_url or ""))
    except Exception:
        return False
    if parsed.scheme not in ("http", "https"):
        return False
    if parsed.hostname not in ("localhost", "127.0.0.1"):
        return False
    if parsed.path not in CALLBACK_PATHS:
        return False
    query = parsed.query or ""
    return "code=" in query and "state=" in query


def build_cpa_oauth_panel_url(cpa_url: str) -> str:
    url = str(cpa_url or "").strip()
    if not url:
        raise RuntimeError("CPA 地址未配置")
    parsed = urlparse(url)
    if not parsed.scheme or not parsed.netloc:
        raise RuntimeError(f"CPA 地址格式无效: {cpa_url}")
    if parsed.fragment == "/oauth" or parsed.fragment == "oauth":
        return url
    if parsed.path.endswith("/management.html"):
        return f"{parsed.scheme}://{parsed.netloc}{parsed.path}#/oauth"
    return f"{parsed.scheme}://{parsed.netloc}/management.html#/oauth"


def _wait_until(predicate: Callable[[], bool], timeout_sec: int, interval_ms: int = 300) -> bool:
    deadline = time.time() + timeout_sec
    while time.time() < deadline:
        try:
            if predicate():
                return True
        except Exception:
            pass
        time.sleep(interval_ms / 1000)
    return False


def _safe_page_wait(page, timeout_ms: int) -> None:
    try:
        page.wait_for_timeout(timeout_ms)
    except Exception:
        time.sleep(timeout_ms / 1000)


def _try_click_first_visible(page, selector: str, log, *, label: str = "") -> bool:
    loc = page.locator(selector)
    count = loc.count()
    for i in range(count):
        item = loc.nth(i)
        try:
            if item.is_visible() and item.is_enabled():
                item.click(timeout=4000)
                log(f"[cpa-oauth] 点击 {label or selector} (idx={i}) 成功")
                return True
        except Exception as exc:
            log(f"[cpa-oauth] 点击 {label or selector} (idx={i}) 失败: {exc!r}")
    return False


def _fill_first_visible(page, selectors: tuple[str, ...], value: str) -> str:
    for selector in selectors:
        loc = page.locator(selector)
        count = loc.count()
        for i in range(count):
            item = loc.nth(i)
            try:
                if item.is_visible():
                    item.fill(value)
                    return selector
            except Exception:
                continue
    return ""


def _read_first_visible_text(page, selectors: tuple[str, ...]) -> str:
    for selector in selectors:
        loc = page.locator(selector)
        count = loc.count()
        for i in range(count):
            item = loc.nth(i)
            try:
                if item.is_visible():
                    text = (item.inner_text(timeout=1000) or "").strip()
                    if text.startswith(("http://", "https://")):
                        return text
            except Exception:
                continue
    return ""


def _has_visible(page, selectors: tuple[str, ...]) -> bool:
    for selector in selectors:
        loc = page.locator(selector)
        count = loc.count()
        for i in range(count):
            try:
                if loc.nth(i).is_visible():
                    return True
            except Exception:
                continue
    return False


def _context_pages(page) -> list:
    context = getattr(page, "context", None)
    pages_attr = getattr(context, "pages", None) if context is not None else None
    if pages_attr is None:
        return []
    try:
        pages = pages_attr() if callable(pages_attr) else pages_attr
    except Exception:
        return []
    try:
        return list(pages or [])
    except Exception:
        return []


def _find_localhost_callback_url(page, log=None) -> str:
    current_url = str(getattr(page, "url", "") or "")
    if is_localhost_oauth_callback_url(current_url):
        return current_url
    for candidate in _context_pages(page):
        if candidate is page:
            continue
        candidate_url = str(getattr(candidate, "url", "") or "")
        if is_localhost_oauth_callback_url(candidate_url):
            if log:
                log("[cpa-oauth] 从同一浏览器上下文的其他页面捕获 localhost callback")
            return candidate_url
    return ""


def _login_cpa_management_if_needed(page, management_key: str, log) -> bool:
    if not management_key:
        return False
    filled_selector = _fill_first_visible(
        page,
        CPA_MANAGEMENT_KEY_SELECTORS,
        management_key,
    )
    if not filled_selector:
        return False
    log(f"[cpa-oauth] 已填写 CPA 管理密钥 selector={filled_selector}")
    for selector in (
        'button:has-text("登录")',
        'button:has-text("Login")',
        'button:has-text("Sign in")',
        'button[type="submit"]',
    ):
        if _try_click_first_visible(page, selector, log, label=f"cpa-management-login({selector})"):
            _safe_page_wait(page, 1200)
            return True
    return False


def _ensure_cpa_oauth_panel_ready(
    page,
    *,
    panel_url: str,
    management_key: str,
    log: Callable[[str], None],
    timeout_sec: int,
) -> str:
    """确保 CPA 面板已越过管理登录页，返回已存在的 OAuth URL（如有）。"""
    deadline = time.time() + timeout_sec
    management_login_attempts = 0
    while time.time() < deadline:
        oauth_url = _read_first_visible_text(page, CPA_OAUTH_URL_SELECTORS)
        if oauth_url:
            return oauth_url

        url = str(getattr(page, "url", "") or "").lower()
        login_visible = _has_visible(page, CPA_MANAGEMENT_KEY_SELECTORS)
        if "#/oauth" in url and not login_visible:
            return ""

        if login_visible:
            management_login_attempts += 1
            if management_login_attempts > 3:
                raise RuntimeError(f"CPA 管理登录失败，请检查管理密钥或权限 URL={page.url}")
            _login_cpa_management_if_needed(page, management_key, log)
            _safe_page_wait(page, 800)
            continue

        if "#/oauth" not in url:
            log(f"[cpa-oauth] CPA 登录后未在 OAuth 路由，重新打开 {panel_url}")
            page.goto(panel_url, wait_until="domcontentloaded", timeout=60000)
            _safe_page_wait(page, 800)
            continue

        nav_clicked = False
        for selector in CPA_OAUTH_NAV_SELECTORS:
            if _try_click_first_visible(page, selector, log, label=f"cpa-oauth-nav({selector})"):
                nav_clicked = True
                break
        _safe_page_wait(page, 800 if nav_clicked else 300)

    raise RuntimeError(f"CPA 管理登录后未进入 OAuth 面板 URL={page.url}")


def fetch_cpa_oauth_url(
    page,
    *,
    cpa_url: str,
    management_key: str,
    log: Callable[[str], None] = print,
    timeout_sec: int = 45,
) -> str:
    panel_url = build_cpa_oauth_panel_url(cpa_url)
    log(f"[cpa-oauth] 打开 CPA OAuth 面板 {panel_url}")
    page.goto(panel_url, wait_until="domcontentloaded", timeout=60000)
    try:
        page.wait_for_timeout(1200)
    except Exception:
        pass

    oauth_url = _ensure_cpa_oauth_panel_ready(
        page,
        panel_url=panel_url,
        management_key=management_key,
        log=log,
        timeout_sec=timeout_sec,
    )

    if oauth_url:
        log(f"[cpa-oauth] CPA 面板已有 OAuth URL {oauth_url[:80]}...")
        return oauth_url

    clicked = False
    for selector in (
        'button:has-text("登录")',
        'button:has-text("Login")',
        'button:has-text("OAuth")',
    ):
        if _try_click_first_visible(page, selector, log, label=f"cpa-oauth-login({selector})"):
            clicked = True
            break
    if not clicked:
        raise RuntimeError(f"CPA OAuth 面板未找到登录按钮 URL={page.url}")

    ok = _wait_until(
        lambda: bool(_read_first_visible_text(page, CPA_OAUTH_URL_SELECTORS)),
        timeout_sec,
    )
    if not ok:
        raise RuntimeError(f"点击 CPA OAuth 登录后未出现授权链接 URL={page.url}")
    oauth_url = _read_first_visible_text(page, CPA_OAUTH_URL_SELECTORS)
    log(f"[cpa-oauth] 获取 OAuth URL {oauth_url[:80]}...")
    return oauth_url


def _click_openai_oauth_continue(page, log) -> bool:
    for selector in (
        'button[type="submit"]',
        'button:has-text("Continue")',
        'button:has-text("Allow")',
        'button:has-text("Authorize")',
        'button:has-text("继续")',
        'button:has-text("允许")',
        'button:has-text("授权")',
        'input[type="submit"]',
    ):
        if _try_click_first_visible(page, selector, log, label=f"openai-oauth-continue({selector})"):
            return True
    return False


def _fill_openai_email_if_present(page, email_hint: str, log) -> bool:
    email = str(email_hint or "").strip()
    if not email:
        return False
    filled_selector = _fill_first_visible(page, OPENAI_EMAIL_SELECTORS, email)
    if not filled_selector:
        return False
    log(f"[cpa-oauth] 已填写 OpenAI OAuth 登录邮箱 selector={filled_selector}")
    if _click_openai_oauth_continue(page, log):
        _safe_page_wait(page, 1500)
    return True


def _drive_openai_oauth_until_callback(page, email_hint: str, log, timeout_sec: int) -> str:
    deadline = time.time() + timeout_sec
    acted = False
    email_submitted = False
    while time.time() < deadline:
        callback_url = _find_localhost_callback_url(page, log)
        if callback_url:
            return callback_url

        if not email_submitted:
            email_submitted = _fill_openai_email_if_present(page, email_hint, log)
            callback_url = _find_localhost_callback_url(page, log)
            if callback_url:
                return callback_url
            if email_submitted:
                acted = True
                _safe_page_wait(page, 500)
                continue

        clicked = _click_openai_oauth_continue(page, log)
        acted = acted or clicked
        if clicked:
            _safe_page_wait(page, 1200)
            callback_url = _find_localhost_callback_url(page, log)
            if callback_url:
                return callback_url
        else:
            _safe_page_wait(page, 500)
            callback_url = _find_localhost_callback_url(page, log)
            if callback_url:
                return callback_url
        callback_url = _find_localhost_callback_url(page, log)
        if callback_url:
            return callback_url
        if email_submitted:
            acted = True

    if acted:
        raise RuntimeError(f"OAuth 授权后未捕获 localhost callback URL={page.url}")
    raise RuntimeError(f"OAuth 页面未找到可继续的登录或授权按钮 URL={page.url}")


def approve_openai_oauth_and_capture_callback(
    page,
    oauth_url: str,
    *,
    email_hint: str = "",
    log: Callable[[str], None] = print,
    timeout_sec: int = 120,
) -> str:
    if not str(oauth_url or "").startswith(("http://", "https://")):
        raise RuntimeError(f"OAuth URL 无效: {oauth_url}")
    log(f"[cpa-oauth] 打开 OpenAI OAuth URL {oauth_url[:80]}...")
    try:
        page.goto(oauth_url, wait_until="domcontentloaded", timeout=60000)
    except Exception as exc:
        if not _find_localhost_callback_url(page, log):
            raise
        log(f"[cpa-oauth] localhost callback 导航报错但 URL 已捕获: {exc!r}")
    try:
        page.wait_for_timeout(1500)
    except Exception:
        pass

    callback_url = _find_localhost_callback_url(page, log)
    if callback_url:
        return callback_url

    callback_url = _drive_openai_oauth_until_callback(page, email_hint, log, timeout_sec)
    log(f"[cpa-oauth] 捕获 callback {callback_url[:80]}...")
    return callback_url


def submit_oauth_callback_to_cpa(
    page,
    *,
    cpa_url: str,
    callback_url: str,
    management_key: str = "",
    log: Callable[[str], None] = print,
    wait_for_success: bool = True,
    timeout_sec: int = 120,
) -> str:
    if not is_localhost_oauth_callback_url(callback_url):
        raise RuntimeError("CPA OAuth callback URL 无效")

    panel_url = build_cpa_oauth_panel_url(cpa_url)
    log(f"[cpa-oauth] 回到 CPA 面板提交 callback {panel_url}")
    page.goto(panel_url, wait_until="domcontentloaded", timeout=60000)
    try:
        page.wait_for_timeout(1200)
    except Exception:
        pass

    _ensure_cpa_oauth_panel_ready(
        page,
        panel_url=panel_url,
        management_key=management_key,
        log=log,
        timeout_sec=45,
    )

    filled_selector = _fill_first_visible(
        page,
        (
            'input[placeholder*="localhost"]',
            '[class*="callbackSection"] input.input',
            'input.input',
        ),
        callback_url,
    )
    if not filled_selector:
        raise RuntimeError(f"CPA 面板未找到 callback 输入框 URL={page.url}")
    log(f"[cpa-oauth] 已填写 callback selector={filled_selector}")

    submitted = False
    for selector in (
        'button:has-text("提交回调 URL")',
        'button:has-text("Submit Callback URL")',
        'button:has-text("Callback URL")',
        '[class*="callbackActions"] button',
        '[class*="callbackSection"] button',
        'button.btn',
    ):
        if _try_click_first_visible(page, selector, log, label=f"cpa-callback-submit({selector})"):
            submitted = True
            break
    if not submitted:
        raise RuntimeError(f"CPA 面板未找到 callback 提交按钮 URL={page.url}")

    if not wait_for_success:
        return ""

    success_texts = ("认证成功", "Authentication successful", "Аутентификация успешна")
    failure_texts = ("认证失败", "回调 URL 提交失败", "oauth flow is not pending", "callback url submit failed")

    def _status_text() -> str:
        try:
            body = page.locator("body").first
            return body.inner_text(timeout=1000) or ""
        except Exception:
            return ""

    def _is_done() -> bool:
        text = _status_text()
        lowered = text.lower()
        if any(marker.lower() in lowered for marker in failure_texts):
            raise RuntimeError(f"CPA OAuth 回调提交失败: {text[:300]}")
        return any(marker.lower() in lowered for marker in success_texts)

    ok = _wait_until(_is_done, timeout_sec)
    if not ok:
        raise RuntimeError(f"等待 CPA OAuth 认证成功超时 URL={page.url}")
    log("[cpa-oauth] CPA OAuth 认证成功")
    return "Authentication successful!"


def authorize_codex_oauth_to_cpa(
    page,
    *,
    cpa_url: str,
    management_key: str,
    email_hint: str = "",
    log: Callable[[str], None] = print,
) -> dict:
    oauth_url = fetch_cpa_oauth_url(page, cpa_url=cpa_url, management_key=management_key, log=log)
    oauth_page = page
    close_oauth_page = False
    try:
        context = getattr(page, "context", None)
        if context is not None and hasattr(context, "new_page"):
            oauth_page = context.new_page()
            close_oauth_page = True
            log("[cpa-oauth] 已打开独立 OpenAI OAuth 页面，CPA 面板页保持打开")
    except Exception as exc:
        log(f"[cpa-oauth] 打开独立 OAuth 页面失败，将复用当前页: {exc!r}")
        oauth_page = page
        close_oauth_page = False

    try:
        callback_url = approve_openai_oauth_and_capture_callback(
            oauth_page,
            oauth_url,
            email_hint=email_hint,
            log=log,
        )
    finally:
        if close_oauth_page:
            try:
                oauth_page.close()
            except Exception as exc:
                log(f"[cpa-oauth] 关闭 OpenAI OAuth 页面异常: {exc!r}")

    status = submit_oauth_callback_to_cpa(
        page,
        cpa_url=cpa_url,
        callback_url=callback_url,
        management_key=management_key,
        log=log,
    )
    return {
        "fileName": "",
        "email": email_hint,
        "planType": "codex-oauth",
        "hasRefreshToken": True,
        "status": status,
        "callbackUrl": callback_url,
        "oauthUrl": oauth_url,
    }