feat(data): add formal item identity and text contracts

.env.example

@@ -11,6 +11,16 @@ CK_DB_NAME=creation_knowledge_prod
 CK_DB_USER=ck_app
 CK_DB_PASSWORD=
 CK_DB_SCHEMA=creation_knowledge
+CK_DB_SSH_TUNNEL=
+
+# Optional migration/admin path. Runtime writes should use CK_DB_USER above;
+# schema migrations should use an owner/admin channel such as root SSH -> postgres.
+CK_DB_ADMIN_SSH_USER=root
+CK_DB_ADMIN_SSH_HOST=
+CK_DB_ADMIN_SSH_KEY=
+CK_DB_ADMIN_PG_SYSTEM_USER=postgres
+CK_DB_ADMIN_PSQL="runuser -u postgres -- psql -d creation_knowledge_prod -v ON_ERROR_STOP=1"
+CK_DB_MIGRATION_APPLY="ssh -i <key> -o BatchMode=yes -o ConnectTimeout=8 -o StrictHostKeyChecking=accept-new root@<host> \"runuser -u postgres -- psql -d creation_knowledge_prod -v ON_ERROR_STOP=1\" < path/to/migration.sql"
 
 # -----------------------------------------------------------------------------
 # Upstream Open AIGC / category-tree database

acquisition/classification/coarse.py

@@ -15,6 +15,7 @@ from acquisition.classify import (
 )
 from core.config import Settings
 from core.prompts import load_prompt
+from core.text_limits import CLASSIFY_BODY_MAX_CHARS, clip_text
 
 ROOT = Path(__file__).resolve().parents[2]
 
@@ -50,7 +51,7 @@ def classify_imgtext(payload: dict[str, Any], settings: Settings) -> tuple:
             "text": (
                 f"平台：{payload.get('platform')}\n"
                 f"标题：{payload.get('title', '')}\n"
-                f"正文：{(payload.get('body_text') or '')[:1500]}\n"
+                f"正文：{clip_text(payload.get('body_text') or '', CLASSIFY_BODY_MAX_CHARS)}\n"
                 "（下附帖子图片，请一并看完）"
             ),
         }
@@ -76,13 +77,36 @@ def classify_video(payload: dict[str, Any], settings: Settings) -> tuple:
 def coarse_classify_item(
     *,
     platform: str,
+    content_mode: str | None = None,
     title: str = "",
     body_text: str = "",
     image_urls: list[str] | None = None,
     video_url: str = "",
     settings: Settings,
 ) -> ClassificationResult:
-    if platform == "douyin" or video_url:
+    if content_mode == "unsupported":
+        return ClassificationResult(
+            is_creation_knowledge=None,
+            label="unsupported_content_mode",
+            confidence=None,
+            reason="内容模态暂不支持，跳过粗筛",
+            prompt_version="content_mode_guard",
+            result_payload={"content_mode": content_mode},
+            status="skipped",
+            error_message="unsupported_content_mode",
+        )
+    if content_mode == "video_post" and not video_url:
+        return ClassificationResult(
+            is_creation_knowledge=None,
+            label="video_missing",
+            confidence=None,
+            reason="视频帖缺少可处理的视频地址，跳过粗筛",
+            prompt_version="content_mode_guard",
+            result_payload={"content_mode": content_mode, "video_url_missing": True},
+            status="skipped",
+            error_message="video_url_missing",
+        )
+    if content_mode == "video_post" or (content_mode is None and video_url):
         version = prompt_version("classify_video")
         is_creation, reason, knowledge, points = classify_video(
             {

acquisition/classify.py

@@ -25,6 +25,12 @@ import httpx
 
 from core.config import Settings, load_env_file
 from core.prompts import load_prompt
+from core.text_limits import (
+    CLASSIFY_BODY_MAX_CHARS,
+    ERROR_MESSAGE_MAX_CHARS,
+    REASON_MAX_CHARS,
+    clip_text,
+)
 
 ROOT = Path(__file__).resolve().parent.parent
 PLATFORMS = ["xiaohongshu", "weixin", "douyin"]   # 默认重判全部；可传平台名覆盖
@@ -108,8 +114,8 @@ def _compress(mp4: Path) -> Path:
 def _parse_judge_content(content: str) -> tuple:
     d = json.loads(content)
     if bool(d.get("is_empty")):
-        return 0, str(d.get("reason", ""))[:60], "", ""
-    return 1, str(d.get("reason", ""))[:60], str(d.get("knowledge", "") or ""), ""
+        return 0, clip_text(d.get("reason", ""), REASON_MAX_CHARS), "", ""
+    return 1, clip_text(d.get("reason", ""), REASON_MAX_CHARS), str(d.get("knowledge", "") or ""), ""
 
 
 def _env_first(env: dict, *keys: str) -> str:
@@ -284,7 +290,7 @@ def _judge(messages: list, settings: Settings, timeout: float) -> tuple:
                 if resp.status_code in (401, 403, 404):
                     break
             except Exception as exc:
-                last = f"{name} {str(exc)[:50]}"
+                last = f"{name} {clip_text(exc, ERROR_MESSAGE_MAX_CHARS)}"
             if "http " not in last or any(f"http {code}" in last for code in RETRYABLE_STATUS_CODES):
                 time.sleep(2 * (attempt + 1))
             else:
@@ -297,7 +303,7 @@ def _judge(messages: list, settings: Settings, timeout: float) -> tuple:
 def classify_imgtext(p: dict, settings: Settings) -> tuple:
     """图文：标题+正文+全部图，用收紧的 classify_imgtext.txt 判 is_empty 并提取知识点。"""
     user = [{"type": "text", "text": f"平台：{p.get('platform')}\n标题：{p.get('title', '')}\n"
-             f"正文：{(p.get('body_text') or '')[:1500]}\n（下附帖子图片，请一并看完）"}]
+             f"正文：{clip_text(p.get('body_text') or '', CLASSIFY_BODY_MAX_CHARS)}\n（下附帖子图片，请一并看完）"}]
     for im in (p.get("images") or [])[:MAX_CARDS]:
         if _is_http_url(im):
             user.append({"type": "image_url", "image_url": {"url": im}})
@@ -331,8 +337,14 @@ def classify_video(p: dict, settings: Settings) -> tuple:
                     use.unlink()
                 except Exception:
                     pass
+    user_text = (
+        "判断这条视频是不是创作知识。\n"
+        f"平台：{p.get('platform', '')}\n"
+        f"标题：{p.get('title', '')}\n"
+        f"正文/文案：{clip_text(p.get('body_text') or '', CLASSIFY_BODY_MAX_CHARS)}"
+    )
     messages = [{"role": "system", "content": load_prompt("classify_video")},
-                {"role": "user", "content": [{"type": "text", "text": "判断这条视频是不是创作知识。"},
+                {"role": "user", "content": [{"type": "text", "text": user_text},
                                              {"type": "video_url", "video_url": {"url": media}}]}]
     return _judge(messages, settings, timeout=300)
 

acquisition/content_mode.py

@@ -0,0 +1,108 @@
+"""Business-level content mode helpers for acquisition."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Literal
+
+
+ContentMode = Literal["image_post", "video_post", "article", "unsupported"]
+
+IMAGE_POST: ContentMode = "image_post"
+VIDEO_POST: ContentMode = "video_post"
+ARTICLE: ContentMode = "article"
+UNSUPPORTED: ContentMode = "unsupported"
+
+
+@dataclass(frozen=True)
+class ContentModeGuard:
+    can_process: bool
+    reason: str = ""
+    label: str = ""
+    metadata: dict[str, Any] | None = None
+
+
+def _has_any(values: list[str] | None) -> bool:
+    return any(bool(value) for value in values or [])
+
+
+def _raw_type(raw: dict[str, Any] | None) -> str:
+    if not isinstance(raw, dict):
+        return ""
+    candidates = [
+        raw.get("type"),
+        raw.get("content_type"),
+        raw.get("note_type"),
+        (raw.get("note_card") or {}).get("type") if isinstance(raw.get("note_card"), dict) else None,
+        raw.get("aweme_type"),
+    ]
+    for value in candidates:
+        if value is not None and str(value).strip():
+            return str(value).strip().lower()
+    return ""
+
+
+def infer_content_mode(
+    *,
+    platform: str,
+    content_type: str | None = None,
+    body_text: str | None = None,
+    image_urls: list[str] | None = None,
+    video_urls: list[str] | None = None,
+    raw: dict[str, Any] | None = None,
+) -> ContentMode:
+    """Infer the business mode without overwriting the platform raw type."""
+
+    platform_key = (platform or "").lower()
+    raw_kind = _raw_type(raw)
+    raw_content_type = (content_type or "").strip().lower()
+    has_image = _has_any(image_urls)
+    has_video = _has_any(video_urls)
+    has_text = bool((body_text or "").strip())
+
+    if platform_key == "weixin":
+        return ARTICLE
+    if has_video:
+        return VIDEO_POST
+    if has_image or has_text:
+        return IMAGE_POST
+    if raw_kind in {"normal", "note", "image", "image_post", "图文"}:
+        return IMAGE_POST
+    if raw_content_type in {"normal", "note", "image", "image_post", "图文"}:
+        return IMAGE_POST
+    return UNSUPPORTED
+
+
+def guard_content_for_processing(
+    *,
+    content_mode: str | None,
+    video_urls: list[str] | None = None,
+) -> ContentModeGuard:
+    """Return whether this item should enter media/classify/decode processing."""
+
+    if content_mode == UNSUPPORTED:
+        return ContentModeGuard(
+            can_process=False,
+            reason="unsupported_content_mode",
+            label="unsupported_content_mode",
+            metadata={"skip_reason": "unsupported_content_mode"},
+        )
+    if content_mode == VIDEO_POST and not _has_any(video_urls):
+        return ContentModeGuard(
+            can_process=False,
+            reason="video_url_missing",
+            label="video_missing",
+            metadata={"skip_reason": "video_url_missing", "video_url_missing": True},
+        )
+    return ContentModeGuard(can_process=True)
+
+
+__all__ = [
+    "ARTICLE",
+    "ContentMode",
+    "ContentModeGuard",
+    "IMAGE_POST",
+    "UNSUPPORTED",
+    "VIDEO_POST",
+    "guard_content_for_processing",
+    "infer_content_mode",
+]

acquisition/domain.py

@@ -14,6 +14,7 @@ from pydantic import BaseModel, ConfigDict, Field
 
 JsonDict = dict[str, Any]
 Platform = Literal["xiaohongshu", "weixin", "douyin"]
+ContentMode = Literal["image_post", "video_post", "article", "unsupported"]
 AcquisitionStatus = Literal[
     "draft",
     "pending",
@@ -95,11 +96,14 @@ class CandidateItem(DomainModel):
     query_id: UUID | None = None
     platform: str
     platform_item_id: str | None = None
+    unique_key: str | None = None
     canonical_url: str | None = None
     content_type: str | None = None
+    content_mode: ContentMode | None = None
     title: str | None = None
     author_name: str | None = None
     published_at: datetime | None = None
+    body_text: str | None = None
     raw_summary: str | None = None
     status: str = "candidate"
     source_payload: JsonDict = Field(default_factory=dict)

acquisition/repositories/base.py

@@ -107,10 +107,13 @@ class AcquisitionRepository(Protocol):
         job_id: UUID | None = None,
         query_id: UUID | None = None,
         platform_item_id: str | None = None,
+        unique_key: str | None = None,
         canonical_url: str | None = None,
         content_type: str | None = None,
+        content_mode: str | None = None,
         title: str | None = None,
         author_name: str | None = None,
+        body_text: str | None = None,
         raw_summary: str | None = None,
         status: str = "candidate",
         source_payload: dict[str, Any] | None = None,
@@ -119,6 +122,19 @@ class AcquisitionRepository(Protocol):
     ) -> CandidateItem:
         ...
 
+    def get_candidate_item_by_unique_key(self, unique_key: str) -> CandidateItem | None:
+        ...
+
+    def attach_existing_candidate_item(
+        self,
+        item_id: UUID,
+        *,
+        job_id: UUID,
+        query_id: UUID,
+        metadata: dict[str, Any] | None = None,
+    ) -> CandidateItem:
+        ...
+
     def add_media_asset(
         self,
         *,
@@ -156,6 +172,9 @@ class AcquisitionRepository(Protocol):
     def get_query_detail(self, *, run_id: UUID, query_id: UUID) -> dict[str, Any]:
         ...
 
+    def get_query_detail_for_batch(self, *, batch_id: UUID, query_id: UUID) -> dict[str, Any]:
+        ...
+
     def list_creation_candidate_items(
         self,
         *,

acquisition/repositories/postgres.py

@@ -264,10 +264,13 @@ class PostgresAcquisitionRepository:
         job_id: UUID | None = None,
         query_id: UUID | None = None,
         platform_item_id: str | None = None,
+        unique_key: str | None = None,
         canonical_url: str | None = None,
         content_type: str | None = None,
+        content_mode: str | None = None,
         title: str | None = None,
         author_name: str | None = None,
+        body_text: str | None = None,
         raw_summary: str | None = None,
         status: str = "candidate",
         source_payload: dict[str, Any] | None = None,
@@ -275,31 +278,46 @@ class PostgresAcquisitionRepository:
         error_message: str | None = None,
     ) -> CandidateItem:
         existing_id = None
-        if platform_item_id:
+        if unique_key:
             row = self._one_or_none(
                 """
                 SELECT id FROM candidate_items
-                WHERE platform = %s AND platform_item_id = %s
+                WHERE unique_key = %s
                 ORDER BY created_at DESC
                 LIMIT 1
                 """,
-                (platform, platform_item_id),
+                (unique_key,),
             )
             existing_id = row["id"] if row else None
 
+        if platform_item_id:
+            if existing_id is None:
+                row = self._one_or_none(
+                    """
+                    SELECT id FROM candidate_items
+                    WHERE platform = %s AND platform_item_id = %s
+                    ORDER BY created_at DESC
+                    LIMIT 1
+                    """,
+                    (platform, platform_item_id),
+                )
+                existing_id = row["id"] if row else None
+
         if existing_id:
             row = self._one(
                 """
                 UPDATE candidate_items SET
                     job_id = %s,
                     query_id = %s,
+                    unique_key = COALESCE(%s, unique_key),
                     canonical_url = %s,
                     content_type = %s,
+                    content_mode = %s,
                     title = %s,
                     author_name = %s,
+                    body_text = %s,
                     raw_summary = %s,
                     status = %s,
-                    source_payload = %s,
                     metadata = %s,
                     error_message = %s
                 WHERE id = %s
@@ -308,13 +326,15 @@ class PostgresAcquisitionRepository:
                 (
                     job_id,
                     query_id,
+                    unique_key,
                     canonical_url,
                     content_type,
+                    content_mode,
                     title,
                     author_name,
+                    body_text,
                     raw_summary,
                     status,
-                    Json(source_payload or {}),
                     Json(metadata or {}),
                     error_message,
                     existing_id,
@@ -324,11 +344,11 @@ class PostgresAcquisitionRepository:
             row = self._one(
                 """
                 INSERT INTO candidate_items(
-                    job_id, query_id, platform, platform_item_id, canonical_url,
-                    content_type, title, author_name, raw_summary, status,
+                    job_id, query_id, platform, platform_item_id, unique_key, canonical_url,
+                    content_type, content_mode, title, author_name, body_text, raw_summary, status,
                     source_payload, metadata, error_message
                 )
-                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                 RETURNING *
                 """,
                 (
@@ -336,10 +356,13 @@ class PostgresAcquisitionRepository:
                     query_id,
                     platform,
                     platform_item_id,
+                    unique_key,
                     canonical_url,
                     content_type,
+                    content_mode,
                     title,
                     author_name,
+                    body_text,
                     raw_summary,
                     status,
                     Json(source_payload or {}),
@@ -355,6 +378,44 @@ class PostgresAcquisitionRepository:
             row = cur.fetchone()
             return dict(row) if row else None
 
+    def get_candidate_item_by_unique_key(self, unique_key: str) -> CandidateItem | None:
+        row = self._one_or_none(
+            """
+            SELECT * FROM candidate_items
+            WHERE unique_key = %s
+            ORDER BY created_at DESC
+            LIMIT 1
+            """,
+            (unique_key,),
+        )
+        return CandidateItem.model_validate(row) if row else None
+
+    def attach_existing_candidate_item(
+        self,
+        item_id: UUID,
+        *,
+        job_id: UUID,
+        query_id: UUID,
+        metadata: dict[str, Any] | None = None,
+    ) -> CandidateItem:
+        row = self._one(
+            """
+            UPDATE candidate_items SET
+                job_id = %s,
+                query_id = %s,
+                metadata = metadata || %s
+            WHERE id = %s
+            RETURNING *
+            """,
+            (
+                job_id,
+                query_id,
+                Json(metadata or {}),
+                item_id,
+            ),
+        )
+        return CandidateItem.model_validate(row)
+
     def add_media_asset(
         self,
         *,
@@ -509,6 +570,7 @@ class PostgresAcquisitionRepository:
         item_ids = [row["id"] for row in items]
         media: list[dict[str, Any]] = []
         classifications: list[dict[str, Any]] = []
+        decode_summaries: list[dict[str, Any]] = []
         if item_ids:
             media = self._all(
                 """
@@ -526,23 +588,107 @@ class PostgresAcquisitionRepository:
                 """,
                 (item_ids,),
             )
+            decode_summaries = self._all(
+                """
+                SELECT DISTINCT ON (dr.item_id)
+                    dr.item_id,
+                    dr.status AS decode_status,
+                    COUNT(DISTINCT pd.id)::int AS payload_count
+                FROM decode_results dr
+                LEFT JOIN payload_drafts pd ON pd.item_id = dr.item_id
+                WHERE dr.item_id = ANY(%s)
+                GROUP BY dr.item_id, dr.status, dr.created_at
+                ORDER BY dr.item_id, dr.created_at DESC
+                """,
+                (item_ids,),
+            )
         return {
             "query": query,
             "jobs": jobs,
             "items": items,
             "media_assets": media,
             "classifications": classifications,
+            "decode_summaries": decode_summaries,
+        }
+
+    def get_query_detail_for_batch(self, *, batch_id: UUID, query_id: UUID) -> dict[str, Any]:
+        query = self._one(
+            "SELECT * FROM queries WHERE id = %s AND batch_id = %s",
+            (query_id, batch_id),
+        )
+        jobs = self._all(
+            """
+            SELECT aj.* FROM acquisition_jobs aj
+            JOIN acquisition_runs ar ON ar.id = aj.run_id
+            WHERE ar.batch_id = %s AND aj.query_id = %s
+            ORDER BY aj.created_at, aj.platform
+            """,
+            (batch_id, query_id),
+        )
+        items = self._all(
+            """
+            SELECT ci.* FROM candidate_items ci
+            JOIN acquisition_jobs aj ON aj.id = ci.job_id
+            JOIN acquisition_runs ar ON ar.id = aj.run_id
+            WHERE ar.batch_id = %s AND ci.query_id = %s
+            ORDER BY ci.platform, ci.created_at
+            """,
+            (batch_id, query_id),
+        )
+        item_ids = [row["id"] for row in items]
+        media: list[dict[str, Any]] = []
+        classifications: list[dict[str, Any]] = []
+        decode_summaries: list[dict[str, Any]] = []
+        if item_ids:
+            media = self._all(
+                """
+                SELECT * FROM media_assets
+                WHERE item_id = ANY(%s)
+                ORDER BY item_id, position
+                """,
+                (item_ids,),
+            )
+            classifications = self._all(
+                """
+                SELECT * FROM item_classifications
+                WHERE item_id = ANY(%s)
+                ORDER BY created_at DESC
+                """,
+                (item_ids,),
+            )
+            decode_summaries = self._all(
+                """
+                SELECT DISTINCT ON (dr.item_id)
+                    dr.item_id,
+                    dr.status AS decode_status,
+                    COUNT(DISTINCT pd.id)::int AS payload_count
+                FROM decode_results dr
+                LEFT JOIN payload_drafts pd ON pd.item_id = dr.item_id
+                WHERE dr.item_id = ANY(%s)
+                GROUP BY dr.item_id, dr.status, dr.created_at
+                ORDER BY dr.item_id, dr.created_at DESC
+                """,
+                (item_ids,),
+            )
+        return {
+            "query": query,
+            "jobs": jobs,
+            "items": items,
+            "media_assets": media,
+            "classifications": classifications,
+            "decode_summaries": decode_summaries,
         }
 
     def get_latest_singleton_overview(self) -> dict[str, Any]:
         batch = self._one_or_none(
             """
-            SELECT * FROM query_batches
-            WHERE generation_method = %s
-            ORDER BY created_at DESC
+            SELECT qb.* FROM query_batches qb
+            JOIN acquisition_runs ar ON ar.batch_id = qb.id
+            GROUP BY qb.id
+            ORDER BY MAX(ar.created_at) DESC
             LIMIT 1
             """,
-            ("creation_singleton_v1",),
+            (),
         )
         if batch is None:
             return {"batch": None, "run": None, "queries": [], "decoded_items": []}
@@ -550,11 +696,11 @@ class PostgresAcquisitionRepository:
         run = self._one_or_none(
             """
             SELECT * FROM acquisition_runs
-            WHERE batch_id = %s AND run_key LIKE %s
+            WHERE batch_id = %s
             ORDER BY created_at DESC
             LIMIT 1
             """,
-            (batch["id"], "singleton-acquisition:%"),
+            (batch["id"],),
         )
         if run is None:
             queries = self._all(
@@ -584,16 +730,21 @@ class PostgresAcquisitionRepository:
                 COUNT(DISTINCT ci.id)::int AS candidate_count,
                 COUNT(DISTINCT ic.id) FILTER (
                     WHERE ic.is_creation_knowledge IS TRUE
-                )::int AS creation_hit_count
+                )::int AS creation_hit_count,
+                COUNT(DISTINCT dr.id)::int AS decoded_count,
+                COUNT(DISTINCT pd.id)::int AS payload_count
             FROM queries q
-            LEFT JOIN acquisition_jobs aj ON aj.query_id = q.id AND aj.run_id = %s
+            LEFT JOIN acquisition_jobs aj ON aj.query_id = q.id
+            LEFT JOIN acquisition_runs ar ON ar.id = aj.run_id AND ar.batch_id = q.batch_id
             LEFT JOIN candidate_items ci ON ci.job_id = aj.id
             LEFT JOIN item_classifications ic ON ic.item_id = ci.id
+            LEFT JOIN decode_results dr ON dr.item_id = ci.id
+            LEFT JOIN payload_drafts pd ON pd.item_id = ci.id
             WHERE q.batch_id = %s
             GROUP BY q.id, q.query_text, q.metadata, q.sort_order
             ORDER BY q.sort_order, q.created_at
             """,
-            (run["id"], batch["id"]),
+            (batch["id"],),
         )
         decoded_items = self._all(
             """
@@ -606,13 +757,14 @@ class PostgresAcquisitionRepository:
                 COUNT(DISTINCT pd.id)::int AS payload_count
             FROM candidate_items ci
             JOIN acquisition_jobs aj ON aj.id = ci.job_id
+            JOIN acquisition_runs ar ON ar.id = aj.run_id
             JOIN decode_results dr ON dr.item_id = ci.id
             LEFT JOIN payload_drafts pd ON pd.item_id = ci.id
-            WHERE aj.run_id = %s
+            WHERE ar.batch_id = %s
             GROUP BY ci.id, ci.query_id, ci.title, ci.platform, dr.status, dr.created_at
             ORDER BY dr.created_at DESC
             """,
-            (run["id"],),
+            (batch["id"],),
         )
         return {
             "batch": batch,

acquisition/unique_key.py

@@ -0,0 +1,56 @@
+"""Stable business identity helpers for acquired candidate items."""
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import parse_qs, urlparse
+
+
+def _first(params: dict[str, list[str]], key: str) -> str:
+    values = params.get(key) or []
+    return values[0].strip() if values and values[0] else ""
+
+
+def parse_weixin_unique_key(url: str | None) -> str | None:
+    """Build the Weixin article key from URL params when all parts exist."""
+    if not url:
+        return None
+    params = parse_qs(urlparse(url).query, keep_blank_values=False)
+    biz = _first(params, "__biz")
+    mid = _first(params, "mid")
+    idx = _first(params, "idx")
+    sn = _first(params, "sn")
+    if not all([biz, mid, idx, sn]):
+        return None
+    return f"wx:{biz}:{mid}:{idx}:{sn}"
+
+
+def _raw_nested(raw_payload: dict[str, Any] | None, path: tuple[str, ...]) -> str:
+    current: Any = raw_payload or {}
+    for key in path:
+        if not isinstance(current, dict):
+            return ""
+        current = current.get(key)
+    return str(current).strip() if current else ""
+
+
+def build_unique_key(
+    platform: str,
+    *,
+    platform_item_id: str | None = None,
+    url: str | None = None,
+    raw_payload: dict[str, Any] | None = None,
+) -> str | None:
+    """Return the formal candidate item unique_key, or None if unknown."""
+    item_id = (platform_item_id or "").strip()
+    if platform == "xiaohongshu":
+        item_id = item_id or _raw_nested(raw_payload, ("candidate", "source_id"))
+        item_id = item_id or _raw_nested(raw_payload, ("detail", "data", "data", "channel_content_id"))
+        return f"xhs:{item_id}" if item_id else None
+    if platform == "douyin":
+        item_id = item_id or _raw_nested(raw_payload, ("candidate", "source_id"))
+        item_id = item_id or _raw_nested(raw_payload, ("candidate", "raw", "id"))
+        item_id = item_id or _raw_nested(raw_payload, ("detail", "data", "data", "channel_content_id"))
+        return f"dy:{item_id}" if item_id else None
+    if platform == "weixin":
+        return parse_weixin_unique_key(url)
+    return None

app/schemas.py

@@ -66,18 +66,25 @@ class ItemClassificationSchema(ApiSchema):
     confidence: float | None = None
     reason: str | None = None
     status: str
+    error_message: str | None = None
 
 
 class CandidateItemSchema(ApiSchema):
     id: UUID
     platform: str
     platform_item_id: str | None = None
+    unique_key: str | None = None
     canonical_url: str | None = None
     content_type: str | None = None
+    content_mode: str | None = None
     title: str | None = None
     author_name: str | None = None
+    body_text: str | None = None
     raw_summary: str | None = None
     status: str
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    error_message: str | None = None
+    decode_summary: dict[str, Any] | None = None
     media_assets: list[MediaAssetSchema] = Field(default_factory=list)
     classification: ItemClassificationSchema | None = None
 

core/jsonio.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 import json
 from typing import Any
 
+from core.text_limits import JSON_ERROR_SNIPPET_MAX_CHARS, clip_text
+
 
 def extract_json_object(text: str) -> dict:
     """取第一个 { 到最后一个 } 的切片解析为 dict。
@@ -12,7 +14,7 @@ def extract_json_object(text: str) -> dict:
     """
     start, end = text.find("{"), text.rfind("}")
     if start == -1 or end == -1 or end < start:
-        raise ValueError(f"no json object in output: {text[:120]!r}")
+        raise ValueError(f"no json object in output: {clip_text(text, JSON_ERROR_SNIPPET_MAX_CHARS)!r}")
     return json.loads(text[start : end + 1])
 
 

core/models.py

@@ -28,10 +28,12 @@ class Post(BaseModel):
 
     id: str  # xhs_<content_id>，复用为 ingest source.id
     platform: str = "xiaohongshu"
+    provider: str = ""
     url: str
     content_id: str
     title: str = ""
     content_type: str = ""  # video / 图文
+    content_mode: str = ""  # image_post / video_post / article / unsupported
     body_text: str = ""
     topic_list: list[str] = Field(default_factory=list)
     image_urls: list[str] = Field(default_factory=list)

core/text_limits.py

@@ -0,0 +1,24 @@
+"""Shared text length limits for the formal creation-knowledge pipeline."""
+from __future__ import annotations
+
+from typing import Any
+
+BODY_TEXT_MAX_CHARS = 50_000
+CLASSIFY_BODY_MAX_CHARS = 50_000
+RAW_SUMMARY_MAX_CHARS = 5_000
+REASON_MAX_CHARS = 10_000
+DIRECTIVE_CONTEXT_MAX_CHARS = 10_000
+QUERY_FILTER_REASON_MAX_CHARS = 5_000
+ERROR_MESSAGE_MAX_CHARS = 5_000
+TITLE_FALLBACK_MAX_CHARS = 1_000
+JSON_ERROR_SNIPPET_MAX_CHARS = 5_000
+
+
+def clip_text(value: Any, limit: int) -> str:
+    """Return text capped at a shared business limit."""
+    if value is None:
+        return ""
+    text = str(value)
+    if limit <= 0:
+        return ""
+    return text[:limit]

db/migrations/002_candidate_items_unique_key.sql

@@ -0,0 +1,54 @@
+-- 002: add formal candidate item unique_key.
+-- Applied to database: creation_knowledge_prod
+
+ALTER TABLE creation_knowledge.candidate_items
+ADD COLUMN IF NOT EXISTS unique_key text;
+
+WITH extracted AS (
+    SELECT
+        id,
+        platform,
+        COALESCE(
+            NULLIF(platform_item_id, ''),
+            NULLIF(source_payload #>> '{candidate,source_id}', ''),
+            NULLIF(source_payload #>> '{detail,data,data,channel_content_id}', ''),
+            NULLIF(substring(canonical_url from '/explore/([^/?#]+)'), '')
+        ) AS xhs_id,
+        COALESCE(
+            NULLIF(platform_item_id, ''),
+            NULLIF(source_payload #>> '{candidate,source_id}', ''),
+            NULLIF(source_payload #>> '{candidate,raw,id}', ''),
+            NULLIF(source_payload #>> '{detail,data,data,channel_content_id}', ''),
+            NULLIF(substring(canonical_url from '/video/([^/?#]+)'), '')
+        ) AS dy_id,
+        substring(canonical_url from '[?&]__biz=([^&#]+)') AS wx_biz,
+        substring(canonical_url from '[?&]mid=([^&#]+)') AS wx_mid,
+        substring(canonical_url from '[?&]idx=([^&#]+)') AS wx_idx,
+        substring(canonical_url from '[?&]sn=([^&#]+)') AS wx_sn
+    FROM creation_knowledge.candidate_items
+    WHERE unique_key IS NULL
+)
+UPDATE creation_knowledge.candidate_items ci
+SET unique_key = CASE
+    WHEN e.platform = 'xiaohongshu' AND e.xhs_id IS NOT NULL THEN 'xhs:' || e.xhs_id
+    WHEN e.platform = 'douyin' AND e.dy_id IS NOT NULL THEN 'dy:' || e.dy_id
+    WHEN e.platform = 'weixin'
+        AND e.wx_biz IS NOT NULL
+        AND e.wx_mid IS NOT NULL
+        AND e.wx_idx IS NOT NULL
+        AND e.wx_sn IS NOT NULL
+        THEN 'wx:' || e.wx_biz || ':' || e.wx_mid || ':' || e.wx_idx || ':' || e.wx_sn
+    ELSE NULL
+END
+FROM extracted e
+WHERE ci.id = e.id;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_candidate_items_unique_key
+ON creation_knowledge.candidate_items(unique_key)
+WHERE unique_key IS NOT NULL;
+
+INSERT INTO creation_knowledge.schema_migrations(version, description)
+VALUES ('002_candidate_items_unique_key', 'Add candidate_items.unique_key and backfill existing candidate identities')
+ON CONFLICT (version) DO UPDATE
+SET description = EXCLUDED.description,
+    applied_at = now();

db/migrations/003_candidate_items_content_mode.sql

@@ -0,0 +1,29 @@
+-- 003: add formal candidate item content_mode.
+-- Applied to database: creation_knowledge_prod
+
+ALTER TABLE creation_knowledge.candidate_items
+ADD COLUMN IF NOT EXISTS content_mode text;
+
+UPDATE creation_knowledge.candidate_items ci
+SET content_mode = CASE
+    WHEN ci.platform = 'weixin' THEN 'article'
+    WHEN EXISTS (
+        SELECT 1 FROM creation_knowledge.media_assets ma
+        WHERE ma.item_id = ci.id AND ma.media_type = 'video'
+    ) THEN 'video_post'
+    WHEN EXISTS (
+        SELECT 1 FROM creation_knowledge.media_assets ma
+        WHERE ma.item_id = ci.id AND ma.media_type IN ('image', 'cover', 'frame')
+    ) OR NULLIF(ci.raw_summary, '') IS NOT NULL THEN 'image_post'
+    ELSE 'unsupported'
+END
+WHERE ci.content_mode IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_candidate_items_content_mode
+ON creation_knowledge.candidate_items(content_mode);
+
+INSERT INTO creation_knowledge.schema_migrations(version, description)
+VALUES ('003_candidate_items_content_mode', 'Add candidate_items.content_mode and backfill business media modes')
+ON CONFLICT (version) DO UPDATE
+SET description = EXCLUDED.description,
+    applied_at = now();

db/migrations/004_candidate_items_body_text.sql

@@ -0,0 +1,15 @@
+-- 004: add full body text storage for formal candidate items.
+
+ALTER TABLE creation_knowledge.candidate_items
+ADD COLUMN IF NOT EXISTS body_text text;
+
+UPDATE creation_knowledge.candidate_items
+SET body_text = LEFT(NULLIF(source_payload #>> '{detail,data,data,body_text}', ''), 50000)
+WHERE body_text IS NULL
+  AND NULLIF(source_payload #>> '{detail,data,data,body_text}', '') IS NOT NULL;
+
+INSERT INTO creation_knowledge.schema_migrations(version, description)
+VALUES ('004_candidate_items_body_text', 'Add candidate_items.body_text for full post text')
+ON CONFLICT (version) DO UPDATE SET
+    description = EXCLUDED.description,
+    applied_at = now();

decode_content/gates.py

@@ -6,6 +6,7 @@ from typing import Any, Callable
 
 from core.llm import chat_json as default_chat_json
 from core.prompts import load_prompt
+from core.text_limits import DIRECTIVE_CONTEXT_MAX_CHARS, clip_text
 from decode_content.models import GateResult
 
 
@@ -82,7 +83,7 @@ def how_gate(knowledge: dict[str, Any], *, chat_json_fn: ChatJsonFn = default_ch
             "steps": [
                 {
                     "input": step.get("input"),
-                    "方法": (step.get("directive") or "")[:300],
+                    "方法": clip_text(step.get("directive") or "", DIRECTIVE_CONTEXT_MAX_CHARS),
                     "产出": step.get("output"),
                 }
                 for step in knowledge.get("steps", [])

decode_content/readers/service.py

@@ -1,8 +1,9 @@
 """Unified post reader for image-text and video decode inputs."""
 from __future__ import annotations
 
-from typing import Protocol
+from typing import Any, Protocol
 
+from acquisition.content_mode import infer_content_mode
 from acquisition.domain import CandidateItem, MediaAsset
 from core.config import Settings
 from core.models import Card, ExtractedContent, Post
@@ -21,6 +22,31 @@ class VideoReader(Protocol):
         ...
 
 
+class UnsupportedPostModeError(ValueError):
+    pass
+
+
+def _nested_text(payload: dict, *path: str) -> str:
+    current: Any = payload
+    for key in path:
+        if not isinstance(current, dict):
+            return ""
+        current = current.get(key)
+    return current if isinstance(current, str) else ""
+
+
+def _body_text_from_payload(source_payload: dict) -> str:
+    return (
+        _nested_text(source_payload, "body_text")
+        or _nested_text(source_payload, "detail", "body_text")
+        or _nested_text(source_payload, "detail", "data", "data", "body_text")
+        or _nested_text(source_payload, "candidate", "body_text")
+        or _nested_text(source_payload, "candidate", "raw", "body_text")
+        or _nested_text(source_payload, "desc")
+        or _nested_text(source_payload, "content")
+    )
+
+
 def _media_url(asset: MediaAsset) -> str | None:
     return asset.cdn_url or asset.oss_url or asset.source_url
 
@@ -51,21 +77,25 @@ def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]
                 )
             )
     source_payload = item.source_payload or {}
+    body_text = item.body_text or _body_text_from_payload(source_payload) or item.raw_summary or ""
     content_id = item.platform_item_id or (str(item.id) if item.id else "")
+    content_mode = item.content_mode or infer_content_mode(
+        platform=item.platform,
+        content_type=item.content_type or "",
+        body_text=body_text,
+        image_urls=image_urls,
+        video_urls=video_urls,
+        raw=source_payload if isinstance(source_payload, dict) 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 ""
-        ),
+        content_type=item.content_type or ("video" if content_mode == "video_post" else "图文"),
+        content_mode=content_mode,
+        body_text=body_text,
         topic_list=source_payload.get("topic_list") or source_payload.get("topics") or [],
         image_urls=image_urls,
         video_urls=video_urls,
@@ -133,7 +163,12 @@ def read_post(
     image_reader: ImageReader | None = None,
     video_reader: VideoReader | None = None,
 ) -> ReadResult:
-    if post.video_urls:
+    mode = post.content_mode or ("video_post" if post.video_urls else "image_post")
+    if mode == "unsupported":
+        raise UnsupportedPostModeError("unsupported_content_mode")
+    if mode == "video_post":
+        if not post.video_urls:
+            raise UnsupportedPostModeError("video_url_missing")
         if video_reader is not None:
             extracted = video_reader(post)
         else:
@@ -165,6 +200,7 @@ def read_item(
 
 __all__ = [
     "ImageReader",
+    "UnsupportedPostModeError",
     "VideoReader",
     "post_from_candidate_item",
     "read_item",

decode_content/scoping.py

@@ -7,6 +7,7 @@ from typing import Any, Protocol
 
 from core.llm import chat_json as default_chat_json
 from core.prompts import load_prompt
+from core.text_limits import DIRECTIVE_CONTEXT_MAX_CHARS, clip_text
 from decode_content.contracts import REUSE_THRESHOLD, SCOPE_TYPE_CN, SkillContract, load_contract
 from decode_content.gates import ChatJsonFn
 
@@ -31,7 +32,7 @@ def slim_knowledges(knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:
                 {
                     "id": step.get("id"),
                     "input": step.get("input"),
-                    "directive": (step.get("directive") or "")[:500],
+                    "directive": clip_text(step.get("directive") or "", DIRECTIVE_CONTEXT_MAX_CHARS),
                     "output": step.get("output"),
                 }
                 for step in knowledge.get("steps", [])

pipeline/acquisition_runner.py

@@ -8,6 +8,7 @@ from uuid import UUID
 from acquisition.repositories.base import AcquisitionRepository
 from acquisition.runner import RunBatchResult, run_batch
 from core.config import Settings
+from core.text_limits import ERROR_MESSAGE_MAX_CHARS, clip_text
 from pipeline.models import PipelineJob, PipelineRun
 from pipeline.repository import PipelineRepository
 
@@ -47,5 +48,9 @@ def run_acquisition_stage(
         return AcquisitionStageResult(pipeline_job=job, acquisition=result)
     except Exception as exc:
         if pipeline_repo and job and job.id:
-            pipeline_repo.mark_job_status(job.id, status="failed", error_message=str(exc)[:300])
+            pipeline_repo.mark_job_status(
+                job.id,
+                status="failed",
+                error_message=clip_text(exc, ERROR_MESSAGE_MAX_CHARS),
+            )
         raise

pipeline/dedupe.py

@@ -15,6 +15,8 @@ class DedupeDecision:
 
 
 def item_dedupe_key(item: CandidateItem) -> str:
+    if item.unique_key:
+        return f"unique_key:{item.unique_key}"
     if item.platform_item_id:
         return f"{item.platform}:id:{item.platform_item_id}"
     if item.canonical_url:
@@ -43,4 +45,6 @@ def should_decode_item(
     decoded = decoded_item_ids or set()
     if str(item.id) in decoded or key in decoded:
         return DedupeDecision(keep=False, key=key, reason="already_decoded")
+    if item.content_mode == "unsupported":
+        return DedupeDecision(keep=False, key=key, reason="unsupported_content_mode")
     return DedupeDecision(keep=True, key=key)

prompts/classify_imgtext.txt

@@ -38,5 +38,5 @@
 
 只输出一个 JSON 对象：
 {"is_empty": true/false,
- "reason": "一句话理由：是「教了哪种内容创作方法」，还是「属于应试/制作/学科/作品本身的哪一类」（≤30字）",
+ "reason": "判断理由：说明它教了哪种内容创作方法，或属于应试/制作/学科/作品本身的哪一类；保留必要证据，不要为了简短而省略关键判断依据",
  "knowledge": "is_empty=false 时：帖子里教的具体创作知识点全文（分条）；is_empty=true 时：空字符串"}

prompts/classify_video.txt

@@ -38,5 +38,5 @@
 
 只输出一个 JSON 对象：
 {"is_empty": true/false,
- "reason": "一句话理由：是「教了哪种内容创作方法」，还是「属于讲观点/应试/制作/学科/作品本身的哪一类」（≤30字）",
+ "reason": "判断理由：说明它教了哪种内容创作方法，或属于讲观点/应试/制作/学科/作品本身的哪一类；保留必要证据，不要为了简短而省略关键判断依据",
  "knowledge": "is_empty=false 时：视频里教的具体创作知识点全文（分条）；is_empty=true 时：空字符串"}

tests/test_app_api.py

@@ -96,10 +96,13 @@ class FakeAcquisitionRepo:
                     "id": self.item_id,
                     "platform": "xiaohongshu",
                     "platform_item_id": "xhs1",
+                    "unique_key": "xhs:xhs1",
                     "canonical_url": "https://xhs.test/1",
                     "content_type": "图文",
+                    "content_mode": "image_post",
                     "title": "脚本开头",
                     "author_name": "作者",
+                    "body_text": "先定受众，再写开头钩子",
                     "raw_summary": "先定受众",
                     "status": "candidate",
                 }
@@ -125,6 +128,7 @@ class FakeAcquisitionRepo:
                     "confidence": 0.98,
                     "reason": "可复用",
                     "status": "done",
+                    "error_message": None,
                 }
             ],
         }
@@ -239,6 +243,8 @@ def test_app_health_and_formal_acquisition_routes():
 
         detail = client.get(f"/api/acquisition/runs/{repo.run_id}/queries/{repo.query_id}").json()
         item = detail["platforms"]["xiaohongshu"]["items"][0]
+        assert item["content_mode"] == "image_post"
+        assert item["body_text"] == "先定受众，再写开头钩子"
         assert item["media_assets"][0]["cdn_url"] == "https://cdn.test/1.jpg"
         assert item["classification"]["is_creation_knowledge"] is True
     finally:
@@ -255,6 +261,15 @@ def test_query_generation_preview_defaults_to_first_two_families():
     assert data["metadata"]["active_family_keys"] == ["f1", "f2"]
 
 
+def test_query_generation_preview_can_return_full_cartesian_combinations():
+    client = TestClient(app)
+    data = client.get("/api/query-generation/preview?per=0&batch_n=1").json()
+
+    assert data["summary"]["family_keys"] == ["f1", "f2"]
+    assert data["summary"]["query_count"] == 36
+    assert [len(family["items"]) for family in data["families"]] == [18, 18]
+
+
 def test_formal_app_does_not_expose_legacy_static_mounts():
     paths = {route.path for route in app.routes if hasattr(route, "path")}
     assert "/data" not in paths

tests/test_classification_coarse.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import acquisition.classification.coarse as coarse
 import acquisition.classify as legacy_classify
 from core.config import PgConfig, Settings
+from core.text_limits import CLASSIFY_BODY_MAX_CHARS
 
 
 def _settings() -> Settings:
@@ -46,6 +47,30 @@ def test_formal_classify_imgtext_passes_http_image_urls(monkeypatch):
     assert {"type": "image_url", "image_url": {"url": "https://cdn.test/a.jpg"}} in user_content
 
 
+def test_formal_classify_imgtext_uses_wide_body_limit(monkeypatch):
+    captured = {}
+
+    def fake_judge(messages, settings, timeout):
+        captured["messages"] = messages
+        return 1, "ok", "长正文方法", ""
+
+    monkeypatch.setattr(coarse, "_judge", fake_judge)
+    long_body = "甲" * (CLASSIFY_BODY_MAX_CHARS + 7)
+
+    coarse.classify_imgtext(
+        {
+            "platform": "xiaohongshu",
+            "title": "长文",
+            "body_text": long_body,
+            "images": [],
+        },
+        _settings(),
+    )
+
+    user_text = captured["messages"][1]["content"][0]["text"]
+    assert user_text.count("甲") == CLASSIFY_BODY_MAX_CHARS
+
+
 def test_legacy_classify_imgtext_also_passes_http_image_urls(monkeypatch):
     captured = {}
 
@@ -70,6 +95,31 @@ def test_legacy_classify_imgtext_also_passes_http_image_urls(monkeypatch):
     assert {"type": "image_url", "image_url": {"url": "https://cdn.test/w.jpg"}} in user_content
 
 
+def test_legacy_classify_video_includes_title_and_body_text(monkeypatch):
+    captured = {}
+
+    def fake_judge(messages, settings, timeout):
+        captured["messages"] = messages
+        return 1, "ok", "视频方法", ""
+
+    monkeypatch.setattr(legacy_classify, "_judge", fake_judge)
+
+    legacy_classify.classify_video(
+        {
+            "platform": "douyin",
+            "title": "视频标题",
+            "body_text": "视频文案",
+            "video": "https://cdn.test/video.mp4",
+        },
+        _settings(),
+    )
+
+    user_text = captured["messages"][1]["content"][0]["text"]
+    assert "平台：douyin" in user_text
+    assert "标题：视频标题" in user_text
+    assert "正文/文案：视频文案" in user_text
+
+
 def test_coarse_classify_item_maps_judge_result(monkeypatch):
     monkeypatch.setattr(
         coarse,
@@ -88,3 +138,49 @@ def test_coarse_classify_item_maps_judge_result(monkeypatch):
     assert result.is_creation_knowledge is False
     assert result.label == "not_creation"
     assert result.status == "classified"
+
+
+def test_coarse_classify_item_routes_douyin_image_post_to_imgtext(monkeypatch):
+    called = {}
+
+    def fake_imgtext(payload, settings):
+        called["imgtext"] = payload
+        return 1, "是创作知识", "图片方法", ""
+
+    def fake_video(payload, settings):
+        called["video"] = payload
+        return 1, "不应调用", "", ""
+
+    monkeypatch.setattr(coarse, "classify_imgtext", fake_imgtext)
+    monkeypatch.setattr(coarse, "classify_video", fake_video)
+
+    result = coarse.coarse_classify_item(
+        platform="douyin",
+        content_mode="image_post",
+        title="图片帖",
+        body_text="图片正文",
+        image_urls=["https://cdn.test/dy.jpg"],
+        settings=_settings(),
+    )
+
+    assert result.is_creation_knowledge is True
+    assert "imgtext" in called
+    assert "video" not in called
+
+
+def test_coarse_classify_item_skips_unsupported_and_missing_video():
+    unsupported = coarse.coarse_classify_item(
+        platform="douyin",
+        content_mode="unsupported",
+        settings=_settings(),
+    )
+    missing_video = coarse.coarse_classify_item(
+        platform="douyin",
+        content_mode="video_post",
+        settings=_settings(),
+    )
+
+    assert unsupported.status == "skipped"
+    assert unsupported.label == "unsupported_content_mode"
+    assert missing_video.status == "skipped"
+    assert missing_video.label == "video_missing"

tests/test_db_migration_contract.py

@@ -4,12 +4,20 @@ from pathlib import Path
 
 
 MIGRATION = Path("db/migrations/001_creation_knowledge_schema.sql")
+MIGRATIONS_DIR = Path("db/migrations")
 
 
 def _migration_sql() -> str:
     return MIGRATION.read_text(encoding="utf-8")
 
 
+def _all_migration_sql() -> str:
+    return "\n".join(
+        path.read_text(encoding="utf-8")
+        for path in sorted(MIGRATIONS_DIR.glob("*.sql"))
+    )
+
+
 def test_creation_knowledge_migration_declares_formal_business_tables():
     sql = _migration_sql()
     expected_tables = [
@@ -49,7 +57,7 @@ def test_migration_is_replayable_and_not_sqlite_shaped():
 
 
 def test_migration_keeps_state_rows_traceable_and_resumable():
-    sql = _migration_sql()
+    sql = _all_migration_sql()
 
     for table in [
         "acquisition_runs",
@@ -70,3 +78,33 @@ def test_migration_keeps_state_rows_traceable_and_resumable():
     assert "response_payload jsonb NOT NULL DEFAULT '{}'::jsonb" in sql
     assert "UNIQUE (run_id, query_id, platform)" in sql
     assert "idx_candidate_items_platform_item" in sql
+
+
+def test_candidate_items_unique_key_migration_is_declared():
+    sql = _all_migration_sql()
+
+    assert "ADD COLUMN IF NOT EXISTS unique_key text" in sql
+    assert "idx_candidate_items_unique_key" in sql
+    assert "WHERE unique_key IS NOT NULL" in sql
+    assert "'002_candidate_items_unique_key'" in sql
+
+
+def test_candidate_items_content_mode_migration_is_declared():
+    sql = _all_migration_sql()
+
+    assert "ADD COLUMN IF NOT EXISTS content_mode text" in sql
+    assert "idx_candidate_items_content_mode" in sql
+    assert "video_post" in sql
+    assert "image_post" in sql
+    assert "article" in sql
+    assert "unsupported" in sql
+    assert "'003_candidate_items_content_mode'" in sql
+
+
+def test_candidate_items_body_text_migration_is_declared():
+    sql = _all_migration_sql()
+
+    assert "ADD COLUMN IF NOT EXISTS body_text text" in sql
+    assert "source_payload #>> '{detail,data,data,body_text}'" in sql
+    assert "LEFT(NULLIF(source_payload #>> '{detail,data,data,body_text}', ''), 50000)" in sql
+    assert "'004_candidate_items_body_text'" in sql

tests/test_decode_readers_service.py

@@ -4,7 +4,12 @@ from uuid import uuid4
 
 from acquisition.domain import CandidateItem, MediaAsset
 from core.models import Card, CardExtract, ExtractedContent, Post
-from decode_content.readers.service import post_from_candidate_item, read_item, read_post
+from decode_content.readers.service import (
+    UnsupportedPostModeError,
+    post_from_candidate_item,
+    read_item,
+    read_post,
+)
 
 
 def test_post_from_candidate_item_prefers_cdn_then_oss_then_source():
@@ -13,7 +18,9 @@ def test_post_from_candidate_item_prefers_cdn_then_oss_then_source():
         id=item_id,
         platform="xiaohongshu",
         platform_item_id="abc",
+        unique_key="xhs:abc",
         canonical_url="https://xhs/item/abc",
+        content_mode="image_post",
         title="脚本结构",
         author_name="sam",
         raw_summary="正文",
@@ -27,12 +34,47 @@ def test_post_from_candidate_item_prefers_cdn_then_oss_then_source():
     post = post_from_candidate_item(item, media)
 
     assert post.id == "xiaohongshu_abc"
+    assert post.content_id == "abc"
+    assert post.content_mode == "image_post"
     assert post.image_urls == ["oss0", "cdn1"]
     assert [card.index for card in post.cards] == [1, 2]
     assert post.body_text == "正文"
     assert post.topic_list == ["脚本"]
 
 
+def test_post_from_candidate_item_prefers_full_body_text_over_summary():
+    item_id = uuid4()
+    item = CandidateItem(
+        id=item_id,
+        platform="weixin",
+        platform_item_id="wx1",
+        content_mode="article",
+        title="长文",
+        body_text="完整正文",
+        raw_summary="摘要",
+    )
+
+    post = post_from_candidate_item(item, [])
+
+    assert post.body_text == "完整正文"
+
+
+def test_post_from_candidate_item_reads_legacy_nested_body_before_summary():
+    item_id = uuid4()
+    item = CandidateItem(
+        id=item_id,
+        platform="xiaohongshu",
+        platform_item_id="x1",
+        content_mode="image_post",
+        raw_summary="摘要",
+        source_payload={"detail": {"data": {"data": {"body_text": "旧 raw 正文"}}}},
+    )
+
+    post = post_from_candidate_item(item, [])
+
+    assert post.body_text == "旧 raw 正文"
+
+
 def test_read_post_imgtext_merges_text_cards_and_media():
     post = Post(
         id="xhs_1",
@@ -60,7 +102,13 @@ def test_read_post_imgtext_merges_text_cards_and_media():
 
 def test_read_item_routes_video_to_video_reader():
     item_id = uuid4()
-    item = CandidateItem(id=item_id, platform="douyin", platform_item_id="v1", title="视频")
+    item = CandidateItem(
+        id=item_id,
+        platform="douyin",
+        platform_item_id="v1",
+        content_mode="video_post",
+        title="视频",
+    )
     media = [MediaAsset(item_id=item_id, media_type="video", cdn_url="https://cdn/video.mp4", position=1)]
 
     def video_reader(post: Post) -> ExtractedContent:
@@ -73,3 +121,27 @@ def test_read_item_routes_video_to_video_reader():
     assert result.media["type"] == "video"
     assert result.cards[0].kind == "segment"
     assert result.cards[0].content == "片段知识"
+
+
+def test_read_post_rejects_unsupported_and_missing_video():
+    unsupported = Post(
+        id="dy_x",
+        platform="douyin",
+        url="u",
+        content_id="x",
+        content_mode="unsupported",
+    )
+    missing_video = Post(
+        id="dy_v",
+        platform="douyin",
+        url="u",
+        content_id="v",
+        content_mode="video_post",
+    )
+
+    import pytest
+
+    with pytest.raises(UnsupportedPostModeError, match="unsupported_content_mode"):
+        read_post(unsupported)
+    with pytest.raises(UnsupportedPostModeError, match="video_url_missing"):
+        read_post(missing_video)

tests/test_pipeline_formal.py

@@ -39,14 +39,32 @@ class FakePipelineRepo:
         return None
 
 
-def test_dedupe_candidate_items_prefers_platform_item_id_then_url():
-    first = CandidateItem(id=uuid4(), platform="xiaohongshu", platform_item_id="x1", canonical_url="https://a")
-    dup = CandidateItem(id=uuid4(), platform="xiaohongshu", platform_item_id="x1", canonical_url="https://b")
+def test_dedupe_candidate_items_prefers_unique_key_then_platform_item_id_then_url():
+    first = CandidateItem(
+        id=uuid4(),
+        platform="xiaohongshu",
+        platform_item_id="x1",
+        unique_key="xhs:x1",
+        canonical_url="https://a",
+    )
+    dup = CandidateItem(
+        id=uuid4(),
+        platform="xiaohongshu",
+        platform_item_id="x1",
+        unique_key="xhs:x1",
+        canonical_url="https://b",
+    )
+    platform_id_only = CandidateItem(id=uuid4(), platform="douyin", platform_item_id="dy1")
     url_only = CandidateItem(id=uuid4(), platform="weixin", canonical_url="HTTPS://MP.TEST/A")
 
-    assert item_dedupe_key(first) == "xiaohongshu:id:x1"
+    assert item_dedupe_key(first) == "unique_key:xhs:x1"
+    assert item_dedupe_key(platform_id_only) == "douyin:id:dy1"
     assert item_dedupe_key(url_only) == "url:https://mp.test/a"
-    assert dedupe_candidate_items([first, dup, url_only]) == [first, url_only]
+    assert dedupe_candidate_items([first, dup, platform_id_only, url_only]) == [
+        first,
+        platform_id_only,
+        url_only,
+    ]
     assert should_decode_item(first, decoded_item_ids={str(first.id)}).keep is False
 
 

tests/test_postgres_repository_contract.py

@@ -9,14 +9,20 @@ class RecordingPostgresRepo(PostgresAcquisitionRepository):
     def __init__(self):
         super().__init__(conn=None)
         self.one_calls = []
+        self.one_or_none_calls = []
         self.all_calls = []
         self.one_results = []
+        self.one_or_none_results = []
         self.all_results = []
 
     def _one(self, sql, params):
         self.one_calls.append((sql, params))
         return self.one_results.pop(0)
 
+    def _one_or_none(self, sql, params):
+        self.one_or_none_calls.append((sql, params))
+        return self.one_or_none_results.pop(0)
+
     def _all(self, sql, params):
         self.all_calls.append((sql, params))
         return self.all_results.pop(0)
@@ -79,6 +85,7 @@ def test_postgres_repository_lists_only_creation_candidates_for_decode():
                 "id": item_id,
                 "platform": "xiaohongshu",
                 "platform_item_id": "x1",
+                "unique_key": "xhs:x1",
                 "canonical_url": "https://xhs.test/1",
                 "status": "candidate",
             }
@@ -97,6 +104,63 @@ def test_postgres_repository_lists_only_creation_candidates_for_decode():
     assert items[0].platform == "xiaohongshu"
 
 
+def test_postgres_upsert_candidate_item_prefers_unique_key_for_existing_rows():
+    item_id = uuid4()
+    repo = RecordingPostgresRepo()
+    repo.one_or_none_results.append({"id": item_id})
+    repo.one_results.append(
+        {
+            "id": item_id,
+            "job_id": None,
+            "query_id": None,
+            "platform": "xiaohongshu",
+            "platform_item_id": "x1",
+            "unique_key": "xhs:x1",
+            "canonical_url": "https://xhs.test/1",
+            "content_type": "图文",
+            "content_mode": "image_post",
+            "title": "更新标题",
+            "author_name": None,
+            "published_at": None,
+            "body_text": "完整正文",
+            "raw_summary": None,
+            "status": "candidate",
+            "source_payload": {},
+            "metadata": {},
+            "error_message": None,
+            "created_at": None,
+            "updated_at": None,
+        }
+    )
+
+    item = repo.upsert_candidate_item(
+        platform="xiaohongshu",
+        platform_item_id="x1",
+        unique_key="xhs:x1",
+        canonical_url="https://xhs.test/1",
+        content_type="图文",
+        content_mode="image_post",
+        title="更新标题",
+        body_text="完整正文",
+    )
+
+    lookup_sql, lookup_params = repo.one_or_none_calls[0]
+    update_sql, update_params = repo.one_calls[0]
+    assert "WHERE unique_key = %s" in lookup_sql
+    assert lookup_params == ("xhs:x1",)
+    assert "source_payload =" not in update_sql
+    assert "unique_key = COALESCE(%s, unique_key)" in update_sql
+    assert "content_mode = %s" in update_sql
+    assert "body_text = %s" in update_sql
+    assert "INSERT INTO candidate_items" not in update_sql
+    assert update_params[2] == "xhs:x1"
+    assert update_params[5] == "image_post"
+    assert update_params[8] == "完整正文"
+    assert item.unique_key == "xhs:x1"
+    assert item.content_mode == "image_post"
+    assert item.body_text == "完整正文"
+
+
 def test_postgres_query_detail_loads_media_and_classifications_only_when_items_exist():
     run_id = uuid4()
     query_id = uuid4()
@@ -109,13 +173,56 @@ def test_postgres_query_detail_loads_media_and_classifications_only_when_items_e
             [{"id": item_id, "platform": "weixin", "query_id": query_id, "status": "candidate"}],
             [{"id": uuid4(), "item_id": item_id, "media_type": "image", "status": "done"}],
             [{"id": uuid4(), "item_id": item_id, "status": "classified"}],
+            [{"item_id": item_id, "decode_status": "decoded", "payload_count": 2}],
         ]
     )
 
     detail = repo.get_query_detail(run_id=run_id, query_id=query_id)
 
-    assert len(repo.all_calls) == 4
+    assert len(repo.all_calls) == 5
     assert "SELECT * FROM media_assets" in repo.all_calls[2][0]
     assert "SELECT * FROM item_classifications" in repo.all_calls[3][0]
+    assert "FROM decode_results dr" in repo.all_calls[4][0]
     assert detail["items"][0]["id"] == item_id
     assert detail["media_assets"][0]["item_id"] == item_id
+    assert detail["decode_summaries"][0]["payload_count"] == 2
+
+
+def test_postgres_attach_existing_candidate_updates_query_job_and_metadata():
+    item_id = uuid4()
+    job_id = uuid4()
+    query_id = uuid4()
+    repo = RecordingPostgresRepo()
+    repo.one_results.append(
+        {
+            "id": item_id,
+            "job_id": job_id,
+            "query_id": query_id,
+            "platform": "weixin",
+            "platform_item_id": "wx1",
+            "unique_key": "wx:key",
+            "canonical_url": "https://mp.weixin.qq.com/s?...",
+            "content_mode": "article",
+            "body_text": None,
+            "status": "candidate",
+            "source_payload": {},
+            "metadata": {"acquisition_match_status": "existing"},
+            "created_at": None,
+            "updated_at": None,
+        }
+    )
+
+    item = repo.attach_existing_candidate_item(
+        item_id,
+        job_id=job_id,
+        query_id=query_id,
+        metadata={"acquisition_match_status": "existing"},
+    )
+
+    sql, params = repo.one_calls[0]
+    assert "UPDATE candidate_items SET" in sql
+    assert "metadata = metadata || %s" in sql
+    assert params[0] == job_id
+    assert params[1] == query_id
+    assert params[3] == item_id
+    assert item.metadata["acquisition_match_status"] == "existing"

tests/test_unique_key.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from acquisition.unique_key import build_unique_key, parse_weixin_unique_key
+
+
+def test_build_unique_key_for_xiaohongshu_and_douyin():
+    assert (
+        build_unique_key("xiaohongshu", platform_item_id="685a477d00000000120211af")
+        == "xhs:685a477d00000000120211af"
+    )
+    assert (
+        build_unique_key("douyin", platform_item_id="7612592616106388138")
+        == "dy:7612592616106388138"
+    )
+
+
+def test_build_unique_key_reads_raw_payload_fallbacks():
+    assert (
+        build_unique_key("xiaohongshu", raw_payload={"candidate": {"source_id": "xhs-1"}})
+        == "xhs:xhs-1"
+    )
+    assert (
+        build_unique_key("douyin", raw_payload={"candidate": {"raw": {"id": "dy-1"}}})
+        == "dy:dy-1"
+    )
+
+
+def test_parse_weixin_unique_key_from_article_url():
+    url = (
+        "https://mp.weixin.qq.com/s?__biz=MzUxMDg4NDY1Ng=="
+        "&mid=2247484898&idx=2&sn=5069e65919f414b96b7aec53671dc1b5"
+        "&chksm=abc#rd"
+    )
+
+    assert (
+        parse_weixin_unique_key(url)
+        == "wx:MzUxMDg4NDY1Ng==:2247484898:2:5069e65919f414b96b7aec53671dc1b5"
+    )
+    assert build_unique_key("weixin", url=url) == parse_weixin_unique_key(url)
+
+
+def test_weixin_unique_key_is_none_when_url_lacks_article_parts():
+    assert parse_weixin_unique_key("https://mp.weixin.qq.com/s/a") is None
+    assert build_unique_key("weixin", platform_item_id="old-hash", url="https://mp.weixin.qq.com/s/a") is None