routes.py 60 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717
  1. import io
  2. import logging
  3. from flask import current_app, jsonify, request, send_file
  4. from minio import Minio
  5. from minio.error import S3Error
  6. from sqlalchemy import or_
  7. from app import db
  8. from app.api.meta_data import bp
  9. from app.core.common.identifiers import ensure_governance_uid
  10. from app.core.meta_data import (
  11. check_redundancy_for_add,
  12. check_redundancy_for_update,
  13. convert_tag_ids_to_tags,
  14. get_file_content,
  15. get_formatted_time,
  16. handle_id_unstructured,
  17. handle_txt_graph,
  18. meta_impact_graph,
  19. meta_kinship_graph,
  20. meta_list,
  21. normalize_tag_inputs,
  22. parse_text,
  23. solve_unstructured_data,
  24. text_resource_solve,
  25. )
  26. from app.core.system.auth import require_auth
  27. from app.models.metadata_review import (
  28. MetadataReviewRecord,
  29. MetadataVersionHistory,
  30. update_review_record_resolution,
  31. )
  32. from app.models.result import failed, success
  33. from app.services.neo4j_driver import neo4j_driver
  34. logger = logging.getLogger("app")
  35. def get_minio_client():
  36. """获取 MinIO 客户端实例"""
  37. return Minio(
  38. current_app.config["MINIO_HOST"],
  39. access_key=current_app.config["MINIO_USER"],
  40. secret_key=current_app.config["MINIO_PASSWORD"],
  41. secure=current_app.config["MINIO_SECURE"],
  42. )
  43. def get_minio_config():
  44. """获取 MinIO 配置"""
  45. return {
  46. "MINIO_BUCKET": current_app.config["MINIO_BUCKET"],
  47. "PREFIX": current_app.config["PREFIX"],
  48. "ALLOWED_EXTENSIONS": current_app.config["ALLOWED_EXTENSIONS"],
  49. }
  50. def allowed_file(filename):
  51. """检查文件扩展名是否允许"""
  52. if "." not in filename:
  53. return False
  54. ext = filename.rsplit(".", 1)[1].lower()
  55. return ext in get_minio_config()["ALLOWED_EXTENSIONS"]
  56. # 元数据列表
  57. @bp.route("/node/list", methods=["POST"])
  58. def meta_node_list():
  59. try:
  60. payload = request.get_json() or {}
  61. if not isinstance(payload, dict):
  62. return jsonify(failed("请求数据格式错误,应为 JSON 对象"))
  63. def to_int(value, default):
  64. try:
  65. return int(value)
  66. except (TypeError, ValueError):
  67. return default
  68. # 分页参数(size 未传则返回全部记录)
  69. raw_page_size = payload.get("size", None)
  70. if raw_page_size is None:
  71. page_size = None
  72. page = 1
  73. else:
  74. page_size = to_int(raw_page_size, 10)
  75. page = to_int(payload.get("current", 1), 1)
  76. # 过滤参数
  77. name_en_filter = payload.get("name_en") or None
  78. name_zh_filter = payload.get("name_zh") or None
  79. category_filter = payload.get("category") or None
  80. time_filter = payload.get("time") or None
  81. logger.info(
  82. f"[node/list] 过滤参数: name_zh={name_zh_filter}, "
  83. f"name_en={name_en_filter}, category={category_filter}"
  84. )
  85. tag_filter = payload.get("tag")
  86. if tag_filter is not None and not isinstance(tag_filter, list):
  87. tag_filter = None
  88. # 调用核心业务逻辑
  89. result, total_count = meta_list(
  90. page,
  91. page_size,
  92. "",
  93. name_en_filter,
  94. name_zh_filter,
  95. category_filter,
  96. time_filter,
  97. tag_filter,
  98. )
  99. # 返回结果
  100. response_size = total_count if page_size is None else page_size
  101. response_page = 1 if page_size is None else page
  102. return jsonify(
  103. success(
  104. {
  105. "records": result,
  106. "total": total_count,
  107. "size": response_size,
  108. "current": response_page,
  109. }
  110. )
  111. )
  112. except Exception as e:
  113. logger.error(f"获取元数据列表失败: {str(e)}")
  114. return jsonify(failed(str(e)))
  115. # 元数据图谱
  116. @bp.route("/node/graph", methods=["POST"])
  117. def meta_node_graph():
  118. try:
  119. if not request.json:
  120. return jsonify(failed("请求数据不能为空"))
  121. # 从请求中获取节点ID
  122. node_id = request.json.get("nodeId")
  123. if node_id is None:
  124. return jsonify(failed("nodeId 不能为空"))
  125. try:
  126. node_id_int = int(node_id)
  127. except (TypeError, ValueError):
  128. return jsonify(failed("nodeId 必须为整数"))
  129. # 调用核心业务逻辑
  130. graph = meta_kinship_graph(node_id_int)
  131. is_dict = isinstance(graph, dict)
  132. nodes = graph.get("nodes", []) if is_dict else []
  133. relationships = graph.get("relationships", []) if is_dict else []
  134. # 当前节点属性
  135. node_info = next(
  136. (n for n in nodes if n.get("id") == node_id_int),
  137. {},
  138. )
  139. # 关联节点(包含属性,便于前端展示名称等)
  140. related_nodes = [n for n in nodes if n.get("id") != node_id_int]
  141. payload = {
  142. "node": node_info,
  143. "related_nodes": related_nodes,
  144. "relationships": relationships,
  145. }
  146. return jsonify(success(payload))
  147. except Exception as e:
  148. logger.error(f"获取元数据图谱失败: {str(e)}")
  149. return jsonify(failed(str(e)))
  150. # 删除元数据
  151. @bp.route("/node/delete", methods=["POST"])
  152. def meta_node_delete():
  153. try:
  154. if not request.json:
  155. return jsonify(failed("请求数据不能为空"))
  156. # 从请求中获取节点ID
  157. node_id = request.json.get("id")
  158. # 删除节点逻辑
  159. with neo4j_driver.get_session() as session:
  160. cypher = "MATCH (n) WHERE id(n) = $node_id DETACH DELETE n"
  161. session.run(cypher, node_id=int(node_id))
  162. # 返回结果
  163. return jsonify(success({}))
  164. except Exception as e:
  165. logger.error(f"删除元数据失败: {str(e)}")
  166. return jsonify(failed(str(e)))
  167. # 编辑元数据
  168. @bp.route("/node/edit", methods=["POST"])
  169. def meta_node_edit():
  170. try:
  171. if not request.json:
  172. return jsonify(failed("请求数据不能为空"))
  173. # 从请求中获取节点ID
  174. node_id = request.json.get("id")
  175. if not node_id:
  176. return jsonify(failed("节点ID不能为空"))
  177. # 获取节点
  178. with neo4j_driver.get_session() as session:
  179. # 查询节点信息
  180. cypher = """
  181. MATCH (n:DataMeta)
  182. WHERE id(n) = $node_id
  183. RETURN n
  184. """
  185. result = session.run(cypher, node_id=int(node_id))
  186. node = result.single()
  187. if not node or not node["n"]:
  188. return jsonify(failed("节点不存在"))
  189. # 获取节点数据
  190. node_data = dict(node["n"])
  191. node_data["id"] = node["n"].id
  192. # 获取标签信息
  193. tag_cypher = """
  194. MATCH (n:DataMeta)-[:LABEL]->(t:DataLabel)
  195. WHERE id(n) = $node_id
  196. RETURN t
  197. """
  198. tag_result = session.run(tag_cypher, node_id=int(node_id))
  199. tags: list[dict] = []
  200. for record in tag_result:
  201. tag_node = record.get("t")
  202. if tag_node:
  203. tags.append(
  204. {
  205. "id": tag_node.id,
  206. "name_zh": tag_node.get("name_zh", ""),
  207. "name_en": tag_node.get("name_en", ""),
  208. }
  209. )
  210. # 获取主数据信息
  211. master_data_cypher = """
  212. MATCH (n:DataMeta)-[:master_data]->(m:master_data)
  213. WHERE id(n) = $node_id
  214. RETURN m
  215. """
  216. master_data_result = session.run(master_data_cypher, node_id=int(node_id))
  217. master_data = master_data_result.single()
  218. # 构建返回数据
  219. response_data = [
  220. {
  221. "master_data": (
  222. master_data["m"].id
  223. if master_data and master_data["m"]
  224. else None
  225. ),
  226. "name_zh": node_data.get("name_zh", ""),
  227. "name_en": node_data.get("name_en", ""),
  228. "create_time": node_data.get("create_time", ""),
  229. "update_time": node_data.get("update_time", ""),
  230. "status": bool(node_data.get("status", True)),
  231. "data_type": node_data.get("data_type", ""),
  232. "tag": tags,
  233. "affiliation": node_data.get("affiliation"),
  234. "category": node_data.get("category"),
  235. "alias": node_data.get("alias"),
  236. "describe": node_data.get("describe"),
  237. }
  238. ]
  239. logger.info(f"成功获取元数据节点: ID={node_data['id']}")
  240. return jsonify(success(response_data))
  241. except Exception as e:
  242. logger.error(f"获取元数据节点失败: {str(e)}")
  243. return jsonify(failed(str(e)))
  244. # 增加元数据
  245. @bp.route("/check", methods=["GET"])
  246. def meta_check():
  247. """
  248. 检查元数据中文名是否已存在
  249. 请求参数:
  250. - name_zh: 元数据中文名(URL参数)
  251. 返回:
  252. - exists: true/false 表示是否存在
  253. """
  254. try:
  255. name_zh = request.args.get("name_zh")
  256. if not name_zh:
  257. return jsonify(failed("缺少name_zh参数"))
  258. # 查询数据库检查是否存在
  259. with neo4j_driver.get_session() as session:
  260. cypher = """
  261. MATCH (n:DataMeta {name_zh: $name_zh})
  262. RETURN count(n) > 0 as exists
  263. """
  264. result = session.run(cypher, name_zh=name_zh)
  265. record = result.single()
  266. if record:
  267. exists = record["exists"]
  268. logger.info(f"检查元数据 '{name_zh}': {'存在' if exists else '不存在'}")
  269. return jsonify(
  270. success({"exists": exists, "name_zh": name_zh}, "查询成功")
  271. )
  272. else:
  273. return jsonify(
  274. success({"exists": False, "name_zh": name_zh}, "查询成功")
  275. )
  276. except Exception as e:
  277. logger.error(f"检查元数据失败: {str(e)}")
  278. return jsonify(failed(f"检查失败: {str(e)}"))
  279. @bp.route("/node/add", methods=["POST"])
  280. def meta_node_add():
  281. """
  282. 新增元数据节点
  283. 在创建前会进行冗余检测:
  284. - 如果存在完全匹配的元数据,返回已存在的节点信息
  285. - 如果存在疑似重复的元数据,创建审核记录并返回提示
  286. - 如果无重复,正常创建新节点
  287. """
  288. try:
  289. if not request.json:
  290. return jsonify(failed("请求数据不能为空"))
  291. # 从请求中获取节点信息
  292. node_name_zh = request.json.get("name_zh")
  293. node_type = request.json.get("data_type")
  294. node_category = request.json.get("category")
  295. node_alias = request.json.get("alias")
  296. node_affiliation = request.json.get("affiliation")
  297. node_tag = request.json.get("tag")
  298. node_desc = request.json.get("describe")
  299. node_status = bool(request.json.get("status", True))
  300. node_name_en = request.json.get("name_en")
  301. # 是否强制创建(跳过冗余检测)
  302. force_create = bool(request.json.get("force_create", False))
  303. if not node_name_zh:
  304. return jsonify(failed("节点名称不能为空"))
  305. if not node_type:
  306. return jsonify(failed("节点类型不能为空"))
  307. # 统一处理标签ID
  308. tag_ids = normalize_tag_inputs(node_tag)
  309. # ========== 冗余检测 ==========
  310. has_suspicious_duplicates = False
  311. suspicious_candidates = []
  312. if not force_create:
  313. redundancy_result = check_redundancy_for_add(
  314. name_zh=node_name_zh,
  315. name_en=node_name_en or "",
  316. data_type=node_type,
  317. tag_ids=tag_ids,
  318. )
  319. # 存在完全匹配的元数据,直接返回,不做任何操作
  320. if redundancy_result["has_exact_match"]:
  321. exact_id = redundancy_result["exact_match_id"]
  322. logger.info(
  323. f"元数据已存在(完全匹配): name_zh={node_name_zh}, "
  324. f"existing_id={exact_id}"
  325. )
  326. return jsonify(
  327. failed(
  328. f"元数据已存在(完全匹配),无需重复创建。"
  329. f"已存在的元数据ID: {exact_id}"
  330. )
  331. )
  332. # 存在疑似重复的元数据,标记状态,稍后创建节点后再写入审核记录
  333. if redundancy_result["has_candidates"]:
  334. has_suspicious_duplicates = True
  335. suspicious_candidates = redundancy_result["candidates"]
  336. logger.info(
  337. f"发现疑似重复元数据: name_zh={node_name_zh}, "
  338. f"候选数量={len(suspicious_candidates)}"
  339. )
  340. # ========== 创建节点 ==========
  341. with neo4j_driver.get_session() as session:
  342. cypher = """
  343. MERGE (n:DataMeta {name_zh: $name_zh})
  344. ON CREATE SET n.uid = $uid,
  345. n.name_en = $name_en,
  346. n.data_type = $data_type,
  347. n.category = $category,
  348. n.alias = $alias,
  349. n.affiliation = $affiliation,
  350. n.describe = $describe,
  351. n.create_time = $create_time,
  352. n.updateTime = $update_time,
  353. n.status = $status,
  354. n.name_en = $name_en
  355. ON MATCH SET n.uid = coalesce(n.uid, $uid),
  356. n.data_type = $data_type,
  357. n.category = $category,
  358. n.alias = $alias,
  359. n.affiliation = $affiliation,
  360. n.describe = $describe,
  361. n.updateTime = $update_time,
  362. n.status = $status,
  363. n.name_en = $name_en
  364. RETURN n
  365. """
  366. create_time = update_time = get_formatted_time()
  367. node_properties = {
  368. "name_zh": node_name_zh,
  369. "data_type": node_type,
  370. "category": node_category,
  371. "alias": node_alias,
  372. "affiliation": node_affiliation,
  373. "describe": node_desc,
  374. "create_time": create_time,
  375. "update_time": update_time,
  376. "status": node_status,
  377. "name_en": node_name_en,
  378. }
  379. ensure_governance_uid(node_properties)
  380. result = session.run(
  381. cypher,
  382. **node_properties,
  383. )
  384. node = result.single()
  385. if node and node["n"]:
  386. node_data = dict(node["n"])
  387. node_data["id"] = node["n"].id
  388. # 如果提供了标签列表,创建标签关系
  389. tag_nodes = []
  390. if tag_ids:
  391. for tag_id in tag_ids:
  392. # 获取标签节点信息
  393. tag_fetch = session.run(
  394. "MATCH (t:DataLabel) WHERE id(t) = $tag_id RETURN t",
  395. tag_id=tag_id,
  396. ).single()
  397. if not tag_fetch or not tag_fetch.get("t"):
  398. logger.warning(f"未找到标签节点: {tag_id}")
  399. continue
  400. tag_node = tag_fetch["t"]
  401. tag_nodes.append(
  402. {
  403. "id": tag_node.id,
  404. "name_zh": tag_node.get("name_zh", ""),
  405. "name_en": tag_node.get("name_en", ""),
  406. }
  407. )
  408. tag_cypher = """
  409. MATCH (n:DataMeta), (t:DataLabel)
  410. WHERE id(n) = $node_id AND id(t) = $tag_id
  411. MERGE (n)-[r:LABEL]->(t)
  412. RETURN r
  413. """
  414. session.run(tag_cypher, node_id=node["n"].id, tag_id=tag_id)
  415. node_data["tag"] = tag_nodes
  416. logger.info(
  417. f"成功创建或更新元数据节点: "
  418. f"ID={node_data['id']}, name={node_name_zh}"
  419. )
  420. # ========== 处理疑似重复情况 ==========
  421. # 如果存在疑似重复,创建审核记录
  422. if has_suspicious_duplicates and suspicious_candidates:
  423. from app.core.meta_data.redundancy_check import (
  424. write_redundancy_review_record_with_new_id,
  425. )
  426. # 构建新元数据快照(包含新创建的节点ID)
  427. new_meta_snapshot = {
  428. "id": node_data["id"],
  429. "name_zh": node_name_zh,
  430. "name_en": node_name_en or "",
  431. "data_type": node_type,
  432. "tag_ids": tag_ids,
  433. }
  434. # 写入审核记录
  435. write_redundancy_review_record_with_new_id(
  436. new_meta=new_meta_snapshot,
  437. candidates=suspicious_candidates,
  438. source="api",
  439. )
  440. # 返回成功创建,但提示疑似重复
  441. candidate_names = [
  442. c.get("name_zh", "") for c in suspicious_candidates[:3]
  443. ]
  444. return jsonify(
  445. success(
  446. node_data,
  447. message=(
  448. f"元数据创建成功,但发现疑似重复元数据。"
  449. f"疑似重复: {', '.join(candidate_names)}。"
  450. f"已创建审核记录,请前往元数据审核页面进行处理。"
  451. ),
  452. )
  453. )
  454. return jsonify(success(node_data))
  455. else:
  456. logger.error(f"创建元数据节点失败: {node_name_zh}")
  457. return jsonify(failed("创建元数据节点失败"))
  458. except Exception as e:
  459. logger.error(f"添加元数据失败: {str(e)}")
  460. return jsonify(failed(str(e)))
  461. # 搜索元数据
  462. @bp.route("/search", methods=["GET"])
  463. def search_metadata_route():
  464. try:
  465. keyword = request.args.get("keyword", "")
  466. if not keyword:
  467. return jsonify(success([]))
  468. cypher = """
  469. MATCH (n:DataMeta)
  470. WHERE n.name_zh CONTAINS $keyword
  471. RETURN n LIMIT 100
  472. """
  473. with neo4j_driver.get_session() as session:
  474. result = session.run(cypher, keyword=keyword)
  475. metadata_list = [dict(record["n"]) for record in result]
  476. return jsonify(success(metadata_list))
  477. except Exception as e:
  478. logger.error(f"搜索元数据失败: {str(e)}")
  479. return jsonify(failed(str(e)))
  480. # 全文检索查询
  481. @bp.route("/full/text/query", methods=["POST"])
  482. def full_text_query():
  483. try:
  484. if not request.json:
  485. return jsonify(failed("请求数据不能为空"))
  486. # 获取查询条件
  487. search_term = request.json.get("query", "")
  488. if not search_term:
  489. return jsonify(failed("查询条件不能为空"))
  490. # 执行Neo4j全文索引查询
  491. with neo4j_driver.get_session() as session:
  492. cypher = """
  493. CALL db.index.fulltext.queryNodes("DataMetaFulltext", $term)
  494. YIELD node, score
  495. RETURN node, score
  496. ORDER BY score DESC
  497. LIMIT 20
  498. """
  499. result = session.run(cypher, term=search_term)
  500. # 处理查询结果
  501. search_results = []
  502. for record in result:
  503. node_data = dict(record["node"])
  504. node_data["id"] = record["node"].id
  505. node_data["score"] = record["score"]
  506. search_results.append(node_data)
  507. return jsonify(success(search_results))
  508. except Exception as e:
  509. logger.error(f"全文检索查询失败: {str(e)}")
  510. return jsonify(failed(str(e)))
  511. # 非结构化文本查询
  512. @bp.route("/unstructure/text/query", methods=["POST"])
  513. def unstructure_text_query():
  514. try:
  515. if not request.json:
  516. return jsonify(failed("请求数据不能为空"))
  517. # 获取查询参数
  518. node_id = request.json.get("id")
  519. if not node_id:
  520. return jsonify(failed("节点ID不能为空"))
  521. # 获取节点信息
  522. node_data = handle_id_unstructured(node_id)
  523. if not node_data:
  524. return jsonify(failed("节点不存在"))
  525. # 获取对象路径
  526. object_name = node_data.get("url")
  527. if not object_name:
  528. return jsonify(failed("文档路径不存在"))
  529. # 获取 MinIO 配置
  530. minio_client = get_minio_client()
  531. config = get_minio_config()
  532. bucket_name = config["MINIO_BUCKET"]
  533. # 从MinIO获取文件内容
  534. file_content = get_file_content(minio_client, bucket_name, object_name)
  535. # 解析文本内容
  536. parsed_data = parse_text(file_content)
  537. # 返回结果
  538. result = {
  539. "node": node_data,
  540. "parsed": parsed_data,
  541. "content": (
  542. file_content[:1000] + "..."
  543. if len(file_content) > 1000
  544. else file_content
  545. ),
  546. }
  547. return jsonify(success(result))
  548. except Exception as e:
  549. logger.error(f"非结构化文本查询失败: {str(e)}")
  550. return jsonify(failed(str(e)))
  551. # 文件上传
  552. @bp.route("/resource/upload", methods=["POST"])
  553. def upload_file():
  554. try:
  555. # 检查请求中是否有文件
  556. if "file" not in request.files:
  557. return jsonify(failed("没有找到上传的文件"))
  558. file = request.files["file"]
  559. # 检查文件名
  560. if not file.filename:
  561. return jsonify(failed("未选择文件"))
  562. # 保存文件名到本地变量(确保类型安全)
  563. filename = file.filename
  564. # 检查文件类型
  565. if not allowed_file(filename):
  566. return jsonify(failed("不支持的文件类型"))
  567. # 获取 MinIO 配置
  568. minio_client = get_minio_client()
  569. config = get_minio_config()
  570. # 上传到MinIO
  571. file_content = file.read()
  572. file_size = len(file_content)
  573. file_type = filename.rsplit(".", 1)[1].lower()
  574. # 提取文件名(不包含扩展名)
  575. filename_without_ext = filename.rsplit(".", 1)[0]
  576. # 生成紧凑的时间戳 (yyyyMMddHHmmss)
  577. import time
  578. timestamp = time.strftime("%Y%m%d%H%M%S", time.localtime())
  579. # 生成唯一文件名
  580. object_name = (
  581. f"{config['PREFIX']}/{filename_without_ext}_{timestamp}.{file_type}"
  582. )
  583. # 上传文件
  584. minio_client.put_object(
  585. config["MINIO_BUCKET"],
  586. object_name,
  587. io.BytesIO(file_content),
  588. file_size,
  589. content_type=f"application/{file_type}",
  590. )
  591. # 返回结果
  592. return jsonify(
  593. success(
  594. {
  595. "filename": file.filename,
  596. "size": file_size,
  597. "type": file_type,
  598. "url": object_name,
  599. }
  600. )
  601. )
  602. except Exception as e:
  603. logger.error(f"文件上传失败: {str(e)}")
  604. return jsonify(failed(str(e)))
  605. # 文件下载显示
  606. @bp.route("/resource/display", methods=["POST"])
  607. def upload_file_display():
  608. response = None
  609. try:
  610. if not request.json:
  611. return jsonify(failed("请求数据不能为空"))
  612. object_name = request.json.get("url")
  613. if not object_name:
  614. return jsonify(failed("文件路径不能为空"))
  615. # 获取 MinIO 配置
  616. minio_client = get_minio_client()
  617. config = get_minio_config()
  618. # 获取文件内容
  619. response = minio_client.get_object(config["MINIO_BUCKET"], object_name)
  620. file_data = response.read()
  621. # 获取文件名
  622. file_name = object_name.split("/")[-1]
  623. # 确定文件类型
  624. file_extension = file_name.split(".")[-1].lower()
  625. # 为不同文件类型设置合适的MIME类型
  626. mime_types = {
  627. "pdf": "application/pdf",
  628. "doc": "application/msword",
  629. "docx": (
  630. "application/vnd.openxmlformats-"
  631. "officedocument.wordprocessingml.document"
  632. ),
  633. "xls": "application/vnd.ms-excel",
  634. "xlsx": (
  635. "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
  636. ),
  637. "txt": "text/plain",
  638. "csv": "text/csv",
  639. }
  640. content_type = mime_types.get(file_extension, "application/octet-stream")
  641. # 返回结果
  642. return jsonify(
  643. success(
  644. {
  645. "filename": file_name,
  646. "type": file_extension,
  647. "contentType": content_type,
  648. "size": len(file_data),
  649. "url": f"/api/meta/resource/download?url={object_name}",
  650. }
  651. )
  652. )
  653. except S3Error as e:
  654. logger.error(f"MinIO操作失败: {str(e)}")
  655. return jsonify(failed(f"文件访问失败: {str(e)}"))
  656. except Exception as e:
  657. logger.error(f"文件显示信息获取失败: {str(e)}")
  658. return jsonify(failed(str(e)))
  659. finally:
  660. if response:
  661. response.close()
  662. response.release_conn()
  663. # 文件下载接口
  664. @bp.route("/resource/download", methods=["GET"])
  665. def download_file():
  666. response = None
  667. try:
  668. object_name = request.args.get("url")
  669. if not object_name:
  670. return jsonify(failed("文件路径不能为空"))
  671. # URL解码,处理特殊字符
  672. import urllib.parse
  673. object_name = urllib.parse.unquote(object_name)
  674. # 记录下载请求信息,便于调试
  675. logger.info(f"下载文件请求: {object_name}")
  676. # 获取 MinIO 配置
  677. minio_client = get_minio_client()
  678. config = get_minio_config()
  679. # 获取文件
  680. try:
  681. response = minio_client.get_object(config["MINIO_BUCKET"], object_name)
  682. file_data = response.read()
  683. except S3Error as e:
  684. logger.error(f"MinIO获取文件失败: {str(e)}")
  685. return jsonify(failed(f"文件获取失败: {str(e)}"))
  686. # 获取文件名,并处理特殊字符
  687. file_name = object_name.split("/")[-1]
  688. # 直接从内存返回文件,不创建临时文件
  689. file_stream = io.BytesIO(file_data)
  690. # 返回文件
  691. return send_file(
  692. file_stream,
  693. as_attachment=True,
  694. download_name=file_name,
  695. mimetype="application/octet-stream",
  696. )
  697. except Exception as e:
  698. logger.error(f"文件下载失败: {str(e)}")
  699. return jsonify(failed(str(e)))
  700. finally:
  701. if response:
  702. response.close()
  703. response.release_conn()
  704. # 文本资源翻译
  705. @bp.route("/resource/translate", methods=["POST"])
  706. def text_resource_translate():
  707. try:
  708. if not request.json:
  709. return jsonify(failed("请求数据不能为空"))
  710. # 获取参数
  711. name_zh = request.json.get("name_zh", "")
  712. keyword = request.json.get("keyword", "")
  713. if not name_zh:
  714. return jsonify(failed("名称不能为空"))
  715. # 调用资源处理逻辑
  716. result = text_resource_solve(None, name_zh, keyword)
  717. return jsonify(success(result))
  718. except Exception as e:
  719. logger.error(f"文本资源翻译失败: {str(e)}")
  720. return jsonify(failed(str(e)))
  721. # 创建文本资源节点
  722. @bp.route("/resource/node", methods=["POST"])
  723. def text_resource_node():
  724. try:
  725. if not request.json:
  726. return jsonify(failed("请求数据不能为空"))
  727. # 获取参数
  728. name_zh = request.json.get("name_zh", "")
  729. name_en = request.json.get("name_en", "")
  730. keywords = request.json.get("keywords", [])
  731. keywords_en = request.json.get("keywords_en", [])
  732. object_name = request.json.get("url", "")
  733. if not name_zh or not name_en or not object_name:
  734. return jsonify(failed("参数不完整"))
  735. # 创建节点
  736. with neo4j_driver.get_session() as session:
  737. # 创建资源节点
  738. cypher = """
  739. CREATE (n:DataMeta {
  740. uid: $uid,
  741. name_zh: $name_zh,
  742. name_en: $name_en,
  743. keywords: $keywords,
  744. keywords_en: $keywords_en,
  745. url: $object_name,
  746. create_time: $create_time,
  747. updateTime: $update_time
  748. })
  749. RETURN n
  750. """
  751. create_time = update_time = get_formatted_time()
  752. node_properties = {
  753. "name_zh": name_zh,
  754. "name_en": name_en,
  755. "keywords": keywords,
  756. "keywords_en": keywords_en,
  757. "object_name": object_name,
  758. "create_time": create_time,
  759. "update_time": update_time,
  760. }
  761. ensure_governance_uid(node_properties)
  762. result = session.run(
  763. cypher,
  764. **node_properties,
  765. )
  766. record = result.single()
  767. if not record:
  768. return jsonify(failed("创建节点失败"))
  769. node = record["n"]
  770. # 为每个关键词创建标签节点并关联
  771. for i, keyword in enumerate(keywords):
  772. if keyword:
  773. # 创建标签节点
  774. tag_cypher = """
  775. MERGE (t:Tag {name_zh: $name_zh})
  776. ON CREATE SET t.name_en = $name_en,
  777. t.create_time = $create_time
  778. RETURN t
  779. """
  780. tag_result = session.run(
  781. tag_cypher,
  782. name_zh=keyword,
  783. name_en=keywords_en[i] if i < len(keywords_en) else "",
  784. create_time=create_time,
  785. )
  786. tag_record = tag_result.single()
  787. if not tag_record:
  788. continue
  789. tag_node = tag_record["t"]
  790. # 创建关系
  791. rel_cypher = """
  792. MATCH (n), (t)
  793. WHERE id(n) = $node_id AND id(t) = $tag_id
  794. CREATE (n)-[r:HAS_TAG]->(t)
  795. RETURN r
  796. """
  797. session.run(rel_cypher, node_id=node.id, tag_id=tag_node.id)
  798. # 返回创建的节点
  799. return jsonify(success(dict(node)))
  800. except Exception as e:
  801. logger.error(f"创建文本资源节点失败: {str(e)}")
  802. return jsonify(failed(str(e)))
  803. # 处理非结构化数据
  804. @bp.route("/unstructured/process", methods=["POST"])
  805. def processing_unstructured_data():
  806. try:
  807. if not request.json:
  808. return jsonify(failed("请求数据不能为空"))
  809. # 获取参数
  810. node_id = request.json.get("id")
  811. if not node_id:
  812. return jsonify(failed("节点ID不能为空"))
  813. # 获取 MinIO 配置
  814. minio_client = get_minio_client()
  815. config = get_minio_config()
  816. prefix = config["PREFIX"]
  817. # 调用处理逻辑
  818. result = solve_unstructured_data(node_id, minio_client, prefix)
  819. if result:
  820. return jsonify(success({"message": "处理成功"}))
  821. else:
  822. return jsonify(failed("处理失败"))
  823. except Exception as e:
  824. logger.error(f"处理非结构化数据失败: {str(e)}")
  825. return jsonify(failed(str(e)))
  826. # 创建文本图谱
  827. @bp.route("/text/graph", methods=["POST"])
  828. def create_text_graph():
  829. try:
  830. if not request.json:
  831. return jsonify(failed("请求数据不能为空"))
  832. # 获取参数
  833. node_id = request.json.get("id")
  834. entity_zh = request.json.get("entity_zh")
  835. entity_en = request.json.get("entity_en")
  836. if not all([node_id, entity_zh, entity_en]):
  837. return jsonify(failed("参数不完整"))
  838. # 创建图谱
  839. result = handle_txt_graph(node_id, entity_zh, entity_en)
  840. if result:
  841. return jsonify(success({"message": "图谱创建成功"}))
  842. else:
  843. return jsonify(failed("图谱创建失败"))
  844. except Exception as e:
  845. logger.error(f"创建文本图谱失败: {str(e)}")
  846. return jsonify(failed(str(e)))
  847. @bp.route("/config", methods=["GET"])
  848. @require_auth
  849. def get_meta_config():
  850. """获取元数据配置信息"""
  851. config = get_minio_config()
  852. return jsonify(
  853. {
  854. "bucket_name": config["MINIO_BUCKET"],
  855. "prefix": config["PREFIX"],
  856. "allowed_extensions": list(config["ALLOWED_EXTENSIONS"]),
  857. }
  858. )
  859. # 更新元数据
  860. @bp.route("/node/update", methods=["POST"])
  861. def meta_node_update():
  862. """
  863. 更新元数据节点
  864. 在更新前会进行冗余检测(如果修改了 name_zh/name_en):
  865. - 如果更新后的名称与其他节点完全匹配,返回错误
  866. - 如果存在疑似重复的元数据,创建审核记录并返回提示
  867. - 如果无重复,正常更新节点
  868. """
  869. try:
  870. if not request.json:
  871. return jsonify(failed("请求数据不能为空"))
  872. # 从请求中获取节点ID和更新数据
  873. node_id = request.json.get("id")
  874. if not node_id:
  875. return jsonify(failed("节点ID不能为空"))
  876. # 验证并转换节点ID为整数
  877. try:
  878. node_id = int(node_id)
  879. except (ValueError, TypeError):
  880. return jsonify(failed(f"节点ID必须为整数,当前值: {node_id}"))
  881. # 是否强制更新(跳过冗余检测)
  882. force_update = bool(request.json.get("force_update", False))
  883. # 更新节点
  884. with neo4j_driver.get_session() as session:
  885. # 检查节点是否存在并获取当前值
  886. check_cypher = """
  887. MATCH (n:DataMeta)
  888. WHERE id(n) = $node_id
  889. RETURN n
  890. """
  891. result = session.run(check_cypher, node_id=node_id)
  892. node = result.single()
  893. if not node or not node["n"]:
  894. return jsonify(failed("节点不存在"))
  895. # 获取当前节点属性
  896. current_node = dict(node["n"])
  897. # 处理每个可能的更新字段
  898. fields_to_update = {
  899. "name_zh": request.json.get("name_zh"),
  900. "category": request.json.get("category"),
  901. "alias": request.json.get("alias"),
  902. "affiliation": request.json.get("affiliation"),
  903. "data_type": request.json.get("data_type"),
  904. "describe": request.json.get("describe"),
  905. "status": request.json.get("status"),
  906. "name_en": request.json.get("name_en"),
  907. }
  908. # 计算更新后的值(用于冗余检测)
  909. updated_name_zh = (
  910. fields_to_update["name_zh"]
  911. if fields_to_update["name_zh"] is not None
  912. else current_node.get("name_zh", "")
  913. )
  914. updated_name_en = (
  915. fields_to_update["name_en"]
  916. if fields_to_update["name_en"] is not None
  917. else current_node.get("name_en", "")
  918. )
  919. updated_data_type = (
  920. fields_to_update["data_type"]
  921. if fields_to_update["data_type"] is not None
  922. else current_node.get("data_type", "varchar(255)")
  923. )
  924. # 处理标签
  925. tag = request.json.get("tag")
  926. tag_ids = normalize_tag_inputs(tag) if tag is not None else []
  927. # ========== 冗余检测(仅当修改了 name_zh 或 name_en 时)==========
  928. name_changed = (
  929. fields_to_update["name_zh"] is not None
  930. and fields_to_update["name_zh"] != current_node.get("name_zh")
  931. ) or (
  932. fields_to_update["name_en"] is not None
  933. and fields_to_update["name_en"] != current_node.get("name_en")
  934. )
  935. if name_changed and not force_update:
  936. redundancy_result = check_redundancy_for_update(
  937. node_id=node_id,
  938. name_zh=updated_name_zh,
  939. name_en=updated_name_en,
  940. data_type=updated_data_type,
  941. tag_ids=tag_ids,
  942. )
  943. # 存在完全匹配的其他元数据
  944. if redundancy_result["has_exact_match"]:
  945. exact_id = redundancy_result["exact_match_id"]
  946. logger.warning(
  947. f"更新后元数据与其他节点完全匹配: "
  948. f"node_id={node_id}, existing_id={exact_id}"
  949. )
  950. return jsonify(
  951. failed(
  952. f"更新后的元数据与已有节点(ID={exact_id})完全相同,"
  953. f"请检查是否需要合并或修改名称。"
  954. )
  955. )
  956. # 存在疑似重复的元数据,已创建审核记录
  957. if redundancy_result["review_created"]:
  958. candidates = redundancy_result["candidates"]
  959. candidate_names = [c.get("name_zh", "") for c in candidates[:3]]
  960. logger.info(
  961. f"更新元数据发现疑似重复: node_id={node_id}, "
  962. f"candidates={candidate_names}"
  963. )
  964. return jsonify(
  965. failed(
  966. f"发现疑似重复元数据,已创建审核记录。"
  967. f"疑似重复: {', '.join(candidate_names)}。"
  968. f"请前往元数据审核页面处理,或使用 force_update=true 强制更新。"
  969. )
  970. )
  971. # ========== 执行更新 ==========
  972. # 构建更新语句,只更新提供的属性
  973. update_cypher = """
  974. MATCH (n:DataMeta)
  975. WHERE id(n) = $node_id
  976. SET n.updateTime = $update_time
  977. """
  978. # 准备更新参数
  979. update_params = {"node_id": node_id, "update_time": get_formatted_time()}
  980. # 只更新提供了新值的字段
  981. for field, new_value in fields_to_update.items():
  982. if new_value is not None:
  983. # 特殊处理 data_type 字段映射
  984. if field == "data_type":
  985. update_cypher += f", n.data_type = ${field}\n"
  986. else:
  987. update_cypher += f", n.{field} = ${field}\n"
  988. update_params[field] = new_value
  989. update_cypher += "RETURN n"
  990. result = session.run(
  991. update_cypher, # type: ignore[arg-type]
  992. **update_params,
  993. )
  994. updated_node = result.single()
  995. if updated_node and updated_node["n"]:
  996. node_data = dict(updated_node["n"])
  997. node_data["id"] = updated_node["n"].id
  998. # 如果更新了标签,处理标签关系(支持列表)
  999. if tag is not None:
  1000. # 先删除现有标签关系
  1001. delete_tag_cypher = """
  1002. MATCH (n:DataMeta)-[r:LABEL]->(t:DataLabel)
  1003. WHERE id(n) = $node_id
  1004. DELETE r
  1005. """
  1006. session.run(delete_tag_cypher, node_id=node_id)
  1007. for tag_id in tag_ids:
  1008. create_tag_cypher = """
  1009. MATCH (n:DataMeta), (t:DataLabel)
  1010. WHERE id(n) = $node_id AND id(t) = $tag_id
  1011. MERGE (n)-[r:LABEL]->(t)
  1012. RETURN r
  1013. """
  1014. session.run(create_tag_cypher, node_id=node_id, tag_id=tag_id)
  1015. logger.info(f"成功更新元数据节点: ID={node_data['id']}")
  1016. return jsonify(success(node_data))
  1017. else:
  1018. logger.error(f"更新元数据节点失败: ID={node_id}")
  1019. return jsonify(failed("更新元数据节点失败"))
  1020. except Exception as e:
  1021. logger.error(f"更新元数据失败: {str(e)}")
  1022. return jsonify(failed(str(e)))
  1023. @bp.route("/review/list", methods=["POST"])
  1024. def metadata_review_list():
  1025. """
  1026. 审核记录列表:疑似冗余/变动
  1027. Body:
  1028. - current: 页码(默认1)
  1029. - size: 每页数量(默认10)
  1030. - record_type: redundancy|change(可选)
  1031. - status: pending|resolved|ignored(可选)
  1032. - business_domain_id: 业务领域ID(可选)
  1033. - keyword: 关键字(可选,匹配 new_meta.name_zh/name_en)
  1034. """
  1035. try:
  1036. payload = request.get_json() or {}
  1037. if not isinstance(payload, dict):
  1038. return jsonify(failed("请求数据格式错误,应为 JSON 对象"))
  1039. def to_int(value, default):
  1040. try:
  1041. return int(value)
  1042. except (TypeError, ValueError):
  1043. return default
  1044. page = to_int(payload.get("current", 1), 1)
  1045. page_size = to_int(payload.get("size", 10), 10)
  1046. record_type = payload.get("record_type")
  1047. status = payload.get("status")
  1048. business_domain_id = payload.get("business_domain_id")
  1049. keyword = (payload.get("keyword") or "").strip()
  1050. query = MetadataReviewRecord.query
  1051. if record_type:
  1052. query = query.filter(MetadataReviewRecord.record_type == record_type)
  1053. if status:
  1054. query = query.filter(MetadataReviewRecord.status == status)
  1055. if business_domain_id is not None and str(business_domain_id).strip() != "":
  1056. bd_id_int = int(business_domain_id)
  1057. query = query.filter(MetadataReviewRecord.business_domain_id == bd_id_int)
  1058. if keyword:
  1059. # 兼容:使用JSONB ->> 提取进行模糊匹配
  1060. name_zh_col = MetadataReviewRecord.new_meta["name_zh"].astext
  1061. name_en_col = MetadataReviewRecord.new_meta["name_en"].astext
  1062. query = query.filter(
  1063. or_(
  1064. name_zh_col.contains(keyword),
  1065. name_en_col.contains(keyword),
  1066. )
  1067. )
  1068. total = query.count()
  1069. records = (
  1070. query.order_by(MetadataReviewRecord.created_at.desc())
  1071. .offset((page - 1) * page_size)
  1072. .limit(page_size)
  1073. .all()
  1074. )
  1075. # 将 tag_ids 转换为 tags
  1076. records_data = [convert_tag_ids_to_tags(r.to_dict()) for r in records]
  1077. return jsonify(
  1078. success(
  1079. {
  1080. "records": records_data,
  1081. "total": total,
  1082. "size": page_size,
  1083. "current": page,
  1084. }
  1085. )
  1086. )
  1087. except Exception as e:
  1088. logger.error(f"审核记录列表查询失败: {str(e)}")
  1089. return jsonify(failed("审核记录列表查询失败", error=str(e)))
  1090. @bp.route("/review/create", methods=["POST"])
  1091. def metadata_review_create():
  1092. """
  1093. 创建元数据审核记录
  1094. Body:
  1095. - record_type: 审核记录类型(redundancy: 疑似重复 / change: 疑似变动 / merge: 合并请求)
  1096. - source: 触发来源(默认 "manual")
  1097. - meta1: 第一个元数据信息
  1098. - id: 节点ID
  1099. - name_zh: 中文名
  1100. - name_en: 英文名
  1101. - data_type: 数据类型
  1102. - status: 状态
  1103. - tag_ids: 标签ID列表(可选)
  1104. - meta2: 第二个元数据信息
  1105. - id: 节点ID
  1106. - name_zh: 中文名
  1107. - name_en: 英文名
  1108. - data_type: 数据类型
  1109. - status: 状态
  1110. - tag_ids: 标签ID列表(可选)
  1111. - diff_fields: 差异字段列表(可选,如 ["name_zh", "name_en"])
  1112. - notes: 备注(可选)
  1113. Returns:
  1114. 创建成功的审核记录信息
  1115. """
  1116. try:
  1117. payload = request.get_json() or {}
  1118. if not isinstance(payload, dict):
  1119. return jsonify(failed("请求数据格式错误,应为 JSON 对象"))
  1120. record_type = payload.get("record_type")
  1121. source = payload.get("source", "manual")
  1122. meta1 = payload.get("meta1")
  1123. meta2 = payload.get("meta2")
  1124. diff_fields = payload.get("diff_fields", [])
  1125. notes = payload.get("notes")
  1126. # 参数校验
  1127. if not record_type:
  1128. return jsonify(failed("record_type 不能为空"))
  1129. if record_type not in ("redundancy", "change", "merge"):
  1130. return jsonify(
  1131. failed("record_type 必须是 redundancy、change 或 merge 之一")
  1132. )
  1133. if not meta1 or not isinstance(meta1, dict):
  1134. return jsonify(failed("meta1 不能为空且必须是对象"))
  1135. if not meta2 or not isinstance(meta2, dict):
  1136. return jsonify(failed("meta2 不能为空且必须是对象"))
  1137. if not isinstance(diff_fields, list):
  1138. return jsonify(failed("diff_fields 必须是数组"))
  1139. # 校验元数据必要字段
  1140. required_fields = ["id", "name_zh", "name_en", "data_type", "status"]
  1141. for field in required_fields:
  1142. if field not in meta1:
  1143. return jsonify(failed(f"meta1 缺少必要字段: {field}"))
  1144. if field not in meta2:
  1145. return jsonify(failed(f"meta2 缺少必要字段: {field}"))
  1146. # 构建 new_meta(主元数据信息)
  1147. new_meta = {
  1148. "id": meta1.get("id"),
  1149. "name_zh": meta1.get("name_zh"),
  1150. "name_en": meta1.get("name_en"),
  1151. "data_type": meta1.get("data_type"),
  1152. "status": meta1.get("status"),
  1153. "tag_ids": meta1.get("tag_ids", []),
  1154. }
  1155. # 构建 candidates(候选/对比元数据列表)
  1156. # 格式: [{"snapshot": {...}, "diff_fields": [...], "candidate_meta_id": id}]
  1157. candidates = [
  1158. {
  1159. "snapshot": {
  1160. "id": meta2.get("id"),
  1161. "name_zh": meta2.get("name_zh"),
  1162. "name_en": meta2.get("name_en"),
  1163. "data_type": meta2.get("data_type"),
  1164. "status": meta2.get("status"),
  1165. "tag_ids": meta2.get("tag_ids", []),
  1166. },
  1167. "diff_fields": diff_fields,
  1168. "candidate_meta_id": meta2.get("id"),
  1169. }
  1170. ]
  1171. # 创建审核记录
  1172. review_record = MetadataReviewRecord()
  1173. review_record.record_type = record_type
  1174. review_record.source = source
  1175. review_record.new_meta = new_meta
  1176. review_record.candidates = candidates
  1177. review_record.status = "pending"
  1178. review_record.notes = notes
  1179. db.session.add(review_record)
  1180. db.session.commit()
  1181. logger.info(
  1182. f"创建审核记录成功: id={review_record.id}, "
  1183. f"record_type={record_type}, "
  1184. f"meta1_name={meta1.get('name_zh')}, "
  1185. f"meta2_name={meta2.get('name_zh')}"
  1186. )
  1187. return jsonify(
  1188. success(
  1189. {
  1190. "record": review_record.to_dict(),
  1191. "message": "审核记录创建成功,请前往数据审核页面进行处理",
  1192. }
  1193. )
  1194. )
  1195. except Exception as e:
  1196. logger.error(f"创建审核记录失败: {str(e)}")
  1197. db.session.rollback()
  1198. return jsonify(failed("创建审核记录失败", error=str(e)))
  1199. @bp.route("/review/detail", methods=["GET"])
  1200. def metadata_review_detail():
  1201. """
  1202. 审核记录详情
  1203. Query:
  1204. - id: 记录ID
  1205. """
  1206. try:
  1207. record_id = request.args.get("id")
  1208. if not record_id:
  1209. return jsonify(failed("缺少id参数"))
  1210. record = MetadataReviewRecord.query.get(int(record_id))
  1211. if not record:
  1212. return jsonify(failed("记录不存在"))
  1213. # 将 tag_ids 转换为 tags
  1214. data = convert_tag_ids_to_tags(record.to_dict())
  1215. # change 场景:返回受影响元数据的影响关系图谱(若有 meta_id)
  1216. impact_graph = None
  1217. if record.record_type == "change":
  1218. old_meta = record.old_meta or {}
  1219. meta_id = old_meta.get("meta_id")
  1220. if meta_id is not None and str(meta_id).strip() != "":
  1221. try:
  1222. impact_graph = meta_impact_graph(int(meta_id))
  1223. except Exception as e:
  1224. logger.warning(f"获取影响图谱失败: {e}")
  1225. data["impact_graph"] = impact_graph
  1226. return jsonify(success(data))
  1227. except Exception as e:
  1228. logger.error(f"审核记录详情查询失败: {str(e)}")
  1229. return jsonify(failed("审核记录详情查询失败", error=str(e)))
  1230. @bp.route("/review/resolve", methods=["POST"])
  1231. def metadata_review_resolve():
  1232. """
  1233. 处理审核记录
  1234. Body:
  1235. - id: 记录ID
  1236. - action: alias | create_new | accept_change | reject_change | ignore
  1237. - payload: 动作参数(可选)
  1238. - resolved_by: 处理人(可选)
  1239. - notes: 备注(可选)
  1240. action=alias:
  1241. payload: { primary_meta_id: int, alias_meta_id: int }
  1242. 行为:在 DataMeta 节点之间重建 ALIAS 关系
  1243. - 创建 (alias_meta)-[:ALIAS]->(primary_meta) 关系
  1244. - 将所有指向 alias_meta 的 ALIAS 关系转移到 primary_meta
  1245. - primary_meta 已有的 ALIAS 关系保持不变
  1246. - BusinessDomain 的 INCLUDES 关系不受影响
  1247. action=create_new:
  1248. payload: { new_name_zh: str }
  1249. 行为:创建新的 DataMeta(中文名区分)并关联业务领域
  1250. action=accept_change:
  1251. payload: { meta_id?: int }
  1252. 行为:把 new_meta 写回目标 DataMeta,并写入 metadata_version_history(PG)
  1253. action=reject_change/ignore:
  1254. 行为:仅更新审核记录状态
  1255. """
  1256. try:
  1257. payload = request.get_json() or {}
  1258. if not isinstance(payload, dict):
  1259. return jsonify(failed("请求数据格式错误,应为 JSON 对象"))
  1260. record_id = payload.get("id")
  1261. action = payload.get("action")
  1262. action_payload = payload.get("payload") or {}
  1263. resolved_by = payload.get("resolved_by")
  1264. notes = payload.get("notes")
  1265. if not record_id:
  1266. return jsonify(failed("id 不能为空"))
  1267. if not action:
  1268. return jsonify(failed("action 不能为空"))
  1269. record = MetadataReviewRecord.query.get(int(record_id))
  1270. if not record:
  1271. return jsonify(failed("记录不存在"))
  1272. if record.status != "pending":
  1273. return jsonify(failed("记录已处理,无法重复处理"))
  1274. # 需要业务领域上下文的动作
  1275. bd_id = record.business_domain_id
  1276. new_meta = record.new_meta or {}
  1277. if action == "alias":
  1278. primary_meta_id = action_payload.get("primary_meta_id")
  1279. alias_meta_id = action_payload.get("alias_meta_id")
  1280. if not primary_meta_id:
  1281. return jsonify(failed("payload.primary_meta_id 不能为空"))
  1282. if not alias_meta_id:
  1283. return jsonify(failed("payload.alias_meta_id 不能为空"))
  1284. if int(primary_meta_id) == int(alias_meta_id):
  1285. return jsonify(failed("primary_meta_id 和 alias_meta_id 不能相同"))
  1286. # 写入 Neo4j:重建 DataMeta 节点间的 ALIAS 关系
  1287. from app.services.neo4j_driver import neo4j_driver
  1288. with neo4j_driver.get_session() as session:
  1289. # Step 1: 将所有指向 alias_meta 的 ALIAS 关系转移到 primary_meta
  1290. # 查找所有以 alias_meta 为目标的 ALIAS 关系,创建新关系指向 primary_meta,然后删除旧关系
  1291. session.run(
  1292. """
  1293. MATCH (other:DataMeta)-[old_rel:ALIAS]->(alias_meta:DataMeta)
  1294. WHERE id(alias_meta) = $alias_meta_id
  1295. WITH other, old_rel
  1296. MATCH (primary_meta:DataMeta)
  1297. WHERE id(primary_meta) = $primary_meta_id
  1298. MERGE (other)-[:ALIAS]->(primary_meta)
  1299. DELETE old_rel
  1300. """,
  1301. {
  1302. "alias_meta_id": int(alias_meta_id),
  1303. "primary_meta_id": int(primary_meta_id),
  1304. },
  1305. )
  1306. # Step 2: 创建 alias_meta 指向 primary_meta 的 ALIAS 关系
  1307. session.run(
  1308. """
  1309. MATCH (alias_meta:DataMeta), (primary_meta:DataMeta)
  1310. WHERE id(alias_meta) = $alias_meta_id AND id(primary_meta) = $primary_meta_id
  1311. MERGE (alias_meta)-[:ALIAS]->(primary_meta)
  1312. """,
  1313. {
  1314. "alias_meta_id": int(alias_meta_id),
  1315. "primary_meta_id": int(primary_meta_id),
  1316. },
  1317. )
  1318. update_review_record_resolution(
  1319. record,
  1320. action="alias",
  1321. payload={
  1322. "primary_meta_id": int(primary_meta_id),
  1323. "alias_meta_id": int(alias_meta_id),
  1324. },
  1325. resolved_by=resolved_by,
  1326. notes=notes,
  1327. )
  1328. db.session.commit()
  1329. return jsonify(success(record.to_dict()))
  1330. if action == "create_new":
  1331. new_name_zh = (action_payload.get("new_name_zh") or "").strip()
  1332. if not bd_id:
  1333. return jsonify(
  1334. failed("记录缺少 business_domain_id,无法执行 create_new")
  1335. )
  1336. if not new_name_zh:
  1337. return jsonify(failed("payload.new_name_zh 不能为空"))
  1338. from app.core.meta_data import get_formatted_time
  1339. from app.services.neo4j_driver import neo4j_driver
  1340. with neo4j_driver.get_session() as session:
  1341. # 创建新 DataMeta(避免覆盖旧节点)
  1342. node_properties = {
  1343. "name_zh": new_name_zh,
  1344. "name_en": (new_meta.get("name_en") or "").strip(),
  1345. "data_type": (new_meta.get("data_type") or "varchar(255)"),
  1346. "create_time": get_formatted_time(),
  1347. }
  1348. ensure_governance_uid(node_properties)
  1349. result = session.run(
  1350. """
  1351. CREATE (m:DataMeta {
  1352. uid: $uid,
  1353. name_zh: $name_zh,
  1354. name_en: $name_en,
  1355. data_type: $data_type,
  1356. create_time: $create_time,
  1357. status: true
  1358. })
  1359. RETURN m
  1360. """,
  1361. node_properties,
  1362. ).single()
  1363. if not result or not result.get("m"):
  1364. return jsonify(failed("创建新元数据失败"))
  1365. new_meta_id = int(result["m"].id)
  1366. session.run(
  1367. """
  1368. MATCH (n:BusinessDomain), (m:DataMeta)
  1369. WHERE id(n) = $domain_id AND id(m) = $meta_id
  1370. MERGE (n)-[:INCLUDES]->(m)
  1371. """,
  1372. {"domain_id": int(bd_id), "meta_id": new_meta_id},
  1373. )
  1374. update_review_record_resolution(
  1375. record,
  1376. action="create_new",
  1377. payload={"new_name_zh": new_name_zh},
  1378. resolved_by=resolved_by,
  1379. notes=notes,
  1380. )
  1381. db.session.commit()
  1382. return jsonify(success(record.to_dict()))
  1383. if action == "accept_change":
  1384. old_meta = record.old_meta or {}
  1385. meta_id = action_payload.get("meta_id") or old_meta.get("meta_id")
  1386. if not meta_id:
  1387. return jsonify(failed("无法确定需要更新的 meta_id"))
  1388. from app.core.meta_data import get_formatted_time
  1389. from app.services.neo4j_driver import neo4j_driver
  1390. before_snapshot = old_meta.get("snapshot") or {}
  1391. after_snapshot = new_meta
  1392. # 写入 Neo4j:更新 DataMeta 属性,并尝试同步标签集合
  1393. with neo4j_driver.get_session() as session:
  1394. name_zh_val = (
  1395. after_snapshot.get("name_zh")
  1396. or before_snapshot.get("name_zh")
  1397. or ""
  1398. ).strip()
  1399. name_en_val = (after_snapshot.get("name_en") or "").strip()
  1400. data_type_val = after_snapshot.get("data_type") or "varchar(255)"
  1401. session.run(
  1402. """
  1403. MATCH (m:DataMeta)
  1404. WHERE id(m) = $meta_id
  1405. SET m.name_zh = $name_zh,
  1406. m.name_en = $name_en,
  1407. m.data_type = $data_type,
  1408. m.updateTime = $update_time,
  1409. m.status = true
  1410. """,
  1411. {
  1412. "meta_id": int(meta_id),
  1413. "name_zh": name_zh_val,
  1414. "name_en": name_en_val,
  1415. "data_type": data_type_val,
  1416. "update_time": get_formatted_time(),
  1417. },
  1418. )
  1419. tag_ids = after_snapshot.get("tag_ids") or []
  1420. tag_ids = [int(t) for t in tag_ids if t is not None]
  1421. if tag_ids:
  1422. session.run(
  1423. """
  1424. MATCH (m:DataMeta)-[r:LABEL]->(:DataLabel)
  1425. WHERE id(m) = $meta_id
  1426. DELETE r
  1427. """,
  1428. {"meta_id": int(meta_id)},
  1429. )
  1430. session.run(
  1431. """
  1432. MATCH (m:DataMeta)
  1433. WHERE id(m) = $meta_id
  1434. WITH m
  1435. UNWIND $tag_ids AS tid
  1436. MATCH (t:DataLabel) WHERE id(t) = tid
  1437. MERGE (m)-[:LABEL]->(t)
  1438. """,
  1439. {"meta_id": int(meta_id), "tag_ids": tag_ids},
  1440. )
  1441. # 写入版本历史(PG)
  1442. history = MetadataVersionHistory()
  1443. history.meta_id = int(meta_id) if meta_id is not None else 0
  1444. history.change_source = "ddl"
  1445. history.before_snapshot = (
  1446. before_snapshot if before_snapshot is not None else {}
  1447. )
  1448. history.after_snapshot = (
  1449. after_snapshot if after_snapshot is not None else {}
  1450. )
  1451. history.created_by = resolved_by if resolved_by is not None else ""
  1452. db.session.add(history)
  1453. update_review_record_resolution(
  1454. record,
  1455. action="accept_change",
  1456. payload={"meta_id": int(meta_id)},
  1457. resolved_by=resolved_by,
  1458. notes=notes,
  1459. )
  1460. db.session.commit()
  1461. return jsonify(success(record.to_dict()))
  1462. if action in ("reject_change", "ignore"):
  1463. update_review_record_resolution(
  1464. record,
  1465. action=action,
  1466. payload=action_payload,
  1467. resolved_by=resolved_by,
  1468. notes=notes,
  1469. )
  1470. db.session.commit()
  1471. return jsonify(success(record.to_dict()))
  1472. return jsonify(failed(f"不支持的action: {action}"))
  1473. except Exception as e:
  1474. logger.error(f"处理审核记录失败: {str(e)}")
  1475. db.session.rollback()
  1476. return jsonify(failed("处理审核记录失败", error=str(e)))