feat(site-import): POST/GET API 与挂载

Task 7(spec §4):/api/platform/tenants/{slug}/site-import 两端点,
_tenant_guard 守租户(两角色)。POST 校验链:鉴权 → 租户存在/未停用
(404/403) → 形态预检(retry 只认 generate+task_id 非空;新建 URL 复用
safe_fetch._validate_url 单一事实源——scheme/内网字面量/端口,POST 即拒
400「仅支持公网网站地址」) → task.start;异常映射 ImportRateLimited→429 /
ConcurrentLimit→409 / LookupError→404 / ValueError→409(str(exc) 作
detail,任务层白话文案透出)。GET 五键 status/phase/pages/draft/error,
(slug,task_id) 联合索引即越权防线,过期 404「任务不存在或已过期」。
main.py 照 config_routes 位置挂载。测试 15 条(管线全打桩)。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-20 21:33:00 +08:00
parent e1e97dac2c
commit 57adf8e2a6
3 changed files with 397 additions and 0 deletions
+7
View File
@@ -238,6 +238,9 @@ from app.platform.routes.config_routes import (
settings_router as platform_settings_router,
)
from app.platform.routes.data_routes import router as platform_data_router
from app.platform.routes.site_import_routes import (
router as platform_site_import_router,
)
from app.platform.routes.stats_routes import router as platform_stats_router
from app.platform.routes.tenant_routes import router as platform_tenant_router
from app.platform.routes.upload_routes import router as platform_upload_router
@@ -263,6 +266,10 @@ app.include_router(platform_data_router)
# stats.py,逐库查询应用层合并(禁 ATTACH,spec §7.6);合并端点 require_super
# (?tenant= 单租户下钻),单租户端点 _tenant_guard。
app.include_router(platform_stats_router)
# 从网站导入 API(site-import Task 7): POST 启动/GET 轮询(_tenant_guard,
# 两角色)——管线在 app/platform/site_import/task.py 的内存任务表,本层只做
# 校验与异常映射(429/409/404)。
app.include_router(platform_site_import_router)
# 平台认证路由(Task 2):本任务即挂——已知租户 Host(如 jiancai)下
# /api/platform/auth/* 经 Host 中间件直达,TDD 循环闭环。Task 3 将为
# /api/platform/ 前缀加中间件豁免(平台后台不依赖租户域名解析);
+95
View File
@@ -0,0 +1,95 @@
"""site_import API(spec §4)——2 端点,两角色可用,_tenant_guard 守租户。
- POST /{slug}/site-import:新建(body.url)或复活(body.retry="generate"
+ body.task_id)。校验次序:鉴权(_tenant_guard 依赖链,401/403/409 门
连锁)→ 租户存在且未停用(404/403)→ 形态预检(retry 只认 generate 且
task_id 非空;新建 URL 过 safe_fetch 的 URL 层预检——scheme/host 字面量
禁段/端口白名单,spec §5「内网/非法地址 POST 即拒」,连接层二次兜底)
→ task.start。异常映射(Task 7 备忘):ImportRateLimited→429 /
ConcurrentLimit→409 / LookupError→404(retry 目标不存在)/ ValueError→
409(非 generate-failed 态),str(exc) 作 detail——任务层的白话文案
原样透出,API 层不二次翻译(FetchError 已在任务层白话化,同理由)。
- GET /{slug}/site-import/{task_id}:轮询。task.get 的 (slug, task_id)
联合索引即越权防线(跨租户持他人 id → None → 404);30 分钟过期同理
404「任务不存在或已过期」(前端按终态「任务已过期,请重新导入」处理)。
响应五键(spec §4):status(=任务 state)/ phase / pages(=任务
pages_fetched)/ draft(仅 done,spec §4 草稿包整包)/ error。
启动只做校验+建任务——pipeline 由任务模块的后台 aio task 跑,本层不
感知 crawler/generator(测试也因此能整体打桩 task.start/task.get)。
"""
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
from app.platform.routes.tenant_routes import _get_tenant_or_404, _tenant_guard
from app.platform.site_import import safe_fetch, task
router = APIRouter(prefix="/api/platform/tenants", tags=["platform-site-import"])
class StartRequest(BaseModel):
url: str = ""
retry: str = "" # "generate" 复活模式(spec §3.3)
task_id: str = "" # retry 时指定任务
@router.post("/{slug}/site-import")
async def start_import(slug: str, body: StartRequest, request: Request,
user=Depends(_tenant_guard)):
"""启动/复活导入任务 → {task_id}。校验次序与异常映射见模块 docstring。"""
engine = request.app.state.platform_engine
async with AsyncSession(engine, expire_on_commit=False) as session:
tenant = await _get_tenant_or_404(session, slug)
if tenant.status != "active":
raise HTTPException(
status_code=403, detail=f"租户 {slug!r} 已停用,无法发起导入")
url: str | None = None
retry_task_id: str | None = None
if body.retry:
if body.retry != "generate":
raise HTTPException(
status_code=400,
detail=f"retry 只支持 generate(实际 {body.retry!r})")
if not body.task_id:
raise HTTPException(
status_code=400, detail="retry 形态必须带 task_id")
retry_task_id = body.task_id
else:
# URL 层预检单一事实源在 safe_fetch(scheme/禁段字面量/端口)——
# 私有口内部复用,不另抄一份漂移;不公网形态统一白话 400
try:
safe_fetch._validate_url(body.url)
except safe_fetch.FetchError as exc:
raise HTTPException(
status_code=400, detail="仅支持公网网站地址") from exc
url = body.url
try:
task_id = await task.start(slug, url, retry_task_id=retry_task_id)
except task.ImportRateLimited as exc:
raise HTTPException(status_code=429, detail=str(exc)) from exc
except task.ConcurrentLimit as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except LookupError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
return {"task_id": task_id}
@router.get("/{slug}/site-import/{task_id}")
async def get_import_task(slug: str, task_id: str,
user=Depends(_tenant_guard)):
"""轮询任务状态(spec §4 五键;404=不存在/已过期/跨租户)。"""
t = task.get(slug, task_id)
if t is None:
raise HTTPException(status_code=404, detail="任务不存在或已过期")
return {
"status": t.state,
"phase": t.phase,
"pages": t.pages_fetched,
"draft": t.draft if t.state == task.TaskState.DONE else None,
"error": t.error,
}
+295
View File
@@ -0,0 +1,295 @@
"""site_import API(Task 7)——路由/鉴权/异常映射;管线全打桩(spec §4/§5)。
夹具:conftest 的 db_source_app + super_client(超管);租户管理员照
test_tenant_admin_api 建法(创建向导带管理员 → 登录 + 首登改密过 409 门)。
task.start / task.get 打桩——本文件只测路由层(鉴权/校验/异常映射/响应
形状),管线本身归 test_site_import_task.py。
"""
from app.platform.site_import import task
TENANTS = "/api/platform/tenants"
LOGIN = "/api/platform/auth/login"
CHANGE_PW = "/api/platform/auth/change-password"
# spec §4 草稿包样例(generator.build 的输出形状——六键整包)
DRAFT = {
"brand": {"title": "Acme", "color": "#123456",
"whatsapp": "+8613800138000", "email": "sales@acme.com",
"welcome": "Hi! What project?", "topics": ["guardrail"],
"placeholder": "Type your message...", "privacy_url": ""},
"avatar": None,
"prompts": {"system": "s...", "company_facts": {"主营": "guardrail"},
"extractor": "e...", "evaluator": "v..."},
"wall_tails": {"en": "tail", "zh": "尾巴", "es": None},
"contact": {"cc_completion": True, "email_light_ask": True},
"warnings": ["规格判定:不公开(依据:产品页参数表判定)"],
}
def _auth(client):
return {"Authorization": f"Bearer {client.platform_token}"}
def _bearer(token):
return {"Authorization": f"Bearer {token}"}
def _create_tenant(client, slug, *, profile="jiancai", admin=None,
password=None):
"""创建租户便捷 POST(test_tenant_admin_api._create 同款收敛)。"""
r = client.post(TENANTS, headers=_auth(client), json={
"name": f"{slug} 租户", "slug": slug, "profile": profile,
"admin_username": admin or f"{slug}-admin",
"admin_password": password or f"{slug}-pass-1",
"domains": []})
assert r.status_code == 201, r.text
return r
def _tenant_admin_token(client, username, old_password,
new_password="tn-pass-1"):
"""租户管理员登录 + 首登改密(过 must_change 409 门),返回可用 token。"""
r = client.post(LOGIN, json={"username": username,
"password": old_password})
assert r.status_code == 200
token = r.json()["token"]
r = client.post(CHANGE_PW, headers=_bearer(token), json={
"old_password": old_password, "new_password": new_password})
assert r.status_code == 200
return token
class _FakeStart:
"""task.start 桩:记录调用 (slug, url, retry_task_id);error 非空即抛。"""
def __init__(self, error=None, result="tid-stub"):
self.error = error
self.result = result
self.calls = []
async def __call__(self, slug, url, *, retry_task_id=None):
self.calls.append((slug, url, retry_task_id))
if self.error is not None:
raise self.error
return self.result
def _stub_start(monkeypatch, error=None, result="tid-stub"):
stub = _FakeStart(error, result)
monkeypatch.setattr(task, "start", stub)
return stub
def _stub_get(monkeypatch, t):
"""task.get 桩:(slug, task_id) 联合索引语义(他人 slug → None)。"""
monkeypatch.setattr(
task, "get",
lambda slug, task_id: t if (slug, task_id) == (t.slug, t.task_id)
else None)
class TestPostStart:
def test_超管新建任务200(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
stub = _stub_start(monkeypatch, result="tid-1")
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_auth(super_client),
json={"url": "https://acme.com/"})
assert r.status_code == 200, r.text
assert r.json() == {"task_id": "tid-1"}
assert stub.calls == [("demo", "https://acme.com/", None)]
def test_租户管理员自己租户200(self, super_client, monkeypatch):
_create_tenant(super_client, "demo", admin="demo-admin",
password="demo-pass-1")
token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
stub = _stub_start(monkeypatch)
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_bearer(token),
json={"url": "https://acme.com/"})
assert r.status_code == 200, r.text
assert r.json()["task_id"] == "tid-stub"
assert stub.calls == [("demo", "https://acme.com/", None)]
def test_租户管理员打别的租户403(self, super_client, monkeypatch):
_create_tenant(super_client, "demo", admin="demo-admin",
password="demo-pass-1")
_create_tenant(super_client, "other", profile="blank")
token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
stub = _stub_start(monkeypatch)
# 已存在租户与不存在 slug 都是 403——_tenant_guard 依赖先于处理器
# (勿断言 404,spec §5)
assert super_client.post(
f"{TENANTS}/other/site-import", headers=_bearer(token),
json={"url": "https://a.com/"}).status_code == 403
assert super_client.post(
f"{TENANTS}/ghost/site-import", headers=_bearer(token),
json={"url": "https://a.com/"}).status_code == 403
assert stub.calls == [] # 处理器未触达
# 对照:超管打不存在 slug → 处理器内租户存在性检查 404
assert super_client.post(
f"{TENANTS}/ghost/site-import", headers=_auth(super_client),
json={"url": "https://a.com/"}).status_code == 404
def test_未认证401(self, super_client):
_create_tenant(super_client, "demo")
assert super_client.post(
f"{TENANTS}/demo/site-import",
json={"url": "https://a.com/"}).status_code == 401
def test_停用租户403(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
json={"status": "disabled"})
assert r.status_code == 200
stub = _stub_start(monkeypatch)
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_auth(super_client),
json={"url": "https://a.com/"})
assert r.status_code == 403
assert "停用" in r.json()["detail"]
assert stub.calls == []
def test_非http协议400(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
stub = _stub_start(monkeypatch)
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_auth(super_client),
json={"url": "ftp://x.com/"})
assert r.status_code == 400
assert r.json()["detail"] == "仅支持公网网站地址"
# spec §5:内网 IP 字面量 POST 即拒(连接层只是二次兜底)
assert super_client.post(
f"{TENANTS}/demo/site-import", headers=_auth(super_client),
json={"url": "http://127.0.0.1/x"}).status_code == 400
# 空 URL(无 retry 形态)同 400
assert super_client.post(
f"{TENANTS}/demo/site-import", headers=_auth(super_client),
json={}).status_code == 400
assert stub.calls == []
def test_频控429(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
_stub_start(monkeypatch, error=task.ImportRateLimited(
"导入太频繁,15 分钟内最多 3 次,稍后再试"))
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_auth(super_client),
json={"url": "https://a.com/"})
assert r.status_code == 429
assert "导入太频繁" in r.json()["detail"]
def test_并发409(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
_stub_start(monkeypatch, error=task.ConcurrentLimit(
"已有导入任务进行中,稍后再试"))
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_auth(super_client),
json={"url": "https://a.com/"})
assert r.status_code == 409
assert "已有导入任务进行中" in r.json()["detail"]
def test_retry不存在task404(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
stub = _stub_start(monkeypatch, error=LookupError(
"task 'nope' not found for tenant 'demo'"))
# retry 形态预检:task_id 空 / 非法 retry 值 → 400
assert super_client.post(
f"{TENANTS}/demo/site-import", headers=_auth(super_client),
json={"retry": "generate", "task_id": ""}).status_code == 400
assert super_client.post(
f"{TENANTS}/demo/site-import", headers=_auth(super_client),
json={"retry": "crawl", "task_id": "x"}).status_code == 400
# 任务不存在 → 404(str(exc) 作 detail)
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_auth(super_client),
json={"retry": "generate", "task_id": "nope"})
assert r.status_code == 404
assert "nope" in r.json()["detail"]
assert len(stub.calls) == 1 # 形态预检 400 的两次未触达桩
def test_retry非generatefailed态409(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
_stub_start(monkeypatch, error=ValueError(
"任务当前状态不可重试(仅 generate 阶段失败可复活)"))
r = super_client.post(f"{TENANTS}/demo/site-import",
headers=_auth(super_client),
json={"retry": "generate", "task_id": "tid-x"})
assert r.status_code == 409
assert "不可重试" in r.json()["detail"]
class TestGetTask:
def test_进行中返回state与pages(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
t = task.ImportTask("demo", "tid-1", "https://a.com/")
t.state = task.TaskState.CRAWLING
t.phase = "crawling"
t.pages_fetched = 3
_stub_get(monkeypatch, t)
r = super_client.get(f"{TENANTS}/demo/site-import/tid-1",
headers=_auth(super_client))
assert r.status_code == 200
assert r.json() == {"status": "crawling", "phase": "crawling",
"pages": 3, "draft": None, "error": None}
# 未认证 401(依赖链先于处理器)
assert super_client.get(
f"{TENANTS}/demo/site-import/tid-1").status_code == 401
def test_完成返回draft草稿包(self, super_client, monkeypatch):
_create_tenant(super_client, "demo", admin="demo-admin",
password="demo-pass-1")
token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
t = task.ImportTask("demo", "tid-2", "https://a.com/")
t.state = task.TaskState.DONE
t.pages_fetched = 5
t.draft = DRAFT
t.finished_at = 1234.5
_stub_get(monkeypatch, t)
# 租户管理员角色走 GET(两角色 200 的另一半;超管在其余用例覆盖)
r = super_client.get(f"{TENANTS}/demo/site-import/tid-2",
headers=_bearer(token))
assert r.status_code == 200
body = r.json()
assert body["status"] == "done"
assert body["draft"] == DRAFT # spec §4 草稿包整包透传
assert body["pages"] == 5
assert body["phase"] is None
assert body["error"] is None
def test_失败返回白话error(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
t = task.ImportTask("demo", "tid-4", "https://a.com/")
t.fail("网站打不开或响应太慢——检查 URL", phase="crawl")
_stub_get(monkeypatch, t)
r = super_client.get(f"{TENANTS}/demo/site-import/tid-4",
headers=_auth(super_client))
assert r.status_code == 200
assert r.json() == {"status": "failed", "phase": None, "pages": 0,
"draft": None,
"error": "网站打不开或响应太慢——检查 URL"}
def test_他人taskid404(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
_create_tenant(super_client, "other", profile="blank")
t = task.ImportTask("demo", "tid-3", "https://a.com/")
t.state = task.TaskState.GENERATING
t.phase = "generating"
_stub_get(monkeypatch, t)
# (slug, task_id) 联合索引:他人 slug 同 task_id → miss → 404
assert super_client.get(
f"{TENANTS}/other/site-import/tid-3",
headers=_auth(super_client)).status_code == 404
# 本租户同 task_id → 200
assert super_client.get(
f"{TENANTS}/demo/site-import/tid-3",
headers=_auth(super_client)).status_code == 200
def test_过期后404(self, super_client, monkeypatch):
_create_tenant(super_client, "demo")
monkeypatch.setattr(task, "get", lambda slug, task_id: None)
r = super_client.get(f"{TENANTS}/demo/site-import/gone",
headers=_auth(super_client))
assert r.status_code == 404
assert r.json()["detail"] == "任务不存在或已过期"