"""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 os import sqlite3 from pathlib import Path from typing import Any, Iterable, Optional from core.config import load_env_file 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 resolve_db_path(db_path: Path | str | None = None, *, env_file: str | Path | None = None) -> Path: """Resolve the SQLite path at call time. `DB_PATH` stays as the default new-data location for compatibility, while CK_SQLITE_PATH can point old demo/API runs at legacy_data/app.db. """ if db_path: p = Path(db_path) else: source = load_env_file(env_file or os.getenv("CK_ENV_FILE", ".env")) raw = os.getenv("CK_SQLITE_PATH") or source.get("CK_SQLITE_PATH") or str(DB_PATH) p = Path(raw) return p if p.is_absolute() else ROOT / p def connect(db_path: Path | str | None = None) -> sqlite3.Connection: """打开（必要时建）库，建表，行按 dict 取。busy_timeout 让多进程并发写不报锁。""" p = resolve_db_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