feat(decode): add decode content contracts and readers

core/media_download.py

@@ -0,0 +1,22 @@
+"""Shared media download helpers for acquisition and decode adapters."""
+from __future__ import annotations
+
+import httpx
+
+
+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/",
+    "weixin": "https://mp.weixin.qq.com/",
+}
+IOS_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15"
+
+
+def download_media_bytes(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 response:
+        response.raise_for_status()
+        return b"".join(response.iter_bytes())

decode_content/__init__.py

@@ -0,0 +1,14 @@
+"""Formal decode-content package."""
+
+from decode_content.models import DecodeResult, KnowledgeParticle, PayloadDraft, ReadResult
+from decode_content.service import DecodeService, decode_item, decode_post
+
+__all__ = [
+    "DecodeResult",
+    "DecodeService",
+    "KnowledgeParticle",
+    "PayloadDraft",
+    "ReadResult",
+    "decode_item",
+    "decode_post",
+]

decode_content/contracts.py

@@ -0,0 +1,288 @@
+"""Skill contract loading for the formal decode-content workflow."""
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterable
+
+from core.prompts import PROMPTS_DIR
+from decode_content.models import ContractSnapshot
+
+
+ROOT = Path(__file__).resolve().parent.parent
+DEFAULT_SKILL_DIR = ROOT / "创作知识提取-skill"
+SCOPE_TREE_DIR = ROOT / "scope_trees"
+
+KNOWLEDGE_TYPES = ("how", "what", "why")
+BUSINESS_STAGES = ("灵感", "选题", "脚本")
+CREATION_STAGES = ("定向", "构思", "结构", "成文", "打磨")
+WHAT_KINDS = ("子集", "多维关系", "序列")
+SCOPE_TYPES = ("substance", "form", "feeling", "effect", "intent")
+REUSE_THRESHOLD = 0.90
+SCOPE_TYPE_CN = {
+    "substance": "实质",
+    "form": "形式",
+    "feeling": "感受",
+    "effect": "作用",
+    "intent": "意图",
+}
+DIM_CREATIONS = ("创作",)
+DIM_ATTRIBUTE_BY_TYPE = {"how": "how工序", "what": "what构成", "why": "why原理"}
+CUSTOM_EXT_KEYS_BY_TYPE = {
+    "how": ("业务阶段", "创作阶段", "动作", "出自"),
+    "what": ("业务阶段", "概要", "出自"),
+    "why": ("业务阶段", "出自"),
+}
+PROMPT_NAMES = (
+    "gate_admit",
+    "gate_refute",
+    "gate_tiebreak",
+    "normalize_scope",
+    "gate_how_admit",
+    "gate_how_refute",
+    "gate_how_tiebreak",
+    "gate_why_refute",
+)
+SCOPE_TREE_FILES = ("trees.json", "trees_index.json", "trees_embeddings.npy")
+DRIFT_NOTES = (
+    {
+        "key": "dim_attributes_schema_drift",
+        "formal": DIM_ATTRIBUTE_BY_TYPE,
+        "note": "phase3/taxonomy use how工序/what构成/why原理; legacy schemas still contain how/what for two payload types.",
+    },
+    {
+        "key": "content_shape_drift",
+        "note": "formal ingest boundary uses string content; what/why objects are serialized before payload draft.",
+    },
+    {
+        "key": "how_step_wording_drift",
+        "note": "formal code uses input/directive/output even when older phase text mentions 目的/指引/产出.",
+    },
+)
+
+
+@dataclass(frozen=True)
+class ContractArtifact:
+    name: str
+    contract_type: str
+    source_path: Path
+    content: str
+    content_hash: str
+
+    def snapshot(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "contract_type": self.contract_type,
+            "relative_path": str(self.source_path.relative_to(ROOT)),
+        }
+
+
+@dataclass(frozen=True)
+class SkillContract:
+    """Immutable view of the extraction skill docs used by one decode run."""
+
+    skill_dir: Path
+    artifacts: tuple[ContractArtifact, ...]
+    scope_tree_cache: tuple[dict[str, Any], ...] = ()
+
+    @property
+    def content_hash(self) -> str:
+        h = hashlib.sha256()
+        for artifact in self.artifacts:
+            h.update(artifact.name.encode("utf-8"))
+            h.update(artifact.content_hash.encode("ascii"))
+        for item in self.scope_tree_cache:
+            h.update(str(item.get("path", "")).encode("utf-8"))
+            h.update(str(item.get("content_hash", "")).encode("ascii"))
+        return h.hexdigest()
+
+    def get(self, name: str) -> ContractArtifact:
+        for artifact in self.artifacts:
+            if artifact.name == name:
+                return artifact
+        raise KeyError(name)
+
+    @property
+    def phase1(self) -> str:
+        return self.get("extraction/phase1-frame.md").content
+
+    @property
+    def phase2(self) -> str:
+        return self.get("extraction/phase2-scope.md").content
+
+    @property
+    def phase3(self) -> str:
+        return self.get("extraction/phase3-assemble.md").content
+
+    def schema_for_particle_type(self, particle_type: str) -> dict[str, Any]:
+        artifact = self.get(f"format/ingest-payload-{particle_type}.schema.json")
+        return json.loads(artifact.content)
+
+    def prompt_sources(self) -> list[dict[str, Any]]:
+        return [a.snapshot() | {"content_hash": a.content_hash} for a in self.artifacts if a.contract_type == "prompt"]
+
+    def scope_tree_sources(self) -> list[dict[str, Any]]:
+        return list(self.scope_tree_cache)
+
+    def constants(self) -> dict[str, Any]:
+        return {
+            "knowledge_types": list(KNOWLEDGE_TYPES),
+            "business_stages": list(BUSINESS_STAGES),
+            "creation_stages": list(CREATION_STAGES),
+            "what_kinds": list(WHAT_KINDS),
+            "scope_types": list(SCOPE_TYPES),
+            "scope_type_cn": SCOPE_TYPE_CN,
+            "dim_creations": list(DIM_CREATIONS),
+            "dim_attribute_by_type": DIM_ATTRIBUTE_BY_TYPE,
+            "custom_ext_keys_by_type": {k: list(v) for k, v in CUSTOM_EXT_KEYS_BY_TYPE.items()},
+        }
+
+    def snapshots(self, *, version_label: str | None = None) -> list[ContractSnapshot]:
+        snapshots = [
+            ContractSnapshot(
+                contract_name="创作知识提取-skill",
+                contract_type="skill",
+                version_label=version_label,
+                content_hash=self.content_hash,
+                source_path=str(self.skill_dir.relative_to(ROOT)),
+                snapshot={
+                    "artifact_count": len(self.artifacts),
+                    "artifacts": [a.snapshot() for a in self.artifacts],
+                    "scope_tree_cache": self.scope_tree_sources(),
+                    "constants": self.constants(),
+                    "drift_notes": list(DRIFT_NOTES),
+                },
+            )
+        ]
+        snapshots.extend(
+            ContractSnapshot(
+                contract_name=artifact.name,
+                contract_type=artifact.contract_type,
+                version_label=version_label,
+                content_hash=artifact.content_hash,
+                source_path=str(artifact.source_path.relative_to(ROOT)),
+                snapshot=artifact.snapshot(),
+            )
+            for artifact in self.artifacts
+        )
+        return snapshots
+
+
+def _hash_text(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def _hash_bytes(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _artifact_type(path: Path) -> str:
+    parts = path.parts
+    if "extraction" in parts:
+        return "prompt_phase"
+    if "format" in parts:
+        return "schema"
+    if "taxonomy" in parts:
+        return "taxonomy"
+    if path.name == "README.md":
+        return "readme"
+    if "tools" in parts:
+        return "tool_contract"
+    return "contract"
+
+
+def _scope_tree_cache(scope_tree_dir: Path = SCOPE_TREE_DIR) -> tuple[dict[str, Any], ...]:
+    out: list[dict[str, Any]] = []
+    for name in SCOPE_TREE_FILES:
+        path = scope_tree_dir / name
+        if not path.exists():
+            continue
+        stat = path.stat()
+        out.append(
+            {
+                "path": str(path.relative_to(ROOT)),
+                "size": stat.st_size,
+                "mtime": int(stat.st_mtime),
+                "content_hash": _hash_bytes(path.read_bytes()),
+            }
+        )
+    return tuple(out)
+
+
+def _iter_contract_files(skill_dir: Path) -> Iterable[Path]:
+    preferred = [
+        skill_dir / "README.md",
+        skill_dir / "extraction" / "phase1-frame.md",
+        skill_dir / "extraction" / "phase2-scope.md",
+        skill_dir / "extraction" / "phase3-assemble.md",
+        skill_dir / "format" / "ingest-payload-how.schema.json",
+        skill_dir / "format" / "ingest-payload-what.schema.json",
+        skill_dir / "format" / "ingest-payload-why.schema.json",
+        skill_dir / "taxonomy" / "业务阶段.json",
+        skill_dir / "taxonomy" / "作用域.md",
+        skill_dir / "taxonomy" / "创作阶段.json",
+        skill_dir / "taxonomy" / "知识类型.json",
+    ]
+    seen: set[Path] = set()
+    for path in preferred:
+        if path.exists():
+            seen.add(path)
+            yield path
+    for path in sorted(skill_dir.rglob("*")):
+        if path.is_file() and path.suffix in {".md", ".json"} and path not in seen:
+            yield path
+
+
+def _iter_prompt_files() -> Iterable[Path]:
+    for name in PROMPT_NAMES:
+        path = PROMPTS_DIR / f"{name}.txt"
+        if path.exists():
+            yield path
+
+
+def load_contract(skill_dir: Path | str = DEFAULT_SKILL_DIR) -> SkillContract:
+    root = Path(skill_dir)
+    if not root.exists():
+        raise FileNotFoundError(f"skill contract directory not found: {root}")
+    artifacts: list[ContractArtifact] = []
+    for path in _iter_contract_files(root):
+        text = path.read_text(encoding="utf-8")
+        artifacts.append(
+            ContractArtifact(
+                name=str(path.relative_to(root)),
+                contract_type=_artifact_type(path.relative_to(root)),
+                source_path=path,
+                content=text,
+                content_hash=_hash_text(text),
+            )
+        )
+    for path in _iter_prompt_files():
+        text = path.read_text(encoding="utf-8")
+        artifacts.append(
+            ContractArtifact(
+                name=f"prompts/{path.name}",
+                contract_type="prompt",
+                source_path=path,
+                content=text,
+                content_hash=_hash_text(text),
+            )
+        )
+    return SkillContract(skill_dir=root, artifacts=tuple(artifacts), scope_tree_cache=_scope_tree_cache())
+
+
+def load_decode_contracts(skill_dir: Path | str = DEFAULT_SKILL_DIR) -> SkillContract:
+    return load_contract(skill_dir)
+
+
+def compute_contract_hash(contract: SkillContract | None = None) -> str:
+    return (contract or load_contract()).content_hash
+
+
+def iter_contract_snapshots(
+    contract: SkillContract | None = None,
+    *,
+    version_label: str | None = None,
+) -> list[ContractSnapshot]:
+    return (contract or load_contract()).snapshots(version_label=version_label)

decode_content/readers/__init__.py

@@ -0,0 +1,5 @@
+"""Formal content readers package."""
+
+from decode_content.readers.service import post_from_candidate_item, read_item, read_post
+
+__all__ = ["post_from_candidate_item", "read_item", "read_post"]

decode_content/readers/imgtext.py

@@ -0,0 +1,172 @@
+"""Image-text multimodal reader for creation knowledge decode."""
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Mapping, Optional
+
+import httpx
+
+from core.config import load_env_file
+from core.jsonio import extract_json_object, to_bool
+from core.models import Card, CardExtract, ExtractedContent, Post
+from core.prompts import load_prompt
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MODEL = "qwen-vl-plus"
+DEFAULT_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
+DEFAULT_TIMEOUT = 120.0
+MAX_CARDS = 12
+
+
+def _card_label(card: Card) -> str:
+    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 = (
+    "你是创作知识提取助手。从给定的小红书帖子（标题、正文、图片、视频）中，"
+    "提取真正能指导『如何创作内容』的知识；知识常在图片/视频里而非正文。"
+    "忠实提取、不编造。只输出一个 JSON 对象，不要解释或 markdown。"
+)
+
+
+class ExtractorError(RuntimeError):
+    pass
+
+
+class BailianExtractor:
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        model: str = DEFAULT_MODEL,
+        base_url: str = DEFAULT_BASE_URL,
+        timeout_seconds: float = DEFAULT_TIMEOUT,
+        http_post: Callable[..., Any] = httpx.post,
+        max_cards: int = MAX_CARDS,
+    ) -> None:
+        if not api_key:
+            raise ExtractorError("missing ALIYUN_BAILIAN_API_KEY")
+        self.api_key = api_key
+        self.model = model
+        self.base_url = base_url.rstrip("/")
+        self.timeout_seconds = timeout_seconds
+        self.http_post = http_post
+        self.max_cards = max_cards
+
+    @classmethod
+    def from_env(cls, env: Mapping[str, str] | None = None, env_file: str = ".env") -> "BailianExtractor":
+        source = dict(load_env_file(env_file))
+        if env:
+            source.update(env)
+        api_key = source.get("ALIYUN_BAILIAN_API_KEY") or ""
+        return cls(
+            api_key=api_key,
+            model=source.get("ALIYUN_BAILIAN_VL_MODEL") or source.get("ALIYUN_BAILIAN_MODEL") or DEFAULT_MODEL,
+            base_url=source.get("ALIYUN_BAILIAN_BASE_URL") or DEFAULT_BASE_URL,
+            timeout_seconds=float(source.get("ALIYUN_BAILIAN_TIMEOUT_SECONDS") or DEFAULT_TIMEOUT),
+            max_cards=int(source.get("CK_MAX_CARDS") or MAX_CARDS),
+        )
+
+    def _cards(self, post: Post) -> list[Card]:
+        cards = post.cards or [
+            Card(index=i, kind="image", url=url)
+            for i, url in enumerate(post.image_urls, start=1)
+        ]
+        if len(cards) > self.max_cards:
+            dropped = [card.index for card in cards[self.max_cards :]]
+            logger.warning(
+                "post %s card count %d exceeds MAX_CARDS=%d, dropping cards %s",
+                post.id,
+                len(cards),
+                self.max_cards,
+                dropped,
+            )
+            cards = cards[: self.max_cards]
+        return cards
+
+    def build_messages(self, post: Post) -> list[dict[str, Any]]:
+        user_text = load_prompt("extract").format(
+            title=post.title or "(无)",
+            topics="、".join(post.topic_list) or "(无)",
+            body=post.body_text or "(空)",
+        )
+        parts: list[dict[str, Any]] = [{"type": "text", "text": user_text}]
+        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},
+        ]
+
+    def extract(self, post: Post) -> ExtractedContent:
+        messages = self.build_messages(post)
+        last_exc: Optional[Exception] = None
+        for attempt in range(2):
+            try:
+                resp = self.http_post(
+                    f"{self.base_url}/chat/completions",
+                    headers={
+                        "Authorization": f"Bearer {self.api_key}",
+                        "Content-Type": "application/json",
+                    },
+                    json={"model": self.model, "messages": messages},
+                    timeout=self.timeout_seconds,
+                )
+                resp.raise_for_status()
+                content = resp.json()["choices"][0]["message"]["content"]
+                data = extract_json_object(content)
+                cards = [
+                    CardExtract(index=int(card["index"]), content=str(card.get("content") or ""))
+                    for card in (data.get("cards") or [])
+                    if isinstance(card, dict) and card.get("index") is not None
+                ]
+                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")),
+                )
+            except httpx.HTTPError as exc:
+                last_exc = exc
+                if attempt == 0:
+                    continue
+                raise ExtractorError(f"bailian_http_error: {exc}") from exc
+            except (KeyError, IndexError, TypeError, ValueError) as exc:
+                last_exc = exc
+                if attempt == 0:
+                    continue
+                raise ExtractorError(f"bailian_response_invalid: {exc}") from exc
+        raise ExtractorError(f"bailian_unknown_error: {last_exc}")
+
+
+GeminiExtractor = BailianExtractor
+
+
+def extract_content(
+    post: Post,
+    *,
+    client: Optional[BailianExtractor] = None,
+    env_file: str = ".env",
+) -> ExtractedContent:
+    client = client or GeminiExtractor.from_env(env_file=env_file)
+    return client.extract(post)
+
+
+def read_imgtext(post: Post, *, extractor: BailianExtractor | None = None) -> ExtractedContent:
+    client = extractor or BailianExtractor.from_env()
+    return client.extract(post)
+
+
+__all__ = [
+    "BailianExtractor",
+    "ExtractorError",
+    "GeminiExtractor",
+    "extract_content",
+    "read_imgtext",
+]

decode_content/readers/service.py

@@ -0,0 +1,173 @@
+"""Unified post reader for image-text and video decode inputs."""
+from __future__ import annotations
+
+from typing import Protocol
+
+from acquisition.domain import CandidateItem, MediaAsset
+from core.config import Settings
+from core.models import Card, ExtractedContent, Post
+from decode_content.models import ReadCard, ReadResult
+from decode_content.readers.imgtext import BailianExtractor, read_imgtext
+from decode_content.readers.video import read_video
+
+
+class ImageReader(Protocol):
+    def __call__(self, post: Post) -> ExtractedContent:
+        ...
+
+
+class VideoReader(Protocol):
+    def __call__(self, post: Post) -> ExtractedContent:
+        ...
+
+
+def _media_url(asset: MediaAsset) -> str | None:
+    return asset.cdn_url or asset.oss_url or asset.source_url
+
+
+def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]) -> Post:
+    """Build the reader input from formal acquisition DB objects."""
+
+    ordered = sorted(media_assets, key=lambda asset: asset.position)
+    image_urls: list[str] = []
+    video_urls: list[str] = []
+    cards: list[Card] = []
+    for asset in ordered:
+        url = _media_url(asset)
+        if not url:
+            continue
+        media_type = asset.media_type.lower()
+        if media_type == "video":
+            video_urls.append(url)
+            continue
+        if media_type in {"image", "cover", "frame"}:
+            image_urls.append(url)
+            cards.append(
+                Card(
+                    index=len(cards) + 1,
+                    kind="frame" if media_type == "frame" else "image",
+                    url=url,
+                    timestamp=asset.metadata.get("timestamp"),
+                )
+            )
+    source_payload = item.source_payload or {}
+    content_id = item.platform_item_id or (str(item.id) if item.id else "")
+    return Post(
+        id=f"{item.platform}_{content_id}" if content_id and not content_id.startswith(f"{item.platform}_") else content_id,
+        platform=item.platform,
+        url=item.canonical_url or "",
+        content_id=content_id,
+        title=item.title or "",
+        content_type=item.content_type or ("video" if video_urls else "图文"),
+        body_text=(
+            item.raw_summary
+            or source_payload.get("body_text")
+            or source_payload.get("desc")
+            or source_payload.get("content")
+            or ""
+        ),
+        topic_list=source_payload.get("topic_list") or source_payload.get("topics") or [],
+        image_urls=image_urls,
+        video_urls=video_urls,
+        cards=[] if video_urls else cards,
+        author_name=item.author_name,
+        raw=source_payload,
+    )
+
+
+def read_result_from_extracted(post: Post, extracted: ExtractedContent) -> ReadResult:
+    cmap = {card.index: card.content for card in extracted.cards}
+    cards: list[ReadCard] = []
+    source_cards = post.cards
+    if not source_cards and post.image_urls:
+        source_cards = [
+            {"index": idx, "kind": "image", "url": url}
+            for idx, url in enumerate(post.image_urls, start=1)
+        ]
+    for raw in source_cards:
+        if isinstance(raw, dict):
+            index = int(raw.get("index") or 0)
+            kind = str(raw.get("kind") or "image")
+            url = raw.get("url")
+            timestamp = raw.get("timestamp")
+            start = raw.get("start")
+            end = raw.get("end")
+        else:
+            index = raw.index
+            kind = raw.kind
+            url = raw.url
+            timestamp = raw.timestamp
+            start = raw.start
+            end = raw.end
+        cards.append(
+            ReadCard(
+                index=index,
+                kind=kind,
+                url=url,
+                timestamp=timestamp,
+                start=start,
+                end=end,
+                content=cmap.get(index, ""),
+            )
+        )
+    parts = [extracted.text]
+    if extracted.from_image:
+        parts.append("【图片要点】\n" + extracted.from_image)
+    if extracted.from_video:
+        parts.append("【视频要点】\n" + extracted.from_video)
+    parts.extend(f"【卡片{card.index}】{card.content}" for card in extracted.cards if card.content)
+    text = "\n\n".join(p for p in parts if p)
+    media = (
+        {"type": "video", "video_url": post.video_urls[0] if post.video_urls else None, "images": []}
+        if post.video_urls
+        else {"type": "image", "video_url": None, "images": list(post.image_urls)}
+    )
+    return ReadResult(text=text, cards=cards, media=media, is_empty=bool(extracted.is_empty))
+
+
+def read_post(
+    post: Post,
+    *,
+    settings: Settings | None = None,
+    extractor: BailianExtractor | None = None,
+    image_reader: ImageReader | None = None,
+    video_reader: VideoReader | None = None,
+) -> ReadResult:
+    if post.video_urls:
+        if video_reader is not None:
+            extracted = video_reader(post)
+        else:
+            if settings is None:
+                raise ValueError("settings is required when reading video posts")
+            extracted = read_video(post, settings=settings)
+    else:
+        extracted = image_reader(post) if image_reader is not None else read_imgtext(post, extractor=extractor)
+    return read_result_from_extracted(post, extracted)
+
+
+def read_item(
+    item: CandidateItem,
+    media_assets: list[MediaAsset],
+    *,
+    settings: Settings | None = None,
+    extractor: BailianExtractor | None = None,
+    image_reader: ImageReader | None = None,
+    video_reader: VideoReader | None = None,
+) -> ReadResult:
+    return read_post(
+        post_from_candidate_item(item, media_assets),
+        settings=settings,
+        extractor=extractor,
+        image_reader=image_reader,
+        video_reader=video_reader,
+    )
+
+
+__all__ = [
+    "ImageReader",
+    "VideoReader",
+    "post_from_candidate_item",
+    "read_item",
+    "read_post",
+    "read_result_from_extracted",
+]

decode_content/readers/video.py

@@ -0,0 +1,177 @@
+"""Native whole-video reader for creation knowledge decode."""
+from __future__ import annotations
+
+import base64
+import logging
+import os
+import re
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+import httpx
+
+from core.config import Settings
+from core.jsonio import extract_json_object
+from core.media_download import download_media_bytes
+from core.models import Card, CardExtract, ExtractedContent, Post
+from core.prompts import load_prompt
+
+logger = logging.getLogger(__name__)
+
+class VideoExtractError(RuntimeError):
+    pass
+
+
+def _mmss_to_sec(value: Any) -> Optional[float]:
+    if value is None:
+        return None
+    if isinstance(value, (int, float)):
+        return float(value)
+    parts = str(value).strip().split(":")
+    try:
+        nums = [float(part) for part in parts]
+    except ValueError:
+        return None
+    sec = 0.0
+    for number in nums:
+        sec = sec * 60 + number
+    return sec
+
+
+def _default_download(url: str, platform: str, timeout: float = 180.0) -> bytes:
+    return download_media_bytes(url, platform, timeout=timeout)
+
+
+def _seg_content(segment: dict[str, Any]) -> str:
+    parts = [segment.get("title") or ""]
+    for label, key in (("What", "what"), ("Why", "why"), ("How", "how")):
+        value = segment.get(key)
+        if value and str(value).strip().lower() not in {"null", "none", ""}:
+            parts.append(f"{label}：{value}")
+    return "。".join(part for part in parts if part)
+
+
+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,
+    save_path: Optional[Path] = None,
+    public_url: Optional[str] = None,
+    oss_video_url: Optional[str] = None,
+) -> ExtractedContent:
+    key = settings.openrouter_api_key
+    if not key:
+        raise VideoExtractError("missing OPENROUTER_API_KEY")
+
+    if oss_video_url:
+        media_url = oss_video_url
+        public_url = public_url or oss_video_url
+    else:
+        if post.platform == "bilibili" and not (video_path and os.path.exists(video_path)):
+            raise VideoExtractError("B站视频暂未接入：需取 voice_data 音轨 + ffmpeg 合并")
+
+        if video_path and os.path.exists(video_path):
+            data = open(video_path, "rb").read()
+            logger.info("video_extract using local file %s (%d bytes)", video_path, len(data))
+        else:
+            if not post.video_urls:
+                raise VideoExtractError(f"post {post.id} has no video_urls and no 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("视频字节为空")
+
+        if save_path is not None:
+            save_path.parent.mkdir(parents=True, exist_ok=True)
+            Path(save_path).write_bytes(data)
+            logger.info("video_extract saved whole video %s (%d bytes)", save_path, len(data))
+
+        media_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": media_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 []
+    cards: list[Card] = []
+    card_extracts: list[CardExtract] = []
+    for idx, segment in enumerate(segments, start=1):
+        if not isinstance(segment, dict):
+            continue
+        cards.append(
+            Card(
+                index=idx,
+                kind="segment",
+                url=public_url,
+                start=_mmss_to_sec(segment.get("start")),
+                end=_mmss_to_sec(segment.get("end")),
+            )
+        )
+        card_extracts.append(CardExtract(index=idx, content=_seg_content(segment)))
+    post.cards = cards
+    return ExtractedContent(
+        text=str(obj.get("overall") or obj.get("video_title") or ""),
+        cards=card_extracts,
+        is_empty=len(card_extracts) == 0,
+    )
+
+
+def read_video(
+    post: Post,
+    *,
+    settings: Settings,
+    http_post: Callable[..., Any] | None = None,
+    video_path: str | None = None,
+    downloader: Callable[[str, str], bytes] | None = None,
+    save_path: Path | None = None,
+    public_url: str | None = None,
+    oss_video_url: str | None = None,
+) -> ExtractedContent:
+    kwargs: dict[str, Any] = {
+        "settings": settings,
+        "video_path": video_path,
+        "downloader": downloader,
+        "save_path": save_path,
+        "public_url": public_url or oss_video_url,
+        "oss_video_url": oss_video_url,
+    }
+    if http_post is not None:
+        kwargs["http_post"] = http_post
+    return extract_video(post, **kwargs)
+
+
+__all__ = ["VideoExtractError", "extract_video", "read_video"]

decode_content/readers/video_frames.py

@@ -0,0 +1,172 @@
+"""Video frame extraction fallback for multimodal readers."""
+from __future__ import annotations
+
+import logging
+import re
+import subprocess
+from pathlib import Path
+from typing import Optional
+
+import httpx
+
+from core.models import Card
+
+logger = logging.getLogger(__name__)
+
+SCENE_THRESHOLD = 0.3
+MIN_GAP_SECONDS = 1.5
+FALLBACK_MIN = 4
+FALLBACK_TARGET = 8
+SCALE_VF = "scale='min(720,iw)':-2"
+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 response:
+        response.raise_for_status()
+        with open(dst, "wb") as file:
+            for chunk in response.iter_bytes():
+                file.write(chunk)
+
+
+def _run(cmd: list[str]) -> str:
+    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)])
+    match = re.search(r"Duration:\s*(\d+):(\d+):(\d+\.?\d*)", err)
+    if not match:
+        return 0.0
+    hours, minutes, seconds = match.groups()
+    return int(hours) * 3600 + int(minutes) * 60 + float(seconds)
+
+
+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(ts) for ts 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, timestamp in paired:
+        if timestamp is not None and last is not None and timestamp - last < min_gap:
+            continue
+        kept.append((path, timestamp))
+        if timestamp is not None:
+            last = timestamp
+        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)
+    result: list[tuple[Path, Optional[float]]] = []
+    for idx in range(count):
+        timestamp = duration * (idx + 0.5) / count
+        path = out / f"uni_{idx:04d}.jpg"
+        _run(
+            [
+                ffmpeg,
+                "-hide_banner",
+                "-ss",
+                f"{timestamp:.2f}",
+                "-i",
+                str(video),
+                "-frames:v",
+                "1",
+                "-vf",
+                SCALE_VF,
+                "-q:v",
+                "3",
+                str(path),
+            ]
+        )
+        if path.exists():
+            result.append((path, round(timestamp, 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]:
+    out = Path(out_dir)
+    out.mkdir(parents=True, exist_ok=True)
+    ffmpeg = _ffmpeg_bin()
+
+    src = Path(video_src)
+    if not src.exists():
+        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("scene frames only %d, falling back to uniform sampling", len(frames))
+        duration = _probe_duration(ffmpeg, src)
+        frames = _uniform_frames(ffmpeg, src, out, duration, FALLBACK_TARGET)
+    frames = frames[:max_frames]
+    if not frames:
+        raise VideoFramesError(f"未能从视频抽出任何帧: {video_src}")
+
+    return [
+        Card(index=start_index + idx, kind="frame", url=f"{url_prefix.rstrip('/')}/{path.name}", timestamp=timestamp)
+        for idx, (path, timestamp) in enumerate(frames)
+    ]

scripts/smoke_extract.py

@@ -14,7 +14,7 @@ import sys
 from pathlib import Path
 
 from acquisition.crawler import parse_detail_response
-from creation_knowledge.integrations.extractor import GeminiExtractor
+from decode_content.readers.imgtext import GeminiExtractor
 
 FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"
 DEFAULT_IDS = ["67e2e39b0000000003028ff0", "680659e8000000001a007a11"]

创作知识提取-skill/taxonomy/作用域.md

@@ -23,6 +23,6 @@ python 创作知识提取-skill/tools/scope-link.py "撕裂共识" --type effect
 - 0.75–0.90 → 看 top-K 里有没有真同义的，有就复用，没有就用自己的候选值（新值）。
 - < 0.75 → 保留候选值为新值。
 
-底层：`scripts/scope_link.py`（ScopeLinker）+ `creation_knowledge/embedding.py`（火山 Doubao-embedding-vision，2048 维）+ `scope_trees/`（1855 个 live 节点的本地向量缓存）。树更新后重跑 `scripts/dump_trees.py` + `scripts/embed_trees.py` 刷新缓存。
+底层：`scripts/scope_link.py`（ScopeLinker）+ `core/embedding.py`（火山 Doubao-embedding-vision，2048 维）+ `scope_trees/`（1855 个 live 节点的本地向量缓存）。树更新后重跑 `scripts/dump_trees.py` + `scripts/embed_trees.py` 刷新缓存。
 
 > `--type` 用英文 key（substance/form/feeling/effect/intent）或中文（实质/形式/感受/作用/意图）都行；底层按 source_type 中文名过滤。