"""本地 SQLite 管理层 — 加工数据中心"""

import sqlite3
import re
import config
from pathlib import Path


def get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(config.LOCAL_DB)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    return conn


def init_db():
    """初始化本地 DB 表结构"""
    conn = get_conn()
    conn.executescript("""
    -- 消息处理追踪
    CREATE TABLE IF NOT EXISTS processed_messages (
        msg_id       INTEGER PRIMARY KEY,
        chatroom_id  TEXT,
        talker       TEXT,
        msg_type     INTEGER,
        processed_at TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 群聊日总结
    CREATE TABLE IF NOT EXISTS chat_summaries (
        id           INTEGER PRIMARY KEY AUTOINCREMENT,
        chatroom_id  TEXT,
        date         TEXT,
        summary      TEXT,
        msg_count    INTEGER DEFAULT 0,
        created_at   TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 文章队列 (旧表，保留兼容)
    CREATE TABLE IF NOT EXISTS article_queue (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        source_msg_id INTEGER,
        chatroom_id   TEXT,
        talker        TEXT,
        url           TEXT,
        title         TEXT,
        status        TEXT DEFAULT 'pending',
        extracted_at  TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 文章主表 (一篇一行，URL 去重)
    CREATE TABLE IF NOT EXISTS articles (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        url           TEXT UNIQUE,
        title         TEXT,
        first_seen_at TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 文章分享记录 (哪个群、谁、几点几分分享了哪篇)
    CREATE TABLE IF NOT EXISTS article_shares (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        article_id    INTEGER REFERENCES articles(id),
        chatroom_id   TEXT,
        talker        TEXT,
        source_msg_id INTEGER,
        msg_create_time INTEGER,  -- 原始消息时间戳(ms)
        shared_at     TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 图片分析
    CREATE TABLE IF NOT EXISTS image_analysis (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        source_msg_id INTEGER,
        chatroom_id   TEXT,
        talker        TEXT,
        image_path    TEXT,
        ocr_text      TEXT,
        description   TEXT,
        analyzed_at   TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 日报
    CREATE TABLE IF NOT EXISTS daily_reports (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        chatroom_id   TEXT,
        chatroom_name TEXT,
        report_date   TEXT,
        content       TEXT,
        msg_count     INTEGER DEFAULT 0,
        image_count   INTEGER DEFAULT 0,
        article_count INTEGER DEFAULT 0,
        dedup_key     TEXT,
        article_url   TEXT,
        created_at    TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 日报图片
    CREATE TABLE IF NOT EXISTS report_images (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        report_id     INTEGER,
        image_path    TEXT,
        style         TEXT DEFAULT 'wsj',
        created_at    TEXT DEFAULT (datetime('now','localtime'))
    );

    -- Hermes 子进程任务追踪
    CREATE TABLE IF NOT EXISTS hermes_jobs (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        task_id       TEXT UNIQUE NOT NULL,
        job_type      TEXT NOT NULL,
        job_name      TEXT,
        status        TEXT DEFAULT 'running',
        log_file      TEXT,
        started_at    TEXT DEFAULT (datetime('now','localtime')),
        completed_at  TEXT
    );

    -- 学习吸收管线
    CREATE TABLE IF NOT EXISTS learn_queue (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        article_url   TEXT,
        title         TEXT,
        dedup_key     TEXT,
        learn_type    TEXT NOT NULL,
        learn_target  TEXT,
        learn_content TEXT,
        outline       TEXT,
        confidence    TEXT DEFAULT 'medium',
        score         INTEGER DEFAULT 0,
        status        TEXT DEFAULT 'pending_review',
        error_msg     TEXT,
        created_at    TEXT DEFAULT (datetime('now','localtime')),
        reviewed_at   TEXT
    );

    -- 学习结果记录（自学习机制的核心输出）
    CREATE TABLE IF NOT EXISTS learning_results (
        id                INTEGER PRIMARY KEY AUTOINCREMENT,
        learn_queue_id    INTEGER REFERENCES learn_queue(id),
        article_url       TEXT,
        article_title     TEXT,
        outline           TEXT,
        result_type       TEXT NOT NULL,
        result_summary    TEXT,
        created_skill     TEXT,
        updated_skill     TEXT,
        learned_methodology TEXT,
        verification_result TEXT,
        architecture_change TEXT,
        status            TEXT DEFAULT 'applied',
        human_feedback    TEXT,
        auto_applied      INTEGER DEFAULT 0,
        created_at        TEXT DEFAULT (datetime('now','localtime'))
    );

    -- 自动回复白名单（在此名单内的人/群可被直接自动回复）
    CREATE TABLE IF NOT EXISTS reply_whitelist (
        id       INTEGER PRIMARY KEY AUTOINCREMENT,
        kind     TEXT NOT NULL,          -- 'contact' | 'group'
        wxid     TEXT NOT NULL UNIQUE,   -- wxid 或 chatroom id
        name     TEXT DEFAULT '',
        role     TEXT DEFAULT 'member',  -- 'admin'=完整hermes工具 | 'member'=简单hermes -z
        added_at TEXT DEFAULT (datetime('now','localtime'))
    );

    CREATE TABLE IF NOT EXISTS reply_queue (
        id          INTEGER PRIMARY KEY AUTOINCREMENT,
        msg_id      INTEGER,
        talker      TEXT,                -- 会话对象（发送目标）
        sender      TEXT,                -- 实际发言人 wxid
        sender_name TEXT DEFAULT '',
        kind        TEXT,                -- private/group_at_me/group_quote_me
        incoming    TEXT,                -- 对方原文
        draft       TEXT,                -- 生成的回复草稿
        result_file TEXT,                -- hermes 任务结果文件路径
        action      TEXT,                -- 'auto' | 'review'
        reason      TEXT DEFAULT '',     -- 判定理由
        status      TEXT DEFAULT 'pending', -- pending|ack_sent|report_ready|auto_sent|failed|cancelled
        created_at  TEXT DEFAULT (datetime('now','localtime')),
        sent_at     TEXT
    );

    -- AI 聊天：会话表
    CREATE TABLE IF NOT EXISTS conversations (
        id         INTEGER PRIMARY KEY AUTOINCREMENT,
        title      TEXT DEFAULT '新对话',
        created_at TEXT DEFAULT (datetime('now','localtime')),
        updated_at TEXT DEFAULT (datetime('now','localtime'))
    );

    -- AI 聊天：消息表
    CREATE TABLE IF NOT EXISTS messages (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        conversation_id INTEGER NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        role            TEXT NOT NULL CHECK(role IN ('user','assistant','system')),
        content         TEXT NOT NULL,
        created_at      TEXT DEFAULT (datetime('now','localtime'))
    );
    """)
    conn.commit()

    # 为已有表补加列（兼容旧表）
    migrations = [
        "ALTER TABLE daily_reports ADD COLUMN dedup_key TEXT",
        "ALTER TABLE daily_reports ADD COLUMN article_url TEXT",
        "ALTER TABLE learn_queue ADD COLUMN outline TEXT",
        "ALTER TABLE learn_queue ADD COLUMN score INTEGER DEFAULT 0",
        "ALTER TABLE learn_queue ADD COLUMN priority INTEGER DEFAULT 0",
        "ALTER TABLE reply_whitelist ADD COLUMN role TEXT DEFAULT 'member'",
        # 2026-10-04：CREATE TABLE articles 未含 dedup_key，线上表却有 ⇒ 走迁移补列。
        # 新库也必须补，否则 article_upsert/状态机按 dedup_key 匹配会失败。
        "ALTER TABLE articles ADD COLUMN dedup_key TEXT",
        # 2026-10-04 摘要状态机（需求 reczz28JC9dkrLDn）：
        # 区分「未排上摘要(pending)」与「反复摘不出来(failed)」，failed 不拦日报。
        "ALTER TABLE articles ADD COLUMN summary_status TEXT DEFAULT 'pending'",
        "ALTER TABLE articles ADD COLUMN summary_attempts INTEGER DEFAULT 0",
        "ALTER TABLE articles ADD COLUMN summary_updated_at TEXT",
    ]
    for sql in migrations:
        try:
            conn.execute(sql)
        except sqlite3.OperationalError:
            pass  # 列已存在

    # 一次性幂等回填（2026-10-04）：已摘好的文章标记 'ok'。
    # 只动 summary_status='pending' 的行，重复执行无副作用。
    try:
        conn.execute(
            """UPDATE articles SET summary_status = 'ok'
               WHERE COALESCE(summary_status, 'pending') = 'pending'
                 AND (
                   dedup_key IN (SELECT dedup_key FROM daily_reports
                                 WHERE report_date = 'summary' AND dedup_key IS NOT NULL
                                   AND content NOT LIKE '%生成失败%' AND length(content) > 50)
                   OR url IN (SELECT article_url FROM daily_reports
                              WHERE report_date = 'summary' AND article_url IS NOT NULL
                                AND content NOT LIKE '%生成失败%' AND length(content) > 50)
                 )"""
        )
        conn.commit()
    except Exception:
        pass  # 回填失败不阻塞启动

    # 修复时区：UTC → 北京时间 (one-time migration, idempotent)
    tz_fixes = [
        # hermse_jobs: rebuild table with localtime default
        ("hermes_jobs",
         """CREATE TABLE hermes_jobs (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        task_id       TEXT UNIQUE NOT NULL,
        job_type      TEXT NOT NULL,
        job_name      TEXT,
        status        TEXT DEFAULT 'running',
        log_file      TEXT,
        started_at    TEXT DEFAULT (datetime('now','localtime')),
        completed_at  TEXT
    )"""),
        # daily_reports: rebuild table with localtime default
        ("daily_reports",
         """CREATE TABLE daily_reports (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        chatroom_id   TEXT,
        chatroom_name TEXT,
        report_date   TEXT,
        content       TEXT,
        msg_count     INTEGER DEFAULT 0,
        image_count   INTEGER DEFAULT 0,
        article_count INTEGER DEFAULT 0,
        dedup_key     TEXT,
        article_url   TEXT,
        created_at    TEXT DEFAULT (datetime('now','localtime'))
    )"""),
    ]
    for table_name, create_sql in tz_fixes:
        try:
            # Check if table has old UTC default
            old_schema = conn.execute(
                f"SELECT sql FROM sqlite_master WHERE name='{table_name}'"
            ).fetchone()
            if old_schema and "datetime('now','localtime')" not in (old_schema["sql"] or ""):
                conn.execute(f"ALTER TABLE {table_name} RENAME TO {table_name}_old")
                conn.execute(create_sql)
                conn.execute(f"INSERT INTO {table_name} SELECT * FROM {table_name}_old")
                conn.execute(f"DROP TABLE {table_name}_old")
                # Shift existing timestamps from UTC to Beijing time
                time_cols = ["started_at", "completed_at"] if table_name == "hermes_jobs" else ["created_at"]
                for col in time_cols:
                    conn.execute(
                        f"UPDATE {table_name} SET {col} = datetime({col}, '+8 hours') "
                        f"WHERE {col} IS NOT NULL AND {col} LIKE '% 0_:%'"
                    )
                conn.commit()
        except Exception:
            pass  # already migrated or no data

    # 用户认证
    conn.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id           INTEGER PRIMARY KEY AUTOINCREMENT,
            username     TEXT UNIQUE NOT NULL,
            password_hash TEXT NOT NULL,
            role         TEXT NOT NULL DEFAULT 'admin',
            created_at   TEXT DEFAULT (datetime('now','localtime'))
        )
    """)
    # 老库补 role 列（幂等）
    cols = [r[1] for r in conn.execute("PRAGMA table_info(users)").fetchall()]
    if 'role' not in cols:
        conn.execute("ALTER TABLE users ADD COLUMN role TEXT NOT NULL DEFAULT 'admin'")
    # 默认管理员 admin / wechat2026
    import hashlib
    default_pw = hashlib.sha256("wechat2026".encode()).hexdigest()
    conn.execute(
        "INSERT OR IGNORE INTO users (username, password_hash) VALUES (?, ?)",
        ("admin", default_pw)
    )

    conn.commit()

    # 加索引加速去重查询
    try:
        conn.execute("CREATE INDEX IF NOT EXISTS idx_daily_reports_dedup ON daily_reports(dedup_key, report_date)")
    except Exception:
        pass
    conn.commit()
    conn.close()


# ── 日报图片管理 ────────────────────────────────────

def report_image_list(limit: int = 100) -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT ri.*, r.chatroom_name, r.report_date
           FROM report_images ri
           LEFT JOIN daily_reports r ON ri.report_id = r.id
           ORDER BY ri.id DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def report_image_add(report_id: int, image_path: str, style: str = "wsj") -> int:
    conn = get_conn()
    cur = conn.execute(
        "INSERT INTO report_images (report_id, image_path, style) VALUES (?, ?, ?)",
        (report_id, image_path, style)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def report_image_delete(image_id: int):
    conn = get_conn()
    conn.execute("DELETE FROM report_images WHERE id = ?", (image_id,))
    conn.commit()
    conn.close()


# ── 日报管理 ────────────────────────────────────────

def report_list(limit: int = 100, report_type: str = "") -> list[dict]:
    conn = get_conn()
    if report_type == "digest":
        # Daily digests: report_date is a date string like "2026-07-08"
        rows = conn.execute(
            "SELECT * FROM daily_reports WHERE report_date NOT IN ('summary','learn') ORDER BY id DESC LIMIT ?",
            (limit,)
        ).fetchall()
    elif report_type:
        rows = conn.execute(
            "SELECT * FROM daily_reports WHERE report_date = ? ORDER BY id DESC LIMIT ?",
            (report_type, limit)
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM daily_reports ORDER BY id DESC LIMIT ?",
            (limit,)
        ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def report_get(report_id: int) -> dict | None:
    conn = get_conn()
    row = conn.execute("SELECT * FROM daily_reports WHERE id = ?", (report_id,)).fetchone()
    conn.close()
    return dict(row) if row else None


def report_add(chatroom_id: str, chatroom_name: str, report_date: str,
               content: str, msg_count: int = 0, image_count: int = 0,
               article_count: int = 0, dedup_key: str = "",
               article_url: str = "") -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO daily_reports (chatroom_id, chatroom_name, report_date, content,
           msg_count, image_count, article_count, dedup_key, article_url)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (chatroom_id, chatroom_name, report_date, content,
         msg_count, image_count, article_count,
         dedup_key or None, article_url or None)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def report_delete(report_id: int):
    conn = get_conn()
    conn.execute("DELETE FROM daily_reports WHERE id = ?", (report_id,))
    conn.commit()
    conn.close()


# ── 文章摘要去重查询 ────────────────────────────────

def find_article_summary(dedup_key: str, report_type: str) -> dict | None:
    """查某篇文章是否已有摘要/学习结果（排除失败记录）。
    report_type: 'summary' 或 'learn'
    """
    conn = get_conn()
    row = conn.execute(
        """SELECT * FROM daily_reports
           WHERE dedup_key = ? AND report_date = ?
             AND content NOT LIKE '%生成失败%'
             AND length(content) > 50
           ORDER BY id DESC LIMIT 1""",
        (dedup_key, report_type)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def find_article_summaries_batch(dedup_keys: list[str], report_type: str = "summary") -> dict[str, dict]:
    """批量查哪些 dedup_key 已有摘要。
    返回 {dedup_key: {id, content, created_at}, ...}
    report_type: 'summary' 或 'learn' 或 ''（查任何类型）
    """
    if not dedup_keys:
        return {}
    conn = get_conn()
    # 2026-08-28：非公众号链接（如 changdunovel 互动剧）dedup_key 就是 URL 本身，但
    # gen_agg_digest 的 API 调用方在 _extract_dedup_key 得 None 时会以 "t:<title>" 传 key
    # → 双通道匹配：dedup_key 精确 + 标题 LIKE。
    dk_keys = [k for k in dedup_keys if not k.startswith("t:")]
    t_keys = [k[2:] for k in dedup_keys if k.startswith("t:")]
    result = {}

    if dk_keys:
        placeholders = ",".join(["?"] * len(dk_keys))
        if report_type:
            sql = f"""SELECT dedup_key, id, content, report_date, created_at
                      FROM daily_reports
                      WHERE dedup_key IN ({placeholders}) AND report_date = ?
                        AND content NOT LIKE '%生成失败%'
                        AND length(content) > 50
                      ORDER BY id DESC"""
            rows = conn.execute(sql, [*dk_keys, report_type]).fetchall()
        else:
            sql = f"""SELECT dedup_key, id, content, report_date, created_at
                      FROM daily_reports
                      WHERE dedup_key IN ({placeholders})
                        AND content NOT LIKE '%生成失败%'
                        AND length(content) > 50
                      ORDER BY id DESC"""
            rows = conn.execute(sql, dk_keys).fetchall()
        for row in rows:
            d = dict(row)
            dk = d.pop("dedup_key")
            if dk and dk not in result:
                result[dk] = d

    if t_keys:
        # LIKE 粗筛候选：标题前 12 字 + 书名号内标题（分享标题常带动词前缀，如「跟我一起免费看《X》」）
        like_vals = []
        like_conds = []
        for t in t_keys:
            like_conds.append("content LIKE ?")
            like_vals.append(f"%{t[:12]}%")
            m_book = re.search(r"《(.+?)》", t)
            if m_book:
                like_conds.append("content LIKE ?")
                like_vals.append(f"%{m_book.group(1)}%")
        cond = " OR ".join(like_conds)
        sql = f"""SELECT dedup_key, id, content, report_date, created_at
                  FROM daily_reports
                  WHERE report_date = 'summary' AND ({cond})
                    AND content NOT LIKE '%生成失败%'
                    AND length(content) > 50
                  ORDER BY id DESC"""
        for row in conn.execute(sql, like_vals).fetchall():
            d = dict(row)
            c = d.get("content") or ""
            first = c.split(chr(10))[0] if c else ""
            m_title = re.match(r"^#\s*(.+?)摘要", first)
            doc_title = m_title.group(1).strip() if m_title else ""
            for t in t_keys:
                # 双向包含：内容含标题 或 摘要首行书名标题是标题的子串（2026-08-28 柠萌萌互动剧）
                if (t in c or (doc_title and doc_title in t)) and f"t:{t}" not in result:
                    result[f"t:{t}"] = d
                    break
    conn.close()
    return result

# ── 文章队列 ────────────────────────────────────────

def article_list(status: str = "pending", limit: int = 100) -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT * FROM article_queue WHERE status = ? ORDER BY id DESC LIMIT ?""",
        (status, limit)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def article_add(source_msg_id: int, chatroom_id: str, talker: str,
                url: str, title: str) -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO article_queue (source_msg_id, chatroom_id, talker, url, title)
           VALUES (?, ?, ?, ?, ?)""",
        (source_msg_id, chatroom_id, talker, url, title)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def article_update_status(article_id: int, status: str):
    conn = get_conn()
    conn.execute(
        "UPDATE article_queue SET status = ? WHERE id = ?",
        (status, article_id)
    )
    conn.commit()
    conn.close()


def article_find_by_url(url: str) -> dict | None:
    """Check if article URL already exists (dedup)."""
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM article_queue WHERE url = ?", (url,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def article_get_unsummarized(limit: int = 20) -> list[dict]:
    """Articles in article_queue that have no summary in daily_reports."""
    conn = get_conn()
    conn.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r))
    rows = conn.execute(
        """SELECT a.id, a.url, a.title, a.talker, a.chatroom_id, a.extracted_at
           FROM article_queue a
           WHERE a.status = 'pending'
             AND a.id NOT IN (
               SELECT DISTINCT a2.id FROM article_queue a2
               INNER JOIN daily_reports r ON r.article_url = a2.url AND r.report_date = 'summary'
             )
           ORDER BY a.id DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


# ── 新文章表 (articles + article_shares) ──────────────

def article_count() -> int:
    conn = get_conn()
    cnt = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
    conn.close()
    return cnt


def article_upsert(url: str, title: str, return_was_new: bool = False) -> int | tuple[int, bool]:
    """Insert or get article_id (dedup by dedup_key from biz|mid|idx|sn).
    If return_was_new=True, returns (article_id, was_new)."""
    dedup_key = _extract_dedup_key(url)
    conn = get_conn()
    if dedup_key:
        row = conn.execute("SELECT id FROM articles WHERE dedup_key = ?", (dedup_key,)).fetchone()
        if row:
            conn.close()
            return (row[0], False) if return_was_new else row[0]
    else:
        row = conn.execute("SELECT id FROM articles WHERE url = ?", (url,)).fetchone()
        if row:
            conn.close()
            return (row[0], False) if return_was_new else row[0]
    cur = conn.execute("INSERT INTO articles (url, title, dedup_key) VALUES (?, ?, ?)",
                       (url, title, dedup_key))
    conn.commit()
    rid = cur.lastrowid
    conn.close()
    return (rid, True) if return_was_new else rid


def _extract_dedup_key(url: str) -> str:
    """Extract canonical dedup key: biz|mid|idx|sn"""
    if not url:
        return ""
    if "mp.weixin.qq.com" in url:
        biz = re.search(r"__biz=([^&]+)", url)
        mid = re.search(r"mid=(\d+)", url)
        idx = re.search(r"idx=(\d+)", url)
        sn   = re.search(r"sn=([a-f0-9]+)", url)
        if biz and mid and idx and sn:
            return f"{biz.group(1)}|{mid.group(1)}|{idx.group(1)}|{sn.group(1)}"
    return url


def article_share_add(article_id: int, chatroom_id: str, talker: str,
                      source_msg_id: int, msg_create_time: int):
    """Record that article was shared in a group at a specific time by a specific person."""
    conn = get_conn()
    conn.execute(
        """INSERT INTO article_shares (article_id, chatroom_id, talker, source_msg_id, msg_create_time)
           VALUES (?, ?, ?, ?, ?)""",
        (article_id, chatroom_id, talker, source_msg_id, msg_create_time)
    )
    conn.commit()
    conn.close()


def share_exists(source_msg_id: int) -> bool:
    """Check if a share with this WeChat msgId already exists."""
    conn = get_conn()
    row = conn.execute(
        "SELECT 1 FROM article_shares WHERE source_msg_id = ?", (source_msg_id,)
    ).fetchone()
    conn.close()
    return row is not None


def article_list_new(page: int = 1, per_page: int = 20,
                     date_from: str = "", date_to: str = "") -> tuple[int, list[dict]]:
    """List articles from articles+article_shares, grouped by URL."""
    conn = get_conn()
    params = []
    where = ""
    if date_from:
        where += " AND s.shared_at >= ?"
        params.append(date_from + " 00:00:00")
    if date_to:
        where += " AND s.shared_at < ?"
        params.append(date_to + " 23:59:59")

    total = conn.execute(
        f"SELECT COUNT(DISTINCT a.id) FROM articles a JOIN article_shares s ON a.id = s.article_id WHERE 1=1{where}",
        params
    ).fetchone()[0]

    conn.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r))
    rows = conn.execute(
        f"""SELECT a.id, a.url, a.title,
            MAX(s.msg_create_time) as createTime,
            COUNT(DISTINCT s.chatroom_id) as shareCount,
            GROUP_CONCAT(DISTINCT s.chatroom_id) as groups,
            MIN(s.shared_at) as first_seen
        FROM articles a
        JOIN article_shares s ON a.id = s.article_id
        WHERE 1=1{where}
        GROUP BY a.id
        ORDER BY MAX(s.msg_create_time) DESC
        LIMIT ? OFFSET ?""",
        params + [per_page, (page - 1) * per_page]
    ).fetchall()
    conn.close()

    articles = []
    for r in rows:
        groups = r["groups"].split(",") if r["groups"] else []
        articles.append({
            "url": r["url"], "title": r["title"],
            "createTime": r["createTime"] or 0,
            "shareCount": r["shareCount"], "groups": groups,
        })
    return total, articles


def _url_dedup_key(url: str) -> str:
    """Extract WeChat article dedup key (biz|mid|idx|sn) or fall back to the full URL."""
    if not url or "mp.weixin.qq.com" not in url:
        return url or ""
    biz = re.search(r"__biz=([^&]+)", url)
    mid = re.search(r"mid=(\d+)", url)
    idx = re.search(r"idx=(\d+)", url)
    sn = re.search(r"sn=([a-f0-9]+)", url)
    if biz and mid and idx and sn:
        return f"{biz.group(1)}|{mid.group(1)}|{idx.group(1)}|{sn.group(1)}"
    return url


def article_get_unsummarized_new(limit: int = 20) -> list[dict]:
    """Articles without summaries (from new tables), newest first.

    A summary is considered present if a daily_reports row exists with
    report_date='summary' and a matching dedup_key or article_url.
    """
    conn = get_conn()
    conn.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r))

    # Load all articles with shares and all summary reports once.
    articles = conn.execute(
        """SELECT DISTINCT a.id, a.url, a.title, MIN(s.chatroom_id) as chatroom_id,
               MAX(s.msg_create_time) as latest_share
           FROM articles a
           JOIN article_shares s ON a.id = s.article_id
           GROUP BY a.id"""
    ).fetchall()

    reports = conn.execute(
        """SELECT dedup_key, article_url FROM daily_reports
           WHERE report_date = 'summary'
             AND content NOT LIKE '%生成失败%'
             AND length(content) > 50"""
    ).fetchall()

    # Build a set of dedup_keys and normalized article_urls for fast lookup.
    report_keys = set()
    report_urls = set()
    for r in reports:
        if r["dedup_key"]:
            report_keys.add(r["dedup_key"])
        if r["article_url"]:
            report_urls.add(r["article_url"])
            report_urls.add(r["article_url"].replace("&amp;", "&"))

    unsummarized = []
    for a in articles:
        dk = _url_dedup_key(a["url"])
        if dk in report_keys or a["url"] in report_urls:
            continue
        unsummarized.append(dict(a))

    unsummarized.sort(key=lambda x: x.get("latest_share") or 0, reverse=True)
    conn.close()
    return unsummarized[:limit]


def article_count_summarized() -> int:
    """Return number of articles that have a cached summary (report_date='summary').

    Matches by dedup_key or normalized article_url, handling historical data where
    article_url may contain &amp; or even be stored as a dedup_key.
    """
    conn = get_conn()
    articles = conn.execute("SELECT id, url FROM articles").fetchall()
    reports = conn.execute(
        """SELECT dedup_key, article_url FROM daily_reports
           WHERE report_date = 'summary'
             AND content NOT LIKE '%生成失败%'
             AND length(content) > 50"""
    ).fetchall()
    report_keys = set()
    report_urls = set()
    for r in reports:
        if r["dedup_key"]:
            report_keys.add(r["dedup_key"])
        if r["article_url"]:
            report_urls.add(r["article_url"])
            report_urls.add(r["article_url"].replace("&amp;", "&"))
    count = 0
    for a in articles:
        dk = _url_dedup_key(a["url"])
        if dk in report_keys or a["url"] in report_urls:
            count += 1
    conn.close()
    return count


def article_count_learned() -> int:
    """Return number of articles that have a cached learn result (report_date='learn')."""
    conn = get_conn()
    articles = conn.execute("SELECT id, url FROM articles").fetchall()
    reports = conn.execute(
        """SELECT dedup_key, article_url FROM daily_reports
           WHERE report_date = 'learn' AND article_url IS NOT NULL"""
    ).fetchall()
    report_keys = set()
    report_urls = set()
    for r in reports:
        if r["dedup_key"]:
            report_keys.add(r["dedup_key"])
        if r["article_url"]:
            report_urls.add(r["article_url"])
            report_urls.add(r["article_url"].replace("&amp;", "&"))
    count = 0
    for a in articles:
        dk = _url_dedup_key(a["url"])
        if dk in report_keys or a["url"] in report_urls:
            count += 1
    conn.close()
    return count


# ── 文章摘要状态机（2026-10-04，需求 reczz28JC9dkrLDn）────────────
# 语义：pending=未排上摘要（应阻塞日报）；summarizing=已派发抓取中；
#       ok=已摘好；failed=连续失败达上限（不阻塞日报，正文以「缺摘要」可见）。
SUMMARY_STATUS_PENDING = "pending"
SUMMARY_STATUS_SUMMARIZING = "summarizing"
SUMMARY_STATUS_OK = "ok"
SUMMARY_STATUS_FAILED = "failed"
SUMMARY_MAX_ATTEMPTS = 3          # 连续失败达此次数 ⇒ failed
SUMMARY_STALE_MINUTES = 10        # summarizing 悬挂超此分钟数视为可重派


def _article_match_clause(dedup_key: str | None, url: str | None):
    """构造 articles 行匹配条件：dedup_key 或 url 任一命中。返回 (where, params)。"""
    conds, params = [], []
    if dedup_key:
        conds.append("dedup_key = ?")
        params.append(dedup_key)
    if url:
        conds.append("url = ?")
        params.append(url)
    if not conds:
        return None, []
    return "(" + " OR ".join(conds) + ")", params


def article_summary_get_state(dedup_key: str | None = None,
                              url: str | None = None) -> dict | None:
    """读取某篇文章的摘要状态行；找不到返回 None。"""
    where, params = _article_match_clause(dedup_key, url)
    if not where:
        return None
    conn = get_conn()
    row = conn.execute(
        f"""SELECT id, url, dedup_key, summary_status, summary_attempts, summary_updated_at
            FROM articles WHERE {where} LIMIT 1""",
        params
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def article_summary_set_dispatching(dedup_key: str | None = None,
                                    url: str | None = None) -> bool:
    """派发摘要前调用：attempts+1、status='summarizing'、updated_at=now。

    - status='failed' 的行不再自增/改状态（防重派，返回 False）。
    - summarizing 悬挂（updated_at 超 SUMMARY_STALE_MINUTES）视为可重派（照常 attempts+1）。
    返回是否命中并更新了行。
    """
    where, params = _article_match_clause(dedup_key, url)
    if not where:
        return False
    conn = get_conn()
    cur = conn.execute(
        f"""UPDATE articles
            SET summary_attempts = COALESCE(summary_attempts, 0) + 1,
                summary_status = 'summarizing',
                summary_updated_at = datetime('now','localtime')
            WHERE {where} AND COALESCE(summary_status, 'pending') != 'failed'""",
        params
    )
    conn.commit()
    changed = cur.rowcount > 0
    conn.close()
    return changed


def article_summary_mark_failed(dedup_key: str | None = None,
                                url: str | None = None) -> str:
    """记录一次摘要失败：按 attempts 决定 → 达上限转 'failed'，否则回落 'pending'。

    返回写入后的状态（'' 表示未命中行）。
    """
    where, params = _article_match_clause(dedup_key, url)
    if not where:
        return ""
    conn = get_conn()
    row = conn.execute(
        f"SELECT summary_attempts FROM articles WHERE {where} LIMIT 1", params
    ).fetchone()
    attempts = (row["summary_attempts"] if row else 0) or 0
    new_status = SUMMARY_STATUS_FAILED if attempts >= SUMMARY_MAX_ATTEMPTS else SUMMARY_STATUS_PENDING
    conn.execute(
        f"""UPDATE articles SET summary_status = ?,
               summary_updated_at = datetime('now','localtime')
            WHERE {where}""",
        [new_status, *params]
    )
    conn.commit()
    conn.close()
    return new_status


def article_summary_mark_ok(dedup_key: str | None = None,
                            url: str | None = None) -> bool:
    """摘要成功入库后调用：status='ok'。返回是否命中行。"""
    where, params = _article_match_clause(dedup_key, url)
    if not where:
        return False
    conn = get_conn()
    cur = conn.execute(
        f"""UPDATE articles SET summary_status = 'ok',
               summary_updated_at = datetime('now','localtime')
            WHERE {where}""",
        params
    )
    conn.commit()
    changed = cur.rowcount > 0
    conn.close()
    return changed


def article_summary_state_counts() -> dict:
    """按状态统计 articles：{pending, summarizing, ok, failed, total}。"""
    counts = {"pending": 0, "summarizing": 0, "ok": 0, "failed": 0}
    conn = get_conn()
    rows = conn.execute(
        """SELECT COALESCE(summary_status,'pending') AS s, COUNT(*) AS c
           FROM articles GROUP BY COALESCE(summary_status,'pending')"""
    ).fetchall()
    conn.close()
    for r in rows:
        key = r["s"] if r["s"] in counts else "pending"
        counts[key] += r["c"]
    counts["total"] = sum(counts[k] for k in ("pending", "summarizing", "ok", "failed"))
    return counts


def article_summary_failed_titles(limit: int = 100) -> list[str]:
    """列出 failed 状态的文章标题。"""
    conn = get_conn()
    rows = conn.execute(
        """SELECT title FROM articles
           WHERE summary_status = 'failed'
           ORDER BY summary_updated_at DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [r["title"] for r in rows if r["title"]]


def article_get_by_group(chatroom_id: str, limit: int = 50) -> list[dict]:
    """Get all articles shared in a specific group, newest first."""
    conn = get_conn()
    conn.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r))
    rows = conn.execute(
        """SELECT DISTINCT a.id, a.url, a.title,
            MAX(s.msg_create_time) as createTime,
            r.content as summary, r.id as report_id
           FROM articles a
           JOIN article_shares s ON a.id = s.article_id
           LEFT JOIN daily_reports r ON r.article_url = a.url AND r.report_date = 'summary'
           WHERE s.chatroom_id = ?
           GROUP BY a.id
           ORDER BY MAX(s.msg_create_time) DESC
           LIMIT ?""",
        (chatroom_id, limit)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


# ── 图片分析 ────────────────────────────────────────

def image_list(limit: int = 100) -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM image_analysis ORDER BY id DESC LIMIT ?",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def image_add(source_msg_id: int, chatroom_id: str, talker: str,
              image_path: str, ocr_text: str = "", description: str = "") -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO image_analysis (source_msg_id, chatroom_id, talker, image_path, ocr_text, description)
           VALUES (?, ?, ?, ?, ?, ?)""",
        (source_msg_id, chatroom_id, talker, image_path, ocr_text, description)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


# ── 群聊总结 ────────────────────────────────────────

def summary_list(limit: int = 50) -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM chat_summaries ORDER BY date DESC, id DESC LIMIT ?",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def summary_add(chatroom_id: str, date: str, summary: str, msg_count: int = 0) -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO chat_summaries (chatroom_id, date, summary, msg_count)
           VALUES (?, ?, ?, ?)""",
        (chatroom_id, date, summary, msg_count)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


# ── Hermes 子进程追踪 ───────────────────────────────

def job_add(task_id: str, job_type: str, job_name: str = "", log_file: str = "") -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO hermes_jobs (task_id, job_type, job_name, log_file)
           VALUES (?, ?, ?, ?)""",
        (task_id, job_type, job_name, log_file)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def job_update_status(task_id: str, status: str):
    conn = get_conn()
    if status in ("completed", "failed"):
        conn.execute(
            "UPDATE hermes_jobs SET status = ?, completed_at = datetime('now','localtime') WHERE task_id = ?",
            (status, task_id)
        )
    else:
        conn.execute(
            "UPDATE hermes_jobs SET status = ? WHERE task_id = ?",
            (status, task_id)
        )
    conn.commit()
    conn.close()


def job_list(limit: int = 100) -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM hermes_jobs ORDER BY id DESC LIMIT ?", (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def job_get(task_id: str) -> dict | None:
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM hermes_jobs WHERE task_id = ?", (task_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def job_delete(task_id: str):
    conn = get_conn()
    conn.execute("DELETE FROM hermes_jobs WHERE task_id = ?", (task_id,))
    conn.commit()
    conn.close()


# ── 学习吸收管线 ────────────────────────────────────

def learn_add(article_url: str, title: str, dedup_key: str,
              learn_type: str, learn_target: str = "",
              learn_content: str = "", confidence: str = "medium",
              score: int = 0, status: str = "pending_review",
              outline: str = "") -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO learn_queue (article_url, title, dedup_key, learn_type,
           learn_target, learn_content, confidence, score, status, outline)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (article_url, title, dedup_key, learn_type,
         learn_target, learn_content, confidence, score, status, outline)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def learn_list(status: str = "pending_review", limit: int = 200) -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM learn_queue WHERE status = ? ORDER BY id DESC LIMIT ?",
        (status, limit)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def learn_get(item_id: int) -> dict | None:
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM learn_queue WHERE id = ?", (item_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def learn_update_status(item_id: int, status: str, error_msg: str = ""):
    conn = get_conn()
    if status in ("approved", "rejected", "applied"):
        conn.execute(
            """UPDATE learn_queue SET status = ?, reviewed_at = datetime('now','localtime'),
               error_msg = ? WHERE id = ?""",
            (status, error_msg, item_id)
        )
    else:
        conn.execute(
            "UPDATE learn_queue SET status = ?, error_msg = ? WHERE id = ?",
            (status, error_msg, item_id)
        )
    conn.commit()
    conn.close()


def learn_find_by_dedup(dedup_key: str, learn_type: str = "") -> dict | None:
    """Check if an article already has a learn queue entry"""
    conn = get_conn()
    if learn_type:
        row = conn.execute(
            "SELECT * FROM learn_queue WHERE dedup_key = ? AND learn_type = ? ORDER BY id DESC LIMIT 1",
            (dedup_key, learn_type)
        ).fetchone()
    else:
        row = conn.execute(
            "SELECT * FROM learn_queue WHERE dedup_key = ? ORDER BY id DESC LIMIT 1",
            (dedup_key,)
        ).fetchone()
    conn.close()
    return dict(row) if row else None


def learn_get_unclassified_summaries(limit: int = 50) -> list[dict]:
    """Find article summaries that haven't been classified yet"""
    conn = get_conn()
    rows = conn.execute(
        """SELECT dr.* FROM daily_reports dr
           WHERE dr.report_date = 'summary'
           AND dr.dedup_key IS NOT NULL
           AND dr.dedup_key NOT IN (
               SELECT DISTINCT lq.dedup_key FROM learn_queue lq
               WHERE lq.dedup_key IS NOT NULL
           )
           ORDER BY dr.id DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


# ── 学习结果记录 ────────────────────────────────────

def learn_result_add(learn_queue_id: int, article_url: str = "",
                     article_title: str = "", outline: str = "",
                     result_type: str = "", result_summary: str = "",
                     created_skill: str = "", updated_skill: str = "",
                     learned_methodology: str = "",
                     verification_result: str = "",
                     architecture_change: str = "",
                     status: str = "applied",
                     auto_applied: int = 0) -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO learning_results
           (learn_queue_id, article_url, article_title, outline,
            result_type, result_summary, created_skill, updated_skill,
            learned_methodology, verification_result, architecture_change,
            status, auto_applied)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (learn_queue_id, article_url, article_title, outline,
         result_type, result_summary, created_skill, updated_skill,
         learned_methodology, verification_result, architecture_change,
         status, auto_applied)
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def learn_result_list(limit: int = 100) -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM learning_results ORDER BY id DESC LIMIT ?",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def learn_result_get(item_id: int) -> dict | None:
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM learning_results WHERE id = ?", (item_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def learn_result_get_by_learn_id(learn_queue_id: int) -> dict | None:
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM learning_results WHERE learn_queue_id = ? ORDER BY id DESC LIMIT 1",
        (learn_queue_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def learn_result_get_apply_audit(learn_queue_id: int) -> dict | None:
    """取 Step 6 知识应用审计报告。"""
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM learning_results WHERE learn_queue_id = ? AND result_type='apply_audit' ORDER BY id DESC LIMIT 1",
        (learn_queue_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def learn_result_get_step4(learn_queue_id: int) -> dict | None:
    """取 Step 4 吸收报告（排除 apply_audit）。"""
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM learning_results WHERE learn_queue_id = ? AND result_type != 'apply_audit' ORDER BY id DESC LIMIT 1",
        (learn_queue_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def learn_result_add_feedback(item_id: int, feedback: str):
    conn = get_conn()
    conn.execute(
        "UPDATE learning_results SET human_feedback = ? WHERE id = ?",
        (feedback, item_id)
    )
    conn.commit()
    conn.close()


def learn_result_update_status(item_id: int, status: str):
    conn = get_conn()
    conn.execute(
        "UPDATE learning_results SET status = ? WHERE id = ?",
        (status, item_id)
    )
    conn.commit()
    conn.close()


def learn_dagger_corrections(limit: int = 50) -> list[dict]:
    """DAgger Correction 模式查询：找出通过 REVISE 重试后成功的吸收项。

    RoboTTT 实证 (2026-08-03): 保留失败为上下文的纠正学习 (+36%)
    价值是纯成功路径 (+9%) 的 4x。此查询表面这些高价值案例供离线分析。

    检测逻辑: error_msg 含 action=REVISE 但 status=applied（说明重试成功了）。
    """
    conn = get_conn()
    rows = conn.execute(
        """SELECT id, title, learn_type, learn_target, score, error_msg, created_at
           FROM learn_queue
           WHERE status = 'applied'
             AND error_msg LIKE '%action=REVISE%'
           ORDER BY id DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def learn_dagger_stats() -> dict:
    """DAgger 统计：计算纠正学习 vs 纯成功的比例。

    Returns: {total_applied, dagger_count, dagger_ratio, pure_count}
    """
    conn = get_conn()
    total = conn.execute(
        "SELECT COUNT(*) FROM learn_queue WHERE status = 'applied'"
    ).fetchone()[0]
    dagger = conn.execute(
        "SELECT COUNT(*) FROM learn_queue WHERE status = 'applied'"
        " AND error_msg LIKE '%action=REVISE%'"
    ).fetchone()[0]
    conn.close()
    return {
        "total_applied": total,
        "dagger_count": dagger,
        "dagger_ratio": round(dagger / total * 100, 1) if total > 0 else 0,
        "pure_count": total - dagger,
    }


# ── 分类路由阈值：单一事实来源（校准可推导，禁止散落字面量）──
# 判据（pplx-decider 决策模型知识, learn_queue #bcf45b01, 2026-10-13）：
#   阈值只有在置信度/分数**经校准**后才可信——「说 90% 把握时真实准确率也接近 90%」，
#   否则概率只是装饰品，阈值就是拍板常量。实测口径见 learn_calibration_stats()。
#   改阈值只改这里（5 处散落字面量已收敛 ⇒ 校准结论有落点）。
LEARN_ROUTE_LEARN = 7    # score >= 此值 → pending_learn（自动吸收）
LEARN_ROUTE_REVIEW = 4   # score >= 此值 → pending_review（人工审核）

# 校准判据常量（确定性事实，非 LLM 评分）
CALIBRATION_MIN_SAMPLE = 20       # 每档最少已决样本；低于此不出结论（小样本 = 假置信）
CALIBRATION_TARGET_SUCCESS = 0.8  # "high" 档目标成功率下限


def learn_calibration_stats() -> dict:
    """分类器评分校准统计（奖励黑客解毒剂，2026-08-05）。

    Hermes 官方设计哲学知识：评价指标本身会被污染——分类器的高分文章
    如果被人工标记为「不值得学」，说明评分被污染，需要校准。
    核心 KPI：score 准确率 = 高分项最终被吸收（applied）的比例。

    2026-10-13 增（pplx-decider 决策模型，learn_queue #bcf45b01）：
    校准比分数本身更值钱 ⇒ 补三件确定性事实：
      ① confidence_calibration 经验校准曲线（分档真实成功率）——
         旧版只有 confidence_high_failed 计数，无法回答「high 到底可不可信」；
      ② 分母含 failed（旧 high_precision 只看 applied+rejected，漏最大失败档 ⇒ 假阴）；
      ③ n_decided / reliable 样本量守门（decided=1 也回 100% 是小样本假置信）。

    Returns:
      - auto_high_score_applied: score>=7 且最终 applied（分类器正确）
      - auto_high_score_rejected: score>=7 但人工/系统 rejected（评分被推翻）
      - high_precision: auto_high_score_applied / (applied+rejected) 中 score>=7 的成功率
      - confidence_calibration: 分档 {n_decided, applied, failed, rejected, success_rate}
      - confidence_verdict: CALIBRATED / BELOW_TARGET / MISCALIBRATED / INSUFFICIENT_SAMPLE
      - threshold_at: 学习路由阈值处的实测精度 + 样本量 + reliable 标志
      - confidence_calibration: confidence=high 但 failed/rejected 的数量（置信度虚高）
      - score_distribution: 按 score 分档的最终状态分布
    """
    conn = get_conn()

    def _q(sql, args=()):
        return conn.execute(sql, args).fetchone()[0]

    _THR = LEARN_ROUTE_LEARN

    high_applied = _q("SELECT COUNT(*) FROM learn_queue WHERE score >= ? AND status = 'applied'", (_THR,))
    high_rejected = _q("SELECT COUNT(*) FROM learn_queue WHERE score >= ? AND status = 'rejected'", (_THR,))
    high_pending = _q("SELECT COUNT(*) FROM learn_queue WHERE score >= ? AND status IN ('pending_review','pending_learn','approved')", (_THR,))
    mid_applied = _q("SELECT COUNT(*) FROM learn_queue WHERE score >= ? AND score < ? AND status = 'applied'", (LEARN_ROUTE_REVIEW, _THR))
    mid_rejected = _q("SELECT COUNT(*) FROM learn_queue WHERE score >= ? AND score < ? AND status = 'rejected'", (LEARN_ROUTE_REVIEW, _THR))
    conf_high_failed = _q("SELECT COUNT(*) FROM learn_queue WHERE confidence = 'high' AND status IN ('failed','rejected')")

    # ── 已决结果计数（分母含 failed：failed 是吸收管线最大的失败档，漏掉即假阴）──
    def _decided(where_sql, args=()):
        row = conn.execute(
            "SELECT SUM(status='applied') a, SUM(status='failed') f, SUM(status='rejected') r "
            f"FROM learn_queue WHERE {where_sql}", args
        ).fetchone()
        a, f, r = row["a"] or 0, row["f"] or 0, row["r"] or 0
        n = a + f + r
        return {
            "n_decided": n, "applied": a, "failed": f, "rejected": r,
            "success_rate": round(a / n, 3) if n else None,
        }

    # ── ① 经验校准曲线：confidence 档位 vs 真实成功率 ──
    confidence_calibration = {}
    for bucket in ("high", "medium", "low"):
        entry = _decided("confidence = ?", (bucket,))
        entry["reliable"] = entry["n_decided"] >= CALIBRATION_MIN_SAMPLE
        confidence_calibration[bucket] = entry

    # ── ② 阈值处实测精度（含 failed）──
    threshold_at = _decided("score >= ?", (_THR,))
    threshold_at["reliable"] = threshold_at["n_decided"] >= CALIBRATION_MIN_SAMPLE
    threshold_at["threshold"] = _THR

    # ── ③ 确定性 verdict（判据来自 DB 计数，非 LLM 评分）──
    hi_c, mid_c = confidence_calibration["high"], confidence_calibration["medium"]
    if not hi_c["reliable"] or not mid_c["reliable"]:
        confidence_verdict = "INSUFFICIENT_SAMPLE"
    elif (hi_c["success_rate"] or 0) < (mid_c["success_rate"] or 0):
        # high 档比 medium 档更差 ⇒ 置信度与结果反相关，阈值不可信
        confidence_verdict = "MISCALIBRATED"
    elif (hi_c["success_rate"] or 0) < CALIBRATION_TARGET_SUCCESS:
        confidence_verdict = "BELOW_TARGET"
    else:
        confidence_verdict = "CALIBRATED"

    # 按 score 分档的状态分布
    dist = []
    for lo, hi in ((9, 10), (_THR, _THR + 1), (LEARN_ROUTE_REVIEW, _THR - 1), (1, LEARN_ROUTE_REVIEW - 1)):
        rows = conn.execute(
            "SELECT status, COUNT(*) c FROM learn_queue WHERE score >= ? AND score <= ? "
            "GROUP BY status ORDER BY c DESC",
            (lo, hi)
        ).fetchall()
        dist.append({
            "band": f"{lo}-{hi}",
            "statuses": {r["status"]: r["c"] for r in rows},
        })
    conn.close()

    decided = high_applied + high_rejected
    return {
        "high_precision": round(high_applied / decided * 100, 1) if decided > 0 else None,
        "high_applied": high_applied,
        "high_rejected": high_rejected,
        "high_pending": high_pending,
        "mid_applied": mid_applied,
        "mid_rejected": mid_rejected,
        "confidence_high_failed": conf_high_failed,
        "confidence_calibration": confidence_calibration,
        "confidence_verdict": confidence_verdict,
        "threshold_at": threshold_at,
        "threshold_learn": LEARN_ROUTE_LEARN,
        "threshold_review": LEARN_ROUTE_REVIEW,
        "min_sample": CALIBRATION_MIN_SAMPLE,
        "target_success": CALIBRATION_TARGET_SUCCESS,
        "score_distribution": dist,
        "note": "high_precision 越高说明分类器评分越可信；rejected 占比高 = 评分被污染，应下调分类阈值。"
                "confidence_verdict=MISCALIBRATED 时阈值不可信（先校准置信度再改阈值），"
                "口径见 confidence_calibration（分母含 failed）。",
    }


# ── 自动回复白名单 ──────────────────────────────────

def whitelist_list() -> list[dict]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM reply_whitelist ORDER BY kind, name, id DESC"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def whitelist_add(kind: str, wxid: str, name: str = "", role: str = "member") -> int:
    conn = get_conn()
    cur = conn.execute(
        "INSERT OR IGNORE INTO reply_whitelist (kind, wxid, name, role) VALUES (?, ?, ?, ?)",
        (kind, wxid, name, role)
    )
    if not cur.lastrowid:  # 已存在则更新名字+kind（不覆盖 role）
        conn.execute("UPDATE reply_whitelist SET name=?, kind=? WHERE wxid=?",
                     (name, kind, wxid))
    conn.commit()
    rid = cur.lastrowid
    conn.close()
    return rid


def whitelist_remove(item_id: int):
    conn = get_conn()
    conn.execute("DELETE FROM reply_whitelist WHERE id = ?", (item_id,))
    conn.commit()
    conn.close()


def whitelist_roles() -> dict[str, str]:
    """给策略引擎用：所有白名单 wxid → role 的映射。"""
    conn = get_conn()
    rows = conn.execute("SELECT wxid, role FROM reply_whitelist").fetchall()
    conn.close()
    return {r["wxid"]: (r["role"] or "member") for r in rows}


def whitelist_set_role(item_id: int, role: str):
    """修改白名单条目的角色：'admin' 或 'member'。"""
    conn = get_conn()
    conn.execute("UPDATE reply_whitelist SET role=? WHERE id=?", (role, item_id))
    conn.commit()
    conn.close()


# ── 自动回复队列 ────────────────────────────────────

def reply_find_by_msg_id(msg_id) -> dict | None:
    """按 msg_id 查找已入队条目。

    PAP 授权层③ 幂等基础设施：同一 source message 只应产生一条待处理回复。
    加固「回滚 orch_state.json 后重放 drain」场景（重放不应产生第二条待审批条目）。
    """
    if not msg_id:
        return None
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM reply_queue WHERE msg_id = ? ORDER BY id LIMIT 1", (msg_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def reply_add(msg_id, talker, sender, sender_name, kind, incoming,
              draft, action, reason, status="pending") -> int:
    # 幂等守卫（PAP Operations 唯一 ID）：msg_id>0 的真实消息不重复入队，
    # 返回已存在条目 id 而非新建。msg_id=0/None 是「无消息来源」哨兵（手工/测试路径），不去重。
    if msg_id:
        existing = reply_find_by_msg_id(msg_id)
        if existing:
            return existing["id"]
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO reply_queue
           (msg_id, talker, sender, sender_name, kind, incoming, draft, action, reason, status)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (msg_id, talker, sender, sender_name, kind, incoming, draft, action, reason, status)
    )
    conn.commit()
    rid = cur.lastrowid
    conn.close()
    return rid


def reply_list(status: str = "", limit: int = 100) -> list[dict]:
    conn = get_conn()
    if status:
        rows = conn.execute(
            "SELECT * FROM reply_queue WHERE status = ? ORDER BY id DESC LIMIT ?",
            (status, limit)
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM reply_queue ORDER BY id DESC LIMIT ?", (limit,)
        ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def reply_get(item_id: int) -> dict | None:
    conn = get_conn()
    row = conn.execute("SELECT * FROM reply_queue WHERE id = ?", (item_id,)).fetchone()
    conn.close()
    return dict(row) if row else None


def reply_update(item_id: int, status: str, draft: str | None = None):
    conn = get_conn()
    if status in ("sent", "auto_sent", "failed"):
        if draft is not None:
            conn.execute(
                "UPDATE reply_queue SET status=?, draft=?, sent_at=datetime('now','localtime') WHERE id=?",
                (status, draft, item_id))
        else:
            conn.execute(
                "UPDATE reply_queue SET status=?, sent_at=datetime('now','localtime') WHERE id=?",
                (status, item_id))
    else:
        if draft is not None:
            conn.execute("UPDATE reply_queue SET status=?, draft=? WHERE id=?",
                         (status, draft, item_id))
        else:
            conn.execute("UPDATE reply_queue SET status=? WHERE id=?", (status, item_id))
    conn.commit()
    conn.close()


# ── AI 聊天：会话 CRUD ────────────────────────────────

def conv_list() -> list[dict]:
    """列出所有会话，按 updated_at 倒序"""
    conn = get_conn()
    rows = conn.execute(
        "SELECT id, title, created_at, updated_at FROM conversations ORDER BY updated_at DESC"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def conv_create(title: str = "新对话") -> int:
    """新建会话，返回 id"""
    conn = get_conn()
    cur = conn.execute(
        "INSERT INTO conversations (title) VALUES (?)", (title,)
    )
    conn.commit()
    cid = cur.lastrowid
    conn.close()
    return cid


def conv_delete(conv_id: int):
    """删除会话（messages 通过 ON DELETE CASCADE 自动删除）"""
    conn = get_conn()
    conn.execute("DELETE FROM conversations WHERE id = ?", (conv_id,))
    conn.commit()
    conn.close()


def conv_update_title(conv_id: int, title: str):
    """更新会话标题"""
    conn = get_conn()
    conn.execute(
        "UPDATE conversations SET title = ? WHERE id = ?", (title, conv_id)
    )
    conn.commit()
    conn.close()


def conv_update_time(conv_id: int):
    """更新会话 updated_at 为当前时间"""
    conn = get_conn()
    conn.execute(
        "UPDATE conversations SET updated_at = datetime('now','localtime') WHERE id = ?",
        (conv_id,)
    )
    conn.commit()
    conn.close()


def conv_get(conv_id: int) -> dict | None:
    """获取单个会话"""
    conn = get_conn()
    row = conn.execute(
        "SELECT id, title, created_at, updated_at FROM conversations WHERE id = ?",
        (conv_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


# ── AI 聊天：消息 CRUD ────────────────────────────────

def msg_list(conv_id: int) -> list[dict]:
    """获取某会话的所有消息，按 created_at 正序"""
    conn = get_conn()
    rows = conn.execute(
        "SELECT id, conversation_id, role, content, created_at FROM messages WHERE conversation_id = ? ORDER BY created_at ASC",
        (conv_id,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def msg_add(conv_id: int, role: str, content: str) -> int:
    """新增一条消息，返回 id。同时更新会话 updated_at。"""
    conn = get_conn()
    cur = conn.execute(
        "INSERT INTO messages (conversation_id, role, content) VALUES (?, ?, ?)",
        (conv_id, role, content)
    )
    conn.execute(
        "UPDATE conversations SET updated_at = datetime('now','localtime') WHERE id = ?",
        (conv_id,)
    )
    conn.commit()
    mid = cur.lastrowid
    conn.close()
    return mid


def msg_count(conv_id: int) -> int:
    """获取某会话的消息数"""
    conn = get_conn()
    cnt = conn.execute(
        "SELECT COUNT(*) FROM messages WHERE conversation_id = ?", (conv_id,)
    ).fetchone()[0]
    conn.close()
    return cnt


def msg_get(msg_id: int) -> dict | None:
    """获取单条消息"""
    conn = get_conn()
    row = conn.execute(
        "SELECT id, conversation_id, role, content, created_at FROM messages WHERE id = ?",
        (msg_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def msg_update(msg_id: int, content: str):
    """更新消息内容"""
    conn = get_conn()
    conn.execute(
        "UPDATE messages SET content = ? WHERE id = ?",
        (content, msg_id)
    )
    conn.commit()
    conn.close()


def msg_delete_since(conv_id: int, since_id: int):
    """删除指定会话中 id > since_id 的所有消息（用于重新生成时清空后续回复）"""
    conn = get_conn()
    conn.execute(
        "DELETE FROM messages WHERE conversation_id = ? AND id > ?",
        (conv_id, since_id)
    )
    conn.commit()
    conn.close()