docs: 网站自动导入实施计划(10 任务 5 批)

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-20 16:46:49 +08:00
parent 45b0ff6891
commit 196df2dea7
@@ -0,0 +1,362 @@
# 网站自动导入知识库 实施计划
> **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.py 记档)——结果写进 Task 3 的常量
- 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 3 的 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 要求的版本钉位)。"""
t = httpx.AsyncHTTPTransport()
assert hasattr(t, "_pool")
import httpcore
assert hasattr(httpcore, "AsyncConnectionPool")
assert hasattr(httpcore.AsyncConnectionPool, "network_backend") or True # 构造参数
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)
with pytest.raises(safe_fetch.BlockedAddressError):
await backend.connect_tcp("some-host", 80, local_address="10.0.0.5") # 按实现签名适配
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")
```
(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 对齐
if port not in ALLOWED_PORTS:
raise BlockedAddressError(f"port {port} not allowed")
# 照抄 httpcore 默认后端: anyio.getaddrinfo 解析
infos = await anyio.getaddrinfo(host, port, family=anyio.AF_UNSPEC,
type=anyio.SOCK_STREAM)
addr = None
for fam, _, _, sockaddr in infos:
ip = sockaddr[0]
if isinstance(ip, bytes): # ipv6 bytes 形态
ip = ip.decode("idna")
if is_blocked_ip(str(ip)):
raise BlockedAddressError(f"{host} resolves to blocked {ip}")
addr = (fam, sockaddr)
break
# 连已验 sockaddr——实现上委托 anyio.open_tcp_stream 或照默认后端
#(以 httpcore._backends.anyio 的 connect_tcp 实现为准照抄,替换解析源)
...
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:抓取器 + 骨架模板 + 生成器
### 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 近邻跳过)、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 版(通过 GET 生产 API 或直接读两租户 DB 配置——**用本地 curl 拿一次存文件**,勿在测试里连生产)划分:通用纪律段(语言规则/职能边界/红线/磨价/联系方式策略/对话推进/上下文追踪/禁止说的话)原样保留;行业段(首段身份、§三类请求中产品举例、§说话风格的推荐句、固定追问句、好差示例组、关注信息清单)替换为占位符 `{{身份描述}}`、`{{产品追问路径}}`、`{{推荐建议}}`、`{{好差示例组}}`、`{{关注信息}}` 等;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): ...
```
- [ ] **Step 2: RED** → **Step 3: 实现 generator.py**(`analyze(pages_text) -> Analysis`、`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 判定、内容不足判定)
- Test: `tests/test_site_import_task.py`
- [ ] **Step 1: 失败测试**(crawler 打桩——monkeypatch safe_fetch.fetch 返回 fixture Response):状态机流转、(slug, task_id) 联合索引跨租户 404、并发 ≤2 第三个拒绝、**频控 3 次/15 分钟**(复用 config_routes.SlidingWindowCounter 或同款新实例)、7 分钟总超时强 failed、generate 失败可 retry 同 task 复活且不重抓、30 分钟惰性清理、强引用(任务对象在表内有引用)
- [ ] **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: 失败测试**(照 test_config_api.py 的 SETTINGS 夹具形态):
- 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 `<img>` 预览 + 「不上传头像」勾选);右列四件套 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 验证版本递增与内容落位 → 新会话对话验证行业身份
- [ ] **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 → 3 → 6(fetch 原语 → 抽取 → 编排);Task 4 → 5;Task 5 → 6 → 7;Task 8 → 9;Task 10 收口
- 后端 Batch A-C 严格顺序;Batch D 依赖 C 的 API 形状(可先写 model 纯函数后写 UI)