autoChatGPTPlus-Paypal/mail_provider.py

"""邮件 provider：a4sky 邮箱生成 + 通过 hotmail-helper 协议拉验证码。

helper 协议（Chrome 扩展原版）：
  POST {helper_url}/imap-code
  body: {
    targetEmail, mailbox: "INBOX", top: 60,
    senderFilters, subjectFilters, excludeCodes, filterAfterTimestamp
  }
  返回: { ok, code, message: { id, mailbox, subject, receivedTimestamp, ... }, usedTimeFallback, transport }
helper 自己读 IMAP 配置（data/a4sky-imap.local.json），客户端不传账号密码。
"""
from __future__ import annotations

import json
import time
import urllib.error
import urllib.request
from datetime import datetime
from typing import Callable

DEFAULT_SENDER_FILTERS = ["openai", "noreply", "verify", "auth", "duckduckgo", "forward", "chatgpt"]
DEFAULT_SUBJECT_FILTERS = ["verify", "verification", "code", "验证码", "confirm"]


def build_a4sky_email(domain: str = "edu.a4sky.com") -> str:
    """格式：n<YYYYMMDDHHMMSS>@<domain>，与 Chrome 扩展一致。"""
    ts = datetime.now().strftime("%Y%m%d%H%M%S")
    return f"n{ts}@{domain}"


def _post_json(url: str, body: dict, timeout: int = 30) -> dict:
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="POST")
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json")
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        text = resp.read().decode("utf-8", errors="replace")
    if not text:
        return {}
    return json.loads(text)


def request_imap_code(
    helper_url: str,
    target_email: str,
    *,
    filter_after_ts_ms: int = 0,
    exclude_codes: list[str] | None = None,
    sender_filters: list[str] | None = None,
    subject_filters: list[str] | None = None,
    top: int = 60,
    timeout: int = 30,
) -> dict:
    """单次请求 helper /imap-code。返回原始响应。"""
    base = (helper_url or "").rstrip("/")
    if not base:
        raise RuntimeError("未配置邮件助手地址")
    url = f"{base}/imap-code"
    payload = {
        "targetEmail": (target_email or "").strip().lower(),
        "mailbox": "INBOX",
        "top": top,
        "senderFilters": sender_filters or DEFAULT_SENDER_FILTERS,
        "subjectFilters": subject_filters or DEFAULT_SUBJECT_FILTERS,
        "excludeCodes": exclude_codes or [],
        "filterAfterTimestamp": int(filter_after_ts_ms or 0),
    }
    return _post_json(url, payload, timeout=timeout)


def poll_signup_code(
    helper_url: str,
    target_email: str,
    *,
    started_at_ms: int,
    interval_sec: int = 4,
    max_attempts: int = 60,
    exclude_codes: list[str] | None = None,
    log: Callable[[str], None] = print,
) -> str:
    """轮询邮箱直到拿到验证码。返回 6 位 code 字符串。"""
    if not target_email:
        raise RuntimeError("未指定目标邮箱")

    log(f"[mail] 开始轮询验证码 helper={helper_url} email={target_email} interval={interval_sec}s max_attempts={max_attempts}")
    last_err = None
    for attempt in range(1, max_attempts + 1):
        try:
            resp = request_imap_code(
                helper_url,
                target_email,
                filter_after_ts_ms=started_at_ms,
                exclude_codes=exclude_codes,
            )
        except urllib.error.HTTPError as exc:
            try:
                body = exc.read().decode("utf-8", errors="replace")
            except Exception:
                body = ""
            last_err = f"HTTP {exc.code} {body[:200]}"
            log(f"[mail] 第{attempt}次轮询请求 HTTPError: {last_err}")
        except Exception as exc:
            last_err = repr(exc)
            log(f"[mail] 第{attempt}次轮询异常: {last_err}")
        else:
            ok = bool(resp.get("ok"))
            code = str(resp.get("code") or "").strip()
            msg = resp.get("message") or {}
            subj = (msg or {}).get("subject", "")
            ts = (msg or {}).get("receivedTimestamp", 0)
            transport = resp.get("transport", "")
            if ok and code:
                log(f"[mail] 第{attempt}次轮询命中 code={code} subject={subj!r} ts={ts} transport={transport}")
                return code
            log(f"[mail] 第{attempt}次轮询未命中 ok={ok} code={code!r} usedTimeFallback={resp.get('usedTimeFallback')} subject={subj!r}")

        if attempt < max_attempts:
            time.sleep(interval_sec)

    raise TimeoutError(f"轮询验证码超时（{max_attempts} 次），最后错误：{last_err}")