feat(classify)：改用忠实的「读懂+创作闸」重判 + 微信补下载完整正文

不再用截取的定义+偏向，改成直接复用 pipeline 的判断（import 调用，不改 skill/creation_knowledge）：
- 图文(小红书/微信)：GeminiExtractor（完整 prompts/extract.txt，带【卡片N】喂全部图）→ is_empty 即创作闸。
- 视频(抖音)：extract_video（完整 prompts/extract_video.txt，整段视频原生喂 Gemini）→ is_empty
  （看完视频没提到创作知识段落=非创作）。大视频先 ffmpeg 压 480p（保留口播音轨）再 base64。
- search.fetch_weixin_detail + backfill_weixin：微信正文/内文图实测不需 token，补下到本地写回 extra。
- store：busy_timeout 并发安全、posts_to_classify、微信补下载辅助。classify 可传平台名重判。
并发跑、与微信补下载并行。实测理由来自真实内容，小红书/抖音重判质量明显提升。

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

acquisition/backfill_weixin.py

@@ -0,0 +1,68 @@
+"""补下载微信公众号帖子的【完整正文 + 内文图】到本地，写回 app.db。
+
+微信搜索只回标题+封面，正文没下；分类只看标题会判不准。本脚本对库里所有微信帖逐个拉详情
+（wei_xin/detail，实测不需 token）→ 下全部内文图到 data/search/weixin/<hash>/img_*.jpg
+→ 把 body_text + images 写回 search_results.extra。之后微信就和小红书一样是「完整内容」。
+限流：详情接口 10~12s 随机一次（图片下载走 CDN，不计入闸）。已补过的(extra 有 body_text)跳过，可重跑。
+用法：PYTHONPATH=. CK_ENV_FILE=.env python -m acquisition.backfill_weixin
+"""
+from __future__ import annotations
+
+import hashlib
+import time
+from pathlib import Path
+
+from acquisition import store
+from acquisition.crawler import RateLimiter
+from acquisition.search import fetch_weixin_detail
+from core.config import Settings
+from creation_knowledge.integrations.video_extract import _default_download
+
+ROOT = Path(__file__).resolve().parent.parent
+DATA = ROOT / "data"
+GATE_MIN, GATE_MAX = 10.0, 12.0     # 微信详情接口两次调用随机间隔（秒）
+
+
+def _dl(url: str, dst: Path, public: str):
+    try:
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        dst.write_bytes(_default_download(url, "weixin"))
+        return public
+    except Exception:
+        return None
+
+
+def main() -> None:
+    settings = Settings.from_env()
+    conn = store.connect()
+    urls = store.weixin_urls_missing_body(conn)
+    if not urls:
+        print("微信帖正文都已补过。"); conn.close(); return
+    gate = RateLimiter(min_interval_seconds=GATE_MIN, max_interval_seconds=GATE_MAX)
+    total, ok, fail, img_total = len(urls), 0, 0, 0
+    print(f"待补正文的微信帖 {total} 个（详情 {GATE_MIN}~{GATE_MAX}s 一次）...")
+    for i, url in enumerate(urls, 1):
+        try:
+            body, img_urls = fetch_weixin_detail(url, settings=settings, rate_limiter=gate)
+        except Exception as exc:
+            fail += 1
+            print(f"  [{i}/{total}] 详情失败: {str(exc)[:50]}")
+            continue
+        h = hashlib.md5(url.encode()).hexdigest()[:16]
+        base, pub = DATA / "search" / "weixin" / h, f"/data/search/weixin/{h}"
+        imgs = []
+        for j, u in enumerate(img_urls):
+            local = _dl(u, base / f"img_{j}.jpg", f"{pub}/img_{j}.jpg")
+            if local:
+                imgs.append(local)
+        store.update_post_content(conn, url, body, imgs)
+        ok += 1
+        img_total += len(imgs)
+        if i % 10 == 0 or i == total:
+            print(f"  [{i}/{total}] 正文 {len(body)} 字 / 图 {len(imgs)} 张（累计 {ok} 成 / {fail} 败 / {img_total} 图）")
+    conn.close()
+    print(f"完成：{ok} 帖补到正文，{fail} 失败，共下 {img_total} 张内文图。")
+
+
+if __name__ == "__main__":
+    main()

acquisition/classify.py

@@ -1,106 +1,144 @@
-"""把已下载的帖子分成「创作知识 / 非创作知识」（只判定，不解构、不提知识点）。
-
-复用 prompts/extract.txt 里的「创作知识」定义（只读，绝不改 skill/creation_knowledge）。
-方案 2：标题 + 正文 + 封面/帖内图一起喂 Gemini（OpenRouter 多模态），并发 10 + 失败重试。
-结果写 data/app.db 的 post_class 表（按 url），前端按 url 显示「创作知识 / 非创作知识」角标。
-不碰视频/帖子文件本身，不入 ingest。重跑只补未分类的（幂等）。
+"""帖子级「创作知识 / 非创作知识」分类——忠实复用 pipeline 的「读懂 + 创作闸」，不简化提示词。
+
+图文（小红书/微信）：用 creation_knowledge 的 GeminiExtractor（**完整 prompts/extract.txt**，
+  带【卡片N】把全部图喂给 Gemini）→ ExtractedContent.is_empty 即创作闸（true=非创作）。
+视频（抖音）：用 extract_video（**完整 prompts/extract_video.txt**，整段视频原生喂 Gemini）→
+  is_empty（=看完视频没提到任何创作知识段落）。大视频先 ffmpeg 压到 480p（保留口播音轨）再 base64。
+结果按 url 写 app.db 的 post_class（upsert 覆盖，可重判）。并发；与微信补下载并行安全（busy_timeout）。
+只读 prompts、import 调用 pipeline 函数——不改 skill/creation_knowledge，不解构、不入 ingest。
 用法：PYTHONPATH=. CK_ENV_FILE=.env python -m acquisition.classify
 """
 from __future__ import annotations
 
 import base64
 import concurrent.futures as cf
-import json
+import subprocess
+import sys
+import tempfile
 import time
 from pathlib import Path
 
-import httpx
-
 from acquisition import store
 from core.config import Settings
+from core.models import Card, Post
+from creation_knowledge.integrations.extractor import GeminiExtractor
+from creation_knowledge.integrations.video_extract import extract_video
 
 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 ""
+PLATFORMS = ["xiaohongshu", "douyin"]    # 本轮重判平台（微信等正文下完再单独跑）
+IMG_WORKERS = 8                          # 图文并发
+VID_WORKERS = 3                          # 视频并发（含 ffmpeg 压制，别太高）
+MAX_CARDS = 12                           # 图文最多送几张图（与 extractor 对齐）
+COMPRESS_OVER_MB = 12                    # 视频超过此大小先压再喂
+COMPRESS_H = 480
 
 
-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。"""
+def _data_url(public_path: str):
     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 _imgtext_post(p: dict) -> Post:
+    cards = []
+    for i, im in enumerate((p.get("images") or [])[:MAX_CARDS], start=1):
+        u = _data_url(im)
+        if u:
+            cards.append(Card(index=i, kind="image", url=u))
+    return Post(id=f"{p['platform']}_cls", platform=p["platform"], url=p["url"],
+                content_id=p["url"], title=p.get("title", ""),
+                body_text=p.get("body_text", ""), cards=cards)
+
+
+def _compress(mp4: Path) -> Path:
+    """大视频压到 480p（保留音轨——extract_video 要听口播）→ 临时 mp4；失败回原文件。"""
+    out = Path(tempfile.gettempdir()) / f"ck_{mp4.parent.name}_480.mp4"
+    try:
+        import imageio_ffmpeg
+        ff = imageio_ffmpeg.get_ffmpeg_exe()
+        subprocess.run([ff, "-y", "-i", str(mp4), "-vf", f"scale=-2:{COMPRESS_H}",
+                        "-c:v", "libx264", "-crf", "30", "-preset", "veryfast",
+                        "-c:a", "aac", "-b:a", "64k", str(out)],
+                       capture_output=True, timeout=300)
+        if out.exists() and out.stat().st_size > 0:
+            return out
+    except Exception:
+        pass
+    return mp4
+
+
+def classify_imgtext(p: dict, extractor: GeminiExtractor) -> tuple:
+    ec = extractor.extract(_imgtext_post(p))           # 完整 extract.txt → is_empty
+    if ec.is_empty:
+        return 0, "判为非创作：无可迁移的创作方法"
+    return 1, (ec.text or "")[:60]
+
+
+def classify_video(p: dict, settings: Settings) -> tuple:
+    rel = p.get("video") or ""
+    mp4 = ROOT / rel.lstrip("/")
+    if not rel or not mp4.exists():
+        return None, "无本地视频"
+    use = _compress(mp4) if mp4.stat().st_size > COMPRESS_OVER_MB * 1048576 else mp4
+    post = Post(id="douyin_cls", platform="douyin", url=p["url"], content_id=p["url"],
+                title=p.get("title", ""), video_urls=["local"])
+    try:
+        ec = extract_video(post, settings=settings, video_path=str(use), timeout=300)  # 完整 extract_video.txt
+    finally:
+        if use != mp4:
+            try:
+                use.unlink()
+            except Exception:
+                pass
+    if ec.is_empty:
+        return 0, "判为非创作：视频无创作知识段落"
+    return 1, (ec.text or "")[:60]
+
+
+def _safe(fn, *a) -> tuple:
+    try:
+        return fn(*a)
+    except Exception as exc:
+        return None, f"判定失败: {str(exc)[:60]}"
 
 
 def main() -> None:
     settings = Settings.from_env()
-    system = SYS_TMPL.format(definition=_definition())
+    extractor = GeminiExtractor.from_env()
     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}
+    platforms = sys.argv[1:] or PLATFORMS    # 可传平台名重判，如：... -m acquisition.classify weixin
+    posts = store.posts_to_classify(conn, platforms)
+    imgs = [p for p in posts if p["platform"] != "douyin"]
+    vids = [p for p in posts if p["platform"] == "douyin"]
+    total = len(posts)
+    print(f"忠实重判：图文 {len(imgs)}（并发{IMG_WORKERS}）+ 抖音视频 {len(vids)}（并发{VID_WORKERS}，大视频先压）")
+    ts = int(time.time())
+    done = {"n": 0, "fail": 0}
+
+    def _write(p, res):
+        ic, reason = res
+        if ic is None:
+            done["fail"] += 1
+        else:
+            store.upsert_class(conn, p["url"], ic, reason, ts)
+        done["n"] += 1
+        if done["n"] % 20 == 0:
+            print(f"  {done['n']}/{total}（失败 {done['fail']}）")
+
+    with cf.ThreadPoolExecutor(IMG_WORKERS) as ex:
+        futs = {ex.submit(_safe, classify_imgtext, p, extractor): p for p in imgs}
         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}）")
+            _write(futs[fut], fut.result())
+    with cf.ThreadPoolExecutor(VID_WORKERS) as ex:
+        futs = {ex.submit(_safe, classify_video, p, settings): p for p in vids}
+        for fut in cf.as_completed(futs):
+            _write(futs[fut], fut.result())
+
     c = store.class_counts(conn)
     conn.close()
-    print(f"完成：创作知识 {c['creation']} / 非创作知识 {c['non_creation']}（本轮失败 {fail}，可重跑补判）")
+    print(f"完成（含历史微信）：创作知识 {c['creation']} / 非创作知识 {c['non_creation']}"
+          f"（本轮失败 {done['fail']}，可重跑补判）")
 
 
 if __name__ == "__main__":

acquisition/search.py

@@ -185,6 +185,46 @@ def search_weixin(
             client.close()
 
 
+def fetch_weixin_detail(
+    url: str,
+    *,
+    settings: Optional[Settings] = None,
+    http_client: Any = None,
+    rate_limiter: Optional[RateLimiter] = None,
+    env_file: str = ".env",
+) -> tuple[str, list[str]]:
+    """微信公众号文章详情：content_link → (body_text, image_urls)。实测不需 token。失败抛 SearchError。"""
+    settings = settings or Settings.from_env(env_file)
+    rate_limiter = rate_limiter or RateLimiter()
+    owns_client = http_client is None
+    client = http_client or httpx.Client()
+    try:
+        rate_limiter.wait("weixin_detail")
+        api = urljoin(settings.crawler_base_url, "/crawler/wei_xin/detail")
+        try:
+            resp = client.post(api, json={"content_link": url},
+                               headers={"Content-Type": "application/json"},
+                               timeout=settings.crawler_timeout)
+            resp.raise_for_status()
+            data = resp.json()
+        except httpx.HTTPError as exc:
+            raise SearchError(f"http_error: {exc}") from exc
+        except ValueError as exc:
+            raise SearchError("bad_json") from exc
+        if data.get("code") not in (0, "0"):
+            raise SearchError(f"business_error: code={data.get('code')} msg={data.get('msg')}")
+        inner = ((data.get("data") or {}).get("data")) or {}
+        imgs = []
+        for im in inner.get("image_url_list") or []:
+            u = im.get("image_url") if isinstance(im, dict) else im
+            if u:
+                imgs.append(u)
+        return inner.get("body_text") or "", imgs
+    finally:
+        if owns_client:
+            client.close()
+
+
 # ---------- 小红书搜索（封面/标题/作者已在搜索回包的 note_card 里，无需 detail）----------
 def parse_xiaohongshu(response: Any, *, limit: int = 5) -> list[dict]:
     """小红书搜索回包 {code,data:{data:[{id, note_card:{image_list,display_title,desc,user,type}}]}}

acquisition/store.py

@@ -65,11 +65,12 @@ CREATE TABLE IF NOT EXISTS post_class (
 
 
 def connect(db_path: Path | str = DB_PATH) -> sqlite3.Connection:
-    """打开（必要时建）库，建表，行按 dict 取。"""
+    """打开（必要时建）库，建表，行按 dict 取。busy_timeout 让多进程并发写不报锁。"""
     p = Path(db_path)
     p.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(str(p))
+    conn = sqlite3.connect(str(p), timeout=30)
     conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA busy_timeout=30000")   # 撞锁等 30s 而非报错（与微信补下载并发安全）
     conn.executescript(_SCHEMA)
     return conn
 
@@ -226,6 +227,23 @@ def upsert_class(conn: sqlite3.Connection, url: str, is_creation: int, reason: s
     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}}，给取数时挂分类。"""
     urls = [u for u in urls if u]
@@ -242,6 +260,30 @@ def class_counts(conn: sqlite3.Connection) -> dict:
     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(