数据目录导出导入:导出为「<数据源名>-数据目录.json」附件(全部状态的项、起草要求);导入先预览再写入,表名、列名按表结构的写法再忽略大小写对(带 schema 前缀的去掉前缀再对),关联关系两端换成目标的写法并重算编号;已确认、已驳回的项照带(来源、状态不变),目标里已确认、已驳回而不一样的保留目标的并记冲突,推断、有确证的只填空缺;表结构里没有的列、两端对不上的关系、目标遮罩列的码值、不合规的项跳过并写原因;逐表乐观锁、一张表写不进去不影响别的;有起草任务时 409,文件格式不对 422,没有表结构 409;起草要求只在勾选且目标还没有时写入

This commit is contained in:
ThearchyHelios
2026-10-14 08:06:45 +08:00
parent adc56da6cf
commit 2298cf848a
3 changed files with 927 additions and 1 deletions
+65 -1
View File
@@ -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]:
+484
View File
@@ -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"]
+378
View File
@@ -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