"""E-posta kullanımını takip eden yardımcı sınıflar.""" from __future__ import annotations import threading from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Optional, Set from email_utils import normalize_email_key @dataclass class EmailUsageRecord: """Tek bir e-posta adresinin kullanım bilgileri.""" base_email: str email_order_id: str last_used: datetime use_count: int def clone(self) -> "EmailUsageRecord": return EmailUsageRecord( base_email=self.base_email, email_order_id=self.email_order_id, last_used=self.last_used, use_count=self.use_count, ) def to_line(self) -> str: return "|".join( [ self.base_email, self.email_order_id, self.last_used.astimezone(timezone.utc).isoformat(), str(self.use_count), ] ) @staticmethod def from_line(line: str) -> Optional["EmailUsageRecord"]: parts = [part.strip() for part in (line or "").split("|")] if len(parts) < 4: return None base_email, email_order_id, last_used_raw, use_count_raw = parts[:4] if not base_email: return None try: last_used = datetime.fromisoformat(last_used_raw) if last_used.tzinfo is None: last_used = last_used.replace(tzinfo=timezone.utc) except ValueError: last_used = datetime.fromtimestamp(0, tz=timezone.utc) try: use_count = int(use_count_raw) except ValueError: use_count = 0 return EmailUsageRecord(base_email, email_order_id, last_used, use_count) class EmailUsageTracker: """E-posta kullanımını dosya üzerinde izler.""" def __init__(self, path: Path, max_uses: int, reuse_after: timedelta) -> None: self._path = Path(path) self._lock = threading.Lock() self._max_uses = max(1, int(max_uses or 1)) self._reuse_after = reuse_after self._records: Dict[str, EmailUsageRecord] = {} self._in_use: Set[str] = set() self._load() @property def max_uses(self) -> int: return self._max_uses @property def reuse_after(self) -> timedelta: return self._reuse_after def normalize(self, email: str) -> str: return normalize_email_key(email) def _load(self) -> None: if not self._path.exists(): return try: text = self._path.read_text(encoding="utf-8") except OSError: return for line in text.splitlines(): record = EmailUsageRecord.from_line(line) if not record: continue key = self.normalize(record.base_email) if not key: continue self._records[key] = record def _save_locked(self) -> None: lines = [self._records[key].to_line() for key in sorted(self._records.keys())] if lines: if not self._path.parent.exists(): self._path.parent.mkdir(parents=True, exist_ok=True) self._path.write_text("\n".join(lines) + "\n", encoding="utf-8") else: if self._path.exists(): self._path.unlink() def iter_ready(self) -> List[EmailUsageRecord]: now = datetime.now(timezone.utc) ready: List[EmailUsageRecord] = [] with self._lock: for key, record in self._records.items(): if record.use_count >= self._max_uses: continue if key in self._in_use: continue if record.last_used + self._reuse_after > now: continue ready.append(record.clone()) ready.sort(key=lambda rec: rec.last_used) return ready def reserve(self, record: EmailUsageRecord) -> bool: key = self.normalize(record.base_email) now = datetime.now(timezone.utc) with self._lock: stored = self._records.get(key) if stored is None: return False if stored.use_count >= self._max_uses: return False if key in self._in_use: return False if stored.last_used + self._reuse_after > now: return False self._in_use.add(key) return True def reserve_first_ready(self) -> Optional[EmailUsageRecord]: now = datetime.now(timezone.utc) with self._lock: for key, record in self._records.items(): if record.use_count >= self._max_uses: continue if key in self._in_use: continue if record.last_used + self._reuse_after > now: continue self._in_use.add(key) return record.clone() return None def release(self, email: str) -> None: key = self.normalize(email) if not key: return with self._lock: self._in_use.discard(key) def record_usage(self, email: str, email_order_id: str) -> EmailUsageRecord: base_email = str(email or "").strip() if not base_email: raise ValueError("Geçersiz e-posta adresi") key = self.normalize(base_email) if not key: raise ValueError("Geçersiz e-posta adresi") now = datetime.now(timezone.utc) with self._lock: record = self._records.get(key) if record is None: record = EmailUsageRecord(base_email, email_order_id, now, 1) else: record.base_email = base_email if email_order_id: record.email_order_id = email_order_id record.last_used = now record.use_count += 1 self._records[key] = record self._in_use.discard(key) if record.use_count >= self._max_uses: # maksimum kullanım sayısına ulaşıldıysa tekrar kuyruğa alınmasın pass self._save_locked() return record.clone() def is_exhausted(self, record: EmailUsageRecord) -> bool: key = self.normalize(record.base_email) with self._lock: stored = self._records.get(key) if stored is None: return False return stored.use_count >= self._max_uses class EmailReuseManager: """Hazır e-postaları belirli aralıklarla yeniden kullanan modül.""" def __init__( self, tracker: EmailUsageTracker, *, poll_interval: float = 60.0, ) -> None: self._tracker = tracker self._poll_interval = max(5.0, float(poll_interval)) self._queue: List[EmailUsageRecord] = [] self._queued_keys: Set[str] = set() self._lock = threading.Lock() self._stop_event = threading.Event() self._thread: Optional[threading.Thread] = None def start(self) -> None: if self._thread and self._thread.is_alive(): return self._thread = threading.Thread(target=self._run, name="EmailReuse", daemon=True) self._thread.start() def stop(self) -> None: self._stop_event.set() thread = self._thread if thread and thread.is_alive(): thread.join(timeout=5.0) def _run(self) -> None: while not self._stop_event.wait(self._poll_interval): self._populate_queue() def _populate_queue(self) -> None: ready = self._tracker.iter_ready() if not ready: return with self._lock: for record in ready: key = self._tracker.normalize(record.base_email) if key in self._queued_keys: continue if self._tracker.is_exhausted(record): continue self._queue.append(record) self._queued_keys.add(key) def acquire_available_email(self) -> Optional[EmailUsageRecord]: with self._lock: while self._queue: record = self._queue.pop(0) key = self._tracker.normalize(record.base_email) self._queued_keys.discard(key) if self._tracker.is_exhausted(record): continue if self._tracker.reserve(record): return record # Kuyruk boşsa anlık olarak kontrol et record = self._tracker.reserve_first_ready() return record def notify_failure(self, email: str) -> None: self._tracker.release(email)