Create-knowledge-find-decode/acquisition/store.py

"""SQLite 数据层：query 生成结果 + 真实搜索结果，统一进 data/app.db。

为什么不再用散落 json：量大了「一次性 fetch 整个 json」撑不住，要能分页/筛选/多 run 累积。
设计：
  queries        —— 一条生成的 query 一行；各打法字段不同，统一塞进 axes(json) 列
  search_results —— 一条「query × 平台」一行（把 douyin/weixin 嵌套拍平），方便按平台/成功筛
媒体文件仍躺 data/search/，库里只存相对路径（cover/video）。
纯 stdlib sqlite3，无新依赖。写入幂等：同 run 重导先清该 run。
"""
from __future__ import annotations

import json
import sqlite3
from pathlib import Path
from typing import Any, Iterable, Optional

ROOT = Path(__file__).resolve().parent.parent
DB_PATH = ROOT / "data" / "app.db"

_SCHEMA = """
CREATE TABLE IF NOT EXISTS queries (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id    TEXT NOT NULL,
    method    TEXT NOT NULL,          -- 打法名（实质 × 创作阶段 × 需求 等）
    query     TEXT NOT NULL,
    axes      TEXT NOT NULL DEFAULT '{}',  -- 该打法的正交轴 json（实质/形式/阶段…）
    created_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_queries_run    ON queries(run_id);
CREATE INDEX IF NOT EXISTS ix_queries_method ON queries(method);

CREATE TABLE IF NOT EXISTS search_results (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id    TEXT NOT NULL,
    method    TEXT NOT NULL,
    query     TEXT NOT NULL,
    platform  TEXT NOT NULL,          -- douyin / weixin
    ok        INTEGER NOT NULL,       -- 1=搜到且有媒体, 0=失败/空
    title     TEXT,
    url       TEXT,
    cover     TEXT,                   -- /data/search/... 相对路径
    video     TEXT,                   -- 抖音才有
    extra     TEXT NOT NULL DEFAULT '{}',  -- nick/error 等附加 json
    created_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_sr_run      ON search_results(run_id);
CREATE INDEX IF NOT EXISTS ix_sr_platform ON search_results(platform);
CREATE INDEX IF NOT EXISTS ix_sr_method   ON search_results(method);

CREATE TABLE IF NOT EXISTS runs (
    run_id    TEXT PRIMARY KEY,
    kind      TEXT NOT NULL,          -- queries / search
    note      TEXT,
    created_at INTEGER NOT NULL
);

-- 帖子级「创作知识 / 非创作知识」分类（按 url，一帖判一次）。acquisition/classify.py 写入。
CREATE TABLE IF NOT EXISTS post_class (
    url       TEXT PRIMARY KEY,
    is_creation INTEGER,              -- 1=创作知识 / 0=非
    reason    TEXT,
    knowledge TEXT,                   -- 提取出的完整创作知识（创作帖才有）
    points    TEXT,                   -- 结构化知识点 json（按卡片/分段）
    created_at INTEGER NOT NULL
);

-- 430 query 真实采集：run / query×platform job / item / AI 判断
CREATE TABLE IF NOT EXISTS creation_search_runs (
    run_id       TEXT PRIMARY KEY,
    status       TEXT NOT NULL,
    total_queries INTEGER NOT NULL DEFAULT 0,
    note         TEXT,
    started_at   INTEGER NOT NULL,
    finished_at  INTEGER
);

CREATE TABLE IF NOT EXISTS creation_search_jobs (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id        TEXT NOT NULL,
    query         TEXT NOT NULL,
    platform      TEXT NOT NULL,
    status        TEXT NOT NULL,
    attempts      INTEGER NOT NULL DEFAULT 0,
    searched_count INTEGER NOT NULL DEFAULT 0,
    display_count INTEGER NOT NULL DEFAULT 0,
    error         TEXT,
    started_at    INTEGER,
    finished_at   INTEGER,
    UNIQUE(run_id, query, platform)
);
CREATE INDEX IF NOT EXISTS ix_csj_query ON creation_search_jobs(query);
CREATE INDEX IF NOT EXISTS ix_csj_run_platform ON creation_search_jobs(run_id, platform);

CREATE TABLE IF NOT EXISTS creation_search_items (
    id             INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id         TEXT NOT NULL,
    query          TEXT NOT NULL,
    platform       TEXT NOT NULL,
    rank           INTEGER NOT NULL,
    source_id      TEXT,
    url            TEXT,
    title          TEXT,
    author         TEXT,
    body_text      TEXT,
    cover_url      TEXT,
    image_urls     TEXT NOT NULL DEFAULT '[]',
    video_url      TEXT,
    media_json     TEXT NOT NULL DEFAULT '{}',
    raw_json       TEXT NOT NULL DEFAULT '{}',
    is_displayable INTEGER NOT NULL DEFAULT 0,
    error          TEXT,
    created_at     INTEGER NOT NULL,
    UNIQUE(run_id, query, platform, rank)
);
CREATE INDEX IF NOT EXISTS ix_csi_query ON creation_search_items(query);
CREATE INDEX IF NOT EXISTS ix_csi_run_platform ON creation_search_items(run_id, platform);

CREATE TABLE IF NOT EXISTS creation_item_classifications (
    item_id        INTEGER PRIMARY KEY,
    is_creation    INTEGER,
    reason         TEXT,
    knowledge      TEXT,
    prompt_version TEXT,
    error          TEXT,
    classified_at  INTEGER NOT NULL,
    FOREIGN KEY(item_id) REFERENCES creation_search_items(id) ON DELETE CASCADE
);
"""

# 旧库迁移：post_class 早期没有 knowledge/points 列，补上（已存在则忽略）
_MIGRATIONS = [
    "ALTER TABLE post_class ADD COLUMN knowledge TEXT",
    "ALTER TABLE post_class ADD COLUMN points TEXT",
]


def connect(db_path: Path | str = DB_PATH) -> sqlite3.Connection:
    """打开（必要时建）库，建表，行按 dict 取。busy_timeout 让多进程并发写不报锁。"""
    p = Path(db_path)
    p.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(p), timeout=30)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA busy_timeout=30000")   # 撞锁等 30s 而非报错（与微信补下载并发安全）
    conn.executescript(_SCHEMA)
    for sql in _MIGRATIONS:
        try:
            conn.execute(sql)
        except sqlite3.OperationalError:
            pass                                 # 列已存在
    return conn


def _now(ts: Optional[int]) -> int:
    # 调用方传入时间戳（脚本侧用 int(time.time())）；缺省 0，避免本模块碰 Date/random。
    return int(ts) if ts is not None else 0


def _record_run(conn: sqlite3.Connection, run_id: str, kind: str, note: str, ts: int) -> None:
    conn.execute(
        "INSERT OR REPLACE INTO runs(run_id, kind, note, created_at) VALUES(?,?,?,?)",
        (run_id, kind, note, ts),
    )


# ---- 写入 ----------------------------------------------------------------

# 哪些键是「轴」（除 query 外的结构化字段），存进 axes blob
_NON_AXIS = {"query"}


def insert_queries(conn: sqlite3.Connection, run_id: str, method: str,
                   items: Iterable[dict], *, ts: Optional[int] = None) -> int:
    """写一批生成的 query。items=[{query, <各轴>...}]。同 (run_id,method) 先清后写（幂等）。"""
    t = _now(ts)
    conn.execute("DELETE FROM queries WHERE run_id=? AND method=?", (run_id, method))
    rows = []
    for it in items:
        q = (it or {}).get("query")
        if not q:
            continue
        axes = {k: v for k, v in it.items() if k not in _NON_AXIS}
        rows.append((run_id, method, q, json.dumps(axes, ensure_ascii=False), t))
    conn.executemany(
        "INSERT INTO queries(run_id, method, query, axes, created_at) VALUES(?,?,?,?,?)", rows)
    _record_run(conn, run_id, "queries", method, t)
    conn.commit()
    return len(rows)


def _flatten_search(rec: dict) -> list[dict]:
    """把一条 {method, query, douyin/weixin/xiaohongshu:{ok:[...],error}} 拍成「每个结果一行」。
    某渠道有结果 → 每个结果一行 ok=1；空/失败 → 一行 ok=0 记 error（保留「搜过但无结果」状态）。
    只处理记录里实际出现的渠道（向后兼容只有抖音/微信的旧记录）。"""
    out = []
    method, query = rec.get("method", ""), rec.get("query", "")
    for platform in ("douyin", "weixin", "xiaohongshu"):
        if platform not in rec:
            continue
        v = rec.get(platform) or {}
        hits = v.get("ok") or []
        if hits:
            for h in hits:
                extra = {k: h[k] for k in ("nick", "images", "body_text") if k in h}
                out.append({"platform": platform, "ok": 1, "title": h.get("title"),
                            "url": h.get("url"), "cover": h.get("cover"),
                            "video": h.get("video"), "extra": extra})
        else:
            out.append({"platform": platform, "ok": 0, "extra": {"error": v.get("error") or "未搜到"}})
    return [{**r, "method": method, "query": query} for r in out]


def insert_search_results(conn: sqlite3.Connection, run_id: str,
                          records: Iterable[dict], *, ts: Optional[int] = None) -> int:
    """写一批真实搜索结果。records 为 run_search.py 产出的嵌套结构，按平台拍平入库。"""
    t = _now(ts)
    conn.execute("DELETE FROM search_results WHERE run_id=?", (run_id,))
    rows = []
    for rec in records:
        for r in _flatten_search(rec):
            rows.append((run_id, r["method"], r["query"], r["platform"], r["ok"],
                         r.get("title"), r.get("url"), r.get("cover"), r.get("video"),
                         json.dumps(r.get("extra", {}), ensure_ascii=False), t))
    conn.executemany(
        "INSERT INTO search_results(run_id, method, query, platform, ok, title, url, "
        "cover, video, extra, created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)", rows)
    _record_run(conn, run_id, "search", f"{len(rows)} rows", t)
    conn.commit()
    return len(rows)


# ---- 读取（分页 + 筛选）---------------------------------------------------

def _page(conn: sqlite3.Connection, base: str, where: list[str], params: list[Any],
          page: int, size: int) -> dict:
    """通用分页：返回 {total, page, size, items}。"""
    size = max(1, min(int(size), 200))
    page = max(1, int(page))
    clause = (" WHERE " + " AND ".join(where)) if where else ""
    total = conn.execute(f"SELECT COUNT(*) FROM ({base}{clause})", params).fetchone()[0]
    rows = conn.execute(f"{base}{clause} ORDER BY id LIMIT ? OFFSET ?",
                        params + [size, (page - 1) * size]).fetchall()
    return {"total": total, "page": page, "size": size, "items": [dict(r) for r in rows]}


def list_queries(conn: sqlite3.Connection, *, method: Optional[str] = None,
                 run_id: Optional[str] = None, page: int = 1, size: int = 30) -> dict:
    where, params = [], []
    if method:
        where.append("method=?"); params.append(method)
    if run_id:
        where.append("run_id=?"); params.append(run_id)
    res = _page(conn, "SELECT * FROM queries", where, params, page, size)
    for it in res["items"]:
        it["axes"] = json.loads(it.get("axes") or "{}")
    return res


def list_search(conn: sqlite3.Connection, *, platform: Optional[str] = None,
                method: Optional[str] = None, ok: Optional[bool] = None,
                run_id: Optional[str] = None, query: Optional[str] = None,
                page: int = 1, size: int = 30) -> dict:
    where, params = [], []
    if platform:
        where.append("platform=?"); params.append(platform)
    if method:
        where.append("method=?"); params.append(method)
    if ok is not None:
        where.append("ok=?"); params.append(1 if ok else 0)
    if run_id:
        where.append("run_id=?"); params.append(run_id)
    if query:
        where.append("query=?"); params.append(query)   # 按 query 文本取该条的全部搜索结果
    res = _page(conn, "SELECT * FROM search_results", where, params, page, size)
    cm = class_map(conn, {it.get("url") for it in res["items"] if it.get("url")})
    for it in res["items"]:
        it["extra"] = json.loads(it.get("extra") or "{}")
        it["cls"] = cm.get(it.get("url"))                # 创作知识分类（无则 None=未分类）
    return res


# ---- 帖子级创作知识分类 ----------------------------------------------------

def get_unclassified_posts(conn: sqlite3.Connection) -> list[dict]:
    """库里有结果、但还没分类的去重帖子（按 url）。带 title/body_text/本地图片，供 classify 用。"""
    rows = conn.execute(
        "SELECT url, platform, title, cover, extra FROM search_results "
        "WHERE ok=1 AND url IS NOT NULL AND url NOT IN (SELECT url FROM post_class) "
        "GROUP BY url"
    ).fetchall()
    out = []
    for r in rows:
        extra = json.loads(r["extra"] or "{}")
        imgs = extra.get("images") or ([r["cover"]] if r["cover"] else [])
        out.append({"url": r["url"], "platform": r["platform"], "title": r["title"] or "",
                    "body_text": extra.get("body_text", ""), "images": imgs})
    return out


def upsert_class(conn: sqlite3.Connection, url: str, is_creation: int, reason: str, ts: int,
                 knowledge: str = "", points: str = "") -> None:
    conn.execute(
        "INSERT OR REPLACE INTO post_class(url, is_creation, reason, knowledge, points, created_at) "
        "VALUES(?,?,?,?,?,?)", (url, is_creation, reason, knowledge, points, ts))
    conn.commit()


def posts_to_classify(conn: sqlite3.Connection, platforms: list[str]) -> list[dict]:
    """指定平台的全部去重帖（ok=1），带 title/body_text/本地图/本地视频，供（重）分类。
    不管是否已分类——上层 upsert 覆盖，便于换更忠实的判法重判。"""
    qs = ",".join("?" * len(platforms))
    rows = conn.execute(
        f"SELECT url, platform, title, cover, video, extra FROM search_results "
        f"WHERE ok=1 AND url IS NOT NULL AND platform IN ({qs}) GROUP BY url", platforms
    ).fetchall()
    out = []
    for r in rows:
        e = json.loads(r["extra"] or "{}")
        imgs = e.get("images") or ([r["cover"]] if r["cover"] else [])
        out.append({"url": r["url"], "platform": r["platform"], "title": r["title"] or "",
                    "body_text": e.get("body_text", ""), "images": imgs, "video": r["video"]})
    return out


def class_map(conn: sqlite3.Connection, urls) -> dict:
    """{url: {is_creation, reason, knowledge, points}}，给取数时挂分类 + 提取的知识点。"""
    urls = [u for u in urls if u]
    if not urls:
        return {}
    qs = ",".join("?" * len(urls))
    rows = conn.execute(
        f"SELECT url, is_creation, reason, knowledge, points FROM post_class WHERE url IN ({qs})", urls
    ).fetchall()
    out = {}
    for r in rows:
        out[r["url"]] = {"is_creation": r["is_creation"], "reason": r["reason"],
                         "knowledge": r["knowledge"] or "",
                         "points": json.loads(r["points"]) if r["points"] else []}
    return out


def class_counts(conn: sqlite3.Connection) -> dict:
    y = conn.execute("SELECT COUNT(*) FROM post_class WHERE is_creation=1").fetchone()[0]
    n = conn.execute("SELECT COUNT(*) FROM post_class WHERE is_creation=0").fetchone()[0]
    return {"creation": y, "non_creation": n}


# ---- 微信正文补下载 --------------------------------------------------------

def weixin_urls_missing_body(conn: sqlite3.Connection) -> list[str]:
    """所有微信帖（distinct url, ok=1）中 extra 还没补正文(body_text)的，供补下载。"""
    rows = conn.execute(
        "SELECT url, extra FROM search_results WHERE platform='weixin' AND ok=1 "
        "AND url IS NOT NULL GROUP BY url"
    ).fetchall()
    return [r["url"] for r in rows if not json.loads(r["extra"] or "{}").get("body_text")]


def update_post_content(conn: sqlite3.Connection, url: str, body_text: str,
                        images: list[str]) -> None:
    """把正文 + 本地图片写回该 url 的所有行的 extra（其余字段保留）。"""
    for r in conn.execute("SELECT id, extra FROM search_results WHERE url=?", (url,)).fetchall():
        e = json.loads(r["extra"] or "{}")
        e["body_text"] = body_text
        if images:
            e["images"] = images
        conn.execute("UPDATE search_results SET extra=? WHERE id=?",
                     (json.dumps(e, ensure_ascii=False), r["id"]))
    conn.commit()


def search_summary(conn: sqlite3.Connection) -> dict:
    """每条 query 搜到几个结果：{query文本: {total, ok}}。给列表页判断按钮显不显示、显几个。"""
    rows = conn.execute(
        "SELECT query, COUNT(*) AS total, SUM(ok) AS ok_n FROM search_results GROUP BY query"
    ).fetchall()
    return {r["query"]: {"total": r["total"], "ok": r["ok_n"] or 0} for r in rows}


def list_runs(conn: sqlite3.Connection) -> list[dict]:
    rows = conn.execute("SELECT * FROM runs ORDER BY created_at DESC, run_id").fetchall()
    return [dict(r) for r in rows]


def list_methods(conn: sqlite3.Connection, table: str = "queries") -> list[str]:
    """某表里有哪些 method（给前端筛选下拉）。"""
    if table not in ("queries", "search_results"):
        raise ValueError(table)
    rows = conn.execute(f"SELECT DISTINCT method FROM {table} ORDER BY method").fetchall()
    return [r[0] for r in rows]


# ---- 430 query 真实采集 ----------------------------------------------------

CREATION_PLATFORMS = ("xiaohongshu", "weixin", "douyin")


def create_creation_run(conn: sqlite3.Connection, run_id: str, *,
                        total_queries: int, note: str = "", ts: Optional[int] = None,
                        status: str = "running") -> None:
    t = _now(ts)
    conn.execute(
        "INSERT OR REPLACE INTO creation_search_runs"
        "(run_id, status, total_queries, note, started_at, finished_at) VALUES(?,?,?,?,?,NULL)",
        (run_id, status, int(total_queries), note, t),
    )
    conn.commit()


def finish_creation_run(conn: sqlite3.Connection, run_id: str, *,
                        status: str = "finished", ts: Optional[int] = None) -> None:
    conn.execute(
        "UPDATE creation_search_runs SET status=?, finished_at=? WHERE run_id=?",
        (status, _now(ts), run_id),
    )
    conn.commit()


def creation_run_exists(conn: sqlite3.Connection, run_id: str) -> bool:
    row = conn.execute(
        "SELECT 1 FROM creation_search_runs WHERE run_id=? LIMIT 1", (run_id,)
    ).fetchone()
    return row is not None


def ensure_creation_job(conn: sqlite3.Connection, run_id: str, query: str, platform: str,
                        *, ts: Optional[int] = None) -> int:
    t = _now(ts)
    conn.execute(
        "INSERT OR IGNORE INTO creation_search_jobs"
        "(run_id, query, platform, status, attempts, started_at) VALUES(?,?,?,?,?,?)",
        (run_id, query, platform, "pending", 0, t),
    )
    conn.commit()
    row = conn.execute(
        "SELECT id FROM creation_search_jobs WHERE run_id=? AND query=? AND platform=?",
        (run_id, query, platform),
    ).fetchone()
    return int(row["id"])


def update_creation_job(conn: sqlite3.Connection, run_id: str, query: str, platform: str,
                        *, status: str, attempts: Optional[int] = None,
                        searched_count: Optional[int] = None,
                        display_count: Optional[int] = None,
                        error: Optional[str] = None, ts: Optional[int] = None) -> None:
    ensure_creation_job(conn, run_id, query, platform, ts=ts)
    sets = ["status=?"]
    params: list[Any] = [status]
    if attempts is not None:
        sets.append("attempts=?"); params.append(int(attempts))
    if searched_count is not None:
        sets.append("searched_count=?"); params.append(int(searched_count))
    if display_count is not None:
        sets.append("display_count=?"); params.append(int(display_count))
    if error is not None:
        sets.append("error=?"); params.append(error)
    if status in ("done", "failed", "partial"):
        sets.append("finished_at=?"); params.append(_now(ts))
    elif status == "running":
        sets.append("started_at=?"); params.append(_now(ts))
    params.extend([run_id, query, platform])
    conn.execute(
        f"UPDATE creation_search_jobs SET {', '.join(sets)} "
        "WHERE run_id=? AND query=? AND platform=?",
        params,
    )
    conn.commit()


def creation_job_is_done(conn: sqlite3.Connection, run_id: str, query: str, platform: str,
                         *, display_limit: int = 5) -> bool:
    row = conn.execute(
        "SELECT status, display_count FROM creation_search_jobs "
        "WHERE run_id=? AND query=? AND platform=?",
        (run_id, query, platform),
    ).fetchone()
    return bool(row and row["status"] == "done" and int(row["display_count"] or 0) >= display_limit)


def upsert_creation_item(conn: sqlite3.Connection, *, run_id: str, query: str, platform: str,
                         rank: int, source_id: str = "", url: str = "", title: str = "",
                         author: str = "", body_text: str = "", cover_url: str = "",
                         image_urls: Optional[list[str]] = None, video_url: str = "",
                         media: Optional[dict] = None, raw: Optional[dict] = None,
                         is_displayable: bool = False, error: str = "",
                         ts: Optional[int] = None) -> int:
    t = _now(ts)
    conn.execute(
        "INSERT INTO creation_search_items"
        "(run_id, query, platform, rank, source_id, url, title, author, body_text, cover_url, "
        "image_urls, video_url, media_json, raw_json, is_displayable, error, created_at) "
        "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) "
        "ON CONFLICT(run_id, query, platform, rank) DO UPDATE SET "
        "source_id=excluded.source_id, url=excluded.url, title=excluded.title, "
        "author=excluded.author, body_text=excluded.body_text, cover_url=excluded.cover_url, "
        "image_urls=excluded.image_urls, video_url=excluded.video_url, media_json=excluded.media_json, "
        "raw_json=excluded.raw_json, is_displayable=excluded.is_displayable, "
        "error=excluded.error, created_at=excluded.created_at",
        (
            run_id, query, platform, int(rank), source_id, url, title, author, body_text,
            cover_url, json.dumps(image_urls or [], ensure_ascii=False), video_url,
            json.dumps(media or {}, ensure_ascii=False), json.dumps(raw or {}, ensure_ascii=False),
            1 if is_displayable else 0, error, t,
        ),
    )
    conn.commit()
    row = conn.execute(
        "SELECT id FROM creation_search_items WHERE run_id=? AND query=? AND platform=? AND rank=?",
        (run_id, query, platform, int(rank)),
    ).fetchone()
    return int(row["id"])


def upsert_creation_classification(conn: sqlite3.Connection, item_id: int,
                                   is_creation: Optional[int], reason: str = "",
                                   knowledge: str = "", prompt_version: str = "",
                                   error: str = "", ts: Optional[int] = None) -> None:
    conn.execute(
        "INSERT INTO creation_item_classifications"
        "(item_id, is_creation, reason, knowledge, prompt_version, error, classified_at) "
        "VALUES(?,?,?,?,?,?,?) "
        "ON CONFLICT(item_id) DO UPDATE SET "
        "is_creation=excluded.is_creation, reason=excluded.reason, knowledge=excluded.knowledge, "
        "prompt_version=excluded.prompt_version, error=excluded.error, classified_at=excluded.classified_at",
        (int(item_id), is_creation, reason, knowledge, prompt_version, error, _now(ts)),
    )
    conn.commit()


def latest_creation_run_id(conn: sqlite3.Connection) -> Optional[str]:
    row = conn.execute(
        "SELECT run_id FROM creation_search_runs ORDER BY started_at DESC, run_id DESC LIMIT 1"
    ).fetchone()
    return row["run_id"] if row else None


def creation_search_summary(conn: sqlite3.Connection, run_id: Optional[str] = None) -> dict:
    rid = run_id or latest_creation_run_id(conn)
    if not rid:
        return {"run_id": None, "queries": {}}
    def default_query_entry() -> dict:
        return {
            "platforms": {},
            "done": 0,
            "failed": 0,
            "display_count": 0,
            "imgtext_creation_count": 0,
            "imgtext_classified_count": 0,
            "imgtext_total_count": 0,
            "imgtext_target_count": 10,
        }

    rows = conn.execute(
        "SELECT query, platform, status, attempts, searched_count, display_count, error "
        "FROM creation_search_jobs WHERE run_id=? ORDER BY query, platform",
        (rid,),
    ).fetchall()
    out: dict[str, dict] = {}
    for r in rows:
        q = r["query"]
        ent = out.setdefault(q, default_query_entry())
        p = {
            "status": r["status"],
            "attempts": r["attempts"],
            "searched_count": r["searched_count"],
            "display_count": r["display_count"],
            "error": r["error"],
        }
        ent["platforms"][r["platform"]] = p
        ent["display_count"] += r["display_count"] or 0
        if r["status"] == "done":
            ent["done"] += 1
        if r["status"] == "failed":
            ent["failed"] += 1
    count_rows = conn.execute(
        "SELECT i.query, COUNT(*) AS total_count, "
        "SUM(CASE WHEN c.item_id IS NOT NULL THEN 1 ELSE 0 END) AS classified_count, "
        "SUM(CASE WHEN c.is_creation=1 THEN 1 ELSE 0 END) AS creation_count "
        "FROM creation_search_items i "
        "LEFT JOIN creation_item_classifications c ON c.item_id=i.id "
        "WHERE i.run_id=? AND i.is_displayable=1 "
        "AND i.platform IN ('xiaohongshu', 'weixin') "
        "GROUP BY i.query",
        (rid,),
    ).fetchall()
    for r in count_rows:
        ent = out.setdefault(r["query"], default_query_entry())
        ent["imgtext_creation_count"] = int(r["creation_count"] or 0)
        ent["imgtext_classified_count"] = int(r["classified_count"] or 0)
        ent["imgtext_total_count"] = int(r["total_count"] or 0)
    return {"run_id": rid, "queries": out}


def get_creation_query_detail(conn: sqlite3.Connection, query: str, *,
                              run_id: Optional[str] = None, display_limit: int = 5) -> dict:
    rid = run_id or latest_creation_run_id(conn)
    platforms = {p: {"status": "not_started", "items": [], "error": None} for p in CREATION_PLATFORMS}
    if not rid:
        return {"run_id": None, "query": query, "platforms": platforms}
    job_rows = conn.execute(
        "SELECT platform, status, attempts, searched_count, display_count, error "
        "FROM creation_search_jobs WHERE run_id=? AND query=?",
        (rid, query),
    ).fetchall()
    for r in job_rows:
        platforms[r["platform"]].update({
            "status": r["status"],
            "attempts": r["attempts"],
            "searched_count": r["searched_count"],
            "display_count": r["display_count"],
            "error": r["error"],
        })
    item_rows = conn.execute(
        "SELECT i.*, c.is_creation, c.reason AS class_reason, c.knowledge, "
        "c.prompt_version, c.error AS class_error, c.classified_at "
        "FROM creation_search_items i "
        "LEFT JOIN creation_item_classifications c ON c.item_id=i.id "
        "WHERE i.run_id=? AND i.query=? AND i.is_displayable=1 "
        "ORDER BY i.platform, i.rank",
        (rid, query),
    ).fetchall()
    per_platform_counts = {p: 0 for p in CREATION_PLATFORMS}
    for r in item_rows:
        p = r["platform"]
        if p not in platforms or per_platform_counts[p] >= display_limit:
            continue
        item = dict(r)
        item["image_urls"] = json.loads(item.get("image_urls") or "[]")
        item["media"] = json.loads(item.pop("media_json") or "{}")
        item.pop("raw_json", None)  # 原始回包很大；详情 API 只返回前端展示需要的字段
        item["classification"] = {
            "is_creation": item.pop("is_creation"),
            "reason": item.pop("class_reason") or "",
            "knowledge": item.pop("knowledge") or "",
            "prompt_version": item.pop("prompt_version") or "",
            "error": item.pop("class_error") or "",
            "classified_at": item.pop("classified_at"),
        }
        platforms[p]["items"].append(item)
        per_platform_counts[p] += 1
    return {"run_id": rid, "query": query, "platforms": platforms}


def creation_items_to_classify(conn: sqlite3.Connection, *, run_id: Optional[str] = None,
                               platforms: Optional[list[str]] = None,
                               retry_failed: bool = True,
                               limit: Optional[int] = None) -> list[dict]:
    rid = run_id or latest_creation_run_id(conn)
    if not rid:
        return []
    where = ["i.run_id=?", "i.is_displayable=1"]
    params: list[Any] = [rid]
    if platforms:
        qs = ",".join("?" * len(platforms))
        where.append(f"i.platform IN ({qs})")
        params.extend(platforms)
    if retry_failed:
        where.append("(c.item_id IS NULL OR c.is_creation IS NULL)")
    else:
        where.append("c.item_id IS NULL")
    sql = (
        "SELECT i.id, i.platform, i.title, i.body_text, i.image_urls, i.video_url "
        "FROM creation_search_items i "
        "LEFT JOIN creation_item_classifications c ON c.item_id=i.id "
        f"WHERE {' AND '.join(where)} ORDER BY i.platform, i.id"
    )
    if limit is not None and int(limit) > 0:
        sql += " LIMIT ?"
        params.append(int(limit))
    rows = conn.execute(sql, params).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        d["image_urls"] = json.loads(d.get("image_urls") or "[]")
        out.append(d)
    return out