宜博 / pixelphone

"""本地 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'",
    ]
    for sql in migrations:
        try:
            conn.execute(sql)
        except sqlite3.OperationalError:
            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.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()
    placeholders = ",".join(["?"] * len(dedup_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, [*dedup_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, dedup_keys).fetchall()
    conn.close()
    result = {}
    for row in rows:
        d = dict(row)
        dk = d.pop("dedup_key")
        if dk and dk not in result:
            result[dk] = d
    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 article_get_unsummarized_new(limit: int = 20) -> list[dict]:
    """Articles without summaries (from new tables), 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, 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
           WHERE a.id NOT IN (
               SELECT DISTINCT a2.id FROM articles a2
               INNER JOIN daily_reports r ON r.article_url = a2.url AND r.report_date = 'summary'
           )
           GROUP BY a.id
           ORDER BY latest_share DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


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 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_add(msg_id, talker, sender, sender_name, kind, incoming,
              draft, action, reason, status="pending") -> int:
    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()