视频改为原生整段视频（取代抽帧）+ 段卡溯源

把视频「时间段」当作第三种 card（image/frame/segment）：
- models.py：CardKind 加 segment；Card 加 start/end，url 改可空
- integrations/video_extract.py：mp4→base64→OpenRouter video_url→gemini-3-flash-preview
  返回带时间戳 segments → 段卡(post.cards) + ExtractedContent.cards；下载可注入/可传本地文件
- prompts/extract_video.txt：按时间戳分段提炼 What/Why/How
- pipeline：提取分发（video→原生 / 图文→逐图），移除抽帧步；视频段卡在提取后落库
- video_frames.py 降级为可选缩略图/兜底，不接主流程
- config：douyin_ratio(默认540p)；api /api/prompts 增 extract_video；PROMPT_VERSION v5
- web：段卡渲染为时间段 chip（00:18–01:44），3 处 img src=c.url 加 url 守卫，证据→片段
- docs：架构 §12 重写为原生视频+网络分裂约束、设计 §10.4、开发顺序 M11
- 测试：+test_video_extract（断言段卡 start/end + base64 请求）；23 passed
- 真机：抖音视频走完整流水线落库 stage=done，4 段卡 kind=segment，item.source_cards=[1] 溯源到段

注：原生视频走 OPENROUTER_API_KEY（无直连 GEMINI key）；网络分裂（抖音下载需非新加坡出口、
Gemini 需可达 Google）由用户后续解决，本次用本地 mp4 注入验证。

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

creation_knowledge/api.py

@@ -70,8 +70,10 @@ def get_prompts() -> dict:
     return {
         "version": PROMPT_VERSION,
         "items": [
-            {"key": "extract", "label": "多模态提取", "model": s.video_model,
+            {"key": "extract", "label": "图文提取", "model": s.video_model,
              "system": extractor_mod._SYSTEM_PROMPT, "user": load_prompt("extract")},
+            {"key": "extract_video", "label": "视频提炼（原生整段）", "model": s.video_model,
+             "system": "（无独立 system，提示词自含）", "user": load_prompt("extract_video")},
             {"key": "screen", "label": "筛选", "model": s.llm_model,
              "system": screen_stage.SYSTEM, "user": load_prompt("screen")},
             {"key": "split", "label": "拆分", "model": s.llm_model,

creation_knowledge/config.py

@@ -89,6 +89,7 @@ class Settings:
     # 卡片 / 抽帧
     max_cards: int
     frames_dir: str
+    douyin_ratio: str  # 视频下载偏好码率（控成本/体积），如 540p
 
     @classmethod
     def from_env(cls, env_file: str | Path = ".env") -> "Settings":
@@ -120,4 +121,5 @@ class Settings:
             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"),
+            douyin_ratio=env_value("CK_DOUYIN_RATIO", file_env, "540p"),
         )

creation_knowledge/integrations/video_extract.py

@@ -0,0 +1,143 @@
+"""原生整段视频提炼：把整段 mp4 经 OpenRouter base64 video_url 发给 Gemini，
+按时间戳分段提炼 What/Why/How。取代抽帧作为视频内容提炼主路。
+
+每个 segment → 一张段卡 Card(kind="segment", start, end)，写入 post.cards；
+同时产出 ExtractedContent.cards=[{index, content}]，让下游 split 照常按【卡片N】溯源。
+
+网络分裂：下载抖音视频需非新加坡出口，调 OpenRouter 需可达 Google。下载做成可注入
+（downloader）或可传入本地文件（video_path），便于"下载节点 / 推理节点"分离。
+"""
+from __future__ import annotations
+
+import base64
+import logging
+import os
+import re
+from typing import Any, Callable, Optional
+
+import httpx
+
+from creation_knowledge.config import Settings
+from creation_knowledge.jsonio import extract_json_object
+from creation_knowledge.models import Card, CardExtract, ExtractedContent, Post
+from creation_knowledge.prompts import load_prompt
+
+logger = logging.getLogger(__name__)
+
+_REFERER = {
+    "douyin": "https://www.douyin.com/",
+    "kuaishou": "https://www.kuaishou.com/",
+    "bilibili": "https://www.bilibili.com/",
+    "shipinhao": "https://channels.weixin.qq.com/",
+}
+_IOS_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15"
+
+
+class VideoExtractError(RuntimeError):
+    pass
+
+
+def _mmss_to_sec(value: Any) -> Optional[float]:
+    """'MM:SS' / 'HH:MM:SS' / 数字 → 秒。"""
+    if value is None:
+        return None
+    if isinstance(value, (int, float)):
+        return float(value)
+    parts = str(value).strip().split(":")
+    try:
+        nums = [float(p) for p in parts]
+    except ValueError:
+        return None
+    sec = 0.0
+    for n in nums:
+        sec = sec * 60 + n
+    return sec
+
+
+def _default_download(url: str, platform: str, timeout: float = 180.0) -> bytes:
+    headers = {"User-Agent": _IOS_UA, "Referer": _REFERER.get(platform, "")}
+    with httpx.stream("GET", url, headers=headers, timeout=timeout,
+                      follow_redirects=True) as r:
+        r.raise_for_status()
+        return b"".join(r.iter_bytes())
+
+
+def _seg_content(seg: dict) -> str:
+    parts = [seg.get("title") or ""]
+    for label, key in (("What", "what"), ("Why", "why"), ("How", "how")):
+        v = seg.get(key)
+        if v and str(v).strip().lower() not in ("null", "none", ""):
+            parts.append(f"{label}：{v}")
+    return "。".join(p for p in parts if p)
+
+
+def extract_video(
+    post: Post,
+    *,
+    settings: Settings,
+    http_post: Callable[..., Any] = httpx.post,
+    video_path: Optional[str] = None,
+    downloader: Optional[Callable[[str, str], bytes]] = None,
+    timeout: float = 600.0,
+) -> ExtractedContent:
+    """对视频帖做原生整段提炼，就地写好 post.cards（段卡），返回 ExtractedContent。"""
+    key = settings.openrouter_api_key
+    if not key:
+        raise VideoExtractError("missing OPENROUTER_API_KEY")
+
+    # 1) 拿到 mp4 字节
+    if video_path and os.path.exists(video_path):
+        data = open(video_path, "rb").read()
+        logger.info("video_extract 用本地文件 %s (%d bytes)", video_path, len(data))
+    else:
+        if not post.video_urls:
+            raise VideoExtractError(f"post {post.id} 无 video_urls 且未提供 video_path")
+        url = post.video_urls[0]
+        if post.platform == "douyin" and "ratio=" in url:  # 偏好较小码率控成本
+            url = re.sub(r"ratio=[^&]+", f"ratio={settings.douyin_ratio}", url)
+        dl = downloader or _default_download
+        try:
+            data = dl(url, post.platform)
+        except Exception as exc:
+            raise VideoExtractError(f"视频下载失败: {exc}") from exc
+    if not data:
+        raise VideoExtractError("视频字节为空")
+
+    # 2) base64 + 3) OpenRouter 原生视频
+    data_url = "data:video/mp4;base64," + base64.b64encode(data).decode()
+    prompt = load_prompt("extract_video").format()
+    body = {"model": settings.video_model, "messages": [{"role": "user", "content": [
+        {"type": "text", "text": prompt},
+        {"type": "video_url", "video_url": {"url": data_url}},
+    ]}]}
+    try:
+        resp = http_post(
+            f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
+            headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
+            json=body, timeout=timeout)
+        resp.raise_for_status()
+        content = resp.json()["choices"][0]["message"]["content"]
+    except httpx.HTTPError as exc:
+        raise VideoExtractError(f"openrouter_http_error: {exc}") from exc
+    except (KeyError, IndexError, TypeError, ValueError) as exc:
+        raise VideoExtractError(f"openrouter_response_invalid: {exc}") from exc
+
+    obj = extract_json_object(content)
+    segments = obj.get("segments") or []
+
+    # 4) 段卡 + 每段内容
+    cards: list[Card] = []
+    card_extracts: list[CardExtract] = []
+    for i, seg in enumerate(segments, start=1):
+        if not isinstance(seg, dict):
+            continue
+        cards.append(Card(index=i, kind="segment",
+                          start=_mmss_to_sec(seg.get("start")),
+                          end=_mmss_to_sec(seg.get("end"))))
+        card_extracts.append(CardExtract(index=i, content=_seg_content(seg)))
+    post.cards = cards  # 就地写入段卡，供 upsert 落库 + 前端展示
+    return ExtractedContent(
+        text=str(obj.get("overall") or obj.get("video_title") or ""),
+        cards=card_extracts,
+        is_empty=len(card_extracts) == 0,
+    )

creation_knowledge/models.py

@@ -8,16 +8,19 @@ from pydantic import BaseModel, Field
 KnowledgeType = Literal["what", "why", "how"]
 ScopeType = Literal["substance", "form", "feeling", "effect", "intent"]
 Stage = Literal["灵感", "选题", "脚本"]
-CardKind = Literal["image", "frame"]
+CardKind = Literal["image", "frame", "segment"]
 
 
 class Card(BaseModel):
-    """一张"卡片"：图文帖的一张图，或视频抽出的一帧。下游溯源只认 index。"""
+    """一张"卡片"：图文帖的一张图、视频抽出的一帧，或原生视频提炼出的一个时间段。
+    下游溯源只认 index（1-based）。"""
 
-    index: int  # 帖内顺序号，从 0 起
+    index: int
     kind: CardKind = "image"
-    url: str
+    url: Optional[str] = None  # 段卡可无图；图片/帧有 url
     timestamp: Optional[float] = None  # 仅 frame：该帧在视频中的秒数
+    start: Optional[float] = None  # 仅 segment：起始秒
+    end: Optional[float] = None  # 仅 segment：结束秒
 
 
 class Post(BaseModel):

creation_knowledge/pipeline.py

@@ -12,6 +12,7 @@ from creation_knowledge.integrations.crawler import CrawlerError, fetch_post_det
 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.integrations.video_extract import VideoExtractError, extract_video
 from creation_knowledge.ingest import IngestError, ingest as real_ingest
 from creation_knowledge.models import ExtractedContent, Post
 from creation_knowledge.stages import (
@@ -42,30 +43,17 @@ def _process_one(
         post = fetch_fn(url)
     except CrawlerError as exc:
         return {"url": url, "status": "fetch_failed", "error": str(exc)}
+    store.upsert_post(post)  # 图文卡片在此落；视频段卡在提取后补落
 
-    # 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) 多模态提取
+    # 2) 多模态提取（图文=逐图；视频=原生整段视频→段卡，extract 内写 post.cards）
     try:
         content = extract_fn(post)
-        store.set_extracted(post.id, content.model_dump())  # stage=extracted
-    except ExtractorError as exc:
+    except (ExtractorError, VideoExtractError) as exc:
         store.update_stage(post.id, "failed")
         return {"url": url, "post_id": post.id, "status": "extract_failed", "error": str(exc)}
+    if post.cards:
+        store.upsert_post(post)  # 视频段卡落库（图文重复 upsert 无害）
+    store.set_extracted(post.id, content.model_dump())  # stage=extracted
 
     # 3) 筛选
     screening = screen_post(post, content, chat=chat)
@@ -110,7 +98,16 @@ def run_pipeline(
     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
+    if extract_fn is None:
+        _image_client = GeminiExtractor.from_env(env_file=env_file)
+
+        def _dispatch_extract(post: Post) -> ExtractedContent:
+            # 视频帖 → 原生整段视频（OpenRouter base64）；图文帖 → 逐图提取
+            if (post.content_type or "").lower() == "video" or post.video_urls:
+                return extract_video(post, settings=settings)
+            return _image_client.extract(post)
+
+        extract_fn = _dispatch_extract
     chat = chat or default_chat(env_file)
 
     results = []

creation_knowledge/prompts.py

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

prompts/extract_video.txt

@@ -0,0 +1,23 @@
+你在从一条短视频里提炼「创作知识」——能指导别人怎么做内容的方法、原理、清单。
+
+<什么算创作知识>
+能迁移、能复用、能教别人"怎么做内容"的东西（选题法、爆款结构、起号逻辑、脚本/文案/镜头技巧、为什么某种做法有效）。不算的：一条具体作品本身、纯展示内容。
+</什么算创作知识>
+
+<工作要点>
+- 视频是口播 + 画面 + 字幕，请**同时听口播、看画面与字幕**——口播里往往是核心知识，别只看画面。
+- 以视频实际内容为准，忠实提炼，原文没有的不要编造、不要拔高。
+- **按视频时间顺序分段**，每段给时间戳和这段讲到的创作知识；一段对应一个相对完整的知识点/小主题。
+- 每段标 What/Why/How（这段讲的是"是什么/有哪些"、"为什么有效"、"怎么做"），有哪个填哪个，没有的填 null。
+</工作要点>
+
+<输出>
+只输出一个 JSON 对象：
+{{"video_title": "视频在讲什么的简短标题",
+  "overall": "全片在教什么，一两句话",
+  "segments": [
+    {{"start": "MM:SS", "end": "MM:SS", "title": "这段小标题",
+      "knowledge_types": ["what 和/或 why 和/或 how"],
+      "what": "或 null", "why": "或 null", "how": "或 null"}}
+  ]}}
+</输出>

scripts/smoke_video_native.py

@@ -0,0 +1,115 @@
+"""一次性实验：抖音视频 → 直连 Gemini 原生视频 → 按时间戳分段提炼创作知识。
+
+验证"整段视频交给 Gemini 能否详细提炼"，对比抽帧。不接流水线，仅评估。
+直连 Gemini File API（GEMINI_API_KEY），不走 OpenRouter（其对 Gemini 视频仅支持 YouTube）。
+
+用法：python scripts/smoke_video_native.py [env_file] [content_id]
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+
+import httpx
+
+BASE = "https://generativelanguage.googleapis.com"
+DETAIL = "http://crawler.aiddit.com/crawler/dou_yin/detail"
+IOS_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15"
+
+PROMPT = (
+    "你在从一条短视频里提炼「创作知识」——能指导别人怎么做内容的方法/原理/清单。\n"
+    "视频是口播+画面，请**同时听口播、看画面与字幕**。按视频时间顺序分段，"
+    "每段给时间戳和这段讲到的创作知识。原文（含口播）没有的不要编造。只输出 JSON：\n"
+    '{"video_title": "", "overall": "全片在教什么，一两句话",\n'
+    ' "segments": [{"start": "MM:SS", "end": "MM:SS", "title": "这段小标题",\n'
+    '   "knowledge_types": ["what/why/how"], "what": "或null", "why": "或null", "how": "或null"}]}'
+)
+
+
+def load_env(path):
+    env = {}
+    for line in open(path):
+        line = line.strip()
+        if line and not line.startswith("#") and "=" in line:
+            k, v = line.split("=", 1)
+            env[k.strip()] = v.strip().strip('"').strip("'")
+    return env
+
+
+def main():
+    args = sys.argv[1:]
+    env_file = next((a for a in args if a.endswith(".env")), ".env")
+    video_file = next((a for a in args if a.endswith((".mp4", ".bin"))), None)
+    cid = next((a for a in args if not a.endswith((".env", ".mp4", ".bin"))),
+               "7612631899479648866")
+    env = load_env(env_file)
+    key = env.get("GEMINI_API_KEY") or os.environ.get("GEMINI_API_KEY")
+    if not key:
+        raise SystemExit("缺 GEMINI_API_KEY（env_file 或环境变量）")
+    model = (env.get("CONTENT_AGENT_VIDEO_LLM_MODEL") or "gemini-3-flash-preview").split("/")[-1]
+
+    if video_file:  # 已有本地 mp4（如别处下载后传来），跳过抓取
+        data = open(video_file, "rb").read()
+        print(f"[1-2] 使用本地视频 {video_file}：{len(data)} bytes ({len(data)//1024//1024} MB)")
+    else:
+        print(f"[1] 取新鲜视频直链 content_id={cid}")
+        inner = httpx.post(DETAIL, json={"content_id": cid}, timeout=40).json()["data"]["data"]
+        vurl = inner["video_url_list"][0]["video_url"]
+        print("    title:", inner.get("title"), "| dur:",
+              inner["video_url_list"][0].get("video_duration"), "s")
+        print("[2] 下载 mp4")
+        with httpx.stream("GET", vurl, headers={"User-Agent": IOS_UA, "Referer": "https://www.douyin.com/"},
+                          timeout=180, follow_redirects=True) as r:
+            r.raise_for_status()
+            data = b"".join(r.iter_bytes())
+        print(f"    {len(data)} bytes ({len(data)//1024//1024} MB)")
+
+    print("[3] File API 上传")
+    start = httpx.post(
+        f"{BASE}/upload/v1beta/files?key={key}",
+        headers={"X-Goog-Upload-Protocol": "resumable", "X-Goog-Upload-Command": "start",
+                 "X-Goog-Upload-Header-Content-Length": str(len(data)),
+                 "X-Goog-Upload-Header-Content-Type": "video/mp4",
+                 "Content-Type": "application/json"},
+        json={"file": {"display_name": "douyin_" + cid}}, timeout=60)
+    start.raise_for_status()
+    up_url = start.headers["X-Goog-Upload-URL"]
+    up = httpx.post(up_url, headers={"X-Goog-Upload-Offset": "0",
+                                     "X-Goog-Upload-Command": "upload, finalize"},
+                    content=data, timeout=600)
+    up.raise_for_status()
+    fobj = up.json()["file"]
+    name, uri = fobj["name"], fobj["uri"]
+    print("    file:", name)
+
+    print("[4] 等待处理 ACTIVE")
+    for _ in range(80):
+        f = httpx.get(f"{BASE}/v1beta/{name}?key={key}", timeout=30).json()
+        st = f.get("state")
+        if st == "ACTIVE":
+            break
+        if st == "FAILED":
+            raise SystemExit("文件处理失败: " + json.dumps(f, ensure_ascii=False))
+        time.sleep(3)
+    print("    state:", st)
+
+    print(f"[5] generateContent（model={model}，按时间戳提炼）")
+    body = {"contents": [{"parts": [{"file_data": {"mime_type": "video/mp4", "file_uri": uri}},
+                                    {"text": PROMPT}]}]}
+    for m in [model, "gemini-2.5-flash", "gemini-flash-latest"]:
+        g = httpx.post(f"{BASE}/v1beta/models/{m}:generateContent?key={key}", json=body, timeout=300)
+        if g.status_code == 200:
+            model = m
+            break
+        print(f"    model {m} -> HTTP {g.status_code}: {g.text[:160]}")
+    g.raise_for_status()
+    text = g.json()["candidates"][0]["content"]["parts"][0]["text"]
+    print("    使用模型:", model)
+    print("==== 提炼结果 ====")
+    print(text)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/smoke_video_openrouter.py

@@ -0,0 +1,67 @@
+"""实验B：本地 mp4 → base64 → 经 OpenRouter 把整段视频发给 Gemini，按时间戳提炼。
+
+我们没有直连 GEMINI_API_KEY，Gemini 走 OpenRouter。OpenRouter 对 Gemini 视频支持
+base64 data URL（Vertex 通道）。本脚本验证这条路能否拿到详细的时间戳分段提炼。
+
+用法：python scripts/smoke_video_openrouter.py <env_file> <video.mp4>
+"""
+from __future__ import annotations
+
+import base64
+import sys
+
+import httpx
+
+PROMPT = (
+    "你在从一条短视频里提炼「创作知识」——能指导别人怎么做内容的方法/原理/清单。\n"
+    "视频是口播+画面，请**同时听口播、看画面与字幕**。按视频时间顺序分段，"
+    "每段给时间戳和这段讲到的创作知识。原文（含口播）没有的不要编造。只输出 JSON：\n"
+    '{"video_title": "", "overall": "全片在教什么，一两句话",\n'
+    ' "segments": [{"start": "MM:SS", "end": "MM:SS", "title": "这段小标题",\n'
+    '   "knowledge_types": ["what/why/how"], "what": "或null", "why": "或null", "how": "或null"}]}'
+)
+
+
+def load_env(path):
+    env = {}
+    for line in open(path):
+        line = line.strip()
+        if line and not line.startswith("#") and "=" in line:
+            k, v = line.split("=", 1)
+            env[k.strip()] = v.strip().strip('"').strip("'")
+    return env
+
+
+def main():
+    env_file = next((a for a in sys.argv[1:] if a.endswith(".env")), ".env")
+    video = next((a for a in sys.argv[1:] if a.endswith((".mp4", ".bin"))), None)
+    env = load_env(env_file)
+    key = env.get("OPENROUTER_API_KEY") or env.get("OPEN_ROUTER_API_KEY")
+    base = (env.get("OPENROUTER_BASE_URL") or "https://openrouter.ai/api/v1").rstrip("/")
+    model = env.get("CONTENT_AGENT_VIDEO_LLM_MODEL") or "google/gemini-3-flash-preview"
+    if not key:
+        raise SystemExit("缺 OPENROUTER_API_KEY")
+
+    raw = open(video, "rb").read()
+    print(f"video {video}: {len(raw)} bytes ({len(raw)//1024//1024} MB) -> base64")
+    data_url = "data:video/mp4;base64," + base64.b64encode(raw).decode()
+
+    body = {"model": model, "messages": [{"role": "user", "content": [
+        {"type": "text", "text": PROMPT},
+        {"type": "video_url", "video_url": {"url": data_url}},
+    ]}]}
+    print(f"POST {base}/chat/completions  model={model}  payload≈{len(data_url)//1024//1024}MB")
+    r = httpx.post(f"{base}/chat/completions",
+                   headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
+                   json=body, timeout=600)
+    print("HTTP", r.status_code)
+    if r.status_code != 200:
+        print(r.text[:800]); raise SystemExit(1)
+    out = r.json()
+    print("provider:", out.get("provider"), "| usage:", out.get("usage"))
+    print("==== 提炼结果 ====")
+    print(out["choices"][0]["message"]["content"])
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/smoke_video_pipeline.py

@@ -0,0 +1,72 @@
+"""真机验证：抖音视频走完整流水线（原生提取→筛选→拆分→解构→落库），用本地 mp4 绕过下载网络。
+
+用法：python scripts/smoke_video_pipeline.py <env_file> <video.mp4> [content_id]
+"""
+from __future__ import annotations
+
+import sys
+
+import httpx
+
+from creation_knowledge.config import Settings
+from creation_knowledge.integrations.db import CkStore
+from creation_knowledge.integrations.llm import default_chat
+from creation_knowledge.integrations.video_extract import extract_video
+from creation_knowledge.models import Post
+from creation_knowledge.stages import (
+    build_ingest_payload, deconstruct_item, screen_post, split_post)
+
+
+def main() -> int:
+    args = sys.argv[1:]
+    env_file = next((a for a in args if a.endswith(".env")), ".env")
+    video = next((a for a in args if a.endswith((".mp4", ".bin"))), None)
+    cid = next((a for a in args if not a.endswith((".env", ".mp4", ".bin"))),
+               "7612631899479648866")
+    settings = Settings.from_env(env_file)
+    chat = default_chat(env_file)
+
+    det = httpx.post("http://crawler.aiddit.com/crawler/dou_yin/detail",
+                     json={"content_id": cid}, timeout=40).json()["data"]["data"]
+    post = Post(id=f"dy_{cid}", platform="douyin",
+                url=f"https://www.douyin.com/video/{cid}", content_id=cid,
+                title=det.get("title") or "", content_type="video",
+                body_text=det.get("body_text") or "",
+                video_urls=[det["video_url_list"][0]["video_url"]],
+                raw={"data": {"data": det}})
+
+    store = CkStore(settings.pg)
+    store.delete_post(post.id)
+    store.upsert_post(post)
+
+    print("[extract] 原生整段视频 …")
+    content = extract_video(post, settings=settings, video_path=video)
+    store.upsert_post(post)
+    store.set_extracted(post.id, content.model_dump())
+    print(f"  段数={len(post.cards)}  overall={content.text[:50]}")
+    print(f"  段卡示例={[(c.index, c.start, c.end) for c in post.cards[:3]]}")
+
+    scr = screen_post(post, content, chat=chat)
+    store.set_screening(post.id, scr.model_dump())
+    print(f"[screen] passed={scr.passed} score={scr.score}")
+    if scr.passed:
+        items = split_post(post, content, chat=chat)
+        print(f"[split] {len(items)} 条知识")
+        for it in items:
+            deco = deconstruct_item(it, chat=chat)
+            payload = build_ingest_payload(post, it, deco)
+            store.save_item(post.id, it.model_dump(), deco.model_dump(), payload.model_dump())
+        store.update_stage(post.id, "done")
+
+    p = store.read_post(post.id)
+    its = store.read_items(post.id)
+    print(f"[DB] stage={p['stage']} cards={len(p.get('cards') or [])} "
+          f"kinds={set(c['kind'] for c in (p.get('cards') or []))} items={len(its)}")
+    if its:
+        print(f"  item0: source_cards={its[0]['item'].get('source_cards')} "
+              f"evidence0={(its[0]['item'].get('evidence') or [None])[0]}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_pipeline_e2e.py

@@ -23,6 +23,7 @@ def _settings() -> Settings:
         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, max_cards=12, frames_dir="runtime/frames",
+        douyin_ratio="540p",
     )
 
 

tests/test_video_extract.py

@@ -0,0 +1,72 @@
+"""原生整段视频提取离线测试：注入假 http_post 返回 segments，断言段卡 + 内容。"""
+from __future__ import annotations
+
+import json
+
+from creation_knowledge.config import PgConfig, Settings
+from creation_knowledge.integrations.video_extract import extract_video
+from creation_knowledge.models import Post
+
+SEGMENTS = json.dumps({
+    "video_title": "AI视频教程", "overall": "教 AI 出片",
+    "segments": [
+        {"start": "00:00", "end": "00:18", "title": "问题与简介",
+         "knowledge_types": ["what"], "what": "常见翻车", "why": None, "how": None},
+        {"start": "00:19", "end": "00:48", "title": "去噪原理",
+         "knowledge_types": ["why"], "what": None, "why": "噪声去噪类似雕刻", "how": None},
+    ]}, ensure_ascii=False)
+
+
+class _Resp:
+    def __init__(self, content): self._c = content
+    def raise_for_status(self): return None
+    def json(self): return {"choices": [{"message": {"content": self._c}}]}
+
+
+def _settings() -> Settings:
+    return Settings(
+        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
+        crawler_base_url="x", crawler_key="", crawler_timeout=30,
+        video_model="google/gemini-3-flash-preview", gemini_api_key="",
+        openrouter_base_url="https://openrouter.ai/api/v1", openrouter_api_key="k",
+        llm_model="m", knowhub_api="x", ingest_enabled=False, max_cards=12,
+        frames_dir="runtime/frames", douyin_ratio="540p")
+
+
+def _fake_post():
+    captured = {}
+
+    def post(url, headers=None, json=None, timeout=None):
+        captured["url"] = url
+        captured["json"] = json
+        return _Resp(SEGMENTS)
+
+    return post, captured
+
+
+def test_video_extract_makes_segment_cards(tmp_path):
+    vf = tmp_path / "v.mp4"
+    vf.write_bytes(b"\x00\x00\x00\x18ftypmp42fakevideobytes")
+    fake, captured = _fake_post()
+    post = Post(id="dy_1", platform="douyin", url="u", content_id="1",
+                content_type="video", video_urls=["http://x/play?ratio=default"])
+
+    out = extract_video(post, settings=_settings(), http_post=fake, video_path=str(vf))
+
+    # ExtractedContent.cards = 每段一条内容
+    assert len(out.cards) == 2
+    assert out.cards[0].index == 1 and "问题" in out.cards[0].content
+    assert out.is_empty is False and out.text  # overall 进 text
+
+    # post.cards = 段卡，带 start/end（秒）
+    assert len(post.cards) == 2
+    assert post.cards[0].kind == "segment"
+    assert post.cards[0].start == 0.0 and post.cards[1].start == 19.0
+    assert post.cards[1].end == 48.0
+    assert post.cards[0].url is None  # 段卡默认无图
+
+    # 请求体含 base64 video_url
+    parts = captured["json"]["messages"][0]["content"]
+    assert any(p["type"] == "video_url"
+               and p["video_url"]["url"].startswith("data:video/mp4;base64,")
+               for p in parts)

web/index.html

@@ -95,6 +95,7 @@
   .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}
+  .segchip{display:inline-flex;align-items:center;font-size:12px;padding:4px 10px;border-radius:8px;background:#eef4fb;color:#185fa5;border:1px solid #cfe0f5;flex:none}
 </style>
 </head>
 <body>
@@ -144,6 +145,12 @@ function PromptModal({item,onClose}){
 }
 
 function fmtTs(s){s=Math.floor(s||0);return String(Math.floor(s/60)).padStart(2,'0')+":"+String(s%60).padStart(2,'0');}
+function cardTime(c){
+  if(!c) return "";
+  if(c.kind==="segment"&&c.start!=null) return fmtTs(c.start)+"–"+fmtTs(c.end);
+  if(c.kind==="frame"&&c.timestamp!=null) return fmtTs(c.timestamp);
+  return "";
+}
 
 function KPCard({it,idx,cardMap,onZoom}){
   const item=it.item||{}, deco=it.deconstruction||{}, payload=it.ingest_payload||{};
@@ -156,10 +163,12 @@ function KPCard({it,idx,cardMap,onZoom}){
 
     {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>)}
+      {srcCards.map(c=> c.url
+        ? <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">{cardTime(c)||("卡片"+c.index)}</span></span>
+        : <span className="segchip" key={c.index}>片段{c.index}{cardTime(c)?" · "+cardTime(c):""}</span>
+      )}
     </div>}
 
     <div className="part">
@@ -197,7 +206,9 @@ function KPCard({it,idx,cardMap,onZoom}){
             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>}
+              {c&&(c.url
+                ? <span className="evcard" onClick={()=>onZoom(c.url)}>卡片{card}</span>
+                : <span className="evcard" style={{cursor:"default"}} title="视频片段">片段{card}{cardTime(c)?" · "+cardTime(c):""}</span>)}
               <span>{text}</span></div>;
           })}</div>
         </details>}
@@ -228,10 +239,12 @@ function Pipeline({postId,onZoom,prompts,onPrompt}){
       <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">{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>
+      <div className="imgs">{cards.map(c=> c.url
+        ? <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}{cardTime(c)?" · "+cardTime(c):""}</span></span>
+        : <span className="segchip" key={c.index}>片段{c.index}{cardTime(c)?" · "+cardTime(c):""}</span>
+      )}</div>
     </Stage>
 
     <Stage n="2" title="筛选" sub="判断这篇帖子值不值得提取成创作知识"

创作知识-重构设计.md

@@ -469,14 +469,15 @@ for url in post_urls:
 
 `source_cards` 与带卡片号的 `evidence` 进 `custom_ext`（「来源卡片」「原文证据」），不破坏 ingest 契约。
 
-### 10.4 视频抽帧
+### 10.4 视频：原生整段视频（取代抽帧）
 
-视频帖无现成卡片，需抽帧生成：
+视频内容提炼**走原生整段视频**，不抽帧。原因（真机验证）：知识类视频多为口播，知识在**语音**里；抽帧只取画面会丢核心。把整段 mp4 交给 Gemini（经 OpenRouter base64），它能听口播、看画面字幕，按**时间段**输出 What/Why/How。
 
-- **主策略**：场景切换检测（ffmpeg `select='gt(scene,0.3)'`），知识类视频常是字幕卡/场景切换，正好对应卡片；解析每帧时间戳。
-- **护栏**：相邻帧最小间隔 1.5s 去近重复；最多 `MAX_CARDS`(默认12) 帧控成本；下采样长边 ≤720。
-- **回退**：场景帧不足 4 帧时，按时长均匀采样约 8 帧。
-- 每帧 → 卡片（带时间戳）。**已知限制**：抽帧只覆盖视觉，不含口播语音；语音的 ASR/字幕作为后续项，本版不做。
+- 每个时间段 → 一张**段卡**（`kind=segment`，带 start/end 秒）。溯源即"溯源到 02:07–03:43 这一段"。
+- 通道：`OPENROUTER_API_KEY`（无直连 GEMINI key）；模型 `google/gemini-3-flash-preview`；下载偏好 540p 码率控成本。
+- **网络分裂约束**：抖音视频下载需非新加坡出口、调 Gemini 需可达 Google，单机两头难兼得 → 下载与推理可拆节点（代码把下载做成可注入/可传本地文件）。
+- **抽帧降级**：`video_frames.py` 保留为可选缩略图/兜底，不接主流程。
+- **已知限制**：原生默认 1 FPS 采样，快速小字可能漏；段卡默认无缩略图（纯时间段）。
 
 ### 10.5 卡片上限
 

技术文档/开发顺序.md

@@ -116,12 +116,13 @@ tests/{test_crawler,test_extractor,test_stages,test_pipeline_e2e}.py
 - **统一卡片抽象**：图文每张图、视频每帧 = 卡片（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/字幕为后续项。
+### M11 · 视频：原生整段视频 ✅（已实现，取代抽帧）
+- **文件**：`integrations/video_extract.py::extract_video`；`prompts/extract_video.txt`；`pipeline.py`(提取分发：视频→原生，图文→逐图；移除抽帧步)；`config.py`(douyin_ratio)；前端 `web/index.html`(段卡=时间段 chip)。
+- **策略**：整段 mp4 → base64 → OpenRouter `video_url` → `gemini-3-flash-preview` 返回带时间戳的 segments → 段卡(start/end) + 每段内容；下游 split 照常溯源到段。
+- **通道**：`OPENROUTER_API_KEY`（无直连 GEMINI key）；下载偏好 540p；下载可注入/可传本地文件以应对网络分裂。
+- **抽帧降级**：`video_frames.py` 保留为可选缩略图/兜底，不接主流程；`MAX_IMAGES→MAX_CARDS` 仍约束图片卡。
+- **验证**：`tests/test_video_extract.py`（注入假响应断言段卡 start/end + 请求含 base64 video_url）；真机用本地 mp4 注入跑通 extract→split→落库。
+- **限制**：1 FPS 采样快速小字可能漏；段卡默认无缩略图。
 
 ## 3. 可延展点（现在不做，留好接口）
 

技术文档/技术架构.md

@@ -218,18 +218,21 @@ FastAPI（`creation_knowledge/api.py`），`uvicorn creation_knowledge.api:app`
 
 把图文每张图、视频每帧统一成「卡片」（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。
+- 卡片三种 `kind`：`image`（图文每张图）/ `frame`（抽帧，已降级）/ `segment`（原生视频的时间段，带 `start`/`end` 秒、`url` 可空）。
+- 数据模型：`Post.cards: [Card{index,kind,url?,timestamp?,start?,end?}]`；`ExtractedContent.cards: [{index,content}]`；`KnowledgeItem.source_cards: [int]`；`evidence: [{text, card}]`。
+- 归因在提取步：图文给每图打 `【卡片N】` 标签按卡输出；视频原生提炼时模型按时间段输出（每段=一张段卡）；`split` 据此填 source_cards + evidence.card。视频的溯源即"溯源到 02:07–03:43 这一段"。
 - 存储：`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）
+## 12. 视频：原生整段视频提炼（native video）
 
-`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`。
+- 模块 `integrations/video_extract.py::extract_video`：取/传入 mp4 → base64 → 经 **OpenRouter** `video_url` 发给 `google/gemini-3-flash-preview` → 返回 `segments[]`（start/end MM:SS、title、knowledge_types、what/why/how）。
+- 映射：每个 segment → 一张段卡 `Card(kind="segment", start, end)` 写入 `post.cards`；`ExtractedContent.cards=[{index,content}]`、`text=overall`。下游 screen/split/deconstruct 照常复用。
+- 提取分发（`pipeline`）：`post.content_type=="video"` 或 `post.video_urls` 非空 → `extract_video`；否则图文逐图 `GeminiExtractor`。
+- 请求契约：`POST {OPENROUTER_BASE_URL}/chat/completions`，content=`[{type:text},{type:video_url,video_url:{url:"data:video/mp4;base64,..."}}]`；provider 实际落 Google AI Studio，按 video_tokens 计费（¥0.1 量级/4 分钟）。**无直连 `GEMINI_API_KEY`，只走 `OPENROUTER_API_KEY`。**
+- 下载：抖音详情 `video_url_list[0].video_url`（按 `CK_DOUYIN_RATIO` 默认 540p 改码率），iOS UA + `Referer` + redirects；下载做成可注入（`downloader`）或传入本地 `video_path`，以应对网络分裂。
+- ⚠️ 网络分裂（生产约束）：抖音视频下载需**非新加坡出口**（新加坡云端被 302 挡），调 OpenRouter/Gemini 需**可达 Google**——单机两头难兼得，需"下载节点 + 推理节点"或一个两通的网络。
+- 抽帧 `integrations/video_frames.py` **保留但不接主流程**，仅作可选缩略图/兜底。
+- 已知限制：原生视频默认 1 FPS 采样，快速小字可能漏（口播教程无碍）；段卡默认无缩略图（纯时间段）。`MAX_CARDS` 仅约束图片卡，不限视频段数。