From adc56da6cffff274a553f1224e694fa655930c72 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 14 Oct 2026 07:59:51 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E5=AE=A1?= =?UTF-8?q?=E9=98=85=E9=98=9F=E5=88=97=EF=BC=9A=E6=8E=A8=E6=96=AD=E7=9A=84?= =?UTF-8?q?=E9=A1=B9=E6=8C=89=E3=80=8C=E4=BD=BF=E7=94=A8=E6=AC=A1=E6=95=B0?= =?UTF-8?q?=E6=9D=83=E9=87=8D=20=C3=97=20=E9=A3=8E=E9=99=A9=E6=9D=83?= =?UTF-8?q?=E9=87=8D=20=C3=97=20=E6=9D=A5=E6=BA=90=E6=9D=83=E9=87=8D?= =?UTF-8?q?=E3=80=8D=E6=8E=92=E5=A5=BD=EF=BC=88=E6=9C=89=E6=95=88=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E6=9D=A1=E4=BB=B6=E3=80=81=E5=8E=BB=E9=87=8D=E8=A7=84?= =?UTF-8?q?=E5=88=99=E3=80=81=E4=B8=9A=E5=8A=A1=E6=97=A5=E6=9C=9F=E3=80=81?= =?UTF-8?q?=E5=85=B3=E8=81=94=E5=85=B3=E7=B3=BB=E6=9C=80=E9=AB=98=EF=BC=8C?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E8=B5=B7=E8=8D=89=E3=80=81=E5=91=BD=E5=90=8D?= =?UTF-8?q?=E6=8E=A8=E6=96=AD=E6=AF=94=E5=BA=93=E6=B3=A8=E9=87=8A=E4=BC=98?= =?UTF-8?q?=E5=85=88=EF=BC=89=EF=BC=8C=E5=B8=A6=E6=8E=92=E5=9C=A8=E8=BF=99?= =?UTF-8?q?=E9=87=8C=E7=9A=84=E7=90=86=E7=94=B1=EF=BC=8C=E5=8F=AF=E6=8C=89?= =?UTF-8?q?=E5=BD=B1=E5=93=8D=E6=95=B0=E5=AD=97=E7=9A=84=E9=A1=B9=20/=20?= =?UTF-8?q?=E5=90=8D=E7=A7=B0=E5=92=8C=E8=AF=B4=E6=98=8E=E3=80=81=E6=8C=89?= =?UTF-8?q?=E8=A1=A8=E7=AD=9B=E9=80=89=E5=92=8C=E5=88=86=E9=A1=B5=EF=BC=9B?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E7=A1=AE=E8=AE=A4=E3=80=81=E9=A9=B3=E5=9B=9E?= =?UTF-8?q?=E3=80=81=E6=92=A4=E9=94=80=E5=AE=A1=E9=98=85=E9=80=90=E8=A1=A8?= =?UTF-8?q?=E6=8C=89=E7=89=88=E6=9C=AC=E4=B9=90=E8=A7=82=E9=94=81=EF=BC=8C?= =?UTF-8?q?=E5=90=8C=E4=B8=80=E5=BC=A0=E8=A1=A8=E7=9A=84=E5=87=A0=E6=9D=A1?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=E5=86=99=E5=85=A5=EF=BC=8C=E4=B8=80=E5=BC=A0?= =?UTF-8?q?=E8=A1=A8=E5=86=B2=E7=AA=81=E4=B8=8D=E5=BD=B1=E5=93=8D=E5=88=AB?= =?UTF-8?q?=E7=9A=84=E8=A1=A8=EF=BC=9B=E8=A1=A8=E8=A2=AB=E5=93=AA=E4=BA=9B?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E7=94=A8=E8=BF=87=EF=BC=88=E6=96=B0=E7=9A=84?= =?UTF-8?q?=E5=9C=A8=E5=89=8D=EF=BC=8C=E5=8F=96=E6=B3=95=E5=90=8C=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E6=AC=A1=E6=95=B0=EF=BC=8C=E6=8A=BD=E5=87=BA=20query?= =?UTF-8?q?=5Fsnapshot=5Ftables=20=E5=85=B1=E7=94=A8=EF=BC=89=EF=BC=9B?= =?UTF-8?q?=E5=AE=A1=E9=98=85=E8=BF=9B=E5=BA=A6=EF=BC=88=E5=B8=B8=E7=94=A8?= =?UTF-8?q?=2020=20=E5=BC=A0=E8=A1=A8=E3=80=81=E5=85=A8=E9=83=A8=E3=80=81?= =?UTF-8?q?=E5=BD=B1=E5=93=8D=E6=95=B0=E5=AD=97=E7=9A=84=E9=A1=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/catalog.py | 62 +++- backend/app/data/catalog.py | 42 ++- backend/app/data/catalog_review.py | 404 +++++++++++++++++++++ backend/tests/test_catalog_review_queue.py | 357 ++++++++++++++++++ 4 files changed, 850 insertions(+), 15 deletions(-) create mode 100644 backend/app/data/catalog_review.py create mode 100644 backend/tests/test_catalog_review_queue.py diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index 0405a0f7..7680eb54 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -13,6 +13,9 @@ - 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） +- GET /catalog/review-queue 审阅队列：推断的项按使用次数、风险、来源排好；POST /catalog/review/batch 批量审阅， + GET /catalog/review-progress 审阅进度，GET /catalog/{table}/usage 用过这张表的运行 + （data/catalog_review.py） 规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径 不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。 @@ -24,7 +27,7 @@ from __future__ import annotations from typing import Any, Literal -from fastapi import APIRouter, Body, Depends, Header, HTTPException +from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query from fastapi.exceptions import RequestValidationError from pydantic import BaseModel, Field, ValidationError from sqlalchemy.ext.asyncio import AsyncSession @@ -35,6 +38,7 @@ from app.api.coded import ( ) from app.api.runs import actor_of from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, introspect, table_versions +from app.data import catalog_review from app.data.engine import SnapshotTampered from app.db.base import get_session from app.db.models import DataSource @@ -234,6 +238,30 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi } +# ---- 审阅队列、审阅进度（要登记在 GET /catalog/{table} 前面：不然 review-queue、review-progress 会被当成表名） + + +@router.get("/{source_id}/catalog/review-queue") +async def catalog_review_queue(source_id: str, + limit: int = Query(50, ge=1, le=catalog_review.QUEUE_LIMIT_MAX), + offset: int = Query(0, ge=0), + kind: Literal["all", "facts", "names"] = Query("all"), + table: str | None = Query(None, max_length=255), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """审阅队列：只列推断的项，按使用次数、风险、来源排好（data/catalog_review.py）。kind 筛影响数字的项（facts）或名称 + 和说明（names），table 只看一张表。""" + _, source = await _resolved(session, source_id) + return await catalog_review.review_queue(session, source, kind=kind, table=table or None, limit=limit, + offset=offset) + + +@router.get("/{source_id}/catalog/review-progress") +async def catalog_review_progress(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """审阅进度：常用 20 张表、全部、影响数字的项各自确认了多少。""" + _, source = await _resolved(session, source_id) + return await catalog_review.review_progress(session, source) + + @router.get("/{source_id}/catalog/{table}") async def get_catalog_table(source_id: str, table: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: @@ -527,6 +555,38 @@ async def review_catalog_item(source_id: str, table: str, payload: CatalogReview return await _detail(session, source, key, entry) +class CatalogBatchReviewItem(BaseModel): + table_name: str = Field(min_length=1, max_length=255) + path: str = Field(min_length=1, max_length=600) + action: Literal["confirm", "reject", "reset"] + if_version: int = Field(ge=0) + + +class CatalogBatchReviewIn(BaseModel): + items: list[CatalogBatchReviewItem] = Field(min_length=1, max_length=catalog_review.BATCH_MAX) + + +@router.post("/{source_id}/catalog/review/batch") +async def review_catalog_batch(source_id: str, payload: CatalogBatchReviewIn, + x_actor: str | None = Header(default=None), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """批量审阅：逐条结果和请求一一对应。逐表按 if_version 乐观锁，同一张表的几条一次写入；一张表冲突、路径不对、表不存在 + 只记在那几条上（ok 为假，code 是 catalog_conflict / catalog_path_invalid / table_missing），不影响别的。""" + _, source = await _resolved(session, source_id) + items = [catalog_review.BatchItem(i.table_name, i.path, i.action, i.if_version) for i in payload.items] + return {"results": await catalog_review.review_batch(session, source, items, actor=actor_of(x_actor))} + + +@router.get("/{source_id}/catalog/{table}/usage") +async def catalog_table_runs(source_id: str, table: str, + limit: int = Query(20, ge=1, le=catalog_review.RUNS_LIMIT_MAX), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """用过这张表的运行（total 是几次运行），新的在前；取法同使用次数（catalog.table_usage）。""" + _, source = await _resolved(session, source_id) + key, _ = await _table_key(session, source, table) + return await catalog_review.table_runs(session, source, key, limit=limit) + + def _tables_of(source: Any) -> dict[str, Any] | None: """核对提案用的表结构；没有表结构（尚未探查）时不核对列名。""" return ((source.schema_cache or {}).get("tables") or None) if source is not None else None diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 3dd0bb39..cdef28cf 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -2249,27 +2249,27 @@ def query_snapshot_meta(source: Any, sql: str) -> dict[str, Any]: return {"source_id": getattr(source, "id", None), "source": getattr(source, "name", None), "tables": tables} -async def table_usage(session: AsyncSession, source: Any) -> dict[str, int]: - """每张表被运行查询过的次数：{schema_cache 的表名: 次数}，没被查过的表不出现。 +async def query_snapshot_tables(session: AsyncSession, source: Any) -> list[tuple[str, frozenset[str]]]: + """这个源上属于运行的每一行查询快照引用：(运行 id, 这条查询用到的表——schema_cache 的表名)，没对上任何表的不收。 - 取法：数据源查询工具每次成功查询都存一份查询快照工件（tools/datasource.py），工件引用表 artifacts 里 - 按 (内容哈希, 运行) 各记一行。数「属于某次运行（runs 里有这次运行）、属于这个源、SQL 里用到了这张表」的 - 行：同一次运行里结果一模一样的重复查询只算一次，工具库里的试用（不属于任何运行）不算。 + 使用次数（table_usage）和「这张表被哪些运行用过」（data/catalog_review.py）共用这一份取法：数据源查询工具每次成功 + 查询都存一份查询快照工件（tools/datasource.py），工件引用表 artifacts 里按 (内容哈希, 运行) 各记一行；只收属于某次运行 + （runs 里有这次运行）、属于这个源的行，工具库里的试用（不属于任何运行）不算。 - 表名在查询当时就记进了引用的 meta（query_snapshot_meta），这里只读 artifacts 一张表；升级前的老引用 - 没有 meta，退回读工件内容里的源名和 SQL，读过的按工件 id 缓存（内容永不改变）。SQL 里写的表名去掉 - schema 前缀后按 name_key（只把 ASCII 字母转小写，和 SQLite 比较标识符的口径一致）对到表结构上，对不上的不计。 + 表名在查询当时就记进了引用的 meta（query_snapshot_meta），这里只读 artifacts 一张表；升级前的老引用没有 meta，退回读 + 工件内容里的源名和 SQL，读过的按工件 id 缓存（内容永不改变）。SQL 里写的表名去掉 schema 前缀后按 name_key（只把 + ASCII 字母转小写，和 SQLite 比较标识符的口径一致）对到表结构上，对不上的不计。 """ rows = (await session.execute( - select(Artifact.id, Artifact.meta).join(Run, Run.id == Artifact.run_id) + select(Artifact.id, Artifact.meta, Artifact.run_id).join(Run, Run.id == Artifact.run_id) .where(Artifact.kind == QUERY_SNAPSHOT_KIND) )).all() tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {} lookup: dict[str, str] = {} for name in tables: lookup.setdefault(name_key(name), name) - counts: Counter[str] = Counter() - for artifact_id, meta in rows: + out: list[tuple[str, frozenset[str]]] = [] + for artifact_id, meta, run_id in rows: meta = meta if isinstance(meta, dict) else {} if isinstance(meta.get("tables"), list): source_id, source_name, names = meta.get("source_id"), meta.get("source"), meta["tables"] @@ -2281,8 +2281,21 @@ async def table_usage(session: AsyncSession, source: Any) -> dict[str, int]: continue elif source_name != source.name: continue - hit = {lookup.get(name_key(str(n).rsplit(".", 1)[-1])) for n in names} - counts.update(t for t in hit if t is not None) + hit = frozenset(t for t in (lookup.get(name_key(str(n).rsplit(".", 1)[-1])) for n in names) if t is not None) + if hit: + out.append((str(run_id), hit)) + return out + + +async def table_usage(session: AsyncSession, source: Any) -> dict[str, int]: + """每张表被运行查询过的次数：{schema_cache 的表名: 次数}，没被查过的表不出现。 + + 按查询快照引用的行数（query_snapshot_tables）：同一次运行里结果一模一样的重复查询只算一次，同一次运行里不同的查询 + 各算一次，工具库里的试用不算。 + """ + counts: Counter[str] = Counter() + for _run, hit in await query_snapshot_tables(session, source): + counts.update(hit) return dict(counts) @@ -2337,7 +2350,8 @@ __all__ = [ "fk_relations", "frozen_catalog", "infer_name_relations", "initial_status", "is_date_column", "is_profile_note", "join_paths", "make_item", "merge_notes", "name_words", "now_iso", "parse_path", "plan_patch", "profile_relation_holds", - "query_snapshot_meta", "read_catalog", "read_entry", "relation_graph", "relation_id", "render_table_index", + "query_snapshot_meta", "query_snapshot_tables", "read_catalog", "read_entry", "relation_graph", "relation_id", + "render_table_index", "render_table_notes", "resolve_draft_model", "resolve_table_name", "review_entry", "review_item", "same_notes", "same_profile_note", "save_human_edit", "save_patch", "status_counts", "structure_relation_ids", "system_notes_source", "table_usage", "unreviewed_status", "validate_notes", "visible_notes", "write_entry", diff --git a/backend/app/data/catalog_review.py b/backend/app/data/catalog_review.py new file mode 100644 index 00000000..b03c9e31 --- /dev/null +++ b/backend/app/data/catalog_review.py @@ -0,0 +1,404 @@ +"""数据目录的审阅队列、批量审阅、表被哪些运行用过、审阅进度。 + +**为什么要有它。** 起草（数据库注释、外键、命名推断、数据剖析、模型）一次写出成百上千项，绝大多数停在「推断」：SQL 检查的 +错误级规则只认已确认、有确证的项，推断的项只作提示。逐表点开确认太慢，也看不出先确认哪几项。队列把推断的项摊平，按 +「这张表常不常用 × 这一项错了影响多大 × 这一项从哪来」排好，最该先看的在前面；勾几项一次确认、驳回或撤销审阅。 + +- **队列**（review_queue，GET /catalog/review-queue）：只列推断（proposed）的项。priority = 使用次数权重 × 风险权重 × + 来源权重，从高到低；同分按表名、路径排，结果稳定。 + - 使用次数权重：1 + log2(1 + 使用次数)。使用次数同目录清单（catalog.table_usage）：按对数增长，常用表在前，但不至于 + 让常用表上的中文名压过所有表上影响数字的项。 + - 风险权重（RISK_WEIGHT）：有效记录条件、去重规则、业务日期、关联关系最高（直接改变数字），业务主键、度量类型次之 + （计数和能不能加总），码值、粒度、表类型再次，中文名、说明、含义、单位最低。 + - 来源权重（SOURCE_WEIGHT）：模型起草、命名推断是猜的，比数据库注释优先；数据剖析（码值候选、业务日期、没核实的关系） + 居中。 + - 两组（kind）：facts 影响数字的项（有效记录条件、去重规则、业务日期、关联关系、业务主键、度量类型）；names 名称和说明 + （中文名、说明、含义、单位、码值、粒度、表类型）。表的「说明」和中文名、含义一样只是说明文字，归 names。 + - 表结构里已经没有的表、列上的项不列（确认了也没有用，目录页上另有清理的提示）。 + - reasons 用中文说为什么排在这里：常用表（使用 N 次）、影响数字的项（有效记录条件）、模型起草，未确认…… +- **批量审阅**（review_batch，POST /catalog/review/batch）：逐表按 if_version 乐观锁，同一张表的几条在一次写入里完成， + 一张表冲突、写不进去不影响别的表；路径不对、表不存在逐条报（results 和请求一一对应）。审阅的规则同单项审阅 + （catalog.review_item）。 +- **被哪些运行用过**（table_runs，GET /catalog/{table}/usage）：取法同使用次数（catalog.query_snapshot_tables），按运行 + 去重，新的在前。 +- **进度**（review_progress，GET /catalog/review-progress）：使用次数前 TOP_N 张表（只算被运行用到过的）、全部、影响 + 数字的项各自的确认情况。项的范围和队列一样：表结构里没有的表、列不算。 +""" +from __future__ import annotations + +import math +from collections import Counter +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.data import catalog +from app.db.models import Run + +#: 队列的筛选：全部、影响数字的项、名称和说明 +KINDS = ("all", "facts", "names") +#: 队列一页最多几项；批量审阅一次最多几条；「被哪些运行用过」一次最多几次运行 +QUEUE_LIMIT_MAX = 200 +BATCH_MAX = 200 +RUNS_LIMIT_MAX = 100 +#: 进度里的「常用表」取使用次数前几张 +TOP_N = 20 + +#: 每一项属于哪一组、风险几档：键是 (t, 字段) / (c, 字段) / (r, "")，档 4 最高 +FIELD_GROUP: dict[tuple[str, str], tuple[str, int]] = { + ("t", "valid_filter"): ("facts", 4), ("t", "dedup"): ("facts", 4), ("t", "business_date"): ("facts", 4), + ("r", ""): ("facts", 4), + ("t", "keys"): ("facts", 3), ("c", "measure"): ("facts", 3), + ("c", "codes"): ("names", 2), ("t", "grain"): ("names", 2), ("t", "kind"): ("names", 2), + ("t", "label"): ("names", 1), ("t", "description"): ("names", 1), ("c", "label"): ("names", 1), + ("c", "meaning"): ("names", 1), ("c", "unit"): ("names", 1), +} +RISK_WEIGHT = {4: 4.0, 3: 3.0, 2: 2.0, 1: 1.0} +#: 来源权重：没写的（外键约束、人工填写，它们本来就不是推断）按 1 +SOURCE_WEIGHT = {"llm": 1.5, "name": 1.5, "profile": 1.25, "comment": 1.0} + +#: 来源的界面叫法（和 frontend/src/lib/terms.ts 的 CATALOG_SOURCE_LABEL 一致，有测试核对） +SOURCE_LABEL = {"comment": "数据库注释", "fk": "外键约束", "name": "命名推断", "profile": "数据剖析", "llm": "模型起草", + "human": "人工填写"} +RELATION_LABEL = "关联关系" +#: reasons 的说法 +REASON_TOP = "常用表（使用 {n} 次）" +REASON_USED = "使用 {n} 次" +REASON_RISK = {4: "影响数字的项（{label}）", 3: "影响计数和汇总的项（{label}）", 2: "帮助理解取值的项（{label}）", + 1: "名称和说明（{label}）"} +REASON_PENDING_CODES = "{n} 个码值的含义待填写" +REASON_SOURCE = "{source}，未确认" +#: 码值含义为空（数据剖析写进来的候选）时的说法，同界面的 CODES_TEXT.pending +CODE_PENDING = "含义待填写" + +#: 批量审阅逐条结果的机读码 +CODE_CONFLICT = "catalog_conflict" +CODE_PATH = "catalog_path_invalid" +CODE_INVALID = "catalog_invalid" +CODE_TABLE_MISSING = "table_missing" + +Slot = catalog.Slot + + +# ========================================================================== +# 一项的路径、叫法、显示 +# ========================================================================== + + +def slot_path(key: Slot) -> str: + """槽位 → 审阅接口的路径写法（catalog.parse_path 的反过来）。""" + if key[0] == "t": + return key[1] + if key[0] == "c": + return f"columns.{key[1]}.{key[2]}" + return f"relations.{key[1]}" + + +def field_of(key: Slot) -> tuple[str, str]: + """槽位 → FIELD_GROUP 的键。""" + if key[0] == "r": + return ("r", "") + return (key[0], key[-1]) + + +def field_name(key: Slot) -> str: + """这一项的字段名：表级、列级的项是字段名，关联关系是 relation。""" + return "relation" if key[0] == "r" else key[-1] + + +def field_label(key: Slot) -> str: + """字段的界面叫法：中文名、码值、有效记录条件、关联关系……""" + if key[0] == "t": + return catalog.TABLE_FIELD_LABEL.get(key[1], key[1]) + if key[0] == "c": + return catalog.COLUMN_FIELD_LABEL.get(key[2], key[2]) + return RELATION_LABEL + + +def item_value(key: Slot, item: Mapping[str, Any]) -> Any: + """一项的值：关联关系是两端、基数和覆盖率，别的是 value。""" + if key[0] == "r": + return {k: item.get(k) for k in ("columns", "to_table", "to_columns", "cardinality", "coverage")} + return item.get("value") + + +def _percent(ratio: float) -> str: + """覆盖率写成百分数：整数不带小数，其余留一位；不到 100% 的不写成 100%。""" + text = f"{ratio * 100:.1f}".rstrip("0").rstrip(".") + return "99.9%" if text == "100" and ratio < 1 else f"{text}%" + + +def relation_text(rel: Mapping[str, Any]) -> str: + """关联关系写成一行：member_id → members.id（多对一，覆盖率 98%）。""" + left = "、".join(str(c) for c in rel.get("columns") or []) + right = "、".join(f"{rel.get('to_table')}.{c}" for c in rel.get("to_columns") or []) + extra = [] + if rel.get("cardinality") in catalog.CARDINALITY_LABEL: + extra.append(catalog.CARDINALITY_LABEL[rel["cardinality"]]) + coverage = rel.get("coverage") + if isinstance(coverage, (int, float)) and not isinstance(coverage, bool): + extra.append(f"覆盖率 {_percent(float(coverage))}") + return f"{left} → {right}" + (f"（{'，'.join(extra)}）" if extra else "") + + +def display(key: Slot, value: Any) -> str: + """一项的值写成一行字，用界面上的说法（明细表、可累加、多对一……）。""" + if key[0] == "r": + return relation_text(value if isinstance(value, Mapping) else {}) + field_ = key[-1] + if field_ == "kind": + return catalog.UI_KIND_LABEL.get(value, str(value)) + if field_ == "measure": + return catalog.UI_MEASURE_LABEL.get(value, str(value)) + if field_ == "keys" and isinstance(value, list): + return "、".join(str(v) for v in value) + if field_ == "codes" and isinstance(value, Mapping): + return "、".join(f"{k}={v if str(v).strip() else CODE_PENDING}" for k, v in value.items()) + if field_ == "business_date" and isinstance(value, Mapping): + parts = [str(value.get("column") or "")] + if value.get("rule"): + parts.append(f"规则：{value['rule']}") + if value.get("timezone"): + parts.append(f"时区：{value['timezone']}") + return "；".join(parts) + return value if isinstance(value, str) else str(value) + + +# ========================================================================== +# 排序 +# ========================================================================== + + +def usage_weight(usage: int) -> float: + return 1.0 + math.log2(1 + max(0, int(usage or 0))) + + +def priority(usage: int, key: Slot, source: Any) -> float: + """一项的优先级：使用次数权重 × 风险权重 × 来源权重（不取整；排序和交给界面时取四位小数）。""" + _, tier = FIELD_GROUP[field_of(key)] + return usage_weight(usage) * RISK_WEIGHT[tier] * SOURCE_WEIGHT.get(str(source), 1.0) + + +def top_tables(tables: Mapping[str, Any], usage: Mapping[str, int]) -> list[str]: + """使用次数前 TOP_N 张表：只算被运行用到过的，同样多的按表结构里的顺序。""" + order = {name: i for i, name in enumerate(tables)} + used = [name for name in tables if usage.get(name, 0) > 0] + return sorted(used, key=lambda n: (-usage[n], order[n]))[:TOP_N] + + +def _reasons(key: Slot, item: Mapping[str, Any], usage: int, top: bool) -> list[str]: + out: list[str] = [] + if usage > 0: + out.append((REASON_TOP if top else REASON_USED).format(n=f"{usage:,}")) + _, tier = FIELD_GROUP[field_of(key)] + out.append(REASON_RISK[tier].format(label=field_label(key))) + if key[0] == "c" and key[2] == "codes" and isinstance(item.get("value"), Mapping): + pending = sum(1 for v in item["value"].values() if not str(v).strip()) + if pending: + out.append(REASON_PENDING_CODES.format(n=pending)) + source = str(item.get("source") or "") + out.append(REASON_SOURCE.format(source=SOURCE_LABEL.get(source, source))) + return out + + +def live_slots(notes: Mapping[str, Any] | None, meta: Mapping[str, Any] | None) -> list[tuple[Slot, dict[str, Any]]]: + """一张表目录里还算数的项：表结构里已经没有的列上的项不算（不知道表结构时都算）。""" + columns = catalog._column_names(meta) + return [(key, item) for key, item in catalog._slots(notes).items() + if not (key[0] == "c" and columns is not None and key[1] not in columns)] + + +# ========================================================================== +# 队列 +# ========================================================================== + + +async def review_queue(session: AsyncSession, source: Any, *, kind: str = "all", table: str | None = None, + limit: int = 50, offset: int = 0) -> dict[str, Any]: + """审阅队列：{total, items}。source 是读表结构用的源（上传源是绑定当前快照的视图）。""" + cache = getattr(source, "schema_cache", None) or {} + tables = cache.get("tables") or {} + wanted = None + if table: + wanted = catalog.resolve_table_name(cache, table) or table + entries = await catalog.read_catalog(session, source.id) + usage = await catalog.table_usage(session, source) + top = set(top_tables(tables, usage)) + items: list[dict[str, Any]] = [] + for name, entry in entries.items(): + meta = tables.get(name) + if meta is None or (wanted is not None and name != wanted): + continue + used = usage.get(name, 0) + for key, item in live_slots(entry.notes, meta): + if item.get("status") != "proposed": + continue + group, _ = FIELD_GROUP[field_of(key)] + if kind != "all" and group != kind: + continue + value = item_value(key, item) + items.append({ + "table_name": name, + "path": slot_path(key), + "field": field_name(key), + "field_label": field_label(key), + # 列级的项是哪一列（表级的项、关联关系为 null） + "column": key[1] if key[0] == "c" else None, + "value": value, + "display": display(key, value), + "source": item.get("source"), + "status": item.get("status"), + "note": item.get("note") if isinstance(item.get("note"), str) and item.get("note") else None, + "priority": round(priority(used, key, item.get("source")), 4), + "reasons": _reasons(key, item, used, name in top), + "table_usage": used, + "version": entry.version, + }) + items.sort(key=lambda i: (-i["priority"], i["table_name"], i["path"])) + return {"total": len(items), "items": items[offset:offset + limit]} + + +# ========================================================================== +# 批量审阅 +# ========================================================================== + + +@dataclass(frozen=True) +class BatchItem: + table_name: str + path: str + action: str + if_version: int + + +def _result(item: BatchItem, *, ok: bool, version: int | None = None, error: str | None = None, + code: str | None = None) -> dict[str, Any]: + return {"table_name": item.table_name, "path": item.path, "ok": ok, "version": version if ok else None, + "error": error, "code": code} + + +async def review_batch(session: AsyncSession, source: Any, items: Iterable[BatchItem], *, + actor: str | None) -> list[dict[str, Any]]: + """批量审阅，返回逐条结果（和请求一一对应）。 + + 表名按表结构认（全名、大小写不一致都认）；表结构里没有、但还留着目录的也认（同单项审阅）。同一张表的几条按请求里的 + 顺序在当前目录上依次审阅，一次写入：版本不符（if_version 和库里的不一样）、路径不对的那几条不做，其余照做；写入时撞上 + 别人刚改过，这张表的几条都记冲突。""" + items = list(items) + cache = getattr(source, "schema_cache", None) or {} + results: list[dict[str, Any] | None] = [None] * len(items) + groups: dict[str, list[int]] = {} + known: dict[str, str | None] = {} + for i, item in enumerate(items): + if item.table_name not in known: + key = catalog.resolve_table_name(cache, item.table_name) + if key is None and await catalog.read_entry(session, source.id, item.table_name) is not None: + key = item.table_name + known[item.table_name] = key + key = known[item.table_name] + if key is None: + results[i] = _result(item, ok=False, code=CODE_TABLE_MISSING, + error=f"数据源中没有表 {item.table_name}，可能已被删除或尚未探查结构") + continue + groups.setdefault(key, []).append(i) + at = catalog.now_iso() + for key, indexes in groups.items(): + entry = await catalog.read_entry(session, source.id, key) + current = entry.version if entry else 0 + notes: dict[str, Any] = entry.notes if entry else {} + applied: list[int] = [] + for i in indexes: + item = items[i] + if item.if_version != current: + results[i] = _result(item, ok=False, code=CODE_CONFLICT, error=str(catalog.CatalogConflict(key, current))) + continue + try: + notes = catalog.review_item(notes, item.path, item.action, at=at) + except catalog.CatalogPathError as e: + results[i] = _result(item, ok=False, code=CODE_PATH, error=str(e)) + continue + applied.append(i) + if not applied: + continue + try: + written = await catalog.write_entry(session, source.id, key, notes, if_version=current, actor=actor) + except catalog.CatalogConflict as e: + for i in applied: + results[i] = _result(items[i], ok=False, code=CODE_CONFLICT, error=str(e)) + continue + except catalog.CatalogInvalid as e: + for i in applied: + results[i] = _result(items[i], ok=False, code=CODE_INVALID, error=str(e)) + continue + for i in applied: + results[i] = _result(items[i], ok=True, version=written.version) + return [r for r in results if r is not None] + + +# ========================================================================== +# 被哪些运行用过 +# ========================================================================== + + +async def table_runs(session: AsyncSession, source: Any, table: str, *, limit: int = 20) -> dict[str, Any]: + """用过这张表的运行：{total, runs}，新的在前（按开始时间，没开始的按建立时间）。取法同使用次数。""" + run_ids = {run for run, hit in await catalog.query_snapshot_tables(session, source) if table in hit} + if not run_ids: + return {"total": 0, "runs": []} + started = func.coalesce(Run.started_at, Run.created_at) + rows = (await session.execute( + select(Run.id, Run.workflow_id, Run.workflow_name, Run.run_class, Run.status, Run.started_at) + .where(Run.id.in_(run_ids)).order_by(started.desc(), Run.id).limit(limit) + )).all() + return { + "total": len(run_ids), + "runs": [{"run_id": r.id, "workflow_id": r.workflow_id, "workflow_name": r.workflow_name or "", + "run_class": r.run_class, "status": r.status, + "started_at": r.started_at.isoformat() if r.started_at else None} for r in rows], + } + + +# ========================================================================== +# 进度 +# ========================================================================== + + +async def review_progress(session: AsyncSession, source: Any) -> dict[str, Any]: + """审阅进度：常用 TOP_N 张表（只算被运行用到过的）、全部、影响数字的项。项按状态计数；verified（有确证）另给一项。""" + tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {} + entries = await catalog.read_catalog(session, source.id) + usage = await catalog.table_usage(session, source) + top = set(top_tables(tables, usage)) + every: Counter[str] = Counter() + head: Counter[str] = Counter() + facts: Counter[str] = Counter() + for name, entry in entries.items(): + meta = tables.get(name) + if meta is None: + continue + for key, item in live_slots(entry.notes, meta): + status = str(item.get("status")) + every[status] += 1 + if name in top: + head[status] += 1 + if FIELD_GROUP[field_of(key)][0] == "facts": + facts[status] += 1 + return { + "top_n": TOP_N, + "top": {"tables": len(top), "items_total": sum(head.values()), "items_confirmed": head["confirmed"], + "items_proposed": head["proposed"], "items_verified": head["verified"]}, + "all": {"items_total": sum(every.values()), "items_confirmed": every["confirmed"], + "items_proposed": every["proposed"], "items_rejected": every["rejected"], + "items_verified": every["verified"]}, + "facts": {"proposed": facts["proposed"], "confirmed": facts["confirmed"]}, + } + + +__all__ = ["BATCH_MAX", "BatchItem", "CODE_CONFLICT", "CODE_INVALID", "CODE_PATH", "CODE_TABLE_MISSING", + "FIELD_GROUP", "KINDS", "QUEUE_LIMIT_MAX", "RISK_WEIGHT", "RUNS_LIMIT_MAX", "SOURCE_LABEL", "SOURCE_WEIGHT", + "TOP_N", "display", "field_label", "field_name", "field_of", "item_value", "live_slots", "priority", + "relation_text", "review_batch", "review_progress", "review_queue", "slot_path", "table_runs", "top_tables", + "usage_weight"] diff --git a/backend/tests/test_catalog_review_queue.py b/backend/tests/test_catalog_review_queue.py new file mode 100644 index 00000000..1598103c --- /dev/null +++ b/backend/tests/test_catalog_review_queue.py @@ -0,0 +1,357 @@ +"""数据目录的审阅队列、批量审阅、表被哪些运行用过、审阅进度（data/catalog_review.py）。 + +- GET /catalog/review-queue：只列推断（proposed）的项，按「使用次数权重 × 风险权重 × 来源权重」从高到低，同分按表名、 + 路径；facts（有效记录条件、去重规则、业务日期、关联关系、业务主键、度量类型）/ names（中文名、说明、含义、单位、码值、 + 粒度、表类型）两组；按表筛选；分页；reasons 写为什么排在这里；表结构里已经没有的表、列不列 +- POST /catalog/review/batch：逐表按 if_version 乐观锁，一张表冲突不影响别的表；同一张表的几条一次写入；路径不对、表不存在 + 逐条报 +- GET /catalog/{table}/usage：用过这张表的运行，新的在前，取法同 catalog.table_usage +- GET /catalog/review-progress：常用 20 张表、全部、影响数字的项各自的确认情况 + +表名、列名、业务数字都是虚构的。 +""" +from __future__ import annotations + +import re +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path +from urllib.parse import quote + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.data import catalog, catalog_review +from app.data.catalog import make_item +from app.db.base import SessionLocal +from app.db.models import Artifact, DataSource, Run +from app.main import app as _fastapi_app +from tests.fixtures.sources import drop_source + +ACTOR = {"X-Actor": quote("林晓")} +TERMS = Path(__file__).resolve().parents[2] / "frontend" / "src" / "lib" / "terms.ts" + + +def _col(name: str, type_: str = "INTEGER") -> dict: + return {"name": name, "type": type_, "nullable": True, "comment": None} + + +def _tables() -> dict: + return { + "orders": {"qualified": "orders", "primary_key": ["id"], "comment": None, "is_view": False, + "foreign_keys": [], "unique": [], + "columns": [_col("id"), _col("status"), _col("amount", "REAL"), _col("member_id"), + _col("paid_at", "TEXT")]}, + "members": {"qualified": "members", "primary_key": ["id"], "comment": None, "is_view": False, + "foreign_keys": [], "unique": [], + "columns": [_col("id"), _col("nick", "TEXT"), _col("tier")]}, + "coupons": {"qualified": "coupons", "primary_key": ["id"], "comment": None, "is_view": False, + "foreign_keys": [], "unique": [], "columns": [_col("id"), _col("face_value", "REAL")]}, + } + + +def _member_rel(status: str = "proposed", source: str = "name") -> dict: + return {"id": catalog.relation_id("orders", ["member_id"], "members", ["id"]), "columns": ["member_id"], + "to_table": "members", "to_columns": ["id"], "cardinality": "many_to_one", "coverage": 0.98, + "source": source, "status": status} + + +ORDERS = { + "label": make_item("订单", "llm"), + "grain": make_item("每笔订单一行", "llm", "confirmed"), + "valid_filter": make_item("status <> 9", "llm", note="剖析核对：排除 3.2% 的行（共 12,345 行）"), + "kind": make_item("fact", "llm"), + "columns": { + "status": {"label": make_item("状态", "comment"), "codes": make_item({"1": "已支付", "9": ""}, "profile")}, + "amount": {"measure": make_item("flow", "llm"), "unit": make_item("元", "llm", "rejected")}, + # 表结构里已经没有这一列：不进队列 + "legacy_flag": {"label": make_item("旧标记", "llm")}, + }, + "relations": [_member_rel()], +} +MEMBERS = {"label": make_item("会员", "llm"), "keys": make_item(["id"], "name"), + "columns": {"nick": {"label": make_item("昵称", "comment")}}} +COUPONS = {"label": make_item("优惠券", "comment")} +#: 表结构里已经没有这张表：不进队列 +GONE = {"label": make_item("旧表", "llm")} + +_CREATED: list[str] = [] + + +@pytest.fixture(autouse=True) +async def _cleanup(): + yield + while _CREATED: + await drop_source(_CREATED.pop()) + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c: + yield c + + +async def _write(sid: str, table: str, notes: dict) -> None: + async with SessionLocal() as session: + entry = await catalog.read_entry(session, sid, table) + await catalog.write_entry(session, sid, table, notes, if_version=entry.version if entry else 0, actor=None) + + +async def _use(sid: str, name: str, table_runs: dict[str, int]) -> list[str]: + """给每张表造几次运行查询它（每次运行一份查询快照），返回运行 id（先建的在前、开始得早）。""" + runs: list[str] = [] + base = datetime(2026, 9, 1, tzinfo=timezone.utc) + async with SessionLocal() as session: + for table, n in table_runs.items(): + for _ in range(n): + rid = uuid.uuid4().hex + session.add(Run(id=rid, graph={}, workflow_name=f"周报-{table}", run_class="formal", + status="succeeded", started_at=base + timedelta(hours=len(runs)))) + session.add(Artifact(id=uuid.uuid4().hex, run_id=rid, kind=catalog.QUERY_SNAPSHOT_KIND, + meta={"source_id": sid, "source": name, "tables": [table]})) + runs.append(rid) + await session.commit() + return runs + + +async def _source(*, usage: dict[str, int] | None = None) -> str: + name = f"rq_{uuid.uuid4().hex[:10]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=":memory:", schema_cache={"tables": _tables()}) + session.add(row) + await session.commit() + sid = row.id + _CREATED.append(sid) + for table, notes in (("orders", ORDERS), ("members", MEMBERS), ("coupons", COUPONS), ("gone_table", GONE)): + await _write(sid, table, notes) + if usage: + await _use(sid, name, usage) + return sid + + +def _base(sid: str) -> str: + return f"/api/datasources/{sid}/catalog" + + +# -------------------------------------------------------------------------- +# 排序规则（纯函数） +# -------------------------------------------------------------------------- + + +def test_priority_multiplies_usage_risk_and_source_weights(): + p = catalog_review.priority + # 风险：有效记录条件 > 业务主键 > 码值 > 中文名 + assert p(0, ("t", "valid_filter"), "llm") > p(0, ("t", "keys"), "llm") > p(0, ("c", "x", "codes"), "llm") \ + > p(0, ("c", "x", "label"), "llm") + assert p(0, ("r", "r1"), "name") == p(0, ("t", "dedup"), "name") == p(0, ("t", "business_date"), "name") + assert p(0, ("c", "x", "measure"), "llm") == p(0, ("t", "keys"), "llm") + assert p(0, ("t", "grain"), "llm") == p(0, ("t", "kind"), "llm") == p(0, ("c", "x", "codes"), "llm") + assert p(0, ("c", "x", "meaning"), "llm") == p(0, ("c", "x", "unit"), "llm") == p(0, ("t", "label"), "llm") + # 来源：模型起草、命名推断比数据库注释优先 + assert p(0, ("t", "label"), "llm") == p(0, ("t", "label"), "name") > p(0, ("t", "label"), "comment") + # 使用次数：用得多的在前，但按对数增长，不压过风险 + assert p(37, ("t", "label"), "llm") > p(1, ("t", "label"), "llm") > p(0, ("t", "label"), "llm") + assert p(37, ("t", "label"), "llm") == pytest.approx( + p(0, ("t", "label"), "llm") * p(37, ("t", "label"), "comment") / p(0, ("t", "label"), "comment")) + + +def test_source_labels_follow_the_ui(): + text = TERMS.read_text(encoding="utf-8") + block = re.search(r"export const CATALOG_SOURCE_LABEL\b[^=]*=\s*\{(.*?)\n\}", text, re.S) + assert block + ui = dict(re.findall(r"(\w+):\s*'([^']+)'", block.group(1))) + assert ui == catalog_review.SOURCE_LABEL + + +# -------------------------------------------------------------------------- +# 队列 +# -------------------------------------------------------------------------- + + +async def test_queue_lists_only_proposed_items_by_priority(client): + sid = await _source(usage={"orders": 37, "members": 2}) + r = await client.get(f"{_base(sid)}/review-queue") + assert r.status_code == 200, r.text + body = r.json() + items = body["items"] + paths = [(i["table_name"], i["path"]) for i in items] + # 已确认、已驳回的不列；表结构里没有的表、列不列 + assert ("orders", "grain") not in paths and ("orders", "columns.amount.unit") not in paths + assert not any(t == "gone_table" for t, _ in paths) + assert ("orders", "columns.legacy_flag.label") not in paths + assert body["total"] == len(items) == 11 + assert all(i["status"] == "proposed" for i in items) + # 从高到低，同分按表名、路径 + keys = [(-i["priority"], i["table_name"], i["path"]) for i in items] + assert keys == sorted(keys) + first = items[0] + rel_id = _member_rel()["id"] + assert (first["table_name"], first["path"]) in {("orders", "valid_filter"), ("orders", f"relations.{rel_id}")} + vf = next(i for i in items if i["path"] == "valid_filter") + assert vf["field"] == "valid_filter" and vf["field_label"] == "有效记录条件" + assert vf["value"] == "status <> 9" and vf["display"] == "status <> 9" + assert vf["source"] == "llm" and vf["note"] == "剖析核对：排除 3.2% 的行（共 12,345 行）" + assert vf["table_usage"] == 37 and vf["version"] == 1 + assert vf["reasons"] == ["常用表（使用 37 次）", "影响数字的项（有效记录条件）", "模型起草，未确认"] + # 常用表上的中文名排在没人用的表上的中文名前面；同一张表上影响数字的项排在中文名前面 + order = [(i["table_name"], i["path"]) for i in items] + assert order.index(("orders", "label")) < order.index(("coupons", "label")) + assert order.index(("members", "keys")) < order.index(("members", "label")) + + +async def test_queue_values_are_displayed_in_ui_words(client): + sid = await _source() + items = {(i["table_name"], i["path"]): i for i in (await client.get(f"{_base(sid)}/review-queue")).json()["items"]} + rel = items[("orders", f"relations.{_member_rel()['id']}")] + assert rel["field"] == "relation" and rel["field_label"] == "关联关系" + assert rel["value"] == {"columns": ["member_id"], "to_table": "members", "to_columns": ["id"], + "cardinality": "many_to_one", "coverage": 0.98} + assert rel["display"] == "member_id → members.id（多对一，覆盖率 98%）" + assert rel["reasons"][-1] == "命名推断，未确认" + kind = items[("orders", "kind")] + assert kind["display"] == "明细表" and kind["value"] == "fact" + codes = items[("orders", "columns.status.codes")] + assert codes["field"] == "codes" and codes["field_label"] == "码值" and codes["column"] == "status" + assert codes["display"] == "1=已支付、9=含义待填写" + assert "1 个码值的含义待填写" in codes["reasons"] and "数据剖析，未确认" in codes["reasons"] + assert items[("orders", "columns.amount.measure")]["display"] == "可累加" + assert items[("members", "keys")]["display"] == "id" + assert items[("members", "label")]["reasons"] == ["名称和说明（中文名）", "模型起草，未确认"] + assert items[("coupons", "label")]["reasons"] == ["名称和说明（中文名）", "数据库注释，未确认"] + + +async def test_queue_filters_by_kind_and_table_and_pages(client): + sid = await _source(usage={"orders": 3}) + facts = (await client.get(f"{_base(sid)}/review-queue", params={"kind": "facts"})).json() + assert {i["field"] for i in facts["items"]} == {"valid_filter", "relation", "measure", "keys"} + names = (await client.get(f"{_base(sid)}/review-queue", params={"kind": "names"})).json() + assert {i["field"] for i in names["items"]} == {"label", "codes", "kind"} + assert facts["total"] + names["total"] == 11 + only = (await client.get(f"{_base(sid)}/review-queue", params={"table": "MEMBERS"})).json() + assert {i["table_name"] for i in only["items"]} == {"members"} and only["total"] == 3 + everything = (await client.get(f"{_base(sid)}/review-queue")).json()["items"] + page = (await client.get(f"{_base(sid)}/review-queue", params={"limit": 4, "offset": 4})).json() + assert page["total"] == 11 and page["items"] == everything[4:8] + bad = await client.get(f"{_base(sid)}/review-queue", params={"kind": "other"}) + assert bad.status_code == 422 + assert (await client.get(f"{_base(sid)}/review-queue", params={"limit": 0})).status_code == 422 + missing = await client.get("/api/datasources/nope/catalog/review-queue") + assert missing.status_code == 404 + + +# -------------------------------------------------------------------------- +# 批量审阅 +# -------------------------------------------------------------------------- + + +async def test_batch_review_writes_each_table_once_and_isolates_conflicts(client): + sid = await _source() + rel_path = f"relations.{_member_rel()['id']}" + body = {"items": [ + {"table_name": "orders", "path": "valid_filter", "action": "confirm", "if_version": 1}, + {"table_name": "orders", "path": rel_path, "action": "reject", "if_version": 1}, + {"table_name": "orders", "path": "columns.nothing.label", "action": "confirm", "if_version": 1}, + {"table_name": "members", "path": "label", "action": "confirm", "if_version": 7}, # 版本不对 + {"table_name": "coupons", "path": "label", "action": "reject", "if_version": 1}, + {"table_name": "ghost", "path": "label", "action": "confirm", "if_version": 0}, + ]} + r = await client.post(f"{_base(sid)}/review/batch", json=body, headers=ACTOR) + assert r.status_code == 200, r.text + results = r.json()["results"] + assert [(x["table_name"], x["path"], x["ok"]) for x in results] == [ + ("orders", "valid_filter", True), ("orders", rel_path, True), ("orders", "columns.nothing.label", False), + ("members", "label", False), ("coupons", "label", True), ("ghost", "label", False)] + # 同一张表的两条一次写进去：版本只加 1 + assert results[0]["version"] == results[1]["version"] == 2 + assert results[2]["code"] == "catalog_path_invalid" and "找不到" in results[2]["error"] + assert results[2]["version"] is None + assert results[3]["code"] == "catalog_conflict" and "重新载入" in results[3]["error"] + assert results[4]["version"] == 2 and results[4]["error"] is None and results[4]["code"] is None + assert results[5]["code"] == "table_missing" + async with SessionLocal() as session: + orders = await catalog.read_entry(session, sid, "orders") + members = await catalog.read_entry(session, sid, "members") + coupons = await catalog.read_entry(session, sid, "coupons") + assert orders.version == 2 and orders.updated_by == "林晓" + assert orders.notes["valid_filter"]["status"] == "confirmed" + assert orders.notes["relations"][0]["status"] == "rejected" + assert members.version == 1 and members.notes["label"]["status"] == "proposed" + assert coupons.notes["label"]["status"] == "rejected" + # 撤销审阅：回到来源的初始状态 + again = await client.post(f"{_base(sid)}/review/batch", json={"items": [ + {"table_name": "orders", "path": "valid_filter", "action": "reset", "if_version": 2}]}) + assert again.json()["results"][0]["ok"] is True + async with SessionLocal() as session: + assert (await catalog.read_entry(session, sid, "orders")).notes["valid_filter"]["status"] == "proposed" + + +async def test_batch_review_limits(client): + sid = await _source() + many = [{"table_name": "orders", "path": "label", "action": "confirm", "if_version": 1}] * 201 + assert (await client.post(f"{_base(sid)}/review/batch", json={"items": many})).status_code == 422 + assert (await client.post(f"{_base(sid)}/review/batch", json={"items": []})).status_code == 422 + wrong = [{"table_name": "orders", "path": "label", "action": "approve", "if_version": 1}] + assert (await client.post(f"{_base(sid)}/review/batch", json={"items": wrong})).status_code == 422 + ok = [{"table_name": "orders", "path": "label", "action": "confirm", "if_version": 1}] * 200 + r = await client.post(f"{_base(sid)}/review/batch", json={"items": ok}) + assert r.status_code == 200 and all(x["ok"] for x in r.json()["results"]) + + +# -------------------------------------------------------------------------- +# 被哪些运行用过 +# -------------------------------------------------------------------------- + + +async def test_table_usage_lists_runs_newest_first(client): + sid = await _source() + async with SessionLocal() as session: + name = (await session.get(DataSource, sid)).name + runs = await _use(sid, name, {"orders": 3, "members": 1}) + # 同一次运行查了两次 orders：只算一次运行 + async with SessionLocal() as session: + session.add(Artifact(id=uuid.uuid4().hex, run_id=runs[0], kind=catalog.QUERY_SNAPSHOT_KIND, + meta={"source_id": sid, "source": name, "tables": ["orders", "members"]})) + # 别的数据源的同名表不算 + other = uuid.uuid4().hex + session.add(Run(id=other, graph={}, workflow_name="别处", status="succeeded")) + session.add(Artifact(id=uuid.uuid4().hex, run_id=other, kind=catalog.QUERY_SNAPSHOT_KIND, + meta={"source_id": "elsewhere", "source": "elsewhere", "tables": ["orders"]})) + await session.commit() + r = await client.get(f"{_base(sid)}/orders/usage") + assert r.status_code == 200, r.text + body = r.json() + assert body["total"] == 3 + assert [x["run_id"] for x in body["runs"]] == [runs[2], runs[1], runs[0]] + first = body["runs"][0] + assert first["workflow_name"] == "周报-orders" and first["run_class"] == "formal" + assert first["status"] == "succeeded" and first["started_at"].startswith("2026-09-01T02:00") + assert set(first) == {"run_id", "workflow_id", "workflow_name", "run_class", "status", "started_at"} + limited = (await client.get(f"{_base(sid)}/orders/usage", params={"limit": 1})).json() + assert limited["total"] == 3 and len(limited["runs"]) == 1 + members = (await client.get(f"{_base(sid)}/members/usage")).json() + assert members["total"] == 2 + assert (await client.get(f"{_base(sid)}/ghost/usage")).status_code == 404 + + +# -------------------------------------------------------------------------- +# 进度 +# -------------------------------------------------------------------------- + + +async def test_review_progress_counts_top_tables_all_and_facts(client): + sid = await _source(usage={"orders": 5}) + r = await client.get(f"{_base(sid)}/review-progress") + assert r.status_code == 200, r.text + body = r.json() + assert body["top_n"] == 20 + # 常用表只算被运行用到过的表：这里只有 orders（9 项：表结构里没有的列不算） + assert body["top"] == {"tables": 1, "items_total": 9, "items_confirmed": 1, "items_proposed": 7, + "items_verified": 0} + assert body["all"] == {"items_total": 13, "items_confirmed": 1, "items_proposed": 11, "items_rejected": 1, + "items_verified": 0} + # 影响数字的项：有效记录条件、关联关系、度量类型、业务主键 + assert body["facts"] == {"proposed": 4, "confirmed": 0} + await client.post(f"{_base(sid)}/review/batch", json={"items": [ + {"table_name": "orders", "path": "valid_filter", "action": "confirm", "if_version": 1}]}) + after = (await client.get(f"{_base(sid)}/review-progress")).json() + assert after["top"]["items_confirmed"] == 2 and after["facts"] == {"proposed": 3, "confirmed": 1}