"""Online Disposable Mail API entegrasyonu için yardımcı modüller.""" from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Optional import requests CONFIG_PATH = Path(__file__).with_name("config.json") _BASE_URL = "https://api.online-disposablemail.com/api" _DEFAULT_QUANTITY = 1 _DEFAULT_LINK_PRIORITY = True class DisposableMailAPIError(RuntimeError): """Online Disposable Mail API isteklerinde oluşan hataları temsil eder.""" @dataclass class DisposableMailOrder: """Sipariş sonucu dönen temel bilgiler.""" order_id: str email: str raw: Dict[str, Any] def _coerce_bool(value: Any) -> Optional[bool]: if value is None: return None if isinstance(value, bool): return value if isinstance(value, str): lowered = value.strip().lower() if lowered in {"1", "true", "yes", "on"}: return True if lowered in {"0", "false", "no", "off", ""}: return False return None if isinstance(value, (int, float)): return bool(value) return None def _coerce_int(value: Any) -> Optional[int]: if value is None: return None if isinstance(value, bool): return int(value) try: return int(str(value).strip()) except (TypeError, ValueError): return None class DisposableMailAPI: """Online Disposable Mail hizmeti ile temel HTTP isteklerini gerçekleştiren istemci.""" def __init__(self, config: Dict[str, Any]): try: api_key = str(config["disposablemail_apikey"]).strip() except KeyError as exc: raise DisposableMailAPIError( "config.json içinde 'disposablemail_apikey' bulunamadı." ) from exc if not api_key: raise DisposableMailAPIError("'disposablemail_apikey' değeri boş olamaz.") service_id = _coerce_int(config.get("disposablemail_service_id")) if service_id is None: raise DisposableMailAPIError( "config.json içinde 'disposablemail_service_id' değeri bulunamadı veya geçersiz." ) email_type_id = _coerce_int(config.get("disposablemail_email_type_id")) if email_type_id is None: raise DisposableMailAPIError( "config.json içinde 'disposablemail_email_type_id' değeri bulunamadı veya geçersiz." ) quantity = _coerce_int(config.get("disposablemail_quantity")) or _DEFAULT_QUANTITY link_priority = _coerce_bool(config.get("disposablemail_link_priority")) if link_priority is None: link_priority = _DEFAULT_LINK_PRIORITY self.api_key = api_key self.service_id = service_id self.email_type_id = email_type_id self.quantity = max(1, quantity) self.link_priority = bool(link_priority) def _request( self, path: str, params: Dict[str, Any], *, timeout: int = 30, allow_non_200: bool = False, ) -> Dict[str, Any]: url = f"{_BASE_URL}/{path.lstrip('/')}" encoded_params = {key: value for key, value in params.items() if value is not None} try: response = requests.get(url, params=encoded_params, timeout=timeout) response.raise_for_status() except requests.RequestException as exc: # pragma: no cover - ağ hataları test dışında raise DisposableMailAPIError( f"Online Disposable Mail isteği başarısız oldu: {exc}" ) from exc try: payload = response.json() except ValueError as exc: raise DisposableMailAPIError("Sunucu JSON formatında yanıt döndürmedi.") from exc if not isinstance(payload, dict): raise DisposableMailAPIError("Sunucu beklenmeyen JSON yanıtı döndürdü.") code_value = payload.get("code") if not allow_non_200 and code_value != 200: message = payload.get("msg") or "Bilinmeyen hata" raise DisposableMailAPIError(f"API hatası ({code_value}): {message}") return payload def get_balance(self) -> Dict[str, Any]: params = {"apiKey": self.api_key} return self._request("balance", params) def order_mail(self, quantity: Optional[int] = None) -> Dict[str, Any]: order_quantity = max(1, quantity or self.quantity) params = { "apiKey": self.api_key, "serviceId": self.service_id, "emailTypeId": self.email_type_id, "quantity": order_quantity, "linkPriority": str(self.link_priority).lower(), } payload = self._request("mailbox", params) data = payload.get("data") if not isinstance(data, dict): raise DisposableMailAPIError("Sipariş yanıtı beklenen formatta değil.") orders = data.get("orders") if not isinstance(orders, list) or not orders: raise DisposableMailAPIError("Sipariş sonucu e-posta listesi içermiyor.") first = orders[0] if not isinstance(first, dict): raise DisposableMailAPIError("Sipariş sonucu beklenmeyen veri içeriyor.") order_id = str(first.get("orderId") or "").strip() email_address = str(first.get("email") or "").strip() if not order_id or not email_address: raise DisposableMailAPIError("Sipariş sonucu geçerli orderId veya email içermiyor.") return { "code": payload.get("code", 0), "msg": payload.get("msg"), "data": { "order": { "orderId": order_id, "email": email_address, }, "orders": orders, "links": data.get("links"), }, } def check_mail(self, order_id: str | int) -> Dict[str, Any]: params = { "apiKey": self.api_key, "orderId": str(order_id), "linkPriority": str(self.link_priority).lower(), } payload = self._request("latest/code", params, allow_non_200=True) code = payload.get("code") if code == 401: message_raw = str(payload.get("msg") or "").strip() message = message_raw.lower() waiting_phrase = "waiting to receive verification code" if waiting_phrase not in message: raise DisposableMailAPIError( "API 401: apiKey eksik/yanlış veya IP yetkisiz." ) # Hizmet doğrulama kodunu henüz teslim etmediğinde API 401 döndürüyor. # Çağıran kodun beklemeye devam edebilmesi için yanıt kodunu temizle. payload = dict(payload) payload["code"] = None if message_raw: payload["msg"] = message_raw else: payload.pop("msg", None) return payload if code in (204, 404): return payload if code not in (200, None): raise DisposableMailAPIError(f"API hatası ({code}): {payload.get('msg')}") return payload def create_order_object(self, payload: Dict[str, Any]) -> DisposableMailOrder: data = payload.get("data") if not isinstance(data, dict): raise DisposableMailAPIError("Sipariş bilgisi eksik.") order_info = data.get("order") if not isinstance(order_info, dict): raise DisposableMailAPIError("Sipariş bilgisi bulunamadı.") order_id = str(order_info.get("orderId") or "").strip() email = str(order_info.get("email") or "").strip() if not order_id or not email: raise DisposableMailAPIError("Sipariş nesnesi oluşturulamadı.") return DisposableMailOrder(order_id=order_id, email=email, raw=order_info) def load_config(path: Path = CONFIG_PATH) -> Dict[str, Any]: if not path.exists(): raise FileNotFoundError(f"Config dosyası bulunamadı: {path}") with path.open("r", encoding="utf-8") as config_file: data = json.load(config_file) if not isinstance(data, dict): raise ValueError("config.json beklenmeyen veri döndürdü") return data def resolve_mail_service(config: Dict[str, Any]) -> str: anymessage_flag = _coerce_bool(config.get("anymessage")) if anymessage_flag is None: anymessage_flag = _coerce_bool(config.get("AnyMessage")) disposable_flag = _coerce_bool(config.get("disposablemail")) if disposable_flag is None: disposable_flag = _coerce_bool(config.get("DisposableMail")) anymessage_flag = bool(anymessage_flag) disposable_flag = bool(disposable_flag) if anymessage_flag and disposable_flag: raise ValueError("'anymessage' ve 'disposablemail' aynı anda true olamaz.") if disposable_flag: return "disposablemail" return "anymessage" def create_mail_api_from_file(path: Path = CONFIG_PATH) -> DisposableMailAPI: config = load_config(path) return DisposableMailAPI(config) __all__ = [ "DisposableMailAPI", "DisposableMailAPIError", "DisposableMailOrder", "create_mail_api_from_file", "load_config", "resolve_mail_service", ]