feat(classify)：帖子级「创作知识 / 非创作知识」分类（多模态+并发，前端角标）

acquisition/classify.py：复用 prompts/extract.txt 的创作知识定义（只读），把标题+正文+
封面/帖内图喂 Gemini（OpenRouter 多模态）做二分类，并发 10 + 失败重试，写 app.db 的
post_class 表（按 url，幂等可重跑）。store 取数时挂分类；详情页每张卡显示创作知识/非创作
知识角标（带理由 tooltip）。只判定、不解构、不入 ingest，不碰 skill/creation_knowledge。
实测 386 帖 0 失败：创作 257 / 非创作 129。

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

acquisition/classify.py

@@ -0,0 +1,107 @@
+"""把已下载的帖子分成「创作知识 / 非创作知识」（只判定，不解构、不提知识点）。
+
+复用 prompts/extract.txt 里的「创作知识」定义（只读，绝不改 skill/creation_knowledge）。
+方案 2：标题 + 正文 + 封面/帖内图一起喂 Gemini（OpenRouter 多模态），并发 10 + 失败重试。
+结果写 data/app.db 的 post_class 表（按 url），前端按 url 显示「创作知识 / 非创作知识」角标。
+不碰视频/帖子文件本身，不入 ingest。重跑只补未分类的（幂等）。
+用法：PYTHONPATH=. CK_ENV_FILE=.env python -m acquisition.classify
+"""
+from __future__ import annotations
+
+import base64
+import concurrent.futures as cf
+import json
+import time
+from pathlib import Path
+
+import httpx
+
+from acquisition import store
+from core.config import Settings
+
+ROOT = Path(__file__).resolve().parent.parent
+EXTRACT_PROMPT = ROOT / "prompts" / "extract.txt"
+WORKERS = 10          # 并发路数（实测线性提速；过高易被 OpenRouter 限流）
+MAX_IMG = 2           # 每帖最多喂几张图（控延迟/体积）
+RETRIES = 3           # 失败重试次数（含 429 限流退避）
+
+
+def _definition() -> str:
+    """复用 extract.txt 的 <什么算创作知识> 段（同一判据，只读不改）。"""
+    txt = EXTRACT_PROMPT.read_text("utf-8")
+    a, b = txt.find("<什么算创作知识>"), txt.find("</什么算创作知识>")
+    return txt[a:b + len("</什么算创作知识>")] if a >= 0 and b >= 0 else ""
+
+
+SYS_TMPL = """你是创作知识分类器。给你一篇帖子的平台/标题/正文和图片，判断它是不是「创作知识」。
+{definition}
+只输出 JSON：{{"is_creation": true/false, "reason": "一句话理由（≤25字）"}}。
+范围内但拿不准 → is_creation=true；明确越界（游戏玩法/产品运营/做菜本身/纯工具操作/纯作品展示）→ false。
+注意：视频帖可能只有标题+封面、信息有限，据现有内容审慎判断。"""
+
+
+def _img_data_url(public_path: str):
+    """本地公开路径 /data/search/... → base64 data URL；不存在返回 None。"""
+    fs = ROOT / public_path.lstrip("/")
+    if not fs.exists():
+        return None
+    return "data:image/jpeg;base64," + base64.b64encode(fs.read_bytes()).decode()
+
+
+def classify_one(post: dict, settings: Settings, system: str) -> dict:
+    """单帖多模态判定，带重试。返回 {is_creation:1/0/None, reason}。None=判定失败（留待重跑）。"""
+    user_text = (f"平台：{post.get('platform')}\n标题：{post.get('title') or ''}\n"
+                 f"正文：{(post.get('body_text') or '')[:600]}")
+    content = [{"type": "text", "text": user_text}]
+    for img in (post.get("images") or [])[:MAX_IMG]:
+        url = _img_data_url(img)
+        if url:
+            content.append({"type": "image_url", "image_url": {"url": url}})
+    messages = [{"role": "system", "content": system}, {"role": "user", "content": content}]
+    api = settings.openrouter_base_url.rstrip("/") + "/chat/completions"
+    headers = {"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"}
+    payload = {"model": settings.llm_model, "messages": messages,
+               "response_format": {"type": "json_object"}}
+    last = ""
+    for attempt in range(RETRIES):
+        try:
+            resp = httpx.post(api, headers=headers, json=payload, timeout=90)
+            if resp.status_code == 200:
+                d = json.loads(resp.json()["choices"][0]["message"]["content"])
+                return {"is_creation": 1 if d.get("is_creation") else 0,
+                        "reason": str(d.get("reason", ""))[:50]}
+            last = f"http {resp.status_code}"
+        except Exception as exc:
+            last = str(exc)[:50]
+        time.sleep(1.5 * (attempt + 1))     # 退避后重试
+    return {"is_creation": None, "reason": f"判定失败: {last}"}
+
+
+def main() -> None:
+    settings = Settings.from_env()
+    system = SYS_TMPL.format(definition=_definition())
+    conn = store.connect()
+    posts = store.get_unclassified_posts(conn)
+    if not posts:
+        print("没有待分类帖子（都判过了）。"); conn.close(); return
+    print(f"待分类 {len(posts)} 帖（并发 {WORKERS}，每帖≤{MAX_IMG} 图）...")
+    ts, done, fail = int(time.time()), 0, 0
+    with cf.ThreadPoolExecutor(max_workers=WORKERS) as ex:
+        futs = {ex.submit(classify_one, p, settings, system): p for p in posts}
+        for fut in cf.as_completed(futs):
+            p = futs[fut]
+            r = fut.result()
+            if r["is_creation"] is None:
+                fail += 1
+            else:
+                store.upsert_class(conn, p["url"], r["is_creation"], r["reason"], ts)
+            done += 1
+            if done % 25 == 0:
+                print(f"  {done}/{len(posts)}（失败 {fail}）")
+    c = store.class_counts(conn)
+    conn.close()
+    print(f"完成：创作知识 {c['creation']} / 非创作知识 {c['non_creation']}（本轮失败 {fail}，可重跑补判）")
+
+
+if __name__ == "__main__":
+    main()

acquisition/store.py

@@ -53,6 +53,14 @@ CREATE TABLE IF NOT EXISTS runs (
     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,
+    created_at INTEGER NOT NULL
+);
 """
 
 
@@ -187,11 +195,53 @@ def list_search(conn: sqlite3.Connection, *, platform: Optional[str] = None,
     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) -> None:
+    conn.execute("INSERT OR REPLACE INTO post_class(url, is_creation, reason, created_at) VALUES(?,?,?,?)",
+                 (url, is_creation, reason, ts))
+    conn.commit()
+
+
+def class_map(conn: sqlite3.Connection, urls) -> dict:
+    """{url: {is_creation, reason}}，给取数时挂分类。"""
+    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 FROM post_class WHERE url IN ({qs})", urls).fetchall()
+    return {r["url"]: {"is_creation": r["is_creation"], "reason": r["reason"]} for r in rows}
+
+
+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 search_summary(conn: sqlite3.Connection) -> dict:
     """每条 query 搜到几个结果：{query文本: {total, ok}}。给列表页判断按钮显不显示、显几个。"""
     rows = conn.execute(

acquisition/web/app/dist/index.html

@@ -4,8 +4,8 @@
     <meta charset="UTF-8" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>创作知识 · 找帖子</title>
-    <script type="module" crossorigin src="/app/assets/index-C7iYtVJR.js"></script>
-    <link rel="stylesheet" crossorigin href="/app/assets/index-CIKg5rSi.css">
+    <script type="module" crossorigin src="/app/assets/index-CLGP1gjo.js"></script>
+    <link rel="stylesheet" crossorigin href="/app/assets/index-D_lgEh3q.css">
   </head>
   <body>
     <div id="root"></div>

acquisition/web/app/src/pages/QueryDetail.jsx

@@ -43,6 +43,14 @@ export default function QueryDetail({ query }) {
   )
 }
 
+function ClsBadge({ cls }) {
+  if (!cls || cls.is_creation === null) return <span className="cls none">未分类</span>
+  const yes = cls.is_creation === 1
+  return <span className={`cls ${yes ? 'yes' : 'no'}`} title={cls.reason || ''}>
+    {yes ? '创作知识' : '非创作知识'}
+  </span>
+}
+
 function Result({ r, cls }) {
   const imgs = (r.extra?.images && r.extra.images.length ? r.extra.images : (r.cover ? [r.cover] : []))
   return (
@@ -52,6 +60,7 @@ function Result({ r, cls }) {
         : imgs.length > 1
           ? <div className="gallery">{imgs.map((u, i) => <img key={i} className="gimg" src={u} alt="" />)}</div>
           : imgs[0] && <img className="cover" src={imgs[0]} alt="" />}
+      <ClsBadge cls={r.cls} />
       <div className="t">{r.title || '(无标题)'}</div>
       {r.extra?.body_text && <div className="bt">{r.extra.body_text}</div>}
       <div className="meta">

acquisition/web/app/src/styles.css

@@ -96,6 +96,12 @@ td.q { font-weight: 600; }
 .lane .pl.wx { color: #15a36a; }
 .lane video, .lane img.cover { width: 100%; max-height: 260px; object-fit: cover; border-radius: 8px; background: #000; display: block; }
 .lane .t { font-size: 12.5px; margin: 7px 0 3px; line-height: 1.45; }
+/* 创作知识 / 非创作知识 分类角标 */
+.cls { display: inline-block; font-size: 11px; font-weight: 700; border-radius: 6px;
+  padding: 2px 8px; margin: 8px 0 2px; }
+.cls.yes { background: #e6f7ee; color: var(--ok); }
+.cls.no { background: #fbeaea; color: var(--bad); }
+.cls.none { background: #f0f0f3; color: var(--muted); font-weight: 600; }
 /* 小红书完整帖：图片画廊 + 正文（链接会被反爬封，整帖存本地） */
 .gallery { display: grid; grid-template-columns: repeat(3, 1fr); gap: 4px; }
 .gallery .gimg { width: 100%; aspect-ratio: 1; object-fit: cover; border-radius: 6px; background: #f0f0f3; }