"""
邮件群发引擎 — 完整版
SMTP + DNS直投 + API中转 + EML发送 + 代理IP + 定时发送
暂停时段 + 邮箱黑名单 + 已发送缓存 + 阅读追踪 + 退信检测 + 词语替换
"""
import smtplib
import ssl
import socket
import socks
import threading
import time
import random
import re
import os
import json
import dns.resolver
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
from email.utils import formataddr, formatdate, make_msgid
from typing import List, Optional, Callable, Dict, Set
from datetime import datetime, timedelta

from core.models import MailBox, Email, Variable, Content, SubjectEx, Attachment, SendStats, ProxyIP


# ============================================================
# SOCKS直投辅助 — 并发安全(不污染全局socket)
# ============================================================
def _smtp_recv(sock, timeout=30):
    """读取完整 SMTP 响应 (支持多行 250-.../250 结尾; 处理TCP分段: 行可能只到一半)
    完整判定: 最后一行必须是 "NNN<SP>...\r\n" (行尾已到且第4字符为空格)
    返回 b'' 表示超时/断连"""
    sock.settimeout(timeout)
    buf = b''
    import time as _time
    start = _time.time()
    while len(buf) < 65536:
        # 完整响应判定: 以 \r\n 结尾, 且最后一个完整行满足 NNN<SP>
        if buf.endswith(b'\r\n'):
            last_line = buf[:-2].split(b'\r\n')[-1]
            if len(last_line) >= 4 and last_line[:3].isdigit() and last_line[3:4] == b' ':
                return buf
        try:
            c = sock.recv(4096)
            if not c:
                break
            buf += c
        except socket.timeout:
            break
        except Exception:
            break
        # 总超时保护 (即使不断分段也不死循环)
        if _time.time() - start > timeout:
            break
    return buf


import threading as _th
# SMTP 拟人化节奏模式: 'fast' (不sleep) / 'medium' (握手阶段EHLO后sleep,DATA后全速) / 'slow' (每命令sleep)
_smtp_pace_enabled = {'mode': 'fast'}  # 默认 fast 速度优先

def _smtp_send(sock, data):
    try:
        mode = _smtp_pace_enabled.get('mode', 'fast')
        if mode == 'slow':
            import random as _r
            import time as _t
            try:
                _t.sleep((50 + _r.randint(0, 150)) / 1000.0)
            except Exception:
                pass
        # 'medium' 模式: 只在 EHLO/STARTTLS 后加延迟, 由调用方在 EHLO 后显式 sleep
        # 'fast' 模式: 不 sleep
        sock.sendall(data)
        return True
    except Exception:
        return False


import struct as _struct

# 全局开关: 本地 VPN (Clash/快客/SSR 等本地 HTTP/SOCKS 代理在 127.0.0.1:vpn_port)
# 启用后链路: Python -> 本地VPN -> udealproxy -> 收件域MX:25  (大陆机器必须)
_local_vpn_config = {'enabled': False, 'port': 7890, 'type': 'http'}


def _socks5_handshake(s, proxy_ip, proxy_port, username, pwd, target_ip, target_port, timeout):
    """在已直连到 (proxy_ip, proxy_port) 的 socket 上手写 SOCKS5 协议
    建 SOCKS5 隧道到 (target_ip, target_port)
    返回 True 成功 / False 失败
    """
    try:
        s.settimeout(timeout)
        # 步骤1: 握手 (SOCKS5, 支持no-auth+用户密码)
        if username:
            s.sendall(b'\x05\x02\x00\x02')  # SOCKS5, 2 methods, no-auth + userpass
        else:
            s.sendall(b'\x05\x01\x00')  # SOCKS5, 1 method, no-auth
        m = s.recv(2)
        if len(m) < 2 or m[0] != 0x05:
            return False
        chosen = m[1]
        if chosen == 0x02:
            # 用户密码认证 (RFC 1929)
            u_bytes = (username or '').encode('utf-8')
            p_bytes = (pwd or '').encode('utf-8')
            if len(u_bytes) > 255 or len(p_bytes) > 255:
                return False
            s.sendall(b'\x01' + bytes([len(u_bytes)]) + u_bytes + bytes([len(p_bytes)]) + p_bytes)
            a = s.recv(2)
            if len(a) < 2 or a[1] != 0x00:
                return False
        elif chosen != 0x00:
            return False
        # 步骤2: 连接请求 (IPv4)
        try:
            ip_bytes = bytes([int(x) for x in target_ip.split('.')])
            if len(ip_bytes) != 4:
                raise ValueError
        except Exception:
            # 不是 IPv4, 走域名 (atype=0x03)
            # 这里 mx 是 IP, 不进
            return False
        port_bytes = _struct.pack('>H', target_port)
        s.sendall(b'\x05\x01\x00\x01' + ip_bytes + port_bytes)
        # 步骤3: 等响应
        r = s.recv(10)
        if len(r) < 2 or r[1] != 0x00:
            return False
        return True
    except Exception:
        return False


def _randomize_ttl(sock):
    """随机 IP TTL (64=Linux / 128=Windows / 255=类Unix), 打散 TCP 层设备指纹"""
    try:
        ttl = random.choice([64, 64, 64, 128, 128, 255, 63, 127])
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl)
    except Exception:
        pass


def _open_socks_tunnel(mx_ip, mx_port, proxy, timeout=30):
    """通过SOCKS5/HTTP代理建一条到(mx_ip, mx_port)的隧道
    返回socket对象或None

    ★ 本地 VPN 支持 (大陆机器必须):
    若 _local_vpn_config['enabled']=True, 链路变成:
        Python -> 本地VPN(127.0.0.1:vpn_port) -> udealproxy -> 远端
    实现: 先用 PySocks 让 VPN 连通到 (proxy.ip, proxy.port),
    得到 raw socket, 然后手写 SOCKS5 协议跟 udealproxy 协商抵达 (mx_ip, mx_port)
    """
    if proxy is None:
        return None
    try:
        if _local_vpn_config.get('enabled'):
            # ===== 两层代理: 本地VPN + udealproxy =====
            vpn_port = int(_local_vpn_config.get('port') or 7890)
            vpn_type_str = (_local_vpn_config.get('type') or 'http').lower()
            if vpn_type_str == 'socks5':
                vpn_t = socks.SOCKS5
            elif vpn_type_str == 'socks4':
                vpn_t = socks.SOCKS4
            else:
                vpn_t = socks.HTTP
            # 用本地 VPN 建一条到代理服务器的 raw TCP 连接 (VPN 替你转过去)
            s = socks.socksocket()
            s.set_proxy(vpn_t, '127.0.0.1', vpn_port)
            s.settimeout(timeout)
            _randomize_ttl(s)
            proxy_addr = (proxy.ip, int(proxy.port))
            s.connect(proxy_addr)
            # 现在 s 是直连 udealproxy 服务端口 的 raw socket
            # 手写 SOCKS5 协议跟 udealproxy 协商 (如果代理是 SOCKS5 类型)
            ptype = (proxy.proxy_type or 'socks5').lower()
            if ptype in ('socks5', 'socks4'):
                ok = _socks5_handshake(s, proxy.ip, proxy.port,
                                       proxy.username or '', proxy.pwd or '',
                                       mx_ip, int(mx_port), timeout)
                if not ok:
                    try: s.close()
                    except: pass
                    return None
                # 现在 s 已经是到 (mx_ip, mx_port) 的隧道
                return s
            elif ptype == 'http':
                # udealproxy 如果是 HTTP 类型: 走 HTTP CONNECT
                cred_enc = ''
                if proxy.username:
                    import base64 as _b64
                    cred = (proxy.username + ':' + (proxy.pwd or '')).encode('utf-8')
                    cred_enc = 'Proxy-Authorization: Basic ' + _b64.b64encode(cred).decode() + '\r\n'
                req = ('CONNECT ' + mx_ip + ':' + str(mx_port) + ' HTTP/1.1\r\nHost: ' + mx_ip + ':' + str(mx_port) + '\r\n' + cred_enc + '\r\n').encode()
                s.sendall(req)
                resp = b''
                while b'\r\n\r\n' not in resp:
                    c = s.recv(4096)
                    if not c: break
                    resp += c
                    if len(resp) > 4096: break
                if b'200' not in resp.split(b'\r\n')[0]:
                    try: s.close()
                    except: pass
                    return None
                return s
            return None
        # ===== 单层代理 (香港/海外直连 udealproxy) =====
        ptype = (proxy.proxy_type or 'socks5').lower()
        if ptype == 'socks5':
            socks_t = socks.SOCKS5
        elif ptype == 'socks4':
            socks_t = socks.SOCKS4
        else:
            socks_t = socks.HTTP
        s = socks.socksocket()
        s.set_proxy(socks_t, proxy.ip, proxy.port,
                    True if proxy.username else False,
                    proxy.username or '', proxy.pwd or '')
        s.settimeout(timeout)
        _randomize_ttl(s)
        s.connect((mx_ip, mx_port))
        return s
    except Exception:
        try:
            s.close()
        except Exception:
            pass
        return None


class SpintaxProcessor:
    """Spintax {a|b|c} 旋转语法"""

    @staticmethod
    def process(text: str, rng: random.Random = None) -> str:
        if rng is None:
            rng = random.Random()
        if '{' not in text:
            return text
        result = []
        i = 0
        depth = 0
        start = -1
        while i < len(text):
            if text[i] == '{':
                if depth == 0:
                    start = i
                depth += 1
            elif text[i] == '}':
                depth -= 1
                if depth == 0 and start >= 0:
                    block = text[start + 1:i]
                    options = SpintaxProcessor._split_options(block)
                    if len(options) > 1:
                        chosen = rng.choice(options)
                        result.append(SpintaxProcessor.process(chosen, rng))
                    else:
                        # 不含 | 的 {...} 不是 spintax (CSS/JS花括号等), 原样保留
                        result.append('{' + block + '}')
                    start = -1
            elif depth == 0:
                result.append(text[i])
            i += 1
        if depth > 0 and start >= 0:
            result.append(text[start:])
        return ''.join(result)

    @staticmethod
    def _split_options(block: str) -> List[str]:
        options = []
        current = []
        depth = 0
        for ch in block:
            if ch == '{':
                depth += 1
                current.append(ch)
            elif ch == '}':
                depth -= 1
                current.append(ch)
            elif ch == '|' and depth == 0:
                options.append(''.join(current))
                current = []
            else:
                current.append(ch)
        options.append(''.join(current))
        return [o for o in options if o]


class VariableProcessor:
    """变量替换 [变量名] — 支持is_multi_not_same/change_times"""

    def __init__(self, variables: List[Variable] = None):
        self.variables = {}
        if variables:
            for v in variables:
                self.variables[v.var_name] = v
        self.rng = random.Random()

    def replace(self, text: str, email: Email = None) -> str:
        if not text:
            return text

        def replace_var(match):
            var_name = match.group(1)
            if var_name in self.variables:
                var = self.variables[var_name]
                with var.locker:
                    if var.var_content:
                        if var.is_seq:
                            idx = var.use_times % len(var.var_content)
                            var.use_times += 1
                            return var.var_content[idx]
                        elif var.is_multi_not_same:
                            if var.use_times < len(var.var_content) and var.change_times > 0:
                                idx = var.use_times % len(var.var_content)
                                var.use_times += 1
                                return var.var_content[idx]
                            return self.rng.choice(var.var_content)
                        else:
                            if var.change_times > 0 and var.use_times >= var.change_times:
                                var.use_times = 0
                            var.use_times += 1
                            return self.rng.choice(var.var_content)
            return match.group(0)

        text = re.sub(r'\[([^\]]+)\]', replace_var, text)

        if email:
            text = text.replace('[email]', email.email_address)
            text = text.replace('[name]', email.name or '')
            text = text.replace('[var]', email.var_text or '')

        text = text.replace('[random]', str(self.rng.randint(10000, 99999)))
        text = text.replace('[timestamp]', str(int(time.time())))
        text = text.replace('[date]', time.strftime('%Y-%m-%d'))
        text = text.replace('[time]', time.strftime('%H:%M:%S'))

        return text


class WordReplacer:
    """词语替换 — 敏感词自动替换"""

    def __init__(self, replacements: List[tuple] = None):
        self.replacements = replacements or []

    def replace(self, text: str) -> str:
        for old, new in self.replacements:
            text = text.replace(old, new)
        return text


class MailCache:
    """已发送邮箱缓存 — 断点续发"""

    def __init__(self, cache_file: str = 'sent_cache.json'):
        self.cache_file = cache_file
        self.sent_set: Set[str] = set()
        self._lock = threading.Lock()
        self._load()

    def _load(self):
        if os.path.exists(self.cache_file):
            try:
                with open(self.cache_file, 'r', encoding='utf-8') as f:
                    self.sent_set = set(json.load(f))
            except:
                self.sent_set = set()

    def _save(self):
        try:
            with open(self.cache_file, 'w', encoding='utf-8') as f:
                json.dump(list(self.sent_set), f)
        except:
            pass

    def is_sent(self, email: str) -> bool:
        with self._lock:
            return email in self.sent_set

    def mark_sent(self, email: str):
        with self._lock:
            self.sent_set.add(email)
            if len(self.sent_set) % 50 == 0:
                self._save()

    def clear(self):
        with self._lock:
            self.sent_set.clear()
            self._save()

    def remove(self, email: str):
        with self._lock:
            self.sent_set.discard(email)


class EmailBlacklist:
    """邮箱黑名单"""

    def __init__(self, blacklist_file: str = 'email_blacklist.json'):
        self.blacklist_file = blacklist_file
        self.blacklist: Set[str] = set()
        self._lock = threading.Lock()
        self._load()

    def _load(self):
        if os.path.exists(self.blacklist_file):
            try:
                with open(self.blacklist_file, 'r', encoding='utf-8') as f:
                    self.blacklist = set(json.load(f))
            except:
                self.blacklist = set()

    def _save(self):
        try:
            with open(self.blacklist_file, 'w', encoding='utf-8') as f:
                json.dump(list(self.blacklist), f)
        except:
            pass

    def is_blacklisted(self, email: str) -> bool:
        with self._lock:
            return email.lower() in self.blacklist

    def add(self, email: str):
        with self._lock:
            self.blacklist.add(email.lower())
            self._save()

    def add_batch(self, emails: List[str]):
        with self._lock:
            for e in emails:
                self.blacklist.add(e.lower())
            self._save()

    def remove(self, email: str):
        with self._lock:
            self.blacklist.discard(email.lower())
            self._save()

    def clear(self):
        with self._lock:
            self.blacklist.clear()
            self._save()


class ProxyManager:
    """代理IP轮换管理"""

    def __init__(self, proxies: List[ProxyIP] = None, config: dict = None):
        self.proxies = [p for p in (proxies or []) if p.enable]
        self.config = config or {}
        self.current_idx = 0
        self._lock = threading.Lock()
        self.per_ip_count = self.config.get('per_ip_count', 50)
        self.auto_switch = self.config.get('auto_switch', False)
        self.current_count = 0

    def get_current(self) -> Optional[ProxyIP]:
        with self._lock:
            if not self.proxies:
                return None
            return self.proxies[self.current_idx]

    def switch(self):
        with self._lock:
            if not self.proxies:
                return
            self.current_idx = (self.current_idx + 1) % len(self.proxies)
            self.current_count = 0

    def on_sent(self):
        with self._lock:
            self.current_count += 1
            if self.auto_switch and self.current_count >= self.per_ip_count:
                self.current_idx = (self.current_idx + 1) % len(self.proxies)
                self.current_count = 0

    def on_fail(self, proxy: ProxyIP):
        with self._lock:
            proxy.fail_count += 1
            if self.auto_switch and proxy.fail_count >= 3:
                proxy.enable = False
                self.switch()

    def rotate(self) -> Optional[ProxyIP]:
        """强制切换到下一个可用代理，返回新的代理"""
        with self._lock:
            if not self.proxies:
                return None
            for _ in range(len(self.proxies)):
                self.current_idx = (self.current_idx + 1) % len(self.proxies)
                self.current_count = 0
                if self.proxies[self.current_idx].enable:
                    return self.proxies[self.current_idx]
            return None

    def apply_to_socket(self, proxy: ProxyIP):
        """设置socks代理"""
        if proxy and proxy.proxy_type in ('socks5', 'socks4', 'http'):
            proxy_type = socks.SOCKS5 if proxy.proxy_type == 'socks5' else (
                socks.SOCKS4 if proxy.proxy_type == 'socks4' else socks.HTTP)
            socks.set_default_proxy(proxy_type, proxy.ip, proxy.port,
                                   True, proxy.username, proxy.pwd)
            socket.socket = socks.socksocket


class ADSLManager:
    """ADSL拨号换IP"""

    def __init__(self, config: dict = None):
        self.config = config or {}
        self.connection_name = self.config.get('connection_name', '宽带连接')
        self.username = self.config.get('username', '')
        self.password = self.config.get('password', '')
        self.auto_dial = self.config.get('auto_dial', False)
        self.fail_threshold = self.config.get('fail_threshold', 5)
        self.wait_seconds = self.config.get('wait_seconds', 10)

    def get_current_ip(self) -> str:
        try:
            import requests as req
            for url in ['https://api.ipify.org', 'https://ifconfig.me', 'https://icanhazip.com']:
                try:
                    resp = req.get(url, timeout=10)
                    if resp.status_code == 200:
                        return resp.text.strip()
                except:
                    continue
        except:
            pass
        return ''

    def dial(self) -> bool:
        try:
            import subprocess
            subprocess.run(['rasdial', self.connection_name, '/disconnect'],
                         capture_output=True, timeout=30)
            time.sleep(3)
            result = subprocess.run(
                ['rasdial', self.connection_name, self.username, self.password],
                capture_output=True, timeout=60)
            if result.returncode == 0:
                time.sleep(self.wait_seconds)
                return True
            return False
        except:
            return False


class TrackingInserter:
    """阅读追踪代码插入"""

    @staticmethod
    def insert_pixel(html_content: str, email_addr: str, track_url: str) -> str:
        import hashlib
        track_id = hashlib.md5(f'{email_addr}{time.time()}'.encode()).hexdigest()[:16]
        pixel = f'<img src="{track_url}/open?id={track_id}&email={email_addr}" width="1" height="1" border="0" style="display:none"/>'
        if '</body>' in html_content.lower():
            return re.sub(r'</body>', f'{pixel}</body>', html_content, flags=re.IGNORECASE)
        return html_content + pixel

    @staticmethod
    def insert_link_tracking(html_content: str, track_url: str) -> str:
        def replace_link(match):
            href = match.group(1)
            if href.startswith('http') and track_url not in href:
                return f'href="{track_url}/click?url={href}"'
            return match.group(0)
        return re.sub(r'href="([^"]+)"', replace_link, html_content)


class BounceDetector:
    """退信检测"""

    def __init__(self, bounce_email: str = '', callback: Callable = None):
        self.bounce_email = bounce_email
        self.callback = callback
        self.bounce_count = 0
        self.success_count = 0
        self.invalid_count = 0

    def on_result(self, email: str, success: bool, smtp_code: str, error_msg: str):
        if success:
            self.success_count += 1
        else:
            self.bounce_count += 1
            # 退信分类 (软退信/硬退信/无效): 就地判定, 不依赖可选模块
            category = self._classify_bounce(error_msg, smtp_code)
            if category == 'invalid':
                self.invalid_count += 1
            if self.callback:
                self.callback(email, smtp_code, error_msg, category)

    @staticmethod
    def _classify_bounce(error_msg: str, smtp_code) -> str:
        """退信分类: hard(邮箱不存在/被拒) / soft(临时故障) / invalid(格式错)
        与原 BounceAnalyzer.classify 兼容的返回值"""
        msg = str(error_msg or '').lower()
        try:
            code = int(str(smtp_code)[:3])
        except Exception:
            code = 0
        # 硬退信: 用户不存在/域名无效/被拉黑
        if code in (550, 551, 553) or any(k in msg for k in (
                'user unknown', 'no such user', 'user not found', 'does not exist',
                'unknown user', 'invalid mailbox', 'mailbox unavailable',
                'address rejected', 'no mailbox', 'bad recipient')):
            return 'invalid'
        # 软退信: 临时故障/限流/灰名单
        if code in (421, 450, 451, 452) or any(k in msg for k in (
                'try again', 'temporar', 'greylist', 'grey list', 'deferred',
                'busy', 'too many', 'quota', 'full', 'timeout')):
            return 'soft'
        return 'hard'


class MailSender:
    """邮件发送核心引擎 — 完整版"""

    def __init__(self):
        self.stats = SendStats()
        self._stop = False
        self._threads: List[threading.Thread] = []
        self._lock = threading.Lock()
        self.log_callback: Optional[Callable[[str, str], None]] = None
        self.stats_callback: Optional[Callable[[SendStats], None]] = None
        self.email_callback: Optional[Callable[[int, str, int], None]] = None
        self.smart_callback: Optional[Callable] = None
        self.magic_config: dict = {}
        self.bounce_callback: Optional[Callable] = None

        self.mail_cache: Optional[MailCache] = None
        self.blacklist: Optional[EmailBlacklist] = None
        self.proxy_manager: Optional[ProxyManager] = None
        self.adsl_manager: Optional[ADSLManager] = None
        self.word_replacer: Optional[WordReplacer] = None
        self.bounce_detector: Optional[BounceDetector] = None
        self.track_config: dict = {}
        self.sender_names: List[str] = []
        self.paused = False
        self.pause_until: float = 0

    def log(self, level: str, msg: str):
        if self.log_callback:
            self.log_callback(level, msg)

    def update_stats(self):
        if self.stats_callback:
            self.stats_callback(self.stats)

    def update_email(self, idx: int, info: str, status: int):
        if self.email_callback:
            self.email_callback(idx, info, status)

    def stop(self):
        self._stop = True
        self.log('info', '正在停止发送...')
        for t in self._threads:
            if t.is_alive():
                t.join(timeout=5)
        self._threads.clear()
        self.log('info', '发送已停止')

    def _check_pause(self, box: MailBox) -> bool:
        """检查邮箱暂停时段"""
        if not box.pause_second_limit or box.pause_second <= 0:
            return False
        now = time.time()
        if box.pause_first_time:
            try:
                elapsed = now - float(box.pause_first_time)
                if elapsed < box.pause_second:
                    return True
            except:
                pass
        return False

    def _build_message(self, box: MailBox, email: Email, subject: str,
                       content: str, mail_type: int = 1,
                       attachments: List[Attachment] = None,
                       helo_domain: str = None, sender_name: str = None,
                       reply_to: str = None, custom_headers: dict = None,
                       disguise: bool = False,
                       verp_enabled: bool = False,
                       disguise_opts: dict = None) -> str:
        msg = MIMEMultipart('mixed' if attachments else 'alternative')
        from_addr = box.email or box.user
        display_name = sender_name or box.sender_name or ''
        domain = helo_domain or (from_addr.split('@')[-1] if '@' in from_addr else 'localhost')

        # ===== 反指纹伪装 (可选) =====
        disguise_headers = None
        if disguise:
            try:
                _pd = None
                # 方式1: 直接 import (web_send 已把 jpmail 目录加进 sys.path, 节点部署布局适用)
                try:
                    import print_disguise as _pd
                except ImportError:
                    _pd = None
                if _pd is None:
                    # 方式2: 文件路径加载 (开发布局: 项目根/JP邮箱验证系统/print_disguise.py)
                    import importlib.util as _ilu
                    _this_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
                    for _cand in (os.path.join(_this_dir, 'JP邮箱验证系统', 'print_disguise.py'),
                                  os.path.join(os.path.dirname(_this_dir), 'jpmail', 'print_disguise.py'),
                                  os.path.join(_this_dir, 'jpmail', 'print_disguise.py')):
                        if os.path.exists(_cand):
                            _spec = _ilu.spec_from_file_location('print_disguise', _cand)
                            _pd = _ilu.module_from_spec(_spec)
                            _spec.loader.exec_module(_pd)
                            break
                if _pd is not None:
                    disguise_headers = _pd.generate_disguise_headers(
                        from_addr, display_name, email.email_address, subject, custom_headers,
                        mode=getattr(self, '_disguise_mode', 'mua'),
                        opts=disguise_opts if disguise_opts is not None else getattr(self, '_disguise_opts', None))
                    custom_headers = {k: v for k, v in (custom_headers or {}).items() if not k.startswith('_')}
            except Exception as _e:
                try:
                    self.log('warning', f'伪装头生成失败(忽略): {_e}')
                except Exception:
                    pass

        # ===== 根据 MUA 真实结构重构 multipart =====
        # 默认 Python email lib 用 multipart/alternative
        # 但 Outlook 用 multipart/mixed -> multipart/alternative; Thunderbird 用 multipart/alternative
        if disguise_headers:
            struct_type = disguise_headers.get('_multipart_structure', 'alt')
            if False and struct_type == 'mixed_alt' or attachments:  # 强制不用 mixed_alt(日系邮局解析嵌套有问题)
                # 重新建一个新的 mixed 根 + 内部 alternative 容器
                msg = MIMEMultipart('mixed')
                alt_msg = MIMEMultipart('alternative')
            else:
                msg = MIMEMultipart('alternative')
                alt_msg = None
        elif attachments:
            msg = MIMEMultipart('mixed')
            alt_msg = None
        else:
            msg = MIMEMultipart('alternative')
            alt_msg = None

        # ★ From 显示名 RFC2047 编码 (日文显示名不乱码; ISO-2022-JP 原生编码支持)
        try:
            from email.header import Header
            from email.utils import formataddr as _fa
            _hdr_cs = 'iso-2022-jp' if (getattr(self, '_mail_charset', 'utf-8') or 'utf-8').lower() in ('iso-2022-jp', 'iso2022jp') else 'utf-8'
            # 注意: formataddr 内部会强制 utf-8 重编码, 需先用 Header.encode() 生成编码词
            _name_enc = str(Header(str(display_name), _hdr_cs).encode())
            msg['From'] = _fa((_name_enc, from_addr))
        except Exception:
            msg['From'] = formataddr((display_name, from_addr))
        msg['To'] = formataddr((email.name or '', email.email_address))
        # ★ 主题处理 (个性化 + 避开垃圾词) — 必须在 Subject 写入前完成
        #   (Python email库对 Header 对象重复赋值是追加不是替换, 会产出重复Subject头)
        _sflags = getattr(self, '_feature_flags', {})
        try:
            _pd_subj = locals().get('_pd')  # 伪装块里已加载; 未伪装时为 None
            if _pd_subj is None:
                try:
                    import print_disguise as _pd_subj
                except ImportError:
                    _pd_subj = None
            if _sflags.get('personalize_subject') and _pd_subj is not None:
                subject = _pd_subj.personalize_subject(subject, email.email_address)
            if _sflags.get('avoid_spam_words') and _pd_subj is not None:
                subject = _pd_subj.avoid_spam_words(subject)
        except Exception:
            pass
        # ★ 主题 RFC2047 编码 (不管认证/裸发都要做, 否则日文乱码; ISO-2022-JP 原生编码支持)
        try:
            from email.header import Header
            _hdr_cs2 = 'iso-2022-jp' if (getattr(self, '_mail_charset', 'utf-8') or 'utf-8').lower() in ('iso-2022-jp', 'iso2022jp') else 'utf-8'
            msg['Subject'] = Header(subject, _hdr_cs2)
        except Exception:
            msg['Subject'] = subject

        if disguise_headers:
            # 用伪装头覆盖默认头
            if disguise_headers.get('Message-ID'):
                msg['Message-ID'] = disguise_headers['Message-ID']
            else:
                msg['Message-ID'] = make_msgid(domain=domain)
            if disguise_headers.get('Date'):
                msg['Date'] = disguise_headers['Date']
            else:
                msg['Date'] = formatdate(localtime=True)
            if disguise_headers.get('Return-Path'):
                try:
                    rp = disguise_headers['Return-Path']
                    # VERP: Return-Path 含收件人编码
                    if verp_enabled and '@' in email.email_address:
                        recp_coded = email.email_address.replace('@', '=')
                        from_domain = (box.email or box.user or '').split('@')[-1] if '@' in (box.email or box.user or '') else 'localhost'
                        rp = f'<bounces+{recp_coded}@{from_domain}>'
                    msg['Return-Path'] = rp
                except Exception:
                    pass
            # 伪 Received 头必须最早出现 (邮件头顺序也是指纹之一)
            # ★ 存到 _build_message 的返回载体 (msg 属性), 发送侧用完即弃 —
            #   不再挂共享实例属性 (旧实现跨线程残留: A线程的Received链混进B线程邮件)
            if disguise_headers.get('_RECEIVED_CHAIN'):
                try:
                    msg._received_chain = disguise_headers['_RECEIVED_CHAIN']
                except Exception:
                    pass
            if disguise_headers.get('X-Mailer'):
                msg['X-Mailer'] = disguise_headers['X-Mailer']
            if disguise_headers.get('User-Agent'):
                msg['User-Agent'] = disguise_headers['User-Agent']
            if disguise_headers.get('Feedback-ID'):
                msg['Feedback-ID'] = disguise_headers['Feedback-ID']
            if disguise_headers.get('List-Id'):
                msg['List-Id'] = disguise_headers['List-Id']
            if disguise_headers.get('X-Priority'):
                msg['X-Priority'] = disguise_headers['X-Priority']
            if disguise_headers.get('Importance'):
                msg['Importance'] = disguise_headers['Importance']
            # extra headers (callable 已展开为字符串)
            for k, v in disguise_headers.items():
                if k.startswith('_') or k in ('Message-ID', 'Date', 'Return-Path',
                                                'X-Mailer', 'User-Agent', 'Feedback-ID',
                                                'List-Id', 'X-Priority', 'Importance',
                                                'Received'):
                    continue
                try:
                    msg[k] = v
                except Exception:
                    pass
            # 用伪装的 boundary 替换默认的
            if disguise_headers.get('_boundary'):
                try:
                    msg.set_boundary(disguise_headers['_boundary'])
                except Exception:
                    pass
            # HTML 正文变形 + 脚注 + 深度反指纹 (句序隆替 + 同义词 + 追踪像素剥)
            if mail_type == 1:
                try:
                    # 深度变形 (安全版: 同义词替换+零宽字符穿插, 不截断不破结构)
                    content = _pd.disguise_html_body(content)
                    content = _pd.disguise_html_deep(content)
                    # 注入 MUA 风格 META GENERATOR (反检测)
                    mua_name_for_meta = disguise_headers.get('_mua_name', '')
                    content = _pd.inject_meta_generator(content, mua_name_for_meta)
                    # 客户端购脚
                    content = _pd.add_footer_marker(content, disguise_headers.get('_footer_text', ''))
                    # ★ 进箱黑科技 (5个独立功能, 通过 self._feature_flags 控制)
                    flags = getattr(self, '_feature_flags', {})
                    if flags.get('embed_images'):
                        content = _pd.embed_images_base64(content)
                    if flags.get('invisible_segments'):
                        content = _pd.insert_invisible_segments(content)
                    if flags.get('timestamp_personalize'):
                        content = _pd.add_timestamp_personalization(content)
                except Exception:
                    pass
        else:
            msg['Date'] = formatdate(localtime=True)
            msg['Message-ID'] = make_msgid(domain=domain)

        if reply_to:
            msg['Reply-To'] = reply_to
        # ★ 只加 disguise_headers 里没出现过的 key, 避免重复 (如 LIST-UNSUBSCRIBE 被加2次)
        _disguise_keys = set()
        if disguise_headers:
            _disguise_keys = set(k for k in disguise_headers.keys() if not k.startswith('_'))
        if custom_headers:
            for k, v in custom_headers.items():
                if k.startswith('_') or k in _disguise_keys:
                    continue  # 已被 disguise_headers 加过, 跳过
                try:
                    msg[k] = v
                except Exception:
                    pass

        # ===== 处理 multipart 结构 =====
        # 结构: alt (单一 alternative) → text_part 直接挂到 msg
        #       mixed_alt (Outlook/含附件) → text_part 挂到 alt_msg, alt_msg 挂到 msg
        # ★ 字符集: 默认 utf-8, 支持 ISO-2022-JP 等日文原生编码 (PHPMailer Pro 同款)
        charset = getattr(self, '_mail_charset', 'utf-8') or 'utf-8'
        _cs = 'iso-2022-jp' if charset.lower() in ('iso-2022-jp', 'iso2022jp') else charset
        # ISO-2022-JP 无法表示的字(emoji等)先剔除, 避免编码崩溃
        try:
            if _cs != 'utf-8':
                content = content.encode(_cs, errors='replace').decode(_cs)
        except Exception:
            pass
        if mail_type == 1:
            text_part = MIMEText(content, 'html', _cs)
        else:
            text_part = MIMEText(content, 'plain', _cs)

        if alt_msg is not None:
            # 严格 multipart/mixed -> multipart/alternative 结构
            alt_msg.attach(text_part)
            # 如果有纯文本副本, 也加入 alt_msg
            try:
                text_body = ''
                # 兜底: 主调方传 text_parts 时, 此处可扩展; 当前 mail_type 单一文本即 OK
                # 这里我们用一个最简纯文本 fallback
                if mail_type == 1:
                    # 从 HTML 抽裸文本作为 fallback
                    import re as _re_mod
                    text_body = _re_mod.sub(r'<[^>]+>', '', content or '')
                    text_body = _re_mod.sub(r'\n\s*\n', '\n\n', text_body).strip()
                if text_body:
                    alt_msg.attach(MIMEText(text_body, 'plain', _cs))
            except Exception:
                pass
            msg.attach(alt_msg)
        else:
            msg.attach(text_part)

        if attachments:
            for att in attachments:
                try:
                    with open(att.file_path, 'rb') as f:
                        part = MIMEBase('application', 'octet-stream')
                        part.set_payload(f.read())
                        encoders.encode_base64(part)
                        part.add_header('Content-Disposition',
                                        f'attachment; filename="{att.filename}"')
                        if att.content_id:
                            part.add_header('Content-ID', f'<{att.content_id}>')
                        msg.attach(part)
                except Exception as e:
                    self.log('error', f'附件加载失败 {att.filename}: {e}')
        return msg.as_bytes().decode('utf-8').replace('\n', '\r\n')

    def send_smtp(self, box: MailBox, email: Email, subject: str,
                  content: str, mail_type: int = 1,
                  attachments: List[Attachment] = None,
                  helo_domain: str = None, sender_name: str = None,
                  reply_to: str = None, custom_headers: dict = None,
                  timeout: int = 60, proxy: ProxyIP = None) -> tuple:
        smtp = None
        try:
            msg = self._build_message(box, email, subject, content, mail_type,
                                      attachments, helo_domain, sender_name,
                                      reply_to, custom_headers)
            from_addr = box.email or box.user

            if proxy:
                import socks as s
                proxy_type = s.SOCKS5 if proxy.proxy_type == 'socks5' else s.SOCKS4
                s.set_default_proxy(proxy_type, proxy.ip, proxy.port,
                                   True if proxy.username else False,
                                   proxy.username, proxy.pwd)
                socket.socket = s.socksocket

            if box.ssl_tls == 'ssl':
                context = ssl.create_default_context()
                context.check_hostname = False
                context.verify_mode = ssl.CERT_NONE
                smtp = smtplib.SMTP_SSL(box.smtp, box.port, timeout=timeout, context=context)
            else:
                smtp = smtplib.SMTP(box.smtp, box.port, timeout=timeout)
                if box.ssl_tls in ('tls', 'auto'):
                    try:
                        smtp.starttls()
                    except:
                        pass

            smtp.ehlo(helo_domain or (from_addr.split('@')[-1] if '@' in from_addr else 'localhost'))
            if box.pwd:
                user = box.user if '@' in box.user else box.email
                smtp.login(user, box.pwd)
            smtp.sendmail(from_addr, [email.email_address], msg)
            smtp.quit()
            with box.locker:
                box.sent += 1
                box.sent_succ += 1
                box.day_sent += 1
                box.series_fail = 0
            return True, '250', 'OK'
        except smtplib.SMTPRecipientsRefused as e:
            with box.locker:
                box.sent += 1
                box.series_fail += 1
                box.info = str(e)
            return False, '550', str(e)
        except smtplib.SMTPAuthenticationError as e:
            with box.locker:
                box.sent += 1
                box.series_fail += 1
                box.info = f'认证失败: {e}'
            return False, '535', str(e)
        except smtplib.SMTPConnectError as e:
            with box.locker:
                box.sent += 1
                box.series_fail += 1
                box.info = f'连接失败: {e}'
            return False, 'connect', str(e)
        except socket.timeout:
            with box.locker:
                box.sent += 1
                box.series_fail += 1
                box.info = '连接超时'
            return False, 'timeout', '连接超时'
        except Exception as e:
            with box.locker:
                box.sent += 1
                box.series_fail += 1
                box.info = str(e)
            return False, 'unknown', str(e)
        finally:
            if smtp:
                try:
                    smtp.close()
                except:
                    pass
            if proxy:
                socket.socket = __import__('socket').SocketType

    def send_direct(self, box: MailBox, email: Email, subject: str,
                    content: str, mail_type: int = 1,
                    dns_servers: List[str] = None,
                    helo_domain: str = None, sender_name: str = None,
                    timeout: int = 60, proxy: ProxyIP = None,
                    dkim_signer: 'Callable' = None,
                    custom_headers: dict = None,
                    prebuilt_sock: 'object' = None,
                    verp_enabled: bool = False,
                    disguise_opts: dict = None) -> tuple:
        """SMTP直投 — 支持SOCKS代理隧道 + DKIM签名
        prebuilt_sock: 若调用方已建立好长连接SOCKS隧道(EHLO已握手),直接复用该socket
                       此模式下本函数只执行 MAIL FROM/RCPT TO/DATA 流程
                       返回时按结果决定调用方是否要换IP
        disguise_opts: 本封专属伪装配置 (智能个性化按收件域匹配, 线程安全)
        返回: (success, smtp_code, error_msg, ip_rejected, raw_sock)
              ip_rejected=True 表示IP/隧道被拒,调用方应立即换IP
              raw_sock: 若连接可复用(RSET成功),回传给调用方; None表示连接已废
        """
        # 兼容旧调用约定(无 proxy 无 prebuilt_sock):走真实 smtplib 直投路径
        if proxy is None and prebuilt_sock is None:
            try:
                domain = email.email_address.split('@')[-1]
                mx_records = self._query_mx(domain, dns_servers)
                if not mx_records:
                    raise Exception(f'无MX记录: {domain}')
                mx_host = mx_records[0]
                from_addr = box.email or box.user
                _disguise = getattr(self, '_disguise_default', True)
                msg = self._build_message(box, email, subject, content, mail_type,
                                          None, helo_domain, sender_name,
None, custom_headers, disguise=_disguise, verp_enabled=verp_enabled,
                                          disguise_opts=disguise_opts)
                msg_b = msg.encode('utf-8') if isinstance(msg, str) else msg
                if dkim_signer:
                    try:
                        msg_b = dkim_signer(msg_b, from_addr.split('@')[-1] if '@' in from_addr else helo_domain or 'localhost')
                        if isinstance(msg_b, str): msg_b = msg_b.encode('utf-8')
                    except Exception as _e:
                        self.log('warning', 'DKIM签名失败(忽略): ' + str(_e))
                # ★ 真正通过 smtplib 发送 (原实现缺失此段, 必然失败)
                import smtplib as _smtplib
                helo_d = helo_domain or (from_addr.split('@')[-1] if '@' in from_addr else 'localhost')
                smtp = _smtplib.SMTP(mx_host, 25, timeout=timeout)
                try:
                    code, _ = smtp.ehlo(helo_d)
                    if code == 250:
                        try:
                            smtp.starttls()
                            smtp.ehlo(helo_d)
                        except Exception:
                            pass  # STARTTLS 不可用则明文
                    smtp.sendmail(from_addr, [email.email_address], msg_b)
                    with box.locker:
                        box.sent += 1
                        box.info = '直投成功'
                    return True, '250', 'OK', False, None
                finally:
                    try:
                        smtp.close()
                    except Exception:
                        pass
            except Exception as e:
                with box.locker:
                    box.sent += 1
                    box.series_fail += 1
                    box.info = str(e)
                return False, 'unknown', str(e), False, None

        # ===== SOCKS路径 (proxy 或 prebuilt_sock 非 None) =====
        sock = prebuilt_sock
        from_addr = box.email or box.user
        helo_d = helo_domain or (from_addr.split('@')[-1] if '@' in from_addr else 'localhost')
        ip_rejected = False
        try:
            # ★ 构造邮件 + DKIM签名 (SOCKS路径必须有这步)
            _disguise = getattr(self, '_disguise_default', True)
            _built = self._build_message(box, email, subject, content, mail_type,
                                      None, helo_domain, sender_name,
                                      None, custom_headers, disguise=_disguise, verp_enabled=verp_enabled,
                                      disguise_opts=disguise_opts)
            msg = _built.encode('utf-8') if isinstance(_built, str) else _built
            # ★ 手动插入伪 Received 头 (不能通过Python email库设, 会报 embedded header)
            #   从本封邮件对象取 (用完即弃, 不再跨线程/跨邮件残留)
            received_chain = getattr(_built, '_received_chain', '') or ''
            if received_chain and isinstance(msg, bytes):
                recv_bytes = received_chain.encode('utf-8') + b'\r\n'
                # 在第一个头之前插入
                msg = recv_bytes + msg
            if dkim_signer:
                try:
                    msg = dkim_signer(msg, helo_d)
                    if isinstance(msg, str):
                        msg = msg.encode('utf-8')
                except Exception as _e:
                    self.log('warning', 'DKIM签名失败(忽略): ' + str(_e))

            # MAIL FROM — VERP 启用时拼装 bounce 地址(精确退信归因)
            if verp_enabled and '@' in email.email_address:
                recp_coded = email.email_address.replace('@', '=')
                from_domain = from_addr.split('@')[-1] if '@' in from_addr else 'localhost'
                mail_from_addr = f'bounces+{recp_coded}@{from_domain}'
            else:
                mail_from_addr = from_addr
            mf = f'MAIL FROM:<{mail_from_addr}>\r\n'.encode()
            if not _smtp_send(sock, mf):
                try: sock.close()
                except: pass
                return False, 'sock_closed', '发MAIL FROM时连接断', True, None
            mf_resp = _smtp_recv(sock, timeout=timeout)
            mf_code = mf_resp[:3].decode('utf-8', 'replace')
            if not mf_resp.startswith(b'250'):
                # 5xx MAIL FROM = IP/发件域信誉 → 换IP
                try:
                    sock.close()
                except: pass
                with box.locker:
                    box.sent += 1
                    box.series_fail += 1
                    box.info = f'MAIL FROM被拒({mf_code})'
                return False, mf_code, f'MAIL FROM被拒: {mf_resp[:80]!r}', True, None

            # RCPT TO
            rcpt = f'RCPT TO:<{email.email_address}>\r\n'.encode()
            if not _smtp_send(sock, rcpt):
                try: sock.close()
                except: pass
                return False, 'sock_closed', '发RCPT TO时连接断', True, None
            rcpt_resp = _smtp_recv(sock, timeout=timeout)
            rcpt_code = rcpt_resp[:3].decode('utf-8', 'replace')
            if rcpt_resp.startswith(b'250') or rcpt_resp.startswith(b'251'):
                pass  # 邮箱存在 可继续
            else:
                # 550/551/552 = 邮箱本身错,不是IP错 → 不换IP,但要把连接RSET复用
                # 421/451 etc = 可能是IP问题 → RSET失败就关连接让上层换IP
                if rcpt_resp.startswith(b'4') :
                    # 4xx通常IP问题 → 换IP
                    try:
                        _smtp_send(sock, b'RSET\r\n')
                        _smtp_recv(sock, timeout=timeout)
                        sock.close()
                    except: 
                        try: sock.close()
                        except: pass
                    with box.locker:
                        box.sent += 1
                        box.series_fail += 1
                        box.info = f'RCPT暂时被拒({rcpt_code})'
                    return False, rcpt_code, f'RCPT暂时被拒: {rcpt_resp[:80]!r}', True, None
                else:
                    # 5xx(550/551/552) = 邮箱错,尝试RSET复用连接
                    rset_ok = False
                    try:
                        _smtp_send(sock, b'RSET\r\n')
                        rset_resp = _smtp_recv(sock, timeout=timeout)
                        if rset_resp.startswith(b'250'):
                            rset_ok = True
                    except: pass
                    if not rset_ok:
                        try: sock.close()
                        except: pass
                        # RSET失败,连接废了,IP视角也算"被拒"
                        with box.locker:
                            box.sent += 1
                            box.series_fail += 1
                            box.info = f'邮箱无效且RSET失败({rcpt_code})'
                        return False, rcpt_code, f'邮箱无效: {rcpt_resp[:80]!r}', True, None
                    with box.locker:
                        box.sent += 1
                        box.series_fail += 1
                        box.info = f'邮箱无效({rcpt_code})'
                    return False, rcpt_code, f'邮箱无效: {rcpt_resp[:80]!r}', False, sock

            # DATA
            if not _smtp_send(sock, b'DATA\r\n'):
                try: sock.close()
                except: pass
                return False, 'sock_closed', '发DATA命令时连接断', True, None
            data_resp = _smtp_recv(sock, timeout=timeout)
            if not data_resp.startswith(b'354'):
                # 554 = 内容被定为垃圾 → IP/内容问题 → 换IP
                code = data_resp[:3].decode('utf-8', 'replace')
                try: sock.close()
                except: pass
                with box.locker:
                    box.sent += 1
                    box.series_fail += 1
                    box.info = f'DATA被拒({code})'
                return False, code, f'DATA被拒: {data_resp[:80]!r}', True, None

            # 发内容+CRLF.CRLF
            # msg 可能是 bytes (DKIM签名后) 或 str (未签名)
            if isinstance(msg, bytes):
                data_bytes = msg
            elif hasattr(msg, 'as_bytes'):
                data_bytes = msg.as_bytes()
            elif hasattr(msg, 'encode'):
                data_bytes = msg.encode('utf-8')
            else:
                data_bytes = bytes(msg)
            # ★ DKIM签名后的bytes不再做行尾规范化 (会破坏body hash)
            # 只有未签名的才需要规范化
            if not dkim_signer:
                data_bytes = data_bytes.replace(b'\r\n', b'\n').replace(b'\n', b'\r\n')
            if not data_bytes.endswith(b'\r\n'):
                data_bytes += b'\r\n'
            # ★ RFC 5321 dot-stuffing: 正文里以 '.' 开头的行必须转义为 '..'
            #   否则被服务器当作 DATA 终止符 → 邮件截断 + DKIM body hash 作废
            data_bytes = re.sub(br'(?m)^\.', b'..', data_bytes)
            data_bytes += b'.\r\n'
            if not _smtp_send(sock, data_bytes):
                try: sock.close()
                except: pass
                return False, 'sock_closed', '发邮件正文时连接断', True, None
            final_resp = _smtp_recv(sock, timeout=timeout)
            final_code = final_resp[:3].decode('utf-8', 'replace')
            if final_resp.startswith(b'250'):
                # 成功! RSET准备下一条
                rset_ok = False
                try:
                    _smtp_send(sock, b'RSET\r\n')
                    rset_resp = _smtp_recv(sock, timeout=timeout)
                    if rset_resp.startswith(b'250'):
                        rset_ok = True
                except: pass
                with box.locker:
                    box.sent += 1
                    box.sent_succ += 1
                    box.day_sent += 1
                    box.series_fail = 0
                if not rset_ok:
                    try: sock.close()
                    except: pass
                    return True, '250', 'OK', False, None
                return True, final_code, 'OK', False, sock
            else:
                # 421/554/451 等内容/速率被拒 → 换IP
                try: sock.close()
                except: pass
                with box.locker:
                    box.sent += 1
                    box.series_fail += 1
                    box.info = f'DATA结果被拒({final_code})'
                return False, final_code, f'DATA结果被拒: {final_resp[:80]!r}', True, None

        except socket.timeout:
            try:
                if sock is not None: sock.close()
            except: pass
            return False, 'timeout', 'SMTP命令超时', True, None
        except Exception as e:
            try:
                if sock is not None: sock.close()
            except: pass
            return False, 'unknown', str(e), False, None

    def send_api(self, box: MailBox, email: Email, subject: str,
                 content: str, api_url: str, api_key: str = '',
                 timeout: int = 30) -> tuple:
        """API中转发送"""
        try:
            import requests
            payload = {
                'from': box.email or box.user,
                'to': email.email_address,
                'subject': subject,
                'content': content,
                'name': email.name or '',
            }
            headers = {'Content-Type': 'application/json'}
            if api_key:
                headers['Authorization'] = f'Bearer {api_key}'
            resp = requests.post(api_url, json=payload, headers=headers, timeout=timeout)
            if resp.status_code in (200, 202):
                with box.locker:
                    box.sent += 1
                    box.sent_succ += 1
                    box.day_sent += 1
                    box.series_fail = 0
                return True, '250', 'OK'
            else:
                with box.locker:
                    box.sent += 1
                    box.series_fail += 1
                    box.info = f'API返回{resp.status_code}'
                return False, str(resp.status_code), resp.text[:200]
        except Exception as e:
            with box.locker:
                box.sent += 1
                box.series_fail += 1
                box.info = str(e)
            return False, 'api_error', str(e)

    def send_eml(self, box: MailBox, email: Email, eml_content: str,
                 helo_domain: str = None, timeout: int = 60) -> tuple:
        """从EML内容发送"""
        smtp = None
        try:
            from_addr = box.email or box.user
            if box.ssl_tls in ('ssl',):
                context = ssl.create_default_context()
                context.check_hostname = False
                context.verify_mode = ssl.CERT_NONE
                smtp = smtplib.SMTP_SSL(box.smtp, box.port, timeout=timeout, context=context)
            else:
                smtp = smtplib.SMTP(box.smtp, box.port, timeout=timeout)
            smtp.ehlo(helo_domain or (from_addr.split('@')[-1] if '@' in from_addr else 'localhost'))
            if box.ssl_tls in ('tls', 'auto'):
                smtp.starttls()
                smtp.ehlo(helo_domain or (from_addr.split('@')[-1] if '@' in from_addr else 'localhost'))
            if box.pwd:
                smtp.login(box.user if '@' in box.user else box.email, box.pwd)
            smtp.sendmail(from_addr, [email.email_address], eml_content)
            with box.locker:
                box.sent += 1
                box.sent_succ += 1
                box.day_sent += 1
                box.series_fail = 0
            return True, '250', 'OK'
        except Exception as e:
            with box.locker:
                box.sent += 1
                box.series_fail += 1
                box.info = str(e)
            return False, 'unknown', str(e)
        finally:
            if smtp is not None:
                try:
                    smtp.quit()
                except Exception:
                    try:
                        smtp.close()
                    except Exception:
                        pass

    def _query_mx(self, domain: str, dns_servers: List[str] = None) -> List[str]:
        try:
            resolver = dns.resolver.Resolver()
            if dns_servers:
                resolver.nameservers = dns_servers
            answers = resolver.resolve(domain, 'MX')
            mx_list = sorted([(r.preference, str(r.exchange).rstrip('.')) for r in answers])
            return [mx[1] for mx in mx_list]
        except:
            return []

    def start_sending(self, boxes: List[MailBox], emails: List[Email],
                      subjects: List[SubjectEx], contents: List[Content],
                      thread_count: int = 10, send_interval: float = 0,
                      send_mode: str = 'smtp', dns_servers: List[str] = None,
                      variables: List[Variable] = None,
                      attachments: List[Attachment] = None,
                      helo_domain: str = None, reply_to: str = None,
                      custom_headers: dict = None,
                      magic_config: dict = None,
                      smart_enabled: bool = True,
                      proxy_config: dict = None,
                      proxies: List[ProxyIP] = None,
                      adsl_config: dict = None,
                      track_config: dict = None,
                      word_replacements: List[tuple] = None,
                      sender_names: List[str] = None,
                      use_cache: bool = True,
                      use_blacklist: bool = True,
                      api_url: str = '', api_key: str = '',
                      eml_list: List[str] = None,
                      schedule_start: datetime = None,
                      schedule_interval: int = 0):
        """启动批量发送 — 完整版"""
        self._stop = False
        self.stats = SendStats()
        self.stats.start_time = time.time()
        self.stats.total = len(emails)

        active_subjects = [s for s in subjects if s.enable]
        active_contents = [c for c in contents if c.enable]
        active_boxes = [b for b in boxes if b.can_send()]

        if not active_boxes:
            self.log('error', '没有可用的发件邮箱')
            return
        if not active_subjects:
            self.log('error', '没有可用的主题模板')
            return
        if not active_contents:
            self.log('error', '没有可用的内容模板')
            return

        var_proc = VariableProcessor(variables or [])
        from core.text_engine import MagicEncoder
        magic = MagicEncoder(magic_config or {})
        rng = random.Random()

        if use_cache:
            self.mail_cache = MailCache()
        if use_blacklist:
            self.blacklist = EmailBlacklist()
        if word_replacements:
            self.word_replacer = WordReplacer(word_replacements)
        if proxy_config and proxy_config.get('enabled') and proxies:
            self.proxy_manager = ProxyManager(proxies, proxy_config)
        if adsl_config and adsl_config.get('enabled'):
            self.adsl_manager = ADSLManager(adsl_config)
        if track_config and track_config.get('enabled'):
            self.track_config = track_config
        if sender_names:
            self.sender_names = sender_names
        if track_config and track_config.get('bounce_email'):
            self.bounce_detector = BounceDetector(
                track_config['bounce_email'], self.bounce_callback)

        box_idx = 0
        email_idx = 0
        box_lock = threading.Lock()
        current_interval = send_interval
        series_fail_total = 0

        if schedule_start:
            now = datetime.now()
            if schedule_start > now:
                wait = (schedule_start - now).total_seconds()
                self.log('info', f'定时发送: 等待{wait:.0f}秒到{schedule_start}')
                while not self._stop and time.time() < schedule_start.timestamp():
                    time.sleep(1)
                if self._stop:
                    return

        def worker():
            nonlocal box_idx, email_idx, current_interval, series_fail_total

            while not self._stop:
                with box_lock:
                    if email_idx >= len(emails):
                        break
                    current_email = emails[email_idx]
                    current_idx = email_idx
                    email_idx += 1

                    if self.mail_cache and self.mail_cache.is_sent(current_email.email_address):
                        continue
                    if self.blacklist and self.blacklist.is_blacklisted(current_email.email_address):
                        continue

                    _all_paused_rounds = 0
                    while not self._stop:
                        if box_idx >= len(active_boxes):
                            box_idx = 0
                        box = active_boxes[box_idx]
                        box_idx = (box_idx + 1) % len(active_boxes)
                        box.reset_daily()
                        if self._check_pause(box):
                            _all_paused_rounds += 1
                            if _all_paused_rounds > len(active_boxes):
                                # 全部邮箱暂停/不可用: 睡眠等待, 避免忙等占满CPU
                                time.sleep(30)
                                _all_paused_rounds = 0
                            continue
                        if box.can_send():
                            break
                        _all_paused_rounds += 1
                        if _all_paused_rounds > len(active_boxes):
                            time.sleep(30)
                            _all_paused_rounds = 0

                    if not box.can_send():
                        self.log('error', '没有可用邮箱')
                        break

                try:
                    subject = rng.choice(active_subjects).subject
                    content = rng.choice(active_contents)
                    subject = var_proc.replace(subject, current_email)
                    content_text = var_proc.replace(content.content_text, current_email)
                    subject = SpintaxProcessor.process(subject, rng)
                    content_text = SpintaxProcessor.process(content_text, rng)
                    content_text = magic.process(content_text)

                    if self.word_replacer:
                        content_text = self.word_replacer.replace(content_text)
                        subject = self.word_replacer.replace(subject)

                    sender_name = box.sender_name
                    if box.random_user and self.sender_names:
                        sender_name = rng.choice(self.sender_names)
                    elif box.random_user:
                        sender_name = f'User{rng.randint(1000, 9999)}'

                    domain = helo_domain or (box.email.split('@')[-1] if '@' in box.email else 'localhost')

                    if self.track_config.get('enabled') and self.track_config.get('server_url'):
                        content_text = TrackingInserter.insert_pixel(
                            content_text, current_email.email_address,
                            self.track_config['server_url'])
                        if self.track_config.get('track_clicks'):
                            content_text = TrackingInserter.insert_link_tracking(
                                content_text, self.track_config['server_url'])

                    success = False
                    smtp_code = ''
                    error_msg = ''
                    proxy = None

                    if self.proxy_manager:
                        proxy = self.proxy_manager.get_current()

                    if send_mode == 'smtp':
                        success, smtp_code, error_msg = self.send_smtp(
                            box, current_email, subject, content_text,
                            content.mail_type, attachments, domain,
                            sender_name, reply_to, custom_headers,
                            timeout=60, proxy=proxy)
                    elif send_mode == 'direct':
                        # send_direct 返回 5 元组 (success, code, msg, ip_rejected, raw_sock)
                        success, smtp_code, error_msg, _ipr, _sock = self.send_direct(
                            box, current_email, subject, content_text,
                            content.mail_type, dns_servers, domain, sender_name)
                    elif send_mode == 'api' and api_url:
                        success, smtp_code, error_msg = self.send_api(
                            box, current_email, subject, content_text,
                            api_url, api_key)
                    elif send_mode == 'eml' and eml_list:
                        eml_content = rng.choice(eml_list)
                        success, smtp_code, error_msg = self.send_eml(
                            box, current_email, eml_content, domain)

                    if self.proxy_manager:
                        if success:
                            self.proxy_manager.on_sent()
                        else:
                            self.proxy_manager.on_fail(proxy)

                    if self.bounce_detector:
                        self.bounce_detector.on_result(
                            current_email.email_address, success, smtp_code, error_msg)

                    if self.mail_cache and success:
                        self.mail_cache.mark_sent(current_email.email_address)
                    if self.blacklist and not success and 'invalid' in error_msg.lower():
                        self.blacklist.add(current_email.email_address)

                    with self._lock:
                        if success:
                            self.stats.success += 1
                            self.update_email(current_idx, '成功', 1)
                            series_fail_total = 0
                        else:
                            self.stats.fail += 1
                            self.update_email(current_idx, f'失败: {error_msg}', 0)
                            series_fail_total += 1

                            if self.bounce_callback:
                                self.bounce_callback(current_email.email_address, smtp_code, error_msg)

                            if smart_enabled and self.smart_callback:
                                adjust = self.smart_callback(smtp_code, error_msg, box, self.stats)
                                if adjust:
                                    self.stats.adjust_count += 1
                                    action = adjust.get('action', '')
                                    self.log('warning', f'智能调节: {adjust.get("reason","")} → {action}')
                                    if 'interval' in adjust:
                                        current_interval = adjust['interval']
                                    if 'skip_box' in adjust and adjust['skip_box']:
                                        with box.locker:
                                            box.enable = False
                                            box.info = f'智能调节禁用: {adjust.get("reason","")}'
                                    if action == 'change_box':
                                        with box.locker:
                                            box.enable = False
                                            box.info = '切换邮箱'
                                    if action == 'change_ip':
                                        if self.proxy_manager:
                                            new_proxy = self.proxy_manager.rotate()
                                            if new_proxy:
                                                self.log('info', f'智能调节: 切换代理IP → {new_proxy.ip}:{new_proxy.port}')
                                        if self.adsl_manager and self.adsl_manager.auto_dial:
                                            if self.adsl_manager.dial():
                                                self.log('success', f'智能调节: ADSL换IP → {self.adsl_manager.get_current_ip()}')
                                    if action == 'reduce_speed':
                                        current_interval = max(current_interval + 2, 5)
                                        self.log('info', f'智能调节: 降速 → 间隔{current_interval}秒')
                                    if action == 'change_content':
                                        self.log('warning', '智能调节: 建议切换内容模板(进垃圾箱)')
                                    if action == 'disable_box':
                                        with box.locker:
                                            box.enable = False
                                            box.info = f'连续失败禁用: {adjust.get("reason","")}'

                        self.update_stats()

                    if self.adsl_manager and self.adsl_manager.auto_dial and series_fail_total >= self.adsl_manager.fail_threshold:
                        self.log('warning', f'连续失败{series_fail_total}次，ADSL拨号换IP...')
                        if self.adsl_manager.dial():
                            new_ip = self.adsl_manager.get_current_ip()
                            self.log('success', f'ADSL拨号成功，新IP: {new_ip}')
                            series_fail_total = 0
                        else:
                            self.log('error', 'ADSL拨号失败')

                    if current_interval > 0:
                        time.sleep(current_interval + rng.uniform(0, max(current_interval * 0.5, 0.1)))

                    if schedule_interval > 0 and email_idx % schedule_interval == 0:
                        self.log('info', f'定时暂停: 已发{email_idx}封，等待下一轮...')
                        time.sleep(60)

                except Exception as e:
                    with self._lock:
                        self.stats.fail += 1
                        self.update_email(current_idx, f'异常: {e}', 0)
                        self.update_stats()
                    self.log('error', f'发送异常: {e}')

        self.log('info', f'开始发送: {len(emails)}封, {thread_count}线程, 模式={send_mode}')

        for i in range(thread_count):
            t = threading.Thread(target=worker, daemon=True, name=f'send-{i}')
            self._threads.append(t)
            t.start()

        for t in self._threads:
            t.join()

        if self.mail_cache:
            self.mail_cache._save()

        self.log('info', f'发送完成: 成功{self.stats.success} 失败{self.stats.fail} 调节{self.stats.adjust_count}')