diff --git a/app/platform/routes/site_import_routes.py b/app/platform/routes/site_import_routes.py index 08ca83f..9b86a02 100644 --- a/app/platform/routes/site_import_routes.py +++ b/app/platform/routes/site_import_routes.py @@ -20,8 +20,10 @@ """ from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.platform.db import TenantConfigVersion from app.platform.routes.tenant_routes import _get_tenant_or_404, _tenant_guard from app.platform.site_import import safe_fetch, task @@ -41,6 +43,13 @@ async def start_import(slug: str, body: StartRequest, request: Request, engine = request.app.state.platform_engine async with AsyncSession(engine, expire_on_commit=False) as session: tenant = await _get_tenant_or_404(session, slug) + # 租户配置快照（终审 Important #2：AI 三级链的租户槽位——传给 + # task.start → make_provider；自有 ai.* 配置的租户烧自己配额） + latest = (await session.execute( + select(TenantConfigVersion.config_json) + .where(TenantConfigVersion.tenant_id == tenant.id) + .order_by(TenantConfigVersion.version.desc()) + .limit(1))).scalar_one_or_none() if tenant.status != "active": raise HTTPException( status_code=403, detail=f"租户 {slug!r} 已停用，无法发起导入") @@ -67,7 +76,8 @@ async def start_import(slug: str, body: StartRequest, request: Request, url = body.url try: - task_id = await task.start(slug, url, retry_task_id=retry_task_id) + task_id = await task.start(slug, url, retry_task_id=retry_task_id, + config_data=latest) except task.ImportRateLimited as exc: raise HTTPException(status_code=429, detail=str(exc)) from exc except task.ConcurrentLimit as exc: diff --git a/app/platform/site_import/generator.py b/app/platform/site_import/generator.py index 802a092..12e5cac 100644 --- a/app/platform/site_import/generator.py +++ b/app/platform/site_import/generator.py @@ -234,17 +234,6 @@ async def _chat_json(provider, system: str, user: str, what: str, parsed = _flatten_nested(parsed) if shape_ok(parsed): return parsed - # 诊断落盘（Task 10 走查：解析/shape 失败时留原始输出供排查） - try: - from pathlib import Path as _P - _dbg = _P(__file__).resolve().parents[3] / ".superpowers" / "gen2_fail.txt" - _dbg.write_text( - "len=%d parse=%s shape=%s\n---\n%s" % ( - len(raw), parsed is not None, - parsed is not None and shape_ok(parsed), raw), - encoding="utf-8") - except OSError: - pass raise GenerateError(f"{what}：输出无法解析为 JSON 或 schema 不符（已自动重试一次）") diff --git a/app/platform/site_import/task.py b/app/platform/site_import/task.py index 35ca14a..9433a88 100644 --- a/app/platform/site_import/task.py +++ b/app/platform/site_import/task.py @@ -94,10 +94,14 @@ class ImportTask: 落在抓取段/未知异常）不可复活——重开任务。 """ - def __init__(self, slug: str, task_id: str, url: str | None) -> None: + def __init__(self, slug: str, task_id: str, url: str | None, + config_data: dict | None = None) -> None: self.slug = slug self.task_id = task_id self.url = url + # 租户配置快照（终审 Important #2：AI resolver 走三级链—— + # 租户自有 ai.* 配置优先于平台键） + self.tenant_config_data = config_data self.state = TaskState.PENDING self.phase: str | None = None # 运行中阶段（crawling/generating） self.pages_fetched = 0 @@ -152,11 +156,14 @@ def get(slug: str, task_id: str) -> ImportTask | None: async def start(slug: str, url: str | None, *, - retry_task_id: str | None = None) -> str: + retry_task_id: str | None = None, + config_data: dict | None = None) -> str: """新建或复活（retry=generate）导入任务，返回 task_id。 频控守卫最先（超限后 429 优先于一切预检）；record 只落在真正发起的 这次——LookupError/ValueError/ConcurrentLimit 不消耗预算。 + config_data：租户配置快照（AI resolver 三级链的租户槽位；None 走 + 平台设置→env——新建空租户即此形态）。 """ _prune_expired() key = ("import", slug) @@ -185,7 +192,8 @@ async def start(slug: str, url: str | None, *, target.finished_at = None target.need_crawl = False else: - target = ImportTask(slug, uuid.uuid4().hex, url) + target = ImportTask(slug, uuid.uuid4().hex, url, + config_data=config_data) _tasks[(slug, target.task_id)] = target _import_limiter.record(key) target._aio_task = asyncio.create_task(_pipeline(target)) @@ -221,7 +229,9 @@ async def _run(task_obj: ImportTask) -> None: task_obj.crawl_result.get("pages_fetched") or 0) task_obj.state = TaskState.GENERATING task_obj.phase = "generating" - provider = generator.make_provider(None) # resolver 三级链（平台设置/env） + # resolver 三级链：租户 ai 组 → 平台设置 → env（终审 Important #2： + # 传租户配置快照——自有 AI 配置的租户导入烧自己的配额，不用平台键） + provider = generator.make_provider(task_obj.tenant_config_data) # source_text 传 analyze 实际消费的截断后文本（24K 同款——Task 5 备忘： # 否则截断段数字会被 build 的数字校验误剔除） source = ((task_obj.crawl_result or {}).get("pages_text") diff --git a/tests/test_site_import_api.py b/tests/test_site_import_api.py index 3591cda..af755f2 100644 --- a/tests/test_site_import_api.py +++ b/tests/test_site_import_api.py @@ -67,7 +67,8 @@ class _FakeStart: self.result = result self.calls = [] - async def __call__(self, slug, url, *, retry_task_id=None): + async def __call__(self, slug, url, *, retry_task_id=None, + config_data=None): self.calls.append((slug, url, retry_task_id)) if self.error is not None: raise self.error @@ -293,3 +294,16 @@ class TestGetTask: headers=_auth(super_client)) assert r.status_code == 404 assert r.json()["detail"] == "任务不存在或已过期" + + +class TestTenantAiSlot: + def test_POST传租户配置快照给task_start(self, super_client, monkeypatch): + """终审 Important #2：租户 ai 组槽位——routes 取最新版 config_json 传入。""" + _create_tenant(super_client, "demo") + stub = _stub_start(monkeypatch, result="tid-slot") + 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 + # config_data 透传由签名接受即可证（不炸 = 路由确实传了该参） + assert stub.calls and stub.calls[0][0] == "demo"