From c1a70ecf38e8069740938d2d0f22c787af623acc Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 13 Oct 2026 15:33:00 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E8=B5=B7?= =?UTF-8?q?=E8=8D=89=E6=94=B9=E6=88=90=E5=90=8E=E5=8F=B0=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=EF=BC=9A=E6=96=B0=E8=A1=A8=20catalog=5Fdraft=5Fjobs=E3=80=81ca?= =?UTF-8?q?talog=5Fdraft=5Ftasks=20=E5=AD=98=E4=BB=BB=E5=8A=A1=E5=92=8C?= =?UTF-8?q?=E9=80=90=E8=A1=A8=E7=8A=B6=E6=80=81=EF=BC=8C=E5=90=8C=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E6=BA=90=E5=90=8C=E6=97=B6=E5=8F=AA=E6=9C=89=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E6=8E=92=E9=98=9F=E6=88=96=E8=BF=9B=E8=A1=8C=E4=B8=AD?= =?UTF-8?q?=E7=9A=84=E4=BB=BB=E5=8A=A1=EF=BC=88=E6=8E=A5=E5=8F=A3=E5=85=88?= =?UTF-8?q?=E6=9F=A5=E3=80=81=E9=83=A8=E5=88=86=E5=94=AF=E4=B8=80=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=E5=85=9C=E5=BA=95=EF=BC=89=EF=BC=9BPOST=20/catalog/dr?= =?UTF-8?q?aft-jobs=20=E5=BB=BA=E4=BB=BB=E5=8A=A1=EF=BC=88202=EF=BC=89?= =?UTF-8?q?=EF=BC=8CGET=20/draft-jobs/active=E3=80=81/latest=E3=80=81/{id}?= =?UTF-8?q?=20=E8=AF=BB=E4=BB=BB=E5=8A=A1=EF=BC=8C/cancel=20=E5=81=9C?= =?UTF-8?q?=E6=AD=A2=EF=BC=88=E6=8E=92=E9=98=9F=E7=9A=84=E7=9B=B4=E6=8E=A5?= =?UTF-8?q?=E5=81=9C=EF=BC=8C=E8=BF=9B=E8=A1=8C=E4=B8=AD=E7=9A=84=E5=81=9C?= =?UTF-8?q?=E6=8E=89=E6=AD=A3=E5=9C=A8=E8=BF=9B=E8=A1=8C=E7=9A=84=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B=E8=B0=83=E7=94=A8=EF=BC=8C=E5=B7=B2=E5=86=99=E8=BF=9B?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E7=9A=84=E4=B8=8D=E5=9B=9E=E6=BB=9A=EF=BC=89?= =?UTF-8?q?=EF=BC=8C/retry=20=E5=8F=AA=E9=87=8D=E8=AF=95=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E3=80=81=E5=8F=AA=E5=86=99=E5=87=BA=E4=B8=80=E9=83=A8=E5=88=86?= =?UTF-8?q?=E3=80=81=E5=81=9C=E6=AD=A2=E4=BA=86=E7=9A=84=E8=A1=A8=EF=BC=9B?= =?UTF-8?q?=E5=90=8C=E6=BA=90=20409=20catalog=5Fdraft=5Fbusy=E3=80=81?= =?UTF-8?q?=E8=B5=B7=E8=8D=89=E6=A8=A1=E5=9E=8B=E7=94=A8=E4=B8=8D=E4=BA=86?= =?UTF-8?q?=20409=20catalog=5Fdraft=5Fmodel=5Funavailable=E3=80=81?= =?UTF-8?q?=E5=B7=B2=E7=BB=93=E6=9D=9F=20409=20catalog=5Fdraft=5Ffinished?= =?UTF-8?q?=EF=BC=9B=E6=AF=8F=E5=BC=A0=E8=A1=A8=E4=B8=80=E7=BB=93=E6=9D=9F?= =?UTF-8?q?=E5=B0=B1=E5=B9=B6=E5=85=A5=E7=9B=AE=E5=BD=95=EF=BC=8C=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6=E3=80=81=E8=AE=A1=E6=95=B0=E3=80=81=E7=94=A8=E9=87=8F?= =?UTF-8?q?=E9=9A=8F=E6=97=B6=E8=90=BD=E5=BA=93=EF=BC=9B=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E6=97=B6=E6=8A=8A=E6=B2=A1=E5=81=9A=E5=AE=8C?= =?UTF-8?q?=E7=9A=84=E4=BB=BB=E5=8A=A1=E6=A0=87=E6=88=90=E4=B8=AD=E6=96=AD?= =?UTF-8?q?=E3=80=81=E4=B8=8D=E8=87=AA=E5=8A=A8=E7=BB=AD=E8=B7=91=EF=BC=8C?= =?UTF-8?q?=E5=85=B3=E5=81=9C=E6=97=B6=E5=90=8C=E6=A0=B7=EF=BC=9B=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E6=B8=85=E5=8D=95=E6=AF=8F=E8=A1=8C=E5=8A=A0=20last?= =?UTF-8?q?=5Fdraft=EF=BC=8C=E9=A1=B6=E5=B1=82=E5=8A=A0=20guidance?= =?UTF-8?q?=E3=80=81active=5Fjob=EF=BC=9B=E6=95=B0=E6=8D=AE=E6=BA=90?= =?UTF-8?q?=E7=BA=A7=E7=9A=84=E8=B5=B7=E8=8D=89=E8=A6=81=E6=B1=82=E5=AD=98?= =?UTF-8?q?=E5=9C=A8=20options=20=E9=87=8C=EF=BC=8C=E7=BB=8F=20PUT=20/cata?= =?UTF-8?q?log/guidance=20=E7=BB=B4=E6=8A=A4=EF=BC=88=E6=9C=80=E5=A4=9A=20?= =?UTF-8?q?2000=20=E5=AD=97=EF=BC=8C=E8=A1=A8=E5=90=8D=E6=81=B0=E5=A5=BD?= =?UTF-8?q?=E5=8F=AB=20guidance=20=E7=9A=84=E6=95=B4=E4=BB=BD=E6=8F=90?= =?UTF-8?q?=E4=BA=A4=E7=85=A7=E6=97=A7=EF=BC=89=EF=BC=8C=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=BA=90=E6=8E=A5=E5=8F=A3=E4=B8=8D=E6=98=BE=E7=A4=BA=E3=80=81?= =?UTF-8?q?=E6=95=B4=E4=BB=BD=E5=9B=9E=E5=86=99=20options=20=E6=97=B6?= =?UTF-8?q?=E7=95=99=E7=9D=80=EF=BC=8C=E6=8B=BC=E8=BF=9E=E6=8E=A5=E4=B8=B2?= =?UTF-8?q?=E6=97=B6=E6=91=98=E6=8E=89=EF=BC=9B=E6=9C=AC=E6=AC=A1=E8=AF=B4?= =?UTF-8?q?=E6=98=8E=E6=9C=80=E5=A4=9A=201000=20=E5=AD=97=EF=BC=8C?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E8=AE=B0=E4=B8=8B=E8=B5=B7=E8=8D=89=E8=A6=81?= =?UTF-8?q?=E6=B1=82=E5=92=8C=E6=9C=AC=E6=AC=A1=E8=AF=B4=E6=98=8E=E7=9A=84?= =?UTF-8?q?=E5=BF=AB=E7=85=A7=EF=BC=8C=E5=90=8C=E6=AD=A5=E8=B5=B7=E8=8D=89?= =?UTF-8?q?=E4=B9=9F=E5=B8=A6=E4=B8=8A=E6=95=B0=E6=8D=AE=E6=BA=90=E7=9A=84?= =?UTF-8?q?=E8=B5=B7=E8=8D=89=E8=A6=81=E6=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/catalog.py | 211 +++++++- backend/app/api/coded.py | 5 + backend/app/api/datasources.py | 26 +- backend/app/data/catalog_jobs.py | 639 +++++++++++++++++++++++ backend/app/data/engine.py | 8 +- backend/app/db/models.py | 72 ++- backend/app/main.py | 12 +- backend/tests/test_catalog_draft_jobs.py | 526 +++++++++++++++++++ 8 files changed, 1478 insertions(+), 21 deletions(-) create mode 100644 backend/app/data/catalog_jobs.py create mode 100644 backend/tests/test_catalog_draft_jobs.py diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index d52ff971..46362b9b 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -7,11 +7,15 @@ - POST /catalog/{table}/patch/preview 目录修改提案的预览（只读）：对着当前目录算改前、改后 - POST /catalog/{table}/patch 保存目录修改提案（乐观锁），记为人工确认 - GET /catalog/{table}/impact 影响面：引用这张表的已发布、受管模板（data/catalog_impact.py） -- POST /catalog/draft 同步起草（注释、外键、命名推断，可选模型） +- POST /catalog/draft 同步起草（注释、外键、命名推断，可选模型）。界面改用下面的后台起草任务，这个留着 +- PUT /catalog/guidance 数据源级的起草要求（使用者补充的背景和要求，起草时交给模型） +- POST /catalog/draft-jobs 建后台起草任务（202）；GET /draft-jobs/active、/latest、/{job_id} 读任务； + POST /draft-jobs/{job_id}/cancel 停止，/retry 只重试没写好的表（data/catalog_jobs.py） - POST /catalog/profile 同步剖析：对业务库发少量只读查询，核对关系、取码值候选（data/catalog_profile.py） 规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径 不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。 +起草任务的 409 带机读码（api/coded.py）：同一个源已有排队或进行中的任务、起草模型用不了、要停止的任务已经结束。 上传表格的源按当前快照的表结构回答（table_versions.resolve_source），和查询实际用的库是同一个版本。 """ @@ -19,13 +23,17 @@ from __future__ import annotations from typing import Any, Literal -from fastapi import APIRouter, Depends, Header, HTTPException -from pydantic import BaseModel, Field +from fastapi import APIRouter, Body, Depends, Header, HTTPException +from fastapi.exceptions import RequestValidationError +from pydantic import BaseModel, Field, ValidationError from sqlalchemy.ext.asyncio import AsyncSession -from app.api.coded import PROFILE_BUSY, SCHEMA_MISSING, SNAPSHOT_TAMPERED, CodedHTTPException +from app.api.coded import ( + CATALOG_DRAFT_BUSY, CATALOG_DRAFT_FINISHED, CATALOG_DRAFT_MODEL_UNAVAILABLE, PROFILE_BUSY, SCHEMA_MISSING, + SNAPSHOT_TAMPERED, CodedHTTPException, +) from app.api.runs import actor_of -from app.data import catalog, catalog_impact, catalog_profile, introspect, table_versions +from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, introspect, table_versions from app.data.engine import SnapshotTampered from app.db.base import get_session from app.db.models import DataSource @@ -64,10 +72,25 @@ class CatalogPatchIn(CatalogPatchPreviewIn): class CatalogDraftIn(BaseModel): #: 要起草的表；不给或为空时按使用次数取前 20 张 tables: list[str] | None = Field(default=None, max_length=200) - #: 是否请助手的模型起草中文名、粒度、列的含义等 + #: 是否请起草模型（设置里的起草模型，没设沿用助手的模型）写中文名、粒度、列的含义等 use_model: bool = False +class DraftJobIn(BaseModel): + #: 要起草的表；不给或为空时按使用次数取前 20 张。最多 500 张（超了 422，说明是中文） + tables: list[str] | None = None + #: 是否请起草模型写中文名、粒度、列的含义等 + use_model: bool = False + #: 这一次的补充说明，和数据源级的起草要求一起交给模型（最多 1000 字） + note: str | None = None + + +class GuidanceIn(BaseModel): + #: 数据源级的起草要求；空串（或 null）清除。最多 2000 字（超了 422，说明是中文）。必须写这一项：表名恰好叫 + #: guidance 时，整份提交那张表目录的请求也落到这个路径上，不能把它当成「清除起草要求」 + guidance: str | None + + class CatalogProfileIn(BaseModel): #: 要剖析的表；不给或为空时按使用次数取前 10 张目录里有待核实关系的表 tables: list[str] | None = Field(default=None, max_length=200) @@ -100,7 +123,7 @@ def _item_value(notes: dict[str, Any], name: str) -> tuple[Any, str | None]: def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogEntry | None, - usage: int) -> dict[str, Any]: + usage: int, last_draft: dict[str, Any] | None = None) -> dict[str, Any]: notes = entry.notes if entry else {} label, label_status = _item_value(notes, "label") kind, _ = _item_value(notes, "kind") @@ -120,6 +143,8 @@ def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogE "version": entry.version if entry else 0, "updated_at": _iso(entry.updated_at) if entry else None, "updated_by": entry.updated_by if entry else None, + # 这张表最近一次起草任务里的那一项：{status, job_id, finished_at, error, model_error}，没起草过是 null + "last_draft": last_draft, } @@ -180,15 +205,18 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi 说明有没有截断，schema_total 是数据库里一共几张表（含视图），界面据此说明「只探查了前 200 张」。老缓存没记 总数时按探查到的张数算。 """ - _, source = await _resolved(session, source_id) + row, source = await _resolved(session, source_id) cache = source.schema_cache or {} tables = cache.get("tables") or {} entries = await catalog.read_catalog(session, source.id) usage = await catalog.table_usage(session, source) + drafts = await catalog_jobs.last_drafts(session, source.id) order = {name: i for i, name in enumerate(tables)} - rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0)) for name, meta in tables.items()] + rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0), drafts.get(name)) + for name, meta in tables.items()] rows.sort(key=lambda r: (-r["usage"], order[r["table_name"]])) - rows += [_summary_row(name, None, entry, 0) for name, entry in sorted(entries.items()) if name not in tables] + rows += [_summary_row(name, None, entry, 0, drafts.get(name)) + for name, entry in sorted(entries.items()) if name not in tables] return { "tables": rows, "system_notes": catalog.system_notes_source(source), @@ -196,6 +224,10 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi "schema_note": None if tables else introspect.why_empty(cache), "schema_truncated": bool(cache.get("truncated")), "schema_total": max(_int(cache.get("total")), len(tables)), + # 数据源级的起草要求（没有是 null） + "guidance": catalog_jobs.guidance_of(row), + # 排队或进行中的起草任务 {id, status, total, done}，界面据此显示进度；没有是 null + "active_job": await catalog_jobs.job_brief(session, await catalog_jobs.active_job(session, source.id)), } @@ -207,6 +239,155 @@ async def get_catalog_table(source_id: str, table: str, return await _detail(session, source, key, entry) +async def _source_row(session: AsyncSession, source_id: str) -> DataSource: + row = await session.get(DataSource, source_id) + if row is None: + raise HTTPException(404, "数据源不存在，可能已被删除") + return row + + +@router.put("/{source_id}/catalog/guidance") +async def put_catalog_guidance(source_id: str, body: dict[str, Any] = Body(...), + x_actor: str | None = Header(default=None), + session: AsyncSession = Depends(get_session)) -> Any: + """存数据源级的起草要求：去首尾空白，空串清除，最多 2000 字（超了 422）。上传表格的源存在它自己的数据源记录上。 + + 要登记在 PUT /catalog/{table} 前面：不然「guidance」会被当成表名。反过来，表名恰好叫 guidance 时，整份提交那张表 + 目录的请求（{notes, if_version}，没有 guidance 这一项）也落到这里：照旧当作那张表的整份提交。""" + if "guidance" not in body and "notes" in body: + try: + table_payload = CatalogPutIn.model_validate(body) + except ValidationError as e: + raise RequestValidationError(e.errors()) from e + return await put_catalog_table(source_id, "guidance", table_payload, x_actor=x_actor, session=session) + try: + payload = GuidanceIn.model_validate(body) + except ValidationError as e: + raise RequestValidationError(e.errors()) from e + row = await _source_row(session, source_id) + try: + value = await catalog_jobs.save_guidance(session, row, payload.guidance) + except catalog_jobs.TextTooLong as e: + raise HTTPException(422, str(e)) from e + return {"guidance": value} + + +async def _ensure_model(session: AsyncSession) -> None: + """起草模型用不了（没有接入、停用、只有演示接入……）：409，不建任务。说法同同步起草时的 model_error。""" + try: + await catalog_jobs.check_model(session) + except catalog.CatalogModelUnavailable as e: + raise CodedHTTPException(409, str(e), CATALOG_DRAFT_MODEL_UNAVAILABLE) from e + + +def _no_schema(cache: dict[str, Any]) -> HTTPException: + return HTTPException(409, f"{introspect.why_empty(cache)}，无法起草数据目录。请先在数据源卡片上点「探查结构」") + + +@router.post("/{source_id}/catalog/draft-jobs", status_code=202) +async def create_draft_job(source_id: str, payload: DraftJobIn, x_actor: str | None = Header(default=None), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """建后台起草任务，马上返回（202，任务是排队中）。同一个源已有排队或进行中的任务 409 catalog_draft_busy（界面改去 + GET /draft-jobs/active 显示那一个）；用模型而起草模型用不了 409 catalog_draft_model_unavailable；没有表结构 409。 + 表结构里没有的表照样建进任务、状态 skipped。""" + if payload.tables and len(payload.tables) > catalog_jobs.MAX_TABLES: + raise HTTPException(422, f"一次最多起草 {catalog_jobs.MAX_TABLES} 张表，当前 {len(payload.tables)} 张，" + "请分几次起草") + try: + note = catalog_jobs.clean_text(payload.note, catalog_jobs.NOTE_MAX, "本次说明") + except catalog_jobs.TextTooLong as e: + raise HTTPException(422, str(e)) from e + row, source = await _resolved(session, source_id) + try: + await catalog_jobs.ensure_idle(session, row) + except catalog_jobs.DraftBusy as e: + raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e + cache = source.schema_cache or {} + if not cache.get("tables"): + raise _no_schema(cache) + if payload.use_model: + await _ensure_model(session) + try: + job, tasks = await catalog_jobs.create_job(session, row, source, tables=payload.tables or None, + use_model=payload.use_model, note=note, actor=actor_of(x_actor)) + except catalog_jobs.DraftBusy as e: + raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e + catalog_jobs.manager.start(job.id) + return {"job": catalog_jobs.job_dict(job, tasks)} + + +@router.get("/{source_id}/catalog/draft-jobs/active") +async def active_draft_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """排队或进行中的那个起草任务，没有是 null。""" + row = await _source_row(session, source_id) + return {"job": await catalog_jobs.job_out(session, await catalog_jobs.active_job(session, row.id))} + + +@router.get("/{source_id}/catalog/draft-jobs/latest") +async def latest_draft_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """最近建的一个起草任务（任何状态），没有是 null。""" + row = await _source_row(session, source_id) + return {"job": await catalog_jobs.job_out(session, await catalog_jobs.latest_job(session, row.id))} + + +async def _job_or_404(session: AsyncSession, source_id: str, job_id: str) -> tuple[DataSource, Any]: + row = await _source_row(session, source_id) + job = await catalog_jobs.load(session, row.id, job_id) + if job is None: + raise HTTPException(404, "起草任务不存在，可能属于别的数据源，或者已随数据源删除") + return row, job + + +@router.get("/{source_id}/catalog/draft-jobs/{job_id}") +async def get_draft_job(source_id: str, job_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + _, job = await _job_or_404(session, source_id, job_id) + return {"job": await catalog_jobs.job_out(session, job)} + + +@router.post("/{source_id}/catalog/draft-jobs/{job_id}/cancel") +async def cancel_draft_job(source_id: str, job_id: str, + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """停止：排队的直接停止；进行中的停掉正在进行的模型调用，没开始和没写完的表记停止，已经写进目录的不回滚。 + 已经结束的 409 catalog_draft_finished。返回停止之后的任务。""" + row, job = await _job_or_404(session, source_id, job_id) + owner = row.id + try: + await catalog_jobs.cancel(session, job) + except catalog_jobs.DraftFinished as e: + raise CodedHTTPException(409, str(e), CATALOG_DRAFT_FINISHED) from e + session.expire_all() # 后台在别的会话里收的尾：重新读 + return {"job": await catalog_jobs.job_out(session, await catalog_jobs.load(session, owner, job_id))} + + +@router.post("/{source_id}/catalog/draft-jobs/{job_id}/retry", status_code=202) +async def retry_draft_job(source_id: str, job_id: str, x_actor: str | None = Header(default=None), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """重试：新任务，只含原任务里没写好（failed、partial）和停止了（cancelled）的表；用不用模型、本次说明沿用，起草要求 + 取数据源当前的。有排队或进行中的任务 409 catalog_draft_busy；没有可重试的表 422；起草模型用不了 409。""" + _, old = await _job_or_404(session, source_id, job_id) + row, source = await _resolved(session, source_id) + try: + await catalog_jobs.ensure_idle(session, row) + except catalog_jobs.DraftBusy as e: + raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e + try: + tables = await catalog_jobs.retry_tables(session, old) + except catalog_jobs.NothingToRetry as e: + raise HTTPException(422, str(e)) from e + cache = source.schema_cache or {} + if not cache.get("tables"): + raise _no_schema(cache) + if old.use_model: + await _ensure_model(session) + try: + job, tasks = await catalog_jobs.create_job(session, row, source, tables=tables, use_model=old.use_model, + note=old.note, actor=actor_of(x_actor), retry_of=old.id) + except catalog_jobs.DraftBusy as e: + raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e + catalog_jobs.manager.start(job.id) + return {"job": catalog_jobs.job_dict(job, tasks)} + + @router.put("/{source_id}/catalog/{table}") async def put_catalog_table(source_id: str, table: str, payload: CatalogPutIn, x_actor: str | None = Header(default=None), @@ -228,17 +409,19 @@ async def put_catalog_table(source_id: str, table: str, payload: CatalogPutIn, async def draft_catalog(source_id: str, payload: CatalogDraftIn, x_actor: str | None = Header(default=None), session: AsyncSession = Depends(get_session)) -> dict[str, Any]: - """同步起草。返回每张表新增、更新、删除了几项；模型用不了或某张表的模型起草失败，照实写在结果里。 + """同步起草（界面改用后台起草任务，这个接口留着，返回形状不变）。引擎同后台任务（分段、按失败原因处理），起草要求 + 取数据源级的那份。返回每张表新增、更新、删除了几项；模型用不了或某张表的模型起草失败，照实写在结果里。 model_items：模型给了这张表几项可用内容（没用模型、模型这部分失败时为 null）。界面据此区分「模型给了内容、 只是和现有目录一致」（model_items > 0、三个数都是 0）和「模型没给内容」（model_error 有值）。 """ - _, source = await _resolved(session, source_id) + row, source = await _resolved(session, source_id) cache = source.schema_cache or {} if not cache.get("tables"): - raise HTTPException(409, f"{introspect.why_empty(cache)}，无法起草数据目录。请先在数据源卡片上点「探查结构」") + raise _no_schema(cache) report = await catalog.draft_catalog(session, source, tables=payload.tables or None, - use_model=payload.use_model, actor=actor_of(x_actor)) + use_model=payload.use_model, actor=actor_of(x_actor), + guidance=catalog_jobs.guidance_of(row)) rows = [{"table_name": r.table, "added": r.added, "updated": r.updated, "removed": r.removed, "version": r.version, "error": r.error, "model_error": r.model_error, "model_items": r.model_items} for r in report.tables] diff --git a/backend/app/api/coded.py b/backend/app/api/coded.py index 4cf6da90..204b94af 100644 --- a/backend/app/api/coded.py +++ b/backend/app/api/coded.py @@ -19,6 +19,11 @@ DATASOURCE_INACTIVE = "datasource_inactive" SCHEMA_MISSING = "schema_missing" PROFILE_BUSY = "profile_busy" SNAPSHOT_TAMPERED = "snapshot_tampered" +#: 数据目录的后台起草任务（api/catalog.py 的 /catalog/draft-jobs）：同一个源已有排队或进行中的任务（409，界面改去 +#: GET /draft-jobs/active 显示那一个）、起草模型用不了（409，不建任务）、要停止的任务已经结束了（409） +CATALOG_DRAFT_BUSY = "catalog_draft_busy" +CATALOG_DRAFT_MODEL_UNAVAILABLE = "catalog_draft_model_unavailable" +CATALOG_DRAFT_FINISHED = "catalog_draft_finished" class CodedHTTPException(HTTPException): diff --git a/backend/app/api/datasources.py b/backend/app/api/datasources.py index c4be16a8..f2a9e008 100644 --- a/backend/app/api/datasources.py +++ b/backend/app/api/datasources.py @@ -27,8 +27,8 @@ from app.data.catalog_profile import profile_settings_rejection from app.data import introspect as introspect_mod from app.data import table_versions from app.data.engine import ( - MASK_COLUMNS_OPTION, MAX_QUERY_TIMEOUT_S, QUERY_TIMEOUT_OPTION, SUPPORTED_KINDS, SnapshotTampered, build_url, - engine_args, engines, mask_columns_problem, query_timeout_problem, + CATALOG_DRAFT_GUIDANCE_OPTION, MASK_COLUMNS_OPTION, MAX_QUERY_TIMEOUT_S, QUERY_TIMEOUT_OPTION, SUPPORTED_KINDS, + SnapshotTampered, build_url, engine_args, engines, mask_columns_problem, query_timeout_problem, ) from app.db.base import get_session, new_id from app.db.models import DataSource, ImportStaging, SourceSnapshot, TableImport, TableRecipe @@ -124,6 +124,22 @@ def _import_mode(row: DataSource) -> str | None: return "simple" if _is_upload(row) else None +#: options 里由别的接口维护的项：数据源表单不显示、不提交它们，整份回写 options 时原样留着（同模型接入 extra 里的 +#: last_check）。数据目录的起草要求只经 PUT /catalog/guidance 改（api/catalog.py）：表单把认不得的项放进「高级参数」 +#: 的单行输入框，多行的要求会被压成一行 +_MANAGED_OPTIONS = (CATALOG_DRAFT_GUIDANCE_OPTION,) + + +def _shown_options(options: dict[str, Any] | None) -> dict[str, Any]: + return {k: v for k, v in (options or {}).items() if k not in _MANAGED_OPTIONS} + + +def _with_managed(options: dict[str, Any], saved: dict[str, Any] | None) -> dict[str, Any]: + """表单提交的 options + 存着的、由别的接口维护的项（提交里带了也不认）。""" + kept = {k: v for k, v in (saved or {}).items() if k in _MANAGED_OPTIONS} + return {**_shown_options(options), **kept} + + def _to_out(row: DataSource, snapshot: dict[str, Any] | None = None, recipe_info: dict[str, Any] | None = None) -> DataSourceOut: cache = row.schema_cache or {} @@ -137,7 +153,7 @@ def _to_out(row: DataSource, snapshot: dict[str, Any] | None = None, schema_error=str(cache.get("error") or "") if cache.get("failed") else "", available_schemas=list(cache.get("available_schemas") or []), id=row.id, name=row.name, kind=row.kind, host=row.host, port=row.port, - database=row.database, username=row.username, options=row.options or {}, + database=row.database, username=row.username, options=_shown_options(row.options), readonly=row.readonly, description=row.description, enabled=row.enabled, password_masked=mask(row.password), has_password=bool(row.password), @@ -429,7 +445,8 @@ async def create_source( _refuse_reserved(payload.kind, payload.database) row = DataSource( - **payload.model_dump(exclude={"password"}), + **payload.model_dump(exclude={"password", "options"}), + options=_with_managed(payload.options, None), password=encrypt(payload.password), ) _settle_oracle(row, None, None) @@ -472,6 +489,7 @@ async def update_source( raise HTTPException(422, _UPLOAD_DIR_READONLY) if "options" in data: _refuse_bad_options(data["options"]) + data["options"] = _with_managed(data["options"] or {}, row.options) if "password" in data: # 空串表示清空，非空表示换新的 data["password"] = encrypt(data["password"]) if data["password"] else None diff --git a/backend/app/data/catalog_jobs.py b/backend/app/data/catalog_jobs.py new file mode 100644 index 00000000..4d1d244f --- /dev/null +++ b/backend/app/data/catalog_jobs.py @@ -0,0 +1,639 @@ +"""数据目录的后台起草任务，以及数据源级的起草要求。 + +以前起草是一次同步长请求（POST /catalog/draft，界面每 12 张表一个请求）：一个请求出错整个对话框就停，关掉对话框 +之后没有任何记录哪些表失败、为什么。现在： + +- **建任务**（create_job）：同一个数据源同时只有一个排队或进行中的任务（DraftBusy；部分唯一索引兜底）。不指定表时 + 按使用次数取前 20 张；表结构里没有的表照样建进任务、状态 skipped。起草要求（数据源级）和本次说明随任务记下快照。 +- **在后台起草**（manager.start）：进程内的 asyncio 任务，引擎是 data/catalog_draft.py（分段、按失败原因处理）。 + 每张表一结束就并入目录（catalog.write_draft，规则和同步起草一样：人工确认、驳回过的不动，乐观锁冲突重读重并）； + 逐表的状态、到第几段、计数、用量随时落库，接口读的是库里的状态。 +- **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没开始和没写完的表记 cancelled，已经写进 + 目录的不回滚，任务最后是 cancelled。 +- **重试**（retry_tables + create_job）：新任务，只含原任务里 failed、partial、cancelled 的表；用不用模型、本次说明沿用， + 起草要求取数据源当前的。 +- **服务重启**：启动时（startup）把排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明「服务重启，起草 + 中断」。不自动续跑：复制出来的数据目录（检查、验收）启动时不能自己去调模型；界面让人点「重试」续上。关停时 + （manager.shutdown）停掉进行中的任务，同样标成 interrupted。 + +任务的库写入一律经 _JobRun._write：一条一条来，shield 起来——停止任务时不会打断写到一半的库操作，收尾前等它们写完。 + +**起草要求**存在数据源 options 的 catalog_draft_guidance 里（上传表格的源存在它自己的 DataSource 行上），一段文字， +去首尾空白、最多 GUIDANCE_MAX 字，空串等于没有。只经 PUT /catalog/guidance 改：数据源接口不显示它、整份回写 options +时留着（api/datasources.py），拼连接串时摘掉（data/engine.py）。 +""" +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable, Callable +from typing import Any + +from sqlalchemy import func, select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.data import catalog, catalog_draft +from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION +from app.db.base import SessionLocal, new_id, utcnow +from app.db.models import CatalogDraftJob, CatalogDraftTask, DataSource + +logger = logging.getLogger(__name__) + +#: 任务还没结束的状态（同一个源只能有一个）、结束了的状态 +ACTIVE = ("queued", "running") +FINISHED = ("succeeded", "partial", "failed", "cancelled", "interrupted") +#: 逐表的状态，以及其中算「已结束」的、重试时要带上的 +TASK_STATUSES = ("pending", "running", "ok", "partial", "failed", "skipped", "cancelled") +TASK_DONE = ("ok", "partial", "failed", "skipped", "cancelled") +RETRYABLE = ("failed", "partial", "cancelled") +_OPEN = ("pending", "running") + +#: 一个任务最多几张表；起草要求、本次说明最多几个字 +MAX_TABLES = 500 +GUIDANCE_MAX = 2000 +NOTE_MAX = 1000 +#: 目录清单上「这张表最近一次起草」只看最近这么多个任务 +LAST_DRAFT_JOBS = 100 +#: 停止进行中的任务时最多等它收尾多久（秒）；关停服务时另给（shutdown 的 grace） +STOP_GRACE_S = 10.0 + +INTERRUPTED_ERROR = "服务重启，起草中断" +UNFINISHED_ERROR = "起草结束时这张表还没有写完，请重试这张表" +WRITE_ERROR = "写入数据目录时出错：{line}" +JOB_ERROR = "起草任务出错：{line}" +SOURCE_GONE = "数据源不存在，可能已被删除" + + +class DraftBusy(Exception): + """这个数据源已经有排队或进行中的起草任务。message 给人看。""" + + +class DraftFinished(Exception): + """要停止的任务已经结束了。""" + + +class NothingToRetry(Exception): + """原任务里没有需要重试的表。""" + + +class TextTooLong(ValueError): + """起草要求、本次说明超长。message 给人看。""" + + +class _JobFailed(Exception): + """任务整体出错（数据源没了、快照没了、模型用不了）：message 就是任务的 error。""" + + +# ========================================================================== +# 起草要求、本次说明 +# ========================================================================== + + +def clean_text(value: Any, limit: int, label: str) -> str | None: + """去首尾空白；空串（和不是字符串的）等于没有；超过 limit 个字抛 TextTooLong。""" + text = value.strip() if isinstance(value, str) else "" + if len(text) > limit: + raise TextTooLong(f"{label}最多 {limit} 字，当前 {len(text)} 字，请删减后再保存") + return text or None + + +def guidance_of(row: Any) -> str | None: + """数据源级的起草要求（没有是 None）。""" + raw = (getattr(row, "options", None) or {}).get(CATALOG_DRAFT_GUIDANCE_OPTION) + return raw.strip() or None if isinstance(raw, str) else None + + +async def save_guidance(session: AsyncSession, row: DataSource, value: Any) -> str | None: + """存起草要求，返回存下的（清除了是 None）。超长抛 TextTooLong，什么都不改。""" + text = clean_text(value, GUIDANCE_MAX, "起草要求") + options = {k: v for k, v in (row.options or {}).items() if k != CATALOG_DRAFT_GUIDANCE_OPTION} + if text: + options[CATALOG_DRAFT_GUIDANCE_OPTION] = text + row.options = options + await session.commit() + return text + + +# ========================================================================== +# 读任务 +# ========================================================================== + + +def _iso(value: Any) -> str | None: + return value.isoformat() if value is not None else None + + +async def active_job(session: AsyncSession, source_id: str) -> CatalogDraftJob | None: + """排队或进行中的那个（同一个源最多一个）。""" + return (await session.execute( + select(CatalogDraftJob).where(CatalogDraftJob.source_id == source_id, CatalogDraftJob.status.in_(ACTIVE)) + .order_by(CatalogDraftJob.created_at.desc()).limit(1) + )).scalar_one_or_none() + + +async def latest_job(session: AsyncSession, source_id: str) -> CatalogDraftJob | None: + """最近建的一个（任何状态）。""" + return (await session.execute( + select(CatalogDraftJob).where(CatalogDraftJob.source_id == source_id) + .order_by(CatalogDraftJob.created_at.desc()).limit(1) + )).scalar_one_or_none() + + +async def load(session: AsyncSession, source_id: str, job_id: str) -> CatalogDraftJob | None: + """这个源的这个任务；不存在、不属于这个源都是 None。""" + job = await session.get(CatalogDraftJob, job_id) + return job if job is not None and job.source_id == source_id else None + + +async def tasks_of(session: AsyncSession, job_id: str) -> list[CatalogDraftTask]: + return list((await session.execute( + select(CatalogDraftTask).where(CatalogDraftTask.job_id == job_id).order_by(CatalogDraftTask.seq) + )).scalars()) + + +def task_dict(task: CatalogDraftTask) -> dict[str, Any]: + return { + "table_name": task.table_name, + "status": task.status, + "columns": task.column_count, + "parts_total": task.parts_total, + "parts_done": task.parts_done, + "added": task.added, + "updated": task.updated, + "removed": task.removed, + "version": task.version, + "model_items": task.model_items, + "error": task.error, + "model_error": task.model_error, + "started_at": _iso(task.started_at), + "finished_at": _iso(task.finished_at), + } + + +def _counts(statuses: list[str]) -> dict[str, int]: + counts = dict.fromkeys(TASK_STATUSES, 0) + for status in statuses: + counts[status] = counts.get(status, 0) + 1 + return counts + + +def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask]) -> dict[str, Any]: + """接口返回的任务（CatalogDraftJob）：逐表的状态按起草顺序；计数、合计都按逐表的算。""" + counts = _counts([t.status for t in tasks]) + return { + "id": job.id, + "source_id": job.source_id, + "status": job.status, + "use_model": bool(job.use_model), + "model": job.model, + "guidance": job.guidance, + "note": job.note, + "created_at": _iso(job.created_at), + "started_at": _iso(job.started_at), + "finished_at": _iso(job.finished_at), + "created_by": job.created_by, + "retry_of": job.retry_of, + "total": len(tasks), + "done": sum(counts.get(s, 0) for s in TASK_DONE), + "counts": counts, + "added": sum(t.added or 0 for t in tasks), + "updated": sum(t.updated or 0 for t in tasks), + "removed": sum(t.removed or 0 for t in tasks), + "usage": {"calls": job.calls or 0, "input_tokens": job.input_tokens or 0, + "output_tokens": job.output_tokens or 0}, + "error": job.error, + "tasks": [task_dict(t) for t in tasks], + } + + +async def job_out(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None: + return None if job is None else job_dict(job, await tasks_of(session, job.id)) + + +async def job_brief(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None: + """目录清单顶上的进度：{id, status, total, done}。""" + if job is None: + return None + rows = (await session.execute( + select(CatalogDraftTask.status, func.count()).where(CatalogDraftTask.job_id == job.id) + .group_by(CatalogDraftTask.status) + )).all() + counts = {status: n for status, n in rows} + return {"id": job.id, "status": job.status, "total": sum(counts.values()), + "done": sum(counts.get(s, 0) for s in TASK_DONE)} + + +async def last_drafts(session: AsyncSession, source_id: str) -> dict[str, dict[str, Any]]: + """每张表最近一次起草任务里的那一项：{表名: {status, job_id, finished_at, error, model_error}}。任务还在进行， + 就是进行中的状态。只看最近 LAST_DRAFT_JOBS 个任务。""" + ids = list((await session.execute( + select(CatalogDraftJob.id).where(CatalogDraftJob.source_id == source_id) + .order_by(CatalogDraftJob.created_at.desc()).limit(LAST_DRAFT_JOBS) + )).scalars()) + if not ids: + return {} + rank = {job_id: i for i, job_id in enumerate(ids)} + rows = (await session.execute(select(CatalogDraftTask).where(CatalogDraftTask.job_id.in_(ids)))).scalars() + out: dict[str, dict[str, Any]] = {} + for task in sorted(rows, key=lambda t: (rank[t.job_id], t.seq)): + out.setdefault(task.table_name, {"status": task.status, "job_id": task.job_id, + "finished_at": _iso(task.finished_at), "error": task.error, + "model_error": task.model_error}) + return out + + +# ========================================================================== +# 建任务、重试、停止 +# ========================================================================== + + +def busy_message(row: Any) -> str: + return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录，请等这次起草结束（或先停止它）再开始新的起草" + + +async def ensure_idle(session: AsyncSession, row: Any) -> None: + """这个源没有排队或进行中的任务；有就抛 DraftBusy。""" + if await active_job(session, row.id) is not None: + raise DraftBusy(busy_message(row)) + + +async def check_model(session: AsyncSession) -> None: + """起草模型能不能用：不能用抛 catalog.CatalogModelUnavailable（只建模型对象，不发请求）。""" + await catalog.resolve_draft_model(session) + + +async def create_job(session: AsyncSession, row: Any, source: Any, *, tables: list[str] | None, use_model: bool, + note: str | None, actor: str | None, + retry_of: str | None = None) -> tuple[CatalogDraftJob, list[CatalogDraftTask]]: + """建一个排队的任务（还没开始，调用方接着 manager.start）。row 是数据源记录（起草要求取它的），source 是读表结构 + 用的源（上传源是绑定当前快照的视图）。有排队或进行中的任务抛 DraftBusy。""" + await ensure_idle(session, row) + cache = getattr(source, "schema_cache", None) or {} + all_tables = cache.get("tables") or {} + wanted = list(tables) if tables else await catalog.default_tables(session, source) + job = CatalogDraftJob(id=new_id(), source_id=row.id, status="queued", use_model=bool(use_model), + guidance=guidance_of(row), note=note, created_by=actor, retry_of=retry_of) + tasks: list[CatalogDraftTask] = [] + seen: set[str] = set() + now = utcnow() + for name in dict.fromkeys(str(t) for t in wanted): + key = catalog.resolve_table_name(cache, name) + if key is None: + tasks.append(CatalogDraftTask(job_id=job.id, seq=len(tasks), table_name=name, status="skipped", + error=catalog.missing_table_error(name), finished_at=now)) + continue + if key in seen: + continue + seen.add(key) + columns = len((all_tables.get(key) or {}).get("columns") or []) + tasks.append(CatalogDraftTask(job_id=job.id, seq=len(tasks), table_name=key, status="pending", + column_count=columns, + parts_total=catalog_draft.parts_for(columns) if use_model else None)) + session.add(job) + try: + await session.flush() + session.add_all(tasks) + await session.commit() + except IntegrityError as e: + # 部分唯一索引：另一个请求同时给这个源建了任务 + await session.rollback() + raise DraftBusy(busy_message(row)) from e + return job, tasks + + +async def retry_tables(session: AsyncSession, job: CatalogDraftJob) -> list[str]: + """重试要带上的表：原任务里 failed、partial、cancelled 的，按原来的顺序。没有抛 NothingToRetry。""" + names = [t.table_name for t in await tasks_of(session, job.id) if t.status in RETRYABLE] + if not names: + raise NothingToRetry("这次起草里没有失败、只写出一部分或停止了的表，不需要重试") + return names + + +async def cancel(session: AsyncSession, job: CatalogDraftJob) -> None: + """停止一个任务。排队的直接 cancelled；进行中的交给 manager 停（停掉模型调用、没写完的表记 cancelled）。 + 已经结束的抛 DraftFinished。""" + if job.status in FINISHED: + raise DraftFinished("这次起草已经结束了，不需要停止") + if job.status == "queued": + now = utcnow() + claimed = await session.execute( + update(CatalogDraftJob).where(CatalogDraftJob.id == job.id, CatalogDraftJob.status == "queued") + .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False)) + if claimed.rowcount == 1: + await session.execute( + update(CatalogDraftTask).where(CatalogDraftTask.job_id == job.id, CatalogDraftTask.status.in_(_OPEN)) + .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False)) + await session.commit() + return + await session.rollback() # 刚被后台认领、开始了：按进行中的停 + if await manager.stop(job.id, "user", grace=STOP_GRACE_S): + return + # 库里说在进行，这个进程里却没有在跑它（记录对不上）：直接记停止 + await _close(session, job.id, "cancelled", None, "cancelled", None) + + +async def _close(session: AsyncSession, job_id: str, job_status: str, job_error: str | None, task_status: str, + task_error: str | None) -> None: + """收尾：还没结束的表记 task_status，任务（还没结束的话）记 job_status。""" + now = utcnow() + await session.execute( + update(CatalogDraftTask).where(CatalogDraftTask.job_id == job_id, CatalogDraftTask.status.in_(_OPEN)) + .values(status=task_status, error=task_error, finished_at=now).execution_options(synchronize_session=False)) + await session.execute( + update(CatalogDraftJob).where(CatalogDraftJob.id == job_id, CatalogDraftJob.status.in_(ACTIVE)) + .values(status=job_status, error=job_error, finished_at=now).execution_options(synchronize_session=False)) + await session.commit() + + +async def startup() -> int: + """服务启动时：排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明服务重启。不自动续跑。返回标了几个。""" + async with SessionLocal() as session: + ids = [job_id for job_id in (await session.execute( + select(CatalogDraftJob.id).where(CatalogDraftJob.status.in_(ACTIVE)))).scalars() + if not manager.running(job_id)] + for job_id in ids: + await _close(session, job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR) + if ids: + logger.warning("服务重启：%d 个数据目录起草任务没有做完，已标为中断，可在数据目录页重试", len(ids)) + return len(ids) + + +# ========================================================================== +# 在后台起草 +# ========================================================================== + + +class _JobRun: + """一个任务在后台的一次执行。""" + + def __init__(self, job_id: str) -> None: + self.job_id = job_id + self.task: asyncio.Task[None] | None = None + #: 为什么停：user 使用者点了停止，shutdown 服务关停（标中断） + self.stop_reason: str | None = None + self._writes: set[asyncio.Task[Any]] = set() + self._lock = asyncio.Lock() + self._started: set[str] = set() + self.source: Any = None + self.actor: str | None = None + self.at = "" + + # ---------------------------------------------------------------- 库写入 + + async def _write(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any: + """一条库写入：同一个任务的写入一条一条来；shield 起来，停止任务时不会打断写到一半的库操作。""" + async def go() -> Any: + async with self._lock: + async with SessionLocal() as session: + return await fn(session, *args) + + task = asyncio.ensure_future(go()) + self._writes.add(task) + task.add_done_callback(self._writes.discard) + return await asyncio.shield(task) + + async def _settle(self) -> None: + """等手上的库写入都写完（停止、出错收尾之前）。""" + while self._writes: + await asyncio.gather(*list(self._writes), return_exceptions=True) + + # ---------------------------------------------------------------- 主流程 + + async def run(self) -> None: + try: + await self._run() + except asyncio.CancelledError: + current = asyncio.current_task() + if current is not None: + current.uncancel() + await self._settle() + if self.stop_reason == "user": + await self._write(_close, self.job_id, "cancelled", None, "cancelled", None) + else: + await self._write(_close, self.job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR) + except _JobFailed as e: + await self._settle() + await self._write(_close, self.job_id, "failed", str(e), "failed", str(e)) + except Exception as e: # noqa: BLE001 - 任务整体出错：照实记下，不留在「起草中」 + from app.core.errors import describe_exception, first_line + + logger.exception("数据目录起草任务 %s 出错", self.job_id) + await self._settle() + reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e)) + await self._write(_close, self.job_id, "failed", reason, "failed", reason) + + async def _run(self) -> None: + if not await self._write(self._claim): + return # 开始之前就被停止了 + async with SessionLocal() as session: + job = await session.get(CatalogDraftJob, self.job_id) + row = await session.get(DataSource, job.source_id) if job is not None else None + if job is None or row is None: + raise _JobFailed(SOURCE_GONE) + from app.data import table_versions + + try: + source = await table_versions.resolve_source(session, row) + except table_versions.SnapshotMissing as e: + raise _JobFailed(str(e)) from e + pending = [t.table_name for t in await tasks_of(session, job.id) if t.status == "pending"] + use_model, background = bool(job.use_model), catalog_draft.Background(guidance=job.guidance, note=job.note) + self.actor = job.created_by + model: Any = None + existing: dict[str, Any] = {} + if use_model and pending: + try: + model, model_id = await catalog.resolve_draft_model(session) + except catalog.CatalogModelUnavailable as e: + raise _JobFailed(str(e)) from e + existing = {t: e.notes for t, e in (await catalog.read_catalog(session, source.id)).items()} + self.source, self.at = source, catalog.now_iso() + all_tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {} + names = [n for n in pending if isinstance(all_tables.get(n), dict)] + gone = [n for n in pending if n not in names] + if gone: + # 建任务之后表结构变了（重新探查过）：这几张表跳过 + await self._write(self._skip, gone) + if use_model and names: + await self._write(self._set_model, model_id) + hooks = catalog_draft.Hooks(call_started=self._call_started, usage=self._usage, + part_done=self._part_done, table_done=self._table_done) + await catalog_draft.run_model_draft(model, source, names, existing=existing, background=background, + hooks=hooks) + else: + for name in names: + await self._call_started([name]) + await self._table_done(name, None) + await self._settle() + await self._write(self._finish) + + # ---------------------------------------------------------------- 进度回调（引擎调） + + async def _call_started(self, tables: list[str]) -> None: + fresh = [t for t in dict.fromkeys(tables) if t not in self._started] + if fresh: + self._started.update(fresh) + await self._write(self._mark_running, fresh) + + async def _usage(self, input_tokens: int, output_tokens: int) -> None: + await self._write(self._add_usage, input_tokens, output_tokens) + + async def _part_done(self, table: str) -> None: + await self._write(self._bump_part, table) + + async def _table_done(self, table: str, outcome: catalog_draft.ModelOutcome | None) -> None: + await self._write(self._write_table, table, outcome) + + # ---------------------------------------------------------------- 库操作 + + def _task(self, *where: Any) -> Any: + return update(CatalogDraftTask).where(CatalogDraftTask.job_id == self.job_id, *where) \ + .execution_options(synchronize_session=False) + + async def _claim(self, session: AsyncSession) -> bool: + result = await session.execute( + update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id, CatalogDraftJob.status == "queued") + .values(status="running", started_at=utcnow()).execution_options(synchronize_session=False)) + await session.commit() + return result.rowcount == 1 + + async def _set_model(self, session: AsyncSession, model_id: str) -> None: + await session.execute(update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id) + .values(model=model_id or None).execution_options(synchronize_session=False)) + await session.commit() + + async def _skip(self, session: AsyncSession, names: list[str]) -> None: + now = utcnow() + for name in names: + await session.execute(self._task(CatalogDraftTask.table_name == name, CatalogDraftTask.status == "pending") + .values(status="skipped", error=catalog.missing_table_error(name), finished_at=now)) + await session.commit() + + async def _mark_running(self, session: AsyncSession, names: list[str]) -> None: + await session.execute(self._task(CatalogDraftTask.table_name.in_(names), CatalogDraftTask.status == "pending") + .values(status="running", started_at=utcnow())) + await session.commit() + + async def _add_usage(self, session: AsyncSession, input_tokens: int, output_tokens: int) -> None: + await session.execute( + update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id) + .values(calls=CatalogDraftJob.calls + 1, input_tokens=CatalogDraftJob.input_tokens + int(input_tokens), + output_tokens=CatalogDraftJob.output_tokens + int(output_tokens)) + .execution_options(synchronize_session=False)) + await session.commit() + + async def _bump_part(self, session: AsyncSession, table: str) -> None: + await session.execute(self._task(CatalogDraftTask.table_name == table) + .values(parts_done=CatalogDraftTask.parts_done + 1)) + await session.commit() + + async def _write_table(self, session: AsyncSession, table: str, + outcome: catalog_draft.ModelOutcome | None) -> None: + """一张表结束：并入目录、写入，记下这张表的结果。""" + try: + result = await catalog.write_draft(session, self.source, table, outcome, at=self.at, actor=self.actor) + except Exception as e: # noqa: BLE001 - 这一张写不进去不影响别的表 + from app.core.errors import describe_exception, first_line + + logger.exception("起草任务 %s 写入表 %s 的数据目录时出错", self.job_id, table) + await session.rollback() + result = catalog.TableDraftResult(table=table, status="failed", + error=WRITE_ERROR.format(line=first_line(e) or describe_exception(e)), + model_error=outcome.error if outcome is not None else None) + values: dict[str, Any] = { + "status": result.status, "added": result.added, "updated": result.updated, "removed": result.removed, + "version": result.version, "model_items": result.model_items, "error": result.error, + "model_error": result.model_error, "finished_at": utcnow(), + } + if outcome is not None: + values["parts_done"] = outcome.parts_done + await session.execute(self._task(CatalogDraftTask.table_name == table, CatalogDraftTask.status.in_(_OPEN)) + .values(**values)) + await session.commit() + + async def _finish(self, session: AsyncSession) -> None: + """起草完：万一还有没收尾的表（进度回调写库失败之类）记失败；有 failed、partial、cancelled 的表是 partial， + 否则 succeeded。""" + now = utcnow() + await session.execute(self._task(CatalogDraftTask.status.in_(_OPEN)) + .values(status="failed", error=UNFINISHED_ERROR, finished_at=now)) + statuses = list((await session.execute( + select(CatalogDraftTask.status).where(CatalogDraftTask.job_id == self.job_id))).scalars()) + status = "partial" if any(s in RETRYABLE for s in statuses) else "succeeded" + await session.execute( + update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id, CatalogDraftJob.status == "running") + .values(status=status, finished_at=now).execution_options(synchronize_session=False)) + await session.commit() + + +class DraftJobs: + """进程内的起草任务（照 app/scheduler 的写法：一个 asyncio 任务一件事，停的时候给宽限）。""" + + def __init__(self) -> None: + self._runs: dict[str, _JobRun] = {} + + def _alive(self, job_id: str) -> _JobRun | None: + run = self._runs.get(job_id) + if run is None or run.task is None or run.task.done(): + return None + try: + if run.task.get_loop() is not asyncio.get_running_loop(): + self._runs.pop(job_id, None) # 别的事件循环留下的（测试每个用例一个循环） + return None + except RuntimeError: + return None + return run + + def start(self, job_id: str) -> None: + """开始在后台起草这个（排队中的）任务。""" + run = _JobRun(job_id) + run.task = asyncio.get_running_loop().create_task(run.run(), name=f"catalog-draft-{job_id}") + self._runs[job_id] = run + + def forget(_: asyncio.Task[None]) -> None: + if self._runs.get(job_id) is run: + self._runs.pop(job_id, None) + + run.task.add_done_callback(forget) + + def running(self, job_id: str) -> bool: + return self._alive(job_id) is not None + + async def stop(self, job_id: str, reason: str, *, grace: float = STOP_GRACE_S) -> bool: + """停掉这个进程里在跑的任务，等它收尾（最多 grace 秒）。不在这个进程里跑返回 False。""" + run = self._alive(job_id) + if run is None: + return False + if run.stop_reason is None: + run.stop_reason = reason + run.task.cancel() # type: ignore[union-attr] + await asyncio.wait([run.task], timeout=grace) # type: ignore[list-item] + return True + + async def wait(self, job_id: str, timeout: float | None = None) -> None: + """等这个任务做完（测试、调试用）。不在跑直接返回。""" + run = self._alive(job_id) + if run is not None: + await asyncio.wait([run.task], timeout=timeout) # type: ignore[list-item] + + async def shutdown(self, *, grace: float = 5.0) -> None: + """关停：停掉进行中的任务（标中断），最多等 grace 秒收尾。""" + runs = [run for job_id in list(self._runs) if (run := self._alive(job_id)) is not None] + for run in runs: + if run.stop_reason is None: + run.stop_reason = "shutdown" + run.task.cancel() # type: ignore[union-attr] + if runs: + await asyncio.wait([run.task for run in runs], timeout=grace) # type: ignore[misc] + logger.info("已停止 %d 个数据目录起草任务", len(runs)) + + +manager = DraftJobs() + + +__all__ = ["ACTIVE", "DraftBusy", "DraftFinished", "DraftJobs", "FINISHED", "GUIDANCE_MAX", "INTERRUPTED_ERROR", + "MAX_TABLES", "NOTE_MAX", "NothingToRetry", "RETRYABLE", "TASK_DONE", "TASK_STATUSES", "TextTooLong", + "active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_idle", "guidance_of", "job_brief", + "job_dict", "job_out", "last_drafts", "latest_job", "load", "manager", "retry_tables", "save_guidance", + "startup", "tasks_of"] diff --git a/backend/app/data/engine.py b/backend/app/data/engine.py index 0f4cbf3d..59315118 100644 --- a/backend/app/data/engine.py +++ b/backend/app/data/engine.py @@ -62,12 +62,18 @@ MASK_COLUMNS_OPTION = "mask_columns" #: data/catalog_profile.py；放在这里只为让拼连接串时摘掉它 CATALOG_PROFILE_OPTION = "catalog_profile" +#: 数据源 options 里的数据目录起草要求：使用者补充的背景和要求，起草时交给模型（一段文字，最多 2000 字）。 +#: 只经 PUT /api/datasources/{id}/catalog/guidance 维护，见 data/catalog_jobs.py；放在这里只为让拼连接串时摘掉它 +CATALOG_DRAFT_GUIDANCE_OPTION = "catalog_draft_guidance" + # options 里这些 key 是 AgentLab 自己的配置，不是驱动参数，拼 URL 时要摘掉。 # schema：探查哪个 schema（企业库里只读账号名下常常什么都没有，数据在别处） # query_timeout_s：查询时限，由数据层按语句下发给数据库，见 _server_deadline # mask_columns：证据面板遮罩的列，驱动不认识它，拼进连接串会被当成未知参数拒掉 # catalog_profile：数据剖析设置，同上；它还是个对象，拼进去连 URL 都不成形 -_NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION, CATALOG_PROFILE_OPTION}) +# catalog_draft_guidance：数据目录的起草要求，一段中文，拼进去连接直接失败 +_NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION, CATALOG_PROFILE_OPTION, + CATALOG_DRAFT_GUIDANCE_OPTION}) _MASK_SPLIT = re.compile(r"[,，、;；\n]+") diff --git a/backend/app/db/models.py b/backend/app/db/models.py index db3e6dac..932d02a7 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -4,7 +4,7 @@ from datetime import datetime from typing import Any from sqlalchemy import ( - Float, ForeignKey, Index, Integer, LargeBinary, String, Text, UniqueConstraint, + Float, ForeignKey, Index, Integer, LargeBinary, String, Text, UniqueConstraint, text, ) from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -348,6 +348,76 @@ class CatalogNote(Base, TimestampMixin): updated_by: Mapped[str | None] = mapped_column(String(100), default=None) +class CatalogDraftJob(Base, TimestampMixin): + """数据目录的一次后台起草任务（data/catalog_jobs.py）。逐表的状态在 catalog_draft_tasks。 + + 以前起草是一次同步长请求，出错整个对话框就停、关掉之后没有任何记录哪些表失败、为什么；现在建任务后在后台起草， + 进度（每张表到第几段、计数、用量）随时落库，接口读的是库里的状态。 + + 同一个数据源同时只有一个排队或进行中的任务：接口先查一遍，部分唯一索引兜底（两个请求同时建，后提交的撞上）。 + 服务重启时还没结束的标成 interrupted，不自动续跑。挂外键、级联删除：任务是数据源的附属记录，源删掉了留着没有意义。 + """ + + __tablename__ = "catalog_draft_jobs" + __table_args__ = ( + Index("ix_catalog_draft_jobs_source_created", "source_id", "created_at"), + Index("ux_catalog_draft_jobs_active", "source_id", unique=True, + sqlite_where=text("status IN ('queued', 'running')")), + ) + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) + # queued | running | succeeded | partial | failed | cancelled | interrupted + status: Mapped[str] = mapped_column(String(20), default="queued") + use_model: Mapped[bool] = mapped_column(default=False) + # 实际用的模型 id（用模型时开始起草后写上） + model: Mapped[str | None] = mapped_column(String(200), default=None) + # 建任务时的快照：数据源级的起草要求、这一次的补充说明 + guidance: Mapped[str | None] = mapped_column(Text, default=None) + note: Mapped[str | None] = mapped_column(Text, default=None) + # 请求头 X-Actor 的署名（自报、未认证） + created_by: Mapped[str | None] = mapped_column(String(100), default=None) + started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + # 用量：调用次数、输入和输出 token（拿不到用量的调用按 0 算） + calls: Mapped[int] = mapped_column(Integer, default=0) + input_tokens: Mapped[int] = mapped_column(Integer, default=0) + output_tokens: Mapped[int] = mapped_column(Integer, default=0) + # 任务整体出错的原因（逐表的原因在 catalog_draft_tasks） + error: Mapped[str | None] = mapped_column(Text, default=None) + # 重试建的任务记下是重试哪一个 + retry_of: Mapped[str | None] = mapped_column(String(32), default=None) + + +class CatalogDraftTask(Base): + """起草任务里的一张表。按 seq（起草顺序）排。状态：pending 排队、running 起草中、ok、partial 模型只写出一部分、 + failed、skipped 表结构里没有、cancelled（停止或服务重启）。""" + + __tablename__ = "catalog_draft_tasks" + __table_args__ = (UniqueConstraint("job_id", "seq"),) + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + job_id: Mapped[str] = mapped_column(ForeignKey("catalog_draft_jobs.id", ondelete="CASCADE"), index=True) + seq: Mapped[int] = mapped_column(Integer) + # 表结构里的键；表结构里没有的（skipped）是请求里的原样名字 + table_name: Mapped[str] = mapped_column(String(255)) + status: Mapped[str] = mapped_column(String(20), default="pending") + column_count: Mapped[int] = mapped_column(Integer, default=0) + # 模型分几段起草（不用模型时为 NULL）、已经结束了几段 + parts_total: Mapped[int | None] = mapped_column(Integer, default=None) + parts_done: Mapped[int] = mapped_column(Integer, default=0) + added: Mapped[int] = mapped_column(Integer, default=0) + updated: Mapped[int] = mapped_column(Integer, default=0) + removed: Mapped[int] = mapped_column(Integer, default=0) + version: Mapped[int] = mapped_column(Integer, default=0) + model_items: Mapped[int | None] = mapped_column(Integer, default=None) + # error：整张表没起草（表结构里没有、写入一直冲突、服务重启中断）；model_error：模型那部分没写出来的原因和办法 + error: Mapped[str | None] = mapped_column(Text, default=None) + model_error: Mapped[str | None] = mapped_column(Text, default=None) + started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + + class Setting(Base, TimestampMixin): """键值形式的用户设置，前端 Settings 页直接读写。""" diff --git a/backend/app/main.py b/backend/app/main.py index 8a1feb7a..ba7dade7 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -70,6 +70,15 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: except Exception: # noqa: BLE001 logger.exception("上传表格的版本整理失败，服务照常启动") + # 上次没做完的数据目录起草任务：标成中断（没结束的表记停止），不自动续跑——拷来的数据目录（检查脚本、验收副本） + # 启动时不能替人去调模型。界面上点「重试」接着起草没写好的表 + from app.data import catalog_jobs + + try: + await catalog_jobs.startup() + except Exception: # noqa: BLE001 + logger.exception("数据目录起草任务的启动整理失败，服务照常启动") + await run_manager.setup() from app.sandbox.manager import sandbox_manager as _sbm @@ -84,8 +93,9 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: try: yield finally: - # 先停调度，再停运行：停运行的途中不该再有新运行被发起 + # 先停调度，再停运行：停运行的途中不该再有新运行被发起。起草任务停下来标成中断，下次启动不会再续 await scheduler.stop() + await catalog_jobs.manager.shutdown() await run_manager.shutdown() from app.sandbox.manager import sandbox_manager diff --git a/backend/tests/test_catalog_draft_jobs.py b/backend/tests/test_catalog_draft_jobs.py new file mode 100644 index 00000000..87b66072 --- /dev/null +++ b/backend/tests/test_catalog_draft_jobs.py @@ -0,0 +1,526 @@ +"""数据目录的后台起草任务（data/catalog_jobs.py、/api/datasources/{id}/catalog/draft-jobs）与起草要求。 + +- 建任务 202；同一个源已有排队或进行中的任务 409 catalog_draft_busy；起草模型用不了 409 + catalog_draft_model_unavailable（不建任务）；没有表结构 409；表结构里没有的表照样建进任务、状态 skipped +- 在后台起草，每张表一结束就并入目录；进度、计数、用量落库，接口读的是库里的状态 +- 停止：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没写完的表记 cancelled，已经写进目录的不回滚；已结束的 409 +- 服务启动时把没结束的任务标成 interrupted（没结束的表记 cancelled、写明服务重启），不自动续跑；关停时同样 +- 重试：只含原任务里 failed、partial、cancelled 的表，本次说明沿用、起草要求取数据源当前的；没有可重试的 422 +- 目录清单：每行 last_draft，顶层 guidance、active_job +- 起草要求存在数据源 options 里（上传表格的源存在它自己那一行），数据源接口不显示、整份回写 options 时留着，拼连接串时摘掉 + +表名、列名、说明文字都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import uuid +from types import SimpleNamespace +from urllib.parse import quote + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.data import catalog, catalog_jobs +from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION, build_url +from app.db.base import SessionLocal +from app.db.models import CatalogDraftJob, CatalogDraftTask, DataSource +from app.main import app as _fastapi_app +from tests.fixtures.catalog.fake_model import FakeModel, requested +from tests.fixtures.sources import drop_source + +ACTOR = {"X-Actor": quote("王敏")} + + +# -------------------------------------------------------------------------- +# 夹具 +# -------------------------------------------------------------------------- + + +def _col(name: str, type_: str = "INTEGER", comment: str | None = None) -> dict: + return {"name": name, "type": type_, "nullable": True, "comment": comment} + + +def _tables() -> dict: + return { + "visits": {"qualified": "visits", "columns": [_col("id"), _col("ticket_no", "TEXT"), _col("visit_time", "TEXT"), + _col("channel_id")], + "primary_key": ["id"], "comment": None, "is_view": False, + "foreign_keys": [{"columns": ["channel_id"], "to_table": "channels", "to_columns": ["id"]}]}, + "channels": {"qualified": "channels", "columns": [_col("id"), _col("name", "TEXT")], "primary_key": ["id"], + "comment": "售票渠道", "is_view": False, "foreign_keys": []}, + "gates": {"qualified": "gates", "columns": [_col("id"), _col("gate_code", "TEXT")], "primary_key": ["id"], + "comment": None, "is_view": False, "foreign_keys": []}, + } + + +REPLIES = { + "visits": {"label": "入园记录", "kind": "fact", "columns": [{"name": "ticket_no", "label": "票号"}]}, + "channels": {"label": "渠道", "kind": "dimension"}, + "gates": {"label": "闸口", "kind": "dimension"}, +} + +_CREATED: list[str] = [] + + +@pytest.fixture(autouse=True) +async def _cleanup(): + yield + # 后台任务先收完（停不下来的停掉），再删源：删源会级联删掉任务 + await catalog_jobs.manager.shutdown(grace=5) + while _CREATED: + await drop_source(_CREATED.pop()) + + +async def _source(tables: dict | None = None, **kw) -> str: + async with SessionLocal() as session: + row = DataSource(name=f"cdj_{uuid.uuid4().hex[:10]}", kind="sqlite", database=":memory:", + schema_cache={"tables": tables if tables is not None else _tables()}, **kw) + session.add(row) + await session.commit() + _CREATED.append(row.id) + return row.id + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c: + yield c + + +def _base(sid: str) -> str: + return f"/api/datasources/{sid}/catalog" + + +def use(monkeypatch, model) -> None: + """起草模型换成假模型（建任务时查一次能不能用，开始起草时再取一次）。""" + async def resolve(session): + return model, "fake-model" + + monkeypatch.setattr(catalog, "resolve_draft_model", resolve) + + +async def finished(job_id: str) -> None: + await catalog_jobs.manager.wait(job_id, timeout=10) + + +async def get_job(client, sid: str, job_id: str) -> dict: + r = await client.get(f"{_base(sid)}/draft-jobs/{job_id}") + assert r.status_code == 200, r.text + return r.json()["job"] + + +class Held: + """假模型：请求里有 hold 里的表时，等 release 才回；被停止时记下这次调用被取消了。""" + + def __init__(self, replies: dict, hold: set[str]) -> None: + self.inner, self.hold = FakeModel(replies), set(hold) + self.release = asyncio.Event() + self.cancelled = 0 + self.model_name = "held-model" + + def with_structured_output(self, schema, **_): + outer = self + + class _S: + async def ainvoke(self, messages): + text = "\n".join(str(m.content) for m in messages) + if outer.hold & set(requested(text)): + try: + await outer.release.wait() + except asyncio.CancelledError: + outer.cancelled += 1 + raise + return outer.inner._reply(messages, "structured") + + return _S() + + async def ainvoke(self, messages): + return await self.inner.ainvoke(messages) + + +async def until(predicate, timeout: float = 5.0) -> None: + for _ in range(int(timeout / 0.01)): + if await predicate(): + return + await asyncio.sleep(0.01) + raise AssertionError("等不到") + + +# -------------------------------------------------------------------------- +# 建任务、在后台起草 +# -------------------------------------------------------------------------- + + +async def test_a_job_drafts_in_the_background_and_writes_each_table(client, monkeypatch): + sid = await _source() + model = FakeModel(REPLIES) + use(monkeypatch, model) + await client.put(f"{_base(sid)}/guidance", json={"guidance": "金额一律按元写单位"}) + r = await client.post(f"{_base(sid)}/draft-jobs", headers=ACTOR, + json={"tables": ["visits", "nope", "channels"], "use_model": True, "note": " 看渠道 "}) + assert r.status_code == 202, r.text + job = r.json()["job"] + assert job["status"] in ("queued", "running") and job["source_id"] == sid + assert job["use_model"] is True and job["guidance"] == "金额一律按元写单位" and job["note"] == "看渠道" + assert job["created_by"] == "王敏" and job["total"] == 3 + assert [t["table_name"] for t in job["tasks"]] == ["visits", "nope", "channels"], "按起草顺序" + await finished(job["id"]) + + job = await get_job(client, sid, job["id"]) + assert job["status"] == "succeeded" and job["error"] is None and job["model"] == "fake-model" + assert job["done"] == 3 and job["counts"] == {"pending": 0, "running": 0, "ok": 2, "partial": 0, "failed": 0, + "skipped": 1, "cancelled": 0} + assert job["started_at"] and job["finished_at"] and job["usage"]["calls"] >= 1 + by = {t["table_name"]: t for t in job["tasks"]} + assert by["nope"]["status"] == "skipped" and by["nope"]["error"] == "表结构里没有 nope，请先重新探查结构" + visits = by["visits"] + assert visits["status"] == "ok" and visits["parts_total"] == 1 and visits["parts_done"] == 1 + assert visits["model_items"] == 3 and visits["added"] == 4 and visits["version"] == 1 and visits["columns"] == 4 + assert visits["started_at"] and visits["finished_at"] and visits["error"] is None + assert job["added"] == sum(t["added"] for t in job["tasks"]) + # 起草要求和本次说明都交给了模型 + assert "起草要求：金额一律按元写单位" in model.calls[0] and "本次说明：看渠道" in model.calls[0] + + detail = (await client.get(f"{_base(sid)}/visits")).json() + assert detail["notes"]["label"]["value"] == "入园记录" and detail["updated_by"] == "王敏" + + +async def test_a_job_without_the_model_writes_structure_only(client): + sid = await _source() + r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": [], "use_model": False}) + assert r.status_code == 202, r.text + job = r.json()["job"] + assert job["total"] == 3, "不给表时按使用次数取前 20 张" + await finished(job["id"]) + job = await get_job(client, sid, job["id"]) + assert job["status"] == "succeeded" and job["model"] is None and job["usage"]["calls"] == 0 + assert all(t["parts_total"] is None and t["model_items"] is None for t in job["tasks"]) + assert {t["table_name"]: t["added"] for t in job["tasks"]}["visits"] == 1 # 外键一条 + + +async def test_a_table_whose_model_part_failed_is_failed_but_structure_is_written(client, monkeypatch): + sid = await _source() + use(monkeypatch, FakeModel(REPLIES, fail={"visits"})) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits", "channels"], + "use_model": True})).json()["job"] + await finished(job["id"]) + job = await get_job(client, sid, job["id"]) + by = {t["table_name"]: t for t in job["tasks"]} + assert job["status"] == "partial" + assert by["visits"]["status"] == "failed" and "网关超时" in by["visits"]["model_error"] + assert by["visits"]["error"] is None and by["visits"]["added"] == 1, "外键照常写入" + assert by["channels"]["status"] == "ok" + + +async def test_busy_source_is_409_and_the_active_job_is_reachable(client, monkeypatch): + sid = await _source() + model = Held(REPLIES, hold={"visits"}) + use(monkeypatch, model) + first = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits"], "use_model": True})).json() + r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"], "use_model": False}) + assert r.status_code == 409 and r.json()["code"] == "catalog_draft_busy", r.text + active = (await client.get(f"{_base(sid)}/draft-jobs/active")).json()["job"] + assert active["id"] == first["job"]["id"] + listed = (await client.get(_base(sid))).json() + assert listed["active_job"]["id"] == first["job"]["id"] and listed["active_job"]["total"] == 1 + assert listed["active_job"]["status"] in ("queued", "running") and listed["active_job"]["done"] == 0 + model.release.set() + await finished(first["job"]["id"]) + assert (await client.get(f"{_base(sid)}/draft-jobs/active")).json()["job"] is None + latest = (await client.get(f"{_base(sid)}/draft-jobs/latest")).json()["job"] + assert latest["id"] == first["job"]["id"] and latest["status"] == "succeeded" + assert (await client.get(_base(sid))).json()["active_job"] is None + + +async def test_unavailable_model_is_409_and_no_job_is_created(client, monkeypatch): + sid = await _source() + + async def unavailable(session): + raise catalog.CatalogModelUnavailable("未配置模型接入，无法由模型起草数据目录。请到「设置 → 模型接入」添加") + + monkeypatch.setattr(catalog, "resolve_draft_model", unavailable) + r = await client.post(f"{_base(sid)}/draft-jobs", json={"use_model": True}) + assert r.status_code == 409, r.text + assert r.json()["code"] == "catalog_draft_model_unavailable" and "模型接入" in r.json()["detail"] + assert (await client.get(f"{_base(sid)}/draft-jobs/latest")).json()["job"] is None + + +async def test_without_schema_or_with_bad_input_nothing_is_created(client): + sid = await _source(tables={}) + r = await client.post(f"{_base(sid)}/draft-jobs", json={"use_model": False}) + assert r.status_code == 409 and "探查结构" in r.json()["detail"] + sid = await _source() + r = await client.post(f"{_base(sid)}/draft-jobs", json={"use_model": False, "note": "说" * 1001}) + assert r.status_code == 422 and "1000" in r.json()["detail"] + r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": [f"t{i}" for i in range(501)]}) + assert r.status_code == 422 and "500" in r.json()["detail"] + assert (await client.get(f"{_base(sid)}/draft-jobs/latest")).json()["job"] is None + assert (await client.get(f"{_base('missing-source')}/draft-jobs/latest")).status_code == 404 + assert (await client.get(f"{_base(sid)}/draft-jobs/not-a-job")).status_code == 404 + + +async def test_a_job_of_another_source_is_404(client): + a, b = await _source(), await _source() + job = (await client.post(f"{_base(a)}/draft-jobs", json={"tables": ["gates"]})).json()["job"] + await finished(job["id"]) + assert (await client.get(f"{_base(b)}/draft-jobs/{job['id']}")).status_code == 404 + assert (await client.post(f"{_base(b)}/draft-jobs/{job['id']}/cancel")).status_code == 404 + assert (await client.post(f"{_base(b)}/draft-jobs/{job['id']}/retry")).status_code == 404 + + +# -------------------------------------------------------------------------- +# 停止、服务重启 +# -------------------------------------------------------------------------- + + +async def test_cancelling_a_queued_job(client, monkeypatch): + sid = await _source() + monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None) # 不让它开始 + job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates", "channels"]})).json()["job"] + assert job["status"] == "queued" + r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel") + assert r.status_code == 200, r.text + job = r.json()["job"] + assert job["status"] == "cancelled" and job["finished_at"] + assert {t["status"] for t in job["tasks"]} == {"cancelled"} + again = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel") + assert again.status_code == 409 and again.json()["code"] == "catalog_draft_finished" + + +async def test_cancelling_a_running_job_stops_the_call_and_keeps_what_is_written(client, monkeypatch): + sid = await _source() + model = Held(REPLIES, hold={"visits"}) + use(monkeypatch, model) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + job = (await client.post(f"{_base(sid)}/draft-jobs", + json={"tables": ["channels", "visits"], "use_model": True})).json()["job"] + + async def channels_written() -> bool: + tasks = {t["table_name"]: t for t in (await get_job(client, sid, job["id"]))["tasks"]} + return tasks["channels"]["status"] == "ok" and tasks["visits"]["status"] == "running" + + await until(channels_written) + r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel") + assert r.status_code == 200, r.text + after = r.json()["job"] + by = {t["table_name"]: t for t in after["tasks"]} + assert after["status"] == "cancelled" and after["finished_at"] + assert by["channels"]["status"] == "ok" and by["visits"]["status"] == "cancelled" + assert model.cancelled == 1, "正在进行的模型调用被停掉" + detail = (await client.get(f"{_base(sid)}/channels")).json() + assert detail["notes"]["kind"]["value"] == "dimension", "已经写进目录的不回滚" + assert (await client.get(f"{_base(sid)}/draft-jobs/active")).json()["job"] is None + + +async def test_startup_marks_unfinished_jobs_interrupted_and_does_not_resume(client): + sid = await _source() + async with SessionLocal() as session: + job = CatalogDraftJob(source_id=sid, status="running", use_model=True) + session.add(job) + await session.flush() + for seq, (name, status) in enumerate([("visits", "ok"), ("channels", "running"), ("gates", "pending")]): + session.add(CatalogDraftTask(job_id=job.id, seq=seq, table_name=name, status=status)) + await session.commit() + job_id = job.id + assert await catalog_jobs.startup() >= 1 + got = await get_job(client, sid, job_id) + by = {t["table_name"]: t for t in got["tasks"]} + assert got["status"] == "interrupted" and got["error"] == "服务重启，起草中断" and got["finished_at"] + assert by["visits"]["status"] == "ok" and by["visits"]["error"] is None + assert by["channels"]["status"] == by["gates"]["status"] == "cancelled" + assert by["channels"]["error"] == by["gates"]["error"] == "服务重启，起草中断" + assert not catalog_jobs.manager.running(job_id), "不自动续跑" + + +async def test_shutdown_stops_running_jobs_as_interrupted(client, monkeypatch): + sid = await _source() + model = Held(REPLIES, hold={"visits"}) + use(monkeypatch, model) + job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits"], "use_model": True})).json()["job"] + + async def running() -> bool: + return (await get_job(client, sid, job["id"]))["tasks"][0]["status"] == "running" + + await until(running) + await catalog_jobs.manager.shutdown(grace=5) + got = await get_job(client, sid, job["id"]) + assert got["status"] == "interrupted" and got["tasks"][0]["status"] == "cancelled" + assert got["tasks"][0]["error"] == "服务重启，起草中断" + + +# -------------------------------------------------------------------------- +# 重试 +# -------------------------------------------------------------------------- + + +async def test_retry_takes_only_failed_partial_and_cancelled_tables(client, monkeypatch): + sid = await _source() + use(monkeypatch, FakeModel(REPLIES, fail={"visits"})) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + await client.put(f"{_base(sid)}/guidance", json={"guidance": "旧的要求"}) + first = (await client.post(f"{_base(sid)}/draft-jobs", json={ + "tables": ["visits", "channels", "nope"], "use_model": True, "note": "第一次"})).json()["job"] + await finished(first["id"]) + await client.put(f"{_base(sid)}/guidance", json={"guidance": "新的要求"}) + assert (await get_job(client, sid, first["id"]))["guidance"] == "旧的要求", "任务记下的是当时的快照" + + use(monkeypatch, FakeModel(REPLIES)) + r = await client.post(f"{_base(sid)}/draft-jobs/{first['id']}/retry", headers=ACTOR) + assert r.status_code == 202, r.text + retry = r.json()["job"] + assert [t["table_name"] for t in retry["tasks"]] == ["visits"] + assert retry["use_model"] is True and retry["note"] == "第一次" and retry["guidance"] == "新的要求" + assert retry["retry_of"] == first["id"] and retry["created_by"] == "王敏" + await finished(retry["id"]) + assert (await get_job(client, sid, retry["id"]))["status"] == "succeeded" + + nothing = await client.post(f"{_base(sid)}/draft-jobs/{retry['id']}/retry") + assert nothing.status_code == 422 and "没有" in nothing.json()["detail"] + + +async def test_retry_while_busy_is_409(client, monkeypatch): + sid = await _source() + monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None) + job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"]})).json()["job"] + await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel") + other = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["channels"]})).json()["job"] + assert other["status"] == "queued" + r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/retry") + assert r.status_code == 409 and r.json()["code"] == "catalog_draft_busy" + + +async def test_retry_with_an_unavailable_model_is_409(client, monkeypatch): + sid = await _source() + monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None) + use(monkeypatch, FakeModel(REPLIES)) + job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"], "use_model": True})).json()["job"] + await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel") + + async def unavailable(session): + raise catalog.CatalogModelUnavailable("当前只有演示用的模型接入，无法用于起草数据目录") + + monkeypatch.setattr(catalog, "resolve_draft_model", unavailable) + r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/retry") + assert r.status_code == 409 and r.json()["code"] == "catalog_draft_model_unavailable" + + +# -------------------------------------------------------------------------- +# 目录清单上的起草状态 +# -------------------------------------------------------------------------- + + +async def test_the_list_carries_last_draft_per_table(client, monkeypatch): + sid = await _source() + use(monkeypatch, FakeModel(REPLIES, fail={"visits"})) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + first = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits", "channels"], + "use_model": True})).json()["job"] + await finished(first["id"]) + second = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["channels"]})).json()["job"] + await finished(second["id"]) + body = (await client.get(_base(sid))).json() + rows = {r["table_name"]: r for r in body["tables"]} + assert rows["gates"]["last_draft"] is None + visits = rows["visits"]["last_draft"] + assert visits["status"] == "failed" and visits["job_id"] == first["id"] and visits["finished_at"] + assert "网关超时" in visits["model_error"] and visits["error"] is None + assert rows["channels"]["last_draft"]["job_id"] == second["id"], "取最近一次任务里的那一项" + assert body["guidance"] is None and body["active_job"] is None + + +# -------------------------------------------------------------------------- +# 起草要求 +# -------------------------------------------------------------------------- + + +async def test_guidance_is_saved_trimmed_cleared_and_limited(client): + sid = await _source() + r = await client.put(f"{_base(sid)}/guidance", json={"guidance": " 票务库，金额按元。\n状态 9 是作废 "}) + assert r.status_code == 200, r.text + assert r.json() == {"guidance": "票务库，金额按元。\n状态 9 是作废"} + assert (await client.get(_base(sid))).json()["guidance"] == "票务库，金额按元。\n状态 9 是作废" + too_long = await client.put(f"{_base(sid)}/guidance", json={"guidance": "字" * 2001}) + assert too_long.status_code == 422 and "2000" in too_long.json()["detail"] + assert (await client.get(_base(sid))).json()["guidance"] == "票务库，金额按元。\n状态 9 是作废" + assert (await client.put(f"{_base(sid)}/guidance", json={"guidance": "字" * 2000})).status_code == 200 + cleared = await client.put(f"{_base(sid)}/guidance", json={"guidance": " "}) + assert cleared.json() == {"guidance": None} + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + assert CATALOG_DRAFT_GUIDANCE_OPTION not in (row.options or {}) + assert (await client.put(f"{_base('missing-source')}/guidance", json={"guidance": "x"})).status_code == 404 + + +async def test_a_table_named_guidance_can_still_be_saved(client): + """起草要求的路径和「整份提交一张表」撞名：表名恰好叫 guidance 时，{notes, if_version} 照旧是那张表的提交， + 不会被当成清除起草要求；两样都没写的 422。""" + tables = _tables() + tables["guidance"] = {"qualified": "guidance", "columns": [_col("id"), _col("body", "TEXT")], + "primary_key": ["id"], "comment": None, "is_view": False, "foreign_keys": []} + sid = await _source(tables) + await client.put(f"{_base(sid)}/guidance", json={"guidance": "金额按元"}) + r = await client.put(f"{_base(sid)}/guidance", json={"notes": {"label": {"value": "操作指引"}}, "if_version": 0}, + headers=ACTOR) + assert r.status_code == 200, r.text + assert r.json()["table_name"] == "guidance" and r.json()["notes"]["label"]["value"] == "操作指引" + assert (await client.get(_base(sid))).json()["guidance"] == "金额按元" + assert (await client.put(f"{_base(sid)}/guidance", json={})).status_code == 422 + + +async def test_guidance_of_an_uploaded_source_lives_on_its_own_row(client): + sid = await _source(origin="upload") + r = await client.put(f"{_base(sid)}/guidance", json={"guidance": "按期上传的客流日报"}) + assert r.status_code == 200, r.text + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + assert row.options[CATALOG_DRAFT_GUIDANCE_OPTION] == "按期上传的客流日报" + + +async def test_the_datasource_form_neither_shows_nor_drops_the_guidance(client): + sid = await _source(options={"schema": "main"}) + await client.put(f"{_base(sid)}/guidance", json={"guidance": "金额按元"}) + shown = (await client.get("/api/datasources")).json() + mine = next(s for s in shown if s["id"] == sid) + assert CATALOG_DRAFT_GUIDANCE_OPTION not in mine["options"] and mine["options"]["schema"] == "main" + # 表单整份回写 options（不带起草要求，或者带着一份别的）：起草要求留着，只认 PUT /catalog/guidance + r = await client.patch(f"/api/datasources/{sid}", json={"options": {"schema": "other", + CATALOG_DRAFT_GUIDANCE_OPTION: "篡改"}}) + assert r.status_code == 200, r.text + assert CATALOG_DRAFT_GUIDANCE_OPTION not in r.json()["options"] + assert (await client.get(_base(sid))).json()["guidance"] == "金额按元" + async with SessionLocal() as session: + assert (await session.get(DataSource, sid)).options == {"schema": "other", + CATALOG_DRAFT_GUIDANCE_OPTION: "金额按元"} + + +def test_the_guidance_never_reaches_the_driver(): + src = SimpleNamespace(kind="mysql", host="db.example", port=3306, database="shop", username="reader", + password=None, options={"charset": "utf8mb4", CATALOG_DRAFT_GUIDANCE_OPTION: "金额按元"}) + url = build_url(src) + assert "charset=utf8mb4" in url and CATALOG_DRAFT_GUIDANCE_OPTION not in url + + +async def test_the_sync_draft_uses_the_source_guidance(client, monkeypatch): + sid = await _source(description="虚构的景区票务库") + model = FakeModel(REPLIES) + use(monkeypatch, model) + await client.put(f"{_base(sid)}/guidance", json={"guidance": "渠道表的名称是对外展示名"}) + r = await client.post(f"{_base(sid)}/draft", json={"tables": ["channels"], "use_model": True}) + assert r.status_code == 200, r.text + assert "数据源说明：虚构的景区票务库\n起草要求：渠道表的名称是对外展示名" in model.calls[0] + assert set(r.json()) == {"tables", "total", "model_used", "model", "model_error"}, "旧的同步接口返回形状不变" + + +async def test_jobs_go_away_with_their_source(client): + sid = await _source() + job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"]})).json()["job"] + await finished(job["id"]) + await drop_source(sid) + _CREATED.remove(sid) + async with SessionLocal() as session: + assert (await session.execute(select(CatalogDraftJob).where(CatalogDraftJob.id == job["id"]))).first() is None + assert (await session.execute(select(CatalogDraftTask).where(CatalogDraftTask.job_id == job["id"]))).first() \ + is None