feat(api)：SQLite 数据层 + 分页/筛选取数接口

acquisition/store.py：query 与真实搜索结果统一进 data/app.db（搜索按「每结果一行」拍平），
分页/筛选/按 query 取/汇总。acquisition/web_api.py：/api/{runs,methods,queries,search,
search/summary}。creation_knowledge/api.py 挂上 /api 路由，为前端供数（取代「前端读死 json」）。
scripts/import_to_db.py 一键把现有 json 导库。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

acquisition/store.py

@@ -0,0 +1,210 @@
+"""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
+);
+"""
+
+
+def connect(db_path: Path | str = DB_PATH) -> sqlite3.Connection:
+    """打开（必要时建）库，建表，行按 dict 取。"""
+    p = Path(db_path)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(str(p))
+    conn.row_factory = sqlite3.Row
+    conn.executescript(_SCHEMA)
+    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:{ok:[...],error}, weixin:{ok:[...],error}} 拍成「每个结果一行」。
+    某渠道有结果 → 每个结果一行 ok=1；空/失败 → 一行 ok=0 记 error（保留「搜过但无结果」状态）。"""
+    out = []
+    method, query = rec.get("method", ""), rec.get("query", "")
+    for platform in ("douyin", "weixin"):
+        v = rec.get(platform) or {}
+        hits = v.get("ok") or []
+        if hits:
+            for h in hits:
+                extra = {k: h[k] for k in ("nick",) 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)
+    for it in res["items"]:
+        it["extra"] = json.loads(it.get("extra") or "{}")
+    return res
+
+
+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]

acquisition/web_api.py

@@ -0,0 +1,69 @@
+"""前端取数接口：从 app.db 分页/筛选返回 query 与真实搜索结果。
+
+挂到现有 FastAPI（creation_knowledge/api.py）。前端从「读死 json」改成调这些接口：
+  GET /api/runs                              有哪些批次
+  GET /api/methods?table=queries|search      某表的打法清单（筛选下拉）
+  GET /api/queries?method=&run_id=&page=&size=
+  GET /api/search?platform=&method=&ok=&run_id=&page=&size=
+每个请求开一个 sqlite 连接（轻、够用）；连接出错时接口照常返回空集，不 500。
+"""
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import APIRouter, Query
+
+from acquisition import store
+
+router = APIRouter(prefix="/api")
+
+
+@router.get("/runs")
+def runs() -> dict:
+    conn = store.connect()
+    try:
+        return {"runs": store.list_runs(conn)}
+    finally:
+        conn.close()
+
+
+@router.get("/methods")
+def methods(table: str = Query("queries")) -> dict:
+    conn = store.connect()
+    try:
+        tbl = "search_results" if table == "search" else "queries"
+        return {"methods": store.list_methods(conn, tbl)}
+    finally:
+        conn.close()
+
+
+@router.get("/queries")
+def queries(method: Optional[str] = None, run_id: Optional[str] = None,
+            page: int = 1, size: int = 30) -> dict:
+    conn = store.connect()
+    try:
+        return store.list_queries(conn, method=method, run_id=run_id, page=page, size=size)
+    finally:
+        conn.close()
+
+
+@router.get("/search")
+def search(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:
+    conn = store.connect()
+    try:
+        return store.list_search(conn, platform=platform, method=method, ok=ok,
+                                 run_id=run_id, query=query, page=page, size=size)
+    finally:
+        conn.close()
+
+
+@router.get("/search/summary")
+def search_summary() -> dict:
+    """{query文本: {total, ok}}——列表页据此给每条 query 显示「搜索结果」按钮与计数。"""
+    conn = store.connect()
+    try:
+        return {"summary": store.search_summary(conn)}
+    finally:
+        conn.close()

creation_knowledge/api.py

@@ -12,14 +12,13 @@ from pathlib import Path
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import RedirectResponse
 from fastapi.staticfiles import StaticFiles
 
-from creation_knowledge.config import Settings
+from core.config import Settings
 
 _ROOT = Path(__file__).resolve().parent.parent
-WEB_DIR = _ROOT / "web"
-SKILL_EX_DIR = _ROOT / "创作知识提取-skill" / "extraction"  # ②③ 提示词真源
-READ_PROMPT_DIR = _ROOT / "prompts"  # ① 读懂提示词真源
+WEB_DIR = _ROOT / "acquisition" / "web"  # 前端工程目录（仅 app/ 下的 Vite 产物）
 
 
 def _env_file() -> str:
@@ -46,12 +45,16 @@ _data_dir = Path(_settings().data_dir or "data")
 _data_dir.mkdir(parents=True, exist_ok=True)
 app.mount("/data", StaticFiles(directory=str(_data_dir)), name="data")
 
-# 提示词真源（前端「看提示词」直接 fetch，不再写死副本，避免漂移）
-if SKILL_EX_DIR.exists():
-    app.mount("/prompt-skill", StaticFiles(directory=str(SKILL_EX_DIR)), name="prompt-skill")
-if READ_PROMPT_DIR.exists():
-    app.mount("/prompt-read", StaticFiles(directory=str(READ_PROMPT_DIR)), name="prompt-read")
+# 取数接口（分页/筛选）
+from acquisition.web_api import router as api_router  # noqa: E402
+app.include_router(api_router)
 
-# 单页前端挂在最后（catch-all），不影响 /frames、/data、/prompt-*
-if WEB_DIR.exists():
-    app.mount("/", StaticFiles(directory=str(WEB_DIR), html=True), name="web")
+# 根路径直接送到前端
+@app.get("/")
+def _root() -> RedirectResponse:
+    return RedirectResponse(url="/app/")
+
+# 唯一前端：Vite+React 构建产物挂在 /app（旧静态看板已下线）
+_APP_DIST = WEB_DIR / "app" / "dist"
+if _APP_DIST.exists():
+    app.mount("/app", StaticFiles(directory=str(_APP_DIST), html=True), name="app")

scripts/import_to_db.py

@@ -0,0 +1,42 @@
+"""把现有 data/queries/demo.json + search_results.json 导入 data/app.db。
+
+一次性迁移 / 重导用。run_id 用固定名（demo / search-demo），重导幂等覆盖。
+用法：PYTHONPATH=. python scripts/import_to_db.py
+"""
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+
+from acquisition import store
+
+ROOT = Path(__file__).resolve().parent.parent
+DEMO = ROOT / "data" / "queries" / "demo.json"
+SEARCH = ROOT / "data" / "queries" / "search_results.json"
+
+
+def main() -> None:
+    ts = int(time.time())
+    conn = store.connect()
+    try:
+        if DEMO.exists():
+            demo = json.loads(DEMO.read_text("utf-8"))
+            for key in ("tactic1", "tactic2", "tactic3", "tactic4"):
+                t = demo.get(key) or {}
+                items = t.get("items") or []
+                if not items:
+                    continue
+                n = store.insert_queries(conn, "demo", t.get("name", key), items, ts=ts)
+                print(f"queries[{t.get('name', key)}]: {n} 条")
+        if SEARCH.exists():
+            recs = json.loads(SEARCH.read_text("utf-8"))
+            n = store.insert_search_results(conn, "search-demo", recs, ts=ts)
+            print(f"search_results: {n} 行（{len(recs)} 条 query × 平台拍平）")
+        print("runs:", [r["run_id"] for r in store.list_runs(conn)])
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()

tests/test_store.py

@@ -0,0 +1,74 @@
+"""SQLite 数据层离线测：写入幂等 + 拍平 + 分页筛选（临时库，不碰 data/app.db）。"""
+from __future__ import annotations
+
+from acquisition import store
+
+
+def _db(tmp_path):
+    return store.connect(tmp_path / "t.db")
+
+
+def test_insert_queries_idempotent_and_axes(tmp_path):
+    c = _db(tmp_path)
+    items = [{"query": "q1", "实质": "军人", "阶段": "灵感"}, {"query": "q2", "形式": "排比"}]
+    assert store.insert_queries(c, "r1", "多轴", items, ts=100) == 2
+    # 重导同 (run,method) → 覆盖不叠加
+    assert store.insert_queries(c, "r1", "多轴", items, ts=100) == 2
+    res = store.list_queries(c, method="多轴")
+    assert res["total"] == 2
+    assert res["items"][0]["axes"] == {"实质": "军人", "阶段": "灵感"}   # query 不进 axes
+    assert "query" not in res["items"][0]["axes"]
+
+
+def test_insert_queries_skips_empty_query(tmp_path):
+    c = _db(tmp_path)
+    n = store.insert_queries(c, "r1", "m", [{"query": ""}, {"实质": "x"}, {"query": "ok"}], ts=1)
+    assert n == 1
+
+
+def test_search_flatten_topk_and_ok(tmp_path):
+    c = _db(tmp_path)
+    recs = [
+        {"method": "M", "query": "q1",
+         "douyin": {"ok": [{"title": "t", "url": "u", "cover": "/c.jpg", "video": "/v.mp4"},
+                           {"title": "t2", "url": "u2", "cover": "/c2.jpg", "video": "/v2.mp4"}],
+                    "error": None},
+         "weixin": {"ok": [{"title": "w", "url": "wu", "cover": "/wc.jpg", "nick": "号"}], "error": None}},
+        {"method": "M", "query": "q2",
+         "douyin": {"ok": [], "error": "未搜到"},
+         "weixin": {"ok": [], "error": "搜索失败: x"}},
+    ]
+    # q1: 2 抖音 + 1 微信 = 3 个 ok 行；q2: 抖音空 + 微信空 = 2 个 ok=0 行 → 共 5
+    assert store.insert_search_results(c, "s1", recs, ts=5) == 5
+    ok_dy = store.list_search(c, platform="douyin", ok=True)
+    assert ok_dy["total"] == 2                                   # q1 的两个视频
+    assert {i["video"] for i in ok_dy["items"]} == {"/v.mp4", "/v2.mp4"}
+    # 搜过但无结果 → ok=0 行带 error（按 query 取得到，按钮才显示「搜过·无结果」）
+    bad = store.list_search(c, platform="weixin", ok=False)
+    assert bad["total"] == 1 and "失败" in bad["items"][0]["extra"]["error"]
+    good_wx = store.list_search(c, platform="weixin", ok=True)
+    assert good_wx["items"][0]["extra"]["nick"] == "号"
+    # summary：q1 有 3 个结果，q2 有 0 个但被记为搜过（total=2 行）
+    s = store.search_summary(c)
+    assert s["q1"]["ok"] == 3 and s["q2"]["ok"] == 0 and s["q2"]["total"] == 2
+
+
+def test_pagination(tmp_path):
+    c = _db(tmp_path)
+    store.insert_queries(c, "r", "m", [{"query": f"q{i}"} for i in range(25)], ts=1)
+    p1 = store.list_queries(c, page=1, size=10)
+    p3 = store.list_queries(c, page=3, size=10)
+    assert p1["total"] == 25 and len(p1["items"]) == 10
+    assert len(p3["items"]) == 5            # 余 5 条
+    assert p1["items"][0]["query"] != p3["items"][0]["query"]
+
+
+def test_runs_and_methods(tmp_path):
+    c = _db(tmp_path)
+    store.insert_queries(c, "rq", "方法A", [{"query": "a"}], ts=2)
+    store.insert_search_results(c, "rs", [{"method": "方法B", "query": "b",
+                                           "douyin": None, "weixin": None}], ts=3)
+    runs = {r["run_id"]: r["kind"] for r in store.list_runs(c)}
+    assert runs == {"rq": "queries", "rs": "search"}
+    assert store.list_methods(c, "queries") == ["方法A"]
+    assert store.list_methods(c, "search_results") == ["方法B"]