feat(classify)：收紧创作知识判据到「自媒体内容创作」，显式排除应试/制作/学科

依据 8123 金标准 + 全量审计（约一半误判）+ 业界界定，把「创作知识」窄化为：教你创作
【自媒体/社媒内容】的可迁移方法（选题/钩子/脚本/结构/标题/封面/叙事/定位/涨粉）。
三条一票否决（命中即非创作）：① 应试/学术写作（申论/作文/范文素材/答题模板/论文/公文）
② 制作/工具操作（AI生成/提示词/参数/排版/剪辑/导出 = 制作知识）③ 学科知识/评论/作品本身。

- 新增 prompts/classify_imgtext.txt（图文收紧判据），重写 prompts/classify_video.txt 补三条排除。
- classify.py 图文改用收紧 prompt + 直连（不再用 pipeline 的 extract.txt），与视频统一走 _judge；
  两者都输出 {is_empty, reason, knowledge}，默认重判三平台。不改 skill/creation_knowledge。

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

acquisition/classify.py

@@ -1,12 +1,13 @@
-"""帖子级「创作知识 / 非创作知识」分类——忠实复用 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。
+判据已收紧到「**自媒体内容创作**」，并显式排除三类越界（见 prompts）：① 应试/学术写作
+② 制作/工具操作（=制作知识）③ 学科知识/评论/作品本身。两套真实提示词：
+  图文（小红书/微信）：prompts/classify_imgtext.txt（标题+正文+全部图喂 Gemini）
+  视频（抖音）：prompts/classify_video.txt（整段视频原生喂 Gemini，大视频先 ffmpeg 压 480p 保音轨）
+都输出 {is_empty, reason, knowledge}：is_empty 即创作闸；is_empty=false 时连带提炼出具体创作知识点。
 结果按 url 写 app.db 的 post_class（upsert 覆盖，可重判）。并发；与微信补下载并行安全（busy_timeout）。
-只读 prompts、import 调用 pipeline 函数——不改 skill/creation_knowledge，不解构、不入 ingest。
-用法：PYTHONPATH=. CK_ENV_FILE=.env python -m acquisition.classify
+只读 prompts、走 OpenRouter——不改 skill/creation_knowledge，不解构、不入 ingest。
+用法：PYTHONPATH=. CK_ENV_FILE=.env python -m acquisition.classify [平台名...]
 """
 from __future__ import annotations
 
@@ -23,15 +24,13 @@ import httpx
 
 from acquisition import store
 from core.config import Settings
-from core.models import Card, Post
 from core.prompts import load_prompt
-from creation_knowledge.integrations.extractor import GeminiExtractor
 
 ROOT = Path(__file__).resolve().parent.parent
-PLATFORMS = ["xiaohongshu", "douyin"]    # 本轮重判平台（微信等正文下完再单独跑）
+PLATFORMS = ["xiaohongshu", "weixin", "douyin"]   # 默认重判全部；可传平台名覆盖
 IMG_WORKERS = 8                          # 图文并发
 VID_WORKERS = 3                          # 视频并发（含 ffmpeg 压制，别太高）
-MAX_CARDS = 12                           # 图文最多送几张图（与 extractor 对齐）
+MAX_CARDS = 12                           # 图文最多送几张图
 COMPRESS_OVER_MB = 12                    # 视频超过此大小先压再喂
 COMPRESS_H = 480
 
@@ -43,19 +42,8 @@ def _data_url(public_path: str):
     return "data:image/jpeg;base64," + base64.b64encode(fs.read_bytes()).decode()
 
 
-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；失败回原文件。"""
+    """大视频压到 480p（保留口播音轨）→ 临时 mp4；失败回原文件。"""
     out = Path(tempfile.gettempdir()) / f"ck_{mp4.parent.name}_480.mp4"
     try:
         import imageio_ffmpeg
@@ -71,18 +59,44 @@ def _compress(mp4: Path) -> Path:
     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, "判为非创作：无可迁移的创作方法", "", ""
-    points = json.dumps([{"index": c.index, "content": c.content} for c in (ec.cards or [])],
-                        ensure_ascii=False)
-    return 1, (ec.text or "")[:60], (ec.text or ""), points   # reason 取开头，knowledge 存全文
+def _judge(messages: list, settings: Settings, timeout: float) -> tuple:
+    """调 Gemini（OpenRouter，强制 JSON）解析 {is_empty, reason, knowledge}，带重试。
+    返回 (is_creation 1/0/None, reason, knowledge, points)。"""
+    api = settings.openrouter_base_url.rstrip("/") + "/chat/completions"
+    headers = {"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"}
+    payload = {"model": settings.video_model, "messages": messages,
+               "response_format": {"type": "json_object"}}
+    last = ""
+    for attempt in range(3):
+        try:
+            resp = httpx.post(api, headers=headers, json=payload, timeout=timeout)
+            if resp.status_code == 200:
+                d = json.loads(resp.json()["choices"][0]["message"]["content"])
+                if bool(d.get("is_empty")):
+                    return 0, str(d.get("reason", ""))[:60], "", ""
+                return 1, str(d.get("reason", ""))[:60], str(d.get("knowledge", "") or ""), ""
+            last = f"http {resp.status_code}"
+        except Exception as exc:
+            last = str(exc)[:50]
+        time.sleep(2 * (attempt + 1))
+    return None, f"判定失败: {last}", "", ""
+
+
+def classify_imgtext(p: dict, settings: Settings) -> tuple:
+    """图文：标题+正文+全部图，用收紧的 classify_imgtext.txt 判 is_empty 并提取知识点。"""
+    user = [{"type": "text", "text": f"平台：{p.get('platform')}\n标题：{p.get('title', '')}\n"
+             f"正文：{(p.get('body_text') or '')[:1500]}\n（下附帖子图片，请一并看完）"}]
+    for im in (p.get("images") or [])[:MAX_CARDS]:
+        u = _data_url(im)
+        if u:
+            user.append({"type": "image_url", "image_url": {"url": u}})
+    messages = [{"role": "system", "content": load_prompt("classify_imgtext")},
+                {"role": "user", "content": user}]
+    return _judge(messages, settings, timeout=120)
 
 
 def classify_video(p: dict, settings: Settings) -> tuple:
-    """看完整段视频，用与图文同款的【严格创作知识判据】（classify_video.txt：两轴+越界+拦观点拔高）判 is_empty。
-    不再用 extract_video 的松「提炼」闸（它会把讲观点的视频拔高成创作知识）。"""
+    """视频：看完整段视频，用收紧的 classify_video.txt 判 is_empty 并提取知识点。"""
     rel = p.get("video") or ""
     mp4 = ROOT / rel.lstrip("/")
     if not rel or not mp4.exists():
@@ -99,26 +113,7 @@ def classify_video(p: dict, settings: Settings) -> tuple:
     messages = [{"role": "system", "content": load_prompt("classify_video")},
                 {"role": "user", "content": [{"type": "text", "text": "判断这条视频是不是创作知识。"},
                                              {"type": "video_url", "video_url": {"url": media}}]}]
-    api = settings.openrouter_base_url.rstrip("/") + "/chat/completions"
-    headers = {"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"}
-    payload = {"model": settings.video_model, "messages": messages,
-               "response_format": {"type": "json_object"}}
-    last = ""
-    for attempt in range(3):
-        try:
-            resp = httpx.post(api, headers=headers, json=payload, timeout=300)
-            if resp.status_code == 200:
-                d = json.loads(resp.json()["choices"][0]["message"]["content"])
-                is_empty = bool(d.get("is_empty"))
-                if is_empty:
-                    return 0, str(d.get("reason", ""))[:60], "", ""
-                kn = str(d.get("knowledge", "") or "")
-                return 1, str(d.get("reason", ""))[:60], kn, ""
-            last = f"http {resp.status_code}"
-        except Exception as exc:
-            last = str(exc)[:50]
-        time.sleep(2 * (attempt + 1))
-    return None, f"判定失败: {last}", "", ""
+    return _judge(messages, settings, timeout=300)
 
 
 def _safe(fn, *a) -> tuple:
@@ -130,14 +125,13 @@ def _safe(fn, *a) -> tuple:
 
 def main() -> None:
     settings = Settings.from_env()
-    extractor = GeminiExtractor.from_env()
     conn = store.connect()
-    platforms = sys.argv[1:] or PLATFORMS    # 可传平台名重判，如：... -m acquisition.classify weixin
+    platforms = sys.argv[1:] or PLATFORMS
     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}，大视频先压）")
+    print(f"收紧重判：图文 {len(imgs)}（并发{IMG_WORKERS}）+ 抖音视频 {len(vids)}（并发{VID_WORKERS}，大视频先压）")
     ts = int(time.time())
     done = {"n": 0, "fail": 0}
 
@@ -152,7 +146,7 @@ def main() -> None:
             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}
+        futs = {ex.submit(_safe, classify_imgtext, p, settings): p for p in imgs}
         for fut in cf.as_completed(futs):
             _write(futs[fut], fut.result())
     with cf.ThreadPoolExecutor(VID_WORKERS) as ex:
@@ -162,8 +156,7 @@ def main() -> None:
 
     c = store.class_counts(conn)
     conn.close()
-    print(f"完成（含历史微信）：创作知识 {c['creation']} / 非创作知识 {c['non_creation']}"
-          f"（本轮失败 {done['fail']}，可重跑补判）")
+    print(f"完成：创作知识 {c['creation']} / 非创作知识 {c['non_creation']}（本轮失败 {done['fail']}，可重跑补判）")
 
 
 if __name__ == "__main__":

prompts/classify_imgtext.txt

@@ -0,0 +1,27 @@
+你在判断一篇图文帖（小红书 / 微信公众号，含标题、正文、图片）是不是「创作知识」——能教别人"如何创作【自媒体 / 社媒内容】"的可迁移方法、原理、清单。请把图片也看完（知识常在图里）再判断。
+
+<什么算创作知识>
+创作知识 = 教你"怎么让一条**面向大众的自媒体内容**（图文 / 短视频 / 脚本 / 封面 / 标题）被点开、被看完、被转化、并把账号做大"的**可迁移方法**。典型：
+- 选题 / 钩子开头 / 脚本与结构 / 标题公式 / 封面设计 / 叙事与呈现 / 账号定位 / 起号涨粉 / 数据复盘。
+- 必须是**方法 / 原理 / 公式 / 清单**（能套用、能迁移），不是一篇具体作品、不是一份素材本身。
+</什么算创作知识>
+
+<一票否决·命中任一即判非创作（is_empty=true）>
+整篇主体只要落在下面任一类，就判**非创作**，别拔高、别勉强：
+1. 【应试 / 学术写作】申论、高考/中考作文、作文或申论范文、人物/作文**素材库**、答题模板、解题套路、考研/学术论文、公文 / 讲话稿 / 材料写作。
+   —— 目标是"拿分数 / 合规范"，不是"做面向大众的自媒体内容、争夺注意力"。即便都在"写"，也不算。
+2. 【制作 / 工具操作 = 制作知识，不是创作】用 AI / 某 App / 小程序 / 提示词 / 参数把内容生成或做出来（文生视频、AI 写文案/出图、提示词框架、参数设置），以及排版 / 秀米 / 剪辑 / 调色 / 导出等编辑器操作。
+   —— 这是"怎么把成品**做出来**（制作 / 执行）"，不是"怎么**构思出**更好的内容（创作）"。
+3. 【学科知识 / 评论 / 作品本身】只是在讲某学科知识（历史 / 政治 / 外交 / 经济本身）、输出某个观点 / 感悟 / 人生道理、一篇时政评论或一份具体作品、一份纯素材 / 范文合集。
+   —— 这是"内容 / 作品本身"，不是"教你怎么创作内容的方法"。
+</一票否决>
+
+<判定>
+- 口诀：教的是"怎么让面向大众的内容被点开 / 看完 / 转化 / 涨粉"的可迁移方法 → 算（is_empty=false）；"应试达标 / 用工具把成品做出来 / 某学科观点或作品本身 / 纯素材范文" → 不算（is_empty=true）。
+- 范围内但拿不准有没有方法 → 算（false）；但"是不是上面三类越界"要**果断**，明显越界就判非创作（true）。
+- 若 is_empty=false，把帖子里教的**具体创作知识点**忠实提炼出来（分条、不编造、不拔高）。
+
+只输出一个 JSON 对象：
+{"is_empty": true/false,
+ "reason": "一句话理由：是「教了哪种自媒体创作方法」，还是「属于应试/制作/学科/作品本身的哪一类」（≤30字）",
+ "knowledge": "is_empty=false 时：帖子里教的具体创作知识点全文（分条）；is_empty=true 时：空字符串"}

prompts/classify_video.txt

@@ -1,33 +1,25 @@
-你在判断一条短视频是不是「创作知识」——能教别人"如何创作内容"的方法、原理、认知或清单。请看完整段视频（听口播、看画面与字幕）再判断。只做判断，不提炼、不分段。
+你在判断一条短视频是不是「创作知识」——能教别人"如何创作【自媒体 / 社媒内容】"的可迁移方法、原理、清单。请看完整段视频（听口播、看画面与字幕）再判断。
 
 <什么算创作知识>
-能迁移、能复用、能教别人"怎么创作图文/视频内容"的方法、原理、清单。**判断要同时过两条轴，缺一不算：**
-
-【轴A·产出物是不是「内容」】这条知识帮你创作出来的成品，是不是 图文/视频/脚本/剧本/小说 这类**内容**？题材不限（美食/游戏/健身/人生感悟…都行），卡的是产出物：
-- 算：产出"关于某题材的**内容**"——美食视频拍法、游戏解说脚本、健身图文写法。
-- 不算：产出"那个题材**本身**"——一道菜的做法、一个游戏的玩法、一个产品、一套运营/选品策略。
-
-【轴B·是「创作」不是「制作」】教的是"怎么构思/选题/写/结构/呈现"（创作），还是"用哪个软件/AI、点哪个按钮把成品产出来"（制作/工具操作）？
-- 算（创作）：选题/构思方法、结构或排版公式、要素清单、为什么某种开头/呈现更抓人、镜头/构图/配色的设计判断。
-- 不算（制作）：打开某 App/AI → 输入 → 点生成 → 导出 这类工具操作；纯器材/剪辑/调色/导出。
-- 特别强调：「用某 AI/App 帮你生成内容成品」的教程也**不算**。
-
-【两轴 AND】两条都过才算；任一不过 → 不算。
+创作知识 = 教你"怎么让一条**面向大众的自媒体内容**（短视频 / 图文 / 脚本 / 封面 / 标题）被点开、被看完、被转化、并把账号做大"的**可迁移方法**。典型：
+- 选题 / 钩子开头 / 脚本与结构 / 标题公式 / 封面设计 / 叙事与呈现 / 账号定位 / 起号涨粉 / 数据复盘。
+- 必须是**方法 / 原理 / 公式 / 清单**（能套用、能迁移），不是一条具体作品、不是一份素材本身。
 </什么算创作知识>
 
-<最常见的误判·务必拦下>
-**一个人对着镜头讲他的人生观/价值观/人生感悟/某个道理/某个话题（如"人生的意义""如何看待 XX""我最有帮助的改变"），这是在表达「观点 / 内容本身」，不是在教别人"怎么创作内容"——判为不算（is_empty=true）。**
-- **不要拔高**：别把"一个人在表达观点/讲道理"说成"教你如何做观点输出/口播/治愈系内容"。除非视频里**明确在教可复用的创作方法**（选题怎么选、脚本怎么搭、开头怎么钩、结构怎么排、起号怎么做、文案怎么写），否则就是作品/观点本身。
-- 同理不算（作品/内容本身）：一段具体文案或作品的展示、纯口播抒情、纯叙事讲故事、纯知识科普（讲某领域知识但不教"怎么做成内容"）、励志鸡汤、个人经历/成长心得分享、新闻报道。
+<一票否决·命中任一即判非创作（is_empty=true）>
+整条视频主体只要落在下面任一类，就判**非创作**，别拔高、别勉强：
+1. 【一个人讲观点 / 道理 / 感悟 = 作品本身】对着镜头讲人生观 / 价值观 / 人生感悟 / 某个道理 / 某个话题（如"人生的意义""如何看待 XX""我最有帮助的改变"）——这是在表达观点 / 内容本身，不是教别人怎么创作内容。**不要**把"一个人在表达观点"拔高成"教你做观点输出 / 口播内容"。
+2. 【应试 / 学术写作】申论、高考/中考作文、作文或申论范文、人物/作文素材、答题模板、解题套路、考研/学术论文、公文 / 讲话稿。目标是拿分数 / 合规范，不是做自媒体内容。
+3. 【制作 / 工具操作 = 制作知识，不是创作】用 AI / 某 App / 小程序 / 提示词 / 参数把内容生成或做出来（文生视频、AI 写文案/出图、提示词框架），以及剪辑 / 调色 / 导出 / 排版等操作。这是"怎么把成品做出来（制作）"，不是"怎么构思出更好的内容（创作）"。
+4. 【学科知识 / 评论 / 作品本身】只是在讲某学科知识（历史 / 政治 / 外交本身）、一篇时政评论、纯叙事讲故事 / 抒情励志 / 个人经历分享、纯素材合集。是内容 / 作品本身，不是创作方法。
+</一票否决>
 
 <判定>
-- 视频里只要有哪怕一点"怎么**创作内容**"的可迁移方法（而非只是在讲观点/讲内容） → is_empty=false（算创作知识）。
-- 通篇只是讲观点 / 讲道理 / 讲某话题 / 展示作品 / 抒情励志 / 分享经历，没有教"怎么做内容"的方法 → is_empty=true（不算）。
-- "范围内但拿不准有没有方法" → is_empty=false；但"到底是讲观点还是教创作"要**果断**——明显是在讲观点 / 作品本身，就判 true，别用"拿不准"放进来。
-
-若 is_empty=false（算创作知识），还要把视频里教的**具体创作知识点**完整提炼出来（忠实视频内容、不编造、不拔高；分条列清：教了什么方法/原理/清单，按视频讲解顺序）。
+- 口诀：教的是"怎么让面向大众的内容被点开 / 看完 / 转化 / 涨粉"的可迁移方法 → 算（is_empty=false）；"讲观点/道理 / 应试达标 / 用工具把成品做出来 / 某学科观点或作品本身" → 不算（is_empty=true）。
+- 范围内但拿不准有没有方法 → 算（false）；但"是不是上面四类越界（尤其是讲观点 / 讲道理）"要**果断**，明显越界就判非创作（true）。
+- 若 is_empty=false，把视频里教的**具体创作知识点**忠实提炼出来（分条、不编造、不拔高）。
 
 只输出一个 JSON 对象：
 {"is_empty": true/false,
- "reason": "一句话理由，点明是「讲观点/作品本身」还是「教了哪种可复用的创作方法」（≤30字）",
+ "reason": "一句话理由：是「教了哪种自媒体创作方法」，还是「属于讲观点/应试/制作/学科/作品本身的哪一类」（≤30字）",
  "knowledge": "is_empty=false 时：视频里教的具体创作知识点全文（分条）；is_empty=true 时：空字符串"}