From 57adf8e2a6e3f009e4d674bd5fbf67f03350af84 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= <email@sample.com>
Date: Tue, 20 Oct 2026 21:33:00 +0800
Subject: [PATCH] =?UTF-8?q?feat(site-import):=20POST/GET=20API=20=E4=B8=8E?=
 =?UTF-8?q?=E6=8C=82=E8=BD=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 app/main.py                               |   7 +
 app/platform/routes/site_import_routes.py |  95 +++++++
 tests/test_site_import_api.py             | 295 ++++++++++++++++++++++
 3 files changed, 397 insertions(+)
 create mode 100644 app/platform/routes/site_import_routes.py
 create mode 100644 tests/test_site_import_api.py

diff --git a/app/main.py b/app/main.py
index d600553..6e04984 100644
--- a/app/main.py
+++ b/app/main.py
@@ -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/ 前缀加中间件豁免(平台后台不依赖租户域名解析);
diff --git a/app/platform/routes/site_import_routes.py b/app/platform/routes/site_import_routes.py
new file mode 100644
index 0000000..08ca83f
--- /dev/null
+++ b/app/platform/routes/site_import_routes.py
@@ -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,
+    }
diff --git a/tests/test_site_import_api.py b/tests/test_site_import_api.py
new file mode 100644
index 0000000..3591cda
--- /dev/null
+++ b/tests/test_site_import_api.py
@@ -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"] == "任务不存在或已过期"