"""
发送节点服务 — 主控远程控制的直投节点
========================================
主控把任务(内容+收件人+代理池+发件箱)推送到节点后, 节点本地持久化,
自主执行; 主控关机/断网都不影响节点继续发, 节点重启自动断点续发。

通信: HTTP + JSON, 端口默认 8898, X-Token 头鉴权

端点:
  GET  /ping                     存活探测
  GET  /status                   任务+引擎状态
  POST /task                     下发任务包(不含收件人也可)
  POST /recipients               追加收件人 (大列表分块推)
  POST /start                    启动直投引擎
  POST /stop                     停止引擎
  GET  /result?offset=&limit=    增量拉取发送结果
  POST /clear                    清空任务/结果/缓存

本地文件 (node_data/):
  task_current.json   当前任务包
  recipients.txt      收件人列表
  state.json          节点状态 (idle/assigned/running/finished/stopped)
  send_result.csv     发送结果 (web_send 写)
  send_cache.json     已发缓存 (断点续发)

启动:  python node_server.py   (或主控通过 systemd/计划任务拉起)
"""
import os
import sys
import json
import time
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
_PROJECT_ROOT = os.path.dirname(_THIS_DIR)
if _PROJECT_ROOT not in sys.path:
    sys.path.insert(0, _PROJECT_ROOT)
if _THIS_DIR not in sys.path:
    sys.path.insert(0, _THIS_DIR)

import web_send

NODE_CONFIG_FILE = os.path.join(_THIS_DIR, 'node_config.json')
NODE_DATA_DIR = os.path.join(_THIS_DIR, 'node_data')
TASK_FILE = os.path.join(NODE_DATA_DIR, 'task_current.json')
RECIPIENTS_FILE = os.path.join(NODE_DATA_DIR, 'recipients.txt')
STATE_FILE = os.path.join(NODE_DATA_DIR, 'state.json')
RESULT_CSV = os.path.join(_THIS_DIR, 'send_result.csv')

_lock = threading.RLock()
_started_at = time.time()


def _ensure_dirs():
    if not os.path.isdir(NODE_DATA_DIR):
        try:
            os.makedirs(NODE_DATA_DIR, exist_ok=True)
        except Exception:
            pass


def load_node_config():
    cfg = {'port': 8898, 'token': 'node-token-change-me'}
    if os.path.exists(NODE_CONFIG_FILE):
        try:
            with open(NODE_CONFIG_FILE, 'r', encoding='utf-8-sig') as f:
                loaded = json.load(f)
            cfg.update(loaded or {})
        except Exception:
            pass
    return cfg


def _read_json(path, default):
    try:
        if os.path.exists(path):
            with open(path, 'r', encoding='utf-8') 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


def _load_task():
    return _read_json(TASK_FILE, None)


def _save_task(task):
    return _write_json(TASK_FILE, task)


def _load_state():
    return _read_json(STATE_FILE, {'status': 'idle'})


def _save_state(st):
    return _write_json(STATE_FILE, st)


def _on_engine_finish(status, snapshot):
    """引擎跑完/被停止后的回调: 落盘节点状态"""
    try:
        with _lock:
            st = _load_state()
            st['status'] = status
            st['finished_at'] = time.strftime('%Y-%m-%d %H:%M:%S')
            st['final_stats'] = {k: snapshot.get(k) for k in
                                 ('total', 'processed', 'success', 'fail',
                                  'bounced', 'ip_rejected', 'switches', 'elapsed', 'speed')}
            _save_state(st)
    except Exception:
        pass


def _load_dkim_keys(boxes):
    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.email if hasattr(b, 'email') else ''
        if '@' not in email:
            continue
        domain = email.split('@')[-1].lower()
        sel = '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 build_cfg_from_task(task):
    from core.models import MailBox, ProxyIP
    content = task.get('content') or {}
    send_cfg = task.get('send_cfg') or {}
    proxies = []
    for p in task.get('proxies') or []:
        proxies.append(ProxyIP(
            ip=p.get('host', ''), port=int(p.get('port') or 0),
            proxy_type=p.get('proxy_type', 'socks5'),
            username=p.get('user', ''), pwd=p.get('pwd', ''),
            enable=True,
        ))
    boxes = []
    for b in task.get('boxes') or []:
        boxes.append(MailBox(
            enable=True, email=b.get('email', ''), user=b.get('email', ''),
            pwd=b.get('pwd', ''), sender_name=b.get('sender_name', ''),
            random_user=bool(b.get('random_user', False)),
        ))
    return {
        'emails_file': RECIPIENTS_FILE,
        'subjects': content.get('subjects') or ['お知らせ'],
        'html_parts': content.get('html_parts') or [''],
        'text_parts': content.get('text_parts') or [],
        'boxes': boxes,
        'proxies': proxies,
        'thread_count': int(send_cfg.get('thread_count') or 50),
        'send_interval': float(send_cfg.get('send_interval') or 0),
        'send_timeout': int(send_cfg.get('send_timeout') or 60),
        'helo_domain': send_cfg.get('helo_domain') or '',
        'reply_to': send_cfg.get('reply_to') or '',
        'custom_headers': send_cfg.get('custom_headers') or {},
        'magic_config': {'enabled': False},
        'dkim_keys': task.get('dkim_keys') or _load_dkim_keys(boxes),
        'use_cache': True,
        'ip_reject_threshold': int(send_cfg.get('ip_reject_threshold') or 1),
        'dns_servers': send_cfg.get('dns_servers') or None,
        'use_starttls': bool(send_cfg.get('use_starttls', True)),
        'disguise_enabled': bool(send_cfg.get('disguise_enabled', True)),
        'disguise_mode': send_cfg.get('disguise_mode', 'mua'),
        'disguise_opts': send_cfg.get('disguise_opts') or None,
        'dkim_extended': bool(send_cfg.get('dkim_extended', True)),
        'dkim_header_mode': send_cfg.get('dkim_header_mode') or None,
        'ip_proactive_threshold': int(send_cfg.get('ip_proactive_threshold') or 30),
        'mx_rate_per_minute': int(send_cfg.get('mx_rate_per_minute') or 0),
        'domain_rate_per_minute': int(send_cfg.get('domain_rate_per_minute') or 0),
        'smtp_pace_mode': send_cfg.get('smtp_pace_mode', 'medium'),
        'schedule_pause': bool(send_cfg.get('schedule_pause', False)),
        'schedule_pause_start': int(send_cfg.get('schedule_pause_start') or 0),
        'schedule_pause_end': int(send_cfg.get('schedule_pause_end') or 6),
        'verp_enabled': False,
        'warmup_enabled': False,
        'utls_forwarder': {'host': '127.0.0.1', 'port': 2526,
                           'enabled': bool(send_cfg.get('utls_forwarder_enabled', True))},
        'avoid_spam_words': bool(send_cfg.get('avoid_spam_words', False)),
        'mail_charset': send_cfg.get('mail_charset') or 'utf-8',
        'smart_profile': send_cfg.get('smart_profile') or {'enabled': False},
        'run_async': True,
        'finish_callback': _on_engine_finish,
    }


def start_engine_from_task():
    """从本地任务文件启动直投引擎 (节点重启自动恢复也走这里)"""
    task = _load_task()
    if not task:
        return False, '节点尚未收到任务 (请先在顶部「📁数据上传」弹窗勾选名单和节点并点「确认分配到节点」)'
    cfg = build_cfg_from_task(task)
    ok = web_send.start_send(cfg)
    if ok:
        with _lock:
            st = _load_state()
            st['status'] = 'running'
            st['task_id'] = task.get('task_id', '')
            st['started_at'] = time.strftime('%Y-%m-%d %H:%M:%S')
            st.pop('finished_at', None)
            st.pop('final_stats', None)
            _save_state(st)
        return True, 'ok'
    # 失败: 透出真实中文原因 + 最近日志
    snap = web_send.snapshot()
    err = snap.get('error_msg') or ''
    logs = (snap.get('log_lines') or [])[-3:]
    detail = err or ''
    for l in logs:
        if 'ERROR' in l or '失败' in l or '空' in l or '未配置' in l:
            detail = (detail + ' | ' if detail else '') + l
    if not detail and logs:
        detail = logs[-1]
    return False, detail or '引擎启动失败 (原因不明, 请查看节点日志)'


def auto_resume():
    """节点启动时: 上次是 running → 自动恢复 (断点续发跳过已发)"""
    st = _load_state()
    if st.get('status') == 'running':
        print('[NODE] 检测到未完成任务, 自动恢复...')
        ok, msg = start_engine_from_task()
        print('[NODE] 恢复结果: ' + msg)


def _test_proxy(p):
    """测试代理可用性 + 检测出口IP国家 (ipinfo.io)
    返回 (ok, msg) — msg 包含出口IP和所属国家
    """
    host = str(p.get('host', '') or '').strip()
    port = int(p.get('port') or 0)
    if not host or not port:
        return False, '代理参数不全'
    try:
        import socks as _socks
        s = _socks.socksocket()
        s.settimeout(20)
        ptype = str(p.get('proxy_type', 'socks5') or 'socks5')
        uname = p.get('user') or None
        pwd = p.get('pwd') or None
        if ptype in ('socks5', 'socks'):
            s.set_proxy(_socks.SOCKS5, host, port, username=uname, password=pwd)
        elif ptype == 'http':
            s.set_proxy(_socks.HTTP, host, port, username=uname, password=pwd)
        else:
            return False, '不支持的代理类型: ' + ptype
        s.connect(('ipinfo.io', 80))
        s.sendall(b'GET /json HTTP/1.1\r\nHost: ipinfo.io\r\nConnection: close\r\n\r\n')
        resp = b''
        while True:
            c = s.recv(4096)
            if not c:
                break
            resp += c
        s.close()
        body = resp.split(b'\r\n\r\n', 1)[-1]
        import json as _json
        info = _json.loads(body.decode('utf-8', 'replace'))
        exit_ip = info.get('ip', '')
        country = info.get('country', '')
        flag = {'JP': '🇯🇵', 'US': '🇺🇸', 'BR': '🇧🇷', 'SG': '🇸🇬', 'HK': '🇭🇰'}.get(country, '🌍')
        jp_hint = '' if country == 'JP' else ' (⚠️非日本IP, 日系邮局基本判垃圾箱, 建议换日本地区代理池)'
        return True, '代理可用 出口IP: ' + exit_ip + ' ' + flag + country + jp_hint
    except Exception as e:
        return False, str(e)[:120]


def _append_recipients(text):
    n = 0
    try:
        with open(RECIPIENTS_FILE, 'a', encoding='utf-8') as f:
            for line in (text or '').splitlines():
                e = line.strip()
                if not e or '@' not in e:
                    continue
                f.write(e.lower() + '\n')
                n += 1
    except Exception:
        pass
    # 维护缓存计数 (状态查询不再扫文件, 低CPU)
    if n:
        with _lock:
            st = _load_state()
            st['recipients_total'] = int(st.get('recipients_total') or 0) + n
            _save_state(st)
    return n


def _recipients_total_cached():
    """收件人总数: 状态缓存 > 引擎快照 > 一次性文件计数"""
    with _lock:
        st = _load_state()
        if st.get('recipients_total'):
            return int(st['recipients_total'])
    snap = web_send.snapshot()
    if snap.get('total'):
        return int(snap['total'])
    n = _count_file_lines(RECIPIENTS_FILE)
    with _lock:
        st = _load_state()
        st['recipients_total'] = n
        _save_state(st)
    return n


class NodeHandler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def _auth_ok(self):
        cfg = load_node_config()
        token = self.headers.get('X-Token', '')
        if not token:
            qs = parse_qs(urlparse(self.path).query)
            token = (qs.get('token') or [''])[0]
        return token == str(cfg.get('token', ''))

    def _send_json(self, code, obj):
        body = json.dumps(obj, ensure_ascii=False).encode('utf-8')
        self.send_response(code)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.send_header('Content-Length', str(len(body)))
        self.send_header('Access-Control-Allow-Origin', '*')
        self.end_headers()
        try:
            self.wfile.write(body)
        except Exception:
            pass

    def _read_body(self):
        length = int(self.headers.get('Content-Length') or 0)
        if length <= 0:
            return b''
        return self.rfile.read(length)

    def _read_json(self):
        try:
            return json.loads(self._read_body().decode('utf-8', 'replace'))
        except Exception:
            return None

    def do_OPTIONS(self):
        self._send_json(204, {})

    def do_GET(self):
        if not self._auth_ok():
            self._send_json(401, {'ok': False, 'msg': 'bad token'})
            return
        parsed = urlparse(self.path)
        qs = parse_qs(parsed.query)
        if parsed.path == '/ping':
            self._send_json(200, {'ok': True, 'version': '2.4.4-node', 'uptime': int(time.time() - _started_at)})
        elif parsed.path == '/status':
            with _lock:
                task = _load_task()
                st = _load_state()
                snap = web_send.snapshot()
            total = _recipients_total_cached()
            processed = int(snap.get('processed') or 0)
            self._send_json(200, {
                'ok': True,
                'node_state': st,
                'task_id': (task or {}).get('task_id', ''),
                'engine': {k: snap.get(k) for k in
                           ('status', 'total', 'processed', 'success', 'fail',
                            'bounced', 'ip_rejected', 'switches', 'elapsed', 'speed',
                            'thread_count', 'proxies_count', 'boxes_count')},
                'recipients_total': total,
                'pending': max(0, total - processed),
                'result_count': processed,
                'logs': (snap.get('log_lines') or [])[-60:],
            })
        elif parsed.path == '/result':
            offset = int((qs.get('offset') or ['0'])[0])
            limit = min(int((qs.get('limit') or ['1000'])[0]), 10000)
            rows, total = _read_result_slice(offset, limit)
            self._send_json(200, {'ok': True, 'rows': rows, 'total': total, 'next_offset': offset + len(rows)})
        else:
            self._send_json(404, {'ok': False, 'msg': 'not found'})

    def do_POST(self):
        if not self._auth_ok():
            self._send_json(401, {'ok': False, 'msg': 'bad token'})
            return
        parsed = urlparse(self.path)
        data = self._read_json()
        if parsed.path == '/task':
            if not data or not data.get('task_id'):
                self._send_json(400, {'ok': False, 'msg': 'task_id required'})
                return
            with _lock:
                _ensure_dirs()
                task = {
                    'task_id': str(data['task_id']),
                    'content': data.get('content') or {},
                    'send_cfg': data.get('send_cfg') or {},
                    'proxies': data.get('proxies') or [],
                    'boxes': data.get('boxes') or [],
                    'dkim_keys': data.get('dkim_keys') or {},
                    'assigned_at': time.strftime('%Y-%m-%d %H:%M:%S'),
                }
                _save_task(task)
                # DKIM 密钥同时落盘到本地 dkim_keys 目录 (节点重启/后续任务复用)
                try:
                    dkim_dir = os.path.join(_THIS_DIR, 'dkim_keys')
                    os.makedirs(dkim_dir, exist_ok=True)
                    for dom, info in (data.get('dkim_keys') or {}).items():
                        sel = (info.get('selector') if isinstance(info, dict) else '') or 'default'
                        pk = (info.get('private_key') if isinstance(info, dict) else '') or ''
                        if dom and pk:
                            with open(os.path.join(dkim_dir, str(dom) + '_' + sel + '.pem'),
                                      'w', encoding='utf-8') as f:
                                f.write(pk)
                except Exception:
                    pass
                # 收件人: 直接内嵌或清空重收
                if os.path.exists(RECIPIENTS_FILE):
                    try:
                        os.remove(RECIPIENTS_FILE)
                    except Exception:
                        pass
                n = 0
                if data.get('recipients_text'):
                    n = _append_recipients(data['recipients_text'])
                st = _load_state()
                st['status'] = 'assigned'
                st['task_id'] = str(data['task_id'])
                st['recipients_total'] = n  # 重置缓存计数
                _save_state(st)
                # ★ 新任务 = 全新发送: 清空已发缓存 (换名单不跳过; 同一任务内重启仍断点续发)
                for _f in (web_send.SENT_CACHE_FILE, web_send.SENT_CACHE_FILE + '-wal',
                           web_send.SENT_CACHE_FILE + '-shm'):
                    if os.path.exists(_f):
                        try:
                            os.remove(_f)
                        except Exception:
                            pass
            self._send_json(200, {'ok': True, 'msg': 'task assigned', 'recipients': n})
        elif parsed.path == '/recipients':
            if not data or not data.get('task_id'):
                self._send_json(400, {'ok': False, 'msg': 'task_id required'})
                return
            with _lock:
                n = _append_recipients(data.get('recipients_text') or '')
            self._send_json(200, {'ok': True, 'recipients_appended': n})
        elif parsed.path == '/content':
            # 两阶段流程: 任务已分配 → 主控后补内容 (节点仍处 assigned, 收到内容后等 /start)
            if not data or not data.get('task_id'):
                self._send_json(400, {'ok': False, 'msg': 'task_id required'})
                return
            with _lock:
                task = _load_task()
                if not task or task.get('task_id') != str(data['task_id']):
                    self._send_json(400, {'ok': False, 'msg': '节点未收到该任务 (请先在主控点「分配配置」)'})
                    return
                if data.get('content'):
                    task['content'] = data['content']
                    _save_task(task)
            self._send_json(200, {'ok': True, 'msg': 'content updated'})
        elif parsed.path == '/start':
            # 支持携带内容一起启动 (主控一键: 内容+启动)
            with _lock:
                if data and data.get('content'):
                    task = _load_task()
                    if task and task.get('task_id') == str(data.get('task_id') or task.get('task_id')):
                        task['content'] = data['content']
                        _save_task(task)
                ok, msg = start_engine_from_task()
            self._send_json(200, {'ok': ok, 'msg': msg})
        elif parsed.path == '/stop':
            web_send.stop_send()
            with _lock:
                st = _load_state()
                st['status'] = 'stopped'
                _save_state(st)
            self._send_json(200, {'ok': True})
        elif parsed.path == '/test_proxy':
            p = (data or {}).get('proxy') or {}
            ok, msg = _test_proxy(p)
            self._send_json(200, {'ok': ok, 'msg': msg})
        elif parsed.path == '/send_test':
            # 进箱测试: 复用当前任务配置(代理/魔改/DKIM)单封同步发送
            to_addr = ((data or {}).get('to') or '').strip()
            subject = (data or {}).get('subject') or 'Inbox Test'
            html = (data or {}).get('html') or '<html><body><p>Inbox test</p></body></html>'
            from_email = ((data or {}).get('from_email') or '').strip()
            from_name = ((data or {}).get('from_name') or '').strip()
            if '@' not in to_addr:
                self._send_json(400, {'ok': False, 'msg': '收件地址无效'})
                return
            with _lock:
                task = _load_task()
            ok, msg, code = web_send.send_single_test(to_addr, subject, html,
                                                       task, from_email, from_name)
            self._send_json(200, {'ok': ok, 'msg': msg, 'code': str(code)})
        elif parsed.path == '/set_boxes':
            # 主控下发发件域名(发件箱+DKIM密钥)到节点任务
            with _lock:
                task = _load_task()
                if not task:
                    self._send_json(400, {'ok': False, 'msg': '节点尚未分配数据 (请先在「数据上传」弹窗分配名单和节点)'})
                    return
                task['boxes'] = data.get('boxes') or []
                if data.get('dkim_keys'):
                    task['dkim_keys'] = data.get('dkim_keys')
                    try:
                        dkim_dir = os.path.join(_THIS_DIR, 'dkim_keys')
                        os.makedirs(dkim_dir, exist_ok=True)
                        for dom, info in data['dkim_keys'].items():
                            sel = (info.get('selector') if isinstance(info, dict) else '') or 'default'
                            pk = (info.get('private_key') if isinstance(info, dict) else '') or ''
                            if dom and pk:
                                with open(os.path.join(dkim_dir, str(dom) + '_' + sel + '.pem'),
                                          'w', encoding='utf-8') as f:
                                    f.write(pk)
                    except Exception:
                        pass
                _save_task(task)
            self._send_json(200, {'ok': True, 'msg': '发件域名已更新'})
        elif parsed.path == '/set_content':
            # 主控下发内容(主题/正文)到节点任务
            with _lock:
                task = _load_task()
                if not task:
                    self._send_json(400, {'ok': False, 'msg': '节点尚未分配数据 (请先在「数据上传」弹窗分配名单和节点)'})
                    return
                if data.get('content'):
                    task['content'] = data['content']
                    _save_task(task)
            self._send_json(200, {'ok': True, 'msg': '内容已更新'})
        elif parsed.path == '/set_proxy':
            # 主控下发新代理池到节点任务 (独立分配代理)
            with _lock:
                task = _load_task()
                if task:
                    task['proxies'] = data.get('proxies') or []
                    if isinstance(task.get('send_cfg'), dict):
                        task['send_cfg']['proxies'] = data.get('proxies') or []
                    _save_task(task)
            # ★ 热更新运行中的引擎 (API直连IP过期后主控推送新池, 不停任务立即生效)
            try:
                from core.models import ProxyIP as _ProxyIP
                _plist = []
                for _p in (data.get('proxies') or []):
                    _plist.append(_ProxyIP(
                        ip=str(_p.get('host', '')), port=int(_p.get('port') or 0),
                        proxy_type=_p.get('proxy_type', 'socks5'),
                        username=_p.get('user', ''), pwd=_p.get('pwd', ''), enable=True))
                web_send.update_live_proxies(_plist)
            except Exception:
                pass
            self._send_json(200, {'ok': True, 'msg': '代理已更新(含热更新)'})
        elif parsed.path == '/set_config':
            # 主控批量下发魔改配置到节点任务
            # ★ 新配置整体替换旧魔改配置 (旧项自动清除, 不再合并残留)
            with _lock:
                task = _load_task()
                if task:
                    cfg = data.get('send_cfg') or {}
                    MK_KEYS = ('disguise_opts', 'dkim_header_mode', 'smtp_pace_mode',
                               'ip_proactive_threshold', 'mx_rate_per_minute', 'domain_rate_per_minute',
                               'utls_forwarder_enabled', 'avoid_spam_words', 'smart_profile',
                               'extra_headers', 'reply_to_enabled', 'reply_to',
                               'disguise_enabled', 'disguise_mode', 'personalize_subject',
                               'invisible_segments', 'embed_images', 'timestamp_personalization',
                               'mail_charset')
                    old = task.get('send_cfg')
                    if isinstance(old, dict):
                        for k in list(old.keys()):
                            if k in MK_KEYS:
                                old.pop(k, None)
                        old.update(cfg)
                        task['send_cfg'] = old
                    else:
                        task['send_cfg'] = dict(cfg)
                    _save_task(task)
            self._send_json(200, {'ok': True, 'msg': '魔改配置已更新'})
        elif parsed.path == '/clear_proxy':
            # 只清任务包里的代理配置 (数据和任务保留)
            with _lock:
                task = _load_task()
                if task:
                    task['proxies'] = []
                    try:
                        task['send_cfg']['proxies'] = []
                    except Exception:
                        pass
                    _save_task(task)
            self._send_json(200, {'ok': True, 'msg': '代理已清空'})
        elif parsed.path == '/clear':
            web_send.stop_send()
            web_send.reset_state()
            with _lock:
                for f in (TASK_FILE, RECIPIENTS_FILE, STATE_FILE):
                    if os.path.exists(f):
                        try:
                            os.remove(f)
                        except Exception:
                            pass
                # 清结果 + 已发缓存 (含WAL/SHM, 清完节点回到全新状态)
                for f in (web_send.SEND_RESULT_CSV, web_send.SENT_CACHE_FILE,
                          web_send.SENT_CACHE_FILE + '-wal', web_send.SENT_CACHE_FILE + '-shm'):
                    if os.path.exists(f):
                        try:
                            os.remove(f)
                        except Exception:
                            pass
            self._send_json(200, {'ok': True, 'msg': 'cleared'})
        else:
            self._send_json(404, {'ok': False, 'msg': 'not found'})


def _count_file_lines(path):
    n = 0
    try:
        with open(path, 'r', encoding='utf-8') as f:
            for _ in f:
                n += 1
    except Exception:
        pass
    return n


def _read_result_slice(offset, limit):
    rows = []
    total = 0
    try:
        if not os.path.exists(RESULT_CSV):
            return rows, 0
        with open(RESULT_CSV, 'r', encoding='utf-8') as f:
            for i, line in enumerate(f):
                if i == 0:
                    continue
                total += 1
                if i - 1 >= offset and len(rows) < limit:
                    rows.append(line.rstrip('\n'))
    except Exception:
        pass
    return rows, total


def main():
    _ensure_dirs()
    cfg = load_node_config()
    port = int(cfg.get('port', 8898) or 8898)
    print('=' * 60)
    print('发送节点服务 (直投引擎节点)')
    print('端口: ' + str(port) + '  token: ' + str(cfg.get('token', ''))[:8] + '...')
    print('任务文件: ' + TASK_FILE)
    print('=' * 60)
    auto_resume()
    server = ThreadingHTTPServer(('0.0.0.0', port), NodeHandler)
    print('[NODE] 监听 0.0.0.0:' + str(port) + ', 等待主控下发任务...')
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print('[NODE] 已停止')


if __name__ == '__main__':
    main()