M5 流水线 + CLI + ingest：5 样例真实跑通入库

- pipeline.run_pipeline：6 步串联，每步落 ck_post/ck_knowledge_item，依赖可注入
- ingest.py：KnowHub 入库，INGEST_ENABLED 开关（开发期关闭，双重保险）
- cli.py：python -m creation_knowledge.cli run --urls ...
- tests/test_pipeline_e2e.py(离线 done/rejected 两路径)；scripts/run_batch.py
- 真实 e2e：5 帖全 done、28 条知识片段入库、ingest_status=pending

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

creation_knowledge/cli.py

@@ -0,0 +1,39 @@
+"""命令行入口：python -m creation_knowledge.cli run --urls <...>"""
+from __future__ import annotations
+
+import argparse
+import json
+
+from creation_knowledge.pipeline import run_pipeline
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="creation_knowledge")
+    sub = parser.add_subparsers(dest="command", required=True)
+    run = sub.add_parser("run", help="跑流水线：拉取->提取->筛选->拆分->解构->组装[->入库]")
+    run.add_argument("--urls", nargs="+", required=True, help="帖子链接或 content_id")
+    run.add_argument("--env-file", default=".env", help="环境变量文件路径")
+    run.add_argument("--ingest", action="store_true",
+                     help="开启真实入库（默认关闭，只组装+存库）")
+    return parser
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    if args.command == "run":
+        results = run_pipeline(
+            args.urls,
+            env_file=args.env_file,
+            ingest_enabled=True if args.ingest else False,
+        )
+        for r in results:
+            print(json.dumps(r, ensure_ascii=False))
+        ok = sum(1 for r in results if r["status"] == "done")
+        rej = sum(1 for r in results if r["status"] == "rejected")
+        fail = len(results) - ok - rej
+        print(f"\n汇总：done={ok} rejected={rej} failed={fail} / 共 {len(results)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

creation_knowledge/ingest.py

@@ -0,0 +1,41 @@
+"""入库：调 KnowHub ingest API。开发期默认关闭，只组装+存库（见 技术文档/技术架构.md）。"""
+from __future__ import annotations
+
+from typing import Any, Callable, Optional
+
+import httpx
+
+from creation_knowledge.config import Settings
+from creation_knowledge.models import IngestPayload
+
+INGEST_PATH = "/api/v1/knowledge/ingest"
+
+
+class IngestError(RuntimeError):
+    pass
+
+
+def ingest(
+    payload: IngestPayload,
+    *,
+    settings: Optional[Settings] = None,
+    env_file: str = ".env",
+    http_post: Callable[..., Any] = httpx.post,
+    timeout: float = 30.0,
+) -> dict:
+    """真实发送一条知识到 KnowHub。仅在 INGEST_ENABLED=true 时允许调用。"""
+    settings = settings or Settings.from_env(env_file)
+    if not settings.ingest_enabled:
+        raise IngestError("INGEST_ENABLED=false：开发期拒绝真实入库")
+    url = settings.knowhub_api.rstrip("/") + INGEST_PATH
+    try:
+        resp = http_post(
+            url,
+            json=payload.model_dump(),
+            headers={"Content-Type": "application/json"},
+            timeout=timeout,
+        )
+        resp.raise_for_status()
+        return resp.json()
+    except httpx.HTTPError as exc:
+        raise IngestError(f"ingest_http_error: {exc}") from exc

creation_knowledge/pipeline.py

@@ -0,0 +1,104 @@
+"""流水线：把 6 步串起来，每步落库。INGEST_ENABLED=False 时只组装+存，不真实发送。
+
+依赖都可注入（fetch/extract/chat/store），便于离线测试与替换平台。
+"""
+from __future__ import annotations
+
+from typing import Callable, Optional
+
+from creation_knowledge.config import Settings
+from creation_knowledge.integrations.crawler import CrawlerError, fetch_post_detail
+from creation_knowledge.integrations.db import CkStore
+from creation_knowledge.integrations.extractor import ExtractorError, GeminiExtractor
+from creation_knowledge.integrations.llm import ChatFn, default_chat
+from creation_knowledge.ingest import IngestError, ingest as real_ingest
+from creation_knowledge.models import ExtractedContent, Post
+from creation_knowledge.stages import (
+    build_ingest_payload,
+    deconstruct_item,
+    screen_post,
+    split_post,
+)
+
+FetchFn = Callable[[str], Post]
+ExtractFn = Callable[[Post], ExtractedContent]
+
+
+def _process_one(
+    url: str,
+    *,
+    settings: Settings,
+    store: CkStore,
+    fetch_fn: FetchFn,
+    extract_fn: ExtractFn,
+    chat: ChatFn,
+    ingest_enabled: bool,
+) -> dict:
+    # 1) 拉取
+    try:
+        post = fetch_fn(url)
+    except CrawlerError as exc:
+        return {"url": url, "status": "fetch_failed", "error": str(exc)}
+    store.upsert_post(post)  # stage=fetched
+
+    # 2) 多模态提取
+    try:
+        content = extract_fn(post)
+        store.set_extracted(post.id, content.model_dump())  # stage=extracted
+    except ExtractorError as exc:
+        store.update_stage(post.id, "failed")
+        return {"url": url, "post_id": post.id, "status": "extract_failed", "error": str(exc)}
+
+    # 3) 筛选
+    screening = screen_post(post, content, chat=chat)
+    store.set_screening(post.id, screening.model_dump())  # stage=screened
+    if not screening.passed:
+        store.update_stage(post.id, "rejected")
+        return {"url": url, "post_id": post.id, "status": "rejected",
+                "score": screening.score, "reason": screening.reason}
+
+    # 4) 拆分 -> 5) 解构 -> 6) 组装（+可选入库）
+    items = split_post(post, content, chat=chat)
+    item_ids = []
+    for item in items:
+        deco = deconstruct_item(item, chat=chat)
+        payload = build_ingest_payload(post, item, deco)
+        item_id = store.save_item(
+            post.id, item.model_dump(), deco.model_dump(), payload.model_dump()
+        )
+        item_ids.append(item_id)
+        if ingest_enabled:
+            try:
+                res = real_ingest(payload, settings=settings)
+                store.update_item_ingest(item_id, "ingested", res.get("knowledge_id"))
+            except IngestError:
+                store.update_item_ingest(item_id, "failed", None)
+    store.update_stage(post.id, "done")
+    return {"url": url, "post_id": post.id, "status": "done", "items": len(item_ids)}
+
+
+def run_pipeline(
+    urls: list[str],
+    *,
+    settings: Optional[Settings] = None,
+    env_file: str = ".env",
+    ingest_enabled: Optional[bool] = None,
+    store: Optional[CkStore] = None,
+    fetch_fn: Optional[FetchFn] = None,
+    extract_fn: Optional[ExtractFn] = None,
+    chat: Optional[ChatFn] = None,
+) -> list[dict]:
+    settings = settings or Settings.from_env(env_file)
+    ingest_enabled = settings.ingest_enabled if ingest_enabled is None else ingest_enabled
+    store = store or CkStore(settings.pg)
+    fetch_fn = fetch_fn or (lambda url: fetch_post_detail(url, settings=settings))
+    extract_fn = extract_fn or GeminiExtractor.from_env(env_file=env_file).extract
+    chat = chat or default_chat(env_file)
+
+    results = []
+    for url in urls:
+        results.append(_process_one(
+            url, settings=settings, store=store, fetch_fn=fetch_fn,
+            extract_fn=extract_fn, chat=chat, ingest_enabled=ingest_enabled,
+        ))
+    return results

scripts/run_batch.py

@@ -0,0 +1,49 @@
+"""M5 真实 e2e：把 5 个样例帖跑完整流水线，落 DB（INGEST_ENABLED=False），再回读校验。
+
+须在能连 DB + OpenRouter 的云端跑（会有 crawler/Gemini/claude 调用费）。
+用法：python scripts/run_batch.py [env_file] [content_id ...]
+"""
+from __future__ import annotations
+
+import json
+import sys
+
+from creation_knowledge.config import PgConfig
+from creation_knowledge.integrations.db import CkStore
+from creation_knowledge.pipeline import run_pipeline
+
+DEFAULT_IDS = [
+    "67e4bdf50000000006028a59",  # HOW+WHAT 海狸
+    "698481e1000000000a02a7c1",  # HOW+WHAT Irvin
+    "67e2e39b0000000003028ff0",  # HOW 拾意
+    "699308fa0000000016009697",  # WHY 方圆
+    "680659e8000000001a007a11",  # HOW+WHY+WHAT 拾意
+]
+
+
+def main() -> int:
+    args = sys.argv[1:]
+    env_file = args[0] if args and args[0].endswith(".env") else ".env"
+    ids = [a for a in args if not a.endswith(".env")] or DEFAULT_IDS
+
+    print(f"=== 跑流水线：{len(ids)} 个样例（ingest 关闭）===")
+    results = run_pipeline(ids, env_file=env_file)
+    for r in results:
+        print(json.dumps(r, ensure_ascii=False))
+
+    print("\n=== DB 回读校验 ===")
+    store = CkStore(PgConfig.from_env(env_file))
+    for r in results:
+        pid = r.get("post_id")
+        if not pid:
+            print(f"{r['url']}  -> {r['status']}")
+            continue
+        post = store.read_post(pid)
+        items = store.read_items(pid)
+        statuses = [it["ingest_status"] for it in items]
+        print(f"{pid}  stage={post['stage']}  items={len(items)}  ingest_status={statuses}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_pipeline_e2e.py

@@ -0,0 +1,108 @@
+"""M5 离线编排测试：注入 fake store/fetch/extract/chat，验证 done 与 rejected 两条路径。
+
+真实落库的 e2e 见 scripts/run_batch.py（在能连 DB 的云端跑）。
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from creation_knowledge.config import PgConfig, Settings
+from creation_knowledge.integrations.crawler import parse_detail_response
+from creation_knowledge.models import ExtractedContent
+from creation_knowledge.pipeline import run_pipeline
+
+FIXTURES = Path(__file__).parent / "fixtures"
+CID = "67e4bdf50000000006028a59"
+
+
+def _settings() -> Settings:
+    return Settings(
+        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
+        crawler_base_url="http://x", crawler_key="", crawler_timeout=30,
+        video_model="m", gemini_api_key="", openrouter_base_url="http://x",
+        openrouter_api_key="k", llm_model="m", knowhub_api="http://x",
+        ingest_enabled=False,
+    )
+
+
+class FakeStore:
+    def __init__(self):
+        self.posts: dict = {}
+        self.items: list = []
+
+    def upsert_post(self, post):
+        self.posts[post.id] = {"stage": "fetched"}
+
+    def set_extracted(self, pid, ex):
+        self.posts[pid].update(stage="extracted", extracted=ex)
+
+    def set_screening(self, pid, sc):
+        self.posts[pid].update(stage="screened", screening=sc)
+
+    def update_stage(self, pid, stage):
+        self.posts[pid]["stage"] = stage
+
+    def save_item(self, pid, item, deco, payload):
+        self.items.append({"post_id": pid, "item": item, "deco": deco, "payload": payload})
+        return len(self.items)
+
+    def update_item_ingest(self, iid, status, kid):
+        self.items[iid - 1].update(ingest_status=status, knowledge_id=kid)
+
+
+def _fetch(url):
+    resp = json.loads((FIXTURES / f"xhs_case_{CID}.json").read_text("utf-8"))
+    return parse_detail_response(resp, fallback_content_id=CID)
+
+
+def _extract(post):
+    return ExtractedContent(text=post.body_text or "内容", is_empty=False)
+
+
+def _chat_pass(system, user):
+    if "筛选" in system:
+        return {"passed": True, "score": 8, "reason": "ok"}
+    if "拆分" in system:
+        return {"items": [{"title": "脚本要素", "knowledge_types": ["how"],
+                           "what": "null", "why": "null", "how": "逐个填要素",
+                           "evidence": ["原句"]}]}
+    if "解构" in system:
+        return {"stages": ["脚本"], "scopes": [{"scope_type": "form", "value": "操作流程"}],
+                "stage_reason": "r", "scope_reason": "r"}
+    raise AssertionError(f"unexpected system: {system}")
+
+
+def _chat_reject(system, user):
+    if "筛选" in system:
+        return {"passed": False, "score": 2, "reason": "只是作品本身"}
+    raise AssertionError("rejected post 不应进入拆分/解构")
+
+
+def test_pipeline_done_path():
+    store = FakeStore()
+    results = run_pipeline(
+        ["https://www.xiaohongshu.com/explore/" + CID],
+        settings=_settings(), ingest_enabled=False, store=store,
+        fetch_fn=_fetch, extract_fn=_extract, chat=_chat_pass,
+    )
+    r = results[0]
+    assert r["status"] == "done" and r["items"] == 1
+    pid = f"xhs_{CID}"
+    assert store.posts[pid]["stage"] == "done"
+    assert len(store.items) == 1
+    saved = store.items[0]
+    assert saved["payload"]["source"]["id"] == pid
+    assert saved["payload"]["dim_attributes"] == ["how"]
+
+
+def test_pipeline_rejected_path():
+    store = FakeStore()
+    results = run_pipeline(
+        [CID], settings=_settings(), ingest_enabled=False, store=store,
+        fetch_fn=_fetch, extract_fn=_extract, chat=_chat_reject,
+    )
+    r = results[0]
+    assert r["status"] == "rejected"
+    assert store.posts[f"xhs_{CID}"]["stage"] == "rejected"
+    assert store.items == []