"""
master_db.py — 主控 SQLite 数据库
==================================
表:
  nodes       节点列表
  tasks       任务记录 (内容/魔改配置/发件箱 JSON)
  task_nodes  任务-节点分配 + 每节点结果拉取游标
  results     发送结果明细 (千万级, 支持筛选统计)
  uploads     收件人文件上传记录

千万级能力:
  - WAL 模式 + 批量事务插入 (每1000条提交)
  - results 按 (task_id, email) 建索引, 聚合统计秒出
"""
import os
import json
import sqlite3
import threading
import time

_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(_THIS_DIR, 'master_data', 'master.db')

_lock = threading.RLock()
_conn = None

SCHEMA = """
CREATE TABLE IF NOT EXISTS nodes (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL DEFAULT '',
    ip TEXT NOT NULL,
    port INTEGER NOT NULL,
    token TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL DEFAULT '',
    node_type TEXT NOT NULL DEFAULT 'direct',
    smtp_port INTEGER NOT NULL DEFAULT 587,
    smtp_user TEXT NOT NULL DEFAULT '',
    smtp_pwd TEXT NOT NULL DEFAULT '',
    day_limit INTEGER NOT NULL DEFAULT 0,
    minute_limit INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS tasks (
    task_id TEXT PRIMARY KEY,
    name TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL DEFAULT '',
    content_json TEXT NOT NULL DEFAULT '{}',
    send_cfg_json TEXT NOT NULL DEFAULT '{}',
    boxes_json TEXT NOT NULL DEFAULT '[]',
    dkim_json TEXT NOT NULL DEFAULT '{}',
    recipients_total INTEGER NOT NULL DEFAULT 0,
    status TEXT NOT NULL DEFAULT 'running',
    finished_at TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS task_nodes (
    task_id TEXT NOT NULL,
    node_id TEXT NOT NULL,
    node_name TEXT NOT NULL DEFAULT '',
    recipients INTEGER NOT NULL DEFAULT 0,
    push_status TEXT NOT NULL DEFAULT 'assigned',
    result_cursor INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (task_id, node_id)
);
CREATE TABLE IF NOT EXISTS results (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    task_id TEXT NOT NULL,
    node_id TEXT NOT NULL DEFAULT '',
    email TEXT NOT NULL,
    email_domain TEXT NOT NULL DEFAULT '',
    status TEXT NOT NULL DEFAULT '',
    code TEXT NOT NULL DEFAULT '',
    msg TEXT NOT NULL DEFAULT '',
    timestamp TEXT NOT NULL DEFAULT '',
    proxy_id TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_results_task ON results(task_id, id);
CREATE INDEX IF NOT EXISTS idx_results_email ON results(email);
CREATE INDEX IF NOT EXISTS idx_results_domain ON results(email_domain);
CREATE TABLE IF NOT EXISTS result_summary (
    task_id TEXT NOT NULL,
    domain TEXT NOT NULL,
    status TEXT NOT NULL,
    cnt INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (task_id, domain, status)
);
CREATE INDEX IF NOT EXISTS idx_summary_task ON result_summary(task_id);
CREATE TABLE IF NOT EXISTS uploads (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    filename TEXT NOT NULL DEFAULT '',
    path TEXT NOT NULL DEFAULT '',
    count INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS proxy_pool (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    host TEXT NOT NULL,
    port INTEGER NOT NULL,
    proxy_type TEXT NOT NULL DEFAULT 'socks5',
    username TEXT NOT NULL DEFAULT '',
    password TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS mail_servers (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL DEFAULT '',
    ip TEXT NOT NULL,
    ssh_port INTEGER NOT NULL DEFAULT 22,
    ssh_user TEXT NOT NULL DEFAULT 'root',
    ssh_pwd TEXT NOT NULL DEFAULT '',
    hostname TEXT NOT NULL DEFAULT '',
    mail_stack TEXT NOT NULL DEFAULT 'haraka',
    smtp_user TEXT NOT NULL DEFAULT '',
    smtp_pwd TEXT NOT NULL DEFAULT '',
    dkim_public TEXT NOT NULL DEFAULT '',
    dkim_selector TEXT NOT NULL DEFAULT 'default',
    status TEXT NOT NULL DEFAULT '',
    dns_status TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL DEFAULT '',
    smtp_port INTEGER NOT NULL DEFAULT 587,
    day_limit INTEGER NOT NULL DEFAULT 0,
    minute_limit INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS content_library (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL DEFAULT '',
    content_json TEXT NOT NULL DEFAULT '{}',
    created_at TEXT NOT NULL DEFAULT '',
    updated_at TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS mk_library (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL DEFAULT '',
    cfg_json TEXT NOT NULL DEFAULT '{}',
    created_at TEXT NOT NULL DEFAULT '',
    updated_at TEXT NOT NULL DEFAULT ''
);
"""


def _now():
    return time.strftime('%Y-%m-%d %H:%M:%S')


_lock = threading.RLock()
_conn = None
_local = threading.local()


def db():
    global _conn
    with _lock:
        conn = getattr(_local, 'conn', None)
        if conn is None:
            os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
            conn = sqlite3.connect(DB_PATH, check_same_thread=False)
            conn.execute('PRAGMA journal_mode=WAL')
            conn.execute('PRAGMA synchronous=NORMAL')
            conn.execute('PRAGMA busy_timeout=30000')
            conn.execute('PRAGMA cache_size=-64000')
            conn.executescript(SCHEMA)
            _migrate_nodes(conn)
            conn.commit()
            _local.conn = conn
            _conn = conn
        return conn


def _migrate_nodes(conn):
    """旧库升级: nodes 表补邮局节点相关列; mail_servers 表补 smtp_port/限速列"""
    try:
        cols = [r[1] for r in conn.execute('PRAGMA table_info(nodes)')]
        for col, ddl in [
            ('node_type', 'TEXT NOT NULL DEFAULT \'direct\''),
            ('smtp_port', 'INTEGER NOT NULL DEFAULT 587'),
            ('smtp_user', 'TEXT NOT NULL DEFAULT \'\''),
            ('smtp_pwd', 'TEXT NOT NULL DEFAULT \'\''),
            ('day_limit', 'INTEGER NOT NULL DEFAULT 0'),
            ('minute_limit', 'INTEGER NOT NULL DEFAULT 0'),
        ]:
            if col not in cols:
                conn.execute('ALTER TABLE nodes ADD COLUMN ' + col + ' ' + ddl)
    except Exception:
        pass
    try:
        cols = [r[1] for r in conn.execute('PRAGMA table_info(mail_servers)')]
        for col, ddl in [
            ('smtp_port', 'INTEGER NOT NULL DEFAULT 587'),
            ('day_limit', 'INTEGER NOT NULL DEFAULT 0'),
            ('minute_limit', 'INTEGER NOT NULL DEFAULT 0'),
        ]:
            if col not in cols:
                conn.execute('ALTER TABLE mail_servers ADD COLUMN ' + col + ' ' + ddl)
    except Exception:
        pass


def _q(sql, params=()):
    c = db()
    cur = c.execute(sql, params)
    return cur.fetchall()


def _q1(sql, params=()):
    rows = _q(sql, params)
    return rows[0] if rows else None


def _exec(sql, params=()):
    c = db()
    c.execute(sql, params)


def _commit():
    db().commit()


# ============================================================
# nodes
# ============================================================
def node_all():
    rows = _q('SELECT id,name,ip,port,token,created_at FROM nodes ORDER BY created_at')
    return [{'id': r[0], 'name': r[1], 'ip': r[2], 'port': r[3],
             'token': r[4], 'created_at': r[5]} for r in rows]


def node_add(node):
    # ★ 按 ip+port 去重: 同一台服务器重复部署/添加时复用已有id, 不产生重复记录
    try:
        _old = _q1('SELECT id FROM nodes WHERE ip=? AND port=?',
                   (node['ip'], int(node.get('port', 8898))))
        if _old:
            node['id'] = _old[0]
    except Exception:
        pass
    _exec('INSERT OR REPLACE INTO nodes(id,name,ip,port,token,created_at) VALUES(?,?,?,?,?,?)',
          (node['id'], node.get('name', ''), node['ip'], int(node.get('port', 8898)),
           node.get('token', ''), node.get('created_at', _now())))
    _commit()
    return node


def node_remove(node_id):
    _exec('DELETE FROM nodes WHERE id=?', (node_id,))
    _exec('DELETE FROM task_nodes WHERE node_id=?', (node_id,))
    _commit()


def node_get(node_id):
    r = _q1('SELECT id,name,ip,port,token,created_at FROM nodes WHERE id=?', (node_id,))
    if not r:
        return None
    return {'id': r[0], 'name': r[1], 'ip': r[2], 'port': r[3],
            'token': r[4], 'created_at': r[5]}


# ============================================================
# tasks
# ============================================================
def task_insert(task):
    _exec('INSERT OR REPLACE INTO tasks(task_id,name,created_at,content_json,send_cfg_json,'
          'boxes_json,dkim_json,recipients_total,status,finished_at) VALUES(?,?,?,?,?,?,?,?,?,?)',
          (task['task_id'], task.get('name', ''), task.get('created_at', _now()),
           json.dumps(task.get('content') or {}, ensure_ascii=False),
           json.dumps(task.get('send_cfg') or {}, ensure_ascii=False),
           json.dumps(task.get('boxes') or [], ensure_ascii=False),
           json.dumps(task.get('dkim_keys') or {}, ensure_ascii=False),
           int(task.get('recipients_total') or 0),
           task.get('status', 'running'), task.get('finished_at', '')))
    for tn in task.get('nodes') or []:
        _exec('INSERT OR REPLACE INTO task_nodes(task_id,node_id,node_name,recipients,push_status,result_cursor) '
              'VALUES(?,?,?,?,?,?)',
              (task['task_id'], tn['node_id'], tn.get('node_name', ''),
               int(tn.get('recipients') or 0), tn.get('push_status', 'assigned'),
               int(tn.get('result_cursor') or 0)))
    _commit()


def task_get(task_id):
    r = _q1('SELECT task_id,name,created_at,content_json,send_cfg_json,boxes_json,dkim_json,'
            'recipients_total,status,finished_at FROM tasks WHERE task_id=?', (task_id,))
    if not r:
        return None
    task = {
        'task_id': r[0], 'name': r[1], 'created_at': r[2],
        'content': json.loads(r[3] or '{}'), 'send_cfg': json.loads(r[4] or '{}'),
        'boxes': json.loads(r[5] or '[]'), 'dkim_keys': json.loads(r[6] or '{}'),
        'recipients_total': r[7], 'status': r[8], 'finished_at': r[9],
        'nodes': [],
    }
    for nr in _q('SELECT node_id,node_name,recipients,push_status,result_cursor FROM task_nodes WHERE task_id=?',
                 (task_id,)):
        task['nodes'].append({'node_id': nr[0], 'node_name': nr[1], 'recipients': nr[2],
                              'push_status': nr[3], 'result_cursor': nr[4]})
    return task


def task_all():
    rows = _q('SELECT task_id,name,created_at,recipients_total,status FROM tasks ORDER BY created_at DESC')
    # 一次性取各任务的节点类型标记 (node_id 以 ms 开头=邮局, 其余=直投节点)
    ms_flags = {}
    for r in _q('SELECT task_id, node_id FROM task_nodes'):
        tid, nid = r[0], str(r[1] or '')
        fl = ms_flags.setdefault(tid, {'has_ms': False, 'has_direct': False})
        if nid.startswith('ms'):
            fl['has_ms'] = True
        else:
            fl['has_direct'] = True
    out = []
    for r in rows:
        st = _q1('SELECT SUM(CASE WHEN status="sent" THEN 1 ELSE 0 END), COUNT(*) FROM results WHERE task_id=?', (r[0],))
        fl = ms_flags.get(r[0], {'has_ms': False, 'has_direct': False})
        out.append({'task_id': r[0], 'name': r[1], 'created_at': r[2],
                    'recipients_total': r[3], 'status': r[4],
                    'processed': int(st[1] or 0), 'success': int(st[0] or 0),
                    'fail': int(st[1] or 0) - int(st[0] or 0),
                    'node_count': _q1('SELECT COUNT(*) FROM task_nodes WHERE task_id=?', (r[0],))[0],
                    'has_ms': fl['has_ms'], 'has_direct': fl['has_direct']})
    return out


def task_remove(task_id):
    _exec('DELETE FROM tasks WHERE task_id=?', (task_id,))
    _exec('DELETE FROM task_nodes WHERE task_id=?', (task_id,))
    _commit()


def task_set_status(task_id, status):
    _exec("UPDATE tasks SET status=?, finished_at=? WHERE task_id=?",
          (status, _now() if status in ('finished', 'stopped') else '', task_id))
    _commit()


def task_update_content(task_id, content):
    _exec('UPDATE tasks SET content_json=? WHERE task_id=?',
          (json.dumps(content or {}, ensure_ascii=False), task_id))
    _commit()


def task_update_send_cfg(task_id, send_cfg):
    _exec('UPDATE tasks SET send_cfg_json=? WHERE task_id=?',
          (json.dumps(send_cfg or {}, ensure_ascii=False), task_id))
    _commit()


def task_cursor_get(task_id, node_id):
    r = _q1('SELECT result_cursor FROM task_nodes WHERE task_id=? AND node_id=?', (task_id, node_id))
    return int(r[0]) if r else 0


def task_cursor_set(task_id, node_id, cursor):
    _exec('UPDATE task_nodes SET result_cursor=? WHERE task_id=? AND node_id=?', (int(cursor), task_id, node_id))
    _commit()


def task_node_status_set(task_id, node_id, push_status):
    _exec('UPDATE task_nodes SET push_status=? WHERE task_id=? AND node_id=?',
          (push_status, task_id, node_id))
    _commit()


def tasks_by_node(node_id):
    """含指定目标(节点/邮局)的任务ID列表 (按创建时间排序)"""
    rows = _q('SELECT DISTINCT t.task_id FROM task_nodes tn JOIN tasks t ON t.task_id=tn.task_id '
              'WHERE tn.node_id=? ORDER BY t.created_at', (node_id,))
    return [r[0] for r in rows]


def result_stats_by_node(task_id, node_id):
    """按节点统计结果: {sent, fail, total} (邮局发信worker实时状态用)"""
    r = _q1('SELECT COUNT(*) FROM results WHERE task_id=? AND node_id=?', (task_id, node_id))
    total = int(r[0]) if r else 0
    r = _q1('SELECT COUNT(*) FROM results WHERE task_id=? AND node_id=? AND status="sent"',
            (task_id, node_id))
    sent = int(r[0]) if r else 0
    return {'sent': sent, 'fail': total - sent, 'total': total}


# ============================================================
# results (亿级: executemany批量写 + 预聚合摘要表 + 键集分页)
# ============================================================
def result_insert_many(rows):
    """批量插入结果 (executemany + 单事务提交 + 同步维护摘要计数)
    rows: [(task_id, node_id, email, status, code, msg, timestamp, proxy_id), ...]
    亿级吞吐: 单批建议1万-5万条, 实测百万条/秒级
    """
    from collections import Counter
    if not rows:
        return 0
    c = db()
    prepared = []
    summary = Counter()
    for r in rows:
        email = (r[2] or '').lower()
        dom = email.split('@')[-1] if '@' in email else ''
        st = r[3] or ''
        prepared.append((r[0], r[1], email, dom, st, str(r[4] or ''),
                         (r[5] or '')[:120], r[6] or '', r[7] or ''))
        summary[(r[0], dom, st)] += 1
    c.executemany('INSERT INTO results(task_id,node_id,email,email_domain,status,code,msg,timestamp,proxy_id) '
                  'VALUES(?,?,?,?,?,?,?,?,?)', prepared)
    c.executemany(
        'INSERT INTO result_summary(task_id,domain,status,cnt) VALUES(?,?,?,?) '
        'ON CONFLICT(task_id,domain,status) DO UPDATE SET cnt=cnt+excluded.cnt',
        [(k[0], k[1], k[2], v) for k, v in summary.items()])
    c.commit()
    return len(prepared)


def _summary_empty(task_id=None):
    where = ' WHERE task_id=?' if task_id else ''
    params = (task_id,) if task_id else ()
    return _q1('SELECT COALESCE(SUM(cnt),0) FROM result_summary' + where, params)[0] == 0


def _rebuild_summary(task_id=None):
    """旧数据迁移: 从 results 全表重建摘要 (一次性)"""
    c = db()
    if task_id:
        c.execute('DELETE FROM result_summary WHERE task_id=?', (task_id,))
        c.execute('INSERT INTO result_summary(task_id,domain,status,cnt) '
                  'SELECT task_id, email_domain, status, COUNT(*) FROM results WHERE task_id=? '
                  'GROUP BY task_id, email_domain, status', (task_id,))
    else:
        c.execute('DELETE FROM result_summary')
        c.execute('INSERT INTO result_summary(task_id,domain,status,cnt) '
                  'SELECT task_id, email_domain, status, COUNT(*) FROM results '
                  'GROUP BY task_id, email_domain, status')
    c.commit()


def result_query(task_id=None, status=None, domain=None, before_id=None, limit=100):
    """键集分页 (ORDER BY id DESC, id < before_id) — 亿级下深分页 O(log n)"""
    sql = ('SELECT id,task_id,node_id,email,status,code,msg,timestamp,proxy_id '
           'FROM results WHERE 1=1')
    params = []
    if task_id:
        sql += ' AND task_id=?'
        params.append(task_id)
    if status:
        sql += ' AND status=?'
        params.append(status)
    if domain:
        sql += ' AND email_domain=?'
        params.append(domain.lower())
    if before_id:
        sql += ' AND id<?'
        params.append(int(before_id))
    sql += ' ORDER BY id DESC LIMIT ?'
    params.append(max(1, min(int(limit), 10000)))
    rows = _q(sql, params)
    return [{'id': r[0], 'task_id': r[1], 'node_id': r[2], 'email': r[3], 'status': r[4],
             'code': r[5], 'msg': r[6], 'timestamp': r[7], 'proxy_id': r[8]} for r in rows]


def result_stats(task_id=None):
    """任务聚合统计 (读预聚合摘要, O(1); 无摘要时自动重建)"""
    if _summary_empty(task_id):
        _rebuild_summary(task_id)
    where = ' WHERE task_id=?' if task_id else ''
    params = (task_id,) if task_id else ()
    total = _q1('SELECT COALESCE(SUM(cnt),0) FROM result_summary' + where, params)[0]
    by_status = {r[0]: r[1] for r in
                 _q('SELECT status, SUM(cnt) FROM result_summary' + where + ' GROUP BY status', params)}
    by_domain = _q('SELECT domain, SUM(cnt), SUM(CASE WHEN status="sent" THEN cnt ELSE 0 END) '
                   'FROM result_summary' + where +
                   ' GROUP BY domain ORDER BY SUM(cnt) DESC LIMIT 30', params)
    sent = by_status.get('sent', 0)
    return {
        'total': total,
        'sent': sent,
        'failed': total - sent,
        'sent_rate': round(sent / total * 100, 2) if total else 0,
        'by_status': by_status,
        'by_domain': [{'domain': d[0], 'total': d[1], 'sent': d[2] or 0} for d in by_domain],
    }


def result_clear(task_id=None):
    if task_id:
        _exec('DELETE FROM results WHERE task_id=?', (task_id,))
        _exec('DELETE FROM result_summary WHERE task_id=?', (task_id,))
    else:
        _exec('DELETE FROM results')
        _exec('DELETE FROM result_summary')
    _commit()


# ============================================================
# uploads
# ============================================================
def upload_add(filename, path, count):
    _exec('INSERT INTO uploads(filename,path,count,created_at) VALUES(?,?,?,?)',
          (filename, path, int(count or 0), _now()))
    _commit()


def upload_all():
    return [{'id': r[0], 'filename': r[1], 'path': r[2], 'count': r[3], 'created_at': r[4]}
            for r in _q('SELECT id,filename,path,count,created_at FROM uploads ORDER BY id DESC')]


# ============================================================
# mail_servers (部署邮局)
# ============================================================
def mail_server_add(ms):
    _exec('INSERT OR REPLACE INTO mail_servers(id,name,ip,ssh_port,ssh_user,ssh_pwd,hostname,'
          'mail_stack,smtp_user,smtp_pwd,dkim_public,dkim_selector,status,dns_status,created_at,'
          'smtp_port,day_limit,minute_limit) '
          'VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)',
          (ms['id'], ms.get('name', ''), ms['ip'], int(ms.get('ssh_port', 22)),
           ms.get('ssh_user', 'root'), ms.get('ssh_pwd', ''), ms.get('hostname', ''),
           ms.get('mail_stack', 'haraka'), ms.get('smtp_user', ''), ms.get('smtp_pwd', ''),
           ms.get('dkim_public', ''), ms.get('dkim_selector', 'default'),
           ms.get('status', ''), ms.get('dns_status', ''), ms.get('created_at', _now()),
           int(ms.get('smtp_port') or 0), int(ms.get('day_limit') or 0), int(ms.get('minute_limit') or 0)))
    _commit()


def mail_server_update(ms_id, **kwargs):
    cols = [k for k in kwargs if k in ('name', 'ip', 'ssh_port', 'ssh_user', 'ssh_pwd', 'hostname',
                                       'mail_stack', 'smtp_user', 'smtp_pwd', 'dkim_public',
                                       'dkim_selector', 'status', 'dns_status',
                                       'smtp_port', 'day_limit', 'minute_limit')]
    if not cols:
        return
    sets = ', '.join(c + '=?' for c in cols)
    _exec('UPDATE mail_servers SET ' + sets + ' WHERE id=?',
          tuple(kwargs[c] for c in cols) + (ms_id,))
    _commit()


def mail_server_del(ms_id):
    _exec('DELETE FROM mail_servers WHERE id=?', (ms_id,))
    _commit()


def mail_server_all():
    rows = _q('SELECT id,name,ip,ssh_port,ssh_user,ssh_pwd,hostname,mail_stack,smtp_user,smtp_pwd,'
              'dkim_public,dkim_selector,status,dns_status,created_at,smtp_port,day_limit,minute_limit '
              'FROM mail_servers ORDER BY created_at')
    return [{'id': r[0], 'name': r[1], 'ip': r[2], 'ssh_port': r[3], 'ssh_user': r[4],
             'ssh_pwd': r[5], 'hostname': r[6], 'mail_stack': r[7], 'smtp_user': r[8],
             'smtp_pwd': r[9], 'dkim_public': r[10], 'dkim_selector': r[11],
             'status': r[12], 'dns_status': r[13], 'created_at': r[14],
             'smtp_port': r[15], 'day_limit': r[16], 'minute_limit': r[17]} for r in rows]


def mail_server_get(ms_id):
    rows = _q('SELECT id,name,ip,ssh_port,ssh_user,ssh_pwd,hostname,mail_stack,smtp_user,smtp_pwd,'
              'dkim_public,dkim_selector,status,dns_status,created_at,smtp_port,day_limit,minute_limit '
              'FROM mail_servers WHERE id=?', (ms_id,))
    if not rows:
        return None
    r = rows[0]
    return {'id': r[0], 'name': r[1], 'ip': r[2], 'ssh_port': r[3], 'ssh_user': r[4],
            'ssh_pwd': r[5], 'hostname': r[6], 'mail_stack': r[7], 'smtp_user': r[8],
            'smtp_pwd': r[9], 'dkim_public': r[10], 'dkim_selector': r[11],
            'status': r[12], 'dns_status': r[13], 'created_at': r[14],
            'smtp_port': r[15], 'day_limit': r[16], 'minute_limit': r[17]}


# ============================================================
# proxy_pool (主控代理池, 任务缺省下发)
# ============================================================
def proxy_all():
    return [{'id': r[0], 'host': r[1], 'port': r[2], 'proxy_type': r[3],
             'username': r[4], 'password': r[5], 'created_at': r[6]}
            for r in _q('SELECT id,host,port,proxy_type,username,password,created_at '
                        'FROM proxy_pool ORDER BY id')]


def proxy_add(p):
    """★ 共存模式: 追加新代理 (同 host:port 已存在则更新凭据, 不再清空旧代理)
    多条网关 → 任务分配卡多选轮询分发 → 并发分压"""
    host = str(p.get('host', '')).strip()
    port = int(p.get('port') or 0)
    if not host or port <= 0:
        return
    proxy_type = p.get('proxy_type', 'socks5')
    user = p.get('user', '') or p.get('username', '')
    pwd = p.get('pwd', '') or p.get('password', '')
    # 同 host+port 去重: 存在则更新 (类型/账密可能换了)
    rows = _q('SELECT id FROM proxy_pool WHERE host=? AND port=?', (host, port))
    if rows:
        _exec('UPDATE proxy_pool SET proxy_type=?, username=?, password=? WHERE host=? AND port=?',
              (proxy_type, user, pwd, host, port))
    else:
        _exec('INSERT INTO proxy_pool(host,port,proxy_type,username,password,created_at) '
              'VALUES(?,?,?,?,?,?)',
              (host, port, proxy_type, user, pwd, _now()))
    _commit()


def proxy_remove(pid):
    _exec('DELETE FROM proxy_pool WHERE id=?', (int(pid),))
    _commit()


def proxy_clear():
    _exec('DELETE FROM proxy_pool')
    _commit()


def proxy_as_list():
    """转成引擎/节点格式 [{'host','port','proxy_type','user','pwd'}]"""
    return [{'host': p['host'], 'port': p['port'], 'proxy_type': p['proxy_type'],
             'user': p['username'], 'pwd': p['password']} for p in proxy_all()]


# ============================================================
# content_library (内容库, 多版本)
# ============================================================
def content_all():
    return [{'id': r[0], 'name': r[1], 'created_at': r[2], 'updated_at': r[3]}
            for r in _q('SELECT id,name,created_at,updated_at FROM content_library ORDER BY id DESC')]


def content_get(cid):
    r = _q1('SELECT id,name,content_json,created_at,updated_at FROM content_library WHERE id=?', (int(cid),))
    if not r:
        return None
    return {'id': r[0], 'name': r[1], 'content': json.loads(r[2] or '{}'),
            'created_at': r[3], 'updated_at': r[4]}


def content_add(name, content):
    now = _now()
    cur = _exec('INSERT INTO content_library(name,content_json,created_at,updated_at) VALUES(?,?,?,?)',
                (name or '未命名内容', json.dumps(content or {}, ensure_ascii=False), now, now))
    _commit()
    return content_all()[0]['id'] if cur else None


def content_update(cid, name, content):
    _exec('UPDATE content_library SET name=?,content_json=?,updated_at=? WHERE id=?',
          (name or '未命名内容', json.dumps(content or {}, ensure_ascii=False), _now(), int(cid)))
    _commit()


def content_del(cid):
    _exec('DELETE FROM content_library WHERE id=?', (int(cid),))
    _commit()


# ============================================================
# mk_library (魔改配置库, 多版本)
# ============================================================
def mk_all():
    return [{'id': r[0], 'name': r[1], 'created_at': r[2], 'updated_at': r[3]}
            for r in _q('SELECT id,name,created_at,updated_at FROM mk_library ORDER BY id DESC')]


def mk_get(mid):
    r = _q1('SELECT id,name,cfg_json,created_at,updated_at FROM mk_library WHERE id=?', (int(mid),))
    if not r:
        return None
    return {'id': r[0], 'name': r[1], 'cfg': json.loads(r[2] or '{}'),
            'created_at': r[3], 'updated_at': r[4]}


def mk_add(name, cfg):
    now = _now()
    cur = _exec('INSERT INTO mk_library(name,cfg_json,created_at,updated_at) VALUES(?,?,?,?)',
                (name or '未命名配置', json.dumps(cfg or {}, ensure_ascii=False), now, now))
    _commit()
    return mk_all()[0]['id'] if cur else None


def mk_update(mid, name, cfg):
    _exec('UPDATE mk_library SET name=?,cfg_json=?,updated_at=? WHERE id=?',
          (name or '未命名配置', json.dumps(cfg or {}, ensure_ascii=False), _now(), int(mid)))
    _commit()


def mk_del(mid):
    _exec('DELETE FROM mk_library WHERE id=?', (int(mid),))
    _commit()