""" 发件域 Warm-up 与冷却管理 (反检测核心) ================================================= 三类机制: M. 新发件域 warm-up: 新域被滥用立刻被定位为群发签名, 必须按曲线放开额度 day1: 5封 / day2: 20 / day3: 50 / day4: 100 / day5+: 200 (该曲线参考主流营销平台 warm-up 流程) N. 同收件人 N 天冷却: 单收件人 N 天内只发1封, 避免重复骚扰触发举报 → ISP 视角抗识别 O. 退信率 backoff: 单发件域 5 分钟内退信率>20% 自动暂停该域 N 分钟 """ import os import time import json import threading from collections import defaultdict, OrderedDict # ============================================================ # 默认 warm-up 曲线 (天数 -> 当日最大额度) # ============================================================ DEFAULT_WARMUP_SCHEDULE = [5, 20, 50, 100, 200, 400, 800] # 收件人冷却时间戳 LRU 上限 (防 502万邮箱撑爆内存) RECIPIENT_COOLDOWN_MAX = 100000 # 持久化时最多保留的最近收件人条数 RECIPIENT_COOLDOWN_SAVE_MAX = 20000 class _RecipientCooldownLRU(OrderedDict): """收件人冷却时间戳 LRU: 超出上限淘汰最旧条目""" def __init__(self, *args, maxsize=RECIPIENT_COOLDOWN_MAX, **kwargs): super().__init__(*args, **kwargs) self.maxsize = maxsize def __setitem__(self, key, value): super().__setitem__(key, value) self.move_to_end(key) while len(self) > self.maxsize: self.popitem(last=False) class WarmupManager: def __init__(self, data_file='warmup_data.json', schedule=None, log_callback=None): self.data_file = data_file self.schedule = schedule or DEFAULT_WARMUP_SCHEDULE self.log_callback = log_callback self._lock = threading.RLock() # 长期状态: {domain: {'first_send_ts': float, 'last_send_date': 'YYYY-MM-DD', 'day_count': int, 'sent_today': int}} self.domain_state = {} # 短期退信窗口: {domain: [{'ts': float, 'bounce': bool}, ...]} self.bounce_window = defaultdict(list) # 被暂停的发件域: {domain: 're-enable_ts'} self.paused_domains = {} self._load() def log(self, level, msg): if self.log_callback: try: self.log_callback(level, msg) except Exception: pass # ---------- M: warm-up ---------- def get_day_quota(self, domain: str) -> int: """返回该发件域今日允许最大发件数 逻辑: - 首次发信时记录 day0, 当日额度=5 - 每过一天推进曲线 (day1/day2/day3...) - 超过曲线最大值时不再受限 (返回大数) """ with self._lock: now = time.time() today = time.strftime('%Y-%m-%d') st = self.domain_state.setdefault(domain, {}) if not st.get('first_send_ts'): st['first_send_ts'] = now st['last_send_date'] = today st['day_count'] = 0 st['sent_today'] = 0 self._save() # 日期推进 if st.get('last_send_date') != today: # 算过了几天 (线性近似); day_count 可超过曲线长度 → 后续不再限制 try: last_dt = time.mktime(time.strptime(st['last_send_date'], '%Y-%m-%d')) cur_dt = time.mktime(time.strptime(today, '%Y-%m-%d')) days_diff = int((cur_dt - last_dt) / 86400) st['day_count'] = st.get('day_count', 0) + max(days_diff, 1) except Exception: st['day_count'] = st.get('day_count', 0) + 1 st['last_send_date'] = today st['sent_today'] = 0 self._save() day_idx = st.get('day_count', 0) if day_idx < len(self.schedule): return self.schedule[day_idx] return 100000 # 已 warm-up 充分 (超过曲线最大天数, 不再限量) def record_send(self, domain: str): """成功发送1封 -> +1""" with self._lock: st = self.domain_state.get(domain) or {} st['sent_today'] = st.get('sent_today', 0) + 1 self.domain_state[domain] = st if st.get('sent_today', 0) % 5 == 0: self._save() def is_warmup_ok(self, domain: str) -> bool: """该发件域今日是否还能继续发 (没超当日 warm-up 配额)""" try: with self._lock: st = self.domain_state.get(domain) or {'sent_today': 0} quota = self.get_day_quota(domain) if st.get('sent_today', 0) >= quota: return False # 暂停中? re_enable = self.paused_domains.get(domain) if re_enable and time.time() < re_enable: return False return True except Exception: return True # ---------- N: 收件人冷却 ---------- def _recip_state(self) -> _RecipientCooldownLRU: """取收件人冷却字典 (自动把旧数据转成 LRU)""" st = self.domain_state.get('__recip__') if not isinstance(st, _RecipientCooldownLRU): st = _RecipientCooldownLRU(st or {}) self.domain_state['__recip__'] = st return st def can_send_to(self, email: str, cooldown_days: int = 7) -> bool: """该收件人最近 cooldown_days 天内是否已发过 -> 若已发返回 False""" # 用 LRU 上限保护内存 (完整方案可改 SQLite, 这里先 LRU + 文件) if cooldown_days <= 0: return True with self._lock: st = self._recip_state() last_ts = st.get(email) if last_ts and (time.time() - last_ts) < cooldown_days * 86400: return False return True def mark_sent_to(self, email: str): with self._lock: st = self._recip_state() st[email] = time.time() # 每 100 条保存一次 try: if len(st) > 10 and len(st) % 100 == 0: self._save() except Exception: pass # ---------- O: 退信率 backoff ---------- def record_send_result(self, domain: str, is_bounce: bool): """记录一次发送结果(bounce=True/False) → 计算窗口退信率""" with self._lock: now = time.time() window_seconds = 300 # 5 分钟 lst = self.bounce_window[domain] lst.append({'ts': now, 'bounce': is_bounce}) # 清掉窗口外 while lst and lst[0]['ts'] < now - window_seconds: lst.pop(0) # 计算退信率 recent = lst[-30:] # 至少 30 封信统计 if len(recent) >= 10: bounces = sum(1 for x in recent if x['bounce']) rate = bounces / len(recent) if rate >= 0.2: # 暂停 10 分钟 pause_sec = 600 self.paused_domains[domain] = now + pause_sec self.log('warning', f'[{domain}] 退信率{rate*100:.0f}% 触发限制, 暂停10分钟') self._save() def is_paused(self, domain: str) -> bool: with self._lock: re_enable = self.paused_domains.get(domain) if re_enable and time.time() < re_enable: return True if re_enable and time.time() >= re_enable: # 解除暂停 + 清退信窗口 self.paused_domains.pop(domain, None) self.bounce_window[domain] = [] self._save() return False # ---------- 持久化 ---------- def _save(self): try: data = { 'domain_state': self._domain_state_for_save(), 'paused_domains': self.paused_domains, } with open(self.data_file, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2, default=str) except Exception: pass def _domain_state_for_save(self) -> dict: """收件人冷却条目只保留最近 RECIPIENT_COOLDOWN_SAVE_MAX 条, 防止 JSON 无限膨胀""" ds = dict(self.domain_state) recip = ds.get('__recip__') if isinstance(recip, dict) and len(recip) > RECIPIENT_COOLDOWN_SAVE_MAX: ds['__recip__'] = dict(list(recip.items())[-RECIPIENT_COOLDOWN_SAVE_MAX:]) return ds def _load(self): try: if os.path.exists(self.data_file): with open(self.data_file, 'r', encoding='utf-8') as f: data = json.load(f) self.domain_state = data.get('domain_state', {}) self.paused_domains = data.get('paused_domains', {}) recip = self.domain_state.get('__recip__') if isinstance(recip, dict) and len(recip) > RECIPIENT_COOLDOWN_SAVE_MAX: self.domain_state['__recip__'] = dict(list(recip.items())[-RECIPIENT_COOLDOWN_SAVE_MAX:]) except Exception: pass def reset(self, domain: str = None): with self._lock: if domain: self.domain_state.pop(domain, None) self.bounce_window.pop(domain, None) self.paused_domains.pop(domain, None) else: self.domain_state = {} self.bounce_window = defaultdict(list) self.paused_domains = {} self._save()