Browser_Agent/workflows/temp_mail.py

# workflows/temp_mail.py
"""
Universeller Temp-Mail Service mit Multi-Provider Fallback (mail.tm & 1secmail).
Bietet asynchrone Postfächer für automatische Registrierungen und OTP/Link-Parsing.
"""

import asyncio
import logging
import random
import re
import string
import httpx
from typing import Optional, Tuple

logger = logging.getLogger(__name__)

MAIL_TM_BASE = "https://api.mail.tm"
ONESECMAIL_BASE = "https://www.1secmail.com/api/v1/"


def generate_random_username(prefix: str = "agent") -> str:
    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
    return f"{prefix}_{suffix}"


def generate_random_password(length: int = 16) -> str:
    chars = string.ascii_letters + string.digits + "!@#$%^&*"
    return "".join(random.choices(chars, k=length))


class TempMailAccount:
    def __init__(self, email: str, password: str, provider: str, token: str = "", account_id: str = ""):
        self.email = email
        self.password = password
        self.provider = provider
        self.token = token
        self.account_id = account_id
        self.username, self.domain = email.split("@") if "@" in email else ("", "")


class TempMailService:
    """
    Verwaltet temporäre E-Mail-Postfächer mit automatischem Polling nach OTP-Codes und Links.
    """

    def __init__(self, timeout: float = 30.0):
        self._http = httpx.AsyncClient(timeout=timeout)

    async def aclose(self):
        await self._http.aclose()

    async def create_inbox(self) -> TempMailAccount:
        """
        Erstellt ein temporäres Postfach. Versucht zuerst mail.tm, fällt bei Bedarf auf 1secmail zurück.
        """
        try:
            return await self._create_mail_tm_inbox()
        except Exception as e:
            logger.warning(f"mail.tm fehlgeschlagen ({e}), wechsle zu 1secmail...")
            return await self._create_1secmail_inbox()

    async def _create_mail_tm_inbox(self) -> TempMailAccount:
        # 1. Domains holen
        domains_resp = await self._http.get(f"{MAIL_TM_BASE}/domains")
        domains_resp.raise_for_status()
        domains = domains_resp.json().get("hydra:member", [])
        if not domains:
            raise RuntimeError("Keine mail.tm Domains verfügbar")

        domain = domains[0]["domain"]
        username = generate_random_username()
        password = generate_random_password()
        email = f"{username}@{domain}"

        # 2. Account registrieren
        reg_resp = await self._http.post(
            f"{MAIL_TM_BASE}/accounts",
            json={"address": email, "password": password},
        )
        reg_resp.raise_for_status()
        account_id = reg_resp.json().get("id", "")

        # 3. JWT Token holen
        token_resp = await self._http.post(
            f"{MAIL_TM_BASE}/token",
            json={"address": email, "password": password},
        )
        token_resp.raise_for_status()
        token = token_resp.json().get("token", "")

        return TempMailAccount(
            email=email,
            password=password,
            provider="mail.tm",
            token=token,
            account_id=account_id,
        )

    async def _create_1secmail_inbox(self) -> TempMailAccount:
        resp = await self._http.get(f"{ONESECMAIL_BASE}?action=genRandomMailbox&count=1")
        resp.raise_for_status()
        emails = resp.json()
        if not emails:
            raise RuntimeError("Keine 1secmail Postfächer verfügbar")

        email = emails[0]
        password = generate_random_password()
        return TempMailAccount(
            email=email,
            password=password,
            provider="1secmail",
        )

    async def poll_for_otp_code(
        self,
        account: TempMailAccount,
        pattern: str = r"\b\d{4,8}\b",
        max_wait_seconds: int = 60,
        poll_interval: float = 3.0,
    ) -> Optional[str]:
        """
        Wartet auf eine eingehende E-Mail und extrahiert den ersten passenden OTP-Verifizierungscode.
        """
        elapsed = 0.0
        while elapsed < max_wait_seconds:
            messages = await self._fetch_messages(account)
            for msg in messages:
                content = await self._fetch_message_content(account, msg)
                match = re.search(pattern, content)
                if match:
                    code = match.group(0)
                    logger.info(f"OTP-Code erfolgreich gefunden: {code}")
                    return code
            await asyncio.sleep(poll_interval)
            elapsed += poll_interval

        logger.warning(f"Kein OTP-Code nach {max_wait_seconds}s empfangen.")
        return None

    async def poll_for_activation_link(
        self,
        account: TempMailAccount,
        url_pattern: str = r"https?://[^\s\"'>]+(?:verify|confirm|activate|token|auth)[^\s\"'>]*",
        max_wait_seconds: int = 60,
        poll_interval: float = 3.0,
    ) -> Optional[str]:
        """
        Wartet auf eine E-Mail und extrahiert den Aktivierungs-/Bestätigungslink.
        """
        elapsed = 0.0
        while elapsed < max_wait_seconds:
            messages = await self._fetch_messages(account)
            for msg in messages:
                content = await self._fetch_message_content(account, msg)
                match = re.search(url_pattern, content, re.IGNORECASE)
                if match:
                    link = match.group(0)
                    logger.info(f"Aktivierungslink gefunden: {link}")
                    return link
            await asyncio.sleep(poll_interval)
            elapsed += poll_interval

        logger.warning(f"Kein Aktivierungslink nach {max_wait_seconds}s empfangen.")
        return None

    async def _fetch_messages(self, account: TempMailAccount) -> list:
        try:
            if account.provider == "mail.tm":
                headers = {"Authorization": f"Bearer {account.token}"}
                resp = await self._http.get(f"{MAIL_TM_BASE}/messages", headers=headers)
                if resp.status_code == 200:
                    return resp.json().get("hydra:member", [])
            elif account.provider == "1secmail":
                url = f"{ONESECMAIL_BASE}?action=getMessages&login={account.username}&domain={account.domain}"
                resp = await self._http.get(url)
                if resp.status_code == 200:
                    return resp.json()
        except Exception as e:
            logger.debug(f"Fehler beim Abrufen der Mails: {e}")
        return []

    async def _fetch_message_content(self, account: TempMailAccount, msg: dict) -> str:
        try:
            if account.provider == "mail.tm":
                msg_id = msg.get("id")
                headers = {"Authorization": f"Bearer {account.token}"}
                resp = await self._http.get(f"{MAIL_TM_BASE}/messages/{msg_id}", headers=headers)
                if resp.status_code == 200:
                    data = resp.json()
                    return data.get("text", "") or data.get("intro", "") or str(data.get("html", ""))
            elif account.provider == "1secmail":
                msg_id = msg.get("id")
                url = f"{ONESECMAIL_BASE}?action=readMessage&login={account.username}&domain={account.domain}&id={msg_id}"
                resp = await self._http.get(url)
                if resp.status_code == 200:
                    data = resp.json()
                    return data.get("textBody", "") or data.get("body", "")
        except Exception as e:
            logger.debug(f"Fehler beim Lesen des E-Mail-Inhalts: {e}")
        return ""