"""
主控端 — 集群控制核心 (节点管理 / 任务分配 / 状态聚合 / 结果汇总)
====================================================================
流程:
  1. 添加节点 (IP:8898 + token)
  2. 建任务: 导入收件人 + 内容 + 选节点 + 分配代理池
  3. 分配: 收件人切分 → 推送各节点 → 远程启动
  4. 之后节点完全自主执行, 主控关机不影响
  5. 主控随时回来: 拉状态 / 增量拉结果 / 停任务

数据 (master_data/):
  nodes.json   节点列表
  tasks.json   任务记录 + 每节点结果拉取游标
  results/     按任务+节点存的拉回结果CSV
"""
import os
import json
import time
import threading
import random

_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
MASTER_DATA_DIR = os.path.join(_THIS_DIR, 'master_data')
RESULTS_DIR = os.path.join(MASTER_DATA_DIR, 'results')
NODES_FILE = os.path.join(MASTER_DATA_DIR, 'nodes.json')
TASKS_FILE = os.path.join(MASTER_DATA_DIR, 'tasks.json')

_lock = threading.RLock()

import master_db

try:
    import requests
    _REQ_OK = True
except ImportError:
    _REQ_OK = False


def _ensure_dirs():
    for d in (MASTER_DATA_DIR, RESULTS_DIR):
        if not os.path.isdir(d):
            try:
                os.makedirs(d, exist_ok=True)
            except Exception:
                pass


def _read_json(path, default):
    try:
        if os.path.exists(path):
            with open(path, 'r', encoding='utf-8-sig') as f:
                return json.load(f)
    except Exception:
        pass
    return default


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


# ============================================================
# 节点管理 (SQLite)
# ============================================================
def load_nodes():
    return master_db.node_all()


def save_nodes(nodes):
    return True  # 兼容旧调用, 数据已在 SQLite


def node_add(name, ip, port, token):
    _ensure_dirs()
    with _lock:
        nodes = load_nodes()
        node = {
            'id': 'n' + str(int(time.time() * 1000)),
            'name': name or ('节点-' + str(len(nodes) + 1)),
            'ip': ip,
            'port': int(port or 8898),
            'token': token,
            'created_at': time.strftime('%Y-%m-%d %H:%M:%S'),
        }
        master_db.node_add(node)
        return node


def node_remove(node_id):
    with _lock:
        return master_db.node_remove(node_id) is not False


def _find_node(node_id):
    for n in load_nodes():
        if n.get('id') == node_id:
            return n
    return None


def _node_call(node, method, path, payload=None, timeout=20):
    """调用节点 HTTP 接口, 返回 (ok, json_dict)"""
    if not _REQ_OK:
        return False, {'msg': 'requests 未安装'}
    url = 'http://' + str(node['ip']) + ':' + str(node['port']) + path
    headers = {'X-Token': str(node.get('token', ''))}
    try:
        if method == 'GET':
            resp = requests.get(url, headers=headers, timeout=timeout)
        else:
            resp = requests.post(url, headers=headers, json=payload, timeout=timeout)
        try:
            data = resp.json()
        except Exception:
            data = {'msg': 'bad response: ' + resp.text[:100]}
        if resp.status_code >= 400:
            return False, data
        # 节点业务层 ok:false (如token错/参数错) 也算失败
        if isinstance(data, dict) and data.get('ok') is False:
            return False, data
        return True, data
    except Exception as e:
        return False, {'msg': str(e)[:120]}


def node_ping(node_id):
    node = _find_node(node_id)
    if not node:
        return {'ok': False, 'msg': '节点不存在'}
    ok, data = _node_call(node, 'GET', '/ping', timeout=8)
    return {'ok': ok, 'node': node['name'], 'data': data}


# ============================================================
# 任务管理 (SQLite)
# ============================================================
def load_tasks():
    return master_db.task_all()


def save_tasks(tasks):
    return True  # 兼容旧调用, 数据已在 SQLite


def _split_recipients(recipients, n_parts):
    """把收件人列表切成 n 等份"""
    n_parts = max(1, int(n_parts))
    total = len(recipients)
    base = total // n_parts
    rem = total % n_parts
    chunks = []
    idx = 0
    for i in range(n_parts):
        size = base + (1 if i < rem else 0)
        chunks.append(recipients[idx:idx + size])
        idx += size
    return chunks


def _load_recipients_lines(source):
    """从主控本地文件读收件人 (去重可选, 返回列表)"""
    lines = []
    try:
        with open(source, 'r', encoding='utf-8') as f:
            for line in f:
                e = line.strip()
                if e and '@' in e:
                    lines.append(e.lower())
    except Exception:
        pass
    return lines


def _load_master_dkim_keys(boxes):
    """主控本地 dkim_keys 目录按发件箱域名加载 DKIM 私钥 (随任务下发节点)"""
    keys = {}
    dkim_dir = os.path.join(_THIS_DIR, 'dkim_keys')
    if not os.path.isdir(dkim_dir):
        return keys
    for b in boxes:
        email = b.get('email', '') if isinstance(b, dict) else ''
        if '@' not in email:
            continue
        domain = email.split('@')[-1].lower()
        sel = (b.get('dkim_selector') if isinstance(b, dict) else '') or 'default'
        pem = os.path.join(dkim_dir, domain + '_' + sel + '.pem')
        if os.path.exists(pem) and domain not in keys:
            try:
                with open(pem, 'r', encoding='utf-8') as f:
                    keys[domain] = {'selector': sel, 'private_key': f.read()}
            except Exception:
                pass
    return keys


def _stream_dedupe_split(source, n_parts):
    """流式按 hash 切分到 n_parts 个临时文件 (亿级名单不爆内存)
    返回 (out_files, counts) 或 (None, None)
    """
    import sqlite3
    _ensure_dirs()
    stamp = str(int(time.time() * 1000))
    db_path = os.path.join(MASTER_DATA_DIR, '_dedupe_' + stamp + '.db')
    out_files = []
    fhs = []
    counts = [0] * n_parts
    for i in range(n_parts):
        p = os.path.join(MASTER_DATA_DIR, '_chunk_' + stamp + '_' + str(i) + '.txt')
        fhs.append(open(p, 'w', encoding='utf-8'))
        out_files.append(p)
    try:
        conn = sqlite3.connect(db_path)
        conn.execute('PRAGMA journal_mode=OFF')
        conn.execute('PRAGMA synchronous=OFF')
        conn.execute('PRAGMA temp_store=MEMORY')
        conn.execute('CREATE TABLE IF NOT EXISTS e (email TEXT, bucket INTEGER)')
        n = 0
        with open(source, 'r', encoding='utf-8') as f:
            for line in f:
                e = line.strip().lower()
                if not e or '@' not in e:
                    continue
                conn.execute('INSERT INTO e(email,bucket) VALUES(?,?)',
                             (e, hash(e) % n_parts))
                n += 1
                if n % 200000 == 0:
                    conn.commit()
        conn.commit()
        for b in range(n_parts):
            for (email,) in conn.execute('SELECT email FROM e WHERE bucket=?', (b,)):
                fhs[b].write(email + '\n')
                counts[b] += 1
        conn.close()
        for fh in fhs:
            fh.close()
        try:
            os.remove(db_path)
        except Exception:
            pass
        return out_files, counts
    except Exception:
        for fh in fhs:
            try:
                fh.close()
            except Exception:
                pass
        try:
            os.remove(db_path)
        except Exception:
            pass
        return None, None


def _push_recipients_file(node, task_id, path, chunk_lines=50000):
    """分块推送收件人文件到节点 (每块 chunk_lines 行, 大文件不爆内存)"""
    try:
        with open(path, 'r', encoding='utf-8') as f:
            buf = []
            pushed = 0
            for line in f:
                e = line.strip()
                if not e or '@' not in e:
                    continue
                buf.append(e)
                if len(buf) >= chunk_lines:
                    ok, data = _node_call(node, 'POST', '/recipients',
                                          {'task_id': task_id, 'recipients_text': '\n'.join(buf)},
                                          timeout=300)
                    if not ok:
                        return False, pushed
                    pushed += len(buf)
                    buf = []
            if buf:
                ok, data = _node_call(node, 'POST', '/recipients',
                                      {'task_id': task_id, 'recipients_text': '\n'.join(buf)},
                                      timeout=300)
                if not ok:
                    return False, pushed
                pushed += len(buf)
        return True, pushed
    except Exception as e:
        return False, 0


def _split_by_quotas(recipients, quotas):
    """按每节点指定数量顺序切片 (配额为正整数的节点才分; 余量追加给最后一个节点)
    quotas: [int, ...] 与节点顺序对应
    """
    chunks = []
    rest = list(recipients)
    for q in quotas:
        q = max(0, int(q or 0))
        if q <= 0:
            chunks.append([])
            continue
        chunks.append(rest[:q])
        rest = rest[q:]
    if rest:
        # 余量追加给最后一个分到数据的节点
        for i in range(len(chunks) - 1, -1, -1):
            if chunks[i]:
                chunks[i].extend(rest)
                break
    return chunks


def _stream_dedupe_split_by_quota(source, quotas):
    """流式按配额顺序切分到临时文件 (亿级名单不爆内存)
    返回 (out_files, counts) 或 (None, None)
    """
    import sqlite3
    _ensure_dirs()
    stamp = str(int(time.time() * 1000))
    db_path = os.path.join(MASTER_DATA_DIR, '_dedupe_' + stamp + '.db')
    out_files = []
    fhs = []
    counts = [0] * len(quotas)
    for i in range(len(quotas)):
        p = os.path.join(MASTER_DATA_DIR, '_chunk_' + stamp + '_' + str(i) + '.txt')
        fhs.append(open(p, 'w', encoding='utf-8'))
        out_files.append(p)
    try:
        conn = sqlite3.connect(db_path)
        conn.execute('PRAGMA journal_mode=OFF')
        conn.execute('PRAGMA synchronous=OFF')
        conn.execute('PRAGMA temp_store=MEMORY')
        conn.execute('CREATE TABLE IF NOT EXISTS e (email TEXT)')
        n = 0
        with open(source, 'r', encoding='utf-8') as f:
            for line in f:
                e = line.strip().lower()
                if not e or '@' not in e:
                    continue
                conn.execute('INSERT INTO e(email) VALUES(?)', (e,))
                n += 1
                if n % 200000 == 0:
                    conn.commit()
        conn.commit()
        cur = conn.execute('SELECT email FROM e ORDER BY rowid')
        for i, quota in enumerate(quotas):
            quota = max(0, int(quota or 0))
            for _ in range(quota):
                row = cur.fetchone()
                if row is None:
                    break
                fhs[i].write(row[0] + '\n')
                counts[i] += 1
        # 余量全部给最后一个节点
        last = len(fhs) - 1
        for row in cur:
            fhs[last].write(row[0] + '\n')
            counts[last] += 1
        conn.close()
        for fh in fhs:
            fh.close()
        try:
            os.remove(db_path)
        except Exception:
            pass
        return out_files, counts
    except Exception:
        for fh in fhs:
            try:
                fh.close()
            except Exception:
                pass
        try:
            os.remove(db_path)
        except Exception:
            pass
        return None, None


def _merge_shuffle_split_stream(sources, n_nodes, quotas=None):
    """多文件合并 + 打乱 + 按域均衡平均分配 (流式, 亿级安全)
    sources: [path, ...]
    quotas: [int] 每节点配额 (0/None=均分)
    核心: 按 email_domain 排序后 round-robin 写入各节点文件
          → 每个运营商域的邮箱被均匀摊到所有节点, 不会一个节点全是同一运营商
    返回 (out_files, counts) 或 (None, None)
    """
    import sqlite3
    _ensure_dirs()
    stamp = str(int(time.time() * 1000))
    db_path = os.path.join(MASTER_DATA_DIR, '_merge_' + stamp + '.db')
    out_files = []
    fhs = []
    counts = [0] * n_nodes
    for i in range(n_nodes):
        p = os.path.join(MASTER_DATA_DIR, '_chunk_' + stamp + '_' + str(i) + '.txt')
        fhs.append(open(p, 'w', encoding='utf-8'))
        out_files.append(p)
    try:
        conn = sqlite3.connect(db_path)
        conn.execute('PRAGMA journal_mode=OFF')
        conn.execute('PRAGMA synchronous=OFF')
        conn.execute('PRAGMA temp_store=MEMORY')
        conn.execute('CREATE TABLE IF NOT EXISTS e (email TEXT, domain TEXT)')
        n = 0
        for src in sources:
            if not src or not os.path.exists(src):
                continue
            with open(src, 'r', encoding='utf-8') as f:
                for line in f:
                    e = line.strip().lower()
                    if not e or '@' not in e:
                        continue
                    conn.execute('INSERT INTO e(email,domain) VALUES(?,?)',
                                 (e, e.split('@')[-1]))
                    n += 1
                    if n % 200000 == 0:
                        conn.commit()
        conn.commit()
        # 按域排序导出, 域内 round-robin 平均摊到各节点
        rem = [int(q or 0) for q in quotas] if quotas else [0] * n_nodes
        use_quota = any(x > 0 for x in rem)
        leftover = []
        rr = 0
        for email, _dom in conn.execute('SELECT email, domain FROM e ORDER BY domain, rowid'):
            placed = False
            for _ in range(n_nodes):
                i = rr % n_nodes
                rr += 1
                if (not use_quota) or rem[i] > 0:
                    fhs[i].write(email + '\n')
                    counts[i] += 1
                    if use_quota:
                        rem[i] -= 1
                    placed = True
                    break
            if not placed:
                leftover.append(email)
        # 配额耗尽后的余量: 只追加给未指定配额(均分)的节点; 配额0的节点绝不超发
        if leftover:
            zero_quota_idx = [i for i, q in enumerate(rem if quotas else []) if int(q or 0) <= 0]
            if not quotas:
                zero_quota_idx = list(range(n_nodes))
            if zero_quota_idx:
                i = 0
                for email in leftover:
                    fhs[zero_quota_idx[i % len(zero_quota_idx)]].write(email + '\n')
                    counts[zero_quota_idx[i % len(zero_quota_idx)]] += 1
                    i += 1
            else:
                print('[WARN] 配额切分: %d 个邮箱因所有节点配额耗尽被丢弃' % len(leftover))
        conn.close()
        for fh in fhs:
            fh.close()
        try:
            os.remove(db_path)
        except Exception:
            pass
        return out_files, counts
    except Exception:
        for fh in fhs:
            try:
                fh.close()
            except Exception:
                pass
        try:
            os.remove(db_path)
        except Exception:
            pass
        return None, None


def _auto_boxes_from_dkim():
    """从主控已生成DKIM密钥的域名自动生成发件箱列表
    (域名管理配过DKIM的域自动可用于发件, 不用在任务里再填一次)"""
    boxes = []
    dkim_dir = os.path.join(_THIS_DIR, 'dkim_keys')
    if not os.path.isdir(dkim_dir):
        return boxes
    seen = set()
    try:
        for fn in sorted(os.listdir(dkim_dir)):
            if not fn.endswith('.pem'):
                continue
            dom = fn.rsplit('_', 1)[0]
            if dom and '.' in dom and dom not in seen:
                seen.add(dom)
                boxes.append({'email': 'noreply@' + dom, 'sender_name': 'お知らせ',
                              'random_user': True, 'dkim_selector': 'default'})
    except Exception:
        pass
    return boxes


def task_create(name, recipients_source, content, send_cfg, boxes,
                node_ids, node_proxies=None, node_threads=None, node_quotas=None,
                boxes_per_node=0, dedupe=True, dkim_keys=None, auto_start=True,
                recipients_sources=None, shuffle_merge=True, data_only=False):
    """建任务并分发启动
    recipients_source: 主控本地的收件人文件路径 或 直接列表
    node_proxies: {node_id: [proxy dict...]} 每节点专属代理池 (缺省用 send_cfg 里的)
    node_threads: {node_id: int}
    boxes_per_node: >0 时按节点均分发件箱 (每节点分到 N 个子域名, 依次轮取; 0=全部共用)
    dedupe: 大文件强制走流式去重+切分 (SQLite, 不爆内存)
    返回 (ok, msg, task_dict)
    """
    _ensure_dirs()
    if not _REQ_OK:
        return False, 'requests 未安装 (pip install requests)', None
    if not node_ids:
        return False, '未选择节点', None
    # ★ 防重复创建: 相同收件人源+相同节点, 8秒内重复提交视为连点/重发, 直接拒绝
    try:
        _recent = master_db._q("SELECT task_id FROM tasks WHERE created_at>=? ORDER BY created_at DESC LIMIT 8",
                               (time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time() - 8)),))
        if _recent:
            for _r in _recent:
                _ot = master_db.task_get(_r[0])
                _otn = ([str(x.get('node_id')) for x in (_ot or {}).get('nodes', [])] or [])
                if _otn and sorted(_otn) == sorted([str(x) for x in node_ids]):
                    return False, '8秒内已创建过同节点任务(防重复点击) — 请勿连点, 稍后再试', None
    except Exception:
        pass
    nodes = []
    ms_list = []
    for nid in node_ids:
        if str(nid or '').startswith('ms'):
            ms = master_db.mail_server_get(str(nid))
            if not ms:
                return False, '邮局服务器不存在: ' + str(nid), None
            if not ms.get('smtp_user') or not ms.get('smtp_pwd'):
                return False, '邮局服务器未部署(无SMTP账号): ' + (ms.get('name') or nid) + ' — 请先在📮邮局部署Tab一键部署', None
            ms_list.append(ms)
        else:
            n = _find_node(nid)
            if not n:
                return False, '节点不存在: ' + str(nid), None
            nodes.append(n)
    all_targets = nodes + [dict(m, id=m['id'], name=m.get('name') or m['hostname']) for m in ms_list]

    # 每节点自定义数量 (node_quotas: {node_id: int}); 全0/缺省=均分
    quotas = [int((node_quotas or {}).get(n['id'], 0) or 0) for n in all_targets]
    use_quota = any(q > 0 for q in quotas)

    # 分配阶段前置校验 (完整模式强制; 只分配数据模式仅自动补全+警告)
    warnings = []
    # ★ 自动补全发件箱: 任务没带发件箱时, 从已配DKIM的域名自动生成
    if not boxes or not any(('@' in str(b.get('email', ''))) for b in boxes):
        auto_boxes = _auto_boxes_from_dkim()
        if auto_boxes:
            boxes = auto_boxes
    if not boxes or not any(('@' in str(b.get('email', ''))) for b in boxes):
        # 纯邮局任务不需要发件箱 (发件地址=邮局的smtp_user); 含直投节点才强制
        if nodes:
            if data_only:
                warnings.append('未配置发件域名/发件箱(启动前需补)')
            else:
                return False, '未配置发件箱 — 请在任务表单填写发件箱JSON, 例: [{"email":"noreply@你的域名.com","sender_name":"お知らせ"}]', None
    # 代理/DKIM 校验只对直投节点生效 (邮局节点不走代理, DKIM由邮局签)
    if nodes:
        # ★ API自动获取模式: 有完整API链接下发时, 不需要静态代理池 (节点自己拉取)
        _api_mode = bool((send_cfg or {}).get('proxy_api') and (send_cfg or {}).get('proxy_api', {}).get('url'))
        eff_proxies = ((node_proxies or {}).get(nodes[0]['id'])
                       or (send_cfg or {}).get('proxies')
                       or master_db.proxy_as_list() or [])
        if not eff_proxies and not _api_mode:
            if data_only:
                warnings.append('未配置代理IP(启动前需在代理IP池添加)')
            else:
                return False, '未配置代理IP — 请在「📦资源配置→代理IP池」添加, 或在任务表单填写代理JSON', None
    if dkim_keys is None:
        dkim_keys = _load_master_dkim_keys(boxes or [])
    if not dkim_keys and nodes:
        if data_only:
            warnings.append('未找到DKIM密钥(启动前需在域名管理生成)')
        else:
            return False, '未找到DKIM密钥 — 请在「域名管理」为发件域名生成DKIM密钥 (否则发信必进垃圾箱)', None

    # 收件人来源: 多文件列表 > 单文件 > 内存列表
    sources = list(recipients_sources or [])
    if not sources and isinstance(recipients_source, str) and os.path.exists(recipients_source):
        sources = [recipients_source]

    stream_mode = False
    node_files = None
    node_counts = None
    recipients = None
    if len(sources) > 1 or (sources and shuffle_merge):
        # 多文件/打乱模式: 合并+去重+按域均衡平均分配 (避免单节点全是同一运营商)
        stream_mode = True
        node_files, node_counts = _merge_shuffle_split_stream(sources, len(all_targets), quotas)
        if node_files is None:
            return False, '多文件合并打乱处理失败', None
        recipients_total = sum(node_counts)
    elif sources:
        src = sources[0]
        if dedupe or os.path.getsize(src) >= 50 * 1024 * 1024:
            stream_mode = True
            if use_quota:
                node_files, node_counts = _stream_dedupe_split_by_quota(src, quotas)
            else:
                node_files, node_counts = _stream_dedupe_split(src, len(all_targets))
            if node_files is None:
                return False, '收件人文件处理失败 (去重/切分出错)', None
            recipients_total = sum(node_counts)
        else:
            recipients = _load_recipients_lines(src)
            recipients_total = len(recipients)
    elif isinstance(recipients_source, (list, tuple)):
        recipients = list(recipients_source)
        if dedupe:
            recipients = list(dict.fromkeys(recipients))
        recipients_total = len(recipients)
    else:
        return False, '收件人文件不存在', None
    if recipients_total <= 0:
        return False, '收件人为空', None

    if not stream_mode:
        if use_quota:
            chunks = _split_by_quotas(recipients, quotas)
        else:
            chunks = _split_recipients(recipients, len(all_targets))
    task_id = 't' + time.strftime('%Y%m%d%H%M%S') + str(int(time.time() * 1000) % 1000)
    # DKIM 密钥: 显式传入 > 主控本地 dkim_keys 目录自动加载
    if dkim_keys is None:
        dkim_keys = _load_master_dkim_keys(boxes or [])
    task = {
        'task_id': task_id,
        'name': name or task_id,
        'created_at': time.strftime('%Y-%m-%d %H:%M:%S'),
        'content': content or {},
        'send_cfg': send_cfg or {},
        'boxes': boxes or [],
        'recipients_total': recipients_total,
        'nodes': [],
        'result_cursor': {},   # node_id -> 已拉取行数
    }

    # 1. 逐目标分发: 直投节点推任务包; 邮局服务器名单落本地文件(发信worker读)
    direct_idx = 0
    for i, target in enumerate(all_targets):
        is_ms = str(target['id'] or '').startswith('ms')
        if is_ms:
            # 邮局服务器: 名单写主控本地, 启动时由发信worker逐封SMTP提交
            ms_file = os.path.join(MASTER_DATA_DIR, '_ms_' + task_id + '_' + target['id'] + '.txt')
            if os.path.exists(ms_file):
                try:
                    os.remove(ms_file)
                except Exception:
                    pass
            chunk_data = []
            if stream_mode:
                with open(node_files[i], 'r', encoding='utf-8') as f:
                    chunk_data = [l.strip() for l in f if l.strip()]
            else:
                chunk_data = chunks[i]
            with open(ms_file, 'w', encoding='utf-8') as f:
                f.write('\n'.join(chunk_data))
            task['nodes'].append({
                'node_id': target['id'],
                'node_name': target.get('name') or target.get('hostname') or target['ip'],
                'recipients': len(chunk_data),
                'push_status': 'assigned',
                'node_type': 'mail_server',
            })
            continue
        node = target
        proxies = ((node_proxies or {}).get(node['id'])
                   or task['send_cfg'].get('proxies')
                   or master_db.proxy_as_list() or [])
        cfg_for_node = dict(task['send_cfg'])
        cfg_for_node['proxies'] = proxies
        if node_threads and node_threads.get(node['id']):
            cfg_for_node['thread_count'] = int(node_threads[node['id']])
        # ★ 每节点子域名分配: boxes_per_node>0 时, 依次给每个节点切 N 个发件箱 (各自独立)
        node_boxes = task['boxes']
        if boxes_per_node > 0 and task['boxes']:
            _n = int(boxes_per_node)
            _start = (direct_idx * _n) % max(1, len(task['boxes']))
            node_boxes = []
            for _k in range(_n):
                node_boxes.append(task['boxes'][(_start + _k) % len(task['boxes'])])
            # 每节点专属 DKIM 密钥 (只带该节点用到的域名)
            node_dkim = _load_master_dkim_keys(node_boxes) or dkim_keys
        else:
            node_dkim = dkim_keys
        # ★ 同IP同MX限速按直投节点数均分: 魔改里填的是"全集群对单MX的总上限"
        #   N 个节点各自只统计自己worker → 每节点配额 = 总上限 / 节点数 (向上取整, 至少1)
        _mx_rate_total = int(task['send_cfg'].get('mx_rate_per_minute') or 0)
        if _mx_rate_total > 0 and len(nodes) > 1:
            import math as _math
            per_node = max(1, _math.ceil(_mx_rate_total / len(nodes)))
            cfg_for_node['mx_rate_per_minute'] = per_node
        payload = {
            'task_id': task_id,
            'content': task['content'],
            'send_cfg': cfg_for_node,
            'proxies': proxies,
            'boxes': node_boxes,
            'dkim_keys': node_dkim,
        }
        got_count = 0
        if stream_mode:
            ok, data = _node_call(node, 'POST', '/task', payload, timeout=120)
            if not ok:
                return False, '下发任务失败 [' + node['name'] + ']: ' + str((data or {}).get('msg', '')), task
            ok, pushed = _push_recipients_file(node, task_id, node_files[i])
            if not ok:
                return False, '收件人推送失败 [' + node['name'] + ']', task
            got_count = pushed
        else:
            payload['recipients_text'] = '\n'.join(chunks[i])
            ok, data = _node_call(node, 'POST', '/task', payload, timeout=600)
            if not ok:
                return False, '下发任务失败 [' + node['name'] + ']: ' + str((data or {}).get('msg', '')), task
            got_count = len(chunks[i])
        task['nodes'].append({
            'node_id': node['id'],
            'node_name': node['name'],
            'recipients': got_count,
            'push_status': 'assigned',
        })
        direct_idx += 1

    # 清理直投临时切分文件 (邮局名单已复制到本地 _ms_*.txt, 临时文件可删)
    if node_files:
        for p in node_files:
            try:
                os.remove(p)
            except Exception:
                pass

    # 2. 两阶段流程: auto_start=False 只分配不启动 (节点待命, 等主控 /task_start)
    started, failed = 0, []
    if auto_start:
        for target in all_targets:
            if str(target['id'] or '').startswith('ms'):
                _start_ms_worker(task, target, content or {}, send_cfg or {})
                started += 1
                for tn in task['nodes']:
                    if tn['node_id'] == target['id']:
                        tn['push_status'] = 'running'
                continue
            node = target
            ok, data = _node_call(node, 'POST', '/start', {}, timeout=30)
            if ok and data.get('ok'):
                started += 1
            else:
                failed.append(node['name'] + ': ' + str((data or {}).get('msg', '')))
            for tn in task['nodes']:
                if tn['node_id'] == node['id']:
                    tn['push_status'] = 'running' if (ok and data.get('ok')) else 'start_failed'

    # 3. 落盘任务记录 (SQLite)
    task['dkim_keys'] = dkim_keys
    task['status'] = 'running' if auto_start else 'assigned'
    with _lock:
        master_db.task_insert(task)

    if not auto_start:
        m = '已分配 ' + str(len(all_targets)) + ' 个目标 (' + str(recipients_total) + ' 收件人), 待命 — 编辑内容后点启动'
        if data_only and warnings:
            m += ' | ⚠️ ' + '; '.join(warnings)
        return True, m, task
    if failed:
        return True, '已启动 ' + str(started) + '/' + str(len(all_targets)) + ' 个目标, 失败: ' + '; '.join(failed), task
    return True, '任务已分发, ' + str(len(all_targets)) + ' 个目标已启动 (' + str(recipients_total) + ' 收件人)', task


# 邮局发信 worker 停止信号: {task_id: {ms_id: threading.Event}}
_MS_STOP_EVENTS = {}


def _ms_stop_event(task_id, ms_id):
    with _lock:
        d = _MS_STOP_EVENTS.setdefault(task_id, {})
        ev = d.get(ms_id)
        if ev is None:
            ev = threading.Event()
            d[ms_id] = ev
        return ev


# 运行中的 worker 标记: {(task_id, ms_id): thread} — 防重复启动双份发送
_MS_WORKERS = {}


def _start_ms_worker(task, ms, content, send_cfg):
    """启动邮局发信 worker (后台线程, 限速逐封SMTP提交)
    ★ 运行中守卫: 同一 (task_id, ms_id) 已有 worker 在跑时拒绝重复启动"""
    task_id = task.get('task_id', '')
    ms_id = ms.get('id', '')
    key = (task_id, ms_id)
    with _lock:
        t_old = _MS_WORKERS.get(key)
        if t_old is not None and t_old.is_alive():
            print('[WARN] 邮局worker已在运行, 拒绝重复启动: %s / %s' % (task_id, ms_id))
            return False
        _ms_stop_event(task_id, ms_id).clear()
        t = threading.Thread(target=_ms_send_worker, daemon=True,
                             args=(task, ms, content, send_cfg))
        _MS_WORKERS[key] = t
        t.start()
    return True


def _is_ascii(s):
    try:
        s.encode('ascii')
        return True
    except Exception:
        return False


def _ms_send_worker(task, ms, content, send_cfg):
    """邮局发信 worker: 读本地名单 → 限速 → SMTP认证提交 → 结果入库 (支持停止+实时进度)"""
    import ms_sender
    task_id = task.get('task_id', '')
    ms_id = ms.get('id', '')
    ms_name = ms.get('name') or ms.get('hostname') or ms.get('ip', '')
    stop_ev = _ms_stop_event(task_id, ms_id)
    fpath = os.path.join(MASTER_DATA_DIR, '_ms_' + task_id + '_' + ms_id + '.txt')
    cursor_path = fpath + '.cursor'
    try:
        ms_cfg = (send_cfg or {}).get('mail_server') or {}
        # ★ 发件显示名优先级: 任务发件箱的sender_name > 魔改邮局专区显示名池 > 默认池
        names_pool = list(ms_cfg.get('sender_names') or [])
        box_names = []
        for b in (task.get('boxes') or []):
            bn = (b.get('sender_name') if isinstance(b, dict) else '').strip() if isinstance(b, dict) else ''
            if bn and _is_ascii(bn) and bn not in box_names:
                box_names.append(bn)
        if box_names:
            names_pool = box_names
        if not names_pool:
            names_pool = ms_sender.DEFAULT_SENDER_NAMES
        charset = ms_cfg.get('mail_charset') or 'utf-8'
        avoid_spam = bool(ms_cfg.get('avoid_spam_words', True))
        reply_to = ms_cfg.get('reply_to') or ''
        list_unsub = ms_cfg.get('list_unsubscribe') or ''
        interval = max(0.3, float(ms_cfg.get('send_interval') or 1))
        minute_limit = int(ms.get('minute_limit') if ms.get('minute_limit') is not None else 0)
        day_limit = int(ms.get('day_limit') if ms.get('day_limit') is not None else 0)
        if minute_limit < 0:
            minute_limit = 0
        if day_limit < 0:
            day_limit = 0
        # 0 = 无限制速度
        subjects = list((content or {}).get('subjects') or ['Information'])
        htmls = list((content or {}).get('html_parts') or ['<html><body></body></html>'])
        from_email = ms.get('smtp_user') or ('admin@' + ms.get('hostname', 'localhost'))
        recipients = []
        if os.path.exists(fpath):
            with open(fpath, 'r', encoding='utf-8') as f:
                recipients = [l.strip() for l in f if l.strip() and '@' in l]
        done = 0
        if os.path.exists(cursor_path):
            try:
                done = int(open(cursor_path, 'r', encoding='utf-8').read().strip() or 0)
            except Exception:
                done = 0
        total = len(recipients)
        master_db.task_node_status_set(task_id, ms_id, 'running')
        sent = fail = 0
        minute_key = int(time.time() // 60)
        minute_cnt = 0
        day_str = time.strftime('%Y-%m-%d')
        day_cnt = 0
        paused = False
        for i, email in enumerate(recipients):
            if i < done:
                continue
            if stop_ev.is_set():
                # ★ 停止: 保留名单文件和游标, 重启后断点续发
                master_db.task_node_status_set(task_id, ms_id, 'stopped')
                return
            now_min = int(time.time() // 60)
            if now_min != minute_key:
                minute_key = now_min
                minute_cnt = 0
            today = time.strftime('%Y-%m-%d')
            if today != day_str:
                day_str = today
                day_cnt = 0
            if minute_limit > 0 and minute_cnt >= minute_limit:
                time.sleep(max(2, 60 - time.time() % 60))
                minute_key = int(time.time() // 60)
                minute_cnt = 0
            if day_limit > 0 and day_cnt >= day_limit:
                paused = True
                break
            try:
                subject = random.choice(subjects).replace('[email]', email).replace('{email}', email)
            except Exception:
                subject = 'Information'
            try:
                html = random.choice(htmls).replace('[email]', email).replace('{email}', email)
            except Exception:
                html = '<html><body></body></html>'
            try:
                sender_name = random.choice([n for n in names_pool if n])
            except Exception:
                sender_name = 'Notice'
            lu = ''
            if list_unsub:
                lu = list_unsub.replace('[email]', email).replace('{email}', email)
            rt = ''
            if reply_to:
                rt = reply_to.replace('[email]', email).replace('{email}', email)
            ok, code, msg = ms_sender.send_one(
                ms, from_email, str(sender_name), email, subject, html,
                mail_charset=charset, avoid_spam_words=avoid_spam,
                reply_to=rt, list_unsubscribe=lu)
            master_db.result_insert_many([(
                task_id, ms_id, email, 'sent' if ok else 'fail',
                str(code or ''), str(msg or '')[:120],
                time.strftime('%Y-%m-%d %H:%M:%S'), '')])
            if ok:
                sent += 1
                day_cnt += 1
                minute_cnt += 1
            else:
                fail += 1
            done = i + 1
            with open(cursor_path, 'w', encoding='utf-8') as f:
                f.write(str(done))
            # 实时进度: 每封更新游标 (主控 task_status 读取显示)
            master_db.task_cursor_set(task_id, ms_id, done)
            time.sleep(interval)
        if stop_ev.is_set():
            master_db.task_node_status_set(task_id, ms_id, 'stopped')
        else:
            master_db.task_node_status_set(task_id, ms_id, 'paused' if paused else 'finished')
        # ★ 只有自然发完才清名单; 停止/暂停保留文件和游标, 重启后断点续发
        if not stop_ev.is_set() and not paused:
            try:
                os.remove(fpath)
            except Exception:
                pass
            try:
                os.remove(cursor_path)
            except Exception:
                pass
    except Exception as e:
        try:
            master_db.task_node_status_set(task_id, ms_id, 'failed')
        except Exception:
            pass
    finally:
        # 运行标记清理: 允许同 (task_id, ms_id) 下次重新启动
        with _lock:
            t = _MS_WORKERS.get((task_id, ms_id))
            if t is not None and not t.is_alive():
                _MS_WORKERS.pop((task_id, ms_id), None)


def task_start(task_id, content=None, node_ids=None):
    """启动已分配的任务: 可选先下发内容 → 各节点 /start; 邮局服务器起本地发信worker
    node_ids: 只启动指定节点 (批量启动支持选择)
    返回 (ok, msg)
    """
    task = _get_task(task_id)
    if not task:
        return False, '任务不存在'
    # 只在内容有效时覆盖 (空内容保持任务已有内容, 防覆盖分配内容时存的正确内容)
    if _content_valid(content):
        task['content'] = content or {}
    started, failed = 0, []
    node_list = task.get('nodes') or []
    if node_ids:
        node_list = [tn for tn in node_list if tn.get('node_id') in node_ids]
    if not node_list:
        return False, '所选节点不在本任务分配范围内 (请先点「数据分配」)'
    for tn in node_list:
        tn_id = tn.get('node_id') or ''
        if str(tn_id).startswith('ms'):
            ms = master_db.mail_server_get(str(tn_id))
            if not ms:
                failed.append(tn.get('node_name', '') + '(邮局不存在)')
                continue
            _start_ms_worker(task, ms, task.get('content') or {}, task.get('send_cfg') or {})
            started += 1
            master_db.task_node_status_set(task_id, tn_id, 'running')
            continue
        node = _find_node(tn_id)
        if not node:
            continue
        if content:
            ok, data = _node_call(node, 'POST', '/content',
                                  {'task_id': task_id, 'content': content}, timeout=30)
            if not ok:
                failed.append(node.get('name', '') + '(内容下发失败)')
                continue
        ok, data = _node_call(node, 'POST', '/start', {}, timeout=30)
        if ok and data.get('ok'):
            started += 1
            master_db._exec('UPDATE task_nodes SET push_status="running" WHERE task_id=? AND node_id=?',
                            (task_id, node['id']))
        else:
            failed.append(node.get('name', '') + ': ' + str((data or {}).get('msg', '')))
    master_db._commit()
    master_db.task_set_status(task_id, 'running')
    if content:
        master_db.task_update_content(task_id, content)
    if failed:
        return True, '已启动 ' + str(started) + '/' + str(len(node_list)) + ' 目标, 失败: ' + '; '.join(failed)
    return True, '任务已启动: ' + str(started) + ' 个目标开始工作'


def node_start(node_id):
    """单独启动某节点的已分配任务"""
    node = _find_node(node_id)
    if not node:
        return False, '节点不存在'
    ok, data = _node_call(node, 'POST', '/start', {}, timeout=30)
    return (ok and (data or {}).get('ok')), ((data or {}).get('msg') or '节点已启动')


def mail_servers_live():
    """邮局服务器列表 + 实时发送状态 (最新任务的状态/待发/处理/成功/失败)"""
    servers = master_db.mail_server_all()
    for s in servers:
        s['live'] = {'task_id': '', 'status': '', 'assigned': 0, 'pending': 0,
                     'processed': 0, 'success': 0, 'fail': 0}
        tasks = master_db.tasks_by_node(s['id'])
        if not tasks:
            continue
        tid = tasks[-1]  # 最新任务
        task = _get_task(tid)
        if not task:
            continue
        assigned = 0
        st = ''
        for tn in task.get('nodes') or []:
            if tn.get('node_id') == s['id']:
                assigned = int(tn.get('recipients') or 0)
                st = tn.get('push_status', '')
                break
        cursor = master_db.task_cursor_get(tid, s['id'])
        stats = master_db.result_stats_by_node(tid, s['id'])
        s['live'] = {'task_id': tid, 'status': st or '', 'assigned': assigned,
                     'pending': max(0, assigned - cursor), 'processed': cursor,
                     'success': stats.get('sent', 0), 'fail': stats.get('fail', 0)}
    return servers


def node_stop(node_id):
    """单独停止某节点"""
    node = _find_node(node_id)
    if not node:
        return False, '节点不存在'
    ok, data = _node_call(node, 'POST', '/stop', {}, timeout=15)
    return ok, ''


def _content_valid(content):
    """内容非空判定: 有非空主题或非空正文才算有效"""
    if not content or not isinstance(content, dict):
        return False
    subs = [s for s in (content.get('subjects') or []) if s and str(s).strip()]
    htmls = [h for h in (content.get('html_parts') or []) if h and str(h).strip()]
    return bool(subs or htmls)


def nodes_start(node_ids, content=None):
    """批量启动多个目标 (直投节点 /start; 邮局服务器起本地发信worker)
    content: 可选, 仅在内容非空时更新任务内容 (避免空编辑器覆盖已有内容)
    """
    if not node_ids:
        return False, '未选择节点'
    succ, fail = 0, 0
    for nid in node_ids:
        if str(nid or '').startswith('ms'):
            ms = master_db.mail_server_get(str(nid))
            if not ms:
                fail += 1
                continue
            started = 0
            for tid in master_db.tasks_by_node(str(nid)):
                task = _get_task(tid)
                if not task:
                    continue
                if _content_valid(content):
                    task['content'] = content or {}
                    master_db.task_update_content(tid, content or {})
                _start_ms_worker(task, ms, task.get('content') or {}, task.get('send_cfg') or {})
                master_db.task_node_status_set(tid, str(nid), 'running')
                master_db.task_set_status(tid, 'running')
                started += 1
            if started:
                succ += 1
            else:
                fail += 1
            continue
        node = _find_node(nid)
        if not node:
            fail += 1
            continue
        ok, _ = _node_call(node, 'POST', '/start', {}, timeout=30)
        if ok:
            succ += 1
        else:
            fail += 1
    return True, '启动完成: 成功 %d 个, 失败 %d 个' % (succ, fail)


def nodes_stop(node_ids):
    """批量停止多个目标 (直投节点 /stop; 邮局服务器置停止信号)"""
    if not node_ids:
        return False, '未选择节点'
    succ, fail = 0, 0
    for nid in node_ids:
        if str(nid or '').startswith('ms'):
            stopped = 0
            for tid in master_db.tasks_by_node(str(nid)):
                _ms_stop_event(tid, str(nid)).set()
                stopped += 1
            if stopped:
                succ += 1
            else:
                fail += 1
            continue
        node = _find_node(nid)
        if not node:
            fail += 1
            continue
        ok, _ = _node_call(node, 'POST', '/stop', {}, timeout=15)
        if ok:
            succ += 1
        else:
            fail += 1
    return True, '停止完成: 成功 %d 个, 失败 %d 个' % (succ, fail)


def nodes_clear(node_ids, clear_proxy_only=False):
    """批量清理目标: 直投节点走远程/clear; 邮局服务器清本地名单+游标+任务分配+结果
    返回 [{node_id, name, ok, msg}]
    """
    results = []
    for nid in node_ids:
        if str(nid or '').startswith('ms'):
            ms = master_db.mail_server_get(str(nid))
            if not ms:
                results.append({'node_id': nid, 'name': nid, 'ok': False, 'msg': '邮局服务器不存在'})
                continue
            if clear_proxy_only:
                results.append({'node_id': nid, 'name': ms.get('name') or ms.get('hostname', nid),
                                'ok': True, 'msg': '邮局不用代理, 无需清理'})
                continue
            # 清数据: 本地名单文件+游标+该邮局的任务分配+结果
            n_rm = 0
            try:
                for f in os.listdir(MASTER_DATA_DIR):
                    if f.startswith('_ms_') and ('_' + str(nid) + '.txt' in f or ('_' + str(nid) + '.txt') in f):
                        try:
                            os.remove(os.path.join(MASTER_DATA_DIR, f))
                            n_rm += 1
                        except Exception:
                            pass
            except Exception:
                pass
            tasks = master_db.tasks_by_node(str(nid))
            for tid in tasks:
                master_db.task_node_status_set(tid, str(nid), 'cleared')
            results.append({'node_id': nid, 'name': ms.get('name') or ms.get('hostname', nid),
                            'ok': True, 'msg': '已清理 %d 个名单文件, %d 个任务标记清理' % (n_rm, len(tasks))})
            continue
        node = _find_node(nid)
        if not node:
            results.append({'node_id': nid, 'name': nid, 'ok': False, 'msg': '节点不存在'})
            continue
        path = '/clear_proxy' if clear_proxy_only else '/clear'
        ok, data = _node_call(node, 'POST', path, {}, timeout=25)
        results.append({'node_id': nid, 'name': node.get('name', nid), 'ok': ok,
                        'msg': (data or {}).get('msg', '')})
    return results


def node_set_boxes(node_ids, boxes, dkim_keys=None):
    """批量给选中节点下发发件域名(发件箱+DKIM密钥)"""
    results = []
    for nid in node_ids:
        node = _find_node(nid)
        if not node:
            results.append({'node_id': nid, 'ok': False, 'msg': '节点不存在'})
            continue
        ok, data = _node_call(node, 'POST', '/set_boxes',
                              {'boxes': boxes or [], 'dkim_keys': dkim_keys or {}}, timeout=25)
        results.append({'node_id': nid, 'name': node.get('name', nid), 'ok': ok,
                        'msg': (data or {}).get('msg', '')})
    return results


def node_set_content(node_ids, content):
    """批量给选中目标下发内容 (直投节点推远程; 邮局服务器更新任务记录内容)"""
    results = []
    for nid in node_ids:
        if str(nid or '').startswith('ms'):
            ms = master_db.mail_server_get(str(nid))
            if not ms:
                results.append({'node_id': nid, 'ok': False, 'msg': '邮局服务器不存在'})
                continue
            # 邮局: 内容存任务记录, 启动时发信worker读取
            ms_tasks = master_db.tasks_by_node(str(nid))
            for tid in ms_tasks:
                master_db.task_update_content(tid, content)
            results.append({'node_id': nid, 'name': ms.get('name') or ms.get('hostname', nid),
                            'ok': True, 'msg': '内容已更新' + (' (%d个任务, 启动时生效)' % len(ms_tasks) if ms_tasks else ' (分配任务后启动时生效)')})
            continue
        node = _find_node(nid)
        if not node:
            results.append({'node_id': nid, 'ok': False, 'msg': '节点不存在'})
            continue
        ok, data = _node_call(node, 'POST', '/set_content', {'content': content or {}}, timeout=25)
        results.append({'node_id': nid, 'name': node.get('name', nid), 'ok': ok,
                        'msg': (data or {}).get('msg', '')})
    return results


def node_set_proxy(node_ids, proxies):
    """批量给选中节点下发代理池 (独立分配代理)"""
    results = []
    for nid in node_ids:
        node = _find_node(nid)
        if not node:
            results.append({'node_id': nid, 'ok': False, 'msg': '节点不存在'})
            continue
        ok, data = _node_call(node, 'POST', '/set_proxy', {'proxies': proxies or []}, timeout=25)
        results.append({'node_id': nid, 'name': node.get('name', nid), 'ok': ok,
                        'msg': (data or {}).get('msg', '')})
    return results


def node_set_config(node_ids, send_cfg):
    """批量给选中目标下发魔改配置 (直投节点推远程; 邮局服务器合并进任务send_cfg, 启动时生效)"""
    results = []
    for nid in node_ids:
        if str(nid or '').startswith('ms'):
            ms = master_db.mail_server_get(str(nid))
            if not ms:
                results.append({'node_id': nid, 'ok': False, 'msg': '邮局服务器不存在'})
                continue
            # 邮局: 魔改配置合并进该邮局相关任务的 send_cfg (mail_server 区生效)
            updated = 0
            for tid in master_db.tasks_by_node(str(nid)):
                task = _get_task(tid)
                if not task:
                    continue
                cfg = dict(task.get('send_cfg') or {})
                cfg['mail_server'] = send_cfg.get('mail_server') or {}
                master_db.task_update_send_cfg(tid, cfg)
                updated += 1
            results.append({'node_id': nid, 'name': ms.get('name') or ms.get('hostname', nid),
                            'ok': True, 'msg': '邮局魔改配置已更新' + (' (%d个任务, 启动时生效)' % updated if updated else ' (新建任务自动带此配置)')})
            continue
        node = _find_node(nid)
        if not node:
            results.append({'node_id': nid, 'ok': False, 'msg': '节点不存在'})
            continue
        ok, data = _node_call(node, 'POST', '/set_config', {'send_cfg': send_cfg or {}}, timeout=25)
        results.append({'node_id': nid, 'name': node.get('name', nid), 'ok': ok,
                        'msg': (data or {}).get('msg', '')})
    return results


def tasks_clear(task_ids):
    """批量清空任务记录 (任务+结果)"""
    n = 0
    for tid in task_ids:
        master_db.task_remove(tid)
        master_db.result_clear(tid)
        n += 1
    return n


def tasks_stop(task_ids):
    """批量停止任务"""
    ok_n, fail_n = 0, 0
    for tid in task_ids:
        ok, _msg = task_stop(tid)
        if ok:
            ok_n += 1
        else:
            fail_n += 1
    return ok_n, fail_n


def tasks_pull(task_ids):
    """批量拉取结果入库"""
    total = 0
    for tid in task_ids:
        r = task_pull_results(tid)
        if r:
            total += int(r.get('rows_pulled') or 0)
    return total


def task_status(task_id):
    """聚合各节点状态 (直投节点远程查询; 邮局服务器本地DB实时状态)"""
    task = _get_task(task_id)
    if not task:
        return None
    nodes_info = []
    agg = {'processed': 0, 'success': 0, 'fail': 0, 'bounced': 0,
           'ip_rejected': 0, 'switches': 0, 'speed': 0, 'elapsed': 0}
    for tn in task.get('nodes') or []:
        tn_id = tn.get('node_id') or ''
        info = {'node_id': tn_id, 'node_name': tn.get('node_name'),
                'assigned': tn.get('recipients', 0), 'online': False,
                'node_state': '', 'engine_status': '', 'recipients_count': 0,
                'result_count': 0, 'engine': {}}
        if str(tn_id).startswith('ms'):
            # 邮局服务器: 本地DB实时状态 (worker每封更新游标+结果)
            info['online'] = True
            st = tn.get('push_status', '')
            cursor = master_db.task_cursor_get(task_id, tn_id)
            stats = master_db.result_stats_by_node(task_id, tn_id)
            info['node_state'] = st
            info['engine_status'] = st
            info['recipients_count'] = tn.get('recipients', 0)
            info['result_count'] = stats.get('total', 0)
            info['engine'] = {
                'status': st, 'total': tn.get('recipients', 0),
                'processed': cursor, 'success': stats.get('sent', 0),
                'fail': stats.get('fail', 0), 'bounced': 0, 'ip_rejected': 0,
                'switches': 0, 'elapsed': 0, 'speed': 0,
            }
            agg['processed'] += cursor
            agg['success'] += stats.get('sent', 0)
            agg['fail'] += stats.get('fail', 0)
        else:
            node = _find_node(tn_id)
            if node:
                ok, data = _node_call(node, 'GET', '/status', timeout=10)
                if ok:
                    info['online'] = True
                    info['node_state'] = (data or {}).get('node_state', {}).get('status', '')
                    eng = (data or {}).get('engine') or {}
                    info['engine_status'] = eng.get('status', '')
                    info['engine'] = eng
                    info['recipients_count'] = (data or {}).get('recipients_count', 0)
                    info['result_count'] = (data or {}).get('result_count', 0)
                    # ★ 节点日志同步: 取最后30行(获取代理/握手成功/失败详情)
                    info['logs'] = (data or {}).get('logs') or []
                    for k in ('processed', 'success', 'fail', 'bounced', 'ip_rejected', 'switches'):
                        agg[k] += int(eng.get(k) or 0)
                    agg['elapsed'] = max(agg['elapsed'], float(eng.get('elapsed') or 0))
        nodes_info.append(info)
    total_processed = agg['processed']
    if total_processed > 0:
        agg['speed'] = round(total_processed / max(agg['elapsed'], 1), 1)
    return {'task': task, 'nodes': nodes_info, 'agg': agg}


def task_stop(task_id):
    task = _get_task(task_id)
    if not task:
        return False, '任务不存在'
    failed = []
    stopped_ms = 0
    for tn in task.get('nodes') or []:
        tn_id = tn.get('node_id') or ''
        if str(tn_id).startswith('ms'):
            # 邮局发信worker: 置停止信号, 线程收到后自行停止并清理
            _ms_stop_event(task_id, tn_id).set()
            stopped_ms += 1
            continue
        node = _find_node(tn_id)
        if node:
            ok, data = _node_call(node, 'POST', '/stop', {}, timeout=15)
            if not ok:
                failed.append(node.get('name', ''))
    msg_parts = []
    if stopped_ms:
        msg_parts.append('邮局发信已停止 ' + str(stopped_ms) + ' 台')
    if failed:
        msg_parts.append('节点停止失败: ' + ';'.join(failed))
    if not msg_parts:
        msg_parts.append('已停止')
    return not failed, '; '.join(msg_parts)


def task_pull_results(task_id):
    """增量拉取各节点结果, 存入 SQLite results 表
    返回 {task_id, rows_pulled, per_node: [...]}
    """
    _ensure_dirs()
    task = _get_task(task_id)
    if not task:
        return None
    import csv as _csv
    rows_pulled = 0
    per_node = []
    all_done = True
    for tn in task.get('nodes') or []:
        node = _find_node(tn.get('node_id'))
        if not node:
            continue
        offset = master_db.task_cursor_get(task_id, node['id'])
        ok, data = _node_call(node, 'GET',
                              '/result?offset=' + str(offset) + '&limit=10000', timeout=60)
        info = {'node_id': node['id'], 'node_name': node.get('name', ''), 'pulled': 0, 'online': False}
        if ok and data.get('ok'):
            info['online'] = True
            rows = data.get('rows') or []
            batch = []
            for line in rows:
                try:
                    p = next(_csv.reader([line]))
                    if len(p) >= 6:
                        batch.append((task_id, node['id'], p[0], p[1], p[2], p[3], p[4], p[5]))
                except Exception:
                    continue
            if batch:
                master_db.result_insert_many(batch)
            master_db.task_cursor_set(task_id, node['id'], int(data.get('next_offset', offset + len(rows))))
            info['pulled'] = len(batch)
            rows_pulled += len(batch)
            node_state = (data.get('node_state') or {}).get('status', '')
            if not node_state:
                # /result 接口不带 node_state, 补查 /status 拿节点真实状态
                try:
                    _ok2, _st_data = _node_call(node, 'GET', '/status', timeout=15)
                    if _ok2 and isinstance(_st_data, dict) and _st_data.get('ok'):
                        node_state = (_st_data.get('node_state') or {}).get('status', '')
                except Exception:
                    pass
            if node_state not in ('finished', 'stopped'):
                all_done = False
            else:
                try:
                    master_db._exec('UPDATE task_nodes SET push_status=? WHERE task_id=? AND node_id=?',
                                    (node_state, task_id, node['id']))
                    master_db._commit()
                except Exception:
                    pass
        else:
            all_done = False
        per_node.append(info)
    # 全部节点完成 → 任务状态收尾
    if all_done and task.get('nodes'):
        cur = master_db.task_get(task_id)
        if cur and cur.get('status') != 'finished':
            master_db.task_set_status(task_id, 'finished')
    return {'task_id': task_id, 'rows_pulled': rows_pulled, 'per_node': per_node}


def _get_task(task_id):
    return master_db.task_get(task_id)


def task_list():
    return load_tasks()


def task_remove(task_id):
    """删除任务: 先停止全部 worker/节点, 再清任务记录+结果"""
    task = _get_task(task_id)
    if not task:
        with _lock:
            master_db.task_remove(task_id)
            master_db.result_clear(task_id)
        return True
    # 1) 停止邮局 worker (置停止信号)
    for tn in task.get('nodes') or []:
        tn_id = tn.get('node_id') or ''
        if str(tn_id).startswith('ms'):
            _ms_stop_event(task_id, tn_id).set()
            continue
        node = _find_node(tn_id)
        if node:
            try:
                _node_call(node, 'POST', '/stop', {}, timeout=15)
            except Exception:
                pass
    # 2) 清记录+结果
    with _lock:
        master_db.task_remove(task_id)
        master_db.result_clear(task_id)
        _MS_STOP_EVENTS.pop(task_id, None)
    return True