diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index 7680eb54..b3554756 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -16,6 +16,8 @@ - GET /catalog/review-queue 审阅队列：推断的项按使用次数、风险、来源排好；POST /catalog/review/batch 批量审阅， GET /catalog/review-progress 审阅进度，GET /catalog/{table}/usage 用过这张表的运行 （data/catalog_review.py） +- GET /catalog/export 导出数据目录（JSON 附件）；POST /catalog/import/preview 导入预览（只读）， + POST /catalog/import 导入（data/catalog_transfer.py） 规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径 不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。 @@ -38,7 +40,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 import catalog_review, catalog_transfer from app.data.engine import SnapshotTampered from app.db.base import get_session from app.db.models import DataSource @@ -262,6 +264,68 @@ async def catalog_review_progress(source_id: str, session: AsyncSession = Depend return await catalog_review.review_progress(session, source) +# ---- 导出导入（导出同样要登记在 GET /catalog/{table} 前面） + + +@router.get("/{source_id}/catalog/export") +async def export_catalog(source_id: str, session: AsyncSession = Depends(get_session)) -> Any: + """导出这个数据源的数据目录：JSON 附件「<数据源名>-数据目录.json」，含全部状态的项（data/catalog_transfer.py）。""" + from fastapi.responses import JSONResponse + + row = await _source_row(session, source_id) + body = await catalog_transfer.export_catalog(session, row) + name = catalog_transfer.attachment_name(row.name) + return JSONResponse(body, headers={"Content-Disposition": catalog_transfer.content_disposition(name, row.id)}) + + +class CatalogImportIn(BaseModel): + #: 导出的那个对象（格式在 data/catalog_transfer.read_file 里核对，不对 422，说明是中文） + content: Any = None + #: 目标还没有起草要求时，是否写入文件里的起草要求 + include_guidance: bool = False + + +async def _import_source(session: AsyncSession, source_id: str) -> tuple[DataSource, Any]: + row, source = await _resolved(session, source_id) + cache = source.schema_cache or {} + if not cache.get("tables"): + raise CodedHTTPException(409, f"{introspect.why_empty(cache)}，无法导入数据目录。请先在数据源卡片上点「探查结构」", + SCHEMA_MISSING) + return row, source + + +@router.post("/{source_id}/catalog/import/preview") +async def preview_catalog_import(source_id: str, payload: CatalogImportIn, + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """导入预览（只读）：每张对上的表新增、更新、保留几项，冲突和跳过的逐条列出。文件格式不对 422；没有表结构 409。""" + row, source = await _import_source(session, source_id) + try: + report = await catalog_transfer.plan_import(session, row, source, payload.content, + include_guidance=payload.include_guidance) + except catalog_transfer.TransferInvalid as e: + raise HTTPException(422, str(e)) from e + return report.as_dict() + + +@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。""" + row, source = await _import_source(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: + report = await catalog_transfer.run_import(session, row, source, payload.content, + include_guidance=payload.include_guidance, + actor=actor_of(x_actor)) + except catalog_transfer.TransferInvalid as e: + raise HTTPException(422, str(e)) from e + return report.as_dict() + + @router.get("/{source_id}/catalog/{table}") async def get_catalog_table(source_id: str, table: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: diff --git a/backend/app/data/catalog_transfer.py b/backend/app/data/catalog_transfer.py new file mode 100644 index 00000000..34954f8d --- /dev/null +++ b/backend/app/data/catalog_transfer.py @@ -0,0 +1,484 @@ +"""数据目录的导出导入：本机整理好的目录带到部署实例上（GET /catalog/export、POST /catalog/import/preview、/import）。 + +**为什么要有它。** 目录（尤其是人工确认、驳回过的项）是在本机对着开发库一项项审出来的；部署实例连的是同一套业务库， +却没有办法把这些结论带过去，只能在那边从头再审一遍。导出是一个 JSON 文件，导入先预览、再确认。 + +**文件**（export_catalog）：{format: "agentlab-catalog", version: 1, exported_at, source: {name, kind}, guidance, +tables: {表名: {notes, version}}}，含全部状态的项，表结构里已经没有的表也带上（目录里有什么就导出什么）。 + +**导入的规矩**（plan_import，预览和导入共用这一份）： + +- 表按名字对：表结构里的写法、忽略大小写（name_key）、带 schema 前缀的对不上时去掉前缀再对（catalog.resolve_table_name）。 + 只对目标数据源表结构里的表；对不上的列在 unmatched_tables。文件里两张表对上目标的同一张时，写法完全一样的那张优先， + 另一张算对不上。列名同样按表结构的写法、再忽略大小写对；业务主键、业务日期、关联关系两端的列和指向的表都换成目标的写法， + 关联关系的编号按目标的表名、列名重算（catalog.relation_id）。 +- 文件里已确认、已驳回的项照带：来源、状态、备注、修改时间都不变。目标那一格空着就新增；目标是推断、有确证的就更新； + 目标也是已确认、已驳回的，和文件一样就保留，不一样（值或状态不同）就保留目标的、记一条冲突。关联关系的「一样」看两端和 + 基数，覆盖率是测量值、不算。 +- 文件里推断、有确证的项只填目标的空缺，目标那一格有东西（哪怕是驳回）就保留目标的，不算冲突。 +- 跳过并写原因：表结构里没有的列、业务主键和业务日期指着的列没有的、关联关系两端对不上的（指向的表、两端的列）、 + 目标数据源遮罩的列上的码值（码值就是这一列的取值，遮罩了就不该进目录）、不认识的字段、取值不合规的项（说法同 + catalog.validate_notes）。 +- 逐表写（乐观锁，撞上别人刚改过就重读重算，最多三次），一张表写不进去只记在那张表上，不影响别的表。 +- 起草要求：勾了 include_guidance、文件里有、目标还没有时才写（action: replace）；目标已有的不覆盖（keep）；没勾或文件里 + 没有是 none。 +- 文件格式不对（不是对象、格式标记或版本不对、表的目录不是对象、起草要求不是文字或超长）抛 TransferInvalid，接口回 422。 + +有排队或进行中的起草任务时不导入（接口先调 catalog_jobs.ensure_idle，409 catalog_draft_busy）；目标数据源没有表结构时 +接口回 409（什么也对不上）。 +""" +from __future__ import annotations + +import copy +import re +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.data import catalog, catalog_jobs +from app.data.catalog_review import field_label, slot_path +from app.data.engine import masked_columns +from app.data.names import name_key + +FORMAT = "agentlab-catalog" +VERSION = 1 +FILE_SUFFIX = "-数据目录.json" +_BAD_FILENAME = re.compile(r'[\\/:*?"<>|\x00-\x1f\x7f]') +_NAME_MAX = 80 +_UNNAMED = "数据源" + +#: 状态：照带的、只填空缺的 +_LOCKED = ("confirmed", "rejected") +_STATUS_LABEL = {"proposed": "推断", "verified": "有确证", "confirmed": "已确认", "rejected": "已驳回"} + +#: 写入一直撞上别人刚改过（重读重算三次仍冲突） +WRITE_CONFLICT_ERROR = "这张表的数据目录正被其他人修改，没有导入，请稍后重新导入" +WRITE_ERROR = "写入这张表的数据目录时出错，没有导入：{line}" + + +class TransferInvalid(ValueError): + """文件格式不对。message 给人看（接口回 422）。""" + + +# ========================================================================== +# 导出 +# ========================================================================== + + +def attachment_name(source_name: str) -> str: + """附件名「<数据源名>-数据目录.json」。文件名里不能有的字换成下划线。""" + name = _BAD_FILENAME.sub("_", source_name or "").strip(" .")[:_NAME_MAX].strip(" .") + return f"{name or _UNNAMED}{FILE_SUFFIX}" + + +def content_disposition(filename: str, source_id: str) -> str: + """附件头：中文名按 RFC 5987 写在 filename* 里；filename 给不认 filename* 的老客户端一个纯 ASCII 的名字。""" + from urllib.parse import quote + + ascii_id = re.sub(r"[^A-Za-z0-9]", "", source_id)[:8] + return f"attachment; filename=\"agentlab-catalog-{ascii_id}.json\"; filename*=UTF-8''{quote(filename, safe='')}" + + +async def export_catalog(session: AsyncSession, row: Any) -> dict[str, Any]: + """导出这个数据源的数据目录（row 是数据源记录，目录按它的 id 存）。""" + entries = await catalog.read_catalog(session, row.id) + return { + "format": FORMAT, + "version": VERSION, + "exported_at": catalog.now_iso(), + "source": {"name": row.name, "kind": row.kind}, + "guidance": catalog_jobs.guidance_of(row), + "tables": {name: {"notes": entry.notes, "version": entry.version} for name, entry in sorted(entries.items())}, + } + + +# ========================================================================== +# 读文件 +# ========================================================================== + + +@dataclass(frozen=True) +class ImportFile: + tables: dict[str, dict[str, Any]] + guidance: str | None + + +def read_file(content: Any, *, include_guidance: bool) -> ImportFile: + """导出的那个对象 → ImportFile。格式不对抛 TransferInvalid（说法给人看）。""" + if not isinstance(content, Mapping): + raise TransferInvalid("文件内容不是数据目录导出的格式，请选择从 AgentLab 导出的数据目录文件") + if content.get("format") != FORMAT: + raise TransferInvalid("这个文件不是 AgentLab 导出的数据目录（文件里的格式标记不对），请重新选择文件") + version = content.get("version") + if version != VERSION: + raise TransferInvalid(f"这个数据目录文件的格式版本是「{version}」，当前只能导入第 {VERSION} 版的文件，" + "请用同一版本的 AgentLab 重新导出") + tables = content.get("tables") + if not isinstance(tables, Mapping): + raise TransferInvalid("数据目录文件里没有表的目录，文件可能不完整，请重新导出") + out: dict[str, dict[str, Any]] = {} + for name, body in tables.items(): + notes = body.get("notes") if isinstance(body, Mapping) else None + if not isinstance(name, str) or not name.strip() or not isinstance(notes, Mapping): + raise TransferInvalid(f"数据目录文件里表「{str(name)[:60]}」的目录格式不正确，文件可能被改动过，请重新导出") + out[name] = copy.deepcopy(dict(notes)) + guidance = content.get("guidance") + if guidance is not None and not isinstance(guidance, str): + raise TransferInvalid("数据目录文件里的起草要求格式不正确，文件可能被改动过，请重新导出") + text = guidance.strip() if isinstance(guidance, str) else "" + if include_guidance and len(text) > catalog_jobs.GUIDANCE_MAX: + raise TransferInvalid(f"数据目录文件里的起草要求有 {len(text)} 字，超过了 {catalog_jobs.GUIDANCE_MAX} 字的上限，" + "请删减后再导入，或者不导入起草要求") + return ImportFile(tables=out, guidance=text or None) + + +# ========================================================================== +# 对表、对列 +# ========================================================================== + + +def match_tables(file_tables: list[str], schema_cache: Mapping[str, Any]) -> tuple[dict[str, str], list[str]]: + """文件里的表 → 目标表结构里的键。返回 ({文件里的表名: 目标表名}, 对不上的表名)。写法完全一样的先认领。""" + tables = (schema_cache or {}).get("tables") or {} + matched: dict[str, str] = {} + claimed: set[str] = set() + ordered = sorted(file_tables, key=lambda n: n not in tables) + for name in ordered: + key = catalog.resolve_table_name(schema_cache, name) + if key is None or key in claimed: + continue + matched[name] = key + claimed.add(key) + unmatched = [name for name in file_tables if name not in matched] + return {name: matched[name] for name in file_tables if name in matched}, unmatched + + +class _Columns: + """一张目标表的列：按表结构的写法认，再忽略大小写认（只认唯一对上的）。""" + + def __init__(self, meta: Mapping[str, Any] | None) -> None: + names = [str(c.get("name")) for c in (meta or {}).get("columns") or [] if isinstance(c, Mapping) + and c.get("name") is not None] + self.names = set(names) + self._loose: dict[str, list[str]] = {} + for name in names: + self._loose.setdefault(name_key(name), []).append(name) + + def find(self, name: Any) -> str | None: + if not isinstance(name, str): + return None + if name in self.names: + return name + found = self._loose.get(name_key(name)) or [] + return found[0] if len(found) == 1 else None + + +# ========================================================================== +# 一张表怎么并 +# ========================================================================== + + +@dataclass +class TablePlan: + """文件里一张表并进目标的计划。writes 是要写进目标的项（目标的槽位 → 项）。""" + + table_name: str + matched_as: str + add: int = 0 + update: int = 0 + keep: int = 0 + conflicts: list[dict[str, Any]] = field(default_factory=list) + skipped: list[dict[str, Any]] = field(default_factory=list) + writes: dict[catalog.Slot, dict[str, Any]] = field(default_factory=dict) + version: int = 0 + error: str | None = None + + def as_dict(self) -> dict[str, Any]: + return {"table_name": self.table_name, "matched_as": self.matched_as, "add": self.add, "update": self.update, + "keep": self.keep, "conflicts": self.conflicts, "skipped": self.skipped, "version": self.version, + "error": self.error} + + +def _raw_items(notes: Mapping[str, Any]) -> list[tuple[str, catalog.Slot | None, Any, str | None]]: + """文件里一张表的每一项：(文件里的路径, 槽位, 项, 不认的原因)。不认识的字段、不是对象的列也逐个列出来（槽位为 None）。""" + out: list[tuple[str, catalog.Slot | None, Any, str | None]] = [] + for name, item in notes.items(): + if name in catalog.TABLE_FIELDS: + out.append((name, ("t", name), item, None)) + elif name not in ("columns", "relations"): + out.append((str(name), None, item, f"数据目录里不认识的字段「{name}」")) + columns = notes.get("columns", {}) + if not isinstance(columns, Mapping): + out.append(("columns", None, columns, "列的目录应为以列名为键的对象")) + columns = {} + for col, items in columns.items(): + if not isinstance(items, Mapping): + out.append((f"columns.{col}", None, items, f"列 {col} 的目录应为对象")) + continue + for name, item in items.items(): + path = f"columns.{col}.{name}" + if name in catalog.COLUMN_FIELDS: + out.append((path, ("c", str(col), name), item, None)) + else: + out.append((path, None, item, f"列 {col} 含有不认识的字段「{name}」")) + relations = notes.get("relations", []) + if not isinstance(relations, list): + out.append(("relations", None, relations, "关联关系应为列表")) + relations = [] + for rel in relations: + rid = rel.get("id") if isinstance(rel, Mapping) else None + out.append((f"relations.{rid}", ("r", str(rid)), rel, None)) + return out + + +def _item_problem(key: catalog.Slot, item: Any) -> str | None: + """这一项本身合不合规（说法同 catalog.validate_notes）。""" + if key[0] == "t": + probe: dict[str, Any] = {key[1]: item} + elif key[0] == "c": + probe = {"columns": {key[1]: {key[2]: item}}} + else: + probe = {"relations": [item]} + problems = catalog.validate_notes(probe) + return problems[0] if problems else None + + +def _decision(key: catalog.Slot, item: Mapping[str, Any]) -> Any: + """判断「一样不一样」看的部分：状态，加上值（码值另看「已列全」）。关联关系的两端由编号定了，再看基数；覆盖率不算。""" + if key[0] == "r": + return item.get("status"), item.get("cardinality") + if key[0] == "c" and key[2] == "codes": + return item.get("status"), item.get("value"), catalog.codes_complete(item) + return item.get("status"), item.get("value") + + +def _conflict_why(file_status: str, target_status: str) -> str: + if target_status == "rejected": + return f"目标里这一项已经驳回，文件里是{_STATUS_LABEL.get(file_status, file_status)}，保留目标的" + if file_status == "rejected": + return "目标里这一项已经人工确认过，文件里是已驳回，保留目标的" + return "目标里这一项已经人工确认过，和文件里的不一样，保留目标的" + + +class _Mapper: + """把文件里的一项换成目标的写法：列名、业务主键和业务日期的列、关联关系两端和编号。对不上返回原因。""" + + def __init__(self, target: str, schema_cache: Mapping[str, Any], masked: set[str]) -> None: + self.target = target + self.cache = schema_cache + self.tables = (schema_cache or {}).get("tables") or {} + self.columns = _Columns(self.tables.get(target)) + self.masked = masked + + def __call__(self, key: catalog.Slot, item: Any) -> tuple[catalog.Slot | None, Any, str | None]: + if not isinstance(item, Mapping): + return None, item, None # 交给合规检查说 + item = copy.deepcopy(dict(item)) + if key[0] == "c": + col = self.columns.find(key[1]) + if col is None: + return None, item, f"目标的表结构里没有列 {key[1]}，跳过" + if key[2] == "codes" and col.lower() in self.masked: + return None, item, f"列 {col} 在目标数据源的设置中被遮罩，码值是这一列的取值，不导入" + return ("c", col, key[2]), item, None + if key[0] == "t": + value = item.get("value") + if key[1] == "keys" and isinstance(value, list): + mapped = [self.columns.find(c) for c in value] + if missing := [str(c) for c, m in zip(value, mapped) if m is None]: + return None, item, f"业务主键里的列「{'、'.join(missing)}」不在目标的表结构里，跳过" + item["value"] = mapped + elif key[1] == "business_date" and isinstance(value, Mapping) and isinstance(value.get("column"), str): + col = self.columns.find(value["column"]) + if col is None: + return None, item, f"业务日期的列「{value['column']}」不在目标的表结构里，跳过" + item["value"] = {**value, "column": col} + return key, item, None + return self._relation(item) + + def _relation(self, rel: dict[str, Any]) -> tuple[catalog.Slot | None, Any, str | None]: + cols, to_table, to_cols = rel.get("columns"), rel.get("to_table"), rel.get("to_columns") + if not (isinstance(cols, list) and isinstance(to_cols, list) and isinstance(to_table, str)): + return None, rel, None + target_table = catalog.resolve_table_name(self.cache, to_table) + if target_table is None: + return None, rel, f"关联关系指向的表 {to_table} 不在目标数据源的表结构里，跳过" + mine = [self.columns.find(c) for c in cols] + if missing := [str(c) for c, m in zip(cols, mine) if m is None]: + return None, rel, f"关联关系的本表字段「{'、'.join(missing)}」不在目标的表结构里，跳过" + theirs_columns = _Columns(self.tables.get(target_table)) + theirs = [theirs_columns.find(c) for c in to_cols] + if missing := [str(c) for c, m in zip(to_cols, theirs) if m is None]: + return None, rel, f"关联关系的目标字段「{'、'.join(missing)}」不在 {target_table} 的表结构里，跳过" + rid = catalog.relation_id(self.target, mine, target_table, theirs) + return ("r", rid), {**rel, "id": rid, "columns": mine, "to_table": target_table, "to_columns": theirs}, None + + +def plan_table(table_name: str, target: str, file_notes: Mapping[str, Any], target_notes: Mapping[str, Any] | None, + schema_cache: Mapping[str, Any], masked: set[str]) -> TablePlan: + """文件里一张表并进目标那张表的计划（纯函数，不写库）。""" + plan = TablePlan(table_name=table_name, matched_as=target) + existing = catalog._slots(target_notes) + mapper = _Mapper(target, schema_cache, masked) + seen: set[catalog.Slot] = set() + for path, key, item, why in _raw_items(file_notes): + if why is not None or key is None: + plan.skipped.append({"path": path, "why": why or "无法识别这一项"}) + continue + mapped, item, why = mapper(key, item) + if why is not None: + plan.skipped.append({"path": path, "why": why}) + continue + problem = _item_problem(mapped or key, item) + if problem is not None or mapped is None: + plan.skipped.append({"path": path, "why": problem or "无法识别这一项"}) + continue + if mapped in seen: + plan.skipped.append({"path": path, "why": f"和文件里前面的{catalog.describe_slot(mapped)}是同一项，只导入第一项"}) + continue + seen.add(mapped) + status = item.get("status") + have = existing.get(mapped) + if have is None: + plan.add += 1 + plan.writes[mapped] = item + elif status not in _LOCKED: + plan.keep += 1 # 推断、有确证的只填空缺 + elif have.get("status") in _LOCKED: + if _decision(mapped, have) == _decision(mapped, item): + plan.keep += 1 + else: + plan.conflicts.append({ + "path": slot_path(mapped), "label": _label(mapped, item), + "file_value": _shown_value(mapped, item), "target_value": _shown_value(mapped, have), + "file_status": status, "target_status": have.get("status"), + "why": _conflict_why(str(status), str(have.get("status"))), + }) + elif _decision(mapped, have) == _decision(mapped, item): + plan.keep += 1 + else: + plan.update += 1 + plan.writes[mapped] = item + return plan + + +def _label(key: catalog.Slot, item: Mapping[str, Any]) -> str: + if key[0] == "t": + return f"表的{field_label(key)}" + if key[0] == "c": + return f"列 {key[1]} 的{field_label(key)}" + return f"指向 {item.get('to_table')} 的关联关系" + + +def _shown_value(key: catalog.Slot, item: Mapping[str, Any]) -> Any: + if key[0] == "r": + return {k: item.get(k) for k in ("columns", "to_table", "to_columns", "cardinality", "coverage")} + return item.get("value") + + +# ========================================================================== +# 预览、导入 +# ========================================================================== + + +@dataclass +class ImportReport: + tables: list[TablePlan] + unmatched: list[str] + guidance: dict[str, Any] + + def as_dict(self) -> dict[str, Any]: + return { + "tables": [t.as_dict() for t in self.tables], + "unmatched_tables": self.unmatched, + "totals": {"tables": len(self.tables), "add": sum(t.add for t in self.tables), + "update": sum(t.update for t in self.tables), "keep": sum(t.keep for t in self.tables), + "conflicts": sum(len(t.conflicts) for t in self.tables), + "skipped": sum(len(t.skipped) for t in self.tables)}, + "guidance": self.guidance, + } + + +def _guidance(file: ImportFile, row: Any, include: bool) -> dict[str, Any]: + target = catalog_jobs.guidance_of(row) + if not include or file.guidance is None: + action = "none" + elif target: + action = "keep" + else: + action = "replace" + return {"file": file.guidance, "target": target, "action": action} + + +async def plan_import(session: AsyncSession, row: Any, source: Any, content: Any, *, + include_guidance: bool) -> ImportReport: + """预览：逐表算出要新增、更新、保留什么，冲突和跳过的项及原因。不写库。row 是数据源记录（遮罩、起草要求按它）， + source 是读表结构用的源（上传源是绑定当前快照的视图）。格式不对抛 TransferInvalid。""" + file = read_file(content, include_guidance=include_guidance) + cache = getattr(source, "schema_cache", None) or {} + matched, unmatched = match_tables(list(file.tables), cache) + masked = {c.lower() for c in masked_columns(getattr(row, "options", None))} + plans: list[TablePlan] = [] + for name, target in matched.items(): + entry = await catalog.read_entry(session, source.id, target) + plan = plan_table(name, target, file.tables[name], entry.notes if entry else None, cache, masked) + plan.version = entry.version if entry else 0 + plans.append(plan) + return ImportReport(plans, unmatched, _guidance(file, row, include_guidance)) + + +async def run_import(session: AsyncSession, row: Any, source: Any, content: Any, *, include_guidance: bool, + actor: str | None) -> ImportReport: + """导入：按 plan_import 的规矩逐表写入（乐观锁，撞上别人刚改过就重读重算，最多三次），再按需写起草要求。 + 一张表写不进去只记在它的 error 上（新增、更新记 0），不影响别的表。""" + file = read_file(content, include_guidance=include_guidance) + cache = getattr(source, "schema_cache", None) or {} + matched, unmatched = match_tables(list(file.tables), cache) + masked = {c.lower() for c in masked_columns(getattr(row, "options", None))} + plans: list[TablePlan] = [] + for name, target in matched.items(): + plans.append(await _import_table(session, source, name, target, file.tables[name], cache, masked, actor)) + guidance = _guidance(file, row, include_guidance) + if guidance["action"] == "replace": + await catalog_jobs.save_guidance(session, row, file.guidance) + return ImportReport(plans, unmatched, guidance) + + +async def _import_table(session: AsyncSession, source: Any, name: str, target: str, notes: Mapping[str, Any], + cache: Mapping[str, Any], masked: set[str], actor: str | None) -> TablePlan: + plan = TablePlan(table_name=name, matched_as=target) + try: + for _attempt in range(3): + entry = await catalog.read_entry(session, source.id, target) + version = entry.version if entry else 0 + plan = plan_table(name, target, notes, entry.notes if entry else None, cache, masked) + plan.version = version + if not plan.writes: + return plan + slots = catalog._slots(entry.notes if entry else None) + slots.update(plan.writes) + try: + written = await catalog.write_entry(session, source.id, target, catalog._assemble(slots), + if_version=version, actor=actor) + except catalog.CatalogConflict: + continue + plan.version = written.version + return plan + plan.error = WRITE_CONFLICT_ERROR + except catalog.CatalogInvalid as e: + await session.rollback() + plan.error = WRITE_ERROR.format(line=str(e)) + except Exception as e: # noqa: BLE001 - 这一张写不进去不影响别的表 + from app.core.errors import describe_exception, first_line + + await session.rollback() + plan.error = WRITE_ERROR.format(line=first_line(e) or describe_exception(e)) + plan.add = plan.update = 0 + return plan + + +__all__ = ["FORMAT", "ImportFile", "ImportReport", "TablePlan", "TransferInvalid", "VERSION", "attachment_name", + "content_disposition", "export_catalog", "match_tables", "plan_import", "plan_table", "read_file", + "run_import"] diff --git a/backend/tests/test_catalog_transfer.py b/backend/tests/test_catalog_transfer.py new file mode 100644 index 00000000..de5164b5 --- /dev/null +++ b/backend/tests/test_catalog_transfer.py @@ -0,0 +1,378 @@ +"""数据目录的导出导入（data/catalog_transfer.py）：本机整理好的目录带到部署实例上。 + +- GET /catalog/export：附件「<数据源名>-数据目录.json」，{format: agentlab-catalog, version: 1, exported_at, + source: {name, kind}, guidance, tables: {表名: {notes, version}}}，含全部状态的项（表结构里已经没有的表也带） +- POST /catalog/import/preview：只读，报告每张对上的表新增、更新、保留、冲突、跳过了什么 +- POST /catalog/import：同样的报告，加每张表写入后的版本。规则： + - 表按名字对（忽略大小写；带 schema 前缀的对不上时去掉前缀再对），对不上的列在 unmatched_tables + - 文件里已确认、已驳回的项照带（来源、状态不变）；目标里已确认、已驳回的项和文件不一样时保留目标的、记冲突 + - 文件里推断、有确证的项只填目标的空缺，不覆盖目标已有的 + - 表结构里没有的列、关系两端对不上的跳过并写原因；目标数据源遮罩的列，码值不导入 + - 逐表乐观锁，一张表写不进去不影响别的；有排队或进行中的起草任务时 409 catalog_draft_busy；文件格式不对 422 + - 起草要求：勾了导入、目标还没有时才写；目标已有的不覆盖 + +表名、列名、业务数字都是虚构的。 +""" +from __future__ import annotations + +import copy +import json +import uuid +from urllib.parse import quote, unquote + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.data import catalog, catalog_jobs, catalog_transfer +from app.data.catalog import make_item +from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION +from app.db.base import SessionLocal +from app.db.models import CatalogDraftJob, DataSource +from app.main import app as _fastapi_app +from tests.fixtures.sources import drop_source + +ACTOR = {"X-Actor": quote("沈岚")} + + +def _col(name: str, type_: str = "INTEGER") -> dict: + return {"name": name, "type": type_, "nullable": True, "comment": None} + + +def _tables(*, upper: bool = False) -> dict: + """两边的表结构：目标那边（upper）表名、列名写法不同，还少了一列 tier。""" + def n(name: str) -> str: + return name.upper() if upper else name + + members_cols = [_col(n("id")), _col(n("nick"), "TEXT")] + ([] if upper else [_col("tier")]) + return { + n("orders"): {"qualified": n("orders"), "primary_key": [n("id")], "comment": None, "is_view": False, + "foreign_keys": [], "unique": [], + "columns": [_col(n("id")), _col(n("status")), _col(n("amount"), "REAL"), _col(n("member_id")), + _col(n("contact_phone"), "TEXT")]}, + n("members"): {"qualified": n("members"), "primary_key": [n("id")], "comment": None, "is_view": False, + "foreign_keys": [], "unique": [], "columns": members_cols}, + n("refunds"): {"qualified": n("refunds"), "primary_key": [n("id")], "comment": None, "is_view": False, + "foreign_keys": [], "unique": [], "columns": [_col(n("id")), _col(n("order_id"))]}, + } + + +def _rel(to_table: str, columns: list[str], to_columns: list[str], *, status: str, source: str = "name", + table: str = "orders") -> dict: + return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, + "to_table": to_table, "to_columns": to_columns, "cardinality": "many_to_one", "coverage": None, + "source": source, "status": status} + + +#: 本机（源）的目录 +SOURCE_ORDERS = { + "label": make_item("订单", "human", "confirmed", at="2026-10-01T08:00:00+00:00"), + "grain": make_item("每笔订单一行", "llm"), + "valid_filter": make_item("status <> 9", "human", "confirmed"), + "dedup": make_item("同一订单号取最后一次", "llm", "rejected"), + "kind": make_item("fact", "llm", "confirmed"), + "columns": { + "status": {"codes": make_item({"1": "已支付", "9": "作废"}, "human", "confirmed", complete=True), + "label": make_item("状态", "llm")}, + "amount": {"unit": make_item("元", "human", "confirmed"), "measure": make_item("flow", "llm")}, + "contact_phone": {"codes": make_item({"13900000001": "测试号"}, "human", "confirmed"), + "label": make_item("联系电话", "human", "confirmed")}, + # 目标表结构里没有这一列 + "gift_flag": {"label": make_item("赠品标记", "human", "confirmed")}, + }, + "relations": [ + _rel("members", ["member_id"], ["id"], status="confirmed"), + # 目标那边没有这张表 + _rel("stores", ["member_id"], ["id"], status="rejected"), + ], +} +SOURCE_MEMBERS = {"label": make_item("会员", "llm"), "keys": make_item(["tier"], "human", "confirmed"), + "columns": {"nick": {"label": make_item("昵称", "llm", "verified")}}} +#: 目标那边没有的表 +SOURCE_GHOST = {"label": make_item("旧表", "human", "confirmed")} + +#: 目标（部署实例）上已有的目录 +TARGET_ORDERS = { + "label": make_item("订单表", "human", "confirmed"), # 和文件不一样、都确认过：冲突，保留目标 + "grain": make_item("一行一单", "llm"), # 文件里是推断：保留目标 + "valid_filter": make_item("STATUS <> 9", "llm"), # 目标是推断、文件确认过：更新 + "kind": make_item("fact", "human", "confirmed"), # 一样：保留 + "columns": {"STATUS": {"label": make_item("状态码", "comment")}}, +} + +_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 _source(name: str, tables: dict, entries: dict[str, dict], **kw) -> str: + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=":memory:", schema_cache={"tables": tables}, **kw) + session.add(row) + await session.commit() + sid = row.id + _CREATED.append(sid) + for table, notes in entries.items(): + await _write(sid, table, notes) + return sid + + +async def _pair() -> tuple[str, str]: + local = await _source(f"本机库_{uuid.uuid4().hex[:6]}", _tables(), + {"orders": SOURCE_ORDERS, "members": SOURCE_MEMBERS, "ghost_table": SOURCE_GHOST}, + options={CATALOG_DRAFT_GUIDANCE_OPTION: "金额一律按元"}) + remote = await _source(f"deploy_{uuid.uuid4().hex[:6]}", _tables(upper=True), {"ORDERS": TARGET_ORDERS}, + options={"mask_columns": "contact_phone"}) + return local, remote + + +def _base(sid: str) -> str: + return f"/api/datasources/{sid}/catalog" + + +async def _export(client, sid: str) -> dict: + r = await client.get(f"{_base(sid)}/export") + assert r.status_code == 200, r.text + return r.json() + + +# -------------------------------------------------------------------------- +# 导出 +# -------------------------------------------------------------------------- + + +async def test_export_is_a_json_attachment_with_every_item(client): + local, _ = await _pair() + r = await client.get(f"{_base(local)}/export") + assert r.status_code == 200, r.text + assert r.headers["content-type"].startswith("application/json") + disposition = r.headers["content-disposition"] + assert disposition.startswith("attachment;") and "filename=\"agentlab-catalog-" in disposition + async with SessionLocal() as session: + name = (await session.get(DataSource, local)).name + assert unquote(disposition.split("filename*=UTF-8''", 1)[1]) == f"{name}-数据目录.json" + body = r.json() + assert body["format"] == "agentlab-catalog" and body["version"] == 1 + assert body["exported_at"] and body["source"] == {"name": name, "kind": "sqlite"} + assert body["guidance"] == "金额一律按元" + assert set(body["tables"]) == {"orders", "members", "ghost_table"} + assert body["tables"]["orders"]["version"] == 1 + # 含全部状态的项：驳回的也带 + assert body["tables"]["orders"]["notes"]["dedup"]["status"] == "rejected" + assert body["tables"]["orders"]["notes"] == SOURCE_ORDERS + assert body["tables"]["ghost_table"] == {"notes": SOURCE_GHOST, "version": 1} + + +def test_attachment_name_drops_characters_files_cannot_have(): + assert catalog_transfer.attachment_name('景区/客流:库?') == "景区_客流_库_-数据目录.json" + assert catalog_transfer.attachment_name(" ") == "数据源-数据目录.json" + + +# -------------------------------------------------------------------------- +# 预览与导入 +# -------------------------------------------------------------------------- + + +def _by(report: dict) -> dict[str, dict]: + return {t["matched_as"]: t for t in report["tables"]} + + +async def test_preview_reports_and_writes_nothing(client): + local, remote = await _pair() + content = await _export(client, local) + r = await client.post(f"{_base(remote)}/import/preview", json={"content": content, "include_guidance": True}) + assert r.status_code == 200, r.text + report = r.json() + assert report["unmatched_tables"] == ["ghost_table"] + tables = _by(report) + assert set(tables) == {"ORDERS", "MEMBERS"} + orders = tables["ORDERS"] + assert orders["table_name"] == "orders" + conflicts = {c["path"]: c for c in orders["conflicts"]} + assert set(conflicts) == {"label"} + label = conflicts["label"] + assert label["label"] == "表的中文名" and label["file_value"] == "订单" and label["target_value"] == "订单表" + assert "保留" in label["why"] + skipped = {s["path"]: s["why"] for s in orders["skipped"]} + assert "columns.gift_flag.label" in skipped and "表结构里没有列 gift_flag" in skipped["columns.gift_flag.label"] + assert "遮罩" in skipped["columns.contact_phone.codes"] + stores = next(p for p in skipped if p.startswith("relations.")) + assert "stores" in skipped[stores] + # 新增：dedup（驳回的照带）、status 码值、amount 两项、联系电话的中文名、关联关系；更新：有效记录条件 + # 保留：粒度（文件里是推断，目标已有）、表类型（一样）、STATUS 的中文名（文件里是推断，目标已有） + assert (orders["add"], orders["update"], orders["keep"]) == (6, 1, 3) + members = tables["MEMBERS"] + # 业务主键指着的列 tier 目标那边没有：跳过；会员的中文名和昵称填空缺 + assert (members["add"], members["update"], members["keep"]) == (2, 0, 0) + assert [s["path"] for s in members["skipped"]] == ["keys"] + assert report["totals"] == {"tables": 2, "add": 8, "update": 1, "keep": 3, "conflicts": 1, + "skipped": len(orders["skipped"]) + 1} + assert report["guidance"] == {"file": "金额一律按元", "target": None, "action": "replace"} + async with SessionLocal() as session: + assert (await catalog.read_entry(session, remote, "ORDERS")).version == 1 + assert await catalog.read_entry(session, remote, "MEMBERS") is None + assert catalog_jobs.guidance_of(await session.get(DataSource, remote)) is None + + +async def test_import_writes_by_the_rules_and_reports_versions(client): + local, remote = await _pair() + content = await _export(client, local) + r = await client.post(f"{_base(remote)}/import", json={"content": content, "include_guidance": True}, + headers=ACTOR) + assert r.status_code == 200, r.text + report = r.json() + tables = _by(report) + assert tables["ORDERS"]["version"] == 2 and tables["MEMBERS"]["version"] == 1 + assert tables["ORDERS"]["error"] is None + assert report["guidance"]["action"] == "replace" + async with SessionLocal() as session: + orders = await catalog.read_entry(session, remote, "ORDERS") + members = await catalog.read_entry(session, remote, "MEMBERS") + row = await session.get(DataSource, remote) + notes = orders.notes + assert orders.updated_by == "沈岚" + assert notes["label"]["value"] == "订单表" # 冲突：保留目标的 + assert notes["grain"]["value"] == "一行一单" # 推断只填空缺 + assert notes["valid_filter"] == {**SOURCE_ORDERS["valid_filter"]} # 照带：来源、状态不变 + assert notes["dedup"]["status"] == "rejected" and notes["dedup"]["source"] == "llm" + codes = notes["columns"]["STATUS"]["codes"] + assert codes["value"] == {"1": "已支付", "9": "作废"} and codes["complete"] is True + assert notes["columns"]["STATUS"]["label"]["value"] == "状态码" + assert notes["columns"]["AMOUNT"]["measure"] == SOURCE_ORDERS["columns"]["amount"]["measure"] + # 遮罩列：中文名照带，码值不导入 + assert notes["columns"]["CONTACT_PHONE"] == {"label": SOURCE_ORDERS["columns"]["contact_phone"]["label"]} + assert "GIFT_FLAG" not in notes["columns"] and "gift_flag" not in notes["columns"] + [rel] = notes["relations"] + assert rel["id"] == catalog.relation_id("ORDERS", ["MEMBER_ID"], "MEMBERS", ["ID"]) + assert (rel["columns"], rel["to_table"], rel["to_columns"]) == (["MEMBER_ID"], "MEMBERS", ["ID"]) + assert rel["status"] == "confirmed" and rel["source"] == "name" + assert catalog.validate_notes(notes) == [] + assert members.notes["label"]["status"] == "proposed" and "keys" not in members.notes + assert members.notes["columns"]["NICK"]["label"]["status"] == "verified" + assert catalog_jobs.guidance_of(row) == "金额一律按元" + # 再导一次：都对上了，什么都不写；目标已经有起草要求，不覆盖 + again = (await client.post(f"{_base(remote)}/import", json={"content": content, "include_guidance": True})).json() + assert again["totals"]["add"] == again["totals"]["update"] == 0 + assert _by(again)["ORDERS"]["version"] == 2 + assert again["guidance"] == {"file": "金额一律按元", "target": "金额一律按元", "action": "keep"} + + +async def test_target_rejection_beats_file_confirmation_and_guidance_is_optional(client): + local, remote = await _pair() + await _write(remote, "MEMBERS", {"label": make_item("会员", "llm", "rejected")}) + async with SessionLocal() as session: + row = await session.get(DataSource, remote) + await catalog_jobs.save_guidance(session, row, "部署实例自己的要求") + content = await _export(client, local) + content["tables"]["members"]["notes"]["label"] = make_item("会员", "human", "confirmed") + report = (await client.post(f"{_base(remote)}/import", + json={"content": content, "include_guidance": False})).json() + members = _by(report)["MEMBERS"] + [conflict] = members["conflicts"] + assert conflict["path"] == "label" and conflict["target_status"] == "rejected" + assert conflict["file_status"] == "confirmed" and "驳回" in conflict["why"] + assert report["guidance"] == {"file": "金额一律按元", "target": "部署实例自己的要求", "action": "none"} + async with SessionLocal() as session: + entry = await catalog.read_entry(session, remote, "MEMBERS") + assert entry.notes["label"]["status"] == "rejected" + assert catalog_jobs.guidance_of(await session.get(DataSource, remote)) == "部署实例自己的要求" + + +async def test_one_table_failing_does_not_stop_the_others(client, monkeypatch): + local, remote = await _pair() + content = await _export(client, local) + real = catalog.write_entry + + async def flaky(session, source_id, table, notes, **kw): + if table == "ORDERS": + raise catalog.CatalogConflict(table, 99) + return await real(session, source_id, table, notes, **kw) + + monkeypatch.setattr(catalog, "write_entry", flaky) + report = (await client.post(f"{_base(remote)}/import", json={"content": content})).json() + tables = _by(report) + assert tables["ORDERS"]["error"] and "稍后" in tables["ORDERS"]["error"] + assert tables["ORDERS"]["add"] == tables["ORDERS"]["update"] == 0 + assert tables["MEMBERS"]["error"] is None and tables["MEMBERS"]["version"] == 1 + + +async def test_import_waits_for_draft_jobs_and_rejects_bad_files(client): + local, remote = await _pair() + content = await _export(client, local) + async with SessionLocal() as session: + session.add(CatalogDraftJob(id=uuid.uuid4().hex, source_id=remote, status="running")) + await session.commit() + busy = await client.post(f"{_base(remote)}/import", json={"content": content}) + assert busy.status_code == 409 and busy.json()["code"] == "catalog_draft_busy" + # 预览是只读的，照常给 + assert (await client.post(f"{_base(remote)}/import/preview", json={"content": content})).status_code == 200 + bad = [ + "不是对象", + {**content, "format": "something-else"}, + {**content, "version": 2}, + {**content, "tables": []}, + {**content, "tables": {"orders": {"notes": "x"}}}, + {**content, "guidance": 3}, + ] + for body in bad: + r = await client.post(f"{_base(local)}/import/preview", json={"content": body}) + assert r.status_code == 422, body + assert any("一" <= ch <= "鿿" for ch in r.json()["detail"]) + too_long = {**content, "guidance": "长" * (catalog_jobs.GUIDANCE_MAX + 1)} + r = await client.post(f"{_base(local)}/import/preview", json={"content": too_long, "include_guidance": True}) + assert r.status_code == 422 and "起草要求" in r.json()["detail"] + + +async def test_invalid_items_are_skipped_not_fatal(client): + local, remote = await _pair() + content = await _export(client, local) + notes = content["tables"]["orders"]["notes"] + notes["kind"] = {"value": "event", "source": "human", "status": "confirmed"} # 取值不在可选值内 + notes["formula"] = {"value": "sum(amount)", "source": "human", "status": "confirmed"} # 不认识的字段 + notes["columns"]["amount"]["colour"] = {"value": "红", "source": "human", "status": "confirmed"} + report = (await client.post(f"{_base(remote)}/import/preview", json={"content": content})).json() + skipped = {s["path"]: s["why"] for s in _by(report)["ORDERS"]["skipped"]} + assert "明细表" in skipped["kind"] + assert "formula" in skipped["formula"] and "columns.amount.colour" in skipped + + +async def test_import_needs_a_schema(client): + local, _ = await _pair() + content = await _export(client, local) + empty = await _source(f"empty_{uuid.uuid4().hex[:6]}", {}, {}) + r = await client.post(f"{_base(empty)}/import/preview", json={"content": content}) + assert r.status_code == 409 and r.json()["code"] == "schema_missing" + + +async def test_round_trip_into_an_empty_copy_reproduces_the_catalog(client): + local, _ = await _pair() + content = await _export(client, local) + copy_sid = await _source(f"copy_{uuid.uuid4().hex[:6]}", _tables(), {}) + report = (await client.post(f"{_base(copy_sid)}/import", json={"content": json.loads(json.dumps(content)), + "include_guidance": True})).json() + assert report["unmatched_tables"] == ["ghost_table"] + assert all(not t["conflicts"] for t in report["tables"]) + async with SessionLocal() as session: + mine = await catalog.read_entry(session, local, "orders") + theirs = await catalog.read_entry(session, copy_sid, "orders") + # 关系两端都对得上的照带；指向目标那边没有的表的跳过 + expected = copy.deepcopy(mine.notes) + expected["relations"] = [r for r in mine.notes["relations"] if r["to_table"] != "stores"] + del expected["columns"]["gift_flag"] # 表结构里没有这一列 + assert theirs.notes == expected