"""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 ); """ # 旧库迁移：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]