卡片溯源 + 视频抽帧 + MAX_CARDS + 文档补齐

统一"卡片"抽象：图文每张图、视频每帧 = 卡片(1-based)，下游溯源只认卡片号。
- 数据模型：Card/Post.cards、ExtractedContent.cards(按卡归因)、KnowledgeItem.source_cards、Evidence{text,card}
- 归因在提取步：extractor 给每卡打【卡片N】标签、Gemini 按卡输出；split 产出 source_cards + evidence.card
- 视频抽帧 integrations/video_frames.py：ffmpeg 场景检测(0.3)+最小间隔/最多帧/下采样+均匀采样回退；帧带时间戳
- MAX_IMAGES=6 → MAX_CARDS(默认12，CK_MAX_CARDS)，超限截断记日志；修复 9 卡帖漏卡
- DB：ck_post 加 cards 列(ALTER 已在库执行)；api 加 /frames 静态服务
- web：知识卡内嵌来源卡片缩略图(可放大)、证据→卡片可点开；原文按 cards 统一展示
- 提示词 v4：extract/split 加按卡片归因
- 测试：+test_video_frames、+per-card 提取/拆分用例，共 22 passed
- 文档补齐(as-built)：架构 §9-12(UI/API/提示词/部署/溯源/抽帧)、设计 §10 溯源、开发顺序 M0-M8✅+M10/M11

真机：重跑 5 帖入库带卡片溯源；9 卡帖 cards=9(不再漏)；item.source_cards/evidence.card 已落库

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

creation_knowledge/api.py

@@ -83,5 +83,11 @@ def get_prompts() -> dict:
 
 
 # 单页前端挂在最后（catch-all），不影响上面的 /api 路由
+# 视频帧静态服务（card.url = /frames/<post_id>/<file>）
+_frames_dir = Path(_settings().frames_dir)
+_frames_dir.mkdir(parents=True, exist_ok=True)
+app.mount("/frames", StaticFiles(directory=str(_frames_dir)), name="frames")
+
+# 单页前端挂在最后（catch-all），不影响上面的 /api、/frames
 if WEB_DIR.exists():
     app.mount("/", StaticFiles(directory=str(WEB_DIR), html=True), name="web")

creation_knowledge/config.py

@@ -86,6 +86,9 @@ class Settings:
     # 入库（M5）
     knowhub_api: str
     ingest_enabled: bool
+    # 卡片 / 抽帧
+    max_cards: int
+    frames_dir: str
 
     @classmethod
     def from_env(cls, env_file: str | Path = ".env") -> "Settings":
@@ -115,4 +118,6 @@ class Settings:
             # 开发期默认关闭真实入库；显式置 true 才发送
             ingest_enabled=env_value("INGEST_ENABLED", file_env, "false").lower()
             in ("1", "true", "yes"),
+            max_cards=int(env_value("CK_MAX_CARDS", file_env, "12")),
+            frames_dir=env_value("CK_FRAMES_DIR", file_env, "runtime/frames"),
         )

creation_knowledge/integrations/crawler.py

@@ -15,7 +15,7 @@ from urllib.parse import urljoin
 import httpx
 
 from creation_knowledge.config import Settings
-from creation_knowledge.models import Post
+from creation_knowledge.models import Card, Post
 
 DETAIL_PATH = "/crawler/xiao_hong_shu/detail"
 RATE_LIMIT_SECONDS = 15.0
@@ -93,6 +93,9 @@ def parse_detail_response(response: dict, fallback_content_id: str = "") -> Post
     link = inner.get("content_link") or (
         f"https://www.xiaohongshu.com/explore/{content_id}" if content_id else ""
     )
+    images = _image_urls(inner)
+    # 图文帖：每张图就是一张卡片（1-based；视频帧卡片在 pipeline 抽帧后补入）
+    cards = [Card(index=i, kind="image", url=u) for i, u in enumerate(images, start=1)]
     return Post(
         id=f"xhs_{content_id}",
         platform="xiaohongshu",
@@ -102,8 +105,9 @@ def parse_detail_response(response: dict, fallback_content_id: str = "") -> Post
         content_type=inner.get("content_type") or "",
         body_text=inner.get("body_text") or "",
         topic_list=list(inner.get("topic_list") or []),
-        image_urls=_image_urls(inner),
+        image_urls=images,
         video_urls=_video_urls(inner),
+        cards=cards,
         author_id=inner.get("channel_account_id"),
         author_name=inner.get("channel_account_name"),
         raw=response,

creation_knowledge/integrations/db.py

@@ -42,20 +42,22 @@ class CkStore:
 
     # ---------- 写 ----------
     def upsert_post(self, post: Post) -> None:
-        """落 fetch 产物：raw 等基础字段，stage=fetched。先 UPDATE 后 INSERT。"""
+        """落 fetch 产物：raw / cards 等基础字段，stage=fetched。先 UPDATE 后 INSERT。"""
+        cards = psycopg2.extras.Json([c.model_dump() for c in post.cards])
         with self._connection_factory() as conn:
             with conn.cursor() as cur:
                 cur.execute(
-                    "UPDATE ck_post SET platform=%s, url=%s, raw=%s, "
+                    "UPDATE ck_post SET platform=%s, url=%s, raw=%s, cards=%s, "
                     "updated_at=now() WHERE id=%s",
-                    (post.platform, post.url, psycopg2.extras.Json(post.raw), post.id),
+                    (post.platform, post.url, psycopg2.extras.Json(post.raw),
+                     cards, post.id),
                 )
                 if cur.rowcount == 0:
                     cur.execute(
-                        "INSERT INTO ck_post (id, platform, url, raw, stage) "
-                        "VALUES (%s, %s, %s, %s, 'fetched')",
+                        "INSERT INTO ck_post (id, platform, url, raw, cards, stage) "
+                        "VALUES (%s, %s, %s, %s, %s, 'fetched')",
                         (post.id, post.platform, post.url,
-                         psycopg2.extras.Json(post.raw)),
+                         psycopg2.extras.Json(post.raw), cards),
                     )
             conn.commit()
 

creation_knowledge/integrations/extractor.py

@@ -12,15 +12,27 @@ from typing import Any, Callable, Mapping, Optional
 
 import httpx
 
+import logging
+
 from creation_knowledge.config import load_env_file
 from creation_knowledge.jsonio import extract_json_object, to_bool
-from creation_knowledge.models import ExtractedContent, Post
+from creation_knowledge.models import Card, CardExtract, ExtractedContent, Post
 from creation_knowledge.prompts import load_prompt
 
+logger = logging.getLogger(__name__)
+
 DEFAULT_MODEL = "google/gemini-3-flash-preview"
 DEFAULT_BASE_URL = "https://openrouter.ai/api/v1"
 DEFAULT_TIMEOUT = 90.0
-MAX_IMAGES = 6
+MAX_CARDS = 12  # 一次提取最多送多少张卡片（图/帧），控成本；超出截断并记日志
+
+
+def _card_label(card: Card) -> str:
+    """给模型看的卡片标签，如 【卡片1】 或 【卡片3 · 00:12】。card.index 为 1-based。"""
+    if card.kind == "frame" and card.timestamp is not None:
+        ts = int(card.timestamp)
+        return f"【卡片{card.index} · {ts // 60:02d}:{ts % 60:02d}】"
+    return f"【卡片{card.index}】"
 
 _SYSTEM_PROMPT = (
     "你是创作知识提取助手。从给定的小红书帖子（标题、正文、图片、视频）中，"
@@ -42,7 +54,7 @@ class GeminiExtractor:
         base_url: str = DEFAULT_BASE_URL,
         timeout_seconds: float = DEFAULT_TIMEOUT,
         http_post: Callable[..., Any] = httpx.post,
-        max_images: int = MAX_IMAGES,
+        max_cards: int = MAX_CARDS,
     ) -> None:
         if not api_key:
             raise ExtractorError("missing OPENROUTER_API_KEY")
@@ -51,7 +63,7 @@ class GeminiExtractor:
         self.base_url = base_url.rstrip("/")
         self.timeout_seconds = timeout_seconds
         self.http_post = http_post
-        self.max_images = max_images
+        self.max_cards = max_cards
 
     @classmethod
     def from_env(cls, env: Mapping[str, str] | None = None, env_file: str = ".env") -> "GeminiExtractor":
@@ -64,8 +76,22 @@ class GeminiExtractor:
             model=source.get("CONTENT_AGENT_VIDEO_LLM_MODEL") or DEFAULT_MODEL,
             base_url=source.get("OPENROUTER_BASE_URL") or DEFAULT_BASE_URL,
             timeout_seconds=float(source.get("CONTENT_AGENT_VIDEO_LLM_TIMEOUT_SECONDS") or DEFAULT_TIMEOUT),
+            max_cards=int(source.get("CK_MAX_CARDS") or MAX_CARDS),
         )
 
+    def _cards(self, post: Post) -> list[Card]:
+        """取要送给模型的卡片：优先 post.cards；为空则回退 image_urls。"""
+        cards = post.cards or [
+            Card(index=i, kind="image", url=u)
+            for i, u in enumerate(post.image_urls, start=1)
+        ]
+        if len(cards) > self.max_cards:
+            dropped = [c.index for c in cards[self.max_cards:]]
+            logger.warning("post %s 卡片数 %d 超过 MAX_CARDS=%d，截断丢弃卡片 %s",
+                           post.id, len(cards), self.max_cards, dropped)
+            cards = cards[: self.max_cards]
+        return cards
+
     def build_messages(self, post: Post) -> list[dict]:
         user_text = load_prompt("extract").format(
             title=post.title or "(无)",
@@ -73,8 +99,10 @@ class GeminiExtractor:
             body=post.body_text or "(空)",
         )
         parts: list[dict] = [{"type": "text", "text": user_text}]
-        for url in post.image_urls[: self.max_images]:
-            parts.append({"type": "image_url", "image_url": {"url": url}})
+        # 每张卡片前插一个【卡片N】标签，再插图，便于模型按卡片归因
+        for card in self._cards(post):
+            parts.append({"type": "text", "text": _card_label(card)})
+            parts.append({"type": "image_url", "image_url": {"url": card.url}})
         return [
             {"role": "system", "content": _SYSTEM_PROMPT},
             {"role": "user", "content": parts},
@@ -97,8 +125,16 @@ class GeminiExtractor:
                 resp.raise_for_status()
                 content = resp.json()["choices"][0]["message"]["content"]
                 data = extract_json_object(content)
+                cards = []
+                for c in data.get("cards") or []:
+                    if isinstance(c, dict) and c.get("index") is not None:
+                        cards.append(CardExtract(
+                            index=int(c["index"]),
+                            content=str(c.get("content") or ""),
+                        ))
                 return ExtractedContent(
                     text=str(data.get("text") or ""),
+                    cards=cards,
                     from_image=str(data.get("from_image") or ""),
                     from_video=str(data.get("from_video") or ""),
                     is_empty=to_bool(data.get("is_empty")),

creation_knowledge/integrations/video_frames.py

@@ -0,0 +1,150 @@
+"""视频抽帧：把视频转成一组"卡片"（帧），供多模态提取 + 溯源使用。
+
+策略（见 技术文档/技术架构.md §12）：
+  主策略 = 场景切换检测（ffmpeg select='gt(scene,T)'），知识类视频常是字幕卡/场景切换，
+           正好对应"卡片"；附带 showinfo 解析每帧时间戳。
+  护栏   = 最小间隔去近重复、最多 max_frames 控成本、下采样到长边≤720。
+  回退   = 场景帧 < FALLBACK_MIN 时，按时长均匀采样 FALLBACK_TARGET 帧。
+
+无 ffprobe 依赖：时长从 ffmpeg stderr 的 "Duration:" 解析。
+帧文件写到 out_dir；Card.url = f"{url_prefix}/{文件名}"，由 API 的 /frames 静态服务。
+"""
+from __future__ import annotations
+
+import logging
+import re
+import subprocess
+from pathlib import Path
+from typing import Callable, Optional
+
+import httpx
+
+from creation_knowledge.models import Card
+
+logger = logging.getLogger(__name__)
+
+SCENE_THRESHOLD = 0.3      # 场景切换阈值（电影 30° 规则）
+MIN_GAP_SECONDS = 1.5      # 相邻保留帧最小间隔，去突发近重复
+FALLBACK_MIN = 4           # 场景帧少于此数则触发均匀采样回退
+FALLBACK_TARGET = 8        # 回退时采样的帧数
+SCALE_VF = "scale='min(720,iw)':-2"  # 长边下采样到 ≤720
+
+# 各平台下载视频需要的 Referer
+_REFERER = {
+    "douyin": "https://www.douyin.com/",
+    "kuaishou": "https://www.kuaishou.com/",
+    "bilibili": "https://www.bilibili.com/",
+    "shipinhao": "https://channels.weixin.qq.com/",
+    "xiaohongshu": "https://www.xiaohongshu.com/",
+}
+
+
+class VideoFramesError(RuntimeError):
+    pass
+
+
+def _ffmpeg_bin() -> str:
+    try:
+        import imageio_ffmpeg
+        return imageio_ffmpeg.get_ffmpeg_exe()
+    except Exception:
+        return "ffmpeg"
+
+
+def _download(url: str, platform: str, dst: Path, timeout: float) -> None:
+    headers = {"User-Agent": "Mozilla/5.0", "Referer": _REFERER.get(platform, "")}
+    with httpx.stream("GET", url, headers=headers, timeout=timeout,
+                      follow_redirects=True) as r:
+        r.raise_for_status()
+        with open(dst, "wb") as f:
+            for chunk in r.iter_bytes():
+                f.write(chunk)
+
+
+def _run(cmd: list[str]) -> str:
+    """跑 ffmpeg，返回 stderr（ffmpeg 把信息打到 stderr）。"""
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    return proc.stderr or ""
+
+
+def _probe_duration(ffmpeg: str, video: Path) -> float:
+    err = _run([ffmpeg, "-hide_banner", "-i", str(video)])
+    m = re.search(r"Duration:\s*(\d+):(\d+):(\d+\.?\d*)", err)
+    if not m:
+        return 0.0
+    h, mn, s = m.groups()
+    return int(h) * 3600 + int(mn) * 60 + float(s)
+
+
+def _scene_frames(ffmpeg: str, video: Path, out: Path, threshold: float,
+                  min_gap: float, max_frames: int) -> list[tuple[Path, Optional[float]]]:
+    pattern = str(out / "scene_%04d.jpg")
+    err = _run([ffmpeg, "-hide_banner", "-i", str(video),
+                "-vf", f"select='gt(scene,{threshold})',showinfo,{SCALE_VF}",
+                "-vsync", "vfr", "-q:v", "3", pattern])
+    times = [float(t) for t in re.findall(r"pts_time:(\d+\.?\d*)", err)]
+    files = sorted(out.glob("scene_*.jpg"))
+    paired = list(zip(files, times + [None] * (len(files) - len(times))))
+    # 最小间隔过滤
+    kept: list[tuple[Path, Optional[float]]] = []
+    last = None
+    for path, ts in paired:
+        if ts is not None and last is not None and ts - last < min_gap:
+            continue
+        kept.append((path, ts))
+        if ts is not None:
+            last = ts
+        if len(kept) >= max_frames:
+            break
+    return kept
+
+
+def _uniform_frames(ffmpeg: str, video: Path, out: Path, duration: float,
+                    count: int) -> list[tuple[Path, Optional[float]]]:
+    if duration <= 0:
+        duration = float(count)  # 兜底，避免全 0
+    result: list[tuple[Path, Optional[float]]] = []
+    for k in range(count):
+        ts = duration * (k + 0.5) / count
+        path = out / f"uni_{k:04d}.jpg"
+        _run([ffmpeg, "-hide_banner", "-ss", f"{ts:.2f}", "-i", str(video),
+              "-frames:v", "1", "-vf", SCALE_VF, "-q:v", "3", str(path)])
+        if path.exists():
+            result.append((path, round(ts, 2)))
+    return result
+
+
+def extract_frames(
+    video_src: str,
+    *,
+    out_dir: str | Path,
+    url_prefix: str,
+    platform: str = "",
+    max_frames: int = 12,
+    start_index: int = 1,
+    timeout: float = 60.0,
+) -> list[Card]:
+    """把视频抽成帧卡片。video_src 可为 URL 或本地文件路径。返回 Card(kind='frame')。"""
+    out = Path(out_dir)
+    out.mkdir(parents=True, exist_ok=True)
+    ffmpeg = _ffmpeg_bin()
+
+    src = Path(video_src)
+    if not src.exists():  # 是 URL，先下载
+        src = out / "_src.mp4"
+        _download(video_src, platform, src, timeout)
+
+    frames = _scene_frames(ffmpeg, src, out, SCENE_THRESHOLD, MIN_GAP_SECONDS, max_frames)
+    if len(frames) < FALLBACK_MIN:
+        logger.info("场景帧仅 %d，回退均匀采样", len(frames))
+        dur = _probe_duration(ffmpeg, src)
+        frames = _uniform_frames(ffmpeg, src, out, dur, FALLBACK_TARGET)
+    frames = frames[:max_frames]
+    if not frames:
+        raise VideoFramesError(f"未能从视频抽出任何帧: {video_src}")
+
+    return [
+        Card(index=start_index + i, kind="frame",
+             url=f"{url_prefix.rstrip('/')}/{path.name}", timestamp=ts)
+        for i, (path, ts) in enumerate(frames)
+    ]

creation_knowledge/models.py

@@ -8,6 +8,16 @@ from pydantic import BaseModel, Field
 KnowledgeType = Literal["what", "why", "how"]
 ScopeType = Literal["substance", "form", "feeling", "effect", "intent"]
 Stage = Literal["灵感", "选题", "脚本"]
+CardKind = Literal["image", "frame"]
+
+
+class Card(BaseModel):
+    """一张"卡片"：图文帖的一张图，或视频抽出的一帧。下游溯源只认 index。"""
+
+    index: int  # 帖内顺序号，从 0 起
+    kind: CardKind = "image"
+    url: str
+    timestamp: Optional[float] = None  # 仅 frame：该帧在视频中的秒数
 
 
 class Post(BaseModel):
@@ -23,17 +33,26 @@ class Post(BaseModel):
     topic_list: list[str] = Field(default_factory=list)
     image_urls: list[str] = Field(default_factory=list)
     video_urls: list[str] = Field(default_factory=list)
+    cards: list[Card] = Field(default_factory=list)  # 统一视觉卡片（图/帧）
     author_id: Optional[str] = None
     author_name: Optional[str] = None
     raw: dict = Field(default_factory=dict)  # 原始响应，整体落 ck_post.raw
 
 
+class CardExtract(BaseModel):
+    """某张卡片上提取到的知识要点（按卡片归因）。"""
+
+    index: int
+    content: str = ""
+
+
 class ExtractedContent(BaseModel):
     """extract_content 的产物：多模态汇总后的真实内容。不直接用 body_text。"""
 
     text: str = ""  # 汇总后的正文/讲解
-    from_image: str = ""  # 图片里提取到的知识要点
-    from_video: str = ""  # 视频里提取到的知识要点
+    cards: list[CardExtract] = Field(default_factory=list)  # 按卡片归因的提取
+    from_image: str = ""  # 兼容保留：图片里提取到的知识要点
+    from_video: str = ""  # 兼容保留：视频里提取到的知识要点
     is_empty: bool = False  # 多模态提取后仍无有效内容
 
 
@@ -45,6 +64,13 @@ class ScreeningResult(BaseModel):
     reason: str = ""
 
 
+class Evidence(BaseModel):
+    """一条原文证据，链接到它出自的卡片（纯正文证据 card=None）。"""
+
+    text: str
+    card: Optional[int] = None
+
+
 class KnowledgeItem(BaseModel):
     """split_post 拆出的一个知识片段。knowledge_types 必须与非空的 what/why/how 一致。"""
 
@@ -53,7 +79,8 @@ class KnowledgeItem(BaseModel):
     what: Optional[str] = None
     why: Optional[str] = None
     how: Optional[str] = None
-    evidence: list[str] = Field(default_factory=list)
+    source_cards: list[int] = Field(default_factory=list)  # 这条知识出自哪些卡片
+    evidence: list[Evidence] = Field(default_factory=list)
 
 
 class Scope(BaseModel):

creation_knowledge/pipeline.py

@@ -4,6 +4,7 @@
 """
 from __future__ import annotations
 
+import logging
 from typing import Callable, Optional
 
 from creation_knowledge.config import Settings
@@ -20,6 +21,8 @@ from creation_knowledge.stages import (
     split_post,
 )
 
+logger = logging.getLogger(__name__)
+
 FetchFn = Callable[[str], Post]
 ExtractFn = Callable[[Post], ExtractedContent]
 
@@ -39,7 +42,22 @@ def _process_one(
         post = fetch_fn(url)
     except CrawlerError as exc:
         return {"url": url, "status": "fetch_failed", "error": str(exc)}
-    store.upsert_post(post)  # stage=fetched
+
+    # 1.5) 视频帖：抽帧补成卡片（best-effort，失败不阻塞整帖）
+    if post.video_urls and not post.cards:
+        try:
+            from creation_knowledge.integrations.video_frames import extract_frames
+            post.cards = extract_frames(
+                post.video_urls[0],
+                out_dir=f"{settings.frames_dir}/{post.id}",
+                url_prefix=f"/frames/{post.id}",
+                platform=post.platform,
+                max_frames=settings.max_cards,
+            )
+        except Exception as exc:  # 抽帧失败不影响图文/正文路径
+            logger.warning("post %s 抽帧失败: %s", post.id, exc)
+
+    store.upsert_post(post)  # stage=fetched（含 cards）
 
     # 2) 多模态提取
     try:

creation_knowledge/prompts.py

@@ -6,7 +6,7 @@ from pathlib import Path
 PROMPTS_DIR = Path(__file__).resolve().parent.parent / "prompts"
 
 # 提示词版本：改提示词时手动 +1，写进 ck 记录便于回溯
-PROMPT_VERSION = "v3.1"
+PROMPT_VERSION = "v4"
 
 
 def load_prompt(name: str) -> str:

creation_knowledge/stages/_common.py

@@ -10,14 +10,19 @@ _EMPTY = {"", "null", "none", "无", "n/a", "na"}
 
 
 def content_for_llm(post: Post, content: ExtractedContent) -> str:
-    """喂给 LLM 的内容 = 多模态提取结果（不是 body_text）。"""
+    """喂给 LLM 的内容 = 多模态提取结果（不是 body_text），含按卡片归因的分块。"""
     parts = []
     if content.text:
-        parts.append(content.text)
-    if content.from_image:
-        parts.append("【图片】" + content.from_image)
-    if content.from_video:
-        parts.append("【视频】" + content.from_video)
+        parts.append("【综合】" + content.text)
+    # 按卡片分块，让拆分步骤能把知识溯源到【卡片N】
+    for c in content.cards:
+        if c.content:
+            parts.append(f"【卡片{c.index}】{c.content}")
+    if not content.cards:  # 兼容旧产物
+        if content.from_image:
+            parts.append("【图片】" + content.from_image)
+        if content.from_video:
+            parts.append("【视频】" + content.from_video)
     return "\n".join(parts) or post.body_text
 
 

creation_knowledge/stages/assemble.py

@@ -22,10 +22,16 @@ def build_ingest_payload(
         ensure_ascii=False,
     )
     custom_ext: list[dict] = []
-    if item.evidence:
+    if item.source_cards:
         custom_ext.append(
-            {"key": "原文证据", "type": "str", "value": " / ".join(item.evidence)}
+            {"key": "来源卡片", "type": "str",
+             "value": "、".join(f"卡片{n}" for n in item.source_cards)}
+        )
+    if item.evidence:
+        ev = " / ".join(
+            f"{e.text}（卡片{e.card}）" if e.card else e.text for e in item.evidence
         )
+        custom_ext.append({"key": "原文证据", "type": "str", "value": ev})
     if deco.stage_reason:
         custom_ext.append(
             {"key": "阶段判断理由", "type": "str", "value": deco.stage_reason}

creation_knowledge/stages/split.py

@@ -4,13 +4,29 @@ from __future__ import annotations
 from typing import Optional
 
 from creation_knowledge.integrations.llm import ChatFn, default_chat
-from creation_knowledge.models import ExtractedContent, KnowledgeItem, Post
+from creation_knowledge.models import Evidence, ExtractedContent, KnowledgeItem, Post
 from creation_knowledge.prompts import load_prompt
 from creation_knowledge.stages._common import content_for_llm, norm_text, norm_types
 
 SYSTEM = "你是创作知识拆分助手，原文有什么提什么，不编造。"
 
 
+def _to_int(value) -> Optional[int]:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _to_evidence(raw) -> Optional[Evidence]:
+    """证据可能是字符串（旧）或 {text, card}（新）。"""
+    if isinstance(raw, dict):
+        text = norm_text(raw.get("text"))
+        return Evidence(text=text, card=_to_int(raw.get("card"))) if text else None
+    text = norm_text(raw)
+    return Evidence(text=str(raw).strip()) if text else None
+
+
 def _to_item(raw: dict) -> Optional[KnowledgeItem]:
     what = norm_text(raw.get("what"))
     why = norm_text(raw.get("why"))
@@ -22,10 +38,11 @@ def _to_item(raw: dict) -> Optional[KnowledgeItem]:
     if not types:
         return None  # 三个都空，丢弃
     title = norm_text(raw.get("title")) or (what or why or how or "")[:20]
-    evidence = [e for e in (raw.get("evidence") or []) if norm_text(e)]
+    evidence = [e for e in (_to_evidence(x) for x in (raw.get("evidence") or [])) if e]
+    source_cards = [n for n in (_to_int(x) for x in (raw.get("source_cards") or [])) if n is not None]
     return KnowledgeItem(
         title=title, knowledge_types=types, what=what, why=why, how=how,
-        evidence=[str(e).strip() for e in evidence],
+        source_cards=source_cards, evidence=evidence,
     )
 
 

prompts/extract.txt

@@ -10,18 +10,18 @@
 <工作要点>
 - 知识常在图片/视频里，正文常常只是话题串（如 "#短剧编剧# #写作#"）。为什么强调：只读正文会系统性漏掉真正的知识，所以必须看图、看视频。
 - 以原始素材为准，忠实转述，不编造、不补全、不替作者总结它没说的结论；有的尽量提全。
-- 分模态填：text=综合所有模态后把创作知识讲清楚（主产物）；from_image=只填图里的；from_video=只填视频里的。
-- is_empty：多模态都看完，确实没有任何可迁移的创作知识（纯作品/纯展示/纯无关）→ true；有一点方法/原理/清单 → false；不确定 → 先如实写进 text、标 false，把好坏交给后续筛选。
+- **按卡片归因**：每张图/帧旁都标了【卡片N】。除了 text（综合所有卡片把创作知识讲清楚，主产物），还要填 cards——每张**有知识**的卡片一条 {{"index": N, "content": "这张卡上的知识要点"}}，N 就是【卡片N】里的号；某张卡没有知识就不列它。为什么：后续要据此把每条知识溯源到具体卡片。
+- is_empty：所有卡片都看完，确实没有任何可迁移的创作知识（纯作品/纯展示/纯无关）→ true；有一点方法/原理/清单 → false；不确定 → 先如实写进 text、标 false，把好坏交给后续筛选。
 </工作要点>
 
 <示例>
-<example>  知识在图里、正文为空
-正文："#短剧编剧# #写作#"；图片：九宫格讲 "前提 = 作者立场+人物+冲突+结论"、"人物三维度：生理/社会/心理"。
-输出：{{"text": "讲解剧本创作三要素：前提（=作者立场+人物+冲突+结论）、人物（生理/社会/心理三维度）、冲突", "from_image": "前提公式；人物三维度清单", "from_video": "", "is_empty": false}}
+<example>  知识在图里、正文为空（多张卡片）
+正文："#短剧编剧# #写作#"；【卡片1】图：前提 = 作者立场+人物+冲突+结论；【卡片2】图：人物三维度——生理/社会/心理。
+输出：{{"text": "讲解剧本创作要素：前提（=作者立场+人物+冲突+结论）、人物（生理/社会/心理三维度）", "cards": [{{"index": 1, "content": "前提公式：作者立场+人物+冲突+结论"}}, {{"index": 2, "content": "人物三维度：生理/社会/心理"}}], "from_image": "", "from_video": "", "is_empty": false}}
 </example>
 <example>  纯作品
-正文："夏日雨后彩虹草原" 加一段画面描述/出图提示词；图片：一张风景图。
-输出：{{"text": "", "from_image": "", "from_video": "", "is_empty": true}}
+正文："夏日雨后彩虹草原" 加一段画面描述/出图提示词；【卡片1】图：一张风景图。
+输出：{{"text": "", "cards": [], "from_image": "", "from_video": "", "is_empty": true}}
 </example>
 </示例>
 
@@ -35,5 +35,7 @@
 
 <输出>
 只输出一个 JSON 对象：
-{{"text": "把创作知识完整忠实讲清楚；没有就空字符串", "from_image": "仅图片里的；没有则空", "from_video": "仅视频里的；没有则空", "is_empty": false}}
+{{"text": "把创作知识完整忠实讲清楚；没有就空字符串",
+  "cards": [{{"index": 卡片号, "content": "这张卡上的知识要点"}}],
+  "from_image": "", "from_video": "", "is_empty": false}}
 </输出>

prompts/split.txt

@@ -15,31 +15,33 @@
 - 忠实：原文有什么提什么，没有的字段填 null，不为凑齐 what/why/how 而编。
 - knowledge_types 与非空字段一致（顺序 what→why→how）。
 - title 用能概括这条知识的具体短语，别用泛词（"技巧""干货"）。
-- evidence：先从提取内容里摘出支持这条知识的原句放进 evidence（可多条），再据此填 what/why/how。先摘引、再判断，能贴原文、少编造；evidence 必须来自素材。
+- 溯源：提取内容按【卡片N】分块（【综合】块来自正文/汇总）。每条知识填 source_cards=它出自哪些卡片号（可多张；纯来自【综合】/正文则填 []）。
+- evidence：先摘出支持这条知识的原句放进 evidence，每条写成 {{"text": "原句", "card": 出自的卡片号或 null}}；先摘引、再判断，证据必须来自素材。
 </拆分规则>
 
 <示例>
-<example>  清单型：每个要素都配了说明，仍合并为 1 条（不要拆成 10 条）
-提取内容：短视频脚本包含 10 个要素——标题（视频的灵魂，要精准点题）、拍摄地点（给故事安排舞台）、分镜（规划镜头顺序）、景别、运镜、音效、字幕……每个要素都要逐一规划好。
+<example>  清单型：每个要素一张卡，仍合并为 1 条；source_cards 列全部相关卡
+提取内容：【综合】短视频脚本含 10 个要素。【卡片1】标题：视频灵魂，精准点题。【卡片2】拍摄地点：给故事搭舞台。【卡片3】分镜：规划镜头顺序……（每张卡讲一个要素）
 输出：{{"items": [
   {{"title": "短视频脚本的 10 个核心要素",
     "knowledge_types": ["what", "how"],
-    "what": "短视频脚本由标题、拍摄地点、分镜、景别、运镜、音效、字幕等 10 个要素构成：标题点题、拍摄地点搭背景、分镜定镜头顺序……",
+    "what": "短视频脚本由标题、拍摄地点、分镜……等 10 个要素构成",
     "why": null,
     "how": "按这份清单逐一规划每个要素，把脚本搭完整",
-    "evidence": ["短视频脚本包含 10 个要素", "每个要素都要逐一规划"]}}
+    "source_cards": [1, 2, 3],
+    "evidence": [{{"text": "标题：视频灵魂，精准点题", "card": 1}}, {{"text": "拍摄地点：给故事搭舞台", "card": 2}}]}}
 ]}}
-（这 10 个要素共同支撑"脚本结构"这一个方法，是同一个知识对象 → 合并 1 条；不要给每个要素各开一条。）
+（10 个要素共同支撑"脚本结构"一个方法 → 合并 1 条；source_cards 列出所有相关卡片。）
 </example>
 <example>  多个独立知识对象 → 多条片段
-提取内容：① 爆款选题要撕裂大众共识，越让你写完感到羞耻越能引发转发。② 起号初期要垂直，连续发同一垂类系统才能给账号打标签。
+提取内容：【卡片1】爆款选题要撕裂大众共识，越让你写完感到羞耻越能引发转发。【卡片2】起号初期要垂直，连续发同一垂类系统才能给账号打标签。
 输出：{{"items": [
-  {{"title": "羞耻感选题法", "knowledge_types": ["why"],
-    "what": null, "why": "撕裂大众共识、让你写完感到羞耻的选题更能激发情绪和转发", "how": null,
-    "evidence": ["越让你写完感到羞耻越能引发转发"]}},
-  {{"title": "起号初期垂直打标签", "knowledge_types": ["why", "how"],
-    "what": null, "why": "连续发同一垂类，系统才能给账号打标签", "how": "起号初期持续发同一垂直领域内容",
-    "evidence": ["起号初期要垂直", "连续发同一垂类才能让系统打标签"]}}
+  {{"title": "羞耻感选题法", "knowledge_types": ["why"], "what": null,
+    "why": "撕裂大众共识、让你写完感到羞耻的选题更能激发转发", "how": null,
+    "source_cards": [1], "evidence": [{{"text": "越让你写完感到羞耻越能引发转发", "card": 1}}]}},
+  {{"title": "起号初期垂直打标签", "knowledge_types": ["why", "how"], "what": null,
+    "why": "连续发同一垂类系统才能打标签", "how": "起号初期持续发同一垂直领域内容",
+    "source_cards": [2], "evidence": [{{"text": "连续发同一垂类才能让系统打标签", "card": 2}}]}}
 ]}}
 </example>
 </示例>
@@ -61,7 +63,7 @@
 
 <输出>
 只输出一个 JSON 对象：
-{{"items": [{{"title": "...", "knowledge_types": ["what/why/how"], "what": "...或null", "why": "...或null", "how": "...或null", "evidence": ["原句"]}}]}}
+{{"items": [{{"title": "...", "knowledge_types": ["what/why/how"], "what": "...或null", "why": "...或null", "how": "...或null", "source_cards": [卡片号...], "evidence": [{{"text": "原句", "card": 卡片号或null}}]}}]}}
 </输出>
 
 <自查>

pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
   "fastapi>=0.115.0",
   "uvicorn>=0.30.0",
   "pytest>=8.2.0",
+  "imageio-ffmpeg>=0.4.9",
 ]
 
 [tool.pytest.ini_options]

scripts/validate_db.py

@@ -12,7 +12,7 @@ from creation_knowledge.integrations.db import CkStore, _connect
 from creation_knowledge.models import Post
 
 EXPECTED = {
-    "ck_post": {"id", "platform", "url", "raw", "extracted", "screening",
+    "ck_post": {"id", "platform", "url", "raw", "cards", "extracted", "screening",
                 "stage", "created_at", "updated_at"},
     "ck_knowledge_item": {"id", "post_id", "item", "deconstruction",
                           "ingest_payload", "ingest_status", "knowledge_id",

sql/creation_knowledge.sql

@@ -17,6 +17,7 @@ CREATE TABLE IF NOT EXISTS ck_post (
     platform    text,                                 -- xiaohongshu 等
     url         text,                                 -- 原始链接
     raw         jsonb,                                 -- fetch_post_detail 原始响应
+    cards       jsonb,                                 -- 统一卡片列表 [{index,kind,url,timestamp}]（图/帧）
     extracted   jsonb,                                -- extract_content 多模态产物
     screening   jsonb,                                -- screen_post 结果 {passed,score,reason}
     stage       text        NOT NULL DEFAULT 'fetched', -- fetched/extracted/screened/split/done/rejected/failed
@@ -45,4 +46,7 @@ CREATE TABLE IF NOT EXISTS ck_knowledge_item (
 CREATE INDEX IF NOT EXISTS idx_ck_item_post   ON ck_knowledge_item (post_id);
 CREATE INDEX IF NOT EXISTS idx_ck_item_status ON ck_knowledge_item (ingest_status);
 
+-- 迁移（表已存在时补列，幂等）：卡片溯源
+ALTER TABLE ck_post ADD COLUMN IF NOT EXISTS cards jsonb;
+
 -- 回滚：DROP SCHEMA creation_knowledge CASCADE;

tests/test_extractor.py

@@ -69,6 +69,24 @@ def test_parse_plain_json():
     assert captured["json"]["model"]
 
 
+def test_build_messages_labels_each_card():
+    client = GeminiExtractor(api_key="k", http_post=lambda **kw: None)
+    parts = client.build_messages(_post())[1]["content"]
+    labels = [p["text"] for p in parts if p["type"] == "text"]
+    assert any("【卡片1】" in t for t in labels)
+    assert any("【卡片2】" in t for t in labels)
+
+
+def test_parse_per_card_output():
+    fake, _ = _fake_post_returning(json.dumps(
+        {"text": "x", "cards": [{"index": 1, "content": "卡1要点"},
+                                {"index": 2, "content": "卡2要点"}],
+         "is_empty": False}, ensure_ascii=False))
+    out = GeminiExtractor(api_key="k", http_post=fake).extract(_post())
+    assert len(out.cards) == 2
+    assert out.cards[0].index == 1 and out.cards[0].content == "卡1要点"
+
+
 def test_parse_json_with_code_fence():
     fenced = "```json\n" + json.dumps(
         {"text": "x", "from_image": "y", "from_video": "", "is_empty": "false"},

tests/test_pipeline_e2e.py

@@ -22,7 +22,7 @@ def _settings() -> Settings:
         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,
+        ingest_enabled=False, max_cards=12, frames_dir="runtime/frames",
     )
 
 

tests/test_stages.py

@@ -5,6 +5,7 @@ import json
 
 from creation_knowledge.models import (
     Deconstruction,
+    Evidence,
     ExtractedContent,
     KnowledgeItem,
     Post,
@@ -67,6 +68,19 @@ def test_split_consistency_and_drop_empty():
     assert set(it.knowledge_types) == {"what", "how"}  # 与非空字段对齐，why 被剔除
 
 
+def test_split_source_cards_and_evidence():
+    def chat(s, u):
+        return {"items": [{"title": "评论区选题法", "knowledge_types": ["how"],
+                           "what": "null", "why": "null", "how": "收集高赞评论改写",
+                           "source_cards": [1, 3],
+                           "evidence": [{"text": "高赞评论反映痛点", "card": 1},
+                                        "无卡片的纯字符串证据"]}]}
+    it = split_post(_post(), _content(), chat=chat)[0]
+    assert it.source_cards == [1, 3]
+    assert it.evidence[0].card == 1 and it.evidence[0].text
+    assert it.evidence[1].card is None  # 字符串证据 → card=None
+
+
 def test_deconstruct_filters_invalid():
     def chat(s, u):
         return {"stages": ["选题", "脚本", "乱写"],
@@ -84,7 +98,7 @@ def test_deconstruct_filters_invalid():
 def test_assemble_matches_design_doc():
     item = KnowledgeItem(title="评论区选题法", knowledge_types=["what", "how"],
                          what="从评论提取痛点", why=None, how="收集高赞评论改写",
-                         evidence=["原句a"])
+                         source_cards=[2], evidence=[Evidence(text="原句a", card=2)])
     deco = Deconstruction(
         stages=["选题", "脚本"],
         scopes=[Scope(scope_type="form", value="公式框架"),
@@ -100,4 +114,5 @@ def test_assemble_matches_design_doc():
     assert {"scope_type": "form", "value": "公式框架"} in p.scopes
     assert json.loads(p.content) == {"what": "从评论提取痛点", "why": None,
                                      "how": "收集高赞评论改写"}
-    assert "原文证据" in [e["key"] for e in p.custom_ext]
+    keys = [e["key"] for e in p.custom_ext]
+    assert "原文证据" in keys and "来源卡片" in keys

tests/test_video_frames.py

@@ -0,0 +1,46 @@
+"""M11 抽帧测试：用 ffmpeg 生成测试视频，验证 extract_frames 产出帧卡片。
+
+无 ffmpeg 时自动跳过（CI/本地可能没装）。
+"""
+from __future__ import annotations
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+
+def _ffmpeg() -> str | None:
+    try:
+        import imageio_ffmpeg
+        return imageio_ffmpeg.get_ffmpeg_exe()
+    except Exception:
+        return shutil.which("ffmpeg")
+
+
+FF = _ffmpeg()
+
+
+@pytest.mark.skipif(not FF, reason="ffmpeg 不可用")
+def test_extract_frames_from_generated_clip(tmp_path):
+    from creation_knowledge.integrations.video_frames import extract_frames
+    from creation_knowledge.models import Card
+
+    clip = tmp_path / "clip.mp4"
+    subprocess.run(
+        [FF, "-y", "-f", "lavfi", "-i",
+         "testsrc=duration=6:size=320x240:rate=10", str(clip)],
+        capture_output=True,
+    )
+    assert clip.exists()
+
+    cards = extract_frames(str(clip), out_dir=str(tmp_path / "f"),
+                           url_prefix="/frames/x", max_frames=8)
+    assert len(cards) >= 1
+    assert all(isinstance(c, Card) and c.kind == "frame" for c in cards)
+    assert all(c.url.startswith("/frames/x/") for c in cards)
+    # 帧文件确实生成
+    assert all((tmp_path / "f" / Path(c.url).name).exists() for c in cards)
+    # 1-based 连续索引
+    assert [c.index for c in cards] == list(range(1, len(cards) + 1))

web/index.html

@@ -85,6 +85,16 @@
   .pmx{margin-left:auto;cursor:pointer;color:var(--mut);font-size:16px;line-height:1}
   .pmlabel{font-size:12px;color:var(--mut);font-weight:600;margin:12px 0 4px}
   .pmnote{font-size:12px;color:var(--warn);background:#faeeda;padding:6px 10px;border-radius:6px;margin-top:4px}
+  .srcrow{display:flex;gap:8px;align-items:center;flex-wrap:wrap;margin:8px 0}
+  .srcrow .lbl{font-size:12px;color:var(--mut);flex:none}
+  .srcthumb{position:relative}
+  .srcthumb img{width:56px;height:56px;object-fit:cover;border-radius:6px;border:1px solid var(--line);cursor:zoom-in}
+  .srcthumb img:hover{border-color:var(--ac)}
+  .srcthumb .cn{position:absolute;left:2px;bottom:2px;font-size:10px;background:rgba(0,0,0,.6);color:#fff;padding:0 4px;border-radius:4px}
+  .cardthumb{position:relative}
+  .cardthumb .cn{position:absolute;left:3px;bottom:3px;font-size:10px;background:rgba(0,0,0,.6);color:#fff;padding:0 5px;border-radius:4px}
+  .evrow{display:flex;gap:8px;align-items:baseline;margin:5px 0;font-size:13px}
+  .evcard{font-size:11px;padding:1px 7px;border-radius:6px;background:#e6f1fb;color:#185fa5;cursor:zoom-in;flex:none}
 </style>
 </head>
 <body>
@@ -133,13 +143,25 @@ function PromptModal({item,onClose}){
   </div>;
 }
 
-function KPCard({it,idx}){
+function fmtTs(s){s=Math.floor(s||0);return String(Math.floor(s/60)).padStart(2,'0')+":"+String(s%60).padStart(2,'0');}
+
+function KPCard({it,idx,cardMap,onZoom}){
   const item=it.item||{}, deco=it.deconstruction||{}, payload=it.ingest_payload||{};
   const types=(item.knowledge_types||[]).map(t=>t.toUpperCase()).join(" + ");
   const stages=deco.stages||[], scopes=deco.scopes||[];
+  const srcCards=(item.source_cards||[]).map(n=>cardMap[n]).filter(Boolean);
+  const evidence=item.evidence||[];
   return <div className="card">
     <h4 className="kptitle">{idx}. {item.title||"(无标题)"}</h4>
 
+    {srcCards.length>0 && <div className="srcrow">
+      <span className="lbl">来源卡片</span>
+      {srcCards.map(c=><span className="srcthumb" key={c.index}>
+        <img src={c.url} title={"卡片"+c.index} onClick={()=>onZoom(c.url)} onError={e=>e.target.style.display='none'}/>
+        <span className="cn">{c.kind==="frame"&&c.timestamp!=null?fmtTs(c.timestamp):"卡片"+c.index}</span>
+      </span>)}
+    </div>}
+
     <div className="part">
       <div className="phead"><span className="pn">1</span>知识类型拆解<span className="ktype">{types||"—"}</span></div>
       {item.what&&<div className="whh w"><b>What · 是什么</b><div>{item.what}</div></div>}
@@ -168,8 +190,17 @@ function KPCard({it,idx}){
     </div>
 
     <div className="foot">
-      {(item.evidence&&item.evidence.length>0)&&
-        <details><summary>原文证据（{item.evidence.length}）</summary><pre>{item.evidence.join("\n")}</pre></details>}
+      {evidence.length>0 &&
+        <details><summary>原文证据（{evidence.length}）</summary>
+          <div>{evidence.map((e,i)=>{
+            const text=typeof e==="string"?e:(e.text||"");
+            const card=(typeof e==="object"&&e)?e.card:null;
+            const c=card!=null?cardMap[card]:null;
+            return <div className="evrow" key={i}>
+              {c&&<span className="evcard" onClick={()=>onZoom(c.url)}>卡片{card}</span>}
+              <span>{text}</span></div>;
+          })}</div>
+        </details>}
       <details><summary>入库数据（ingest payload）</summary><pre>{JSON.stringify(payload,null,2)}</pre></details>
     </div>
   </div>;
@@ -186,15 +217,21 @@ function Pipeline({postId,onZoom,prompts,onPrompt}){
   if(!post) return <div className="empty">加载中…</div>;
   const inner=(post.raw&&post.raw.data&&post.raw.data.data)||{};
   const scr=post.screening||{};
-  const imgs=(inner.image_url_list||[]).map(x=>x.image_url||x);
+  const rawImgs=(inner.image_url_list||[]).map(x=>x.image_url||x);
+  // 统一卡片：优先 post.cards（含图/帧），旧数据回退 image_url_list
+  const cards=(post.cards&&post.cards.length)? post.cards
+    : rawImgs.map((u,i)=>({index:i+1,kind:"image",url:u}));
+  const cardMap={}; cards.forEach(c=>{cardMap[c.index]=c;});
   return <div>
     <Stage n="1" title="原文">
       <a className="link" href={post.url} target="_blank" rel="noreferrer">🔗 打开原帖</a>
-      <p className="kv">作者 {inner.channel_account_name||"—"} · 类型 {inner.content_type||"—"}</p>
+      <p className="kv">作者 {inner.channel_account_name||"—"} · 类型 {inner.content_type||"—"} · {cards.length} 张卡片</p>
       <div><b>{inner.title||"(无标题)"}</b></div>
-      <pre>{inner.body_text||"(正文为空，知识在图片里)"}</pre>
-      <div className="imgs">{imgs.map((u,i)=><img key={i} src={u} title="点击放大"
-        onClick={()=>onZoom(u)} onError={e=>e.target.style.display='none'}/>)}</div>
+      <pre>{inner.body_text||"(正文为空，知识在卡片里)"}</pre>
+      <div className="imgs">{cards.map(c=><span className="cardthumb" key={c.index}>
+        <img src={c.url} title={"卡片"+c.index} onClick={()=>onZoom(c.url)} onError={e=>e.target.style.display='none'}/>
+        <span className="cn">{c.index}{c.kind==="frame"&&c.timestamp!=null?" · "+fmtTs(c.timestamp):""}</span>
+      </span>)}</div>
     </Stage>
 
     <Stage n="2" title="筛选" sub="判断这篇帖子值不值得提取成创作知识"
@@ -222,7 +259,7 @@ function Pipeline({postId,onZoom,prompts,onPrompt}){
         {prompts.split && <PromptBtn label="拆分提示词" onClick={()=>onPrompt('split')}/>}
         {prompts.deconstruct && <PromptBtn label="解构提示词" onClick={()=>onPrompt('deconstruct')}/>}
       </React.Fragment>}>
-      {items.length? items.map((it,i)=><KPCard key={i} it={it} idx={i+1}/>) : <div className="kv">无（被淘汰或未拆出）</div>}
+      {items.length? items.map((it,i)=><KPCard key={i} it={it} idx={i+1} cardMap={cardMap} onZoom={onZoom}/>) : <div className="kv">无（被淘汰或未拆出）</div>}
     </Stage>
   </div>;
 }

创作知识-重构设计.md

@@ -419,10 +419,10 @@ for url in post_urls:
 当前版本只做：
 
 1. 从帖子链接开始。
-2. 拉取帖子详情（文本 + 图片 + 视频）。
-3. 多模态内容理解：文本、图片、视频交给 Gemini（gemini-3-flash-preview） 提取真正内容，不只读 `body_text`。
+2. 拉取帖子详情（文本 + 图片 + 视频），把图文每张图、视频每帧统一成「卡片」（见 §10）。
+3. 多模态内容理解：文本 + 各卡片交给 Gemini（gemini-3-flash-preview），**按卡片归因**提取真正内容，不只读 `body_text`。
 4. 筛选有效帖子。
-5. 拆分 What / Why / How。
+5. 拆分 What / Why / How，每条知识记录它出自哪些卡片（source_cards）、证据链接到卡片。
 6. 解构阶段和作用域。
 7. 组装并调用入库接口。
 
@@ -434,3 +434,37 @@ for url in post_urls:
 4. 作者主页分析。
 5. LangGraph。
 6. 自动改写五棵分类树。
+
+## 10. 卡片溯源（card traceability）
+
+让每条知识能追溯到它出自帖子的哪些「卡片」，运营/审计可对着原图原帧核验。
+
+### 10.1 卡片（card）抽象
+
+把一篇帖子的视觉单元统一成「卡片」，下游只认卡片号（1-based）：
+
+- 图文帖：`image_url_list` 里**每张图 = 一张卡片**（卡片1、卡片2…）。
+- 视频帖：**抽帧**后**每一帧 = 一张卡片**（卡片号 + 时间戳，见 §10.4）。
+
+卡片结构：`{index, kind: image|frame, url, timestamp?}`（timestamp 仅帧有）。
+
+### 10.2 归因发生在「多模态提取」步
+
+只有提取这一步看得到图/帧。提取时给每张卡片打 `【卡片N】` 标签，让 Gemini **按卡片输出**知识：`cards: [{index, content}]`。拆分步据此把每条知识标上 `source_cards`，并把每条 `evidence` 写成 `{text, card}`（纯正文证据 card=None）。
+
+### 10.3 入库映射
+
+`source_cards` 与带卡片号的 `evidence` 进 `custom_ext`（「来源卡片」「原文证据」），不破坏 ingest 契约。
+
+### 10.4 视频抽帧
+
+视频帖无现成卡片，需抽帧生成：
+
+- **主策略**：场景切换检测（ffmpeg `select='gt(scene,0.3)'`），知识类视频常是字幕卡/场景切换，正好对应卡片；解析每帧时间戳。
+- **护栏**：相邻帧最小间隔 1.5s 去近重复；最多 `MAX_CARDS`(默认12) 帧控成本；下采样长边 ≤720。
+- **回退**：场景帧不足 4 帧时，按时长均匀采样约 8 帧。
+- 每帧 → 卡片（带时间戳）。**已知限制**：抽帧只覆盖视觉，不含口播语音；语音的 ASR/字幕作为后续项，本版不做。
+
+### 10.5 卡片上限
+
+提取一次最多送 `MAX_CARDS`（默认 12，env `CK_MAX_CARDS` 可调）张卡片；超出截断并记日志（不静默丢卡）。解决了早期 `MAX_IMAGES=6` 漏卡的问题。

技术文档/开发顺序.md

@@ -51,9 +51,11 @@ tests/fixtures/               # 5 个 xhs_case_*.json
 tests/{test_crawler,test_extractor,test_stages,test_pipeline_e2e}.py
 ```
 
-## 2. 开发顺序（9 个里程碑）
+## 2. 开发顺序（里程碑）
 
 > 每个里程碑都能独立验证后再进下一个。函数签名以 `models.py` 的 pydantic 类型为准。
+>
+> **进度（as-built）**：M0–M8 ✅ 已全量实现并真机验证（拉取→多模态→筛选→拆分→解构→组装→落库 + FastAPI + 暖黄单页 React + 提示词 v4 + `/api/prompts`）。M9 多 sub-agent 审计 ⏳ 待做。M10 卡片溯源 + M11 视频抽帧 ✅ 已实现（见下，含 `MAX_IMAGES→MAX_CARDS`）。详细的 UI/API/提示词/部署 as-built 见 [技术架构.md](技术架构.md) §9–§12。
 
 ### M0 · 脚手架
 - **文件**：`pyproject.toml`(uv)、包目录骨架、`creation_knowledge/config.py`、`creation_knowledge/models.py`。
@@ -109,6 +111,18 @@ tests/{test_crawler,test_extractor,test_stages,test_pipeline_e2e}.py
   多数表决，写 `ck_audit(post_id, item_id, dimension, verdict, reason, model, ext jsonb)`，web 标红/标绿。
 - **验证**：对「海狸 HOW+WHAT」判通过、对「彩虹草原」判否决。
 
+### M10 · 卡片溯源 ✅（已实现）
+- **文件**：`models.py`(Card/cards/source_cards/Evidence)、`crawler.py`(图文 cards)、`extractor.py`+`prompts/extract.txt`(按【卡片N】归因)、`stages/split.py`+`prompts/split.txt`(source_cards+evidence{text,card})、`assemble.py`(写 custom_ext)、`db.py`(cards 列)、`api.py`(返回 cards)、`web/index.html`(来源卡片缩略图+证据跳卡片)、`sql`(ALTER add cards)。
+- **统一卡片抽象**：图文每张图、视频每帧 = 卡片（1-based）；下游只认卡片号。
+- **验证**：`tests/test_stages.py::test_split_source_cards_and_evidence`、`test_extractor.py::test_parse_per_card_output`；真机重跑 5 帖，知识卡内嵌来源卡片、证据可点开。
+
+### M11 · 视频抽帧 ✅（已实现）
+- **文件**：`integrations/video_frames.py::extract_frames`；`pipeline.py`(fetch 后抽帧补 cards)；`config.py`(max_cards/frames_dir)；`api.py`(/frames 静态)；`pyproject`(imageio-ffmpeg)。
+- **策略**：ffmpeg 场景切换检测(0.3) + 最小间隔 1.5s/最多 MAX_CARDS 帧/下采样≤720 + 不足 4 帧回退均匀采样约 8 帧；帧带时间戳。
+- **MAX_IMAGES→MAX_CARDS**：默认 12，env `CK_MAX_CARDS`，超限截断记日志；修复 9 卡帖漏卡。
+- **验证**：`tests/test_video_frames.py`（生成测试视频抽帧，无 ffmpeg 自动跳过）。
+- **限制**：只覆盖视觉，口播语音 ASR/字幕为后续项。
+
 ## 3. 可延展点（现在不做，留好接口）
 
 - **多平台**：`fetch_post_detail` 按 `platform` 分发不同 crawler path，下游环节不动。

技术文档/技术架构.md

@@ -178,3 +178,58 @@ for url in urls:
 2. 不引入 LangGraph / 消息队列 / 微服务。
 3. 不直接写 KnowHub 底层表，也不改数据工程的五棵分类树。
 4. 不为结构完整而编造知识；原文（含图/视频）没有的不补。
+
+---
+
+> 以下 §9–§12 为 as-built（已实现并真机验证），补齐之前滞后的文档。
+
+## 9. Web 可视化（as-built）
+
+单页 React（CDN 引入 React + Babel，**零构建**），由 FastAPI 同实例托管，文件 `web/index.html`。
+
+- **主题**：暖黄背景 `#f3efe4` + 白卡 + 柔和文字色（CSS 变量 `--bg/--panel/--what/--why/--how`）。
+- **布局**：左侧帖子列表（stage + 知识条数），右侧选中帖子的流水线：① 原文 → ② 筛选 → ③ 知识点（N 条）。
+- **知识卡**：一条知识一张卡，卡内三段——「来源卡片」缩略图条 + ① 知识类型拆解（What/Why/How 分色）+ ② 解构·阶段 + ③ 解构·作用域（五棵树分色）；底部折叠「原文证据」「入库数据」。
+- **交互**：图片/卡片点击放大（lightbox）；筛选/知识点环节带「📄 提示词」按钮弹窗看该环节提示词；证据上的「卡片N」可点开对应卡片。
+
+## 10. API 与提示词系统（as-built）
+
+FastAPI（`creation_knowledge/api.py`），`uvicorn creation_knowledge.api:app`：
+
+| 路由 | 用途 |
+|---|---|
+| `GET /api/posts` | 帖子列表（id/platform/stage/item_count） |
+| `GET /api/posts/{id}` | 帖子详情（raw/cards/extracted/screening/stage） |
+| `GET /api/posts/{id}/items` | 该帖知识片段（item/deconstruction/ingest_payload/status） |
+| `GET /api/prompts` | 各环节提示词（system+user+model+version），供前端弹窗 |
+| `/frames/*`（静态） | 视频帧文件（card.url 指向） |
+| `/`（静态） | 单页前端 |
+
+**提示词系统**：四个环节的 user 模板在 `prompts/{extract,screen,split,deconstruct}.txt`（`.format()` 占位 + `{{}}` 转义）；各 stage 的 system 角色串在代码里（`screen.py`/`split.py`/`deconstruct.py` 的 `SYSTEM`，extract 的 `_SYSTEM_PROMPT`）；`prompts.py::PROMPT_VERSION` 手动维护版本号，随产物可溯。改提示词只动 `prompts/*.txt`，GUI 即时可见（`/api/prompts` 每次读盘）。
+
+## 6'. 运行与部署（as-built，修订 §6.2）
+
+- **跑流水线**：`PYTHONPATH=. python -m creation_knowledge.cli run --urls <...>`（或 `scripts/run_batch.py`）。
+- **起 Web/API**：`CK_ENV_FILE=<.env> PYTHONPATH=. uvicorn creation_knowledge.api:app --host 0.0.0.0 --port 8900`。
+- **云端**：开发机连不到内网 GitLab，代码走 **rsync** 上机；复用 `~/demand-agent-new/.venv`（自带 httpx/pydantic/psycopg2/fastapi/uvicorn/imageio-ffmpeg）；`.env` 用 `~/ContentFindAgentNew/.env`（同库同 key）。
+- **本机看 Web**：`ssh -i <pem> -N -L 8900:localhost:8900 sam@47.245.103.121` 后开 http://localhost:8900。
+
+## 11. 卡片溯源（card traceability）
+
+把图文每张图、视频每帧统一成「卡片」（1-based），每条知识溯源到来源卡片。详见业务设计 §10。
+
+- 数据模型：`Post.cards: [Card{index,kind,url,timestamp}]`；`ExtractedContent.cards: [{index,content}]`；`KnowledgeItem.source_cards: [int]`；`evidence: [{text, card}]`。
+- 归因在提取步：`extractor` 给每卡打 `【卡片N】` 标签，Gemini 按卡输出；`split` 据此填 source_cards + evidence.card。
+- 存储：`ck_post` 加 `cards jsonb` 列（迁移 `ALTER TABLE … ADD COLUMN IF NOT EXISTS cards jsonb`）；source_cards/evidence 在 `ck_knowledge_item.item` JSONB 内，无需改列。
+- 前端：知识卡内嵌来源卡片缩略图、证据→卡片可点开。
+
+## 12. 视频抽帧（video frame extraction）
+
+`integrations/video_frames.py::extract_frames`，把视频转成帧卡片：
+
+- 主策略：ffmpeg 场景切换检测（`select='gt(scene,0.3)'`）+ showinfo 解析时间戳。
+- 护栏：最小间隔 1.5s、最多 `MAX_CARDS` 帧、长边下采样 ≤720。
+- 回退：场景帧 <4 时按时长均匀采样约 8 帧。
+- 依赖 `imageio-ffmpeg`（自带 ffmpeg 二进制）；帧存 `runtime/frames/<post_id>/`，经 `/frames` 静态服务。
+- pipeline 在 fetch 后、若 `post.video_urls` 非空则抽帧补入 `post.cards`（best-effort，失败不阻塞图文路径）。
+- 已知限制：只覆盖视觉，不含口播语音（ASR/字幕为后续项）。`MAX_CARDS`（默认 12，env `CK_MAX_CARDS`）统一控制图片卡/抽帧预算，取代旧 `MAX_IMAGES=6`。