Blame for src/services/temp_mail.py

"""
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