Spaces:

webhook
/

codex

Sleeping

Upload 137 files

89419ea verified 7 months ago

28.4 kB

"""
Temp-Mail 邮箱服务实现
基于自部署 Cloudflare Worker 临时邮箱服务
接口文档参见 plan/temp-mail.md
"""

import re
import time
import json
import logging
from datetime import datetime
from email import message_from_string
from email.header import decode_header, make_header
from email.message import Message
from email.policy import default as email_policy
from html import unescape
from typing import Optional, Dict, Any, List

from .base import BaseEmailService, EmailServiceError, EmailServiceType
from ..core.http_client import HTTPClient, RequestConfig
from ..config.constants import OTP_CODE_PATTERN
from .temp_mail_addressing import generate_local_part, choose_domain
from .temp_mail_domain_provider import resolve_temp_mail_domains, summarize_temp_mail_domains


logger = logging.getLogger(__name__)

SEMANTIC_OTP_PATTERNS = (
    re.compile(r"(?:openai\s+verification\s+code\s+is|verification\s+code\s+is|code\s+is)\s*[:：]?\s*(\d{6})", re.I),
    re.compile(r"(?:openai\s+verification\s+code|verification\s+code|验证码)\s*[:：]?\s*(\d{6})", re.I),
)


class TempMailService(BaseEmailService):
    """
    Temp-Mail 邮箱服务
    基于自部署 Cloudflare Worker 的临时邮箱，admin 模式管理邮箱
    不走代理，不使用 requests 库
    """

    def __init__(self, config: Dict[str, Any] = None, name: str = None):
        """
        初始化 TempMail 服务
        Args:
            config: 配置字典，支持以下键:
                - base_url: Worker 域名地址，如 https://mail.example.com (必需)
                - admin_password: Admin 密码，对应 x-admin-auth header (必需)
                - domain: 邮箱域名，如 example.com（可选，作为 domains/worker 回退）
                - domains: 可选域名池，数组或逗号分隔字符串（优先级最高）
                - enable_prefix: 是否启用前缀，默认 False
                - timeout: 请求超时时间，默认 30
                - max_retries: 最大重试次数，默认 3
            name: 服务名称
        """
        super().__init__(EmailServiceType.TEMP_MAIL, name)

        required_keys = ["base_url", "admin_password"]
        missing_keys = [key for key in required_keys if not (config or {}).get(key)]
        if missing_keys:
            raise ValueError(f"缺少必需配置: {missing_keys}")

        default_config = {
            "enable_prefix": False,
            "timeout": 30,
            "max_retries": 3,
        }
        self.config = {**default_config, **(config or {})}

        # 不走代理，proxy_url=None
        http_config = RequestConfig(
            timeout=self.config["timeout"],
            max_retries=self.config["max_retries"],
        )
        self.http_client = HTTPClient(proxy_url=None, config=http_config)

        # 邮箱缓存：email -> {jwt, address}
        self._email_cache: Dict[str, Dict[str, Any]] = {}
        self._verification_state: Dict[str, Dict[str, Any]] = {}
        self._last_verification_debug: Dict[str, Dict[str, Any]] = {}

    def _decode_mime_header(self, value: str) -> str:
        """解码 MIME 头，兼容 RFC 2047 编码主题。"""
        if not value:
            return ""
        try:
            return str(make_header(decode_header(value)))
        except Exception:
            return value

    def _extract_body_from_message(self, message: Message) -> str:
        """从 MIME 邮件对象中提取可读正文。"""
        parts: List[str] = []

        if message.is_multipart():
            for part in message.walk():
                if part.get_content_maintype() == "multipart":
                    continue

                content_type = (part.get_content_type() or "").lower()
                if content_type not in ("text/plain", "text/html"):
                    continue

                try:
                    payload = part.get_payload(decode=True)
                    charset = part.get_content_charset() or "utf-8"
                    text = payload.decode(charset, errors="replace") if payload else ""
                except Exception:
                    try:
                        text = part.get_content()
                    except Exception:
                        text = ""

                if content_type == "text/html":
                    text = re.sub(r"<[^>]+>", " ", text)
                parts.append(text)
        else:
            try:
                payload = message.get_payload(decode=True)
                charset = message.get_content_charset() or "utf-8"
                body = payload.decode(charset, errors="replace") if payload else ""
            except Exception:
                try:
                    body = message.get_content()
                except Exception:
                    body = str(message.get_payload() or "")

            if "html" in (message.get_content_type() or "").lower():
                body = re.sub(r"<[^>]+>", " ", body)
            parts.append(body)

        return unescape("\n".join(part for part in parts if part).strip())

    def _extract_mail_fields(self, mail: Dict[str, Any]) -> Dict[str, str]:
        """统一提取邮件字段，兼容 raw MIME 和不同 Worker 返回格式。"""
        sender = str(
            mail.get("source")
            or mail.get("from")
            or mail.get("from_address")
            or mail.get("fromAddress")
            or ""
        ).strip()
        subject = str(mail.get("subject") or mail.get("title") or "").strip()
        body_text = str(
            mail.get("text")
            or mail.get("body")
            or mail.get("content")
            or mail.get("html")
            or ""
        ).strip()
        raw = str(mail.get("raw") or "").strip()

        if raw:
            try:
                message = message_from_string(raw, policy=email_policy)
                sender = sender or self._decode_mime_header(message.get("From", ""))
                subject = subject or self._decode_mime_header(message.get("Subject", ""))
                parsed_body = self._extract_body_from_message(message)
                if parsed_body:
                    body_text = f"{body_text}\n{parsed_body}".strip() if body_text else parsed_body
            except Exception as e:
                logger.debug(f"解析 TempMail raw 邮件失败: {e}")
                body_text = f"{body_text}\n{raw}".strip() if body_text else raw

        body_text = unescape(re.sub(r"<[^>]+>", " ", body_text))
        return {
            "sender": sender,
            "subject": subject,
            "body": body_text,
            "raw": raw,
        }

    def _extract_body_from_raw_mail(self, raw: str) -> str:
        """Extract a readable body from raw MIME content without searching header noise."""
        if not raw:
            return ""

        try:
            message = message_from_string(raw, policy=email_policy)
            return self._extract_body_from_message(message)
        except Exception as e:
            logger.debug(f"解析 TempMail raw 正文失败: {e}")
            return ""

    @staticmethod
    def _find_code_in_text(text: str, *, semantic_first: bool = True, pattern: str = OTP_CODE_PATTERN) -> Optional[str]:
        normalized = unescape(re.sub(r"<[^>]+>", " ", str(text or ""))).strip()
        if not normalized:
            return None

        if semantic_first:
            for candidate_pattern in SEMANTIC_OTP_PATTERNS:
                match = candidate_pattern.search(normalized)
                if match:
                    return match.group(1)

        match = re.search(pattern, normalized)
        return match.group(1) if match else None

    def _extract_verification_code_from_mail(
        self,
        parsed_mail: Dict[str, str],
        pattern: str = OTP_CODE_PATTERN,
    ) -> Optional[str]:
        """Prefer body-derived OTPs and avoid matching digits from headers/domain names."""
        body_text = str(parsed_mail.get("body") or "").strip()
        subject = str(parsed_mail.get("subject") or "").strip()
        raw_text = str(parsed_mail.get("raw") or "").strip()
        raw_body_text = self._extract_body_from_raw_mail(raw_text)

        for text in (body_text, raw_body_text, subject):
            code = self._find_code_in_text(text, semantic_first=True, pattern=pattern)
            if code:
                return code

        for text in (body_text, raw_body_text):
            code = self._find_code_in_text(text, semantic_first=False, pattern=pattern)
            if code:
                return code

        return self._find_code_in_text(subject, semantic_first=False, pattern=pattern)

    def _get_verification_state(self, email: str) -> Dict[str, Any]:
        state = self._verification_state.setdefault(
            email,
            {
                "stage": "signup_otp",
                "consumed_mail_ids": set(),
                "last_code": None,
                "last_mail_id": None,
                "last_mail_timestamp": None,
                "last_stage": None,
            },
        )
        consumed_mail_ids = state.get("consumed_mail_ids")
        if not isinstance(consumed_mail_ids, set):
            state["consumed_mail_ids"] = set(consumed_mail_ids or [])
        return state

    def set_verification_stage(self, email: str, stage: str) -> None:
        state = self._get_verification_state(email)
        state["stage"] = stage or "signup_otp"

    @staticmethod
    def _extract_mail_timestamp(mail: Dict[str, Any]) -> Optional[float]:
        for key in ("createdAt", "created_at", "receivedAt", "received_at", "timestamp", "date"):
            value = mail.get(key)
            if value in (None, ""):
                continue

            if isinstance(value, (int, float)):
                numeric = float(value)
                return numeric / 1000 if numeric > 1_000_000_000_000 else numeric

            if isinstance(value, str):
                stripped = value.strip()
                if not stripped:
                    continue
                if stripped.isdigit():
                    numeric = float(stripped)
                    return numeric / 1000 if numeric > 1_000_000_000_000 else numeric
                try:
                    return datetime.fromisoformat(stripped.replace("Z", "+00:00")).timestamp()
                except ValueError:
                    continue

        return None

    def _admin_headers(self) -> Dict[str, str]:
        """构造 admin 请求头"""
        return {
            "x-admin-auth": self.config["admin_password"],
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    @staticmethod
    def _address_jwt_headers(jwt: str) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {jwt}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    def _fetch_domains_from_worker(self) -> Dict[str, Any]:
        """Fetch temp-mail domains from worker admin endpoint."""
        return self._make_request("GET", "/admin/domains")

    def get_domain_summary(self, preview_limit: int = 3) -> Dict[str, Any]:
        """Return safe domain summary for API/capabilities output."""
        summary = summarize_temp_mail_domains(
            self.config,
            fetch_domains=self._fetch_domains_from_worker,
            preview_limit=preview_limit,
        )
        summary.pop("domains", None)
        return summary

    def _make_request(self, method: str, path: str, **kwargs) -> Any:
        """
        发送请求并返回 JSON 数据
        Args:
            method: HTTP 方法
            path: 请求路径（以 / 开头）
            **kwargs: 传递给 http_client.request 的额外参数
        Returns:
            响应 JSON 数据
        Raises:
            EmailServiceError: 请求失败
        """
        base_url = self.config["base_url"].rstrip("/")
        url = f"{base_url}{path}"

        # 合并默认 admin headers
        kwargs.setdefault("headers", {})
        for k, v in self._admin_headers().items():
            kwargs["headers"].setdefault(k, v)

        try:
            response = self.http_client.request(method, url, **kwargs)

            if response.status_code >= 400:
                error_msg = f"请求失败: {response.status_code}"
                try:
                    error_data = response.json()
                    error_msg = f"{error_msg} - {error_data}"
                except Exception:
                    error_msg = f"{error_msg} - {response.text[:200]}"
                self.update_status(False, EmailServiceError(error_msg))
                raise EmailServiceError(error_msg)

            try:
                return response.json()
            except json.JSONDecodeError:
                return {"raw_response": response.text}

        except Exception as e:
            self.update_status(False, e)
            if isinstance(e, EmailServiceError):
                raise
            raise EmailServiceError(f"请求失败: {method} {path} - {e}")

    def create_email(self, config: Dict[str, Any] = None) -> Dict[str, Any]:
        """
        通过 admin API 创建临时邮箱
        Returns:
            包含邮箱信息的字典:
            - email: 邮箱地址
            - jwt: 用户级 JWT token
            - service_id: 同 email（用作标识）
        """
        enable_prefix = False  # 强制禁用前缀，忽略数据库配置
        name = generate_local_part(enable_prefix=enable_prefix)
        domains = resolve_temp_mail_domains(
            self.config,
            fetch_domains=self._fetch_domains_from_worker,
        )
        if not domains:
            raise EmailServiceError("TempMail 未找到可用域名，请检查 domains/domain 或 /admin/domains")
        domain = choose_domain(domains)

        body = {
            "enablePrefix": enable_prefix,
            "name": name,
            "domain": domain,
        }

        try:
            response = self._make_request("POST", "/admin/new_address", json=body)

            address = response.get("address", "").strip()
            jwt = response.get("jwt", "").strip()

            if not address:
                raise EmailServiceError(f"API 返回数据不完整: {response}")

            email_info = {
                "email": address,
                "jwt": jwt,
                "service_id": address,
                "id": address,
                "created_at": time.time(),
            }

            # 缓存 jwt，供获取验证码时使用
            self._email_cache[address] = email_info

            logger.info(f"成功创建 TempMail 邮箱: {address}")
            self.update_status(True)
            return email_info

        except Exception as e:
            self.update_status(False, e)
            if isinstance(e, EmailServiceError):
                raise
            raise EmailServiceError(f"创建邮箱失败: {e}")

    def _fetch_api_mails(self, jwt: str) -> Dict[str, Any]:
        """Fetch mailbox contents through the worker address-scoped API."""
        return self._make_request(
            "GET",
            "/api/mails",
            params={"limit": 20, "offset": 0},
            headers=self._address_jwt_headers(jwt),
        )

    def _fetch_admin_mails(self, email: str) -> Dict[str, Any]:
        """Fetch mailbox contents through the worker admin API."""
        return self._make_request(
            "GET",
            "/admin/mails",
            params={"limit": 20, "offset": 0, "address": email},
        )

    def _fetch_mail_batch(
        self,
        email: str,
        jwt: Optional[str],
    ) -> tuple[List[Dict[str, Any]], str, Optional[str]]:
        """Prefer address-scoped API mails, then fall back to admin mails."""
        last_error: Optional[str] = None

        if jwt:
            try:
                response = self._fetch_api_mails(jwt)
                mails = response.get("results", response if isinstance(response, list) else [])
                if isinstance(mails, list):
                    return mails, "/api/mails", None
                last_error = f"/api/mails returned unexpected payload: {response}"
            except Exception as e:
                last_error = str(e)
                cached = {**self._email_cache.get(email, {}), "disable_user_api": True, "disable_api_mails": True}
                self._email_cache[email] = cached
                logger.warning(f"TempMail /api/mails 失败，切换到 admin 拉取邮件: {email} - {e}")

        try:
            response = self._fetch_admin_mails(email)
            mails = response.get("results", response if isinstance(response, list) else [])
            if isinstance(mails, list):
                return mails, "/admin/mails", last_error
            if last_error:
                last_error = f"{last_error}; /admin/mails returned unexpected payload: {response}"
            else:
                last_error = f"/admin/mails returned unexpected payload: {response}"
        except Exception as e:
            admin_error = str(e)
            if last_error:
                last_error = f"{last_error}; {admin_error}"
            else:
                last_error = admin_error

        return [], "unavailable", last_error

    def get_verification_code(
        self,
        email: str,
        email_id: str = None,
        timeout: int = 120,
        pattern: str = OTP_CODE_PATTERN,
        otp_sent_at: Optional[float] = None,
    ) -> Optional[str]:
        """Fetch a verification code from TempMail."""
        logger.info(f"??? TempMail ?? {email} ?????...")

        start_time = time.time()
        seen_mail_ids: set = set()
        poll_count = 0
        last_error: Optional[str] = None
        verification_state = self._get_verification_state(email)
        stage = str(verification_state.get("stage") or "signup_otp")
        consumed_mail_ids = verification_state.setdefault("consumed_mail_ids", set())
        last_code = verification_state.get("last_code")
        last_stage = verification_state.get("last_stage")
        min_timestamp = float(otp_sent_at or 0)
        debug_state = {
            "stage": stage,
            "poll_count": 0,
            "last_status": "waiting",
            "otp_sent_at": otp_sent_at or 0,
            "min_timestamp": min_timestamp,
            "fresh_verification_count": 0,
            "fresh_preferred_sender_count": 0,
            "stale_preferred_sender_count": 0,
            "available_fresh_verification_count": 0,
            "available_fresh_preferred_sender_count": 0,
            "used_fresh_preferred_sender_count": 0,
            "selected_sender": None,
            "selected_code": None,
            "selected_received_timestamp": None,
            "deferred_generic_only_polls": 0,
            "candidate_summaries": [],
        }
        self._last_verification_debug[email.lower()] = debug_state

        cached = self._email_cache.get(email, {})
        jwt = None if cached.get("disable_user_api") or cached.get("disable_api_mails") else cached.get("jwt")

        while time.time() - start_time < timeout:
            poll_count += 1
            debug_state["poll_count"] = poll_count
            try:
                mails, source_path, request_error = self._fetch_mail_batch(email, jwt)
                if request_error:
                    last_error = request_error
                if not isinstance(mails, list):
                    time.sleep(3)
                    continue

                matched_candidates: List[Dict[str, Any]] = []
                debug_candidates: List[Dict[str, Any]] = []

                for index, mail in enumerate(mails):
                    mail_id = mail.get("id")
                    if not mail_id or mail_id in seen_mail_ids:
                        continue

                    seen_mail_ids.add(mail_id)
                    if mail_id in consumed_mail_ids:
                        continue

                    parsed = self._extract_mail_fields(mail)
                    sender = parsed["sender"].lower()
                    subject = parsed["subject"]
                    body_text = parsed["body"]
                    content = f"{sender}\n{subject}\n{body_text}".strip()

                    if "openai" not in sender and "openai" not in content.lower():
                        continue

                    code = self._extract_verification_code_from_mail(parsed, pattern=pattern)
                    if not code:
                        continue

                    mail_ts = self._extract_mail_timestamp(mail)
                    is_fresh = otp_sent_at is None or (mail_ts is not None and mail_ts >= otp_sent_at)
                    freshness_rank = 1 if otp_sent_at is None else 0
                    if otp_sent_at is not None:
                        if mail_ts is not None:
                            if mail_ts < otp_sent_at:
                                debug_state["stale_preferred_sender_count"] += 1
                                continue
                            freshness_rank = 2
                            debug_state["available_fresh_verification_count"] += 1
                            debug_state["available_fresh_preferred_sender_count"] += 1
                            debug_state["fresh_preferred_sender_count"] += 1
                        elif stage == "relogin_otp":
                            debug_state["deferred_generic_only_polls"] += 1
                            continue

                    if stage == "relogin_otp" and code == last_code and last_stage == "signup_otp":
                        continue

                    debug_state["fresh_verification_count"] += int(is_fresh)
                    debug_candidates.append(
                        {
                            "sender": parsed["sender"] or "-",
                            "received_timestamp": mail_ts,
                            "delta_from_otp_sent": None if otp_sent_at is None or mail_ts is None else round(mail_ts - otp_sent_at, 3),
                            "code": code,
                            "preferred": bool(is_fresh),
                        }
                    )
                    matched_candidates.append(
                        {
                            "code": code,
                            "mail_id": mail_id,
                            "sender": parsed["sender"] or "-",
                            "timestamp": mail_ts if mail_ts is not None else float("-inf"),
                            "freshness_rank": freshness_rank,
                            "index": index,
                        }
                    )

                if debug_candidates:
                    debug_state["candidate_summaries"] = debug_candidates[:5]

                if matched_candidates:
                    matched_candidates.sort(
                        key=lambda item: (item["freshness_rank"], item["timestamp"], item["index"]),
                        reverse=True,
                    )
                    selected = matched_candidates[0]
                    code = selected["code"]
                    selected_mail_id = selected["mail_id"]
                    consumed_mail_ids.add(selected_mail_id)
                    verification_state["last_code"] = code
                    verification_state["last_mail_id"] = selected_mail_id
                    verification_state["last_mail_timestamp"] = (
                        None if selected["timestamp"] == float("-inf") else selected["timestamp"]
                    )
                    verification_state["last_stage"] = stage
                    debug_state["last_status"] = "matched"
                    debug_state["selected_sender"] = selected["sender"]
                    debug_state["selected_code"] = code
                    debug_state["selected_received_timestamp"] = (
                        None if selected["timestamp"] == float("-inf") else selected["timestamp"]
                    )
                    if selected["freshness_rank"] >= 2:
                        debug_state["used_fresh_preferred_sender_count"] = 1
                    wait_duration = max(time.time() - start_time, 0.0)
                    logger.info(
                        f"TempMail verification code found for {email}: {code} "
                        f"(source={source_path}, stage={stage}, polls={poll_count}, wait={wait_duration:.2f}s)"
                    )
                    self.update_status(True)
                    return code

            except Exception as e:
                last_error = str(e)
                logger.debug(f"?? TempMail ?????: {e}")

            time.sleep(3)

        wait_duration = max(time.time() - start_time, 0.0)
        debug_state["last_status"] = "timeout"
        logger.warning(
            f"?? TempMail ?????: {email}; polls={poll_count}; wait={wait_duration:.2f}s; "
            f"last_error={last_error or '-'}"
        )
        return None

    def get_last_verification_debug(self, email: str) -> Dict[str, Any]:
        return dict(self._last_verification_debug.get(email.lower(), {}))

    def list_emails(self, limit: int = 100, offset: int = 0, **kwargs) -> List[Dict[str, Any]]:
        """
        列出邮箱
        Args:
            limit: 返回数量上限
            offset: 分页偏移
            **kwargs: 额外查询参数，透传给 admin API
        Returns:
            邮箱列表
        """
        params = {
            "limit": limit,
            "offset": offset,
        }
        params.update({k: v for k, v in kwargs.items() if v is not None})

        try:
            response = self._make_request("GET", "/admin/mails", params=params)
            mails = response.get("results", [])
            if not isinstance(mails, list):
                raise EmailServiceError(f"API 返回数据格式错误: {response}")

            emails: List[Dict[str, Any]] = []
            for mail in mails:
                address = (mail.get("address") or "").strip()
                mail_id = mail.get("id") or address
                email_info = {
                    "id": mail_id,
                    "service_id": mail_id,
                    "email": address,
                    "subject": mail.get("subject"),
                    "from": mail.get("source"),
                    "created_at": mail.get("createdAt") or mail.get("created_at"),
                    "raw_data": mail,
                }
                emails.append(email_info)

                if address:
                    cached = self._email_cache.get(address, {})
                    self._email_cache[address] = {**cached, **email_info}

            self.update_status(True)
            return emails
        except Exception as e:
            logger.warning(f"列出 TempMail 邮箱失败: {e}")
            self.update_status(False, e)
            return list(self._email_cache.values())

    def delete_email(self, email_id: str) -> bool:
        """
        删除邮箱
        Note:
            当前 TempMail admin API 文档未见删除地址接口，这里先从本地缓存移除，
            以满足统一接口并避免服务实例化失败。
        """
        removed = False
        emails_to_delete = []

        for address, info in self._email_cache.items():
            candidate_ids = {
                address,
                info.get("id"),
                info.get("service_id"),
            }
            if email_id in candidate_ids:
                emails_to_delete.append(address)

        for address in emails_to_delete:
            self._email_cache.pop(address, None)
            removed = True

        if removed:
            logger.info(f"已从 TempMail 缓存移除邮箱: {email_id}")
            self.update_status(True)
        else:
            logger.info(f"TempMail 缓存中未找到邮箱: {email_id}")

        return removed

    def check_health(self) -> bool:
        """检查服务健康状态"""
        try:
            self._make_request(
                "GET",
                "/admin/mails",
                params={"limit": 1, "offset": 0},
            )
            self.update_status(True)
            return True
        except Exception as e:
            logger.warning(f"TempMail 健康检查失败: {e}")
            self.update_status(False, e)
            return False

	"""
	Temp-Mail 邮箱服务实现
	基于自部署 Cloudflare Worker 临时邮箱服务
	接口文档参见 plan/temp-mail.md
	"""

	import re
	import time
	import json
	import logging
	from datetime import datetime
	from email import message_from_string
	from email.header import decode_header, make_header
	from email.message import Message
	from email.policy import default as email_policy
	from html import unescape
	from typing import Optional, Dict, Any, List

	from .base import BaseEmailService, EmailServiceError, EmailServiceType
	from ..core.http_client import HTTPClient, RequestConfig
	from ..config.constants import OTP_CODE_PATTERN
	from .temp_mail_addressing import generate_local_part, choose_domain
	from .temp_mail_domain_provider import resolve_temp_mail_domains, summarize_temp_mail_domains


	logger = logging.getLogger(__name__)

	SEMANTIC_OTP_PATTERNS = (
	re.compile(r"(?:openai\s+verification\s+code\s+is\|verification\s+code\s+is\|code\s+is)\s[:：]?\s(\d{6})", re.I),
	re.compile(r"(?:openai\s+verification\s+code\|verification\s+code\|验证码)\s[:：]?\s(\d{6})", re.I),
	)


	class TempMailService(BaseEmailService):
	"""
	Temp-Mail 邮箱服务
	基于自部署 Cloudflare Worker 的临时邮箱，admin 模式管理邮箱
	不走代理，不使用 requests 库
	"""

	def __init__(self, config: Dict[str, Any] = None, name: str = None):
	"""
	初始化 TempMail 服务

	Args:
	config: 配置字典，支持以下键:
	- base_url: Worker 域名地址，如 https://mail.example.com (必需)
	- admin_password: Admin 密码，对应 x-admin-auth header (必需)
	- domain: 邮箱域名，如 example.com（可选，作为 domains/worker 回退）
	- domains: 可选域名池，数组或逗号分隔字符串（优先级最高）
	- enable_prefix: 是否启用前缀，默认 False
	- timeout: 请求超时时间，默认 30
	- max_retries: 最大重试次数，默认 3
	name: 服务名称
	"""
	super().__init__(EmailServiceType.TEMP_MAIL, name)

	required_keys = ["base_url", "admin_password"]
	missing_keys = [key for key in required_keys if not (config or {}).get(key)]
	if missing_keys:
	raise ValueError(f"缺少必需配置: {missing_keys}")

	default_config = {
	"enable_prefix": False,
	"timeout": 30,
	"max_retries": 3,
	}
	self.config = {default_config, (config or {})}

	# 不走代理，proxy_url=None
	http_config = RequestConfig(
	timeout=self.config["timeout"],
	max_retries=self.config["max_retries"],
	)
	self.http_client = HTTPClient(proxy_url=None, config=http_config)

	# 邮箱缓存：email -> {jwt, address}
	self._email_cache: Dict[str, Dict[str, Any]] = {}
	self._verification_state: Dict[str, Dict[str, Any]] = {}
	self._last_verification_debug: Dict[str, Dict[str, Any]] = {}

	def _decode_mime_header(self, value: str) -> str:
	"""解码 MIME 头，兼容 RFC 2047 编码主题。"""
	if not value:
	return ""
	try:
	return str(make_header(decode_header(value)))
	except Exception:
	return value

	def _extract_body_from_message(self, message: Message) -> str:
	"""从 MIME 邮件对象中提取可读正文。"""
	parts: List[str] = []

	if message.is_multipart():
	for part in message.walk():
	if part.get_content_maintype() == "multipart":
	continue

	content_type = (part.get_content_type() or "").lower()
	if content_type not in ("text/plain", "text/html"):
	continue

	try:
	payload = part.get_payload(decode=True)
	charset = part.get_content_charset() or "utf-8"
	text = payload.decode(charset, errors="replace") if payload else ""
	except Exception:
	try:
	text = part.get_content()
	except Exception:
	text = ""

	if content_type == "text/html":
	text = re.sub(r"<[^>]+>", " ", text)
	parts.append(text)
	else:
	try:
	payload = message.get_payload(decode=True)
	charset = message.get_content_charset() or "utf-8"
	body = payload.decode(charset, errors="replace") if payload else ""
	except Exception:
	try:
	body = message.get_content()
	except Exception:
	body = str(message.get_payload() or "")

	if "html" in (message.get_content_type() or "").lower():
	body = re.sub(r"<[^>]+>", " ", body)
	parts.append(body)

	return unescape("\n".join(part for part in parts if part).strip())

	def _extract_mail_fields(self, mail: Dict[str, Any]) -> Dict[str, str]:
	"""统一提取邮件字段，兼容 raw MIME 和不同 Worker 返回格式。"""
	sender = str(
	mail.get("source")
	or mail.get("from")
	or mail.get("from_address")
	or mail.get("fromAddress")
	or ""
	).strip()
	subject = str(mail.get("subject") or mail.get("title") or "").strip()
	body_text = str(
	mail.get("text")
	or mail.get("body")
	or mail.get("content")
	or mail.get("html")
	or ""
	).strip()
	raw = str(mail.get("raw") or "").strip()

	if raw:
	try:
	message = message_from_string(raw, policy=email_policy)
	sender = sender or self._decode_mime_header(message.get("From", ""))
	subject = subject or self._decode_mime_header(message.get("Subject", ""))
	parsed_body = self._extract_body_from_message(message)
	if parsed_body:
	body_text = f"{body_text}\n{parsed_body}".strip() if body_text else parsed_body
	except Exception as e:
	logger.debug(f"解析 TempMail raw 邮件失败: {e}")
	body_text = f"{body_text}\n{raw}".strip() if body_text else raw

	body_text = unescape(re.sub(r"<[^>]+>", " ", body_text))
	return {
	"sender": sender,
	"subject": subject,
	"body": body_text,
	"raw": raw,
	}

	def _extract_body_from_raw_mail(self, raw: str) -> str:
	"""Extract a readable body from raw MIME content without searching header noise."""
	if not raw:
	return ""

	try:
	message = message_from_string(raw, policy=email_policy)
	return self._extract_body_from_message(message)
	except Exception as e:
	logger.debug(f"解析 TempMail raw 正文失败: {e}")
	return ""

	@staticmethod
	def _find_code_in_text(text: str, *, semantic_first: bool = True, pattern: str = OTP_CODE_PATTERN) -> Optional[str]:
	normalized = unescape(re.sub(r"<[^>]+>", " ", str(text or ""))).strip()
	if not normalized:
	return None

	if semantic_first:
	for candidate_pattern in SEMANTIC_OTP_PATTERNS:
	match = candidate_pattern.search(normalized)
	if match:
	return match.group(1)

	match = re.search(pattern, normalized)
	return match.group(1) if match else None

	def _extract_verification_code_from_mail(
	self,
	parsed_mail: Dict[str, str],
	pattern: str = OTP_CODE_PATTERN,
	) -> Optional[str]:
	"""Prefer body-derived OTPs and avoid matching digits from headers/domain names."""
	body_text = str(parsed_mail.get("body") or "").strip()
	subject = str(parsed_mail.get("subject") or "").strip()
	raw_text = str(parsed_mail.get("raw") or "").strip()
	raw_body_text = self._extract_body_from_raw_mail(raw_text)

	for text in (body_text, raw_body_text, subject):
	code = self._find_code_in_text(text, semantic_first=True, pattern=pattern)
	if code:
	return code

	for text in (body_text, raw_body_text):
	code = self._find_code_in_text(text, semantic_first=False, pattern=pattern)
	if code:
	return code

	return self._find_code_in_text(subject, semantic_first=False, pattern=pattern)

	def _get_verification_state(self, email: str) -> Dict[str, Any]:
	state = self._verification_state.setdefault(
	email,
	{
	"stage": "signup_otp",
	"consumed_mail_ids": set(),
	"last_code": None,
	"last_mail_id": None,
	"last_mail_timestamp": None,
	"last_stage": None,
	},
	)
	consumed_mail_ids = state.get("consumed_mail_ids")
	if not isinstance(consumed_mail_ids, set):
	state["consumed_mail_ids"] = set(consumed_mail_ids or [])
	return state

	def set_verification_stage(self, email: str, stage: str) -> None:
	state = self._get_verification_state(email)
	state["stage"] = stage or "signup_otp"

	@staticmethod
	def _extract_mail_timestamp(mail: Dict[str, Any]) -> Optional[float]:
	for key in ("createdAt", "created_at", "receivedAt", "received_at", "timestamp", "date"):
	value = mail.get(key)
	if value in (None, ""):
	continue

	if isinstance(value, (int, float)):
	numeric = float(value)
	return numeric / 1000 if numeric > 1_000_000_000_000 else numeric

	if isinstance(value, str):
	stripped = value.strip()
	if not stripped:
	continue
	if stripped.isdigit():
	numeric = float(stripped)
	return numeric / 1000 if numeric > 1_000_000_000_000 else numeric
	try:
	return datetime.fromisoformat(stripped.replace("Z", "+00:00")).timestamp()
	except ValueError:
	continue

	return None

	def _admin_headers(self) -> Dict[str, str]:
	"""构造 admin 请求头"""
	return {
	"x-admin-auth": self.config["admin_password"],
	"Content-Type": "application/json",
	"Accept": "application/json",
	}

	@staticmethod
	def _address_jwt_headers(jwt: str) -> Dict[str, str]:
	return {
	"Authorization": f"Bearer {jwt}",
	"Content-Type": "application/json",
	"Accept": "application/json",
	}

	def _fetch_domains_from_worker(self) -> Dict[str, Any]:
	"""Fetch temp-mail domains from worker admin endpoint."""
	return self._make_request("GET", "/admin/domains")

	def get_domain_summary(self, preview_limit: int = 3) -> Dict[str, Any]:
	"""Return safe domain summary for API/capabilities output."""
	summary = summarize_temp_mail_domains(
	self.config,
	fetch_domains=self._fetch_domains_from_worker,
	preview_limit=preview_limit,
	)
	summary.pop("domains", None)
	return summary

	def _make_request(self, method: str, path: str, **kwargs) -> Any:
	"""
	发送请求并返回 JSON 数据

	Args:
	method: HTTP 方法
	path: 请求路径（以 / 开头）
	**kwargs: 传递给 http_client.request 的额外参数

	Returns:
	响应 JSON 数据

	Raises:
	EmailServiceError: 请求失败
	"""
	base_url = self.config["base_url"].rstrip("/")
	url = f"{base_url}{path}"

	# 合并默认 admin headers
	kwargs.setdefault("headers", {})
	for k, v in self._admin_headers().items():
	kwargs["headers"].setdefault(k, v)

	try:
	response = self.http_client.request(method, url, **kwargs)

	if response.status_code >= 400:
	error_msg = f"请求失败: {response.status_code}"
	try:
	error_data = response.json()
	error_msg = f"{error_msg} - {error_data}"
	except Exception:
	error_msg = f"{error_msg} - {response.text[:200]}"
	self.update_status(False, EmailServiceError(error_msg))
	raise EmailServiceError(error_msg)

	try:
	return response.json()
	except json.JSONDecodeError:
	return {"raw_response": response.text}

	except Exception as e:
	self.update_status(False, e)
	if isinstance(e, EmailServiceError):
	raise
	raise EmailServiceError(f"请求失败: {method} {path} - {e}")

	def create_email(self, config: Dict[str, Any] = None) -> Dict[str, Any]:
	"""
	通过 admin API 创建临时邮箱

	Returns:
	包含邮箱信息的字典:
	- email: 邮箱地址
	- jwt: 用户级 JWT token
	- service_id: 同 email（用作标识）
	"""
	enable_prefix = False # 强制禁用前缀，忽略数据库配置
	name = generate_local_part(enable_prefix=enable_prefix)
	domains = resolve_temp_mail_domains(
	self.config,
	fetch_domains=self._fetch_domains_from_worker,
	)
	if not domains:
	raise EmailServiceError("TempMail 未找到可用域名，请检查 domains/domain 或 /admin/domains")
	domain = choose_domain(domains)

	body = {
	"enablePrefix": enable_prefix,
	"name": name,
	"domain": domain,
	}

	try:
	response = self._make_request("POST", "/admin/new_address", json=body)

	address = response.get("address", "").strip()
	jwt = response.get("jwt", "").strip()

	if not address:
	raise EmailServiceError(f"API 返回数据不完整: {response}")

	email_info = {
	"email": address,
	"jwt": jwt,
	"service_id": address,
	"id": address,
	"created_at": time.time(),
	}

	# 缓存 jwt，供获取验证码时使用
	self._email_cache[address] = email_info

	logger.info(f"成功创建 TempMail 邮箱: {address}")
	self.update_status(True)
	return email_info

	except Exception as e:
	self.update_status(False, e)
	if isinstance(e, EmailServiceError):
	raise
	raise EmailServiceError(f"创建邮箱失败: {e}")

	def _fetch_api_mails(self, jwt: str) -> Dict[str, Any]:
	"""Fetch mailbox contents through the worker address-scoped API."""
	return self._make_request(
	"GET",
	"/api/mails",
	params={"limit": 20, "offset": 0},
	headers=self._address_jwt_headers(jwt),
	)

	def _fetch_admin_mails(self, email: str) -> Dict[str, Any]:
	"""Fetch mailbox contents through the worker admin API."""
	return self._make_request(
	"GET",
	"/admin/mails",
	params={"limit": 20, "offset": 0, "address": email},
	)

	def _fetch_mail_batch(
	self,
	email: str,
	jwt: Optional[str],
	) -> tuple[List[Dict[str, Any]], str, Optional[str]]:
	"""Prefer address-scoped API mails, then fall back to admin mails."""
	last_error: Optional[str] = None

	if jwt:
	try:
	response = self._fetch_api_mails(jwt)
	mails = response.get("results", response if isinstance(response, list) else [])
	if isinstance(mails, list):
	return mails, "/api/mails", None
	last_error = f"/api/mails returned unexpected payload: {response}"
	except Exception as e:
	last_error = str(e)
	cached = {**self._email_cache.get(email, {}), "disable_user_api": True, "disable_api_mails": True}
	self._email_cache[email] = cached
	logger.warning(f"TempMail /api/mails 失败，切换到 admin 拉取邮件: {email} - {e}")

	try:
	response = self._fetch_admin_mails(email)
	mails = response.get("results", response if isinstance(response, list) else [])
	if isinstance(mails, list):
	return mails, "/admin/mails", last_error
	if last_error:
	last_error = f"{last_error}; /admin/mails returned unexpected payload: {response}"
	else:
	last_error = f"/admin/mails returned unexpected payload: {response}"
	except Exception as e:
	admin_error = str(e)
	if last_error:
	last_error = f"{last_error}; {admin_error}"
	else:
	last_error = admin_error

	return [], "unavailable", last_error

	def get_verification_code(
	self,
	email: str,
	email_id: str = None,
	timeout: int = 120,
	pattern: str = OTP_CODE_PATTERN,
	otp_sent_at: Optional[float] = None,
	) -> Optional[str]:
	"""Fetch a verification code from TempMail."""
	logger.info(f"??? TempMail ?? {email} ?????...")

	start_time = time.time()
	seen_mail_ids: set = set()
	poll_count = 0
	last_error: Optional[str] = None
	verification_state = self._get_verification_state(email)
	stage = str(verification_state.get("stage") or "signup_otp")
	consumed_mail_ids = verification_state.setdefault("consumed_mail_ids", set())
	last_code = verification_state.get("last_code")
	last_stage = verification_state.get("last_stage")
	min_timestamp = float(otp_sent_at or 0)
	debug_state = {
	"stage": stage,
	"poll_count": 0,
	"last_status": "waiting",
	"otp_sent_at": otp_sent_at or 0,
	"min_timestamp": min_timestamp,
	"fresh_verification_count": 0,
	"fresh_preferred_sender_count": 0,
	"stale_preferred_sender_count": 0,
	"available_fresh_verification_count": 0,
	"available_fresh_preferred_sender_count": 0,
	"used_fresh_preferred_sender_count": 0,
	"selected_sender": None,
	"selected_code": None,
	"selected_received_timestamp": None,
	"deferred_generic_only_polls": 0,
	"candidate_summaries": [],
	}
	self._last_verification_debug[email.lower()] = debug_state

	cached = self._email_cache.get(email, {})
	jwt = None if cached.get("disable_user_api") or cached.get("disable_api_mails") else cached.get("jwt")

	while time.time() - start_time < timeout:
	poll_count += 1
	debug_state["poll_count"] = poll_count
	try:
	mails, source_path, request_error = self._fetch_mail_batch(email, jwt)
	if request_error:
	last_error = request_error
	if not isinstance(mails, list):
	time.sleep(3)
	continue

	matched_candidates: List[Dict[str, Any]] = []
	debug_candidates: List[Dict[str, Any]] = []

	for index, mail in enumerate(mails):
	mail_id = mail.get("id")
	if not mail_id or mail_id in seen_mail_ids:
	continue

	seen_mail_ids.add(mail_id)
	if mail_id in consumed_mail_ids:
	continue

	parsed = self._extract_mail_fields(mail)
	sender = parsed["sender"].lower()
	subject = parsed["subject"]
	body_text = parsed["body"]
	content = f"{sender}\n{subject}\n{body_text}".strip()

	if "openai" not in sender and "openai" not in content.lower():
	continue

	code = self._extract_verification_code_from_mail(parsed, pattern=pattern)
	if not code:
	continue

	mail_ts = self._extract_mail_timestamp(mail)
	is_fresh = otp_sent_at is None or (mail_ts is not None and mail_ts >= otp_sent_at)
	freshness_rank = 1 if otp_sent_at is None else 0
	if otp_sent_at is not None:
	if mail_ts is not None:
	if mail_ts < otp_sent_at:
	debug_state["stale_preferred_sender_count"] += 1
	continue
	freshness_rank = 2
	debug_state["available_fresh_verification_count"] += 1
	debug_state["available_fresh_preferred_sender_count"] += 1
	debug_state["fresh_preferred_sender_count"] += 1
	elif stage == "relogin_otp":
	debug_state["deferred_generic_only_polls"] += 1
	continue

	if stage == "relogin_otp" and code == last_code and last_stage == "signup_otp":
	continue

	debug_state["fresh_verification_count"] += int(is_fresh)
	debug_candidates.append(
	{
	"sender": parsed["sender"] or "-",
	"received_timestamp": mail_ts,
	"delta_from_otp_sent": None if otp_sent_at is None or mail_ts is None else round(mail_ts - otp_sent_at, 3),
	"code": code,
	"preferred": bool(is_fresh),
	}
	)
	matched_candidates.append(
	{
	"code": code,
	"mail_id": mail_id,
	"sender": parsed["sender"] or "-",
	"timestamp": mail_ts if mail_ts is not None else float("-inf"),
	"freshness_rank": freshness_rank,
	"index": index,
	}
	)

	if debug_candidates:
	debug_state["candidate_summaries"] = debug_candidates[:5]

	if matched_candidates:
	matched_candidates.sort(
	key=lambda item: (item["freshness_rank"], item["timestamp"], item["index"]),
	reverse=True,
	)
	selected = matched_candidates[0]
	code = selected["code"]
	selected_mail_id = selected["mail_id"]
	consumed_mail_ids.add(selected_mail_id)
	verification_state["last_code"] = code
	verification_state["last_mail_id"] = selected_mail_id
	verification_state["last_mail_timestamp"] = (
	None if selected["timestamp"] == float("-inf") else selected["timestamp"]
	)
	verification_state["last_stage"] = stage
	debug_state["last_status"] = "matched"
	debug_state["selected_sender"] = selected["sender"]
	debug_state["selected_code"] = code
	debug_state["selected_received_timestamp"] = (
	None if selected["timestamp"] == float("-inf") else selected["timestamp"]
	)
	if selected["freshness_rank"] >= 2:
	debug_state["used_fresh_preferred_sender_count"] = 1
	wait_duration = max(time.time() - start_time, 0.0)
	logger.info(
	f"TempMail verification code found for {email}: {code} "
	f"(source={source_path}, stage={stage}, polls={poll_count}, wait={wait_duration:.2f}s)"
	)
	self.update_status(True)
	return code

	except Exception as e:
	last_error = str(e)
	logger.debug(f"?? TempMail ?????: {e}")

	time.sleep(3)

	wait_duration = max(time.time() - start_time, 0.0)
	debug_state["last_status"] = "timeout"
	logger.warning(
	f"?? TempMail ?????: {email}; polls={poll_count}; wait={wait_duration:.2f}s; "
	f"last_error={last_error or '-'}"
	)
	return None

	def get_last_verification_debug(self, email: str) -> Dict[str, Any]:
	return dict(self._last_verification_debug.get(email.lower(), {}))

	def list_emails(self, limit: int = 100, offset: int = 0, **kwargs) -> List[Dict[str, Any]]:
	"""
	列出邮箱

	Args:
	limit: 返回数量上限
	offset: 分页偏移
	**kwargs: 额外查询参数，透传给 admin API

	Returns:
	邮箱列表
	"""
	params = {
	"limit": limit,
	"offset": offset,
	}
	params.update({k: v for k, v in kwargs.items() if v is not None})

	try:
	response = self._make_request("GET", "/admin/mails", params=params)
	mails = response.get("results", [])
	if not isinstance(mails, list):
	raise EmailServiceError(f"API 返回数据格式错误: {response}")

	emails: List[Dict[str, Any]] = []
	for mail in mails:
	address = (mail.get("address") or "").strip()
	mail_id = mail.get("id") or address
	email_info = {
	"id": mail_id,
	"service_id": mail_id,
	"email": address,
	"subject": mail.get("subject"),
	"from": mail.get("source"),
	"created_at": mail.get("createdAt") or mail.get("created_at"),
	"raw_data": mail,
	}
	emails.append(email_info)

	if address:
	cached = self._email_cache.get(address, {})
	self._email_cache[address] = {cached, email_info}

	self.update_status(True)
	return emails
	except Exception as e:
	logger.warning(f"列出 TempMail 邮箱失败: {e}")
	self.update_status(False, e)
	return list(self._email_cache.values())

	def delete_email(self, email_id: str) -> bool:
	"""
	删除邮箱

	Note:
	当前 TempMail admin API 文档未见删除地址接口，这里先从本地缓存移除，
	以满足统一接口并避免服务实例化失败。
	"""
	removed = False
	emails_to_delete = []

	for address, info in self._email_cache.items():
	candidate_ids = {
	address,
	info.get("id"),
	info.get("service_id"),
	}
	if email_id in candidate_ids:
	emails_to_delete.append(address)

	for address in emails_to_delete:
	self._email_cache.pop(address, None)
	removed = True

	if removed:
	logger.info(f"已从 TempMail 缓存移除邮箱: {email_id}")
	self.update_status(True)
	else:
	logger.info(f"TempMail 缓存中未找到邮箱: {email_id}")

	return removed

	def check_health(self) -> bool:
	"""检查服务健康状态"""
	try:
	self._make_request(
	"GET",
	"/admin/mails",
	params={"limit": 1, "offset": 0},
	)
	self.update_status(True)
	return True
	except Exception as e:
	logger.warning(f"TempMail 健康检查失败: {e}")
	self.update_status(False, e)
	return False