"""
代理IP直投群发引擎 (Web版 — 单机直投模式)
=================================================
特性:
  - IP 复用长连接 SMTP 直投 (走 SOCKS 代理 隧道 -> 目标域 MX:25)
  - 连续 3 封拒(421/554/隧道失败等IP视角信号)才换 IP,正常发就继续用同 IP
  - 强制只读 数据/valid_emails.txt 作为收件人源 (避免无效信损耗代理信誉)
  - DKIM 签名 (dkimpy) + 中性 SPF ?all 让发件域轮换也合规
  - 跟 app.py 验证端状态机解耦: 本模块自带 send_state / stop_flag, 不污染验证端
  - 复用 core/sender.py 的 MailSender.send_direct (新路径: 传proxy/prebuilt_sock)

使用:
  from web_send import web_send_engine
  web_send_engine.start(config)
  web_send_engine.stop()
  progress = web_send_engine.snapshot()
"""
import os
import sys
import threading
import time
import json
import random
import string
import socket
import csv
import dns.resolver
from email.utils import formatdate, make_msgid
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from concurrent.futures import ThreadPoolExecutor, as_completed
from queue import Queue

# 把项目根目录加到 sys.path 以便 import core
_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
_PROJECT_ROOT = os.path.dirname(_THIS_DIR)
if _PROJECT_ROOT not in sys.path:
    sys.path.insert(0, _PROJECT_ROOT)

# 复用 core/ 的部分模块
try:
    from core.models import MailBox, Email, ProxyIP
    from core.sender import MailSender, _open_socks_tunnel, _smtp_recv, _smtp_send
    import core.sender as _core_sender_mod
except Exception as _e:
    # 启动期未装依赖时不友好; 真正使用时(后端已 import 过) 一定能 import
    raise ImportError(f'web_send 依赖 core.* 导入失败: {_e}')

try:
    import dkim
    _DKIM_AVAILABLE = True
except ImportError:
    _DKIM_AVAILABLE = False


# ============================================================
# 全局单例 + 状态机
# ============================================================
DATA_DIR = os.path.join(_THIS_DIR, '数据')
VALID_EMAILS_FILE = os.path.join(DATA_DIR, 'valid_emails.txt')
SEND_RESULT_CSV = os.path.join(_THIS_DIR, 'send_result.csv')
SENT_CACHE_FILE = os.path.join(_THIS_DIR, 'send_cache.db')

state_lock = threading.Lock()
stats_lock = threading.Lock()
csv_lock = threading.Lock()
log_lock = threading.Lock()
send_stop_flag = {'stop': False}

# ============================================================
# 桥接模式 (外部 SMTP 中继喂入, 如 PHPMailer Pro / 伯勒群发器)
# ============================================================
bridge_queue = Queue()
bridge_enabled_flag = {'on': False}
_unsub_set = None
_unsub_lock = threading.Lock()


def _load_unsub_set():
    global _unsub_set
    with _unsub_lock:
        if _unsub_set is None:
            s = set()
            f = os.path.join(_THIS_DIR, 'unsubscribe.txt')
            if os.path.exists(f):
                try:
                    with open(f, 'r', encoding='utf-8') as fh:
                        for line in fh:
                            e = line.strip().lower()
                            if e and '@' in e:
                                s.add(e)
                except Exception:
                    pass
            _unsub_set = s
        return _unsub_set


def bridge_push(item):
    """外部(中继接收)喂入一封邮件
    item: {email, from_name, from_email, subject, html, text, headers, mail_from}
    返回 True/False
    """
    if not bridge_enabled_flag['on']:
        return False
    email = (item.get('email') or '').strip().lower()
    if not email or '@' not in email:
        return False
    item['unsub'] = email in _load_unsub_set()
    bridge_queue.put(item)
    with stats_lock:
        send_state['total'] = int(send_state.get('total') or 0) + 1
    return True


def bridge_queue_size():
    return bridge_queue.qsize()

# ★ 代理池热更新容器: 任务运行中主控可推送新代理池, worker 下次取代理即生效
#   (API直连IP会过期, 主控定期拉新IP后推送到节点热切换, 不用停任务)
_live_proxies = {'list': None}


def update_live_proxies(new_proxies):
    """外部(主控 /set_proxy 转发)调用: 热更新运行中任务的代理池
    new_proxies: ProxyIP 对象列表 或 None(清除)"""
    try:
        with state_lock:
            _live_proxies['list'] = list(new_proxies or [])
        _add_log('代理池热更新: ' + str(len(_live_proxies['list'])) + ' 条')
        return True
    except Exception:
        return False


send_state = {
    'status': 'idle',              # idle/running/stopped/finished/error
    'total': 0,
    'processed': 0,
    'success': 0,
    'fail': 0,
    'bounced': 0,
    'ip_rejected': 0,
    'switches': 0,                 # IP 切换次数
    'sent_per_ip_max': 0,          # 单 IP 最大连续成功数(实时)
    'start_time': 0,
    'elapsed': 0,
    'speed': 0,
    'current_proxy_id': None,
    'log_lines': [],
    'error_msg': '',
    # 配置回显
    'proxies_count': 0,
    'thread_count': 0,
    'send_interval': 0,
    'boxes_count': 0,
    # IP 视角连续拒信统计 (总览,不分IP)
    'total_ip_reject_streak': 0,
}


def _add_log(msg):
    with log_lock:
        ts = time.strftime('%H:%M:%S')
        send_state['log_lines'].append('[' + ts + '] ' + msg)
        if len(send_state['log_lines']) > 200:
            send_state['log_lines'] = send_state['log_lines'][-200:]


# ============================================================
# 已发送缓存 (断点续发) — SQLite 版 (千万级不爆内存, 百万级写入不卡)
# ============================================================
class SentCache:
    def __init__(self, path=SENT_CACHE_FILE):
        import sqlite3
        self.path = path
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(path, check_same_thread=False)
        self._conn.execute('PRAGMA journal_mode=WAL')
        self._conn.execute('PRAGMA synchronous=NORMAL')
        self._conn.execute('CREATE TABLE IF NOT EXISTS sent (email TEXT PRIMARY KEY)')
        self._conn.commit()
        self._pending = 0

    def is_sent(self, email):
        with self._lock:
            cur = self._conn.execute('SELECT 1 FROM sent WHERE email=?', (email.lower(),))
            return cur.fetchone() is not None

    def mark_sent(self, email):
        with self._lock:
            self._conn.execute('INSERT OR IGNORE INTO sent(email) VALUES(?)', (email.lower(),))
            self._pending += 1
            if self._pending >= 100:
                self._conn.commit()
                self._pending = 0

    def save(self):
        with self._lock:
            self._conn.commit()

    def close(self):
        with self._lock:
            try:
                self._conn.commit()
                self._conn.close()
            except Exception:
                pass


# ============================================================
# DKIM 签名封装
# ============================================================
class DKIMSignerWrap:
    """供 core/sender.MailSender.send_direct 调用的 dkim_signer callback
    返回(string): 给定 msg 字符串 + 域名 -> 加好 DKIM-Signature 头的 msg 字符串
    """
    def __init__(self, dkim_keys, sign_extended=True, header_mode=None):
        # dkim_keys: dict[domain] -> {'selector': str, 'private_key': PEM str 或 bytes}
        self.keys = dkim_keys or {}
        self.sign_extended = sign_extended  # True=12头扩展 False=7头
        self.header_mode = header_mode  # minimal/core7/replyto8/extended12/full14/random_pool
        self._pd = None
        try:
            import print_disguise as _pd
            self._pd = _pd
        except Exception:
            self._pd = None

    def __call__(self, msg_str, domain):
        if not _DKIM_AVAILABLE or not self.keys or self._pd is None:
            return msg_str
        info = self.keys.get(domain)
        if not info:
            if len(self.keys) == 1:
                info = list(self.keys.values())[0]
            else:
                return msg_str
        try:
            selector = info.get('selector', 'default')
            privkey = info.get('private_key', '')
            if isinstance(privkey, str):
                privkey = privkey.encode('utf-8')
            msg_bytes = msg_str.encode('utf-8') if isinstance(msg_str, str) else msg_str
            # ★ 强制 relaxed/relaxed, 不回退到 simple (simple 会导致 body hash 不匹配)
            signed = self._pd.sign_dkim_relaxed(msg_bytes, selector, domain, privkey,
                                                sign_extended=self.sign_extended,
                                                header_mode=self.header_mode)
            return signed + msg_bytes  # ★ dkimpy只返回签名头, 需要拼接到原始msg前面
        except Exception as _e:
            _add_log('DKIM签名失败(忽略): ' + str(_e)[:80])
            return msg_str


# ============================================================
# MX 解析 (复用验证端思路, 缓存5分钟 + 负缓存 + 悬空MX判定)
# ============================================================
_mx_cache = {}
_mx_lock = threading.Lock()
_mx_ttl = 300      # 正常结果缓存 5 分钟
_mx_neg_ttl = 600  # 负结果(无MX/悬空MX)缓存 10 分钟, 避免每封重复查
_mx_cache_max = 5000


def resolve_mx_detail(domain):
    """解析收件域 MX, 返回 (ips: list|None, status: str)
    status:
      'ok'          解析成功
      'no_mx'       无 MX 记录
      'mx_dangling' MX 存在但主机解析失败(悬空MX, 发信必退)
      'mx_no_a'     MX 主机无 A 记录
      'dns_error'   查询异常(可能是临时网络问题)
    """
    if not domain or '.' not in domain or len(domain) < 4:
        return None, 'no_mx'
    now = time.time()
    with _mx_lock:
        hit = _mx_cache.get(domain)
        if hit is not None:
            ips, status, ts = hit
            ttl = _mx_ttl if status == 'ok' else _mx_neg_ttl
            if now - ts < ttl:
                return (list(ips) if ips else None), status

    ips = []
    status = 'dns_error'
    answers = None
    try:
        answers = list(dns.resolver.resolve(domain, 'MX'))
    except dns.resolver.NoAnswer:
        status = 'no_mx'
        answers = []
    except dns.resolver.NXDOMAIN:
        status = 'no_mx'
        answers = []
    except Exception:
        status = 'dns_error'
        answers = None

    if answers is not None:
        if not answers:
            status = 'no_mx' if status == 'no_mx' else 'dns_error'
        else:
            answers.sort(key=lambda x: x.preference)
            mx_failed = 0
            for rdata in answers:
                mx_name = str(rdata.exchange).rstrip('.')
                try:
                    a_records = dns.resolver.resolve(mx_name, 'A')
                    got = 0
                    for a in a_records:
                        ip = str(a)
                        if ip not in ips:
                            ips.append(ip)
                            got += 1
                    if got == 0:
                        mx_failed += 1
                except dns.resolver.NXDOMAIN:
                    mx_failed += 1
                except dns.resolver.NoAnswer:
                    mx_failed += 1
                except Exception:
                    mx_failed += 1
            if ips:
                status = 'ok'
            elif mx_failed >= len(answers):
                status = 'mx_dangling'
            else:
                status = 'mx_no_a'

    with _mx_lock:
        # LRU 淘汰: 先清过期, 仍满则清一半
        if len(_mx_cache) >= _mx_cache_max:
            for k in list(_mx_cache.keys()):
                v = _mx_cache[k]
                ttl = _mx_ttl if v[1] == 'ok' else _mx_neg_ttl
                if time.time() - v[2] >= ttl:
                    del _mx_cache[k]
            if len(_mx_cache) >= _mx_cache_max:
                for k in list(_mx_cache.keys())[:_mx_cache_max // 2]:
                    del _mx_cache[k]
        _mx_cache[domain] = (list(ips), status, now)
    return (ips if ips else None), status


def resolve_mx_ip(domain):
    """兼容旧调用: 只返回 IP 列表或 None"""
    ips, _status = resolve_mx_detail(domain)
    return ips


# ============================================================
# 隧道建立 + EHLO 握手 (复用 core.sender 助手)
# ============================================================
# uTLS 前置层配置 (Go 程序 jputls, 端口2526, 任意JA3指纹)
_utls_forwarder = {'host': '127.0.0.1', 'port': 2526, 'enabled': True}


def _smtp_recv_line(sock):
    buf = b''
    while b'\n' not in buf and len(buf) < 65536:
        try:
            c = sock.recv(4096)
        except Exception:
            break
        if not c:
            break
        buf += c
    return buf.decode('utf-8', 'replace')


def _establish_via_utls(mx_ip, proxy, ehlo_d, use_starttls):
    """经本机 uTLS 前置层建隧道 + banner + EHLO + STARTTLS(uTLS随机JA3) + EHLO2
    成功后返回的 socket 已桥接到 MX, 可直接继续 MAIL FROM 对话
    """
    if not _utls_forwarder.get('enabled'):
        return None, 'disabled'
    import base64 as _b64
    fields = '\x01'.join([
        str(proxy.proxy_type or 'socks5'),
        str(proxy.ip), str(proxy.port),
        str(proxy.username or ''), str(proxy.pwd or ''),
        str(mx_ip), '25',
        ehlo_d or 'localhost',
        '1' if use_starttls else '0',
    ])
    ctl = 'CTL ' + _b64.b64encode(fields.encode('utf-8')).decode('ascii') + '\n'
    s = None
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(35)
        s.connect((str(_utls_forwarder.get('host', '127.0.0.1')), int(_utls_forwarder.get('port', 2526))))
        s.sendall(ctl.encode('utf-8'))
        line = _smtp_recv_line(s).strip()
        if not line.startswith('+OK'):
            try: s.close()
            except Exception: pass
            return None, 'utls_' + (line[1:40] if line else 'fail')
        parts = line[4:].strip().split(' ')
        if len(parts) < 2:
            try: s.close()
            except Exception: pass
            return None, 'utls_badresp'
        banner = _b64.b64decode(parts[0] or '').decode('utf-8', 'replace')
        ehlo_resp = _b64.b64decode(parts[1] or '').decode('utf-8', 'replace')
        tls_profile = parts[2] if len(parts) > 2 else ''
        if not banner.startswith('220') or not ehlo_resp.startswith('250'):
            try: s.close()
            except Exception: pass
            return None, 'utls_badresp'
        s.settimeout(None)
        s._smtp_extensions = _parse_ehlo_extensions(ehlo_resp.encode('utf-8'))
        s._helo_random_d = ehlo_d
        s._tls_upgraded = bool(tls_profile)
        s._tls_profile = tls_profile or 'plaintext'
        return s, ''
    except Exception:
        if s is not None:
            try: s.close()
            except Exception: pass
        return None, 'utls_fail'


def _parse_ehlo_extensions(ehlo_resp_bytes):
    """解析 EHLO 响应提取 SMTP 扩展能力 (反检测关键: 让我们表现出知道 MTA 扩展支持)
    返回 set[str]: {'STARTTLS', 'PIPELINING', '8BITMIME', 'SIZE 10485760', 'AUTH LOGIN PLAIN', ...}
    """
    exts = set()
    try:
        text = ehlo_resp_bytes.decode('utf-8', errors='replace')
        for line in text.split('\n'):
            line = line.strip().rstrip('\r')
            if not line:
                continue
            # 响应格式 "250-AUTH" 或 "250 AUTH" (最后一条用空格而非横线)
            if line[:3] in ('250', '250-', 'EHLO', '250 '):
                body = line[4:] if line.startswith('250-') else (line[4:] if line.startswith('250 ') else line)
                body = body.strip()
                if not body:
                    continue
                # 第一个 token 是扩展名
                key = body.split()[0].upper()
                exts.add(body)  # 完整保留原扩展字符串
                exts.add(key)   # 也保留 key
    except Exception:
        pass
    return exts


def establish_tunnel(mx_ip, proxy, helo_domain, use_starttls=True):
    """建 SOCKS 隧道到 mx_ip:25, 收 banner, EHLO 握手
    + STARTTLS 自动协商 (若 MX 提供且 use_starttls=True)
    + EHLO 扩展解析
    返回 (sock_or_None, fail_reason)
      fail_reason: 'tunnel' / 'no_banner' / 'ehlo_fail' / 'starttls_fail' / ''
    """
    try:
        import print_disguise as _pd
    except Exception:
        _pd = None

    def _human_pause():
        if _pd is not None:
            try:
                _pd.smtp_human_pause()
            except Exception:
                pass

    # EHLO 域名: 真实手机客户端风格 (与伪造Received的客户端一致, docomo入站头不穿帮)
    # 真实iPhone提交= smtpclient.apple / iphone.local; Android= android.local
    if helo_domain and str(helo_domain).strip():
        ehlo_d = str(helo_domain).strip()
    else:
        _helo_pool = ['smtpclient.apple', 'smtpclient.apple', 'smtpclient.apple',
                      'iphone.local', 'iphone', 'ipad.local', 'android.local']
        ehlo_d = random.choice(_helo_pool)

    # ★ 优先走 uTLS 前置层 (JA3 任意伪造); 失败自动兜底 Python TLS 路径
    u_sock, _u_reason = _establish_via_utls(mx_ip, proxy, ehlo_d, use_starttls)
    if u_sock is not None:
        return u_sock, ''

    _human_pause()
    sock = _open_socks_tunnel(mx_ip, 25, proxy, timeout=30)
    if sock is None:
        return None, 'tunnel'
    banner = _smtp_recv(sock, timeout=5)
    if not banner.startswith(b'220'):
        try: sock.close()
        except: pass
        return None, 'no_banner'
    _human_pause()
    if not _smtp_send(sock, f'EHLO {ehlo_d}\r\n'.encode()):
        try: sock.close()
        except: pass
        return None, 'ehlo_fail'
    ehlo_resp = _smtp_recv(sock, timeout=8)
    if not ehlo_resp.startswith(b'250'):
        try: sock.close()
        except: pass
        return None, 'ehlo_fail'

    # 解析 SMTP 扩展 (反检测关键: 让我们表现像真实 MTA, 知道 MTA 扩展支持)
    exts = _parse_ehlo_extensions(ehlo_resp)
    sock._smtp_extensions = exts

    # STARTTLS 自动协商 (docomo/yahoo 等强烈偏好 TLS)
    if use_starttls and any('STARTTLS' in e for e in exts):
        try:
            _human_pause()
            if not _smtp_send(sock, b'STARTTLS\r\n'):
                try: sock.close()
                except: pass
                return None, 'starttls_fail'
            tls_resp = _smtp_recv(sock, timeout=8)
            if not tls_resp.startswith(b'220'):
                # STARTTLS 被拒, 继续明文也行(部分 MTA 看到拒绝会直接关连接, 我们准备好回退)
                # 但很多 MTA 一旦宣告 STARTTLS + 客户端不协商, 会判为低端 MUA, 直接拒后续业务
                # 这里采取: STARTTLS 失败立即关掉这条 socket, 让上层换 IP (避免误导指纹)
                try: sock.close()
                except: pass
                return None, 'starttls_fail'
            # 升级到 TLS — ★ 随机 TLS 指纹 (JA3 多态: 密码套件/曲线/TLS版本/SNI 每连接不同)
            import ssl as _ssl
            ctx, tls_profile = (_pd.build_random_tls_context()
                                if _pd is not None and hasattr(_pd, 'build_random_tls_context')
                                else (None, {'name': 'default'}))
            if ctx is None:
                ctx = _ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = _ssl.CERT_NONE
            # SNI 随机: 部分连接不带SNI(真实客户端多样行为), 部分带
            sni = mx_ip if random.random() < 0.5 else None
            try:
                tls_sock = ctx.wrap_socket(sock, server_hostname=sni)
            except Exception:
                # 服务器证书无效时尝试不验证
                try:
                    ctx2 = _ssl.create_default_context()
                    ctx2.check_hostname = False
                    ctx2.verify_mode = _ssl.CERT_NONE
                    tls_sock = ctx2.wrap_socket(sock)
                except Exception:
                    try: sock.close()
                    except: pass
                    return None, 'starttls_fail'
            tls_sock._tls_profile = tls_profile.get('name', 'default')
            # 再发一次 EHLO (TLS 后 RFC 5321 要求再 EHLO)
            _human_pause()
            if not _smtp_send(tls_sock, f'EHLO {ehlo_d}\r\n'.encode()):
                try: tls_sock.close()
                except: pass
                return None, 'ehlo_fail'
            ehlo_resp2 = _smtp_recv(tls_sock, timeout=8)
            if not ehlo_resp2.startswith(b'250'):
                try: tls_sock.close()
                except: pass
                return None, 'ehlo_fail'
            exts2 = _parse_ehlo_extensions(ehlo_resp2)
            tls_sock._smtp_extensions = exts2
            tls_sock._helo_random_d = ehlo_d
            tls_sock._tls_upgraded = True
            return tls_sock, ''
        except Exception:
            try: sock.close()
            except: pass
            return None, 'starttls_fail'

    sock._helo_random_d = ehlo_d
    sock._tls_upgraded = False
    return sock, ''


def _count_valid_lines(path):
    n = 0
    try:
        with open(path, 'r', encoding='utf-8') as f:
            for line in f:
                e = line.strip()
                if e and '@' in e:
                    n += 1
    except Exception:
        pass
    return n


# ============================================================
# 主任务入口
# ============================================================
def start_send(config):
    """启动群发任务
    config:
      emails              : List[str]  收件邮箱清单 (强校验: 通常来自 valid_emails.txt)
      subjects            : List[str]  主题池(支持Spintax)
      html_parts          : List[str]  HTML正文池(支持Spintax)
      text_parts          : List[str]  纯文本正文池(可空)
      boxes               : List[MailBox] 发件箱池(轮换发件人)
      proxies             : List[ProxyIP] 代理池(连续3封拒换)
      thread_count        : int 线程数
      send_interval       : float 每封间隔秒(默认0)
      send_timeout        : int SMTP超时(秒)
      helo_domain         : 可选HELO域名(留空时按发件域)
      reply_to            : 可选Reply-To
      custom_headers      : dict 额外邮件头(如 LIST-UNSUBSCRIBE)
      magic_config        : 魔术编码开关(传给 core.text_engine.MagicEncoder)
      dkim_keys           : dict[domain] -> {selector, private_key}  DKIM 签名
      use_cache           : bool 启用断点续发(默认 True)
      ip_reject_threshold : int 连续几封拒才换 IP(默认3)
      dns_servers         : List[str] MX/DNS 解析服务器(默认空)
    """
    global send_stop_flag, bridge_enabled_flag
    send_stop_flag['stop'] = False
    bridge_mode = bool(config.get('bridge_mode', False))
    emails_file = config.get('emails_file') or ''
    run_async = bool(config.get('run_async', False))
    finish_callback = config.get('finish_callback') or None

    # ★ 防双发保护: 引擎运行中禁止重复启动 (主控连点/多端同时启动)
    with state_lock:
        cur_status = send_state.get('status', 'idle')
    if cur_status == 'running':
        _set_error('任务正在发送中, 请勿重复启动 (如需重发请先停止)')
        return False

    emails = config.get('emails') or []
    subjects = config.get('subjects') or ['重要なお知らせ']
    html_parts = config.get('html_parts') or ['']
    text_parts = config.get('text_parts') or []
    boxes = config.get('boxes') or []
    proxies_raw = config.get('proxies') or []
    proxies = [p for p in proxies_raw if p.enable]
    # ★ 热更新容器初始化 (worker 每次取代理从容器读, 支持运行中切换)
    _live_proxies['list'] = list(proxies)
    thread_count = int(config.get('thread_count') or 10)
    send_interval = float(config.get('send_interval') or 0)
    send_timeout = int(config.get('send_timeout') or 15)
    helo_domain = config.get('helo_domain') or ''
    reply_to = config.get('reply_to') or ''
    custom_headers = config.get('custom_headers') or {}
    magic_config = config.get('magic_config') or {'enabled': False}
    dkim_keys = config.get('dkim_keys') or {}
    use_cache = bool(config.get('use_cache', True))
    ip_threshold = int(config.get('ip_reject_threshold', 1))  # 默认1: 单封被拒立即换IP
    dns_servers = config.get('dns_servers') or None
    # ---- 反识别新加 ----
    ip_proactive_threshold = int(config.get('ip_proactive_threshold', 30))  # 单IP主动轮换(默认30封后换)
    mx_rate_per_minute = int(config.get('mx_rate_per_minute', 0))  # 单IP单MX每分钟最大发数, 0=不限
    domain_rate_per_minute = int(config.get('domain_rate_per_minute', 0) or 0)  # 单发件域每分钟限速(每worker), 0=不限
    schedule_pause = bool(config.get('schedule_pause', False))  # 启用凌晨暂停时段
    schedule_pause_start = int(config.get('schedule_pause_start', 0))  # 暂停起始小时(0-23)
    schedule_pause_end = int(config.get('schedule_pause_end', 6))  # 暂停结束小时(0-23)
    worker_paused_log = {}  # worker_id -> bool 已写过暂停日志的旗标(避免刷屏)
    # ---- M+N+O: warm-up + 同收件人冷却 + 退信率backoff ----
    warmup_enabled = False  # 强制关闭,不再受前端影响,避免warmup_skip阻止发信
    warmup_cooldown_days = 0  # 强制关闭冷却
    # ---- 工业级进箱改进 ----
    verp_enabled = False  # 默认关(VERP格式可能被日系邮局判BULK)
    disguise_mode = config.get('disguise_mode', 'mua')  # 'mua' or 'esp'
    dkim_extended = bool(config.get('dkim_extended', True))  # 12 头 DKIM
    dkim_header_mode = config.get('dkim_header_mode') or None  # minimal/core7/replyto8/extended12/full14/random_pool
    unsub_filter_enabled = bool(config.get('unsub_filter_enabled', True))  # 退订过滤
    # SMTP 拟人化分档: 'fast'(不sleep) / 'medium'(握手阶段sleep, DATA后全速) / 'slow'(每命令sleep)
    smtp_pace_mode = config.get('smtp_pace_mode', 'medium')
    # 0-7 = 立即换 IP 集合 (单封 421/554 等立刻换, 不管阈值)
    immediate_switch_codes = {421, 554, 451, 450, 452}
    use_starttls = bool(config.get('use_starttls', True))  # MX 提供则自动 STARTTLS
    # ★ 每封都新IP (动态住宅代理: 每条新SOCKS隧道=不同出口IP)
    always_new_ip = False  # 已废弃:用IP复用模式(验证系统同款)

    # 输入校验
    if not emails and not bridge_mode and not emails_file:
        _set_error('收件邮箱列表为空')
        return False
    if not boxes:
        _set_error('未配置发件箱')
        return False
    if not proxies:
        _set_error('未配置代理IP')
        return False
    if not subjects:
        subjects = ['お知らせ']
    if not html_parts or not any(html_parts):
        html_parts = ['<html><body><p>{email}</p></body></html>']

    # 同步状态
    with state_lock:
        send_state.update({
            'status': 'running',
            'total': 0 if bridge_mode else (_count_valid_lines(emails_file) if emails_file else len(emails)),
            'processed': 0,
            'success': 0,
            'fail': 0,
            'bounced': 0,
            'ip_rejected': 0,
            'switches': 0,
            'sent_per_ip_max': 0,
            'start_time': 0,  # 真启动时再冻
            'elapsed': 0,
            'speed': 0,
            'proxies_count': len(proxies),
            'thread_count': thread_count,
            'send_interval': send_interval,
            'boxes_count': len(boxes),
            'log_lines': [],
            'error_msg': '',
        })

    _add_log('群发任务启动')
    _add_log('收件:' + str(len(emails)) + ' 发件箱:' + str(len(boxes)) + ' 代理:' + str(len(proxies)) + ' 线程:' + str(thread_count))
    _add_log('IP视角拒信阈值:连续' + str(ip_threshold) + '封拒才换IP (1=单封立即换)')
    _add_log('IP主动轮换阈值:' + str(ip_proactive_threshold) + '封后主动换IP')
    _add_log('warm-up: ' + ('启用' if warmup_enabled else '关闭') + ' / cooldown: ' + str(warmup_cooldown_days) + '天')
    # 列出所有代理供用户确认
    for i, p in enumerate(proxies):
        _add_log('代理 #' + str(i+1) + ': ' + p.ip + ':' + str(p.port) + ' (' + p.username + ')')
    if mx_rate_per_minute > 0:
        _add_log('同IP同MX速率限制:' + str(mx_rate_per_minute) + '封/分钟')
    if schedule_pause:
        _add_log('定时暂停:凌晨' + str(schedule_pause_start) + '-' + str(schedule_pause_end) + '点不发')
    if dkim_keys:
        _add_log('DKIM签名: 已启用 ' + str(len(dkim_keys)) + '个域密钥')
        # ★ DKIM DNS 预检 (PHPMailer Pro 同款): 发送前确认公钥已在 DNS 生效, 避免签名无效邮件浪费配额
        try:
            import dns.resolver as _dnsres
            for _dk_dom, _dk_info in dkim_keys.items():
                _sel = (_dk_info.get('selector') if isinstance(_dk_info, dict) else '') or 'default'
                try:
                    _ans = _dnsres.resolve(_sel + '._domainkey.' + _dk_dom, 'TXT', lifetime=5)
                    _txt = ''.join((str(x) for x in _ans))
                    if 'p=' in _txt:
                        _add_log('DKIM DNS预检 ✓ ' + _dk_dom + ' (selector=' + _sel + ' 已发布)')
                    else:
                        _add_log('⚠ DKIM DNS预检: ' + _dk_dom + ' 记录存在但无公钥p=')
                except Exception as _de:
                    _add_log('⚠ DKIM DNS预检: ' + _dk_dom + ' 未找到 ' + _sel + '._domainkey 记录 (' + str(_de)[:50] + ') — 收件方验签将失败, 建议先在域名管理发布DKIM')
        except Exception:
            pass
    else:
        _add_log('DKIM签名: 未配置(忽略)')
    if custom_headers.get('LIST-UNSUBSCRIBE'):
        _add_log('退订头: ' + custom_headers['LIST-UNSUBSCRIBE'])

    # 已发送缓存 (桥接/文件模式下队列由外部喂入, 不走 pending 逻辑)
    sent_cache = SentCache() if (use_cache and not bridge_mode) else None
    pending = []
    if not bridge_mode:
        if emails_file:
            pending = []
        else:
            pending = [e for e in emails if not (sent_cache and sent_cache.is_sent(e))]
            if use_cache:
                _add_log('断点续发: 已发' + str(len(emails) - len(pending)) + '  待发' + str(len(pending)))
            else:
                pending = list(emails)
            if not pending:
                with state_lock:
                    send_state['status'] = 'finished'

    # core MailSender 实例 (只用它的 _build_message / send_direct + 回调)
    sender = MailSender()
    sender.log_callback = lambda lvl, msg: _add_log('[' + lvl + '] ' + msg)
    # ★ 邮件字符集 (默认 utf-8; ISO-2022-JP = 日文原生编码, 日系邮局加分项)
    sender._mail_charset = config.get('mail_charset', 'utf-8') or 'utf-8'
    _add_log('邮件字符集: ' + sender._mail_charset)
    # 反识别伪装 — 默认开启,用户在前端可关
    sender._disguise_default = bool(config.get('disguise_enabled', True))
    sender._disguise_mode = disguise_mode  # 'mua' or 'esp'
    # ★ 魔改配置 (主控下发的伪装细节: 客户端池/伪Received/头开关/Date抖动/Message-ID域名)
    _disguise_opts = config.get('disguise_opts') or None
    sender._disguise_opts = _disguise_opts
    if _disguise_opts:
        _add_log('魔改配置已下发: ' + json.dumps(_disguise_opts, ensure_ascii=False)[:200])
    # ★ 进箱黑科技5个开关 (通过 self._feature_flags 传给 _build_message)
    sender._feature_flags = {
        'personalize_subject': bool(config.get('personalize_subject', False)),
        'avoid_spam_words': bool(config.get('avoid_spam_words', False)),
        'invisible_segments': bool(config.get('invisible_segments', False)),
        'embed_images': bool(config.get('embed_images', False)),
        'timestamp_personalize': bool(config.get('timestamp_personalize', False)),
    }

    # ===== 发件人显示控制 (From 头自定义) =====
    from_display_name_cfg = config.get('from_display_name', '') or ''
    from_domain_mode_cfg = config.get('from_domain_mode', 'auto') or 'auto'
    from_custom_domain_cfg = config.get('from_custom_domain', '') or ''
    if from_display_name_cfg or from_domain_mode_cfg in ('custom', 'hide'):
        _add_log('发件人显示: name="' + from_display_name_cfg + '" mode=' + from_domain_mode_cfg + ' custom_domain=' + from_custom_domain_cfg)
        # 应用到每个 box (生成新的 sender_name + 必要时改 box.email)
        for box in boxes:
            if from_display_name_cfg:
                # 支持 Spintax: 命中即随机
                try:
                    from core.text_engine import SpintaxProcessor as _spin
                    box.sender_name = _spin.process(from_display_name_cfg)
                except Exception:
                    box.sender_name = from_display_name_cfg
            if from_domain_mode_cfg == 'custom' and from_custom_domain_cfg and '@' in (box.email or ''):
                local = box.email.split('@')[0]
                box.email = local + '@' + from_custom_domain_cfg
                if box.user and '@' in box.user:
                    box.user = local + '@' + from_custom_domain_cfg
            # hide 模式: 只显示名字, 邮箱地址保留原域(MUA GUI 默认隐藏, 但邮件头必含地址,
            #           所以"隐藏"实质是改显示名为目标名 + 邮箱地址保留 — 大多数 MUA GUI 只显示 display name)
            # auto 模式: 不改 box
    # SMTP 拟人化分档: fast(全速)/medium(握手阶段sleep)/slow(每命令sleep)
    smtp_pace_mode = config.get('smtp_pace_mode', 'medium')
    try:
        _core_sender_mod._smtp_pace_enabled['mode'] = smtp_pace_mode
    except Exception:
        pass

    # VPN 启用 - 大陆机器必须 (Python -> VPN -> udealproxy)
    use_local_vpn = bool(config.get('use_local_vpn', False))
    local_vpn_port = int(config.get('local_vpn_port', 7890))
    local_vpn_type = config.get('local_vpn_type', 'http') or 'http'
    try:
        _core_sender_mod._local_vpn_config['enabled'] = use_local_vpn
        _core_sender_mod._local_vpn_config['port'] = local_vpn_port
        _core_sender_mod._local_vpn_config['type'] = local_vpn_type
        if use_local_vpn:
            _add_log('★VPN已启用: 127.0.0.1:' + str(local_vpn_port) + ' (' + local_vpn_type + ') 链路: 本机->VPN->代理->MX')
        else:
            _add_log('VPN关闭: 直连代理 (香港/海外服务器可用)')
    except Exception:
        pass
    _add_log('SMTP拟人化节奏: ' + smtp_pace_mode + (' (快速)' if smtp_pace_mode=='fast' else ' (握手/全速)' if smtp_pace_mode=='medium' else ' (每命令sleep)'))
    # uTLS 设备指纹层 (Go 前置层 jputls:2526, 任意JA3)
    _utls_cfg = config.get('utls_forwarder') or {}
    try:
        _utls_forwarder['host'] = str(_utls_cfg.get('host', '127.0.0.1'))
        _utls_forwarder['port'] = int(_utls_cfg.get('port', 2526) or 2526)
        _utls_forwarder['enabled'] = bool(_utls_cfg.get('enabled', True))
        _add_log('uTLS设备指纹层: ' + ('启用 ' + str(_utls_forwarder['host']) + ':' + str(_utls_forwarder['port']) + ' (失败自动兜底Python TLS)'
                                        if _utls_forwarder['enabled'] else '关闭'))
    except Exception:
        pass
    send_mode = config.get('send_mode', 'auth') or 'auth'
    _add_log('发送模式: ' + ('认证模式(DKIM签名+MUA伪装)' if send_mode == 'auth' else '裸发模式(不签DKIM,直接发)'))
    if send_mode == 'raw':
        # ★ 裸发模式: 用自己的域名 + 签DKIM, 不做MUA伪装 (速度快 + 进收件箱)
        dkim_signer = DKIMSignerWrap(dkim_keys, sign_extended=dkim_extended, header_mode=dkim_header_mode) if dkim_keys else None
        sender._disguise_default = False  # 裸发不做MUA伪装
        _add_log('裸发模式: 用自己的域名 + 签DKIM, 不做MUA伪装')
    else:
        # ★ 认证模式: 用自己的域名 + 签DKIM + MUA伪装
        dkim_signer = DKIMSignerWrap(dkim_keys, sign_extended=dkim_extended, header_mode=dkim_header_mode) if dkim_keys else None

    # ===== 智能个性化: 按收件域自动匹配魔改档案 =====
    smart_enabled = False
    smart_overrides = {}
    smart_signers = {}
    try:
        _sp_cfg = config.get('smart_profile') or {}
        smart_enabled = bool(_sp_cfg.get('enabled', False))
        smart_overrides = _sp_cfg.get('profiles') or {}
        if smart_enabled:
            from smart_profile import SMART_PROFILES as _SP_PROFILES
            from smart_profile import profile_for as _sp_profile_for
            for _key, _prof in _SP_PROFILES.items():
                smart_signers[_key] = DKIMSignerWrap(dkim_keys, sign_extended=dkim_extended,
                                                     header_mode=_prof.get('dkim_header_mode'))
            _add_log('智能个性化: 已启用 (按收件域自动匹配 ' + str(len(_SP_PROFILES)) + ' 个档案)')
    except Exception as _e:
        smart_enabled = False
        _add_log('智能个性化初始化失败(忽略): ' + str(_e)[:60])

    # ===== M+N+O warm-up + 同收件人冷却 + 退信率 backoff =====
    if warmup_enabled:
        try:
            from warmup_manager import WarmupManager
            warmup_mgr = WarmupManager(
                data_file=os.path.join(_THIS_DIR, 'warmup_data.json'),
                log_callback=lambda lvl, msg: _add_log('[' + lvl + '] [warmup] ' + msg),
            )
            _add_log('warm-up管理已启用: 收件人' + str(warmup_cooldown_days) + '天冷却 + 退信率backoff')
        except Exception as _e:
            warmup_mgr = None
            _add_log('warm-up管理初始化失败(忽略): ' + str(_e)[:60])
    else:
        warmup_mgr = None
    cooldown_days = warmup_cooldown_days if warmup_enabled else 0

    # 魔术编码引擎 (可选)
    try:
        if magic_config.get('enabled'):
            from core.text_engine import MagicEncoder, SpintaxProcessor, VariableProcessor
            magic = MagicEncoder(magic_config)
            _add_log('魔术编码: 已启用')
        else:
            MagicEncoder = None
            magic = None
            from core.text_engine import VariableProcessor
    except Exception:
        magic = None
    try:
        from core.text_engine import SpintaxProcessor, VariableProcessor
    except Exception:
        SpintaxProcessor = None
        VariableProcessor = None

    rng = random.Random()

    # ===== 退订过滤 (启动前查 unsubscribe.txt, 排除已退订邮箱) =====
    unsub_set = set()
    if unsub_filter_enabled:
        unsub_file = os.path.join(_THIS_DIR, 'unsubscribe.txt')
        if os.path.exists(unsub_file):
            try:
                with open(unsub_file, 'r', encoding='utf-8') as f:
                    for line in f:
                        e = line.strip().lower()
                        if e and '@' in e:
                            unsub_set.add(e)
                if unsub_set:
                    _add_log('退订过滤: 已加载 ' + str(len(unsub_set)) + ' 个退订邮箱, 启动后自动剔除')
            except Exception:
                pass
        if unsub_set:
            before = len(pending)
            pending = [e for e in pending if e.lower() not in unsub_set]
            _add_log('退订过滤生效: 剔除 ' + str(before - len(pending)) + ' 个已退订邮箱, 实际待发 ' + str(len(pending)))

    if bridge_mode:
        email_queue = bridge_queue  # 桥接模式: 队列持续由外部喂入
    elif emails_file:
        # 大文件模式: feeder 线程逐行投喂 (百万级收件人不占内存)
        email_queue = Queue(maxsize=10000)
        feeder_done = {'done': False}

        def _feeder():
            try:
                with open(emails_file, 'r', encoding='utf-8') as f:
                    for line in f:
                        if send_stop_flag['stop']:
                            break
                        e = line.strip().lower()
                        if not e or '@' not in e:
                            continue
                        if sent_cache and sent_cache.is_sent(e):
                            continue
                        if unsub_set and e in unsub_set:
                            continue
                        email_queue.put(e)
            except Exception:
                pass
            finally:
                feeder_done['done'] = True

        threading.Thread(target=_feeder, daemon=True).start()
    else:
        email_queue = Queue()
        for e in pending:
            email_queue.put(e)

    # 准备 CSV 输出 header (断点续发追加模式)
    if not os.path.exists(SEND_RESULT_CSV):
        try:
            with open(SEND_RESULT_CSV, 'w', encoding='utf-8', newline='') as f:
                csv.writer(f).writerow(['email', 'status', 'code', 'msg', 'timestamp'])
        except Exception:
            pass

    proxy_counter = [0]
    # ★ 隧道并发信号量: 不管开多少线程, 同时最多 N 个连接在向代理建隧道 (默认5, 防代理账号并发被打爆)
    tunnel_concurrency = int(config.get('tunnel_concurrency') or 5)
    if tunnel_concurrency < 1:
        tunnel_concurrency = 1
    tunnel_sem = threading.BoundedSemaphore(tunnel_concurrency)
    _add_log('隧道并发限制: 同时最多 ' + str(tunnel_concurrency) + ' 个连接向代理建隧道')

    # 真启动时刻(不计预解析)
    with state_lock:
        send_state['start_time'] = time.time()

    def worker(worker_id):
        """单 worker: 1 个代理 IP 长连接一直发, 连续 ip_threshold 封拒才换 IP
        + 单 IP 主动轮换阈值 + 同 IP 同 MX 速率限制 + 多 MX 轮询 + 发件人随机 + 定时暂停
        """
        import copy as _copy
        # 每个 worker 固定绑定一段代理池范围(可循环), 通过 proxy_counter 全局轮询避免冲突
        local_succ_streak = 0
        ip_reject_streak = 0
        ip_active_succ_streak = 0  # 主动轮换: 单 IP 累计成功数
        ip_succ_per_mx_minute = {}  # 单 IP + 单 MX 的 1 分钟发送计数
        sock = None
        current_proxy = None
        current_mx_ip = None
        box_idx_global = [worker_id % len(boxes)]  # local box cursor
        mx_idx_map = {}  # 同一域内取不同 MX IP (worker 内部循环)

        while not send_stop_flag['stop']:
            # === 定时暂停(凌晨00:00-06:00 自动暂停) ===
            if schedule_pause:
                hour_now = time.localtime().tm_hour
                if schedule_pause_start <= hour_now < schedule_pause_end:
                    # 进入暂停时段, 等60秒后再查
                    if not worker_paused_log.get(worker_id, False):
                        _add_log('worker' + str(worker_id) + ': 凌晨暂停时段 ' + str(schedule_pause_start) + '-' + str(schedule_pause_end) + ' 等待...')
                        worker_paused_log[worker_id] = True
                    time.sleep(60)
                    continue
                else:
                    worker_paused_log[worker_id] = False

            if bridge_mode:
                try:
                    item = email_queue.get(timeout=1.5)
                except Exception:
                    continue  # 队列空, 继续等外部喂入
                email_addr = (item.get('email') or '').strip().lower()
            elif emails_file:
                try:
                    email_addr = email_queue.get(timeout=1.5)
                except Exception:
                    if feeder_done['done'] and email_queue.empty():
                        break  # 文件读完且队列清空
                    continue
            else:
                try:
                    email_addr = email_queue.get_nowait()
                except Exception:
                    break

            def requeue(em):
                """失败重入队 (桥接模式下必须把整个 item 放回, 不能丢内容)"""
                if bridge_mode:
                    email_queue.put(item)
                else:
                    email_queue.put(em)

            # 取发件箱(轮询, 加随机选域) + warm-up 检查
            # 寻找一个 warmup_ok 且非暂停的发件箱(最多试5次, 都不行就跳过此邮箱)
            chosen_box = None
            if bridge_mode:
                # 桥接模式: 发件人来自外部喂入的 item (PHPMailer Pro 的 From)
                chosen_box = _copy.copy(boxes[box_idx_global[0] % len(boxes)])
                box_idx_global[0] += 1
                fe = (item.get('from_email') or '').strip()
                fn = (item.get('from_name') or '').strip()
                if fe and '@' in fe:
                    chosen_box.email = fe
                    chosen_box.user = fe
                if fn:
                    chosen_box.sender_name = fn
                chosen_box.random_user = False
            else:
                for _try in range(5):
                    if send_stop_flag['stop']:
                        break
                    box = boxes[box_idx_global[0] % len(boxes)]
                    box_idx_global[0] += 1
                    if rng.random() < 0.3:
                        box_idx_global[0] += rng.randint(1, 3)
                    box_domain = box.email.split('@')[-1] if '@' in box.email else ''
                    if warmup_mgr is not None:
                        if not box_domain:
                            continue
                        if warmup_mgr.is_paused(box_domain):
                            # 域被退信率 backoff 暂停, 换域
                            continue
                        if not warmup_mgr.is_warmup_ok(box_domain):
                            # 今日 warm-up 配额已满
                            continue
                    chosen_box = box
                    break
            if chosen_box is None:
                # 找不到合适发件箱, 全域都在暂停或额度足够 → 记录跳过
                _record_result(email_addr, False, 'warmup_skip', '所有发件箱暂停或当日配额已满', '0')
                with stats_lock:
                    send_state['fail'] += 1
                    send_state['processed'] += 1
                _add_log('warmup_skip: '+email_addr)
                continue
            box = chosen_box

            # 桥接模式: 退订过滤 (启动时已加载 unsubscribe.txt)
            if bridge_mode and item.get('unsub'):
                _record_result(email_addr, False, 'unsub', '收件人已退订', '0')
                with stats_lock:
                    send_state['fail'] += 1
                    send_state['processed'] += 1
                continue

            # ===== N: 同收件人冷却检查 =====
            if warmup_mgr is not None and cooldown_days > 0:
                if not warmup_mgr.can_send_to(email_addr, cooldown_days):
                    _record_result(email_addr, False, 'cooldown', '收件人冷却期内跳过', '0')
                    with stats_lock:
                        send_state['fail'] += 1
                        send_state['processed'] += 1
                    continue

            # 解析 MX
            try:
                domain = email_addr.split('@')[-1].lower()
            except Exception:
                _record_result(email_addr, False, 'bad', '邮箱格式错', '0')
                with stats_lock:
                    send_state['fail'] += 1
                    send_state['processed'] += 1
                continue

            # ★ 智能个性化: 按收件域匹配档案 (每封独立的伪装/签名/限速)
            cur_opts = None
            cur_signer = dkim_signer
            cur_mx_rate = mx_rate_per_minute
            cur_domain_rate = domain_rate_per_minute
            if smart_enabled:
                try:
                    _pkey, _prof = _sp_profile_for(email_addr, smart_overrides)
                    cur_opts = _prof.get('disguise_opts')
                    cur_signer = smart_signers.get(_pkey) or dkim_signer
                    cur_mx_rate = int(_prof.get('mx_rate_per_minute') or 0)
                    cur_domain_rate = int(_prof.get('domain_rate_per_minute') or 0)
                except Exception:
                    pass
            mx_ips, mx_status = resolve_mx_detail(domain)
            if not mx_ips:
                if mx_status == 'mx_dangling':
                    _record_result(email_addr, False, 'mx_dangling', 'MX存在但主机解析失败(悬空MX,发信必退)', '0')
                elif mx_status == 'dns_error':
                    _record_result(email_addr, False, 'no_mx', 'MX查询异常', '0')
                else:
                    _record_result(email_addr, False, 'no_mx', 'MX解析失败', '0')
                with stats_lock:
                    send_state['fail'] += 1
                    send_state['processed'] += 1
                if mx_status == 'mx_dangling':
                    _add_log('悬空MX跳过:' + email_addr)
                elif mx_status != 'dns_error':
                    _add_log('跳过无MX:' + email_addr)
                continue

            # 多 MX 轮选: 同一 worker 在每次需要新 IP 时, 切换到下一个 MX
            prev_mx_idx = mx_idx_map.get(domain, -1)
            new_mx_idx = (prev_mx_idx + 1) % len(mx_ips)
            mx_idx_map[domain] = new_mx_idx
            target_mx = mx_ips[new_mx_idx]

            # —— IP 复用 / 换 IP 决策 ——
            # ★ 验证系统同款逻辑: sock 一路发到被拒才换 (每封复用sock)
            # 被拒时 worker 的 send_direct 会让 return_sock=None -> 下次循环自动 need_new_ip=True 重建隧道
            need_new_ip = (sock is None) or (current_mx_ip != target_mx)
            if not need_new_ip and ip_reject_streak >= ip_threshold:
                # 连续 N 封拒才换 IP
                with stats_lock:
                    send_state['switches'] += 1
                try:
                    sock.close()
                except Exception:
                    pass
                sock = None
                ip_reject_streak = 0
                ip_active_succ_streak = 0
                _add_log('worker' + str(worker_id) + ': 连续' + str(ip_threshold) + '封拒 → 换IP')
                need_new_ip = True

            if not need_new_ip and ip_active_succ_streak >= ip_proactive_threshold:
                # 主动轮换: 单 IP 连续成功 N 封后, 主动换 IP (避免单 IP 长时间指纹)
                with stats_lock:
                    send_state['switches'] += 1
                try:
                    sock.close()
                except Exception:
                    pass
                sock = None
                ip_reject_streak = 0
                ip_active_succ_streak = 0
                need_new_ip = True

            if need_new_ip:
                # ★ 从热更新容器取当前代理池 (主控推送新池后立即生效)
                _cur_pool = _live_proxies.get('list') or proxies
                if not _cur_pool:
                    time.sleep(1)
                    continue
                with stats_lock:
                    idx = proxy_counter[0] % len(_cur_pool)
                    proxy_counter[0] += 1
                    current_proxy = _cur_pool[idx]
                    send_state['current_proxy_id'] = current_proxy.id or (idx + 1)
                    # (switches 已在上方被拒时统计过, 这里不重复加)
                # ★ 获取代理日志 (同步主控可见)
                _add_log('worker' + str(worker_id) + ': 获取代理#' + str(current_proxy.id or '')
                         + ' (' + str(current_proxy.ip) + ') → 目标MX ' + target_mx
                         + ' (' + email_addr.split('@')[-1] + ')')
                current_mx_ip = target_mx
                ip_succ_per_mx_minute = {}  # 重置速率计数
                sock = None
                consecutive_fail = 0
                # 对齐验证系统: 最多试 15 次 (初一次抖动不放弃)
                # 连续失败 3/6/10 次加退避 sleep, 15 次都失败才算这条代理废
                for attempt in range(15):
                    if send_stop_flag['stop']:
                        break
                    tunnel_sem.acquire()
                    try:
                        sock, fail_reason = establish_tunnel(current_mx_ip, current_proxy,
                                                              helo_domain or (box.email.split('@')[-1] if '@' in box.email else 'localhost'),
                                                              use_starttls=use_starttls)
                    finally:
                        tunnel_sem.release()
                    if sock is not None:
                        # ★ 握手成功日志 (同步主控可见)
                        _add_log('worker' + str(worker_id) + ': 隧道握手成功 → 代理#' + str(current_proxy.id or '')
                                 + ' → ' + current_mx_ip + ' (' + email_addr.split('@')[-1] + ')')
                        break
                    consecutive_fail += 1
                    with stats_lock:
                        send_state['ip_rejected'] += 1
                    # ★ 握手失败日志 (节流: 每worker最多8条, 之后只在关键退避点打)
                    if consecutive_fail <= 8 or consecutive_fail in (10, 15):
                        _add_log('worker' + str(worker_id) + ': 握手失败(第' + str(consecutive_fail) + '次) '
                                 + str(fail_reason or '') + ' → 代理#' + str(current_proxy.id or '')
                                 + ' → ' + current_mx_ip)
                    if consecutive_fail == 3:
                        time.sleep(1)
                    elif consecutive_fail == 6:
                        time.sleep(3)
                    elif consecutive_fail == 10:
                        time.sleep(5)
                    elif consecutive_fail >= 14:
                        time.sleep(2)
                    else:
                        time.sleep(0.3)
                # 15 次都失败 -> 邮箱重入队,等下条 IP 再试 (最多 3 次重入)
                if sock is None:
                    _record_result(email_addr, False, 'tunnel', '代理隧道/握手失败', '0')
                    ip_reject_streak += 1
                    with stats_lock:
                        send_state['switches'] += 1  # 这条IP废了→要换→计入switches
                    retry_key = '_retry_' + email_addr
                    with stats_lock:
                        send_state['ip_rejected'] += 1
                        rt = send_state.get(retry_key, 0)
                        if rt < 3:
                            send_state[retry_key] = rt + 1
                            requeue(email_addr)  # 重入队等下次新IP重试
                            _add_log('worker' + str(worker_id) + ': ' + email_addr + ' 隧道失败重入队 (第' + str(rt+1) + '/3次)')
                        else:
                            send_state['fail'] += 1
                            send_state['processed'] += 1
                            _add_log('worker' + str(worker_id) + ': ' + email_addr + ' 3次重入仍失败 → 计入失败')
                    continue

            # === 同 IP 同 MX 速率限制 ===
            if cur_mx_rate > 0:
                key = str(current_proxy.id) + '|' + current_mx_ip
                now_min = int(time.time() / 60)
                last_min, last_count = ip_succ_per_mx_minute.get(key, (now_min, 0))
                if last_min == now_min and last_count >= cur_mx_rate:
                    # 该 IP 该 MX 本分钟已发够, 等到下分钟
                    wait_s = 60 - int(time.time() % 60) + 1
                    _add_log('worker' + str(worker_id) + ': 速率限制 ' + str(last_count) + '/分 → 等' + str(wait_s) + '秒')
                    time.sleep(wait_s)
                    ip_succ_per_mx_minute[key] = (now_min + 1, 0)

            # === 单发件域限速 (保护域名信誉, 千万级必备; 每worker独立计数) ===
            if cur_domain_rate > 0:
                sd = box.email.split('@')[-1] if '@' in box.email else ''
                dkey = 'd|' + sd
                now_min = int(time.time() / 60)
                lm, lc = ip_succ_per_mx_minute.get(dkey, (now_min, 0))
                if lm == now_min and lc >= cur_domain_rate:
                    wait_s = 60 - int(time.time() % 60) + 1
                    time.sleep(wait_s)
                    ip_succ_per_mx_minute[dkey] = (now_min + 1, 0)

            # 构造邮件内容 (轮换主题+内容)
            if bridge_mode:
                # 桥接模式: 内容来自外部喂入 (PHPMailer Pro 已渲染好变量)
                subject_real = (item.get('subject') or '').replace('[email]', email_addr).replace('{email}', email_addr)
                html_real = (item.get('html') or '').replace('[email]', email_addr).replace('{email}', email_addr)
                text_real = (item.get('text') or '').replace('[email]', email_addr).replace('{email}', email_addr)
                if not html_real:
                    html_real = '<html><body></body></html>'
                if magic:
                    try:
                        html_real = magic.process(html_real)
                    except Exception:
                        pass
            else:
                subject_real = rng.choice(subjects)
                html_real = ''
                for _ in range(3):
                    if SpintaxProcessor:
                        try:
                            subject_real = SpintaxProcessor.process(subject_real, rng)
                        except Exception:
                            pass
                    html_real = rng.choice(html_parts)
                    if SpintaxProcessor:
                        try:
                            html_real = SpintaxProcessor.process(html_real, rng)
                        except Exception:
                            pass
                    break
                # 变量替换(最简)
                html_real = html_real.replace('[email]', email_addr).replace('{email}', email_addr)
                subject_real = subject_real.replace('[email]', email_addr).replace('{email}', email_addr)
                if magic:
                    try:
                        html_real = magic.process(html_real)
                    except Exception:
                        pass

                text_real = ''
                if text_parts:
                    text_real = rng.choice(text_parts).replace('[email]', email_addr).replace('{email}', email_addr)

            # 发件人随机化 — 每封都重选随机子域内账号
            box_for_send = _copy.copy(box)
            box_for_send.email = box.email  # 重置默认
            if box.random_user:
                rnd_user = 'u' + ''.join(rng.choices(string.ascii_lowercase + string.digits, k=6))
                box_for_send.email = rnd_user + '@' + (box.email.split('@')[-1] if '@' in box.email else box.email)
            # 多账号轮换: 如果 box 配了多个真实账号, 30% 概率用第一个, 70% 走随机
            # (本实现由 boxes[] 池轮换体现, 不需要额外账号池)

            # Email 对象
            email_obj = Email(email_address=email_addr)

            # 桥接模式: 透传外部喂入的合规头 (List-Unsubscribe 等)
            hdrs = dict(custom_headers or {})
            if bridge_mode:
                for k, v in (item.get('headers') or {}).items():
                    if v:
                        hdrs[k] = v
            # [email] 变量替换 (退订链接等逐收件人个性化)
            hdrs = {k: (str(v).replace('[email]', email_addr) if isinstance(v, str) else v)
                    for k, v in hdrs.items()}

            # 调 MailSender.send_direct (新路径: 传 proxy + prebuilt_sock + 已开启disguise=True + VERP)
            success, code, msg, ip_rejected, return_sock = sender.send_direct(
                box_for_send, email_obj, subject_real, html_real,
                mail_type=1,
                dns_servers=dns_servers,
                helo_domain=helo_domain,
                sender_name=box.sender_name,
                timeout=send_timeout,
                proxy=current_proxy,
                dkim_signer=cur_signer,
                custom_headers=hdrs,
                prebuilt_sock=sock,
                verp_enabled=verp_enabled,
                disguise_opts=cur_opts,
            )

            sock = return_sock  # 复用或 None

            _record_result(email_addr, success, code, msg, str(current_proxy.id or ''))

            if success:
                local_succ_streak += 1
                ip_active_succ_streak += 1
                ip_reject_streak = 0
                if sent_cache:
                    sent_cache.mark_sent(email_addr)
                with stats_lock:
                    send_state['success'] += 1
                    send_state['processed'] += 1
                if cur_mx_rate > 0:
                    key = str(current_proxy.id) + '|' + current_mx_ip
                    now_min = int(time.time() / 60)
                    lm, lc = ip_succ_per_mx_minute.get(key, (now_min, 0))
                    ip_succ_per_mx_minute[key] = (now_min, lc + 1)
                if cur_domain_rate > 0:
                    sd2 = box.email.split('@')[-1] if '@' in box.email else ''
                    dkey = 'd|' + sd2
                    now_min = int(time.time() / 60)
                    lm, lc = ip_succ_per_mx_minute.get(dkey, (now_min, 0))
                    ip_succ_per_mx_minute[dkey] = (now_min, lc + 1)
                # warm-up: 记录发送成功
                if warmup_mgr is not None:
                    try:
                        sd = box.email.split('@')[-1] if '@' in box.email else ''
                        if sd:
                            warmup_mgr.record_send(sd)
                        warmup_mgr.mark_sent_to(email_addr)
                        warmup_mgr.record_send_result(sd, is_bounce=False)
                    except Exception:
                        pass
            else:
                # ★ 验证系统同款逻辑: IP/网络问题 -> 关sock换IP + 邮箱重入队3次 (不计fail)
                #                  邮箱本身错(550/551/552) -> 直接计fail (不重试)
                code_str_raw = str(code)
                try:
                    code_num = int(code_str_raw)
                except Exception:
                    code_num = 0

                # 邮箱本身问题码 (不重试)
                is_invalid_mailbox = code_num in (550, 551, 552)

                # 是否IP/网络问题 -> 需要关sock换IP
                is_ip_problem = (ip_rejected or code_num in immediate_switch_codes
                                 or code_num in (450, 451, 452, 421)
                                 or code_str_raw in ('timeout', 'sock_closed', 'tunnel', 'no_banner', 'ehlo_fail', 'connect', 'unknown'))

                if is_ip_problem:
                    # IP问题 -> 关sock, 下次循环自动 need_new_ip 重建隧道
                    if sock:
                        try: sock.close()
                        except: pass
                    sock = None
                    ip_reject_streak += 1
                    ip_active_succ_streak = 0
                    with stats_lock:
                        send_state['switches'] += 1  # 换IP 计数
                        send_state['ip_rejected'] += 1
                    # 邮箱重入队用新IP重试 (最多3次) - 不计fail (重试中)
                    retry_key = '_retry_' + email_addr
                    with stats_lock:
                        rt = send_state.get(retry_key, 0)
                        if rt < 3:
                            send_state[retry_key] = rt + 1
                            requeue(email_addr)
                            _add_log('worker' + str(worker_id) + ': ' + email_addr + ' IP问题(' + code_str_raw + ') → 重入队新IP再试 (第' + str(rt+1) + '/3次)')
                        else:
                            send_state['fail'] += 1  # 3次重试都不行, 真正计入失败
                            send_state['processed'] += 1
                            _add_log('worker' + str(worker_id) + ': ' + email_addr + ' 3次重试仍失败(' + code_str_raw + ') → 计入失败')
                elif is_invalid_mailbox:
                    # 邮箱本身错 (550 user unknown / 551 / 552 quota full) - 不重试
                    with stats_lock:
                        send_state['fail'] += 1
                        send_state['processed'] += 1
                        send_state['bounced'] += 1
                    ip_reject_streak = 0  # 不是IP的锅
                else:
                    # 其他失败 - 计入fail, 不重试
                    with stats_lock:
                        send_state['fail'] += 1
                        send_state['processed'] += 1
                    ip_reject_streak = 0

                # warm-up: 记录退信
                if warmup_mgr is not None:
                    try:
                        sd = box.email.split('@')[-1] if '@' in box.email else ''
                        warmup_mgr.record_send_result(sd, is_bounce=is_invalid_mailbox)
                    except Exception:
                        pass

            # 速度统计 (10% 概率更新,减少锁竞争)
            if rng.random() < 0.1:
                with stats_lock:
                    elapsed = time.time() - send_state['start_time'] if send_state['start_time'] else 1
                    send_state['speed'] = round(send_state['processed'] / max(elapsed, 1), 1)
                    send_state['elapsed'] = round(elapsed, 1)

            # 节奏
            if send_interval > 0:
                time.sleep(send_interval + rng.uniform(0, max(send_interval * 0.3, 0.05)))

        # worker 结束清理 socks
        if sock:
            try: sock.close()
            except: pass

    # 实际启动线程池
    _add_log('启动 ' + str(thread_count) + ' 个发送线程' + (' (桥接模式: 等待外部SMTP中继喂入)' if bridge_mode else ''))

    def _run_pool():
        try:
            with ThreadPoolExecutor(max_workers=thread_count) as ex:
                futures = [ex.submit(worker, i) for i in range(thread_count)]
                last_log_t = time.time()
                for f in as_completed(futures):
                    try:
                        f.result()
                    except Exception as _e:
                        _add_log('worker异常(忽略): ' + str(_e)[:80])
                    if time.time() - last_log_t > 10:
                        last_log_t = time.time()
                        with stats_lock:
                            _add_log('进度 ' + str(send_state['processed']) + '/' + str(send_state['total'])
                                     + ' 成=' + str(send_state['success'])
                                     + ' 败=' + str(send_state['fail'])
                                     + ' 退=' + str(send_state['bounced'])
                                     + ' IP被拒=' + str(send_state['ip_rejected'])
                                     + ' 换IP=' + str(send_state['switches']))
        except Exception as _e:
            _set_error('任务异常: ' + str(_e))
            return
        finally:
            if sent_cache:
                sent_cache.save()
            flush_csv()
            if bridge_mode:
                bridge_enabled_flag['on'] = False
            with state_lock:
                send_state['elapsed'] = round(time.time() - send_state['start_time'], 1)
                send_state['status'] = 'stopped' if send_stop_flag['stop'] else 'finished'
                send_state['speed'] = round(send_state['processed'] / max(send_state['elapsed'], 1), 1)
            if bridge_mode:
                _add_log('桥接引擎停止: 成' + str(send_state['success'])
                         + ' 败' + str(send_state['fail']) + ' 退' + str(send_state['bounced'])
                         + ' 用时' + str(send_state['elapsed']) + '秒')
            elif run_async:
                _add_log('异步任务结束: ' + send_state['status']
                         + ' 成' + str(send_state['success'])
                         + ' 败' + str(send_state['fail']) + ' 退' + str(send_state['bounced'])
                         + ' 用时' + str(send_state['elapsed']) + '秒')
            if finish_callback:
                try:
                    finish_callback(send_state['status'], dict(send_state))
                except Exception:
                    pass

    if bridge_mode or run_async:
        # 后台模式: 引擎常驻后台, 立即返回 (桥接喂入 / 节点任务自主执行)
        if bridge_mode:
            bridge_enabled_flag['on'] = True
        threading.Thread(target=_run_pool, daemon=True).start()
        return True

    try:
        _run_pool()
    except Exception as _e:
        _set_error('任务异常: ' + str(_e))
        return False

    with state_lock:
        send_state['elapsed'] = round(time.time() - send_state['start_time'], 1)
        send_state['status'] = 'stopped' if send_stop_flag['stop'] else 'finished'
        send_state['speed'] = round(send_state['processed'] / max(send_state['elapsed'], 1), 1)
    _add_log('群发完成: 成' + str(send_state['success'])
             + ' 败' + str(send_state['fail']) + ' 退' + str(send_state['bounced'])
             + ' 用时' + str(send_state['elapsed']) + '秒')
    return True


# ============================================================
# CSV 结果批量落盘 (缓冲100条, 避免每封一次文件open/close)
# ============================================================
_write_buf = []
_write_buf_max = 100


def _write_rows_to_csv(rows):
    if not rows:
        return
    try:
        with open(SEND_RESULT_CSV, 'a', encoding='utf-8', newline='') as f:
            w = csv.writer(f)
            for row in rows:
                w.writerow(row)
    except Exception:
        pass


def _record_result(email, success, code, msg, proxy_id):
    global _write_buf
    row = [email, 'sent' if success else 'fail',
           str(code), (msg or '')[:120],
           time.strftime('%Y-%m-%d %H:%M:%S'), proxy_id]
    should_flush = False
    with csv_lock:
        _write_buf.append(row)
        if len(_write_buf) >= _write_buf_max:
            buf, _write_buf = _write_buf, []
            should_flush = True
    if should_flush:
        _write_rows_to_csv(buf)


def flush_csv():
    """把缓冲的结果全部落盘 (任务结束/停止时调用)"""
    global _write_buf
    with csv_lock:
        if not _write_buf:
            return
        buf, _write_buf = _write_buf, []
    _write_rows_to_csv(buf)


def _set_error(msg):
    with state_lock:
        send_state['status'] = 'error'
        send_state['error_msg'] = msg
    _add_log('[ERROR] ' + msg)


def stop_send():
    send_stop_flag['stop'] = True
    flush_csv()
    _add_log('收到停止命令...')


def send_single_test(to_addr, subject, html, task=None, from_email='', from_name=''):
    """单封测试发送 (IMAP进箱测试用): 复用节点任务配置(代理/魔改/DKIM), 同步发送
    返回 (ok, msg, code)
    """
    from core.sender import MailSender
    from core.models import MailBox, Email, ProxyIP
    task = task or {}
    send_cfg = task.get('send_cfg') or {}
    proxies = []
    for p in (task.get('proxies') or []):
        proxies.append(ProxyIP(
            ip=p.get('host', ''), port=int(p.get('port') or 0),
            proxy_type=p.get('proxy_type', 'socks5'),
            username=p.get('user', ''), pwd=p.get('pwd', ''), enable=True))
    boxes = []
    for b in (task.get('boxes') or []):
        boxes.append(MailBox(enable=True, email=b.get('email', ''),
                             user=b.get('email', ''), pwd=b.get('pwd', ''),
                             sender_name=b.get('sender_name', ''), random_user=False))
    if not boxes and from_email and '@' in from_email:
        boxes = [MailBox(enable=True, email=from_email, user=from_email,
                         pwd='', sender_name=from_name or '', random_user=False)]
    if not boxes:
        return False, '节点无发件箱配置 (请先在域名管理分配发件域名, 或传入from_email)', '0'
    if not proxies:
        return False, '节点无代理配置 (请先在代理IP池添加并分配)', '0'
    box = boxes[0]
    if from_email and '@' in from_email:
        box.email = from_email
        box.user = from_email
    if from_name:
        box.sender_name = from_name

    sender = MailSender()
    sender.log_callback = lambda lvl, msg: _add_log('[test][' + lvl + '] ' + msg)
    sender._disguise_default = bool(send_cfg.get('disguise_enabled', True))
    sender._disguise_mode = send_cfg.get('disguise_mode', 'mua')
    sender._disguise_opts = send_cfg.get('disguise_opts') or None
    sender._mail_charset = send_cfg.get('mail_charset', 'utf-8') or 'utf-8'
    dkim_signer = DKIMSignerWrap(task.get('dkim_keys') or {},
                                 sign_extended=bool(send_cfg.get('dkim_extended', True)),
                                 header_mode=send_cfg.get('dkim_header_mode') or None)
    email_obj = Email(email_address=to_addr)
    _add_log('[test] 单封测试发送 → ' + to_addr + ' (代理池' + str(len(proxies)) + '条)')
    last_msg, last_code = '未发送', '0'
    if not proxies:
        return False, 'error', '代理池为空, 无法测试发送', '0'
    for attempt in range(min(3, len(proxies) or 1)):
        proxy = proxies[attempt % len(proxies)]
        ok, code, msg, _ipr, sock = sender.send_direct(
            box, email_obj, subject, html, mail_type=1,
            helo_domain=send_cfg.get('helo_domain') or '',
            sender_name=box.sender_name,
            timeout=int(send_cfg.get('send_timeout') or 60),
            proxy=proxy, dkim_signer=dkim_signer,
            custom_headers=send_cfg.get('custom_headers') or {},
            disguise_opts=send_cfg.get('disguise_opts') or None,
        )
        if sock:
            try:
                sock.close()
            except Exception:
                pass
        last_msg, last_code = (msg or ''), str(code or '')
        if ok:
            return True, '发送成功 (' + str(code) + ')', str(code)
        _add_log('[test] 第' + str(attempt + 1) + '次尝试失败: ' + str(code) + ' ' + str(msg))
    return False, last_msg or '发送失败', last_code


def snapshot():
    with state_lock:
        d = dict(send_state)
    d['recent_log'] = d.get('log_lines', [])[-15:]
    return d


def reset_state():
    """重置状态(用于启动前)"""
    with state_lock:
        for k in ('processed', 'success', 'fail', 'bounced',
                  'ip_rejected', 'switches', 'sent_per_ip_max',
                  'elapsed', 'speed'):
            send_state[k] = 0
        send_state['error_msg'] = ''
        send_state['status'] = 'idle'
    # 清空桥接队列残留
    bridge_enabled_flag['on'] = False
    while not bridge_queue.empty():
        try:
            bridge_queue.get_nowait()
        except Exception:
            break


# 对外的单例(导出供 app.py 直接 import)
class _WebSendEngine:
    def start(self, config):
        return start_send(config)
    def stop(self):
        return stop_send()
    def snapshot(self):
        return snapshot()
    def reset(self):
        return reset_state()


web_send_engine = _WebSendEngine()