refactor: modularize registration runtime safely

cf_mail_debug.py

@@ -1,198 +1,47 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-
 import argparse
-import re
-import secrets
-import string
 import time
-from typing import Any, Dict, List, Optional, Tuple
 
 from curl_cffi import requests
+from mail_service import CloudflareMailClient, extract_verification_code
 
 
-def extract_code(text: str, subject: str = "") -> Optional[str]:
-    if subject:
-        m = re.search(r"\b([A-Z0-9]{3}-[A-Z0-9]{3})\b", subject, re.IGNORECASE)
-        if m:
-            return m.group(1)
-    m = re.search(r"\b([A-Z0-9]{3}-[A-Z0-9]{3})\b", text, re.IGNORECASE)
-    if m:
-        return m.group(1)
-    for p in [
-        r"verification\s+code[:\s]+(\d{4,8})",
-        r"your\s+code[:\s]+(\d{4,8})",
-        r"confirm(?:ation)?\s+code[:\s]+(\d{4,8})",
-    ]:
-        m = re.search(p, text, re.IGNORECASE)
-        if m:
-            return m.group(1)
-    return None
-
-
-def json_or_text(resp: requests.Response) -> Tuple[Optional[Dict[str, Any]], str]:
+def create_address(api_base, auth_mode="none", api_key="", create_path="/api/new_address",
+                   domain="", name="", timeout=20):
+    import mail_service as _mail_service
+    client = CloudflareMailClient(
+        api_base, auth_mode=auth_mode, api_key=api_key,
+        create_path=create_path, timeout=timeout,
+    )
+    original_requests = _mail_service.requests
+    _mail_service.requests = requests
     try:
-        data = resp.json()
-        return data, ""
-    except Exception:
-        return None, (resp.text or "")[:400]
-
-
-def generate_username(length: int = 10) -> str:
-    """生成 cloudflare_temp_email admin API 需要的随机邮箱名称。"""
-    chars = string.ascii_lowercase + string.digits
-    return "".join(secrets.choice(chars) for _ in range(length))
-
-
-def normalize_path(path: str, default_path: str) -> str:
-    """标准化 API 路径，避免漏写开头斜杠。"""
-    raw = (path or default_path).strip() or default_path
-    return raw if raw.startswith("/") else f"/{raw}"
-
-
-def build_auth_headers(auth_mode: str, api_key: str, content_type: bool = False) -> Dict[str, str]:
-    """按调试参数构造 Cloudflare 临时邮箱接口鉴权请求头。"""
-    headers = {"Content-Type": "application/json"} if content_type else {}
-    key = (api_key or "").strip()
-    mode = (auth_mode or "none").strip().lower()
-    if not key:
-        return headers
-    if mode == "x-admin-auth":
-        headers["x-admin-auth"] = key
-    elif mode == "x-api-key":
-        headers["X-API-Key"] = key
-    elif mode == "bearer":
-        headers["Authorization"] = f"Bearer {key}"
-    return headers
-
-
-def create_address(
-    api_base: str,
-    auth_mode: str = "none",
-    api_key: str = "",
-    create_path: str = "/api/new_address",
-    domain: str = "",
-    name: str = "",
-) -> Tuple[str, str]:
-    """创建 Cloudflare 临时邮箱地址，支持匿名 API 和 admin API。"""
-    path = normalize_path(create_path, "/api/new_address")
-    is_admin_create = path.rstrip("/").lower() == "/admin/new_address"
-    if is_admin_create:
-        payload: Dict[str, Any] = {
-            "name": name.strip() if name.strip() else generate_username(),
-            "enablePrefix": True,
-        }
-        if domain.strip():
-            payload["domain"] = domain.strip()
-        headers = build_auth_headers(auth_mode, api_key, content_type=True)
-    else:
-        payload = {}
-        if domain.strip():
-            payload["domain"] = domain.strip()
-        headers = {"Content-Type": "application/json"}
-    resp = requests.post(
-        f"{api_base.rstrip('/')}{path}",
-        json=payload,
-        headers=headers,
-        timeout=20,
-    )
-    resp.raise_for_status()
-    data, raw = json_or_text(resp)
-    if not data:
-        raise RuntimeError(f"{path} 非JSON: {raw}")
-    address = str(data.get("address", "")).strip()
-    jwt = str(data.get("jwt", "")).strip()
-    if not address or not jwt:
-        raise RuntimeError(f"{path} 缺少 address/jwt: {data}")
-    return address, jwt
-
-
-def fetch_box(api_base: str, jwt: str, path: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
-    resp = requests.get(
-        f"{api_base.rstrip('/')}{path}",
-        params=params,
-        headers={"Authorization": f"Bearer {jwt}"},
-        timeout=20,
-    )
-    if resp.status_code >= 400:
-        return []
-    data, _ = json_or_text(resp)
-    if not isinstance(data, dict):
-        return []
-    if isinstance(data.get("results"), list):
-        return data["results"]
-    if isinstance(data.get("data"), list):
-        return data["data"]
-    if isinstance(data.get("messages"), list):
-        return data["messages"]
-    return []
-
-
-def probe_all_boxes(api_base: str, jwt: str) -> List[Tuple[str, List[Dict[str, Any]]]]:
-    probes = [
-        ("/api/mails", {"limit": 20, "offset": 0}),
-        ("/api/sendbox", {"limit": 20, "offset": 0}),
-        ("/api/mails", {"limit": 20, "offset": 0, "box": "trash"}),
-        ("/api/mails", {"limit": 20, "offset": 0, "folder": "trash"}),
-        ("/api/mails", {"limit": 20, "offset": 0, "deleted": "1"}),
-        ("/api/mails", {"limit": 20, "offset": 0, "status": "deleted"}),
-    ]
-    out: List[Tuple[str, List[Dict[str, Any]]]] = []
-    for path, params in probes:
-        mails = fetch_box(api_base, jwt, path, params)
-        out.append((f"{path}?{params}", mails))
-    return out
-
-
-def get_detail(api_base: str, jwt: str, mail_id: Any) -> Dict[str, Any]:
-    for url in [
-        f"{api_base.rstrip('/')}/api/mail/{mail_id}",
-        f"{api_base.rstrip('/')}/api/mails/{mail_id}",
-    ]:
-        try:
-            resp = requests.get(url, headers={"Authorization": f"Bearer {jwt}"}, timeout=20)
-            if resp.status_code >= 400:
-                continue
-            data, _ = json_or_text(resp)
-            if isinstance(data, dict):
-                return data
-        except Exception:
-            continue
-    return {}
-
-
-def flatten_mail_text(item: Dict[str, Any], detail: Dict[str, Any]) -> Tuple[str, str]:
-    subject = str(item.get("subject") or detail.get("subject") or "")
-    parts: List[str] = []
-    for src in (item, detail):
-        for k in ("text", "raw", "content", "intro", "body", "snippet"):
-            v = src.get(k)
-            if isinstance(v, str) and v.strip():
-                parts.append(v)
-        html_val = src.get("html")
-        if isinstance(html_val, str):
-            html_val = [html_val]
-        if isinstance(html_val, list):
-            for h in html_val:
-                if isinstance(h, str):
-                    parts.append(re.sub(r"<[^>]+>", " ", h))
-    return subject, "\n".join(parts)
+        return client.create_address(domain=domain, name=name)
+    finally:
+        _mail_service.requests = original_requests
 
 
 def main():
-    ap = argparse.ArgumentParser()
-    ap.add_argument("--api-base", required=True)
-    ap.add_argument("--address", default="")
-    ap.add_argument("--credential", default="")
-    ap.add_argument("--auth-mode", default="none", choices=["none", "bearer", "x-api-key", "x-admin-auth"])
-    ap.add_argument("--api-key", default="")
-    ap.add_argument("--create-path", default="/api/new_address")
-    ap.add_argument("--domain", default="")
-    ap.add_argument("--name", default="")
-    ap.add_argument("--timeout", type=int, default=180)
-    ap.add_argument("--interval", type=int, default=3)
-    args = ap.parse_args()
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--api-base", required=True)
+    parser.add_argument("--address", default="")
+    parser.add_argument("--credential", default="")
+    parser.add_argument("--auth-mode", default="none", choices=["none", "bearer", "x-api-key", "x-admin-auth"])
+    parser.add_argument("--api-key", default="")
+    parser.add_argument("--create-path", default="/api/new_address")
+    parser.add_argument("--domain", default="")
+    parser.add_argument("--name", default="")
+    parser.add_argument("--timeout", type=int, default=180)
+    parser.add_argument("--interval", type=int, default=3)
+    args = parser.parse_args()
 
+    client = CloudflareMailClient(
+        args.api_base,
+        auth_mode=args.auth_mode,
+        api_key=args.api_key,
+        create_path=args.create_path,
+    )
     address = args.address.strip()
     credential = args.credential.strip()
     if not credential:
@@ -204,31 +53,31 @@ def main():
             domain=args.domain,
             name=args.name,
         )
-        print(f"[NEW] address={address}")
-        print(f"[NEW] credential(jwt)={credential}")
+        print("[NEW] address=%s" % address)
+        print("[NEW] credential(jwt)=%s" % credential)
     else:
-        print(f"[USE] address={address or '(unknown, from credential)'}")
+        print("[USE] address=%s" % (address or "(unknown, from credential)"))
 
     deadline = time.time() + max(args.timeout, 1)
     seen_ids = set()
     while time.time() < deadline:
-        boxes = probe_all_boxes(args.api_base, credential)
+        boxes = client.probe_all_boxes(credential)
         total = 0
-        for name, mails in boxes:
+        for box_name, mails in boxes:
             if mails:
-                print(f"[BOX] {name} -> {len(mails)}")
+                print("[BOX] %s -> %s" % (box_name, len(mails)))
             total += len(mails)
-            for m in mails:
-                mail_id = m.get("id") or m.get("mail_id")
+            for item in mails:
+                mail_id = item.get("id") or item.get("mail_id")
                 if not mail_id or mail_id in seen_ids:
                     continue
                 seen_ids.add(mail_id)
-                detail = get_detail(args.api_base, credential, mail_id)
-                subj, text = flatten_mail_text(m, detail)
-                code = extract_code(text, subj)
-                print(f"[MAIL] id={mail_id} subject={subj!r} code={code!r}")
+                detail = client.get_detail(credential, mail_id)
+                subject, text = client.flatten_mail_text(item, detail)
+                code = extract_verification_code(text, subject)
+                print("[MAIL] id=%s subject=%r code=%r" % (mail_id, subject, code))
                 if code:
-                    print(f"[FOUND] {code}")
+                    print("[FOUND] %s" % code)
                     return
         if total == 0:
             print("[INFO] no mails yet")