# 网站自动导入知识库 实施计划 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 按 spec(docs/superpowers/specs/2026-10-20-site-import-design.md)实现「给官网 URL → 抓取 → AI 生成全套行业化草稿 → 管理员审核保存」的自助租户接入功能。 **Architecture:** 后端 `app/platform/site_import/` 三件套(校验 client 的抓取器 / 模板+两段 AI 的生成器 / 内存任务表)+ 2 个 API(POST 启动 / GET 轮询)+ admin 审核页(草稿可改、保存走既有 PUT 版本管线零新写路径)。行为骨架模板化(红线零 AI 参与),AI 只产行业内容与品牌文案。 **Tech Stack:** Python 3.13 / FastAPI 单进程(--workers 1 铁律)/ httpx 0.28 + httpcore 1.0.9(连接层 SSRF 校验用私有 network_backend——版本探针测试钉住)/ zhipuai GLMProvider(max_tokens 透传扩展)/ Vue3+TS admin(model.ts 纯函数 + vitest)。 **基线(2026-10-20 实跑确认):pytest 672 passed + 2 skipped;vitest 188 passed。新增只增不减。** **执行环境注意(Windows):** - 后端:`PYTHONUTF8=1 .venv/Scripts/python -m pytest ...`;前端:`cd admin && pnpm test` / `pnpm build` - **实施首步(Task 1)**:核查配置模型 max_tokens 上限(spec §3.2——resolver 默认 glm-4-flash 旧接口封顶 4095,用 env ZHIPU_API_KEY 直接打一次 `chat.completions.create(max_tokens=8192)` 看是否报错;报错则本计划内 MAX_TOKENS_TARGET 取 4095)——结果写进 **Task 5** 的常量 - dts 行尾纪律照旧(`git diff --ignore-cr-at-eol` 为空才 checkout 还原) - 分支:`feature/site-import`(已建) --- ## Batch A:GLMProvider 扩展 + SSRF 校验 client ### Task 1: GLMProvider max_tokens 透传(TDD) **Files:** - Modify: `app/engine/providers/glm.py`(`__init__` 加可选参数 + `_call` 透传) - Test: `tests/test_glm_provider.py`(新建) - [ ] **Step 0: 模型上限核查**(spec 要求的实施首步)——写一次性脚本打真 API: ```python # PYTHONUTF8=1 .venv/Scripts/python -c 方式,读 .env 的 ZHIPU_API_KEY # 尝试 max_tokens=8192 一次调用 "say ok";报 InvalidParameter/超限错 → 上限 4095 ``` 结论(8192 可用 / 用 4095)记入本任务 commit message 与 **Task 5** 的 MAX_TOKENS_TARGET 注释。 - [ ] **Step 1: 失败测试**——tests/test_glm_provider.py: ```python """GLMProvider max_tokens 透传(site-import spec §3.2)。""" from unittest.mock import MagicMock, patch from app.engine.providers.glm import GLMProvider def _provider(**kw): with patch("app.engine.providers.glm.ZhipuAI"): return GLMProvider(api_key="k", **kw) class TestMaxTokensPassthrough: def test_默认不传_调用无max_tokens参数(self): p = _provider() with patch.object(p, "client") as cli: cli.chat.completions.create.return_value.choices = [MagicMock(message=MagicMock(content="ok"))] import asyncio asyncio.run(p.chat([{"role": "user", "content": "hi"}], "sys")) _, kwargs = cli.chat.completions.create.call_args assert "max_tokens" not in kwargs def test_传入max_tokens_透传到create(self): p = _provider(max_tokens=4096) with patch.object(p, "client") as cli: cli.chat.completions.create.return_value.choices = [MagicMock(message=MagicMock(content="ok"))] import asyncio asyncio.run(p.chat([{"role": "user", "content": "hi"}], "sys")) _, kwargs = cli.chat.completions.create.call_args assert kwargs.get("max_tokens") == 4096 ``` - [ ] **Step 2: RED** → **Step 3: 实现**——`__init__(self, api_key, model="glm-4-flash", base_url="", max_tokens: int | None = None)` 存 `self.max_tokens`;`_call` 的 `create(...)` 加 `**( {"max_tokens": self.max_tokens} if self.max_tokens is not None else {} )` - [ ] **Step 4: GREEN + 全量回归**(pytest 全绿,672+2 不回归) - [ ] **Step 5: Commit** `feat(glm): max_tokens 可选透传(site-import 前置)` ### Task 2: SSRF 校验 transport + 抓取原语(TDD) **Files:** - Create: `app/platform/site_import/__init__.py`(空) - Create: `app/platform/site_import/safe_fetch.py`(校验 transport + fetch 函数) - Test: `tests/test_safe_fetch.py`(新建) - [ ] **Step 1: 版本探针 + 私网拒绝 失败测试**: ```python """safe_fetch:连接层 SSRF 校验(spec §3.1——防 DNS rebinding)。""" import httpx import pytest from app.platform.site_import import safe_fetch class TestVersionPin: def test_httpx私有口存在(self): """httpx/httpcore 升级破坏私有 API 时立即红(spec 要求的版本钉位)——真断言,禁永真。""" import inspect import httpcore t = httpx.AsyncHTTPTransport() assert hasattr(t, "_pool") assert "network_backend" in inspect.signature( httpcore.AsyncConnectionPool.__init__).parameters class TestIsBlockedIp: @pytest.mark.parametrize("ip", [ "127.0.0.1", "10.1.2.3", "172.16.0.1", "172.31.255.255", "192.168.1.1", "169.254.1.1", "0.0.0.0", "100.64.0.1", "::1", "fd00::1", "fe80::1", "::ffff:127.0.0.1", "::ffff:10.0.0.1", ]) def test_私网段全拒(self, ip): assert safe_fetch.is_blocked_ip(ip) is True @pytest.mark.parametrize("ip", ["8.8.8.8", "1.1.1.1", "47.254.176.10", "2606:4700::1"]) def test_公网放行(self, ip): assert safe_fetch.is_blocked_ip(ip) is False class TestValidatingBackend: async def test_connect_tcp_拒绝私网(self): backend = safe_fetch.ValidatingBackend() with pytest.raises(safe_fetch.BlockedAddressError): await backend.connect_tcp("127.0.0.1", 80) # (连接层校验逻辑用 IP 字面量直接钉,不做真实域名 DNS 测试) class TestFetch: async def test_非法端口拒绝(self): with pytest.raises(safe_fetch.FetchError): await safe_fetch.fetch("http://example.com:8080/") async def test_私网URL拒绝(self): with pytest.raises(safe_fetch.FetchError): await safe_fetch.fetch("http://127.0.0.1/x") with pytest.raises(safe_fetch.FetchError): await safe_fetch.fetch("http://10.0.0.1/x") async def test_逐跳重定向每跳重校验(self, monkeypatch): """302 → Location 指向私网 IP → 必须以 FetchError 终结(安全关键路径)。""" # monkeypatch 首跳响应 302 Location=http://10.0.0.2/evil; # 断言重定向跳因私网被拒——每跳走完整校验 ... async def test_请求真实走校验后端(self, monkeypatch): """_make_client 的请求确实路由经 ValidatingBackend(连接层钩子被调用)。""" # monkeypatch ValidatingBackend.connect_tcp 计数;mock 公网请求断言计数>=1 ... async def test_重定向上限5跳(self, monkeypatch): """>5 跳 302 链 → FetchError。""" ... ``` (connect_tcp 的准确签名以 httpcore 1.0.9 `AsyncNetworkBackend` 为准——实施时先读 `httpcore/_backends/anyio.py`,照抄其 connect_tcp 签名与 anyio.getaddrinfo 用法,仅在校验点插桩。测试对签名的适配在这一步完成。) - [ ] **Step 2: RED** → **Step 3: 实现 safe_fetch.py**: ```python """site_import 安全抓取原语(spec §3.1)。 连接层 SSRF 校验:自定义 httpcore 后端在 connect_tcp 处 getaddrinfo(anyio 异步——单进程零同步 IO 纪律)→ IP 黑白名单校验 → 只连已验 IP。DNS rebinding 的 TOCTOU 由"解析与连接同一后端同一结果"闭合。禁止改写 URL host 为 IP(会破坏 TLS 证书校验)——SNI 由 httpcore 以原 hostname 包 TLS。 """ import asyncio import ipaddress from typing import Any import anyio import httpcore import httpx BLOCKED_NETS = [ipaddress.ip_network(n) for n in [ "127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "169.254.0.0/16", "0.0.0.0/8", "100.64.0.0/10", "::1/128", "fc00::/7", "fe80::/10", "::ffff:0:0/96", ]] ALLOWED_PORTS = {80, 443} USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36" class BlockedAddressError(Exception): """目标解析到禁段——连接前拒绝。""" class FetchError(Exception): """抓取失败(URL 非法/端口禁/请求层错误)。""" def is_blocked_ip(ip: str) -> bool: try: addr = ipaddress.ip_address(ip) except ValueError: return True return any(addr in net for net in BLOCKED_NETS) class ValidatingBackend(httpcore.AsyncNetworkBackend): """connect_tcp 钩子:anyio.getaddrinfo 解析→校验→只连已验 IP。""" async def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None): # 签名照 httpcore 1.0.9 对齐 # ⚠️ 评审折入两个实现陷阱: # 1) anyio 不重导出 socket 常量——AF_UNSPEC/SOCK_STREAM 用 socket 模块; # 2) 默认后端 anyio.connect_tcp(remote_host=host) 会内部重新解析 DNS—— # 会重新打开 rebinding TOCTOU!必须把「已验 IP 字面量」传给连接调用, # 禁止把原 hostname 再递给任何会做 DNS 的连接函数。 import socket as _socket if port not in ALLOWED_PORTS: raise BlockedAddressError(f"port {port} not allowed") infos = await anyio.getaddrinfo(host, port, family=_socket.AF_UNSPEC, type=_socket.SOCK_STREAM) candidates = [] for fam, _, _, sockaddr in infos: ip = sockaddr[0] if isinstance(ip, bytes): ip = ip.decode("idna") if is_blocked_ip(str(ip)): raise BlockedAddressError(f"{host} resolves to blocked {ip}") candidates.append((fam, str(ip), port)) # 只连已验 IP 字面量(逐候选尝试的简化 happy-eyeballs) ... def _make_client(timeout: float = 10.0) -> httpx.AsyncClient: transport = httpx.AsyncHTTPTransport() # 私有口替换(版本探针测试钉住): 同一校验后端注入连接池 transport._pool = httpcore.AsyncConnectionPool( network_backend=ValidatingBackend()) return httpx.AsyncClient(transport=transport, timeout=timeout, follow_redirects=False, headers={"User-Agent": USER_AGENT}) async def fetch(url: str, timeout: float = 10.0) -> httpx.Response: """单页抓取: 端口预检 + 逐跳手动重定向(≤5 跳, 每跳重新过 fetch 即重校验)。""" # 1) URL 解析: scheme http(s), port ∈ {80,443}(显式非标端口直接 FetchError) # 2) client.get; 3xx → 取 Location 绝对化 → 递归 fetch(每跳同校验); >5 跳 FetchError ... ``` (`ValidatingBackend.connect_tcp` 的实现以照抄 httpcore 默认 anyio 后端为主干——实施时打开 `site-packages/httpcore/_backends/anyio.py` 参考 `AsyncHTTPConnection`/`AsyncNetworkBackend.connect_tcp` 的默认实现,在解析处插入校验。`_make_client` 的私口替换照 spec §3.1。) - [ ] **Step 4: GREEN(含连接真实公网站点冒烟——如 https://example.com 200;离线环境跳过记档)** - [ ] **Step 5: 全量回归 + Commit** `feat(site-import): SSRF 连接层校验 fetch 原语` ## Batch B:清洗抽取 + 骨架模板 + 生成器(crawler 编排在 Batch C/Task 6) ### Task 3: HTML 清洗与抽取纯函数(TDD) **Files:** - Create: `app/platform/site_import/extract.py` - Test: `tests/test_site_import_extract.py` - [ ] **Step 1: 失败测试**(fixture 内联三段 HTML:WP 形态含内联色/logo img/wa.me/mailto/模板残留 555 电话 + 一段外链 CSS 文本 + 一个 >500KB 模拟页)——覆盖:清洗去 script/style/nav/footer/header/comment、正文文本化、8K 截断、主色统计(排除白黑灰——饱和度阈值 + #fff/#000/#888 近邻跳过;**两处都拿不到 → 默认 `#066fc0`**)、CSS 文本同样统计、logo URL 提取(多 logo 取路径日期新者)、wa.me 归一 E.164、mailto 提取、模板残留命中清单(555 电话/Zemez/Add Your Heading/(800) 123-0045/Glasgow D04 89GR)、选页链接发现(同源 + 关键词 en/es/zh + page/N 分页)、薄子页跳过阈值、入口页 SPA 判定(<200 字符) - [ ] **Step 2: RED** → **Step 3: 实现**(纯函数:`clean_html(html)->str`、`extract_color(html, css_text)->str|None`、`extract_logo_url(html)->str|None`、`extract_contacts(html)->{whatsapp,email}`、`find_residue(texts)->list[str]`、`discover_links(base_url, html)->list[str]`、`is_thin(text)->bool`;**禁第三方解析库**——项目零新依赖,用 stdlib `html.parser` 或正则实现,鲁棒性以 fixture 覆盖为准) - [ ] **Step 4: GREEN + Commit** `feat(site-import): HTML 清洗与抽取纯函数` ### Task 4: 行为骨架模板 + 占位符体系 **Files:** - Create: `app/platform/site_import/template/`(`system_template.md`、`extractor_template.md`、`evaluator_template.md`、`placeholders.json`) - Create: `app/platform/site_import/fewshot/`(`README.md` 注明来源两租户+数据治理记档;`fewshot_excerpts.md`) - Test: `tests/test_site_import_template.py` - [ ] **Step 1: 骨架抽取**——以 `seed_profiles/jiancai/prompts/system.md` 为底,参考生产 yotaienergys/televicsic 版(**来源任选其一、一次性取回存本地文件,勿在测试里连生产**:① `ssh all-kefu-prod` + `docker exec all-kefu-app-1 python -c` 读 `/app/tenants/*.db` 最新版 config;② 生产 admin API `GET /api/platform/tenants/{slug}/config`):通用纪律段(语言规则/职能边界/红线/磨价/联系方式策略/对话推进/上下文追踪/禁止说的话)原样保留;行业段(首段身份、§三类请求中产品举例、§说话风格的推荐句、固定追问句、好差示例组、关注信息清单)替换为占位符 `{{身份描述}}`、`{{产品追问路径}}`、`{{推荐建议}}`、`{{好差示例组}}`、`{{关注信息}}` 等;extractor/evaluator 同法(字段名与示例占位化) - [ ] **Step 2: 失败测试**: ```python class TestTemplate: def test_骨架无行业残留(self): t = read template system_template.md for w in ["管道", "pipe", "HESUPIPE", "储能", "storage", "炉辊", "roller", "TELEVIC"]: assert w.lower() not in t.lower() # fewshot 文件除外——模板本体必须纯净 def test_占位符与清单一致(self): placeholders.json 列出的每个 {{名}} 都在三个模板中出现 模板中的每个 {{名}} 都在 placeholders.json def test_红线段存在(self): assert "红线" in t and "价格数字" in t and "不得" not in ... # 关键纪律锚点逐个断言 ``` - [ ] **Step 3: few-shot 摘录**——从两租户实际配置各摘:身份段/追问路径/好差示例对 3 组/事实卡骨架/提取器字段段/评估器示例段(总量 ≤6K tokens——`utf8Bytes` 口径估算,超了删示例) - [ ] **Step 4: GREEN + Commit** `feat(site-import): 行为骨架模板与 few-shot 语料` ### Task 5: 生成器(AI 两段 + 校验器,TDD) **Files:** - Create: `app/platform/site_import/generator.py` - Test: `tests/test_site_import_generator.py` - [ ] **Step 1: 失败测试**(provider 全 fake——注入返回固定 JSON/文本): ```python class TestCall1Parse: async def test_带markdown栅栏的JSON能解析(self): ... async def test_栅栏+杂文解析失败自动重试一次(self): ... async def test_重试再失败任务failed(self): ... class TestNumbersValidator: def test_全半角与单位归一(self): # "950℃" vs "950°C"、"75~90%" vs "75~90%" 视为同数字 def test_源文没有的数字被剔除并注记(self): ... class TestIndustryBranch: def test_规格公开_人设允许聊参数(self): ... # 生成 system 含参数话术分支 def test_规格不公开_参数转经理(self): ... # 事实卡含"参数未公布"规则句 class TestBuild: async def test_占位符全填充_模板零残留(self): ... async def test_品牌四件在输出(self): ... # brand_title/welcome/topics/placeholder async def test_空占位符出厂兜底加warning(self): ... async def test_生成物中文断言(self): ... async def test_墙尾巴三语(self): ... async def test_es生成失败降级为null(self): ... def test_残留命中在build层从联系方式与事实卡剔除(self): ... ``` - [ ] **Step 2: RED** → **Step 3: 实现 generator.py**(`analyze(pages_text) -> Analysis`(**入口截断总量 ≤24K 字符**——10 页 ×8K 不封顶会爆 token 预算,截断在 analyze 入口做,spec §3.2)、`generate(analysis) -> GenOutput`、`validate_numbers(gen, source_texts)`、`build_draft(crawl_result, gen) -> draft包`;AI 调用经注入的 provider 接口——生产装配用 resolver 的 `create_provider(config_data)` + max_tokens 透传 + `asyncio.wait_for` 120s 包裹;**MAX_TOKENS_TARGET = Task 1 核查结论**;调用 prompt 顶层带数据边界声明与「严格只输出 JSON」指令) - [ ] **Step 4: GREEN + 全量回归 + Commit** `feat(site-import): 两段生成器与数字校验器` ## Batch C:任务表 + API ### Task 6: 任务管理(TDD) **Files:** - Create: `app/platform/site_import/task.py` - Create: `app/platform/site_import/crawler.py`(编排 safe_fetch + extract:BFS 选页≤10、逐页 fetch、CSS/logo 追抓 15s 超时、薄子页跳过、入口页 SPA 判定、内容不足判定、**logo 下载校验**——Content-Type+魔数双验只收 png/jpg/jpeg/webp、≤2MB、SVG/超限放弃头像项+warnings 注记,含枚举测试) - Test: `tests/test_site_import_task.py` - [ ] **Step 1: 失败测试**(crawler 打桩——monkeypatch safe_fetch.fetch 返回 fixture Response):状态机流转、(slug, task_id) 联合索引跨租户 404、**全平台**并发 ≤2 第三个拒绝、**频控 3 次/15 分钟(新建与 retry 同计,超限 429)**(复用 config_routes.SlidingWindowCounter 或同款新实例)、7 分钟总超时强 failed、generate 失败可 retry 同 task 复活且不重抓、30 分钟惰性清理、强引用(任务对象在表内有引用)、**内容不足三触发各自映射失败文案**(<1K 总字符 / 调用 1 产品线为空 / 入口页非 text/html 零页)、**logo 下载格式/大小过滤**(SVG 放弃+warning、>2MB 放弃+warning、png 通过) - [ ] **Step 2: RED** → **Step 3: 实现**(模块级 `_tasks: dict[tuple[str, str], ImportTask]` + `asyncio.Lock`;`run_import(slug, task_id, url)` 为 `asyncio.wait_for(_pipeline(), 7*60)` 的包装;pipeline = crawl → analyze → generate → build → done) - [ ] **Step 4: GREEN + Commit** `feat(site-import): 导入任务编排与内存任务表` ### Task 7: API 路由 + 挂载(TDD) **Files:** - Create: `app/platform/routes/site_import_routes.py` - Modify: `app/main.py`(include_router,照 config_routes 挂载位置) - Test: `tests/test_site_import_api.py` - [ ] **Step 1: 失败测试**(夹具用 conftest 的 `db_source_app` + `super_client`(超管)+ 手建租户管理员 client——注意 test_config_api.py 里的 SETTINGS 是 URL 常量不是夹具,勿按图索骥): - POST 新建 `{url}` → 200 `{task_id}`;非 http(s) 400;停用租户 403;同租户进行中 409;**频控第 4 次 429**;retry 带不存在 task_id 404;retry 非 generate-failed 态 409 - GET 两角色 200(超管 + 租户管理员夹具);未认证 401;他人 task_id 404;过期后 404 - **跨租户:租户管理员打别的租户 slug → 403**(_tenant_guard,勿断言 404) - [ ] **Step 2: RED** → **Step 3: 实现**(router = APIRouter(prefix="/api/platform/tenants", dependencies=[Depends(platform_user)]);slug 级用现有 `_tenant_guard` 同款依赖;启动只做校验+建任务,pipeline 由任务模块跑) - [ ] **Step 4: GREEN + 全量回归(672+新增)+ Commit** `feat(site-import): POST/GET API 与挂载` ## Batch D:admin 前端 ### Task 8: API 模块 + model 纯函数(TDD) **Files:** - Create: `admin/src/api/siteImport.ts` - Create: `admin/src/views/site-import/model.ts` - Test: `admin/src/views/site-import/__tests__/model.test.ts` - [ ] **Step 1: 失败测试**——纯函数:`draftToBrandForm(draft)`、`draftToPromptsTabs(draft)`、`dataUrlToBlob(dataUrl)`、`buildSaveSequence(draft, snapshot)`(顺序=头像→brand[GET 后置]→prompts→wall[池形态分派:单句替换/多句追加]→contact;返回步骤数组含每步的 API 调用描述)、`warningsLevel(w)`(规格判定=info/剔除=warning)、`isTerminal404(err)` - [ ] **Step 2: RED → 实现 → GREEN** - [ ] **Step 3: `pnpm test`(188+新增)+ Commit** `feat(admin): site-import API 与模型纯函数` ### Task 9: 导入对话框 + 审核页 **Files:** - Create: `admin/src/views/site-import/index.vue`(对话框 + 审核页全屏 dialog) - Modify: `admin/src/views/tenants/list/index.vue`(详情抽屉加「从网站导入」按钮) - Modify: `admin/src/views/config/center/index.vue`(页头加按钮) - [ ] **Step 1: 导入对话框**——URL 输入 + 开始(POST → 3s 轮询 GET:pending/crawling「抓取中·第 N 页」/generating「AI 生成中」/done 进审核页/failed 显示白话错误 + retry 按钮(generate 失败时)/7.5 分钟停轮询提示可重试;关闭确认「放弃本次导入?」;404=过期终态文案) - [ ] **Step 2: 审核页**——复用 ConfigCard 壳;左列品牌表单(草稿预填;头像 base64 `` 预览 + 「不上传头像」勾选);右列四件套 tabs(等宽 textarea + 字节计数);**墙尾巴区 en/zh/es 三行单行编辑**;联系方式两开关;顶部 warnings 黄条(分级着色);页头「保存将覆盖当前配置(产生新版本可回滚)」;底部「保存全部」按 buildSaveSequence 依次调(进度显示第几组;部分失败重试跳过已完成组)+「放弃」 - [ ] **Step 3: 两入口按钮接线** - [ ] **Step 4: `pnpm test` + `pnpm build` 全绿 + Commit** `feat(admin): 从网站导入对话框与审核页` ## Batch E:终验 ### Task 10: 全局回归 + 生产实测走查 - [ ] **Step 1: 后端全量** `PYTHONUTF8=1 .venv/Scripts/python -m pytest -q`(672+新增全绿) - [ ] **Step 2: 前端全量 + build** - [ ] **Step 3: 本地真实走查**(uvicorn 8018 + admin dev):对 `https://televicsic.com` 走完整流程——POST → 轮询到 done → 草稿包完整(品牌四件/四件套/三语尾巴/warnings 有规格判定)→ 审核页改两处 → 保存全部 → GET config 验证版本递增与内容落位 → 新会话对话验证行业身份。**两种角色各走一遍**(超管 + 租户管理员——spec §9.4,租户管理员夹具同 Task 7 建法) - [ ] **Step 4: SSRF 实测**——POST `http://127.0.0.1/x` 400;POST `http://10.0.0.1/x` 400 - [ ] **Step 5: 频控实测**——连打 4 次 POST 第 4 次 429 - [ ] **Step 6: 收尾**——临时产物清理、`git status` 干净、batch 记档 commit --- ## 依赖与顺序 - Task 1 → 5(max_tokens 扩展被生成器用);Task 2 → 6(fetch 原语被编排用);Task 3、4 → 5;Task 5 → 6 → 7;Task 8 → 9;Task 10 收口 - 后端 Batch A-C 严格顺序;Batch D 依赖 C 的 API 形状(可先写 model 纯函数后写 UI) ## Task 6 消费方备忘(Task 3 评审折入,实施 Task 6 时带上) - extract_color 返回 None 时 crawler 落默认 `#066fc0`(spec §3.1——函数层不兜底) - discover_links 含入口页自身(href="#"/"/")——crawler 需对 start URL 去重防浪费 ≤10 页名额 - 500KB 截断页走 clean_html 前先由本任务修复的未闭合块二遍清洗兜住 ## Task 5 消费方备忘(Task 4 评审折入) - `{{固定追问句}}` 同一字符串落两处(模板 line 34/102)——build 时单值双填 - `{{产品统称}}`(中文语境)与 `{{产品统称EN}}`(英文语境)按语言分别取值 - system 占位符实为 15 个(计划清单 8 + 实证散点 7:认证清单/预算数字示例/零上下文首轮例句/产品统称EN/已知信息重问示例/品类错推示例/产品统称第二处)——build 的填槽映射按 placeholders.json 为准 - 模板固定段两处大宗单位残留(每吨目标价/MOQ is 5 tons)为 spec 逐字保留的有意取舍——记档非缺陷