# 创作知识解构 · 技术文档

> 配套：本目录《产品文档.md》《创作知识拆解框架.md》。
> 本文 = **A 现有系统架构** + **B 下一迭代详细开发步骤** + **C 测试核验** + **D 实施顺序** + **E 已知问题**。
> 所有 `文件:行号` 可点开核对（行号对应当前 main，实现时以实际为准）。
> 2026-06-30 归档说明：本文描述的是旧本地 demo / 单帖解构实现，已经归入正式版改造的历史说明；正式入口与施工步骤以后看《正式版开发文档.md》《正式版施工文档.md》。

---

# A. 旧本地 Demo 系统架构

## A.1 总览

一条 Python 编排流水线，把帖子拆成知识颗并组装 ingest payload。入口 `scripts/decompose.py`，方法论真源是 `创作知识提取-skill/`（phase 文档直接当 LLM system prompt 喂），旧产物给归档静态页渲染。

```
① 读懂 → ①.5 创作闸 → ② 成形 → ②.5 假how根治 → ②.7 why轻闸 → ③ 作用域 → ⑤ 组装
```

主循环：[`scripts/decompose.py:436`](../scripts/decompose.py) `main()`。每帖逐阶段跑，任一总闸不过则短路跳过。

## A.2 流水线各阶段

| 阶段 | 函数 / 位置 | 干什么 | 用的 prompt |
|---|---|---|---|
| ① 读懂 | `read_one` [decompose.py:98](../scripts/decompose.py) | 图文→`GeminiExtractor` 读图；视频→`video_extract` 下 mp4+base64 喂 Gemini | `extract.txt` / `extract_video.txt` |
| ① 总闸 | [decompose.py:453](../scripts/decompose.py) | `is_empty`（纯展示/无知识）→ 短路 | — |
| ①.5 创作闸 | `creation_gate` [decompose.py:142](../scripts/decompose.py) | admit+refute 两票，分歧 tiebreak，边界倾向排除 | `gate_admit/refute/tiebreak.txt` |
| ② 成形 | `shape` [decompose.py:126](../scripts/decompose.py) | 一次出全部颗+类型+轻标签，作用域留空 | `phase1-frame.md` |
| ②.5 假how | `fix_fake_hows` [decompose.py:216](../scripts/decompose.py) | 代码信号+对抗投票；假how→`reshape_nonhow` 重拆 what/why | `gate_how_admit/refute/tiebreak.txt` |
| ②.7 why闸 | `drop_fake_whys` [decompose.py:232](../scripts/decompose.py) | 废话 drop；疑似错分类标 `inferred` 保留 | `gate_why_refute.txt` |
| 守卫 | [decompose.py:466](../scripts/decompose.py) | 业务阶段/创作阶段/kind 受控值归一、组件 parent 校验 | — |
| ③ 作用域 | `scope_candidates`→`nounify_scopes`→`apply_scopes` [decompose.py:270](../scripts/decompose.py) | 5 类带值候选 → 名词化 → 连接词拆原子 → 向量定位 | `phase2-scope.md` / `normalize_scope.txt` |
| ⑤ 组装 | `build_payload` [decompose.py:398](../scripts/decompose.py) | content 按类型分拼、scopes 并集、custom_ext | — |

**关键常量**（[decompose.py:48-64](../scripts/decompose.py)）：`SOURCES`（写死 9 帖入口）、`CSTAGE`（创作阶段 5 值）、`KINDS`（what.kind 3 值）、`REUSE_THRESHOLD=0.90`、`OUT` env 后缀（`OUT=_v2`→写 `frameworks_v2.json` 不覆盖）。

## A.3 关键模块

| 模块 | 职责 | 要点 |
|---|---|---|
| `integrations/crawler.py` | 取数 | `PLATFORMS` 路由 detail path（[:24](../creation_knowledge/integrations/crawler.py)）；`fetch_post_detail(content_id)`（[:153](../creation_knowledge/integrations/crawler.py)）；`parse_detail_response` 纯函数可离线测；`RateLimiter` 15s/bucket（[:64](../creation_knowledge/integrations/crawler.py)） |
| `integrations/extractor.py` | 读图 | `GeminiExtractor`，**图片直传 `image_url`**（[:105](../creation_knowledge/integrations/extractor.py)），走 OpenRouter |
| `integrations/video_extract.py` | 读视频 | `extract_video` 下载 mp4 → **base64 data_url** 喂模型（[:128](../creation_knowledge/integrations/video_extract.py)）；签名 已有 `save_path/public_url`（[:76](../creation_knowledge/integrations/video_extract.py)）；`_default_download` 带 iOS UA + 平台 Referer 躲防盗链 |
| `integrations/llm.py` | 文本 LLM | `chat_json(system,user)` 强约束 JSON，**3 轮重试 + self-repair**（[:46](../creation_knowledge/integrations/llm.py)），注入式 `http_post` 可离线测 |
| `embedding.py` | 向量 | `embed_text` 走火山方舟 Doubao multimodal，2048 维 |
| `scripts/scope_link.py` | 作用域定位 | `ScopeLinker.link(value, source_type, top_k)` numpy 暴力余弦最近邻；树缓存 `scope_trees/`（trees_index.json + trees_embeddings.npy，**1855 节点**：实质972/形式624/作用185/感受46/意图28） |
| `config.py` | 配置 | `Settings.from_env`（[:93](../creation_knowledge/config.py)）；`crawler_base_url` 默认 `http://crawler.aiddit.com` |

## A.4 数据契约

### frameworks.json（旧前端主数据，[archive/2026-06-30-step1/legacy_web/frameworks.json](../archive/2026-06-30-step1/legacy_web/frameworks.json)）
```
{ count, posts:[ {
    post_id, source_id, title, platform, url,
    media:{ type:"image"|"video", video_url, images:[] },
    cards:[ {index, content, image_url|video_url, start, end} ],
    knowledges:[ {
      id, type:"how"|"what"|"why", role:"主"|"组件", parent:null|{how_id,step}, title, 业务阶段:[],
      // how:  purpose, steps:[{id,input,directive,output,出处,创作阶段,动作,作用域:[]}]
      // what: 概要, kind, 维度拆分规则, body:[{item_name,item_desc,作用域:[]}]
      // why:  阐述
      作用域:[ {scope_type,value,candidate,link:"复用"|"新建",score,top:[{name,score,path}]} ]
    } ]
} ] }
```

### payloads.json（旧入库体，[archive/2026-06-30-step1/legacy_web/payloads.json](../archive/2026-06-30-step1/legacy_web/payloads.json)）
```
[ { source:{id,source_type:"post",title,author,source_metadata:{platform,url}},
    title, content, dim_creations:["创作"], dim_attributes:["how"|"what"|"why"],
    scopes:[{scope_type,value}], custom_ext:[{key,type:"str",value}] } ]
```
- `content`：How=拍平字符串；What=`{name,kind,维度拆分规则,body[]}`；Why=`{name,desc}`（[`build_content` decompose.py:362](../scripts/decompose.py)）。

### Ingest schema（最终裁判）
`创作知识提取-skill/format/ingest-payload-{how,what,why}.schema.json` —— 字段约束与受控词表的权威定义。

## A.5 旧前端数据绑定（[archive/2026-06-30-step1/legacy_web/index.html](../archive/2026-06-30-step1/legacy_web/index.html)）

- 加载：旧 `index.html:362` 使用 `Promise.all([j("/frameworks.json"), j("/payloads.json")])`，`cache:"no-store"`。
- 渲染：HowTable（工序表）/ WhatCard（构成卡）/ WhyCard（阐述卡）；组件颗 cross-ref 跳转；作用域 drawer（候选→top-K→复用/新建）；提示词溯源（点开产出该阶段的 prompt）。
- payload 按 `source.id` 匹配挂到帖子（旧静态页对应逻辑）。
- `web/prompt-read`→`../prompts`、`web/prompt-skill`→`../创作知识提取-skill/extraction` 是**符号链接**（前端静态读 prompt）。

## A.6 运行环境

- Python 3.12，**uv** 管理（`.venv` / `uv.lock`）。跑：`PYTHONPATH=. uv run python scripts/decompose.py`；测：`uv run pytest`。
- 须在能访问内网 `crawler.aiddit.com` 的环境（云端 / VPN）跑取数。
- 测试全离线：`pytest` 用 fixture + 注入假 `http_post`，`testpaths=["tests"]`、`pythonpath=["."]`（pyproject.toml）。

---

# B. 下一迭代 · 详细开发步骤

> 历史说明：本章是旧 demo 阶段的下一迭代计划，不再作为正式版施工计划。正式施工以《正式版施工文档.md》为准。

**目标**：`query 搜索 → OSS 转存 → 模型直读 CDN → run 隔离产出`。**仅小红书**。已实测：小红书搜索 `code:0`、OSS `status:0` 回 `cdn_url`、Gemini 直读 OSS http 视频链 HTTP 200。新 env 已写入 `.env`（`CRAWLER_OSS_UPLOAD_URL` 等 5 个）。

## B.0 实测确认的接口契约

**搜索**（host = `settings.crawler_base_url`）：
```
POST /crawler/xiao_hong_shu/keyword
body { keyword, content_type:"图文"|"视频", sort_type:"综合", publish_time:"", cursor:"" }
→ { code:0, data:{ has_more, next_cursor, data:[ {id:<content_id>, note_card:{...预览...}} ] } }
```
content_id = `data.data[].id`；翻页 = `has_more` / `next_cursor`；搜索只给预览，**全文/全图/视频仍需 `fetch_post_detail` 逐 id 取详情**。

**OSS 转存**：
```
POST http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream
body { src_url, src_type:"video"|"image" }
→ { status:0, oss_object:{ cdn_url:"https://res.cybertogether.net/...", oss_object_key } }
```
成功 = `status==0`，取 `oss_object.cdn_url`。

## B.1 新增 2 模块（`creation_knowledge/integrations/`）

风格对齐 crawler：**纯 parser 函数（离线可测）+ 注入式 http**。

### `search.py`
```python
SEARCH_PATHS = {
    "xiaohongshu": "/crawler/xiao_hong_shu/keyword",   # host = settings.crawler_base_url (aiddit)
    "weixin":      "/crawler/wei_xin/keyword",          # 微信公众号，搜索回包即带封面，无需 detail
    "douyin":      "/crawler/dou_yin/keyword",          # 已解决 code 10000：抖音改走 piaoquantv 后端
}
# 抖音独立后端：host=settings.douyin_base_url(crawapi.piaoquantv.com)，body 必带 account_id + cookie_batch；
# keyword 与 detail 都走此 host。小红书/微信仍走 aiddit（settings.crawler_base_url）。
class SearchError(RuntimeError): ...

def parse_search_response(response: dict, *, platform="xiaohongshu") -> list[dict]:
    """纯函数。信封 {code,msg,data:{has_more,next_cursor,data:[...]}}。
    取 data.data[].id（+note_card 预览）；code!=0 抛 SearchError。"""

def search_keyword(keyword, *, platform="xiaohongshu", content_type="图文",
                   sort_type="综合", limit=5, publish_time="",
                   settings=None, http_client=None, rate_limiter=None, env_file=".env") -> list[str]:
    """搜索 → content_id 列表（翻页直到够 limit 或 has_more=False；内部按 id 去重、截断 limit）。
    用 settings.crawler_base_url + SEARCH_PATHS[platform]；复用 crawler.RateLimiter('{platform}_keyword')。"""
```

### `oss.py`
```python
class OssError(RuntimeError): ...

def parse_oss_response(response: dict) -> str:
    """纯函数。{status:0,oss_object:{cdn_url}} → cdn_url；status!=0 或缺字段抛 OssError。"""

def upload_stream(src_url, *, src_type, settings=None, http_post=httpx.post, env_file=".env") -> str:
    """POST settings.oss_upload_url body{src_url,src_type}；timeout=settings.oss_upload_timeout_seconds。"""

def to_oss(src_url, src_type, *, settings, http_post=httpx.post) -> str:
    """兜底包装：成功返回 cdn_url，任何失败返回原 src_url（永不抛）——decompose 用它，OSS 挂了降级不阻断。"""
```

## B.2 改 4 处

### ① `config.py` — Settings 加 5 字段（**必须带默认值**）
```python
# dataclass 字段（加在 data_dir 后）
oss_upload_url: str = "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream"
oss_upload_timeout_seconds: int = 60
search_default_limit: int = 5
search_content_type: str = "图文"
search_sort_type: str = "综合"
# from_env 加读取
oss_upload_url=env_value("CRAWLER_OSS_UPLOAD_URL", file_env, "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream"),
oss_upload_timeout_seconds=int(env_value("CRAWLER_OSS_UPLOAD_TIMEOUT_SECONDS", file_env, "60")),
search_default_limit=int(env_value("CREATION_SEARCH_DEFAULT_LIMIT", file_env, "5")),
search_content_type=env_value("CREATION_SEARCH_CONTENT_TYPE", file_env, "图文"),
search_sort_type=env_value("CREATION_SEARCH_SORT_TYPE", file_env, "综合"),
```
> ⚠️ **默认值是硬约束**：[`tests/test_video_extract.py:27`](../tests/test_video_extract.py) 用关键字逐字段构造 `Settings(...)` 且不传新字段，无默认会立刻断测。

### ② `video_extract.py` — `extract_video` 加 `oss_video_url`（增量、向后兼容）
```python
def extract_video(post, *, ..., oss_video_url: Optional[str] = None) -> ExtractedContent:
    if oss_video_url:
        media_part = {"type": "video_url", "video_url": {"url": oss_video_url}}   # 跳过下载+base64
        public_url = public_url or oss_video_url                                  # 段卡指向 OSS url
    else:
        ... 原 下载→save_path→base64 ...                                          # fixture/兜底不变
        media_part = {"type": "video_url", "video_url": {"url": data_url}}
```

### ③ `decompose.py` — 入口 + 媒体 + 产出
- **QUERIES 常量**（`SOURCES` 后，并存）：`QUERIES = [{"query":"分镜脚本","platform":"xiaohongshu","content_type":"图文","limit":5}]`。
- **抽 `process_one(src, settings, extractor, linker) -> (post_meta, payloads)`**：把 `main()` 循环体（[:445-481](../scripts/decompose.py)）原样抽出，**零行为变更**；先用 SOURCES 跑出与现 `frameworks.json` diff 为空验证 parity。
- **`expand_queries(QUERIES, settings)`**：逐 query `search_keyword` → ids；`seen` 集**跨 query 去重**；搜索失败打印并跳过该 query。产出 `[{cid,platform,from:"live",query}]`。
- **`read_one` 媒体改 OSS**（取数后**立即**转存躲时效）：
  - 图文（[:108-117](../scripts/decompose.py)）：live 帖每张 `to_oss(url,"image",settings=...)` 替掉 `_download_images`；cdn_url 写回 `post.image_urls` / `post.cards[*].url` **再喂** `extractor.extract`（extractor 直读 `card.url`）。fixture 仍走本地 `/data` 路径。
  - 视频（[:100-107](../scripts/decompose.py)）：`oss_url = to_oss(post.video_urls[0],"video",...)` → `extract_video(post,...,oss_video_url=oss_url)`；`media={"type":"video","video_url":oss_url,...}`。`to_oss` 失败返回原 url → 传 `oss_video_url=None` 回 base64 兜底。
  - 媒体类型：复用现有 `if post.video_urls:` 判定。
- **run 隔离产出**：`slugify(text)`（中英→`-`，去空）；写 `web/runs/<slug>.json`（同名存在则加 `-{timestamp}` 保留历史）+ 配套 `web/runs/<slug>.payloads.json`（前端按 `source.id` 配对，须同行）+ 追加 `web/runs/index.json` 清单 `[{slug,query,count,created_at}]`。**不覆盖 `web/frameworks.json`**。

### ④ 旧 `web/index.html` — run 选择器（历史计划）
- data-load `useEffect` 加：fetch `/runs/index.json` 填一个 `<select>`/pill 行（近 header）。
- 选中某 run → 加载 `/runs/<slug>.json` + `/runs/<slug>.payloads.json`；**默认不选仍加载 `/frameworks.json`**（9 帖样本不动）。
- 渲染树形状兼容（run posts 与 frameworks posts 同形），卡片组件零改。

## B.3 跨切面

| 关注点 | 处理 |
|---|---|
| 去重 | `search_keyword` 内按 id；`expand_queries` 的 `seen` 集跨 query |
| 时效 | 取数后**立即** OSS 转存（抖音/CDN 临时签名链） |
| 限流 | 复用 `RateLimiter` 15s（搜索 `{platform}_keyword`、详情 `{platform}_detail`）；OSS 用 `oss_upload_timeout_seconds`=60 |
| 兜底 | OSS 失败→原 url；搜索失败→跳过该 query；视频 OSS 失败→回 base64；详情失败→现有 error-post 路径不变 |

---

# C. 测试与核验

## C.1 必须保持绿（`uv run pytest`，全离线）
`test_crawler` / `test_extractor` / `test_video_extract` / `test_media` / `test_video_frames`。
保障：`Settings` 新字段带默认（护住 [test_video_extract.py:27](../tests/test_video_extract.py) 的 `Settings(...)`）+ `extract_video` 改动为增量（`oss_video_url=None` 默认走原路）。

## C.2 新增离线单测（沿用 `_Resp` + 注入式 http 模式）
- **`test_search.py`**：`parse_search_response` 取 id；`search_keyword` 去重+翻页+截断（假 client 返回重复 id + `has_more`）；`code 10000` → `SearchError`。新增 fixture `tests/fixtures/xhs_search_*.json`（裸 `{code,msg,data}`）。
- **`test_oss.py`**：`parse_oss_response` 取 cdn_url；`status!=0` → `OssError`；`to_oss` 注入异常 → 返回原 url。
- **`test_decompose_helpers.py`**：`slugify`（中英/防碰撞）；`expand_queries` 跨 query 去重（monkeypatch `search_keyword` 返回重叠 id）。
- **`test_video_extract.py` 补** `test_extract_video_oss_direct_feed`：传 `oss_video_url`，断言**未调用下载** + 请求体 `video_url.url` == 该 http url（用假 `http_post` 捕获）。

## C.3 端到端手测（真网，跑一次）
1. 真实小红书 `QUERIES` → `PYTHONPATH=. uv run python scripts/decompose.py`。
2. 确认 `web/runs/<slug>.json` + `.payloads.json` + `index.json` 生成；`media.images`/`media.video_url` 是 `res.cybertogether.net` cdn_url。
3. 起 `web/` 静态服务 → run 选择器列出新 run → 图/视频无 403 → 三类知识渲染如常。

---

# D. 实施顺序（每步独立可验证）

1. `config.py` 加字段（带默认）+ 读 env → `pytest` 全绿（证默认护住旧构造）。
2. `oss.py` + `test_oss.py`（独立）。
3. `search.py` + `test_search.py` + fixture（独立）。
4. `video_extract.py` 加 `oss_video_url` → 视频测试绿 + 新直喂测试。
5. `decompose.py` 抽 `process_one` 零行为重构 → SOURCES 跑出与现 `frameworks.json` diff 为空。
6. `decompose.py` 接入 QUERIES/expand/OSS/slug/run 写盘+manifest → helper 测 + 小红书 dry run。
7. `web/index.html` run 选择器 + manifest 加载，默认回退 frameworks.json。
8. 端到端手测。

> 步骤 2–4 互相独立可并行；每步可单独提交、单独测。

---

# E. 已知问题 / 暂缓项

| 问题 | 现象 | 处置 |
|---|---|---|
| **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 合并） | 与本迭代无关，沿用现状 |
