"""
mail_deploy.py — 邮局部署引擎 (集成进主控)
===========================================
把部署邮局面板的 SSH 批量部署能力移植进主控:
  - 10种邮件栈安装脚本 (install_scripts/, 与主控代码同仓)
  - 参数注入 (域名/CF token/SMTP密码/DKIM/魔改选项)
  - 部署后自动写 DNS (复用主控 cf_config.json 多账号 + core/dns_manager)
  - 策略更新 / 投递监控
状态轮询: 与 deploy_node 同模式 (_state dict)
"""
import os
import json
import time
import random
import string
import threading
import re

_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
_PROJECT_ROOT = os.path.dirname(_THIS_DIR)
SCRIPTS_DIR = os.path.join(_PROJECT_ROOT, 'install_scripts')
CONFIG_FILE = os.path.join(_THIS_DIR, 'mail_deploy_config.json')
CF_CONFIG_FILE = os.path.join(_THIS_DIR, 'cf_config.json')

STACK_MAP = {
    'postfix': 'install-postfix.sh', 'postfix_outbound': 'install-postfix_outbound.sh',
    'haraka': 'install-haraka.sh', 'maddy': 'install-maddy.sh',
    'opensmtpd': 'install-opensmtpd.sh', 'exim4': 'install-exim4.sh',
    'exim4_outbound': 'install-exim4_outbound.sh', 'courier_mta': 'install-courier_mta.sh',
    'poste_io': 'install-poste_io.sh', 'stalwart': 'install-stalwart.sh',
}

DEFAULT_CONFIG = {
    'mail_stack': 'haraka',
    'acme_ca': 'selfsigned',
    'tls_key_type': 'rsa',
    'dkim_sign_count': 3,
    'dkim_sign_mode': 'mkt_lu_news',
    'dkim_selector': 'default',
    'dmarc_policy': 'none',
    'spf_qualifier': '-all',
    'strip_headers': ('X-深情, X-KumoRef, X-Mailer-Trace, Authentication-Results, '
                      'Original-Authentication-Results, Received-SPF, ARC-Authentication-Results, '
                      'ARC-Message-Signature, ARC-Seal, DKIM-Signature, X-Mailer, X-Haraka, '
                      'X-Haraka-UUID, X-Haraka-Transaction, X-Node, X-NodeMailer, '
                      'X-Originating-Email, X-Originating-IP, X-Priority, Importance, '
                      'X-MSMail-Priority, X-PHP-Originating-Script, X-AntiAbuse, X-Source, '
                      'X-Source-Args, X-Authenticated-Sender, X-Authenticated-User, '
                      'X-Sender-IP, X-SMTPAPI'),
    'rewrite_received': 'true',
    'strip_list_unsubscribe': 'false',
    'outbound_client_mode': 'mobile',
    'received_mode': 'single',
    'helo_name': '',
    'workers': 5,
    # ★ 投递重试策略: "5m"=失败后5分钟重试1次即放弃; "none"=不重试; "1m,5m"=依次重试2次
    'retry_policy': '5m',
    # ★ SMTP 账号前缀: 发信账号 = 前缀@域名 (部署产物, 主控提交时用)
    'smtp_user': 'admin',
}

_state = {
    'running': False, 'action': '', 'logs': [], 'done': False,
    'ok': True, 'msg': '', 'results': [], 'started_at': '',
    # ★ 进度: total=总数, done_n=已完成台数, current=正在进行的台名
    'total': 0, 'done_n': 0, 'current': '',
}
_lock = threading.Lock()


def _progress_mark(current_name=None, done_incr=0):
    """更新进度状态 (线程安全)"""
    with _lock:
        if current_name is not None:
            _state['current'] = str(current_name)
        if done_incr:
            _state['done_n'] += done_incr


def _log(msg):
    line = '[' + time.strftime('%H:%M:%S') + '] ' + str(msg)
    _state['logs'].append(line)
    if len(_state['logs']) > 800:
        _state['logs'] = _state['logs'][-800:]
    print('[MAILDEPLOY] ' + line)


def get_config():
    cfg = dict(DEFAULT_CONFIG)
    if os.path.exists(CONFIG_FILE):
        try:
            with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
                loaded = json.load(f)
            for k, v in (loaded or {}).items():
                # 空字符串不覆盖非空默认值 (防止旧配置把魔改开关冲掉)
                if v is not None and not (isinstance(v, str) and v == '' and cfg.get(k)):
                    cfg[k] = v
        except Exception:
            pass
    return cfg


def save_config(cfg):
    try:
        with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
            json.dump(cfg, f, ensure_ascii=False, indent=2)
        return True
    except Exception:
        return False


def status():
    return dict(_state)


def _gen_smtp_pass():
    return 'Tk' + ''.join(random.choices(string.ascii_letters + string.digits, k=14))


def _ssh_connect(ip, port, user, pwd):
    import paramiko
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(ip, port=int(port or 22), username=user or 'root', password=pwd,
                timeout=30, banner_timeout=30)
    return ssh


def _ssh_run(ssh, cmd, timeout=300):
    stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout)
    out = stdout.read().decode('utf-8', 'replace')
    err = stderr.read().decode('utf-8', 'replace')
    code = stdout.channel.recv_exit_status()
    return code, out, err


def _read_script(stack):
    fname = STACK_MAP.get(stack, STACK_MAP['haraka'])
    path = os.path.join(SCRIPTS_DIR, fname)
    if not os.path.exists(path):
        return None
    with open(path, 'r', encoding='utf-8', errors='replace') as f:
        return f.read().replace('\r\n', '\n').replace('\r', '\n')


_SAFE_VAL_RE = re.compile(r'^[A-Za-z0-9._:@/+-]*$')


def _shell_quote(s):
    """shell 单引号安全包裹 (值内单引号转义为 '\'' — POSIX 标准做法)"""
    return "'" + str(s).replace("'", "'\\''") + "'"


def _valid_smtp_user(val):
    """SMTP账号前缀白名单: 字母数字及 . _ - + (RFC local-part 安全子集), 缺省 admin"""
    import re as _re
    val = str(val or 'admin').strip()
    if not _re.fullmatch(r'[A-Za-z0-9._+-]{1,32}', val):
        return 'admin'
    return val


def _inject_params(script_body, server, cfg):
    """替换安装脚本顶部变量 (与部署面板 _inject_params 同逻辑; 值经白名单+引号包裹, 防注入)"""
    hostname = server.get('hostname', '')
    smtp_user_prefix = _valid_smtp_user(cfg.get('smtp_user', 'admin'))
    def _rv(text, var, new_val):
        # 值白名单: 常规配置项只含安全字符; 含特殊字符的(如 STRIP_HEADERS)由调用方加引号
        return re.sub(rf'^({var})=.*$', lambda m: m.group(1) + '=' + str(new_val), text, flags=re.MULTILINE)
    # ★ 幂等: 数据库已有 SMTP 密码则沿用 (重复部署/策略更新不重置密码, 主控提交永不掉线)
    smtp_pass = server.get('smtp_pwd') or _gen_smtp_pass()
    # ★ 危险值安全化: hostname/selector 若含 shell 元字符 → 拒绝 (部署源头数据被污染时兜底)
    for _v in [hostname, cfg.get('dkim_selector', 'default'), cfg.get('helo_name') or hostname]:
        if _v and not _SAFE_VAL_RE.match(str(_v)):
            raise ValueError('参数含非法字符(拒绝部署): ' + str(_v)[:50])
    for var, val in [
        ('MAIL_DOMAIN', _shell_quote(hostname)), ('MAIL_HOSTNAME', _shell_quote(hostname)),
        ('MAIL_HELO', _shell_quote(cfg.get('helo_name') or hostname)),
        ('MSGID_DOMAIN', _shell_quote(hostname)),
        ('DKIM_SELECTOR', _shell_quote(cfg.get('dkim_selector', 'default'))),
        ('SMTP_AUTH_USER', _shell_quote(smtp_user_prefix + '@' + hostname)),
        ('SMTP_PASS', _shell_quote(smtp_pass)),
        ('ADMIN_EMAIL', _shell_quote('admin@' + hostname)),
        ('ACME_CA', _shell_quote(cfg.get('acme_ca', 'selfsigned'))),
        ('TLS_KEY_TYPE', _shell_quote(cfg.get('tls_key_type', 'rsa'))),
        ('RECEIVED_MODE', _shell_quote(cfg.get('received_mode', 'single'))),
        ('OUTBOUND_CLIENT_MODE', _shell_quote(cfg.get('outbound_client_mode', 'mobile'))),
        ('DKIM_SIGN_COUNT', _shell_quote(str(cfg.get('dkim_sign_count', 3)))),
        ('DKIM_SIGN_MODE', _shell_quote(cfg.get('dkim_sign_mode', 'mkt_lu_news'))),
        ('DMARC_POLICY', _shell_quote(cfg.get('dmarc_policy', 'none'))),
        ('SPF_QUALIFIER', _shell_quote(cfg.get('spf_qualifier', '-all'))),
        ('RETRY_POLICY', _shell_quote(_valid_retry_policy(cfg.get('retry_policy', '5m')))),
        ('STRIP_HEADERS', '"' + (cfg.get('strip_headers') or '') + '"'),
        ('REWRITE_RECEIVED', str(cfg.get('rewrite_received', 'true')).lower()),
        ('STRIP_LIST_UNSUBSCRIBE', str(cfg.get('strip_list_unsubscribe', 'false')).lower()),
    ]:
        script_body = _rv(script_body, var, val)
    # 占位符硬替换
    # ★ 顺序关键: 先替换 SMTPUSER@占位域名 (此时域名占位尚未被换掉),
    #   再替换域名占位 mail1.weicai168.com → 否则 SMTPUSER 会字面写入服务器 (535根因)
    script_body = script_body.replace('SMTPUSER@mail1.weicai168.com',
                                      smtp_user_prefix + '@mail1.weicai168.com')
    script_body = script_body.replace('mail1.weicai168.com', hostname)
    script_body = script_body.replace('qwffHPPQ4966', smtp_pass)
    # 当前主控CF token (部署面板版占位 token 换掉, 不影响其余)
    script_body = script_body.replace(
        'cfut_oea3LitKRIKQl0ursISIpPUD2E24BoNopC5l4oI009ced023', '')
    script_body = script_body.replace('selector=news', 'selector=' + cfg.get('dkim_selector', 'default'))
    # 重试策略硬替换 (outbound.ini 里的占位值)
    script_body = script_body.replace('temp_fail_intervals=5m',
                                      'temp_fail_intervals=' + _valid_retry_policy(cfg.get('retry_policy', '5m')))
    return script_body, smtp_pass


def _valid_retry_policy(val):
    """重试策略白名单: none / 5m / 1m,5m / 5m,15m 格式 (时间单位 s|m|h)"""
    val = str(val or '5m').strip()
    import re as _re
    if not _re.fullmatch(r'none|[0-9]+[smh](,[0-9]+[smh])*', val):
        return '5m'  # 非法输入回落默认
    return val


def _load_cf_accounts():
    if not os.path.exists(CF_CONFIG_FILE):
        return []
    try:
        with open(CF_CONFIG_FILE, 'r', encoding='utf-8') as f:
            cfg = json.load(f)
    except Exception:
        return []
    if isinstance(cfg, list):
        return [c for c in cfg if isinstance(c, dict) and c.get('api_key')]
    if isinstance(cfg, dict) and cfg.get('api_key'):
        return [cfg]
    return []


def _account_from_dict(a):
    from core.models import DNSAccount
    key = (a.get('api_key') or '').strip()
    is_token = key.startswith('cfat_') or key.startswith('cf_') or key.startswith('cfut_') or len(key) >= 40
    if not is_token and not a.get('api_email'):
        return None
    return DNSAccount(provider='cloudflare', api_email=a.get('api_email', ''), api_key=key)


def _clean_subdomain_records(provider, zone, hostname):
    """清空该邮局子域 (hostname) 名下的全部解析记录 (不动 zone 其他子域/主域)
    返回删除条数, 失败返回 -1"""
    try:
        records = provider.get_records(zone)
    except Exception:
        return -1
    hostname_l = hostname.lower().rstrip('.')
    removed = 0
    for r in records:
        rname = str(r.get('name', r.get('Name', r.get('RR', ''))) or '').lower().rstrip('.')
        if rname != hostname_l:
            continue  # 只动 hostname 本身的记录, 子域下的 _domainkey/_dmarc 由后缀匹配
        rid = r.get('id', r.get('RecordId', r.get('record_id', '')))
        if not rid:
            continue
        try:
            provider.delete_record(zone, str(rid))
            removed += 1
        except Exception:
            pass
    # 后缀匹配: xxx._domainkey.hostname / _dmarc.hostname (邮局专属子记录)
    for r in records:
        rname = str(r.get('name', r.get('Name', r.get('RR', ''))) or '').lower().rstrip('.')
        if rname == hostname_l:
            continue  # 上面已处理
        if not (rname.endswith('.' + hostname_l) and rname.split('.')[0] in (
                'default', '_domainkey', '_dmarc')):
            continue
        rid = r.get('id', r.get('RecordId', r.get('record_id', '')))
        if not rid:
            continue
        try:
            provider.delete_record(zone, str(rid))
            removed += 1
        except Exception:
            pass
    return removed


def write_dns(server):
    """给邮局域名写 DNS: A/SPF/DKIM/DMARC (用主控CF多账号自动匹配zone)
    ★ 每次部署先清空该子域的旧解析记录再重写 (zone 内其他子域/主域不受影响)"""
    from core.dns_manager import CloudFlareDNS
    hostname = server.get('hostname', '')
    if '.' not in hostname:
        return '域名无效'
    accounts = _load_cf_accounts()
    for a in accounts:
        acc = _account_from_dict(a)
        if not acc:
            continue
        try:
            provider = CloudFlareDNS(acc)
            zones = [d.domain.lower() for d in provider.get_domains()]
        except Exception:
            continue
        zone = None
        for z in zones:
            if hostname == z or hostname.endswith('.' + z):
                zone = z
                break
        if not zone:
            continue
        sub = hostname[:-len(zone)].rstrip('.')
        # ★ 清空该子域旧记录 (A/SPF/DKIM/DMARC 等 hostname 名下全部 + _domainkey/_dmarc 子记录)
        removed = _clean_subdomain_records(provider, zone, hostname)
        recs = [
            ('A', sub or '@', server.get('ip', '')),
            ('TXT', sub or '@', 'v=spf1 a mx ip4:' + server.get('ip', '') + ' '
             + (server.get('spf_qualifier') or '-all')),
            ('TXT', server.get('dkim_selector', 'default') + '._domainkey' + ('.' + sub if sub else ''),
             'v=DKIM1; k=rsa; p=' + (server.get('dkim_public') or '')),
            ('TXT', '_dmarc' + ('.' + sub if sub else ''),
             'v=DMARC1; p=' + (server.get('dmarc_policy') or 'none') + '; sp=' + (server.get('dmarc_policy') or 'none') + ';'),
        ]
        ok_n = 0
        for rt, name, content in recs:
            if not content or content.endswith('p='):
                continue
            try:
                if provider.upsert_record(zone, rt, name, content, log_cb=lambda l, m: None) == 'failed':
                    continue
                ok_n += 1
            except Exception:
                pass
        return '成功 (清旧' + str(max(removed, 0)) + '条, 写入' + str(ok_n) + '条, zone=' + zone + ')'
    return '未匹配到Cloudflare账号的zone'


def start(server_ids, action='deploy'):
    """启动部署/策略更新/证书签发 (后台线程)"""
    global _state
    if _state['running']:
        return False, '已有部署任务进行中, 请等它完成'
    _state = {'running': True, 'action': action, 'logs': [], 'done': False,
              'ok': True, 'msg': '', 'results': [], 'started_at': time.strftime('%Y-%m-%d %H:%M:%S'),
              'total': len(server_ids or []), 'done_n': 0, 'current': ''}
    t = threading.Thread(target=_worker, daemon=True, args=(server_ids, action))
    t.start()
    label = {'deploy': '并发部署', 'policy': '策略更新', 'cert': '证书签发/续期'}.get(action, action)
    return True, '已启动: ' + label


def _deploy_one(server, cfg, script_body, i, total):
    """部署单台邮局 (供并发调用)"""
    import master_db
    _progress_mark(current_name=server['name'])
    _log('==== [' + str(i + 1) + '/' + str(total) + '] ' + server['name'] + ' / ' + server['ip'] + ' (' + server['hostname'] + ') ====')
    # ★ smtp_user_full 提前定义 (后续日志与写库共用; 之前定义在尾部导致日志行 NameError → 部署全败)
    smtp_user_full = _valid_smtp_user(cfg.get('smtp_user', 'admin')) + '@' + server['hostname']
    try:
        injected, smtp_pass = _inject_params(script_body, server, cfg)
        ssh = _ssh_connect(server['ip'], server['ssh_port'], server['ssh_user'], server['ssh_pwd'])
        remote_path = '/tmp/maildeploy-' + server['mail_stack'] + '-' + str(int(time.time())) + '-' + str(random.randint(1000, 9999)) + '.sh'
        sftp = ssh.open_sftp()
        with sftp.open(remote_path, 'w') as f:
            f.write(injected)
        sftp.close()
        _log('[' + server['name'] + '] 安装脚本已上传, 执行中 (首次安装 3-15 分钟)...')
        stdin, stdout, stderr = ssh.exec_command('bash ' + _shell_quote(remote_path) + ' 2>&1', timeout=1800)
        result_json = ''
        out_all = []
        while True:
            line = stdout.readline()
            if not line:
                break
            l = line.rstrip()
            out_all.append(l)
            if '__KUMOMTA_RESULT__' in l:
                jlines = []
                while True:
                    jl = stdout.readline()
                    if not jl:
                        break
                    jlines.append(jl)
                    if '}' in jl:
                        break
                result_json = ''.join(jlines)
        exit_code = stdout.channel.recv_exit_status()
        dkim_public = ''
        if result_json:
            m = re.search(r'"dkim_public":\s*"([^"]+)"', result_json)
            if m:
                dkim_public = m.group(1)
        if exit_code != 0:
            _log('[' + server['name'] + '] ❌ 安装失败 (exit=' + str(exit_code) + ')')
            for l in out_all[-8:]:
                _log('  ' + l[:130])
            try:
                ssh.close()
            except Exception:
                pass
            master_db.mail_server_update(server['id'], status='失败')
            return False
        _log('[' + server['name'] + '] ✅ 安装成功 (SMTP账号 ' + smtp_user_full + ')')
        # ★ 追加非标准提交端口 10587 (部分机房封出站25/587, 主控从10587提交) — 必须在 ssh.close() 之前
        try:
            stdin2, stdout2, stderr2 = ssh.exec_command(
                "grep -q '10587' /etc/haraka/config/smtp.ini || "
                "sed -i 's/listen=.*/listen=0.0.0.0:25,0.0.0.0:587,0.0.0.0:10587/' /etc/haraka/config/smtp.ini; "
                "systemctl restart haraka 2>/dev/null || true", timeout=30)
            stdout2.channel.recv_exit_status()
            _log('[' + server['name'] + '] 10587提交端口已追加')
        except Exception:
            pass
        ssh.close()
        # ★ 部署成功同步提交端口: Haraka 实际监听 10587 (导入时的587是错的, 曾致全部发送超时)
        master_db.mail_server_update(server['id'], status='成功', smtp_user=smtp_user_full,
                                     smtp_pwd=smtp_pass, dkim_public=dkim_public,
                                     dkim_selector=cfg.get('dkim_selector', 'default'),
                                     smtp_port=10587)
        server['dkim_public'] = dkim_public
        server['spf_qualifier'] = cfg.get('spf_qualifier', '-all')
        server['dmarc_policy'] = cfg.get('dmarc_policy', 'none')
        _log('[' + server['name'] + '] 写 DNS ...')
        dns_msg = write_dns(server)
        _log('[' + server['name'] + '] DNS: ' + dns_msg)
        master_db.mail_server_update(server['id'], dns_status=dns_msg)
        with _lock:
            _state['results'].append({'name': server['name'], 'ip': server['ip'],
                                      'smtp_user': smtp_user_full,
                                      'smtp_pwd': smtp_pass, 'dns': dns_msg})
        return True
    except Exception as e:
        _log('[' + server['name'] + '] ❌ ' + str(e)[:180])
        master_db.mail_server_update(server['id'], status='失败')
        return False


def _cert_one(server, cfg, i, total):
    """给单台邮局签发/续期 Let's Encrypt 证书 (DNS-01 验证, token 取自主控CF账号)
    自动: 装certbot → 写CF凭据 → 签发 hostname 证书 → 装入haraka → 重启 → 配自动续期hook"""
    import master_db
    name = server.get('name') or server.get('hostname')
    hostname = server.get('hostname', '')
    _log('==== [证书 %d/%d] %s / %s (%s) ====' % (i + 1, total, name, server.get('ip'), hostname))
    if '.' not in hostname or not _SAFE_VAL_RE.match(hostname):
        _log('[' + name + '] ❌ 发件域名无效(含非法字符), 无法签发')
        return False
    token = ''
    for a in _load_cf_accounts():
        k = (a.get('api_key') or '').strip()
        if k:
            token = k
            break
    if not token:
        _log('[' + name + '] ❌ 主控未配置Cloudflare账号 (请先在「域名管理」Tab保存API Token)')
        return False
    try:
        ssh = _ssh_connect(server['ip'], server['ssh_port'], server['ssh_user'], server['ssh_pwd'])
    except Exception as e:
        _log('[' + name + '] ❌ SSH连接失败: ' + str(e)[:80])
        return False
    try:
        _log('[' + name + '] 检查/安装 certbot + DNS插件 ...')
        code, out, err = _ssh_run(ssh,
            "export DEBIAN_FRONTEND=noninteractive; "
            "which certbot >/dev/null 2>&1 || { apt-get update -qq && apt-get install -y -qq certbot python3-certbot-dns-cloudflare; }; "
            "python3 -c 'import certbot_dns_cloudflare' >/dev/null 2>&1 || apt-get install -y -qq python3-certbot-dns-cloudflare",
            timeout=900)
        if code != 0:
            _log('[' + name + '] ❌ certbot 安装失败: ' + (err or out)[-160:])
            return False
        code, out, err = _ssh_run(ssh, 'mkdir -p /root/.secrets', timeout=30)
        sftp = ssh.open_sftp()
        try:
            with sftp.open('/root/.secrets/cloudflare.ini', 'w') as f:
                f.write('dns_cloudflare_api_token = ' + token + '\n')
        finally:
            sftp.close()
        code, out, err = _ssh_run(ssh, 'chmod 600 /root/.secrets/cloudflare.ini', timeout=30)
        issue_cmd = ("certbot certonly --dns-cloudflare --dns-cloudflare-credentials /root/.secrets/cloudflare.ini "
                     "-d '%s' --non-interactive --agree-tos -m admin@%s --cert-name mailcert --quiet" % (hostname, hostname))
        code, out, err = _ssh_run(ssh, issue_cmd, timeout=600)
        if code != 0:
            _log('[' + name + '] 首次签发失败: ' + (err or out)[-240:])
            code2, out2, err2 = _ssh_run(ssh, "grep -E 'Failed|problem|error|timeout|invalid|unauthorized|NXDOMAIN|SERVFAIL' /var/log/letsencrypt/letsencrypt.log 2>/dev/null | tail -5", timeout=30)
            if code2 == 0 and out2.strip():
                for l in out2.strip().splitlines()[-5:]:
                    _log('  [lelog] ' + l[-160:])
            _log('[' + name + '] 等待20秒后重试一次 ...')
            time.sleep(20)
            code, out, err = _ssh_run(ssh, issue_cmd, timeout=600)
        if code != 0:
            _log('[' + name + '] 重试仍失败, 走续期通道强制刷新 ...')
            code, out, err = _ssh_run(ssh,
                "certbot renew --cert-name mailcert --dns-cloudflare "
                "--dns-cloudflare-credentials /root/.secrets/cloudflare.ini --force-renewal --quiet",
                timeout=600)
        if code != 0:
            _log('[' + name + '] ❌ 签发失败: ' + (err or out)[-200:])
            return False
        _log('[' + name + '] ✅ 证书签发成功 (' + hostname + ')')
        hook = ('#!/bin/bash\n'
                '[ -f /etc/letsencrypt/live/mailcert/fullchain.pem ] || exit 0\n'
                'cp /etc/letsencrypt/live/mailcert/fullchain.pem /etc/haraka/config/tls/cert.pem\n'
                'cp /etc/letsencrypt/live/mailcert/privkey.pem /etc/haraka/config/tls/key.pem\n'
                'systemctl restart haraka\n')
        code, out, err = _ssh_run(ssh,
            "mkdir -p /etc/letsencrypt/renewal-hooks/deploy && cat > /etc/letsencrypt/renewal-hooks/deploy/99-haraka-cert.sh <<'HOOKEOF'\n"
            + hook + "HOOKEOF\nchmod +x /etc/letsencrypt/renewal-hooks/deploy/99-haraka-cert.sh", timeout=30)
        if code != 0:
            _log('[' + name + '] ⚠ 续期hook写入失败(不影响本次): ' + (err or out)[-120:])
        code, out, err = _ssh_run(ssh,
            "cp /etc/letsencrypt/live/mailcert/fullchain.pem /etc/haraka/config/tls/cert.pem && "
            "cp /etc/letsencrypt/live/mailcert/privkey.pem /etc/haraka/config/tls/key.pem && "
            "systemctl restart haraka && sleep 2 && systemctl is-active haraka", timeout=60)
        if code != 0 or 'active' not in out:
            _log('[' + name + '] ❌ haraka 换证书重启失败: ' + (err or out)[-160:])
            return False
        _log('[' + name + '] ✅ 证书已装入 haraka 并重启 (active)')
        code, out, err = _ssh_run(ssh,
            "certbot delete --cert-name wildcard99dhm --non-interactive >/dev/null 2>&1; "
            "rm -f /etc/letsencrypt/renewal-hooks/deploy/deploy_cert.sh /root/deploy_cert.sh; true", timeout=120)
        _log('[' + name + '] ✅ 完成 (自动续期已配置: 每天检查, 到期自动续+自动装+自动重启)')
        return True
    except Exception as e:
        _log('[' + name + '] ❌ ' + str(e)[:160])
        return False
    finally:
        try:
            ssh.close()
        except Exception:
            pass


def _worker(server_ids, action):
    import master_db
    from concurrent.futures import ThreadPoolExecutor
    cfg = get_config()
    try:
        # ★ server_ids 去重 (保序): 前端勾选异常/重复请求时防止同一台部署两遍
        #   (第二遍会在已装环境上重跑 npm → EEXIST/ENOTEMPTY 假失败)
        seen = set()
        server_ids = [sid for sid in server_ids if not (sid in seen or seen.add(sid))]
        servers = [master_db.mail_server_get(sid) for sid in server_ids]
        servers = [s for s in servers if s]
        if not servers:
            _log('没有选中邮局服务器')
            return
        if action == 'cert':
            # ★ 方案A: 按 zone 分组 — 组间并行, 组内串行 (同zone并发会踩TXT记录)
            #   25台散在多个zone时接近全并行; 同zone的(如 mail1-5 同域)才排队
            zone_groups = {}
            for s in servers:
                hostname = s.get('hostname', '')
                # zone = 邮局记录的 hostname 所属 CF zone (用主控账号匹配; 匹配不到归入独立组=可并行)
                zkey = None
                for a in _load_cf_accounts():
                    acc = _account_from_dict(a)
                    if not acc:
                        continue
                    try:
                        from core.dns_manager import CloudFlareDNS
                        provider = CloudFlareDNS(acc)
                        zones = [d.domain.lower() for d in provider.get_domains()]
                    except Exception:
                        continue
                    for z in zones:
                        if hostname == z or hostname.endswith('.' + z):
                            zkey = z
                            break
                    if zkey:
                        break
                zkey = zkey or ('独立:' + hostname)
                zone_groups.setdefault(zkey, []).append(s)
            # zone 列表缓存后清空 (避免每台重复拉)
            total = len(servers)
            n_groups = len(zone_groups)
            _log("开始签发/续期证书 " + str(total) + " 台 (Let's Encrypt DNS-01, 按 " + str(n_groups) + " 个zone分组: 组间并行/组内串行)")
            for zk, lst in zone_groups.items():
                _log('  zone[' + zk + ']: ' + '、'.join(s.get('name', '') for s in lst))
            ok_n = [0]
            fail_n = [0]
            _cert_idx = [0]  # 全局序号 (进度显示用)

            def _cert_group(group_servers):
                for s in group_servers:
                    with _lock:
                        _cert_idx[0] += 1
                        gi = _cert_idx[0]
                    _progress_mark(current_name=s.get('name', ''))
                    ok = _cert_one(s, cfg, gi, total)
                    _progress_mark(done_incr=1)
                    if ok:
                        ok_n[0] += 1
                    else:
                        fail_n[0] += 1

            with ThreadPoolExecutor(max_workers=max(1, n_groups)) as pool:
                futures = [pool.submit(_cert_group, lst) for lst in zone_groups.values()]
                for f in futures:
                    try:
                        f.result()
                    except Exception:
                        pass
            _state['msg'] = '证书完成: 成功 ' + str(ok_n[0]) + ' 台, 失败 ' + str(fail_n[0]) + ' 台'
            _log('🎉 ' + _state['msg'])
            _state['ok'] = fail_n[0] == 0
            return
        _log('开始' + ('部署' if action == 'deploy' else '策略更新') + ' ' + str(len(servers)) + ' 台 (栈: ' + cfg.get('mail_stack', 'haraka') + ', 并发: ' + str(int(cfg.get('workers', 5))) + ')')
        script_body = _read_script(cfg.get('mail_stack', 'haraka'))
        if not script_body:
            raise Exception('安装脚本不存在: ' + STACK_MAP.get(cfg.get('mail_stack'), '') + ' (主控缺 install_scripts 目录)')
        workers = max(1, min(int(cfg.get('workers', 5)), 20))
        total = len(servers)
        ok_n = [0]
        fail_n = [0]
        # 随机启动延迟 (参考原面板: 错开各台启动时间)
        import random as _rnd
        time.sleep(_rnd.uniform(0, 2))
        with ThreadPoolExecutor(max_workers=workers) as pool:
            futures = []
            for i, server in enumerate(servers):
                _log('[' + server['name'] + '] 排队部署 (并发槽 ' + str(len(futures) % workers + 1) + '/' + str(workers) + ')')
                futures.append(pool.submit(_deploy_one, server, cfg, script_body, i, total))
            for f in futures:
                try:
                    if f.result():
                        ok_n[0] += 1
                    else:
                        fail_n[0] += 1
                except Exception:
                    fail_n[0] += 1
                _progress_mark(done_incr=1)
        _state['msg'] = '完成: 成功 ' + str(ok_n[0]) + ' 台, 失败 ' + str(fail_n[0]) + ' 台'
        _log('🎉 ' + _state['msg'])
        _state['ok'] = fail_n[0] == 0
    except Exception as e:
        _state['ok'] = False
        _state['msg'] = '任务异常: ' + str(e)[:200]
        _log('❌ ' + _state['msg'])
    finally:
        _state['running'] = False
        _state['done'] = True


def monitor(server_ids):
    """投递监控: 拉各邮局发送日志统计 (按收件人去重取最终状态, 反映真实送达)"""
    import master_db
    import re as _re
    lines = []
    total = delivered = bounced = deferred = 0
    attempts = {'delivered': 0, 'bounce': 0, 'deferred': 0}
    for sid in server_ids:
        server = master_db.mail_server_get(sid)
        if not server:
            continue
        try:
            ssh = _ssh_connect(server['ip'], server['ssh_port'], server['ssh_user'], server['ssh_pwd'])
            stdin, stdout, stderr = ssh.exec_command(
                'cat /var/log/haraka/delivery.log 2>/dev/null | tail -3000; '
                'journalctl -u haraka --no-pager -n 3000 2>/dev/null', timeout=60)
            text = stdout.read().decode('utf-8', 'replace')
            ssh.close()
            # ★ 按收件人去重: 每个邮箱只取最新一条记录的状态 (重试会写多条, 最终状态才算数)
            final = {}
            s_attempts = {'delivered': 0, 'bounce': 0, 'deferred': 0}
            for line in text.splitlines():
                if 'haraka_delivery' not in line:
                    continue
                m = _re.search(r' to=(\S+) ', line)
                st = _re.search(r'haraka_delivery (\w+)', line)
                if not m or not st:
                    continue
                email = m.group(1).strip('<>')
                status = st.group(1)
                if status not in s_attempts:
                    continue
                s_attempts[status] += 1
                ts = line[:19] if len(line) >= 19 else ''
                if email not in final or ts >= final[email][1]:
                    final[email] = (status, ts)
            d = sum(1 for v in final.values() if v[0] == 'delivered')
            b = sum(1 for v in final.values() if v[0] == 'bounce')
            df = sum(1 for v in final.values() if v[0] == 'deferred')
            total += d + b + df
            delivered += d
            bounced += b
            deferred += df
            attempts['delivered'] += s_attempts['delivered']
            attempts['bounce'] += s_attempts['bounce']
            attempts['deferred'] += s_attempts['deferred']
            lines.append({'name': server['name'], 'delivered': d, 'bounce': b, 'deferred': df,
                          'attempts': sum(s_attempts.values()),
                          'rate': round(d / (d + b + df) * 100, 1) if (d + b + df) else 0})
        except Exception as e:
            lines.append({'name': server['name'], 'error': str(e)[:80]})
    return {'total': total, 'delivered': delivered, 'bounced': bounced, 'deferred': deferred,
            'attempts_total': sum(attempts.values()),
            'rate': round(delivered / total * 100, 1) if total else 0, 'detail': lines}