From 4f0b60df03e2dcff4667a71eba5bf17eabc603e5 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Thu, 15 Oct 2026 19:12:48 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AF=BC=E5=85=A5=E3=80=81=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E8=B5=B7=E8=8D=89=E6=92=9E=E4=B8=8A=E5=89=96=E6=9E=90=E3=80=81?= =?UTF-8?q?=E8=B5=B7=E8=8D=89=E4=BB=BB=E5=8A=A1=E6=97=B6=EF=BC=8C=E8=AF=B4?= =?UTF-8?q?=E6=B3=95=E6=8C=89=E8=BF=99=E6=AC=A1=E8=A6=81=E5=81=9A=E7=9A=84?= =?UTF-8?q?=E4=BA=8B=E8=AF=B4=E4=B8=8B=E4=B8=80=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 复核 R3。导入目录、同步起草撞上剖析任务回 409 profile_busy 时，原话来自 busy_message，结尾固定是「请等这次剖析结束 （或先停止它）再开始新的剖析」；使用者这时是在导入、起草，不是要开始新的剖析，导入对话框原样显示这句。导入撞上起草 任务时同样说「再开始新的起草」（这是改造前就有的，同一类问题一起改了）。 - catalog_profile_jobs.busy_message / ensure_free 加 then（缺省 THEN_PROFILE「再开始新的剖析」，建剖析任务、重试照旧）。 - catalog_jobs.busy_message / sync_busy_message / ensure_draft_free 同样加 then（缺省 THEN_DRAFT「再开始新的起草」）； ensure_idle 和 _lane_busy 按这次要做的事（kind）选：导入撞上什么都是「再导入」；同步起草撞上剖析是「再起草」，撞上 起草仍是「再开始新的起草」（和建起草任务同一句话，test_catalog_draft_jobs 钉着这一点，没动那个文件）。 - 导入接口的说明补上撞上剖析任务回 profile_busy。 测试：剖析时导入、同步起草、再建剖析任务三句结尾各自对得上；队被剖析占着时（任务刚好在收尾）同样按这次要做的事说。 --- backend/app/api/catalog.py | 5 +- backend/app/data/catalog_jobs.py | 62 ++++++++++++++-------- backend/app/data/catalog_profile_jobs.py | 28 ++++++---- backend/tests/test_catalog_profile_jobs.py | 11 +++- 4 files changed, 71 insertions(+), 35 deletions(-) diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index 41b84944..865929f0 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -346,8 +346,9 @@ async def preview_catalog_import(source_id: str, payload: CatalogImportIn, @router.post("/{source_id}/catalog/import") async def import_catalog(source_id: str, payload: CatalogImportIn, x_actor: str | None = Header(default=None), session: AsyncSession = Depends(get_session)) -> dict[str, Any]: - """导入：规则同预览，逐表写入（乐观锁），返回同样形状的报告和每张表写入后的版本。有排队或进行中的起草任务 - 409 catalog_draft_busy；文件格式不对 422；没有表结构 409。""" + """导入：规则同预览，逐表写入（乐观锁），返回同样形状的报告和每张表写入后的版本。有排队或进行中的起草任务（或正在 + 同步起草、导入）409 catalog_draft_busy，有剖析任务 409 profile_busy（说法的最后一句说「再导入」）；文件格式不对 422； + 没有表结构 409。""" row, source = await _import_source(session, source_id) try: async with catalog_jobs.import_write(session, row): diff --git a/backend/app/data/catalog_jobs.py b/backend/app/data/catalog_jobs.py index 9861e2b9..233338ad 100644 --- a/backend/app/data/catalog_jobs.py +++ b/backend/app/data/catalog_jobs.py @@ -313,12 +313,26 @@ async def last_drafts(session: AsyncSession, source_id: str) -> dict[str, dict[s # ========================================================================== -def busy_message(row: Any) -> str: - return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录，请等这次起草结束（或先停止它）再开始新的起草" +#: 撞上起草时说法的最后一句（「请等这次起草结束{then}」）缺省是建起草任务的；导入目录、同步起草撞上起草、剖析时按这次要做 +#: 的事说（_then） +THEN_DRAFT = "再开始新的起草" -def sync_busy_message(row: Any) -> str: - return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录（同步起草），请等这次起草结束再开始新的起草" +def _then(kind: str, busy: str) -> str: + """同步写操作（kind：import / sync_draft）撞上 busy（draft 起草、profile 剖析）时说法的最后一句：导入撞上什么都是 + 「再导入」；同步起草撞上起草是「再开始新的起草」（和建起草任务同一句话），撞上剖析是「再起草」——那时要做的是起草， + 不是「开始新的剖析」。""" + if kind == "import": + return "再导入" + return THEN_DRAFT if busy == "draft" else "再起草" + + +def busy_message(row: Any, then: str = THEN_DRAFT) -> str: + return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录，请等这次起草结束（或先停止它）{then}" + + +def sync_busy_message(row: Any, then: str = THEN_DRAFT) -> str: + return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录（同步起草），请等这次起草结束{then}" def import_busy_message(row: Any) -> str: @@ -334,33 +348,37 @@ def sync_drafting(source_id: str) -> bool: return source_id in _SYNC -async def ensure_draft_free(session: AsyncSession, row: Any) -> None: +async def ensure_draft_free(session: AsyncSession, row: Any, *, then: str = THEN_DRAFT) -> None: """同种任务互斥：这个源没有在起草——没有排队或进行中的起草任务，也没有在同步起草。有就抛 DraftBusy。建起草任务、 - 重试之前调它。剖析任务不算：同源有剖析任务时起草任务照建，排在它后面。""" + 重试之前调它。剖析任务不算：同源有剖析任务时起草任务照建，排在它后面。then 是说法最后一句要做的事。""" if row.id in _SYNC: - raise DraftBusy(sync_busy_message(row)) + raise DraftBusy(sync_busy_message(row, then)) if await active_job(session, row.id) is not None: - raise DraftBusy(busy_message(row)) + raise DraftBusy(busy_message(row, then)) -async def ensure_idle(session: AsyncSession, row: Any) -> None: - """同步写操作（导入目录、同步起草）要求源空闲：没有在起草（DraftBusy，同 ensure_draft_free），也没有排队或进行中的 - 剖析任务（catalog_profile_jobs.ProfileBusy，说法里写明进行到第几张）。""" +async def ensure_idle(session: AsyncSession, row: Any, kind: str = "import") -> None: + """同步写操作（kind：import 导入目录、sync_draft 同步起草）要求源空闲：没有在起草（DraftBusy，同 ensure_draft_free）， + 也没有排队或进行中的剖析任务（catalog_profile_jobs.ProfileBusy，说法里写明进行到第几张）。说法最后一句按 kind 说 + 这次要做的事（_then）。""" from app.data import catalog_profile_jobs - await ensure_draft_free(session, row) - await catalog_profile_jobs.ensure_free(session, row) + await ensure_draft_free(session, row, then=_then(kind, "draft")) + await catalog_profile_jobs.ensure_free(session, row, then=_then(kind, "profile")) -def _lane_busy(row: Any, holder: runtime.LaneTicket) -> Exception: - """库里查过没有任务、这个源的队却被占着（同步起草、导入目录正在做，或者一个任务刚好在收尾）：按占着的是谁说。""" +def _lane_busy(row: Any, holder: runtime.LaneTicket, kind: str) -> Exception: + """库里查过没有任务、这个源的队却被占着（同步起草、导入目录正在做，或者一个任务刚好在收尾）：按占着的是谁说，最后 + 一句按这次要做的事（kind）说。""" if holder.kind == "profile": from app.data import catalog_profile_jobs - return catalog_profile_jobs.ProfileBusy(catalog_profile_jobs.busy_message(row, None)) + return catalog_profile_jobs.ProfileBusy(catalog_profile_jobs.busy_message(row, None, + then=_then(kind, "profile"))) if holder.kind == "import": return DraftBusy(import_busy_message(row)) - return DraftBusy(sync_busy_message(row) if holder.kind == "sync_draft" else busy_message(row)) + then = _then(kind, "draft") + return DraftBusy(sync_busy_message(row, then) if holder.kind == "sync_draft" else busy_message(row, then)) @contextlib.asynccontextmanager @@ -371,7 +389,7 @@ async def _sync_write(row: Any, kind: str, check: Callable[[], Awaitable[None]]) try: ticket = runtime.lanes.try_enter(row.id, kind) except runtime.LaneBusy as e: - raise _lane_busy(row, e.holder) from e + raise _lane_busy(row, e.holder, kind) from e try: yield finally: @@ -379,12 +397,12 @@ async def _sync_write(row: Any, kind: str, check: Callable[[], Awaitable[None]]) async def _ensure_idle_but_sync(session: AsyncSession, row: Any) -> None: - """同 ensure_idle，只是不看 _SYNC：同步起草自己刚登记在里面。""" + """同 ensure_idle(…, "sync_draft")，只是不看 _SYNC：同步起草自己刚登记在里面。""" from app.data import catalog_profile_jobs if await active_job(session, row.id) is not None: raise DraftBusy(busy_message(row)) - await catalog_profile_jobs.ensure_free(session, row) + await catalog_profile_jobs.ensure_free(session, row, then=_then("sync_draft", "profile")) @contextlib.asynccontextmanager @@ -406,7 +424,7 @@ async def sync_draft(session: AsyncSession, row: Any) -> AsyncIterator[None]: async def import_write(session: AsyncSession, row: Any) -> AsyncIterator[None]: """导入目录期间占着这个源的队：源上有排队或进行中的起草、剖析任务（或者正在同步起草、导入）时抛 DraftBusy / ProfileBusy；导入期间新建的任务排在它后面。""" - async with _sync_write(row, "import", lambda: ensure_idle(session, row)): + async with _sync_write(row, "import", lambda: ensure_idle(session, row, "import")): yield @@ -791,7 +809,7 @@ manager = DraftJobs() __all__ = ["ACTIVE", "BLANK_SAMPLE", "DraftBusy", "DraftFinished", "DraftJobs", "FINISHED", "GUIDANCE_MAX", "INTERRUPTED_ERROR", "INTERRUPTED_WRITTEN", "MAX_TABLES", "NOTE_MAX", "NothingToRetry", "RETRYABLE", - "STOPPED_WRITTEN", "TASK_DONE", "TASK_STATUSES", "TextTooLong", + "STOPPED_WRITTEN", "TASK_DONE", "TASK_STATUSES", "THEN_DRAFT", "TextTooLong", "active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_draft_free", "ensure_idle", "guidance_of", "import_busy_message", "import_write", "job_brief", "job_dict", "job_out", "last_drafts", "latest_job", "load", "manager", "retry_tables", "save_guidance", "startup", "sync_busy_message", "sync_draft", diff --git a/backend/app/data/catalog_profile_jobs.py b/backend/app/data/catalog_profile_jobs.py index d2794560..d1f39f58 100644 --- a/backend/app/data/catalog_profile_jobs.py +++ b/backend/app/data/catalog_profile_jobs.py @@ -390,9 +390,15 @@ def due_reason(meta: dict[str, Any] | None, notes: dict[str, Any] | None, last: # ========================================================================== +#: 撞上剖析任务时说法的最后一句（「请等这次剖析结束（或先停止它）{then}」）缺省是建剖析任务的；导入目录、同步起草撞上 +#: 时各说各的（catalog_jobs）：那时要做的是导入、起草，不是开始新的剖析 +THEN_PROFILE = "再开始新的剖析" + + def busy_message(row: Any, brief: dict[str, Any] | None, *, table: str | None = None, - position: int | None = None) -> str: - """同源已有剖析任务时的说法：写明进行到第几张；从表详情点「剖析这张表」时再说这张表在不在那个任务里、排第几。""" + position: int | None = None, then: str = THEN_PROFILE) -> str: + """同源已有剖析任务时的说法：写明进行到第几张；从表详情点「剖析这张表」时再说这张表在不在那个任务里、排第几；最后 + 一句按调用方要做的事说（then）。""" name = getattr(row, "name", "") or "" if brief is None: text = f"数据源「{name}」正在剖析" @@ -402,11 +408,13 @@ def busy_message(row: Any, brief: dict[str, Any] | None, *, table: str | None = text = f"数据源「{name}」正在剖析：第 {min(brief['done'] + 1, brief['total'])} / {brief['total']} 张表" if table is not None: text += f"，{table} 在这次剖析里，排第 {position} 张" if position else f"，{table} 不在这次剖析里" - return text + "。请等这次剖析结束（或先停止它）再开始新的剖析" + return text + f"。请等这次剖析结束（或先停止它）{then}" -async def ensure_free(session: AsyncSession, row: Any, tables: list[str] | None = None) -> None: - """这个源没有排队或进行中的剖析任务；有就抛 ProfileBusy。只指定了一张表时，说法里带上这张表在不在那个任务里。""" +async def ensure_free(session: AsyncSession, row: Any, tables: list[str] | None = None, *, + then: str = THEN_PROFILE) -> None: + """这个源没有排队或进行中的剖析任务；有就抛 ProfileBusy。只指定了一张表时，说法里带上这张表在不在那个任务里；then 是 + 说法的最后一句要做的事（见 busy_message）。""" job = await active_job(session, row.id) if job is None: return @@ -417,7 +425,7 @@ async def ensure_free(session: AsyncSession, row: Any, tables: list[str] | None table = key names = [r.table_name for r in await task_rows(session, job.id)] position = names.index(key) + 1 if key in names else None - raise ProfileBusy(busy_message(row, brief, table=table, position=position)) + raise ProfileBusy(busy_message(row, brief, table=table, position=position, then=then)) #: 挑出 0 张表时的说法 @@ -876,7 +884,7 @@ manager = ProfileJobs() __all__ = ["INTERRUPTED_ERROR", "KEEP_JOBS", "MAX_TABLES", "NOTHING_AT_ALL", "NOTHING_NEEDED", "NothingToRetry", - "ProfileBusy", "ProfileFinished", "ProfileJobs", "ProfileNothing", "active_job", "busy_message", "cancel", - "create_job", "due_reason", "ensure_free", "job_brief", "job_dict", "job_out", "last_detail", - "last_profiles", "latest_job", "load", "load_task", "manager", "prune", "public_last", "retry_tables", - "startup", "task_detail", "task_row", "task_rows", "unsaved_error", "waiting_for"] + "ProfileBusy", "ProfileFinished", "ProfileJobs", "ProfileNothing", "THEN_PROFILE", "active_job", + "busy_message", "cancel", "create_job", "due_reason", "ensure_free", "job_brief", "job_dict", "job_out", + "last_detail", "last_profiles", "latest_job", "load", "load_task", "manager", "prune", "public_last", + "retry_tables", "startup", "task_detail", "task_row", "task_rows", "unsaved_error", "waiting_for"] diff --git a/backend/tests/test_catalog_profile_jobs.py b/backend/tests/test_catalog_profile_jobs.py index 1f93f3bc..e9dc2995 100644 --- a/backend/tests/test_catalog_profile_jobs.py +++ b/backend/tests/test_catalog_profile_jobs.py @@ -186,11 +186,20 @@ async def test_import_and_sync_draft_are_409_while_profiling(client, make_source job = (await _start(client, sid, ["channels"])).json()["job"] await asyncio.wait_for(gate.entered.wait(), 10) + # 说法按这次要做的事说下一步：是在导入、起草，不是要开始新的剖析 r = await client.post(f"{_base(sid)}/import", json={"content": exported}) assert r.status_code == 409 and r.json()["code"] == "profile_busy", r.text - assert "正在剖析" in r.json()["detail"] + assert "正在剖析" in r.json()["detail"] and r.json()["detail"].endswith("请等这次剖析结束（或先停止它）再导入") r = await client.post(f"{_base(sid)}/draft", json={"tables": ["gates"]}) assert r.status_code == 409 and r.json()["code"] == "profile_busy", r.text + assert r.json()["detail"].endswith("请等这次剖析结束（或先停止它）再起草") + again = await _start(client, sid, ["gates"]) + assert again.json()["detail"].endswith("请等这次剖析结束（或先停止它）再开始新的剖析") + # 库里查过没有任务、队却被剖析占着（任务刚好在收尾）时同样按这次要做的事说 + row, holder = SimpleNamespace(name="x"), SimpleNamespace(kind="profile") + assert str(catalog_jobs._lane_busy(row, holder, "import")).endswith("（或先停止它）再导入") + assert str(catalog_jobs._lane_busy(row, holder, "sync_draft")).endswith("（或先停止它）再起草") + assert str(catalog_jobs._lane_busy(row, SimpleNamespace(kind="draft"), "import")).endswith("（或先停止它）再导入") gate.release.set() await catalog_profile_jobs.manager.wait(job["id"], timeout=10)