""" Official ready-to-use DraxonMails Python integration using `wreq`. Install: pip install wreq Main classes: - DraxonMailsClient - DraxonMailsDiscordVerifier """ from __future__ import annotations import json import re import time from dataclasses import dataclass from typing import Any from wreq import Emulation from wreq.blocking import Client class DraxonMailsError(Exception): pass class DraxonMailsApiError(DraxonMailsError): def __init__(self, status_code: int, payload: Any, message: str): self.status_code = status_code self.payload = payload super().__init__(message) class DraxonMailsNoDiscordLinkError(DraxonMailsError): pass @dataclass(slots=True) class DraxonMailsConfig: base_url: str = "https://mail.draxono.in" api_prefix: str = "/api/v1" api_key: str = "" emulate: Any = Emulation.Chrome137 user_agent: str = "DraxonMails-Wreq-Client/1.1" class DraxonMailsClient: def __init__( self, base_url: str = "https://mail.draxono.in", api_key: str = "", *, api_prefix: str = "/api/v1", emulate: Any = Emulation.Chrome137, user_agent: str = "DraxonMails-Wreq-Client/1.1", ) -> None: self.config = DraxonMailsConfig( base_url=base_url.rstrip("/"), api_prefix=api_prefix if api_prefix.startswith("/") else f"/{api_prefix}", api_key=(api_key or "").strip(), emulate=emulate, user_agent=user_agent, ) self._client = Client( emulation=self.config.emulate, user_agent=self.config.user_agent, cookie_store=True, ) def close(self) -> None: close = getattr(self._client, "close", None) if callable(close): close() def __enter__(self) -> "DraxonMailsClient": return self def __exit__(self, exc_type, exc, tb) -> None: self.close() @property def base_api_url(self) -> str: return f"{self.config.base_url}{self.config.api_prefix}" def set_api_key(self, api_key: str) -> None: self.config.api_key = (api_key or "").strip() def random_address(self) -> dict[str, Any]: return self._request_json("GET", "/random-address") def domains(self) -> dict[str, Any]: return self._request_json("GET", "/domains") def public_domain_status(self) -> dict[str, Any]: return self._request_json("GET", "/public/domains/status") def public_settings(self) -> dict[str, Any]: return self._request_json("GET", "/public/settings") def session_inboxes(self) -> list[dict[str, Any]]: return self._request_json("GET", "/inboxes") def get_inbox( self, address: str, *, password: str = "", domain_secret: str = "", extra_headers: dict[str, str] | None = None, ) -> list[dict[str, Any]]: headers = self._headers(domain_secret=domain_secret, extra_headers=extra_headers) if password: return self._request_json("POST", f"/inbox/{address}", json_body={"password": password}, headers=headers) return self._request_json("GET", f"/inbox/{address}", headers=headers) def claim_inbox(self, address: str, password: str, *, extra_headers: dict[str, str] | None = None) -> dict[str, Any]: return self._request_json( "POST", f"/inbox/{address}/claim", json_body={"email": address, "password": password}, headers=self._headers(extra_headers=extra_headers), ) def client_inbox(self, email: str, password: str = "", *, extra_headers: dict[str, str] | None = None) -> dict[str, Any]: return self._request_json( "POST", "/client/inbox", json_body={"email": email, "password": password}, headers=self._headers(extra_headers=extra_headers), ) def _headers(self, *, domain_secret: str = "", extra_headers: dict[str, str] | None = None) -> dict[str, str]: headers = {"Accept": "application/json", "Content-Type": "application/json"} if self.config.api_key: headers["X-Api-Key"] = self.config.api_key if domain_secret: headers["X-Draxon-Domain-Secret"] = domain_secret if extra_headers: headers.update({str(k): str(v) for k, v in extra_headers.items()}) return headers def _request_json(self, method: str, path: str, *, json_body: dict[str, Any] | None = None, headers: dict[str, str] | None = None) -> Any: method = method.upper().strip() url = f"{self.base_api_url}{path if path.startswith('/') else '/' + path}" kwargs: dict[str, Any] = {"headers": headers or self._headers()} if json_body is not None: kwargs["json"] = json_body response = self._dispatch(method, url, kwargs) data = self._read_json(response) status_code = int(getattr(response, "status_code", 0) or 0) if 200 <= status_code < 300: return data raise DraxonMailsApiError(status_code, data, self._extract_error_message(data, status_code)) def _dispatch(self, method: str, url: str, kwargs: dict[str, Any]): if method == "GET": return self._client.get(url, **kwargs) if method == "POST": return self._client.post(url, **kwargs) if method == "PATCH": return self._client.patch(url, **kwargs) if method == "DELETE": return self._client.delete(url, **kwargs) raise ValueError(f"Unsupported method: {method}") def _read_json(self, response) -> Any: json_method = getattr(response, "json", None) if callable(json_method): try: return json_method() except Exception: pass text_method = getattr(response, "text", None) if callable(text_method): text_value = text_method() if isinstance(text_value, bytes): text_value = text_value.decode("utf-8", errors="ignore") if isinstance(text_value, str) and text_value.strip(): return json.loads(text_value) content = getattr(response, "content", b"") if isinstance(content, bytes) and content: return json.loads(content.decode("utf-8", errors="ignore")) return {} @staticmethod def _extract_error_message(payload: Any, status_code: int) -> str: if isinstance(payload, dict): detail = payload.get("detail") error = payload.get("error") note = payload.get("note") pieces = [str(item).strip() for item in (detail, error, note) if item] if pieces: return " — ".join(pieces) return f"HTTP {status_code}" class DraxonMailsDiscordVerifier: DISCORD_LINK_RE = re.compile( r"https?://[^\s\"'<>]+(?:discord(?:app)?\.com|click\.discord\.com)[^\s\"'<>]*", re.IGNORECASE, ) def __init__(self, client: DraxonMailsClient): self.client = client def random_address(self) -> dict[str, Any]: return self.client.random_address() def extract_verification_link(self, messages: list[dict[str, Any]]) -> str: for message in messages: haystacks = [ str(message.get("html", "") or ""), str(message.get("body", "") or ""), str(message.get("markdown", "") or ""), ] for text in haystacks: for match in self.DISCORD_LINK_RE.findall(text): lowered = match.lower() if "discord" not in lowered: continue if any(token in lowered for token in ("verify", "email", "ls/click", "click.discord")): return match.rstrip(").,;\"'") raise DraxonMailsNoDiscordLinkError("Discord verification email found, but no verification link could be extracted") def get_verification_link(self, address: str, *, password: str = "", domain_secret: str = "") -> str: messages = self.client.get_inbox(address, password=password, domain_secret=domain_secret) return self.extract_verification_link(messages) def wait_for_verification_link( self, address: str, *, password: str = "", domain_secret: str = "", timeout_seconds: int = 120, poll_every_seconds: float = 3.0, ) -> str: deadline = time.time() + max(1, int(timeout_seconds)) last_error: Exception | None = None while time.time() < deadline: try: return self.get_verification_link(address, password=password, domain_secret=domain_secret) except DraxonMailsNoDiscordLinkError as exc: last_error = exc time.sleep(max(0.5, float(poll_every_seconds))) if last_error: raise last_error raise TimeoutError("Timed out waiting for a Discord verification link") __all__ = [ "DraxonMailsClient", "DraxonMailsConfig", "DraxonMailsDiscordVerifier", "DraxonMailsError", "DraxonMailsApiError", "DraxonMailsNoDiscordLinkError", ]