Selaa lähdekoodia

Add formal ingest metadata and API controls

SamLee 2 viikkoa sitten
vanhempi
commit
046f816561
37 muutettua tiedostoa jossa 857 lisäystä ja 85 poistoa
  1. 10 0
      .env.example
  2. 8 0
      app/frontend/dist/assets/index-BJBILN5Y.js
  3. 0 0
      app/frontend/dist/assets/index-DaHdrXy9.css
  4. 0 8
      app/frontend/dist/assets/index-DyD-4keP.js
  5. 2 2
      app/frontend/dist/index.html
  6. 8 0
      app/frontend/src/api/decode.js
  7. 31 2
      app/frontend/src/features/item-detail/ItemDetailPage.jsx
  8. 31 0
      app/frontend/src/styles/item-detail.css
  9. 47 2
      app/routes/decode.py
  10. 19 0
      app/schemas.py
  11. 27 0
      core/config.py
  12. 3 0
      core/models.py
  13. 1 6
      decode_content/contracts.py
  14. 193 0
      decode_content/ingest.py
  15. 89 7
      decode_content/payloads.py
  16. 8 0
      decode_content/readers/service.py
  17. 3 1
      decode_content/repositories/postgres.py
  18. 7 0
      decode_content/repository.py
  19. 7 2
      decode_content/service.py
  20. 14 16
      scripts/run_creation_pipeline.py
  21. 2 12
      scripts/run_creation_singleton.py
  22. 56 1
      tests/test_app_api.py
  23. 2 2
      tests/test_decode_contracts.py
  24. 113 4
      tests/test_decode_payloads.py
  25. 3 0
      tests/test_decode_readers_service.py
  26. 1 1
      tests/test_decode_service.py
  27. 135 0
      tests/test_ingest_client.py
  28. 3 3
      创作知识提取-skill/examples/金标样例.md
  29. 1 1
      创作知识提取-skill/extraction/phase3-assemble.md
  30. 6 0
      创作知识提取-skill/format/ingest-payload-how.schema.json
  31. 6 0
      创作知识提取-skill/format/ingest-payload-what.schema.json
  32. 7 1
      创作知识提取-skill/format/ingest-payload-why.schema.json
  33. 3 3
      创作知识提取-skill/taxonomy/知识类型.json
  34. 4 4
      创作知识提取-skill/tools/lint-payload.py
  35. 5 5
      开发文档/创作知识拆解框架.md
  36. 1 1
      开发文档/技术文档.md
  37. 1 1
      开发文档/正式版施工文档.md

+ 10 - 0
.env.example

@@ -55,6 +55,16 @@ CRAWLER_OSS_UPLOAD_URL=http://crawler-upload-v2.aiddit.com/crawler/oss/upload_st
 CRAWLER_OSS_UPLOAD_TIMEOUT_SECONDS=60
 CK_VIDEO_OSS_RETRY_DELAYS_SECONDS=5,10
 
+# -----------------------------------------------------------------------------
+# External knowledge ingest API
+# Leave CK_INGEST_API_URL empty to keep real ingest disabled. Dry-run ingest does
+# not call this endpoint.
+# -----------------------------------------------------------------------------
+CK_INGEST_API_URL=
+CK_INGEST_TOKEN=
+CK_INGEST_TIMEOUT_SECONDS=30
+CK_INGEST_RETRY_DELAYS_SECONDS=1,3
+
 # -----------------------------------------------------------------------------
 # LLM providers
 # -----------------------------------------------------------------------------

Tiedoston diff-näkymää rajattu, sillä se on liian suuri
+ 8 - 0
app/frontend/dist/assets/index-BJBILN5Y.js


Tiedoston diff-näkymää rajattu, sillä se on liian suuri
+ 0 - 0
app/frontend/dist/assets/index-DaHdrXy9.css


Tiedoston diff-näkymää rajattu, sillä se on liian suuri
+ 0 - 8
app/frontend/dist/assets/index-DyD-4keP.js


+ 2 - 2
app/frontend/dist/index.html

@@ -4,8 +4,8 @@
     <meta charset="UTF-8" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>创作知识</title>
-    <script type="module" crossorigin src="/app/assets/index-DyD-4keP.js"></script>
-    <link rel="stylesheet" crossorigin href="/app/assets/index-D4PSWR2J.css">
+    <script type="module" crossorigin src="/app/assets/index-BJBILN5Y.js"></script>
+    <link rel="stylesheet" crossorigin href="/app/assets/index-DaHdrXy9.css">
   </head>
   <body>
     <div id="root"></div>

+ 8 - 0
app/frontend/src/api/decode.js

@@ -3,3 +3,11 @@ import { request } from './client.js'
 export function getDecodeItemDetail(itemId, { signal } = {}) {
   return request(`/api/decode/items/${encodeURIComponent(itemId)}/detail`, { signal })
 }
+
+export function ingestDecodeItem(itemId, { dryRun = true, signal } = {}) {
+  return request(`/api/decode/items/${encodeURIComponent(itemId)}/ingest`, {
+    method: 'POST',
+    json: { dry_run: dryRun },
+    signal,
+  })
+}

+ 31 - 2
app/frontend/src/features/item-detail/ItemDetailPage.jsx

@@ -1,5 +1,5 @@
 import { useEffect, useMemo, useState } from 'react'
-import { getDecodeItemDetail } from '../../api/decode.js'
+import { getDecodeItemDetail, ingestDecodeItem } from '../../api/decode.js'
 import ErrorState from '../../components/ErrorState.jsx'
 import LoadingState from '../../components/LoadingState.jsx'
 import {
@@ -24,6 +24,7 @@ export default function ItemDetailPage({ itemId }) {
   const [err, setErr] = useState('')
   const [jsonPayload, setJsonPayload] = useState(null)
   const [zoom, setZoom] = useState(null)
+  const [ingestState, setIngestState] = useState({ loading: false, message: '' })
 
   useEffect(() => {
     setData(null)
@@ -41,6 +42,21 @@ export default function ItemDetailPage({ itemId }) {
     return () => controller.abort()
   }, [itemId])
 
+  async function runDryIngest() {
+    setIngestState({ loading: true, message: '' })
+    try {
+      const result = await ingestDecodeItem(itemId, { dryRun: true })
+      const refreshed = await getDecodeItemDetail(itemId)
+      setData(refreshed)
+      setIngestState({
+        loading: false,
+        message: `dry-run 已记录 ${result.ingested_count}/${result.total} 颗`,
+      })
+    } catch (e) {
+      setIngestState({ loading: false, message: e.message || 'dry-run 入库失败' })
+    }
+  }
+
   const scopesByParticle = useMemo(() => {
     const map = new Map()
     for (const scope of data?.scope_results || []) {
@@ -101,8 +117,21 @@ export default function ItemDetailPage({ itemId }) {
         <section className="item-knowledge-pane legacy-decode-page embedded-decode-page">
           <div className="pane-head">
             <span>提取知识</span>
-            <b>{particles.length} 颗</b>
+            <div className="item-ingest-actions">
+              <button
+                type="button"
+                className="item-ingest-btn"
+                disabled={ingestState.loading || particles.length === 0}
+                onClick={runDryIngest}
+              >
+                {ingestState.loading ? '记录中' : 'dry-run 入库'}
+              </button>
+              <b>{particles.length} 颗</b>
+            </div>
           </div>
+          {ingestState.message && (
+            <div className="item-ingest-message">{ingestState.message}</div>
+          )}
           {particles.length === 0 ? (
             <div className="no-knowledge">无知识</div>
           ) : (

+ 31 - 0
app/frontend/src/styles/item-detail.css

@@ -107,6 +107,37 @@
   font-size: 12px;
 }
 
+.item-ingest-actions {
+  display: flex;
+  align-items: center;
+  gap: 8px;
+}
+
+.item-ingest-btn {
+  height: 28px;
+  border: 1px solid #bfdbfe;
+  border-radius: 6px;
+  padding: 0 10px;
+  color: #1d4ed8;
+  font-size: 12px;
+  font-weight: 700;
+  background: #eff6ff;
+  cursor: pointer;
+}
+
+.item-ingest-btn:disabled {
+  color: #94a3b8;
+  cursor: not-allowed;
+  background: #f8fafc;
+}
+
+.item-ingest-message {
+  margin: 10px 12px 0;
+  color: #0f766e;
+  font-size: 12px;
+  font-weight: 700;
+}
+
 .item-post-pane .src {
   margin: 12px;
 }

+ 47 - 2
app/routes/decode.py

@@ -4,18 +4,23 @@ from __future__ import annotations
 from typing import Any
 from uuid import UUID
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Body, Depends, HTTPException
 
-from app.dependencies import get_acquisition_repository, get_decode_repository
+from app.dependencies import _env_file, get_acquisition_repository, get_decode_repository
 from app.schemas import (
     CandidateItemSchema,
     DecodeResultSchema,
+    IngestItemRequest,
+    IngestItemResponse,
+    IngestItemResultSchema,
     IngestRecordSchema,
     KnowledgeParticleSchema,
     MediaAssetSchema,
     PayloadDraftSchema,
     ScopeResultSchema,
 )
+from core.config import IngestApiConfig
+from decode_content.ingest import KnowledgeIngestClient, ingest_payload_draft
 
 router = APIRouter(prefix="/api/decode", tags=["decode"])
 
@@ -99,3 +104,43 @@ def enqueue_decode_job(
         raise HTTPException(status_code=501, detail="decode job writer is not implemented")
     job = create(item_id=item_id, status="pending")
     return {"job": job.model_dump(mode="json") if hasattr(job, "model_dump") else job}
+
+
+@router.post("/items/{item_id}/ingest", response_model=IngestItemResponse)
+def ingest_item_payloads(
+    item_id: UUID,
+    request: IngestItemRequest | None = Body(default=None),
+    repo: Any = Depends(get_decode_repository),
+) -> IngestItemResponse:
+    request = request or IngestItemRequest()
+    list_payloads = getattr(repo, "list_payload_drafts", None)
+    if list_payloads is None:
+        raise HTTPException(status_code=501, detail="payload draft reader is not implemented")
+    drafts = list_payloads(item_id=item_id)
+    client = None
+    if not request.dry_run:
+        config = IngestApiConfig.from_env(_env_file())
+        if not config.url:
+            raise HTTPException(status_code=400, detail="CK_INGEST_API_URL 未配置,真实 ingest 已关闭")
+        client = KnowledgeIngestClient(config)
+
+    records = []
+    for draft in drafts:
+        records.append(ingest_payload_draft(repo, draft, dry_run=request.dry_run, client=client))
+    results = [
+        IngestItemResultSchema(
+            payload_draft_id=record.payload_draft_id,
+            ingest_record=IngestRecordSchema.model_validate(record),
+        )
+        for record in records
+    ]
+    ingested_count = sum(1 for record in records if record.status == "ingested")
+    failed_count = sum(1 for record in records if record.status == "failed")
+    return IngestItemResponse(
+        item_id=item_id,
+        dry_run=request.dry_run,
+        total=len(drafts),
+        ingested_count=ingested_count,
+        failed_count=failed_count,
+        results=results,
+    )

+ 19 - 0
app/schemas.py

@@ -156,4 +156,23 @@ class IngestRecordSchema(ApiSchema):
     target_id: str | None = None
     status: str
     attempt_count: int = 0
+    response_payload: dict[str, Any] = Field(default_factory=dict)
     error_message: str | None = None
+
+
+class IngestItemRequest(ApiSchema):
+    dry_run: bool = True
+
+
+class IngestItemResultSchema(ApiSchema):
+    payload_draft_id: UUID | None = None
+    ingest_record: IngestRecordSchema
+
+
+class IngestItemResponse(ApiSchema):
+    item_id: UUID
+    dry_run: bool
+    total: int
+    ingested_count: int
+    failed_count: int
+    results: list[IngestItemResultSchema] = Field(default_factory=list)

+ 27 - 0
core/config.py

@@ -116,6 +116,33 @@ class CreationDbConfig:
         )
 
 
+@dataclass
+class IngestApiConfig:
+    """External knowledge ingest API configuration.
+
+    Empty url means real ingest is disabled; dry-run paths do not need it.
+    """
+
+    url: str = ""
+    token: str = ""
+    timeout_seconds: float = 30.0
+    retry_delays_seconds: tuple[float, ...] = (1.0, 3.0)
+
+    @classmethod
+    def from_env(cls, env_file: str | Path = ".env") -> "IngestApiConfig":
+        file_env = load_env_file(env_file)
+        return cls(
+            url=env_value("CK_INGEST_API_URL", file_env),
+            token=env_value("CK_INGEST_TOKEN", file_env),
+            timeout_seconds=float(env_value("CK_INGEST_TIMEOUT_SECONDS", file_env, "30")),
+            retry_delays_seconds=env_float_tuple(
+                "CK_INGEST_RETRY_DELAYS_SECONDS",
+                file_env,
+                "1,3",
+            ),
+        )
+
+
 @dataclass
 class Settings:
     """全流程配置聚合。M1 只用到 pg;其余字段供 M2+ 使用。"""

+ 3 - 0
core/models.py

@@ -31,10 +31,13 @@ class Post(BaseModel):
     provider: str = ""
     url: str
     content_id: str
+    unique_key: str = ""
     title: str = ""
     content_type: str = ""  # video / 图文
     content_mode: str = ""  # image_post / video_post / article / unsupported
     body_text: str = ""
+    media_count: int = 0
+    cover_url: Optional[str] = None
     topic_list: list[str] = Field(default_factory=list)
     image_urls: list[str] = Field(default_factory=list)
     video_urls: list[str] = Field(default_factory=list)

+ 1 - 6
decode_content/contracts.py

@@ -29,7 +29,7 @@ SCOPE_TYPE_CN = {
     "intent": "意图",
 }
 DIM_CREATIONS = ("创作",)
-DIM_ATTRIBUTE_BY_TYPE = {"how": "how工序", "what": "what构成", "why": "why原理"}
+DIM_ATTRIBUTE_BY_TYPE = {"how": "how", "what": "what", "why": "why"}
 CUSTOM_EXT_KEYS_BY_TYPE = {
     "how": ("业务阶段", "创作阶段", "动作", "出自"),
     "what": ("业务阶段", "概要", "出自"),
@@ -47,11 +47,6 @@ PROMPT_NAMES = (
 )
 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.",

+ 193 - 0
decode_content/ingest.py

@@ -0,0 +1,193 @@
+"""External knowledge ingest API client and draft-level orchestration."""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Any, Callable
+
+import httpx
+
+from core.config import IngestApiConfig
+from core.text_limits import ERROR_MESSAGE_MAX_CHARS, clip_text
+from decode_content.models import IngestRecord, PayloadDraft
+from decode_content.payloads import IngestPayloadValidationError, validate_ingest_payload
+from decode_content.repository import DecodeRepository
+
+PostFn = Callable[..., Any]
+
+
+class IngestApiError(RuntimeError):
+    def __init__(
+        self,
+        message: str,
+        *,
+        response_payload: dict[str, Any] | None = None,
+        http_status: int | None = None,
+        attempt_count: int = 1,
+    ) -> None:
+        super().__init__(message)
+        self.response_payload = response_payload or {}
+        self.http_status = http_status
+        self.attempt_count = attempt_count
+
+
+@dataclass(frozen=True)
+class IngestApiResult:
+    knowledge_id: str | None
+    source_id: str | None
+    http_status: int
+    attempt_count: int
+    response_payload: dict[str, Any]
+
+
+class KnowledgeIngestClient:
+    """Small synchronous client for POST /api/v1/knowledge/ingest."""
+
+    def __init__(
+        self,
+        config: IngestApiConfig,
+        *,
+        post_fn: PostFn | None = None,
+        sleep_fn: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self.config = config
+        self.post_fn = post_fn or httpx.post
+        self.sleep_fn = sleep_fn
+
+    def ingest(self, payload: dict[str, Any]) -> IngestApiResult:
+        validate_ingest_payload(payload)
+        if not self.config.url:
+            raise IngestApiError(
+                "CK_INGEST_API_URL 未配置,真实 ingest 已关闭",
+                response_payload={"dry_run": False, "missing_config": "CK_INGEST_API_URL"},
+            )
+        headers = {"Content-Type": "application/json"}
+        if self.config.token:
+            headers["Authorization"] = f"Bearer {self.config.token}"
+        delays = tuple(self.config.retry_delays_seconds)
+        last_error: IngestApiError | None = None
+        for attempt_index in range(len(delays) + 1):
+            attempt_count = attempt_index + 1
+            try:
+                response = self.post_fn(
+                    self.config.url,
+                    json=payload,
+                    headers=headers,
+                    timeout=self.config.timeout_seconds,
+                )
+                raw = _response_payload(response)
+                status_code = int(getattr(response, "status_code", 0) or 0)
+                if 200 <= status_code < 300:
+                    knowledge_id = raw.get("knowledge_id") if isinstance(raw, dict) else None
+                    source_id = raw.get("source_id") if isinstance(raw, dict) else None
+                    return IngestApiResult(
+                        knowledge_id=str(knowledge_id) if knowledge_id else None,
+                        source_id=str(source_id) if source_id else None,
+                        http_status=status_code,
+                        attempt_count=attempt_count,
+                        response_payload={
+                            "dry_run": False,
+                            "http_status": status_code,
+                            "knowledge_id": knowledge_id,
+                            "source_id": source_id,
+                            "attempt_count": attempt_count,
+                            "response": raw,
+                        },
+                    )
+                last_error = IngestApiError(
+                    f"ingest API 返回 {status_code}",
+                    response_payload={
+                        "dry_run": False,
+                        "http_status": status_code,
+                        "attempt_count": attempt_count,
+                        "response": raw,
+                    },
+                    http_status=status_code,
+                    attempt_count=attempt_count,
+                )
+            except IngestApiError as exc:
+                last_error = exc
+            except Exception as exc:  # network/client errors
+                last_error = IngestApiError(
+                    str(exc) or exc.__class__.__name__,
+                    response_payload={
+                        "dry_run": False,
+                        "exception_type": exc.__class__.__name__,
+                        "attempt_count": attempt_count,
+                    },
+                    attempt_count=attempt_count,
+                )
+            if attempt_index < len(delays):
+                self.sleep_fn(delays[attempt_index])
+        assert last_error is not None
+        raise last_error
+
+
+def _response_payload(response: Any) -> dict[str, Any] | str:
+    try:
+        value = response.json()
+    except Exception:
+        value = getattr(response, "text", "")
+    return value if isinstance(value, (dict, str)) else {"response": value}
+
+
+def ingest_payload_draft(
+    repo: DecodeRepository,
+    draft: PayloadDraft,
+    *,
+    dry_run: bool = True,
+    client: KnowledgeIngestClient | None = None,
+) -> IngestRecord:
+    payload = draft.payload or {}
+    source_id = ((payload.get("source") or {}).get("id") or "") if isinstance(payload, dict) else ""
+    try:
+        validate_ingest_payload(payload)
+        if dry_run:
+            updated = _mark_ingested(repo, draft)
+            return repo.save_ingest_record(
+                payload_draft_id=draft.id,
+                target_system="dry-run",
+                target_id=str(draft.id) if draft.id else None,
+                status="ingested",
+                attempt_count=1,
+                response_payload={
+                    "dry_run": True,
+                    "source_id": source_id,
+                    "note": "payload validated by formal ingest path; external ingest API not called",
+                    "payload": updated.payload if hasattr(updated, "payload") else payload,
+                },
+            )
+        if client is None:
+            raise IngestApiError("真实 ingest 需要 KnowledgeIngestClient")
+        result = client.ingest(payload)
+        _mark_ingested(repo, draft)
+        return repo.save_ingest_record(
+            payload_draft_id=draft.id,
+            target_system="knowledge-ingest-api",
+            target_id=result.knowledge_id,
+            status="ingested",
+            attempt_count=result.attempt_count,
+            response_payload=result.response_payload,
+        )
+    except (IngestPayloadValidationError, IngestApiError) as exc:
+        response_payload = getattr(exc, "response_payload", {}) or {}
+        response_payload.setdefault("dry_run", dry_run)
+        response_payload.setdefault("source_id", source_id)
+        return repo.save_ingest_record(
+            payload_draft_id=draft.id,
+            target_system="dry-run" if dry_run else "knowledge-ingest-api",
+            target_id=None,
+            status="failed",
+            attempt_count=getattr(exc, "attempt_count", 1),
+            response_payload=response_payload,
+            error_message=clip_text(str(exc), ERROR_MESSAGE_MAX_CHARS),
+        )
+
+
+def _mark_ingested(repo: DecodeRepository, draft: PayloadDraft) -> PayloadDraft:
+    if draft.id is None:
+        return draft
+    marker = getattr(repo, "mark_payload_draft_ingested", None)
+    if marker is None:
+        return draft
+    return marker(draft.id)

+ 89 - 7
decode_content/payloads.py

@@ -9,6 +9,17 @@ from decode_content.contracts import DIM_ATTRIBUTE_BY_TYPE
 
 TYPE2ATTR = DIM_ATTRIBUTE_BY_TYPE
 
+CUSTOM_EXT_TYPES = {"str", "num", "date"}
+DIM_ATTRIBUTES = {"how", "what", "why"}
+
+
+class IngestPayloadValidationError(ValueError):
+    """Raised when a generated payload cannot satisfy the ingest API contract."""
+
+
+def _text(value: Any) -> str:
+    return value if isinstance(value, str) else ""
+
 
 def _post_value(post: Any, key: str, default: Any = "") -> Any:
     if isinstance(post, dict):
@@ -46,7 +57,7 @@ def build_content(knowledge: dict[str, Any]) -> str | dict[str, Any]:
                 for item in (knowledge.get("body") or [])
             ],
         }
-    return knowledge.get("阐述") or knowledge.get("desc") or ""
+    return knowledge.get("阐述") or knowledge.get("desc") or knowledge.get("explanation") or ""
 
 
 def _payload_content(knowledge: dict[str, Any]) -> str:
@@ -54,6 +65,58 @@ def _payload_content(knowledge: dict[str, Any]) -> str:
     return content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
 
 
+def validate_ingest_payload(payload: dict[str, Any]) -> None:
+    """Validate the minimal external ingest API contract before storing/sending."""
+
+    errors: list[str] = []
+    source = payload.get("source")
+    if not isinstance(source, dict):
+        source = {}
+        errors.append("source 必须是对象")
+    source_id = _text(source.get("id")).strip()
+    if not source_id:
+        errors.append("source.id 不能为空")
+    elif len(source_id) > 64:
+        errors.append("source.id 最长 64 字符")
+    source_type = _text(source.get("source_type"))
+    if len(source_type) > 32:
+        errors.append("source.source_type 最长 32 字符")
+    source_title = _text(source.get("title"))
+    if len(source_title) > 512:
+        errors.append("source.title 最长 512 字符")
+    source_author = _text(source.get("author"))
+    if len(source_author) > 128:
+        errors.append("source.author 最长 128 字符")
+    title = payload.get("title")
+    if title is not None and len(_text(title)) > 512:
+        errors.append("title 最长 512 字符")
+    content = payload.get("content")
+    if not isinstance(content, str) or not content.strip():
+        errors.append("content 必须是非空字符串")
+    dim_attributes = payload.get("dim_attributes")
+    if not isinstance(dim_attributes, list) or len(dim_attributes) != 1:
+        errors.append("dim_attributes 必须恰好包含一个值")
+    elif dim_attributes[0] not in DIM_ATTRIBUTES:
+        errors.append("dim_attributes 只能是 how/what/why")
+    for idx, scope in enumerate(payload.get("scopes") or [], 1):
+        if not isinstance(scope, dict):
+            errors.append(f"scopes[{idx}] 必须是对象")
+            continue
+        value = _text(scope.get("value"))
+        if len(value) > 128:
+            errors.append(f"scopes[{idx}].value 最长 128 字符")
+    for idx, ext in enumerate(payload.get("custom_ext") or [], 1):
+        if not isinstance(ext, dict):
+            errors.append(f"custom_ext[{idx}] 必须是对象")
+            continue
+        ext_type = ext.get("type")
+        if ext_type not in CUSTOM_EXT_TYPES:
+            errors.append(f"custom_ext[{idx}].type 必须是 str/num/date")
+    if errors:
+        title_hint = _text(payload.get("title")) or source_id or "unknown"
+        raise IngestPayloadValidationError(f"{title_hint}: {'; '.join(errors)}")
+
+
 def _add_scopes(scopes: list[dict[str, str]], seen: set[tuple[str, str]], items: list[dict[str, Any]]) -> None:
     for item in items:
         scope_type = item.get("scope_type")
@@ -78,9 +141,29 @@ def collect_payload_scopes(knowledge: dict[str, Any]) -> list[dict[str, str]]:
     return scopes
 
 
+def _source_metadata(post: Any) -> dict[str, Any]:
+    metadata = {
+        "platform": _post_value(post, "platform"),
+        "url": _post_value(post, "url"),
+        "unique_key": _post_value(post, "unique_key"),
+        "platform_item_id": _post_value(post, "content_id"),
+        "content_type": _post_value(post, "content_type"),
+        "content_mode": _post_value(post, "content_mode"),
+        "media_count": _post_value(post, "media_count", 0),
+        "cover_url": _post_value(post, "cover_url"),
+    }
+    return {
+        key: value
+        for key, value in metadata.items()
+        if value is not None and value != ""
+    }
+
+
 def build_payload(post: Any, knowledge: dict[str, Any], how_titles: dict[str, str] | None = None) -> dict[str, Any]:
     how_titles = how_titles or {}
     t = knowledge.get("type")
+    if t not in TYPE2ATTR:
+        raise IngestPayloadValidationError(f"{knowledge.get('title') or 'unknown'}: unknown knowledge type {t}")
     ext = [
         {"key": "业务阶段", "type": "str", "value": stage}
         for stage in (knowledge.get("业务阶段") or [])
@@ -110,24 +193,23 @@ def build_payload(post: Any, knowledge: dict[str, Any], how_titles: dict[str, st
                 "value": f"{how_titles.get(parent.get('how_id'), parent.get('how_id'))} 第{parent.get('step')}步",
             }
         )
-    return {
+    payload = {
         "source": {
             "id": _post_value(post, "id"),
             "source_type": "post",
             "title": _post_value(post, "title") or "",
             "author": _post_value(post, "author_name") or "",
-            "source_metadata": {
-                "platform": _post_value(post, "platform"),
-                "url": _post_value(post, "url"),
-            },
+            "source_metadata": _source_metadata(post),
         },
         "title": knowledge.get("title"),
         "content": _payload_content(knowledge),
         "dim_creations": ["创作"],
-        "dim_attributes": [TYPE2ATTR.get(t, "how工序")],
+        "dim_attributes": [TYPE2ATTR[t]],
         "scopes": collect_payload_scopes(knowledge),
         "custom_ext": ext,
     }
+    validate_ingest_payload(payload)
+    return payload
 
 
 def build_payloads(post: Any, knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:

+ 8 - 0
decode_content/readers/service.py

@@ -62,10 +62,13 @@ def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]
     image_urls: list[str] = []
     video_urls: list[str] = []
     cards: list[Card] = []
+    media_count = 0
+    cover_url: str | None = None
     for asset in ordered:
         url = _media_url(asset)
         if not url:
             continue
+        media_count += 1
         media_type = asset.media_type.lower()
         if media_type == "video":
             stable_url = _stable_media_url(asset)
@@ -73,6 +76,8 @@ def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]
                 video_urls.append(stable_url)
             continue
         if media_type in {"image", "cover", "frame"}:
+            if cover_url is None:
+                cover_url = url
             image_urls.append(url)
             cards.append(
                 Card(
@@ -98,10 +103,13 @@ def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]
         platform=item.platform,
         url=item.canonical_url or "",
         content_id=content_id,
+        unique_key=item.unique_key or "",
         title=item.title or "",
         content_type=item.content_type or ("video" if content_mode == "video_post" else "图文"),
         content_mode=content_mode,
         body_text=body_text,
+        media_count=media_count,
+        cover_url=cover_url,
         topic_list=source_payload.get("topic_list") or source_payload.get("topics") or [],
         image_urls=image_urls,
         video_urls=video_urls,

+ 3 - 1
decode_content/repositories/postgres.py

@@ -308,6 +308,7 @@ class PostgresDecodeRepository:
         target_system: str,
         target_id: str | None = None,
         status: str = "pending",
+        attempt_count: int = 1,
         response_payload: dict[str, Any] | None = None,
         error_message: str | None = None,
     ) -> IngestRecord:
@@ -317,7 +318,7 @@ class PostgresDecodeRepository:
                 payload_draft_id, target_system, target_id, status,
                 attempt_count, response_payload, error_message
             )
-            VALUES (%s, %s, %s, %s, 1, %s, %s)
+            VALUES (%s, %s, %s, %s, %s, %s, %s)
             RETURNING *
             """,
             (
@@ -325,6 +326,7 @@ class PostgresDecodeRepository:
                 target_system,
                 target_id,
                 status,
+                attempt_count,
                 _json(response_payload),
                 error_message,
             ),

+ 7 - 0
decode_content/repository.py

@@ -82,6 +82,9 @@ class DecodeRepository(Protocol):
     def list_payload_drafts(self, item_id: UUID | None = None) -> list[PayloadDraft]:
         ...
 
+    def mark_payload_draft_ingested(self, payload_draft_id: UUID) -> PayloadDraft:
+        ...
+
     def save_ingest_record(
         self,
         *,
@@ -89,11 +92,15 @@ class DecodeRepository(Protocol):
         target_system: str,
         target_id: str | None = None,
         status: str = "pending",
+        attempt_count: int = 1,
         response_payload: dict[str, Any] | None = None,
         error_message: str | None = None,
     ) -> IngestRecord:
         ...
 
+    def list_ingest_records(self, item_id: UUID | None = None) -> list[IngestRecord]:
+        ...
+
     def save_contract_snapshot(
         self,
         *,

+ 7 - 2
decode_content/service.py

@@ -13,7 +13,7 @@ from decode_content.contracts import SkillContract, load_contract
 from decode_content.framing import frame_and_clean
 from decode_content.gates import ChatJsonFn, creation_gate
 from decode_content.models import DecodeResult, GateResult, PayloadDraft, ReadResult
-from decode_content.payloads import build_payloads
+from decode_content.payloads import build_payloads, validate_ingest_payload
 from decode_content.readers.service import read_post
 from decode_content.repository import DecodeRepository
 from decode_content.scoping import ScopeLinker, apply_scopes, nounify_scopes, scope_candidates
@@ -219,7 +219,7 @@ class DecodeService:
                 self.repository.save_payload_draft(
                     item_id=item_id,
                     particle_id=particle.id,
-                    payload=payload,
+                    payload=_validated_payload(payload),
                     review_status="pending",
                     ingest_ready=False,
                     status="draft",
@@ -242,6 +242,11 @@ class DecodeService:
         )
 
 
+def _validated_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    validate_ingest_payload(payload)
+    return payload
+
+
 def decode_post(
     *,
     item_id: UUID,

+ 14 - 16
scripts/run_creation_pipeline.py

@@ -9,8 +9,9 @@ from uuid import UUID
 
 from acquisition.repositories.postgres import PostgresAcquisitionRepository
 from acquisition.runner import DEFAULT_PLATFORMS, run_batch
-from core.config import CreationDbConfig, Settings
+from core.config import CreationDbConfig, IngestApiConfig, Settings
 from core.db_session import transaction
+from decode_content.ingest import KnowledgeIngestClient, ingest_payload_draft
 from decode_content.repositories.postgres import PostgresDecodeRepository
 from decode_content.service import DecodeService
 from pipeline.decode_runner import run_decode_stage
@@ -26,24 +27,14 @@ def _model_dump(value):
     return dict(value)
 
 
-def _dry_ingest_payloads(repo: PostgresDecodeRepository, outputs: list) -> list[dict]:
+def _ingest_payloads(repo: PostgresDecodeRepository, outputs: list, *, dry_run: bool, env_file: str) -> list[dict]:
     records: list[dict] = []
+    client = None if dry_run else KnowledgeIngestClient(IngestApiConfig.from_env(env_file))
     for output in outputs:
         for draft in output.payload_drafts:
             if draft.id is None:
                 continue
-            repo.mark_payload_draft_ingested(draft.id)
-            record = repo.save_ingest_record(
-                payload_draft_id=draft.id,
-                target_system="dry-run",
-                target_id=str(draft.id),
-                status="ingested",
-                response_payload={
-                    "dry_run": True,
-                    "note": "payload generated by formal creation pipeline; external ingest API not called",
-                    "payload": draft.payload,
-                },
-            )
+            record = ingest_payload_draft(repo, draft, dry_run=dry_run, client=client)
             records.append(_model_dump(record))
     return records
 
@@ -65,6 +56,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--no-resume", action="store_true")
     parser.add_argument("--no-skip-done", action="store_true")
     parser.add_argument("--no-dry-ingest-record", action="store_true")
+    parser.add_argument("--real-ingest", action="store_true", help="Call CK_INGEST_API_URL instead of dry-run records.")
     return parser.parse_args(argv)
 
 
@@ -100,7 +92,12 @@ def main(argv: list[str] | None = None) -> int:
             run_id=acquisition.run_id,
             limit=args.decode_limit,
         )
-        ingest_records = [] if args.no_dry_ingest_record else _dry_ingest_payloads(decode_repo, decode.outputs)
+        ingest_records = [] if args.no_dry_ingest_record else _ingest_payloads(
+            decode_repo,
+            decode.outputs,
+            dry_run=not args.real_ingest,
+            env_file=args.env_file,
+        )
 
     print(json.dumps(
         {
@@ -109,7 +106,8 @@ def main(argv: list[str] | None = None) -> int:
             "platforms": list(platforms),
             "acquisition": _model_dump(acquisition),
             "decode": _model_dump(decode),
-            "dry_ingest_records": ingest_records,
+            "ingest_records": ingest_records,
+            "dry_ingest_records": ingest_records if not args.real_ingest else [],
         },
         ensure_ascii=False,
         default=str,

+ 2 - 12
scripts/run_creation_singleton.py

@@ -24,6 +24,7 @@ from acquisition.repositories.postgres import PostgresAcquisitionRepository  # n
 from acquisition.runner import DEFAULT_PLATFORMS, run_batch  # noqa: E402
 from core.config import CreationDbConfig, Settings  # noqa: E402
 from core.db_session import transaction  # noqa: E402
+from decode_content.ingest import ingest_payload_draft  # noqa: E402
 from decode_content.repositories.postgres import PostgresDecodeRepository  # noqa: E402
 from decode_content.service import DecodeService  # noqa: E402
 from pipeline.decode_runner import run_decode_stage  # noqa: E402
@@ -73,18 +74,7 @@ def _dry_ingest_payloads(
         for draft in output.payload_drafts:
             if draft.id is None:
                 continue
-            repo.mark_payload_draft_ingested(draft.id)
-            record = repo.save_ingest_record(
-                payload_draft_id=draft.id,
-                target_system="dry-run",
-                target_id=str(draft.id),
-                status="ingested",
-                response_payload={
-                    "dry_run": True,
-                    "note": "payload generated by singleton pipeline; external ingest API not called",
-                    "payload": draft.payload,
-                },
-            )
+            record = ingest_payload_draft(repo, draft, dry_run=True)
             records.append(_model_dump(record))
     return records
 

+ 56 - 1
tests/test_app_api.py

@@ -14,6 +14,7 @@ from app.dependencies import (
     get_decode_repository,
     get_pipeline_repository,
 )
+from app.routes import decode as decode_routes
 from app.routes import manual_queries
 from decode_content.models import (
     DecodeJob,
@@ -218,6 +219,9 @@ class FakeDecodeRepo:
         self.decode_result_id = uuid4()
         self.decode_job_id = uuid4()
         self.payload_id = uuid4()
+        self.other_payload_id = uuid4()
+        self.marked_payload_ids = []
+        self.saved_ingest_records = []
 
     def get_decode_result_for_item(self, item_id):
         return DecodeResult(
@@ -242,13 +246,37 @@ class FakeDecodeRepo:
             PayloadDraft(
                 id=self.payload_id,
                 item_id=item_id or uuid4(),
-                payload={"title": "可审核 payload"},
+                payload={
+                    "source": {"id": "xhs_1", "source_type": "post", "title": "脚本开头", "author": "作者"},
+                    "title": "可审核 payload",
+                    "content": "先定受众,再写开头钩子。",
+                    "dim_attributes": ["how"],
+                    "dim_creations": ["创作"],
+                    "scopes": [{"scope_type": "intent", "value": "吸引注意"}],
+                    "custom_ext": [{"key": "业务阶段", "type": "str", "value": "脚本"}],
+                },
                 review_status="pending",
                 ingest_ready=False,
                 status="draft",
             )
         ]
 
+    def mark_payload_draft_ingested(self, payload_draft_id):
+        self.marked_payload_ids.append(payload_draft_id)
+        return PayloadDraft(
+            id=payload_draft_id,
+            item_id=uuid4(),
+            payload=self.list_payload_drafts()[0].payload,
+            review_status="approved",
+            ingest_ready=True,
+            status="ingested",
+        )
+
+    def save_ingest_record(self, **kwargs):
+        record = IngestRecord(id=uuid4(), **kwargs)
+        self.saved_ingest_records.append(record)
+        return record
+
     def list_knowledge_particles(self, item_id=None):
         return [
             KnowledgeParticle(
@@ -273,6 +301,8 @@ class FakeDecodeRepo:
         ]
 
     def list_ingest_records(self, item_id=None):
+        if self.saved_ingest_records:
+            return self.saved_ingest_records
         return [
             IngestRecord(
                 id=uuid4(),
@@ -535,6 +565,14 @@ def test_formal_decode_payload_pipeline_routes_have_override_success_paths():
         assert drafts[0]["item_id"] == str(item_id)
         assert drafts[0]["payload"]["title"] == "可审核 payload"
 
+        ingest = client.post(f"/api/decode/items/{item_id}/ingest").json()
+        assert ingest["dry_run"] is True
+        assert ingest["total"] == 1
+        assert ingest["ingested_count"] == 1
+        assert ingest["failed_count"] == 0
+        assert decode_repo.marked_payload_ids == [decode_repo.payload_id]
+        assert decode_repo.saved_ingest_records[0].target_system == "dry-run"
+
         detail = client.get(f"/api/decode/items/{item_id}/detail").json()
         assert detail["item"]["id"] == str(item_id)
         assert detail["knowledge_particles"][0]["title"] == "What I know"
@@ -547,6 +585,23 @@ def test_formal_decode_payload_pipeline_routes_have_override_success_paths():
         app.dependency_overrides.clear()
 
 
+def test_item_ingest_real_mode_requires_configured_url(monkeypatch):
+    acquisition_repo = FakeAcquisitionRepo()
+    decode_repo = FakeDecodeRepo()
+    item_id = acquisition_repo.item_id
+    app.dependency_overrides[get_decode_repository] = lambda: decode_repo
+    monkeypatch.setattr(decode_routes.IngestApiConfig, "from_env", lambda _env_file: decode_routes.IngestApiConfig())
+    client = TestClient(app)
+
+    try:
+        response = client.post(f"/api/decode/items/{item_id}/ingest", json={"dry_run": False})
+        assert response.status_code == 400
+        assert "CK_INGEST_API_URL" in response.json()["detail"]
+        assert decode_repo.saved_ingest_records == []
+    finally:
+        app.dependency_overrides.clear()
+
+
 def test_no_legacy_api_dependencies_in_formal_app_sources():
     roots = [Path("app"), Path("pipeline")]
     text = "\n".join(

+ 2 - 2
tests/test_decode_contracts.py

@@ -32,7 +32,7 @@ def test_contract_constants_record_formal_business_terms():
     assert CREATION_STAGES == ("定向", "构思", "结构", "成文", "打磨")
     assert WHAT_KINDS == ("子集", "多维关系", "序列")
     assert SCOPE_TYPES == ("substance", "form", "feeling", "effect", "intent")
-    assert DIM_ATTRIBUTE_BY_TYPE == {"how": "how工序", "what": "what构成", "why": "why原理"}
+    assert DIM_ATTRIBUTE_BY_TYPE == {"how": "how", "what": "what", "why": "why"}
 
 
 def test_contract_snapshots_can_feed_repository():
@@ -41,7 +41,7 @@ def test_contract_snapshots_can_feed_repository():
     assert snapshots[0].contract_name == "创作知识提取-skill"
     assert snapshots[0].contract_type == "skill"
     assert snapshots[0].content_hash and len(snapshots[0].content_hash) == 64
-    assert snapshots[0].snapshot["constants"]["dim_attribute_by_type"]["how"] == "how工序"
+    assert snapshots[0].snapshot["constants"]["dim_attribute_by_type"]["how"] == "how"
     assert any(s.contract_type == "prompt_phase" for s in snapshots)
     assert any(s.contract_type == "schema" for s in snapshots)
     assert any(s.contract_type == "prompt" for s in snapshots)

+ 113 - 4
tests/test_decode_payloads.py

@@ -2,8 +2,10 @@ from __future__ import annotations
 
 import json
 
+import pytest
+
 from core.models import Post
-from decode_content.payloads import build_payload, build_payloads
+from decode_content.payloads import IngestPayloadValidationError, build_payload, build_payloads, validate_ingest_payload
 
 
 def _post() -> Post:
@@ -12,7 +14,12 @@ def _post() -> Post:
         platform="xiaohongshu",
         url="https://xhs/1",
         content_id="1",
+        unique_key="xhs:1",
         title="脚本教程",
+        content_type="note",
+        content_mode="image_post",
+        media_count=2,
+        cover_url="https://cdn.test/cover.webp",
         author_name="作者",
     )
 
@@ -41,7 +48,17 @@ def test_build_how_payload_uses_formal_dim_attribute_and_scope_union():
 
     payload = build_payload(_post(), knowledge)
 
-    assert payload["dim_attributes"] == ["how工序"]
+    assert payload["dim_attributes"] == ["how"]
+    assert payload["source"]["source_metadata"] == {
+        "platform": "xiaohongshu",
+        "url": "https://xhs/1",
+        "unique_key": "xhs:1",
+        "platform_item_id": "1",
+        "content_type": "note",
+        "content_mode": "image_post",
+        "media_count": 2,
+        "cover_url": "https://cdn.test/cover.webp",
+    }
     assert "输入:选题" in payload["content"]
     assert "指引:收窄受众" in payload["content"]
     assert payload["scopes"] == [{"scope_type": "intent", "value": "吸引注意"}]
@@ -71,9 +88,101 @@ def test_build_what_and_why_payload_content_is_string_contract():
 
     what_payload, why_payload = build_payloads(_post(), [what, why])
 
-    assert what_payload["dim_attributes"] == ["what构成"]
+    assert what_payload["dim_attributes"] == ["what"]
     assert json.loads(what_payload["content"])["body"][0]["item_name"] == "开头"
     assert {"key": "概要", "type": "str", "value": "脚本由开头中段结尾构成"} in what_payload["custom_ext"]
-    assert why_payload["dim_attributes"] == ["why原理"]
+    assert why_payload["dim_attributes"] == ["why"]
     assert why_payload["content"] == "冲突能制造期待,所以提升停留。"
     assert "score" not in why_payload["scopes"][0]
+
+
+def test_build_why_payload_accepts_explanation_alias():
+    payload = build_payload(
+        _post(),
+        {
+            "id": "y2",
+            "type": "why",
+            "title": "动作变化带来画面节奏",
+            "explanation": "动作变化能制造连续观看的期待。",
+            "业务阶段": ["脚本"],
+            "作用域": [{"scope_type": "effect", "value": "提升节奏感"}],
+        },
+    )
+
+    assert payload["content"] == "动作变化能制造连续观看的期待。"
+
+
+def test_validate_ingest_payload_blocks_external_api_contract_violations():
+    payload = build_payload(
+        _post(),
+        {
+            "id": "h1",
+            "type": "how",
+            "title": "先定受众再写脚本",
+            "purpose": "写出稳定脚本",
+            "业务阶段": ["脚本"],
+            "steps": [
+                {
+                    "input": "选题",
+                    "directive": "收窄受众",
+                    "output": "受众画像",
+                    "作用域": [{"scope_type": "intent", "value": "吸引注意"}],
+                }
+            ],
+        },
+    )
+
+    bad = {**payload, "content": ""}
+    with pytest.raises(IngestPayloadValidationError, match="content"):
+        validate_ingest_payload(bad)
+
+    bad = {**payload, "source": {**payload["source"], "id": "x" * 65}}
+    with pytest.raises(IngestPayloadValidationError, match="source.id"):
+        validate_ingest_payload(bad)
+
+    bad = {**payload, "title": "x" * 513}
+    with pytest.raises(IngestPayloadValidationError, match="title"):
+        validate_ingest_payload(bad)
+
+    bad = {**payload, "source": {**payload["source"], "author": "x" * 129}}
+    with pytest.raises(IngestPayloadValidationError, match="author"):
+        validate_ingest_payload(bad)
+
+    bad = {**payload, "scopes": [{"scope_type": "intent", "value": "x" * 129}]}
+    with pytest.raises(IngestPayloadValidationError, match="scopes"):
+        validate_ingest_payload(bad)
+
+    bad = {**payload, "custom_ext": [{"key": "推荐指数", "type": "bad", "value": "5"}]}
+    with pytest.raises(IngestPayloadValidationError, match="custom_ext"):
+        validate_ingest_payload(bad)
+
+
+def test_validate_ingest_payload_rejects_legacy_dim_attributes():
+    payload = build_payload(
+        _post(),
+        {
+            "id": "h1",
+            "type": "how",
+            "title": "先定受众再写脚本",
+            "purpose": "写出稳定脚本",
+            "业务阶段": ["脚本"],
+            "steps": [
+                {
+                    "input": "选题",
+                    "directive": "收窄受众",
+                    "output": "受众画像",
+                    "作用域": [{"scope_type": "intent", "value": "吸引注意"}],
+                }
+            ],
+        },
+    )
+    legacy_values = [
+        "how" + "\u5de5\u5e8f",
+        "what" + "\u6784\u6210",
+        "why" + "\u539f\u7406",
+    ]
+
+    for legacy_value in legacy_values:
+        bad = {**payload, "dim_attributes": [legacy_value]}
+        with pytest.raises(IngestPayloadValidationError, match="dim_attributes"):
+            validate_ingest_payload(bad)

+ 3 - 0
tests/test_decode_readers_service.py

@@ -35,8 +35,11 @@ def test_post_from_candidate_item_prefers_cdn_then_oss_then_source():
 
     assert post.id == "xiaohongshu_abc"
     assert post.content_id == "abc"
+    assert post.unique_key == "xhs:abc"
     assert post.content_mode == "image_post"
     assert post.image_urls == ["oss0", "cdn1"]
+    assert post.media_count == 2
+    assert post.cover_url == "oss0"
     assert [card.index for card in post.cards] == [1, 2]
     assert post.body_text == "正文"
     assert post.topic_list == ["脚本"]

+ 1 - 1
tests/test_decode_service.py

@@ -136,5 +136,5 @@ def test_decode_service_full_flow_with_fake_llm():
     assert repo.results[0].status == "decoded"
     assert repo.particles[0].particle_type == "how"
     assert repo.scopes[0].scope_type == "intent"
-    assert repo.payloads[0].payload["dim_attributes"] == ["how工序"]
+    assert repo.payloads[0].payload["dim_attributes"] == ["how"]
     assert repo.snapshots[0].contract_name == "创作知识提取-skill"

+ 135 - 0
tests/test_ingest_client.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from core.config import IngestApiConfig
+from decode_content.ingest import KnowledgeIngestClient, ingest_payload_draft
+from decode_content.models import IngestRecord, PayloadDraft
+
+
+class FakeRepo:
+    def __init__(self):
+        self.marked = []
+        self.records = []
+
+    def mark_payload_draft_ingested(self, payload_draft_id):
+        self.marked.append(payload_draft_id)
+        return PayloadDraft(
+            id=payload_draft_id,
+            payload=_payload(),
+            review_status="approved",
+            ingest_ready=True,
+            status="ingested",
+        )
+
+    def save_ingest_record(self, **kwargs):
+        record = IngestRecord(id=uuid4(), **kwargs)
+        self.records.append(record)
+        return record
+
+
+class FakeResponse:
+    def __init__(self, status_code, payload):
+        self.status_code = status_code
+        self._payload = payload
+        self.text = str(payload)
+
+    def json(self):
+        return self._payload
+
+
+def _payload():
+    return {
+        "source": {"id": "xhs_1", "source_type": "post", "title": "帖子", "author": "作者"},
+        "title": "拍照姿势",
+        "content": "拍照时用手部动作制造自然感。",
+        "dim_attributes": ["how"],
+        "dim_creations": ["创作"],
+        "scopes": [{"scope_type": "intent", "value": "自然感"}],
+        "custom_ext": [{"key": "业务阶段", "type": "str", "value": "脚本"}],
+    }
+
+
+def test_dry_run_ingest_validates_and_does_not_call_http():
+    repo = FakeRepo()
+    draft_id = uuid4()
+    draft = PayloadDraft(id=draft_id, payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
+
+    record = ingest_payload_draft(repo, draft, dry_run=True)
+
+    assert repo.marked == [draft_id]
+    assert record.target_system == "dry-run"
+    assert record.status == "ingested"
+    assert record.response_payload["dry_run"] is True
+
+
+def test_real_ingest_posts_and_records_knowledge_id():
+    calls = []
+
+    def fake_post(url, **kwargs):
+        calls.append((url, kwargs))
+        return FakeResponse(201, {"knowledge_id": "k-1", "source_id": "xhs_1"})
+
+    repo = FakeRepo()
+    draft_id = uuid4()
+    draft = PayloadDraft(id=draft_id, payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
+    client = KnowledgeIngestClient(
+        IngestApiConfig(url="https://ingest.test/api/v1/knowledge/ingest", retry_delays_seconds=()),
+        post_fn=fake_post,
+        sleep_fn=lambda _delay: None,
+    )
+
+    record = ingest_payload_draft(repo, draft, dry_run=False, client=client)
+
+    assert calls and calls[0][0] == "https://ingest.test/api/v1/knowledge/ingest"
+    assert record.target_system == "knowledge-ingest-api"
+    assert record.target_id == "k-1"
+    assert record.response_payload["source_id"] == "xhs_1"
+    assert repo.marked == [draft_id]
+
+
+def test_real_ingest_failure_is_recorded_without_marking_draft():
+    def fake_post(url, **kwargs):
+        return FakeResponse(502, {"detail": "embedding down"})
+
+    repo = FakeRepo()
+    draft = PayloadDraft(id=uuid4(), payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
+    client = KnowledgeIngestClient(
+        IngestApiConfig(url="https://ingest.test/api/v1/knowledge/ingest", retry_delays_seconds=()),
+        post_fn=fake_post,
+        sleep_fn=lambda _delay: None,
+    )
+
+    record = ingest_payload_draft(repo, draft, dry_run=False, client=client)
+
+    assert record.status == "failed"
+    assert record.target_system == "knowledge-ingest-api"
+    assert record.response_payload["http_status"] == 502
+    assert repo.marked == []
+
+
+def test_real_ingest_retries_before_success():
+    calls = []
+    sleeps = []
+
+    def fake_post(url, **kwargs):
+        calls.append(url)
+        if len(calls) == 1:
+            return FakeResponse(502, {"detail": "embedding down"})
+        return FakeResponse(201, {"knowledge_id": "k-2", "source_id": "xhs_1"})
+
+    repo = FakeRepo()
+    draft = PayloadDraft(id=uuid4(), payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
+    client = KnowledgeIngestClient(
+        IngestApiConfig(url="https://ingest.test/api/v1/knowledge/ingest", retry_delays_seconds=(0.01,)),
+        post_fn=fake_post,
+        sleep_fn=lambda delay: sleeps.append(delay),
+    )
+
+    record = ingest_payload_draft(repo, draft, dry_run=False, client=client)
+
+    assert len(calls) == 2
+    assert sleeps == [0.01]
+    assert record.status == "ingested"
+    assert record.attempt_count == 2
+    assert record.target_id == "k-2"

+ 3 - 3
创作知识提取-skill/examples/金标样例.md

@@ -53,7 +53,7 @@
   "title": "撕裂共识选题框架",
   "content": "目标:把一个赛道,按「列共识→定失灵场景→撕裂裂缝写选题→羞耻感验收」做成一个能引爆传播的选题\n步骤1(目的:分析,锁定赛道里被反复重复的正确共识)\n  指引:列出本赛道被反复说、大家默认正确的共识。例:『护肤赛道——早C晚A、含酒精的就是不好』\n  产出:赛道共识清单\n步骤2(目的:构思,找这个共识在哪些场景会失灵)\n  指引:对每条共识,找它不成立/反例的具体场景,越具体越好\n  产出:失灵场景\n步骤3(目的:成文,在裂缝处写出撕裂共识的选题)\n  指引:把失灵场景写成一个挑战默认共识的选题句,制造认知冲突\n  产出:撕裂式选题\n步骤4(目的:打磨,用羞耻感验收选题够不够冒犯)\n  指引:自检这个选题会不会让坚持旧共识的人感到被冒犯/羞耻;不会就再撕深一点\n  产出:定稿选题",
   "dim_creations": ["创作"],
-  "dim_attributes": ["how工序"],
+  "dim_attributes": ["how"],
   "scopes": [
     { "scope_type": "substance", "value": "赛道共识" },
     { "scope_type": "effect", "value": "撕裂共识" },
@@ -97,7 +97,7 @@
     ]
   },
   "dim_creations": ["创作"],
-  "dim_attributes": ["what构成"],
+  "dim_attributes": ["what"],
   "scopes": [ { "scope_type": "substance", "value": "脚本要素" },
               { "scope_type": "form", "value": "分镜骨架" } ],
   "custom_ext": [ { "key": "业务阶段", "type": "str", "value": "脚本" },
@@ -117,7 +117,7 @@
   "title": "标题决定打开,与内容质量无关",
   "content": "标题的唯一作用是让人产生点开的欲望;点不点开和内容好不好没关系——因为信息流里用户先看到的只有标题,内容再好、没人点开也等于零(平台分发逻辑)。",
   "dim_creations": ["创作"],
-  "dim_attributes": ["why原理"],
+  "dim_attributes": ["why"],
   "scopes": [ { "scope_type": "effect", "value": "吸引点击" },
               { "scope_type": "intent", "value": "提升打开率" } ],
   "custom_ext": [ { "key": "业务阶段", "type": "str", "value": "选题" } ]

+ 1 - 1
创作知识提取-skill/extraction/phase3-assemble.md

@@ -9,7 +9,7 @@
 | `source` | 照抄 framework.json 的 `source`(三类一样) | | |
 | `title` | 该颗 `title` | | |
 | `dim_creations` | `["创作"]` 写死 | | |
-| `dim_attributes` | `["how工序"]` | `["what构成"]` | `["why原理"]` |
+| `dim_attributes` | `["how"]` | `["what"]` | `["why"]` |
 | `content` | 目标+步骤(目的/怎么做/产出)拍平(str) | **结构化对象** `{name,kind,维度拆分规则,body[]}`(概要入 custom_ext) | `阐述` 一段(str:是什么+为什么,忠实整合) |
 | `scopes` | **各步 `作用域` value 去重并集** | **颗级 `作用域`** value | 颗级 value |
 | `custom_ext` | 业务阶段 + 创作阶段(各步去重) + 动作(各步) | **业务阶段 + 概要** | **仅业务阶段** |

+ 6 - 0
创作知识提取-skill/format/ingest-payload-how.schema.json

@@ -22,6 +22,12 @@
           "properties": {
             "platform": { "type": "string" },
             "url": { "type": "string" },
+            "unique_key": { "type": "string" },
+            "platform_item_id": { "type": "string" },
+            "content_type": { "type": "string" },
+            "content_mode": { "type": "string" },
+            "media_count": { "type": "integer" },
+            "cover_url": { "type": "string" },
             "date": { "type": "string" }
           }
         }

+ 6 - 0
创作知识提取-skill/format/ingest-payload-what.schema.json

@@ -22,6 +22,12 @@
           "properties": {
             "platform": { "type": "string" },
             "url": { "type": "string" },
+            "unique_key": { "type": "string" },
+            "platform_item_id": { "type": "string" },
+            "content_type": { "type": "string" },
+            "content_mode": { "type": "string" },
+            "media_count": { "type": "integer" },
+            "cover_url": { "type": "string" },
             "date": { "type": "string" }
           }
         }

+ 7 - 1
创作知识提取-skill/format/ingest-payload-why.schema.json

@@ -22,6 +22,12 @@
           "properties": {
             "platform": { "type": "string" },
             "url": { "type": "string" },
+            "unique_key": { "type": "string" },
+            "platform_item_id": { "type": "string" },
+            "content_type": { "type": "string" },
+            "content_mode": { "type": "string" },
+            "media_count": { "type": "integer" },
+            "cover_url": { "type": "string" },
             "date": { "type": "string" }
           }
         }
@@ -42,7 +48,7 @@
     },
     "dim_attributes": {
       "type": "array",
-      "items": { "const": "why原理" },
+      "items": { "const": "why" },
       "minItems": 1,
       "maxItems": 1
     },

+ 3 - 3
创作知识提取-skill/taxonomy/知识类型.json

@@ -7,21 +7,21 @@
   "最终分类树": [
     {
       "分类名称": "how",
-      "dim_attributes值": "how工序",
+      "dim_attributes值": "how",
       "payload形态": "完整(目标 purpose + 有序 steps,每步 目的/指引/产出)",
       "分类说明": "这颗知识是『怎么做』——一条不可跳步的创作工序:把某种输入一路做成一个可交付成品。典型:选题框架、脚本结构搭建法、分镜创作流程。判别口诀:读完主要收获是『知道一步步怎么搭出这个东西』,有明确先后顺序。与 what 边界:how 是过程/步骤,what 是界定/列举。与 why 边界:how 是执行层(具体怎么做),why 是原理层(为什么这样做)。绝大多数创作教学帖都能归到一个 how 框架——优先按 how 提。",
       "分类性质": "工序"
     },
     {
       "分类名称": "what",
-      "dim_attributes值": "what构成",
+      "dim_attributes值": "what",
       "payload形态": "结构化(content 为对象:name + kind + 维度拆分规则(子集) + body[item_name/item_desc];概要入 custom_ext;无 steps)",
       "分类说明": "这颗知识是『是什么 / 由什么构成』——对某创作对象的界定、要素清单、模板、风格示例,但不讲怎么一步步做。典型:脚本三要素、爆款标题的构成要素、某风格的特征清单。判别口诀:读完主要收获是『了解这东西是什么、由哪几块组成』,是界定性/列举性的。注意:what 始终是独立一颗——定死两态:要么单独的一条(role=主),要么对应某 how 的某一步而成组件颗(role=组件,parent={how_id,step})。二者皆可,不存在『并进 step、不单独成颗』;与 why 完全对称。",
       "分类性质": "构成"
     },
     {
       "分类名称": "why",
-      "dim_attributes值": "why原理",
+      "dim_attributes值": "why",
       "payload形态": "简化(content 为一段「阐述」:这条原理是什么+为什么,忠实整合原帖;无 steps;不延伸\"怎么用\")",
       "分类说明": "这颗知识是『为什么』——解释某种创作做法/判断背后的原理、依据、平台逻辑、读者心理。典型:为什么标题决定打开率、为什么先写冲突、为什么这种结构传播好。判别口诀:读完主要收获是『理解了某件事为什么这样做』,是解释性/原理性的,较少具体步骤。注意:why 始终是独立一颗——定死两态:要么单独的一条(role=主),要么对应某 how 的某一步而成组件颗(role=组件,parent={how_id,step})。二者皆可,不存在『写进步骤指引、不单独成颗』;与 what 完全对称。",
       "分类性质": "原理"

+ 4 - 4
创作知识提取-skill/tools/lint-payload.py

@@ -14,6 +14,7 @@ from pathlib import Path
 SCOPE_TYPES = {"substance", "form", "feeling", "effect", "intent"}
 EXT_KEYS = {"业务阶段", "创作阶段", "动作", "出自", "概要"}
 KINDS = {"子集", "多维关系", "序列"}
+DIM_ATTRIBUTES = {"how", "what", "why"}
 业务阶段 = {"灵感", "选题", "脚本"}
 创作阶段 = {"定向", "构思", "结构", "成文", "打磨"}
 
@@ -36,14 +37,13 @@ def lint_one(p: dict, idx: int) -> tuple[list[str], list[str]]:
     if p.get("dim_creations") != ["创作"]:
         err.append(f"{tag} dim_creations 必须恒为 [\"创作\"],实为 {p.get('dim_creations')}")
 
-    # dim_attributes 单值 + 类型前缀
+    # dim_attributes 单值 + 精确类型
     da = p.get("dim_attributes") or []
     if len(da) != 1:
         err.append(f"{tag} dim_attributes 应恰好 1 个,实为 {da}")
-    ktype = (da[0][:3] if da else "")  # how / wha / why
-    kind = {"how": "how", "wha": "what", "why": "why"}.get(ktype)
+    kind = da[0] if da and da[0] in DIM_ATTRIBUTES else None
     if kind is None:
-        err.append(f"{tag} dim_attributes[0] 须以 how/what/why 开头,实为 {da}")
+        err.append(f"{tag} dim_attributes[0] 只能是 how/what/why,实为 {da}")
 
     # scopes
     for s in p.get("scopes") or []:

+ 5 - 5
开发文档/创作知识拆解框架.md

@@ -20,9 +20,9 @@
 
 | 类型 | 它是什么 | 成形模板 | dim_attributes |
 |---|---|---|---|
-| **How** 工序 | 不可跳步地把某输入做成成品 | `目标(purpose)` + 有序 `steps`(每步 目的 / 怎么做 / 产出) | `how工序` |
-| **What** 构成 | 界定 / 列举 / 分类 / 判据某创作对象 | `概要`(必填锚点) + `kind`(子集/多维关系/序列)+ `维度拆分规则`(仅子集)+ `body[{item_name, item_desc}]`(**扁平条目**,无小节/形式) | `what构成` |
-| **Why** 原理 | 解释某做法 / 判断背后的道理 | `阐述`(必填,**一段**:这条原理是什么 + 为什么成立;**忠实整合**原帖、无小节、不延伸"怎么用") | `why原理` |
+| **How** 工序 | 不可跳步地把某输入做成成品 | `目标(purpose)` + 有序 `steps`(每步 目的 / 怎么做 / 产出) | `how` |
+| **What** 构成 | 界定 / 列举 / 分类 / 判据某创作对象 | `概要`(必填锚点) + `kind`(子集/多维关系/序列)+ `维度拆分规则`(仅子集)+ `body[{item_name, item_desc}]`(**扁平条目**,无小节/形式) | `what` |
+| **Why** 原理 | 解释某做法 / 判断背后的道理 | `阐述`(必填,**一段**:这条原理是什么 + 为什么成立;**忠实整合**原帖、无小节、不延伸"怎么用") | `why` |
 
 > **What 的 `kind`(部件之间什么关系)**:`子集`=并列可挑的选项(挑一个或多个);`多维关系`=并列共存的组成部分/维度/层级(含"大套小"的包含层级,彼此无时间先后,不挑);`序列`=沿作品从头到尾的时间/线性先后(换序读不通)。
 > **`维度拆分规则`(仅子集)**:先试**单一轴**("按【X】分,每项都是 X 的取值",命名轴非结果、不拼两个);套不出单一轴就填 **`类型枚举`**。
@@ -86,7 +86,7 @@
 
 | 维度 | How | What | Why | 落到 ingest | 谁定 |
 |---|:--:|:--:|:--:|---|---|
-| 知识类型 | ✓ | ✓ | ✓ | `dim_attributes`(how工序/what构成/why原理) | 下游系统 |
+| 知识类型 | ✓ | ✓ | ✓ | `dim_attributes`(how/what/why) | 下游系统 |
 | 业务阶段(灵感/选题/脚本,颗级多值) | ✓ | ✓ | ✓ | `custom_ext` | 我们 |
 | 创作阶段(定向/构思/结构/成文/打磨,**逐步**) | ✓ | ✗ | ✗ | `custom_ext` | 我们 |
 | 动作(开放手法,**逐步**,从内容提炼) | ✓ | ✗ | ✗ | `custom_ext` | 我们 |
@@ -130,7 +130,7 @@
 | payload 字段 | How | What | Why |
 |---|---|---|---|
 | `content` | 目标 + 步骤(目的/怎么做/产出) 拍平(字符串) | **结构化对象** `{name, kind, 维度拆分规则, body[]}`(概要移入 custom_ext) | `阐述` 一段(字符串:是什么+为什么,忠实整合) |
-| `dim_attributes` | `["how工序"]` | `["what构成"]` | `["why原理"]` |
+| `dim_attributes` | `["how"]` | `["what"]` | `["why"]` |
 | `scopes` | 各步作用域并集 | 颗级作用域 | 颗级作用域 |
 | `custom_ext` | 业务阶段 + 创作阶段 + 动作 | **业务阶段 + 概要** | **仅业务阶段** |
 | `source / title / dim_creations` | 三类一样(`dim_creations` 恒 `["创作"]`;组件颗在 custom_ext 或 content 里标明出自哪个 how 第几步) | | |

+ 1 - 1
开发文档/技术文档.md

@@ -249,7 +249,7 @@ def extract_video(post, *, ..., oss_video_url: Optional[str] = None) -> Extracte
 
 | 问题 | 现象 | 处置 |
 |---|---|---|
-| **dim_attributes 不一致** | 代码发 `["why"]`([`TYPE2ATTR` decompose.py:60](../scripts/decompose.py)),但 `format/ingest-payload-why.schema.json` 要 `const:"why原理"`(how/what 同类:`how工序`/`what构成`) | 本迭代**不改**,待入库 ingest 时统一对齐(确定以哪边为准) |
+| **dim_attributes 不一致** | 代码发 `["why"]`([`TYPE2ATTR` decompose.py:60](../scripts/decompose.py)),但 `format/ingest-payload-why.schema.json` 要 `const:"why"`(how/what 同类:`how`/`what`) | 本迭代**不改**,待入库 ingest 时统一对齐(确定以哪边为准) |
 | **抖音搜索鉴权** | aiddit `POST /crawler/dou_yin/keyword` 实测 `code:10000`(强限流) | **已解决**:抖音改走 piaoquantv 后端(`crawapi.piaoquantv.com`,body 带 `account_id=7450041106378522636` + `cookie_batch=default`),search/detail 实测 `code:0`,拿到视频直链。env:`CK_DOUYIN_BASE_URL/ACCOUNT_ID/COOKIE_BATCH` |
 | **OSS 串行延迟** | 每张图一次 OSS POST(60s timeout),多图帖串行慢 | 暂接受;后续可并行/限流 |
 | **B站视频** | `video_extract` 对 B站抛错(需 voice_data 音轨 + ffmpeg 合并) | 与本迭代无关,沿用现状 |

+ 1 - 1
开发文档/正式版施工文档.md

@@ -743,7 +743,7 @@ rg "web/frameworks|web/payloads|creation_knowledge.integrations|rebuild_payloads
 - 新增 `decode_content/contracts.py`,集中加载 `创作知识提取-skill` 的 README、phase、schema、taxonomy、gate/normalize prompts 和 `scope_trees` cache 元信息,生成合同 hash 与 `ContractSnapshot`。
 - 新增 `decode_content/readers/`,图文、整段视频、抽帧兜底 reader 已迁入正式包;正式 reader 入口支持 `CandidateItem + MediaAsset[] -> ReadResult`,不再依赖旧 `src["from"]`、本地 `/data/demo` 或 web JSON。
 - 新增 `decode_content/gates.py`、`framing.py`、`scoping.py`、`payloads.py`、`service.py`,把创作闸、How/Why 质量闸、What/How/Why 拆颗、作用域定位、payload 组装和单帖 decode 编排拆到正式模块。
-- `decode_content/payloads.py` 已统一正式 `dim_attributes`:`how工序 / what构成 / why原理`;What/Why content 按正式 ingest 边界输出字符串。
+- `decode_content/payloads.py` 已统一正式 `dim_attributes`:`how / what / why`;What/Why content 按正式 ingest 边界输出字符串。
 - `DecodeService` 每次 decode 会写合同快照;空内容或创作闸未通过会短路,只保存 read/gate/decode 状态,不生成 particles/payload drafts。
 - 新增 `core/media_download.py`,把原来散在旧视频 reader 里的下载字节能力上提为共享工具;正式 acquisition/decode 不再 import `creation_knowledge.integrations`。
 - 新增 `scripts/run_decode_content.py` 作为薄 CLI;正式 DB item 加载留到 Step6 pipeline/API 继续接。

Kaikkia tiedostoja ei voida näyttää, sillä liian monta tiedostoa muuttui tässä diffissä