ibo/mailtm.py

#!/usr/bin/env python3
"""
mail.tm  -  Wegwerf-E-Mail komplett per API, ohne Einrichtung/Konto.

Nur Python-Bordmittel (urllib). Ablauf:
    addr, pw, token = create_account()   # zufaellige Adresse anlegen
    ... Adresse ins Kontaktformular eintragen ...
    link = wait_for_message(token, extractor, log)   # auf Mail warten

API-Doku: https://docs.mail.tm
"""

import json
import time
import random
import string
import urllib.request
import urllib.error

BASE = "https://api.mail.tm"


def _req(method, path, data=None, token=None):
    url = BASE + path
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if token:
        headers["Authorization"] = "Bearer " + token
    body = json.dumps(data).encode() if data is not None else None
    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=20) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        detail = e.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"mail.tm {method} {path} -> HTTP {e.code}: {detail}")


def _rand(n):
    return "".join(random.choice(string.ascii_lowercase + string.digits)
                   for _ in range(n))


def _members(data):
    """Liste aus der Antwort holen - egal ob reine Liste oder Hydra-Format."""
    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        return data.get("hydra:member", data.get("member", []))
    return []


def get_domain():
    """Erste aktive mail.tm-Domain holen."""
    data = _req("GET", "/domains?page=1")
    domains = _members(data)
    for d in domains:
        if d.get("isActive", True):
            return d["domain"]
    raise RuntimeError("Keine aktive mail.tm-Domain verfuegbar.")


def create_account():
    """Legt eine zufaellige Wegwerf-Adresse an. Gibt (address, password, token)."""
    domain = get_domain()
    address = f"{_rand(10)}@{domain}"
    password = _rand(12)
    # Account anlegen (kann bei Rate-Limit 429 liefern -> kurz warten & retry)
    for attempt in range(3):
        try:
            _req("POST", "/accounts", {"address": address, "password": password})
            break
        except RuntimeError as e:
            if "429" in str(e) and attempt < 2:
                time.sleep(2)
                continue
            raise
    token = get_token(address, password)
    return address, password, token


def get_token(address, password):
    data = _req("POST", "/token", {"address": address, "password": password})
    return data["token"]


def list_messages(token):
    data = _req("GET", "/messages?page=1", token=token)
    return _members(data)


def get_message_html(token, message_id):
    """Volltext einer Nachricht als HTML (oder Text-Fallback)."""
    data = _req("GET", f"/messages/{message_id}", token=token)
    html = data.get("html")
    if isinstance(html, list):
        return "\n".join(html)
    if isinstance(html, str):
        return html
    return data.get("text", "")


def wait_for_message(token, extractor, log,
                     sender_hint="", timeout=300, poll=10, should_cancel=None):
    """
    Pollt das Postfach, bis eine Mail kommt, aus der `extractor(html)` einen
    Link liefert. `extractor` ist eine Funktion html->str|None.
    `should_cancel` (optional): Funktion ()->bool; liefert sie True, wird das
    Warten sofort abgebrochen (z.B. Item geloescht/Batch gestoppt).
    """
    def _cancelled_sleep(seconds):
        """Schlaeft in kleinen Schritten und bricht sofort ab, wenn should_cancel
        True liefert. Gibt True zurueck, wenn abgebrochen wurde."""
        slept = 0.0
        while slept < seconds:
            if should_cancel and should_cancel():
                return True
            time.sleep(0.5)
            slept += 0.5
        return False

    deadline = time.time() + timeout
    seen = set()
    while time.time() < deadline:
        if should_cancel and should_cancel():
            log("Warten abgebrochen.")
            return None
        try:
            msgs = list_messages(token)
        except RuntimeError as e:
            log(f"mail.tm Fehler: {e}")
            if _cancelled_sleep(poll):
                log("Warten abgebrochen.")
                return None
            continue

        for m in msgs:
            mid = m.get("id")
            if not mid or mid in seen:
                continue
            seen.add(mid)
            frm = ""
            if isinstance(m.get("from"), dict):
                frm = m["from"].get("address", "")
            if sender_hint and sender_hint.lower() not in frm.lower():
                continue
            html = get_message_html(token, mid)
            link = extractor(html)
            if link:
                log(f"Mail von {frm}: Link gefunden.")
                return link

        remaining = int(deadline - time.time())
        log(f"Noch keine passende Mail. Warte {poll}s (Rest {remaining}s) ...")
        if _cancelled_sleep(poll):
            log("Warten abgebrochen.")
            return None

    log("!! Timeout - keine Mail mit Link gefunden.")
    return None


if __name__ == "__main__":
    # Kleiner Selbsttest: Adresse anlegen und Posteingang einmal lesen.
    a, p, t = create_account()
    print("Wegwerf-Adresse:", a)
    print("Passwort:", p)
    print("Nachrichten aktuell:", len(list_messages(t)))
    print("Schick eine Mail an obige Adresse und starte erneut zum Testen.")