model.py 61 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525
  1. """
  2. 数据模型核心业务逻辑模块
  3. 本模块包含了数据模型相关的所有核心业务逻辑函数,包括:
  4. - 数据模型的创建、更新、删除
  5. - 数据模型与数据资源、元数据之间的关系处理
  6. - 数据模型血缘关系管理
  7. - 数据模型图谱生成
  8. - 数据模型层级计算等功能
  9. """
  10. import math
  11. import threading
  12. from concurrent.futures import ThreadPoolExecutor
  13. import pandas as pd
  14. from py2neo import Relationship
  15. import logging
  16. import json
  17. # Configure logger
  18. logger = logging.getLogger(__name__)
  19. from app.core.graph.graph_operations import relationship_exists
  20. from app.core.graph.graph_operations import connect_graph,create_or_get_node,get_node
  21. from app.services.neo4j_driver import neo4j_driver
  22. from app.core.meta_data import get_formatted_time, handle_id_unstructured
  23. from app.core.common import delete_relationships, update_or_create_node, get_node_by_id_no_label
  24. from app.core.data_resource.resource import get_node_by_id, serialize_node_properties
  25. # 根据child关系计算数据模型当前的level自动保存
  26. def calculate_model_level(id):
  27. """
  28. 根据child关系计算数据模型当前的level并自动保存
  29. Args:
  30. id: 数据模型的节点ID(整数)
  31. Returns:
  32. None
  33. """
  34. # 确保id是整数类型
  35. node_id = int(id) if id is not None else None
  36. cql = """
  37. MATCH (start_node:DataModel)
  38. WHERE id(start_node) = $nodeId
  39. CALL {
  40. WITH start_node
  41. OPTIONAL MATCH path = (start_node)-[:child*]->(end_node)
  42. RETURN length(path) AS level
  43. }
  44. WITH coalesce(max(level), 0) AS max_level
  45. RETURN max_level
  46. """
  47. with connect_graph().session() as session:
  48. result = session.run(cql, nodeId=node_id)
  49. record = result.single()
  50. data = record["max_level"] if record and "max_level" in record else 0
  51. # 更新level属性
  52. update_query = """
  53. MATCH (n:DataModel)
  54. WHERE id(n) = $nodeId
  55. SET n.level = $level
  56. RETURN n
  57. """
  58. with connect_graph().session() as session:
  59. session.run(update_query, nodeId=node_id, level=data)
  60. # 处理数据模型血缘关系
  61. def handle_model_relation(resource_ids):
  62. """
  63. 处理数据模型血缘关系
  64. Args:
  65. resource_ids: 数据资源ID
  66. Returns:
  67. 血缘关系数据
  68. """
  69. query = """
  70. MATCH (search:DataResource)-[:connection]->(common_node:meta_node)<-[:connection]-(connect:DataResource)
  71. WHERE id(search) = $resource_Ids
  72. WITH search, connect, common_node
  73. MATCH (search)-[:connection]->(search_node:meta_node)
  74. WITH search, connect, common_node, collect(DISTINCT id(search_node)) AS search_nodes
  75. MATCH (connect)-[:connection]->(connect_node:meta_node)
  76. WITH search, connect, common_node, search_nodes, collect(DISTINCT id(connect_node)) AS connect_nodes
  77. WITH search, connect, search_nodes, connect_nodes, collect(DISTINCT id(common_node)) AS common_nodes
  78. // 剔除 search_nodes 和 connect_nodes 中包含在 common_nodes 中的内容
  79. WITH search, connect, common_nodes,
  80. [node IN search_nodes WHERE NOT node IN common_nodes] AS filtered_search_nodes,
  81. [node IN connect_nodes WHERE NOT node IN common_nodes] AS filtered_connect_nodes
  82. RETURN id(connect) as blood_resources, common_nodes,
  83. filtered_search_nodes as origin_nodes, filtered_connect_nodes as blood_nodes
  84. """
  85. with connect_graph().session() as session:
  86. result = session.run(query, resource_Ids=resource_ids)
  87. return result.data()
  88. # 创建一个数据模型节点
  89. def handle_data_model(data_model, result_list, result, receiver):
  90. """
  91. 创建一个数据模型节点
  92. Args:
  93. data_model: 数据模型名称
  94. result_list: 数据模型英文名列表
  95. result: 序列化的ID列表
  96. receiver: 接收到的请求参数
  97. Returns:
  98. tuple: (id, data_model_node)
  99. """
  100. try:
  101. # 添加数据资源 血缘关系的字段 blood_resource
  102. data_model_en = result_list[0] if result_list and len(result_list) > 0 else ""
  103. receiver['id_list'] = result
  104. add_attribute = {
  105. 'create_time': get_formatted_time(),
  106. 'name_en': data_model_en
  107. }
  108. receiver.update(add_attribute)
  109. data_model_node = get_node('DataModel', name_zh=data_model) or create_or_get_node('DataModel', **receiver)
  110. logger.info(f"通过查询或创建节点获得节点ID111,data_model_node: {data_model_node}")
  111. # 获取节点ID,确保我们能安全地访问节点ID
  112. node_id = data_model_node
  113. if hasattr(data_model_node, 'id'):
  114. logger.info(f"通过节点ID获取节点ID222,data_model_node: {data_model_node}")
  115. node_id = data_model_node.id
  116. else:
  117. logger.info(f"通过查询节点名称获取节点ID333,data_model_node: {data_model_node}")
  118. # 如果节点没有id属性,尝试通过查询获取
  119. query = """
  120. MATCH (n:DataModel {name: $name})
  121. RETURN id(n) as node_id
  122. """
  123. with connect_graph().session() as session:
  124. result = session.run(query, name=data_model)
  125. record = result.single()
  126. logger.info(f"通过查询节点名称获取节点ID444,record: {record}")
  127. if record and "node_id" in record:
  128. logger.info(f"通过查询节点名称获取节点ID555,record: {record}")
  129. node_id = record["node_id"]
  130. # 安全地处理子节点关系
  131. child_list = receiver.get('childrenId', [])
  132. for child_id in child_list:
  133. child_node = get_node_by_id_no_label(child_id)
  134. if child_node:
  135. # 直接使用Cypher查询检查关系是否存在
  136. with connect_graph().session() as session:
  137. rel_query = """
  138. MATCH (a)-[r:child]->(b)
  139. WHERE id(a) = $start_id AND id(b) = $end_id
  140. RETURN count(r) > 0 as exists
  141. """
  142. rel_result = session.run(rel_query,
  143. start_id=int(node_id),
  144. end_id=int(child_node.id)).single()
  145. # 如果关系不存在,则创建关系
  146. if not (rel_result and rel_result["exists"]):
  147. session.execute_write(
  148. lambda tx: tx.run(
  149. "MATCH (a), (b) WHERE id(a) = $a_id AND id(b) = $b_id CREATE (a)-[:child]->(b)",
  150. a_id=int(node_id), b_id=int(child_node.id)
  151. )
  152. )
  153. # 根据传入参数id,和数据标签建立关系
  154. if receiver.get('tag'):
  155. tag = get_node_by_id('DataLabel', receiver['tag'])
  156. if tag:
  157. # 直接使用Cypher查询检查关系是否存在
  158. with connect_graph().session() as session:
  159. rel_query = """
  160. MATCH (a)-[r:LABEL]->(b)
  161. WHERE id(a) = $start_id AND id(b) = $end_id
  162. RETURN count(r) > 0 as exists
  163. """
  164. rel_result = session.run(rel_query,
  165. start_id=int(node_id),
  166. end_id=int(tag.id)).single()
  167. # 如果关系不存在,则创建关系
  168. if not (rel_result and rel_result["exists"]):
  169. session.execute_write(
  170. lambda tx: tx.run(
  171. "MATCH (a), (b) WHERE id(a) = $a_id AND id(b) = $b_id CREATE (a)-[:LABEL]->(b)",
  172. a_id=int(node_id), b_id=int(tag.id)
  173. )
  174. )
  175. # 处理数据源关系 - 创建COME_FROM关系
  176. data_source = receiver.get('data_source')
  177. if data_source:
  178. try:
  179. # 获取数据源的标识(支持多种格式)
  180. data_source_id = None
  181. data_source_name_en = None
  182. # 1. 如果是数字(节点ID)
  183. if isinstance(data_source, (int, float)) or (isinstance(data_source, str) and data_source.isdigit()):
  184. data_source_id = int(data_source)
  185. logger.info(f"data_source 为节点ID: {data_source_id}")
  186. # 2. 如果是字典且包含name_en
  187. elif isinstance(data_source, dict) and data_source.get('name_en'):
  188. data_source_name_en = data_source['name_en']
  189. logger.info(f"data_source 为字典,提取name_en: {data_source_name_en}")
  190. # 3. 如果是字符串(name_en)
  191. elif isinstance(data_source, str):
  192. data_source_name_en = data_source
  193. logger.info(f"data_source 为字符串name_en: {data_source_name_en}")
  194. # 创建数据模型与数据源的关系
  195. with connect_graph().session() as session:
  196. if data_source_id is not None:
  197. # 使用节点ID创建关系
  198. # 首先检查数据源节点是否存在
  199. check_ds_cypher = "MATCH (b:DataSource) WHERE id(b) = $ds_id RETURN b"
  200. check_ds_result = session.run(check_ds_cypher, ds_id=data_source_id)
  201. if not check_ds_result.single():
  202. logger.warning(f"数据源节点不存在: ID={data_source_id},跳过关系创建")
  203. else:
  204. # 检查关系是否已存在
  205. rel_check_query = """
  206. MATCH (a:DataModel)-[r:COME_FROM]->(b:DataSource)
  207. WHERE id(a) = $model_id AND id(b) = $ds_id
  208. RETURN count(r) > 0 as exists
  209. """
  210. rel_check_result = session.run(rel_check_query,
  211. model_id=int(node_id),
  212. ds_id=data_source_id).single()
  213. # 如果关系不存在,则创建COME_FROM关系
  214. if not (rel_check_result and rel_check_result["exists"]):
  215. create_rel_cypher = """
  216. MATCH (a:DataModel), (b:DataSource)
  217. WHERE id(a) = $model_id AND id(b) = $ds_id
  218. CREATE (a)-[r:COME_FROM]->(b)
  219. RETURN r
  220. """
  221. session.run(create_rel_cypher,
  222. model_id=int(node_id),
  223. ds_id=data_source_id)
  224. logger.info(f"已创建数据模型与数据源的COME_FROM关系: model_id={node_id} -> data_source_id={data_source_id}")
  225. else:
  226. logger.info(f"数据模型与数据源的COME_FROM关系已存在: model_id={node_id} -> data_source_id={data_source_id}")
  227. elif data_source_name_en:
  228. # 使用name_en创建关系(兼容旧方式)
  229. # 首先检查数据源节点是否存在
  230. check_ds_cypher = "MATCH (b:DataSource) WHERE b.name_en = $ds_name_en RETURN b"
  231. check_ds_result = session.run(check_ds_cypher, ds_name_en=data_source_name_en)
  232. if not check_ds_result.single():
  233. logger.warning(f"数据源节点不存在: name_en={data_source_name_en},跳过关系创建")
  234. else:
  235. # 检查关系是否已存在
  236. rel_check_query = """
  237. MATCH (a:DataModel)-[r:COME_FROM]->(b:DataSource)
  238. WHERE id(a) = $model_id AND b.name_en = $ds_name_en
  239. RETURN count(r) > 0 as exists
  240. """
  241. rel_check_result = session.run(rel_check_query,
  242. model_id=int(node_id),
  243. ds_name_en=data_source_name_en).single()
  244. # 如果关系不存在,则创建COME_FROM关系
  245. if not (rel_check_result and rel_check_result["exists"]):
  246. create_rel_cypher = """
  247. MATCH (a:DataModel), (b:DataSource)
  248. WHERE id(a) = $model_id AND b.name_en = $ds_name_en
  249. CREATE (a)-[r:COME_FROM]->(b)
  250. RETURN r
  251. """
  252. session.run(create_rel_cypher,
  253. model_id=int(node_id),
  254. ds_name_en=data_source_name_en)
  255. logger.info(f"已创建数据模型与数据源的COME_FROM关系: model_id={node_id} -> name_en={data_source_name_en}")
  256. else:
  257. logger.info(f"数据模型与数据源的COME_FROM关系已存在: model_id={node_id} -> name_en={data_source_name_en}")
  258. else:
  259. logger.warning(f"data_source参数无效,无法识别格式: {data_source}")
  260. except Exception as e:
  261. # 数据源关系创建失败不应该中断主流程
  262. logger.error(f"处理数据源关系失败(不中断主流程): {str(e)}")
  263. # 不再抛出异常,允许主流程继续
  264. return node_id, data_model_node
  265. except Exception as e:
  266. logging.error(f"Error in handle_data_model: {str(e)}")
  267. raise
  268. # (从数据资源中选取)
  269. def resource_handle_meta_data_model(id_lists, data_model_node_id):
  270. """
  271. 处理从数据资源中选取的数据模型与元数据的关系
  272. Args:
  273. id_lists: ID列表
  274. data_model_node_id: 数据模型节点ID
  275. Returns:
  276. None
  277. """
  278. try:
  279. logger.info(f"开始处理数据模型与元数据的关系,数据模型ID: {data_model_node_id}")
  280. # 构建meta_id和resouce_id的列表
  281. resouce_ids = [record['resource_id'] for record in id_lists]
  282. meta_ids = [record['id'] for id_list in id_lists for record in id_list['metaData']]
  283. logger.info(f"资源ID列表: {resouce_ids}")
  284. logger.info(f"元数据ID列表: {meta_ids}")
  285. # 创建与meta_node的关系 组成关系
  286. if meta_ids:
  287. logger.info("开始创建数据模型与元数据的关系")
  288. query = """
  289. MATCH (source:DataModel), (target:DataMeta)
  290. WHERE id(source)=$source_id AND id(target) IN $target_ids
  291. MERGE (source)-[:INCLUDES]->(target)
  292. RETURN count(*) as count
  293. """
  294. with connect_graph().session() as session:
  295. result = session.run(query, source_id=data_model_node_id, target_ids=meta_ids)
  296. count = result.single()["count"]
  297. logger.info(f"成功创建 {count} 个数据模型与元数据的关系")
  298. # 创建与DataResource的关系 资源关系
  299. # 不在创建Modle时创建资源关系,将资源关系创建放在数据流程创建时处理
  300. # 关系名称为DERIVED_FROM
  301. # commented by mxl 2025-06-27
  302. #
  303. # if resouce_ids:
  304. # logger.info("开始创建数据模型与数据资源的关系")
  305. # query = """
  306. # MATCH (source:DataModel), (target:DataResource)
  307. # WHERE id(source)=$source_id AND id(target) IN $target_ids
  308. # MERGE (source)-[:DERIVED_FROM]->(target)
  309. # RETURN count(*) as count
  310. # """
  311. # with connect_graph().session() as session:
  312. # result = session.run(query, source_id=data_model_node_id, target_ids=resouce_ids)
  313. # count = result.single()["count"]
  314. # logger.info(f"成功创建 {count} 个数据模型与数据资源的关系")
  315. except Exception as e:
  316. logger.error(f"处理数据模型与元数据的关系时发生错误: {str(e)}")
  317. raise
  318. # (从数据模型中选取)
  319. def model_handle_meta_data_model(id_lists, data_model_node_id):
  320. """
  321. 处理从数据模型中选取的数据模型与元数据的关系
  322. Args:
  323. id_lists: ID列表
  324. data_model_node_id: 数据模型节点ID
  325. Returns:
  326. None
  327. """
  328. # 构建meta_id和model_id的列表
  329. model_ids = [record['model_id'] for record in id_lists]
  330. meta_ids = [record['id'] for id_list in id_lists for record in id_list['metaData']]
  331. # 创建与meta_node的关系 组成关系
  332. if meta_ids:
  333. query = """
  334. MATCH (source:DataModel), (target:DataMeta)
  335. WHERE id(source)=$source_id AND id(target) IN $target_ids
  336. MERGE (source)-[:component]->(target)
  337. """
  338. with neo4j_driver.get_session() as session:
  339. session.run(query, source_id=data_model_node_id, target_ids=meta_ids)
  340. # 创建与data_model的关系 模型关系
  341. if model_ids:
  342. query = """
  343. MATCH (source:DataModel), (target:DataModel)
  344. WHERE id(source)=$source_id AND id(target) IN $target_ids
  345. MERGE (source)-[:use]->(target)
  346. """
  347. with neo4j_driver.get_session() as session:
  348. session.run(query, source_id=data_model_node_id, target_ids=model_ids)
  349. # (从DDL中选取)
  350. def handle_no_meta_data_model(id_lists, receiver, data_model_node):
  351. """
  352. 处理从DDL中选取的没有元数据的数据模型
  353. Args:
  354. id_lists: ID列表(可以为空)
  355. receiver: 接收到的请求参数
  356. data_model_node: 数据模型节点
  357. Returns:
  358. None
  359. """
  360. # DDL新增时,id_lists可能为空,提前返回
  361. if not id_lists:
  362. logger.info("id_lists为空,跳过资源关系处理")
  363. return
  364. # 构建meta_id和resouce_id的列表
  365. resouce_ids = [record['resource_id'] for record in id_lists if 'resource_id' in record]
  366. meta_ids = [record['id'] for id_list in id_lists for record in id_list.get('metaData', []) if 'id' in record]
  367. # 获取数据模型节点ID
  368. data_model_node_id = None
  369. if hasattr(data_model_node, 'id'):
  370. # data_model_node 是节点对象
  371. data_model_node_id = data_model_node.id
  372. elif isinstance(data_model_node, int):
  373. # data_model_node 直接就是整数ID
  374. data_model_node_id = data_model_node
  375. elif isinstance(data_model_node, dict):
  376. # data_model_node 是字典,尝试通过name_zh查询
  377. query = """
  378. MATCH (n:DataModel {name_zh: $name_zh})
  379. RETURN id(n) as node_id
  380. """
  381. with connect_graph().session() as session:
  382. result = session.run(query, name_zh=data_model_node.get('name_zh'))
  383. record = result.single()
  384. if record:
  385. data_model_node_id = record["node_id"]
  386. else:
  387. # 未知类型,记录警告
  388. logger.warning(f"data_model_node类型未知: {type(data_model_node)}, 值: {data_model_node}")
  389. if not data_model_node_id:
  390. return
  391. # 创建与DataResource的关系 资源关系
  392. if resouce_ids:
  393. query = """
  394. MATCH (source:DataModel), (target:DataResource)
  395. WHERE id(source)=$source_id AND id(target) IN $target_ids
  396. MERGE (source)-[:resource]->(target)
  397. """
  398. with connect_graph().session() as session:
  399. session.run(query, source_id=data_model_node_id, target_ids=resouce_ids)
  400. if meta_ids:
  401. meta_node_list = []
  402. for id in meta_ids:
  403. query = """
  404. MATCH (n)
  405. WHERE id(n) = $node_id
  406. RETURN n
  407. """
  408. with connect_graph().session() as session:
  409. result = session.run(query, node_id=id)
  410. # 必须在 session 作用域内处理结果
  411. if result:
  412. record = result.data()
  413. if record:
  414. meta_node_list.append(record[0]['n'])
  415. # 提取接收到的数据并创建meta_node节点
  416. meta_node = None
  417. resource_ids = []
  418. for item in id_lists:
  419. resource_id = item.get('resource_id')
  420. if resource_id:
  421. resource_ids.append(resource_id)
  422. for meta_item in item.get('metaData', []):
  423. meta_id = meta_item['id']
  424. data_standard = meta_item.get('data_standard', '')
  425. name_en = meta_item.get('name_en', '')
  426. name_zh = meta_item.get('name_zh', '')
  427. # 使用传递的参数创建meta_node节点
  428. meta_params = {
  429. 'name_zh': name_zh,
  430. 'name_en': name_en,
  431. 'standard': data_standard,
  432. 'create_time': get_formatted_time()
  433. }
  434. # 创建meta_node节点
  435. meta_node = create_or_get_node('DataMeta', **meta_params)
  436. # 获取数据模型节点ID
  437. dm_id = data_model_node_id if data_model_node_id is not None else data_model_node
  438. if meta_node:
  439. # 直接使用Cypher查询检查关系是否存在
  440. with connect_graph().session() as session:
  441. rel_query = """
  442. MATCH (a)-[r:INCLUDES]->(b)
  443. WHERE id(a) = $start_id AND id(b) = $end_id
  444. RETURN count(r) > 0 as exists
  445. """
  446. rel_result = session.run(rel_query,
  447. start_id=int(dm_id),
  448. end_id=int(meta_node)).single()
  449. # 如果关系不存在,则创建INCLUDES关系
  450. if not (rel_result and rel_result["exists"]):
  451. session.execute_write(
  452. lambda tx: tx.run(
  453. "MATCH (a), (b) WHERE id(a) = $a_id AND id(b) = $b_id CREATE (a)-[:INCLUDES]->(b)",
  454. a_id=int(dm_id), b_id=int(meta_node)
  455. )
  456. )
  457. # 数据模型-详情接口
  458. def handle_id_model(id):
  459. """
  460. 获取数据模型详情
  461. Args:
  462. id: 数据模型的节点ID
  463. Returns:
  464. dict: 包含数据模型详情的字典,格式为:
  465. {"data_model": {
  466. "resource_selected": [...],
  467. "leader": ...,
  468. "origin": ...,
  469. "frequency": ...,
  470. "childrenId": [...],
  471. "organization": ...,
  472. "name_zh": ...,
  473. "name_en": ...,
  474. "data_sensitivity": ...,
  475. "describe": ...,
  476. "tag": ...,
  477. "create_time": ...,
  478. "category": ...,
  479. "status": ...
  480. }}
  481. """
  482. node_id = id
  483. cql = """
  484. MATCH (n:DataModel) WHERE id(n) = $nodeId
  485. OPTIONAL MATCH (n)-[:INCLUDES]->(meta:DataMeta)
  486. OPTIONAL MATCH (n)-[:DERIVED_FROM]->(resource:DataResource)
  487. OPTIONAL MATCH (n)-[:LABEL]->(tag:DataLabel)
  488. OPTIONAL MATCH (uses:model_use)-[:use]->(n)
  489. OPTIONAL MATCH (n)-[:has_component]->(component)
  490. WITH n,
  491. collect(DISTINCT meta) as meta_nodes,
  492. collect(DISTINCT resource) as resources,
  493. collect(DISTINCT component) as components,
  494. collect(DISTINCT uses) as uses,
  495. collect(DISTINCT tag) as tags,
  496. CASE WHEN n.childrenId IS NOT NULL THEN n.childrenId ELSE [] END as children
  497. RETURN {
  498. // 基本信息
  499. id: id(n),
  500. name_zh: n.name_zh,
  501. name_en: n.name_en,
  502. create_time: n.create_time,
  503. describe: n.describe,
  504. category: n.category,
  505. level: n.level,
  506. tag: CASE WHEN size(tags) > 0 AND tags[0] IS NOT NULL THEN {id: id(tags[0]), name: tags[0].name} ELSE null END,
  507. // 添加其他必需字段
  508. leader: n.leader,
  509. origin: n.origin,
  510. blood_resource: n.blood_resource,
  511. frequency: n.frequency,
  512. organization: n.organization,
  513. data_sensitivity: n.data_sensitivity,
  514. status: n.status,
  515. // 子节点列表
  516. childrenId: children
  517. } AS result,
  518. // 资源列表
  519. [{
  520. data_resource: [resource IN resources WHERE resource IS NOT NULL | {
  521. id: id(resource),
  522. name_zh: resource.name_zh,
  523. name_en: resource.name_en,
  524. description: resource.description
  525. }],
  526. resource_id: [resource IN resources WHERE resource IS NOT NULL | id(resource)],
  527. meta_ids: [meta IN meta_nodes WHERE meta IS NOT NULL | {
  528. id: id(meta),
  529. name_zh: meta.name_zh,
  530. name_en: meta.name_en,
  531. data_type: meta.data_type
  532. }]
  533. }] AS resource_selected
  534. """
  535. with connect_graph().session() as session:
  536. result = session.run(cql, nodeId=node_id)
  537. # 处理查询结果
  538. record = result.single()
  539. logging.info(f"获得查询结果---------->>>{record}")
  540. if record:
  541. # 获取基本属性和资源选择列表
  542. properties = record["result"]
  543. resource_selected = record["resource_selected"]
  544. # 确保所有必需字段都有默认值,避免空值
  545. required_fields = ['tag', 'leader', 'origin', 'blood_resource',
  546. 'frequency', 'describe', 'organization', 'name_zh', 'name_en',
  547. 'data_sensitivity', 'create_time', 'category', 'status', 'childrenId']
  548. for field in required_fields:
  549. if field not in properties or properties[field] is None:
  550. if field == 'tag':
  551. properties[field] = {}
  552. elif field == 'childrenId':
  553. properties[field] = []
  554. else:
  555. properties[field] = ""
  556. # 构建最终返回格式
  557. final_data = {
  558. "resource_selected": resource_selected,
  559. **properties
  560. }
  561. return {"data_model": final_data}
  562. else:
  563. # 如果没有查询到结果,返回空的结构
  564. return {"data_model": {
  565. "resource_selected": [{"meta_ids": [], "data_resource": None, "resource_id": None}],
  566. "leader": None, "origin": None, "frequency": None, "childrenId": [],
  567. "organization": None, "name_zh": None, "name_en": None, "data_sensitivity": None,
  568. "describe": None, "tag": {}, "create_time": None, "category": None, "status": None
  569. }}
  570. # 数据模型列表
  571. def model_list(skip_count, page_size, name_en_filter=None, name_zh_filter=None,
  572. category=None, tag=None, level=None):
  573. """
  574. 获取数据模型列表
  575. Args:
  576. skip_count: 跳过的数量
  577. page_size: 页面大小
  578. name_en_filter: 英文名称过滤条件
  579. name_zh_filter: 名称过滤条件
  580. category: 类别过滤条件
  581. tag: 标签过滤条件
  582. level: 层级过滤条件
  583. Returns:
  584. tuple: (数据模型列表, 总数量)
  585. """
  586. try:
  587. # 构建where子句 - 只针对DataModel节点的过滤条件
  588. datamodel_where_clause = []
  589. params = {}
  590. if name_zh_filter is not None:
  591. datamodel_where_clause.append("n.name_zh =~ $name_zh")
  592. params['name_zh'] = f".*{name_zh_filter}.*"
  593. if name_en_filter is not None:
  594. datamodel_where_clause.append("n.name_en =~ $name_en")
  595. params['name_en'] = f".*{name_en_filter}.*"
  596. if category is not None:
  597. datamodel_where_clause.append("n.category = $category")
  598. params['category'] = category
  599. if level is not None:
  600. datamodel_where_clause.append("n.level = $level")
  601. params['level'] = level
  602. # 处理标签查询
  603. if tag is not None:
  604. # 确保tag参数是整数类型
  605. try:
  606. tag_id = int(tag)
  607. params['tag'] = tag_id
  608. except (ValueError, TypeError):
  609. logger.warning(f"Invalid tag parameter: {tag}, expected integer")
  610. return [], 0
  611. # 有标签查询条件时,需要确保标签关系存在
  612. match_clause = "MATCH (n:DataModel)-[:LABEL]->(t)"
  613. datamodel_where_clause.append("id(t) = $tag")
  614. else:
  615. # 没有标签查询条件时,先匹配DataModel,然后可选连接标签
  616. match_clause = "MATCH (n:DataModel)"
  617. # 构建DataModel节点的WHERE子句
  618. datamodel_where_str = " AND ".join(datamodel_where_clause)
  619. if datamodel_where_str:
  620. datamodel_where_str = f"WHERE {datamodel_where_str}"
  621. # 构建查询
  622. with connect_graph().session() as session:
  623. # 计算总数量
  624. if tag is not None:
  625. # 有标签查询时,直接使用标签连接
  626. count_query = f"""
  627. {match_clause}
  628. {datamodel_where_str}
  629. RETURN COUNT(DISTINCT n) AS count
  630. """
  631. else:
  632. # 无标签查询时,只计算DataModel节点
  633. count_query = f"""
  634. MATCH (n:DataModel)
  635. {datamodel_where_str}
  636. RETURN COUNT(n) AS count
  637. """
  638. logger.debug(f"Count query: {count_query}")
  639. logger.debug(f"Query parameters: {params}")
  640. count_result = session.run(count_query, **params)
  641. count_record = count_result.single()
  642. total = count_record['count'] if count_record else 0
  643. # 查询数据 - 修复OPTIONAL MATCH的笛卡尔积问题
  644. if tag is not None:
  645. # 有标签查询时,直接使用标签连接
  646. query = f"""
  647. {match_clause}
  648. {datamodel_where_str}
  649. RETURN DISTINCT
  650. id(n) as id,
  651. n.name_zh as name_zh,
  652. n.name_en as name_en,
  653. n.create_time as create_time,
  654. n.describe as describe,
  655. n.level as level,
  656. n.category as category,
  657. n.status as status,
  658. n.leader as leader,
  659. n.origin as origin,
  660. n.blood_resource as blood_resource,
  661. n.organization as organization,
  662. id(t) as tag_id,
  663. t.name_zh as tag_name
  664. ORDER BY time DESC
  665. SKIP $skip
  666. LIMIT $limit
  667. """
  668. else:
  669. # 无标签查询时,先过滤DataModel节点,然后可选连接标签
  670. query = f"""
  671. MATCH (n:DataModel)
  672. {datamodel_where_str}
  673. WITH n
  674. OPTIONAL MATCH (n)-[:LABEL]->(t)
  675. RETURN
  676. id(n) as id,
  677. n.name_zh as name_zh,
  678. n.name_en as name_en,
  679. n.create_time as create_time,
  680. n.describe as describe,
  681. n.level as level,
  682. n.category as category,
  683. n.status as status,
  684. n.leader as leader,
  685. n.origin as origin,
  686. n.blood_resource as blood_resource,
  687. n.organization as organization,
  688. id(t) as tag_id,
  689. t.name_zh as tag_name
  690. ORDER BY n.create_time DESC
  691. SKIP $skip
  692. LIMIT $limit
  693. """
  694. logger.debug(f"Main query: {query}")
  695. result = session.run(query, skip=skip_count, limit=page_size, **params)
  696. # 处理结果
  697. data = []
  698. for record in result:
  699. item = {
  700. "id": record['id'],
  701. "name_zh": record['name_zh'],
  702. "name_en": record['name_en'],
  703. "create_time": record['create_time'],
  704. "describe": record['describe'],
  705. "category": record['category'],
  706. "status": record['status'],
  707. "leader": record['leader'],
  708. "origin": record['origin'],
  709. "blood_resource": record['blood_resource'],
  710. "organization": record['organization'],
  711. "level": record['level'],
  712. "tag": {"id": record['tag_id'], "name_zh": record['tag_name']} if record['tag_id'] is not None else None
  713. }
  714. data.append(item)
  715. logger.info(f"Query returned {len(data)} items out of {total} total")
  716. return data, total
  717. except Exception as e:
  718. logger.error(f"Error in model_list: {str(e)}")
  719. import traceback
  720. traceback.print_exc()
  721. return [], 0
  722. # 有血缘关系的数据资源列表
  723. def model_resource_list(skip_count, page_size, name_zh_filter=None, id=None,
  724. category=None, create_time=None):
  725. """
  726. 获取数据模型相关的数据资源列表
  727. Args:
  728. skip_count: 跳过的数量
  729. page_size: 页面大小
  730. name_zh_filter: 名称过滤条件
  731. id: 数据模型ID
  732. category: 类别过滤条件
  733. create_time: 时间过滤条件
  734. Returns:
  735. tuple: (数据资源列表, 总数量)
  736. """
  737. try:
  738. # 构建基础查询
  739. base_query = """
  740. MATCH (n:DataModel)
  741. WHERE id(n) = $nodeId
  742. MATCH (n)-[:children]->(m:DataResource)
  743. """
  744. # 计算总数量
  745. count_query = base_query + """
  746. RETURN COUNT(m) as count
  747. """
  748. with connect_graph().session() as session:
  749. # 执行计数查询
  750. count_result = session.run(count_query, nodeId=id)
  751. count_record = count_result.single()
  752. total = count_record['count'] if count_record else 0
  753. # 使用分页和筛选条件构建主查询
  754. main_query = base_query + """
  755. MATCH (m)-[:LABEL]->(l)
  756. WHERE id(n) = $nodeId and labels(m) <> ['DataMeta']
  757. RETURN m.name_zh as name_zh,
  758. m.name_en as name_en,
  759. id(m) as id,
  760. l.name_zh as label,
  761. m.create_time as create_time,
  762. m.description as description,
  763. m.category as category
  764. ORDER BY m.create_time DESC
  765. SKIP $skip LIMIT $limit
  766. """
  767. # 执行主查询
  768. result = session.run(main_query, nodeId=id, skip=skip_count, limit=page_size)
  769. # 处理结果
  770. data = []
  771. for record in result:
  772. item = {
  773. "name_zh": record['name_zh'],
  774. "name_en": record['name_en'],
  775. "id": record['id'],
  776. "label": record['label'],
  777. "create_time": record['create_time'],
  778. "description": record['description'],
  779. "category": record['category']
  780. }
  781. data.append(item)
  782. return data, total
  783. except Exception as e:
  784. print(f"Error in model_resource_list: {str(e)}")
  785. import traceback
  786. traceback.print_exc()
  787. return [], 0
  788. # 数据模型血缘图谱
  789. def model_kinship_graph(nodeid, meta=False):
  790. """
  791. 生成数据模型的血缘关系图谱
  792. 按照DERIVED_FROM关系进行递归查找,从当前节点作为起点查找所有DERIVED_FROM关系指向的节点
  793. Args:
  794. nodeid: 节点ID
  795. meta: 是否包含元数据
  796. Returns:
  797. dict: 包含节点和连线信息的图谱数据
  798. """
  799. try:
  800. with connect_graph().session() as session:
  801. # 确保nodeid为整数
  802. try:
  803. nodeid_int = int(nodeid)
  804. except (ValueError, TypeError):
  805. logger.error(f"节点ID不是有效的整数: {nodeid}")
  806. return {"nodes": [], "lines": []}
  807. # 查询起始模型节点是否存在
  808. start_node_query = """
  809. MATCH (n:DataModel)
  810. WHERE id(n) = $nodeId
  811. RETURN n
  812. """
  813. start_result = session.run(start_node_query, nodeId=nodeid_int)
  814. start_record = start_result.single()
  815. if not start_record:
  816. logger.error(f"未找到ID为{nodeid_int}的DataModel节点")
  817. return {"nodes": [], "lines": []}
  818. # 递归查找DERIVED_FROM关系
  819. cypher = """
  820. MATCH (start:DataModel)
  821. WHERE id(start) = $nodeId
  822. MATCH path = (start)-[:DERIVED_FROM*0..]->(target)
  823. WHERE target:DataResource OR target:DataModel
  824. RETURN path
  825. """
  826. result = session.run(cypher, nodeId=nodeid_int)
  827. # 收集节点和关系
  828. nodes = {}
  829. lines = {}
  830. for record in result:
  831. # 处理路径
  832. path = record['path']
  833. logger.debug(f"处理路径,长度: {len(path)}, 节点数: {len(path.nodes)}, 关系数: {len(path.relationships)}")
  834. # 处理路径中的所有节点
  835. for node in path.nodes:
  836. node_id = int(node.id) # 直接转换为整数
  837. if node_id not in nodes:
  838. node_dict = serialize_node_properties(node)
  839. node_dict["id"] = str(node_id)
  840. node_dict["node_type"] = list(node.labels)[0] if node.labels else ""
  841. nodes[node_id] = node_dict
  842. logger.debug(f"添加节点: ID={node_id}, 标签={list(node.labels)}")
  843. # 处理路径中的所有关系
  844. for rel in path.relationships:
  845. rel_id = int(rel.id) # 直接转换为整数
  846. if rel_id not in lines:
  847. rel_dict = {
  848. "id": str(rel_id),
  849. "from": str(int(rel.start_node.id)),
  850. "to": str(int(rel.end_node.id)),
  851. "text": rel.type
  852. }
  853. lines[rel_id] = rel_dict
  854. logger.debug(f"添加关系: ID={rel_id}, 类型={rel.type}, 从{int(rel.start_node.id)}到{int(rel.end_node.id)}")
  855. # 如果需要元数据,查询INCLUDES关系
  856. if meta:
  857. meta_cypher = """
  858. MATCH (start:DataModel)-[r:INCLUDES]->(meta:DataMeta)
  859. WHERE id(start) = $nodeId
  860. RETURN start, r, meta
  861. """
  862. meta_result = session.run(meta_cypher, nodeId=nodeid_int)
  863. for meta_record in meta_result:
  864. start_node = meta_record['start']
  865. rel = meta_record['r']
  866. meta_node = meta_record['meta']
  867. # 添加元数据节点
  868. meta_node_id = int(meta_node.id)
  869. if meta_node_id not in nodes:
  870. node_dict = serialize_node_properties(meta_node)
  871. node_dict["id"] = str(meta_node_id)
  872. node_dict["node_type"] = list(meta_node.labels)[0] if meta_node.labels else ""
  873. nodes[meta_node_id] = node_dict
  874. # 添加INCLUDES关系
  875. rel_id = int(rel.id)
  876. if rel_id not in lines:
  877. rel_dict = {
  878. "id": str(rel_id),
  879. "from": str(nodeid_int),
  880. "to": str(meta_node_id),
  881. "text": rel.type
  882. }
  883. lines[rel_id] = rel_dict
  884. logger.info(f"成功获取血缘关系图谱,ID: {nodeid_int}, 节点数: {len(nodes)}, 关系数: {len(lines)}")
  885. return {
  886. "nodes": list(nodes.values()),
  887. "lines": list(lines.values())
  888. }
  889. except Exception as e:
  890. logger.error(f"获取数据模型血缘关系图谱失败: {str(e)}")
  891. import traceback
  892. logger.error(f"错误详情: {traceback.format_exc()}")
  893. return {"nodes": [], "lines": []}
  894. # 数据模型影响图谱
  895. def model_impact_graph(nodeid, meta=False):
  896. """
  897. 生成数据模型的影响关系图谱
  898. 按照DERIVED_FROM关系进行递归查找,从当前节点作为终点查找所有指向这个终点的节点
  899. Args:
  900. nodeid: 节点ID
  901. meta: 是否包含元数据
  902. Returns:
  903. dict: 包含节点和连线信息的图谱数据
  904. """
  905. try:
  906. with connect_graph().session() as session:
  907. # 确保nodeid为整数
  908. try:
  909. nodeid_int = int(nodeid)
  910. except (ValueError, TypeError):
  911. logger.error(f"节点ID不是有效的整数: {nodeid}")
  912. return {"nodes": [], "lines": []}
  913. # 查询起始模型节点是否存在
  914. start_node_query = """
  915. MATCH (n:DataModel)
  916. WHERE id(n) = $nodeId
  917. RETURN n
  918. """
  919. start_result = session.run(start_node_query, nodeId=nodeid_int)
  920. start_record = start_result.single()
  921. if not start_record:
  922. logger.error(f"未找到ID为{nodeid_int}的DataModel节点")
  923. return {"nodes": [], "lines": []}
  924. # 递归查找指向当前节点的DERIVED_FROM关系
  925. cypher = """
  926. MATCH (target:DataModel)
  927. WHERE id(target) = $nodeId
  928. MATCH path = (source)-[:DERIVED_FROM*0..]->(target)
  929. WHERE source:DataResource OR source:DataModel
  930. RETURN path
  931. """
  932. result = session.run(cypher, nodeId=nodeid_int)
  933. # 收集节点和关系
  934. nodes = {}
  935. lines = {}
  936. for record in result:
  937. # 处理路径
  938. path = record['path']
  939. logger.debug(f"处理影响路径,长度: {len(path)}, 节点数: {len(path.nodes)}, 关系数: {len(path.relationships)}")
  940. # 处理路径中的所有节点
  941. for node in path.nodes:
  942. node_id = int(node.id) # 直接转换为整数
  943. if node_id not in nodes:
  944. node_dict = serialize_node_properties(node)
  945. node_dict["id"] = str(node_id)
  946. node_dict["node_type"] = list(node.labels)[0] if node.labels else ""
  947. nodes[node_id] = node_dict
  948. logger.debug(f"添加影响节点: ID={node_id}, 标签={list(node.labels)}")
  949. # 处理路径中的所有关系
  950. for rel in path.relationships:
  951. rel_id = int(rel.id) # 直接转换为整数
  952. if rel_id not in lines:
  953. rel_dict = {
  954. "id": str(rel_id),
  955. "from": str(int(rel.start_node.id)),
  956. "to": str(int(rel.end_node.id)),
  957. "text": rel.type
  958. }
  959. lines[rel_id] = rel_dict
  960. logger.debug(f"添加影响关系: ID={rel_id}, 类型={rel.type}, 从{int(rel.start_node.id)}到{int(rel.end_node.id)}")
  961. # 如果需要元数据,查询INCLUDES关系
  962. if meta:
  963. meta_cypher = """
  964. MATCH (target:DataModel)-[r:INCLUDES]->(meta:DataMeta)
  965. WHERE id(target) = $nodeId
  966. RETURN target, r, meta
  967. """
  968. meta_result = session.run(meta_cypher, nodeId=nodeid_int)
  969. for meta_record in meta_result:
  970. target_node = meta_record['target']
  971. rel = meta_record['r']
  972. meta_node = meta_record['meta']
  973. # 添加元数据节点
  974. meta_node_id = int(meta_node.id)
  975. if meta_node_id not in nodes:
  976. node_dict = serialize_node_properties(meta_node)
  977. node_dict["id"] = str(meta_node_id)
  978. node_dict["node_type"] = list(meta_node.labels)[0] if meta_node.labels else ""
  979. nodes[meta_node_id] = node_dict
  980. # 添加INCLUDES关系
  981. rel_id = int(rel.id)
  982. if rel_id not in lines:
  983. rel_dict = {
  984. "id": str(rel_id),
  985. "from": str(nodeid_int),
  986. "to": str(meta_node_id),
  987. "text": rel.type
  988. }
  989. lines[rel_id] = rel_dict
  990. logger.info(f"成功获取影响关系图谱,ID: {nodeid_int}, 节点数: {len(nodes)}, 关系数: {len(lines)}")
  991. return {
  992. "nodes": list(nodes.values()),
  993. "lines": list(lines.values())
  994. }
  995. except Exception as e:
  996. logger.error(f"获取数据模型影响关系图谱失败: {str(e)}")
  997. import traceback
  998. logger.error(f"错误详情: {traceback.format_exc()}")
  999. return {"nodes": [], "lines": []}
  1000. # 数据模型全部图谱
  1001. def model_all_graph(nodeid, meta=False):
  1002. """
  1003. 生成数据模型的所有关系图谱
  1004. 分别调用model_impact_graph查找影响关系,调用model_kinship_graph查找血缘关系,
  1005. 然后合并两部分数据返回
  1006. Args:
  1007. nodeid: 节点ID
  1008. meta: 是否包含元数据
  1009. Returns:
  1010. dict: 包含节点和连线信息的图谱数据
  1011. """
  1012. try:
  1013. # 获取血缘关系图谱
  1014. kinship_data = model_kinship_graph(nodeid, meta)
  1015. # 获取影响关系图谱
  1016. impact_data = model_impact_graph(nodeid, meta)
  1017. # 合并节点数据,使用字典去重
  1018. merged_nodes = {}
  1019. merged_lines = {}
  1020. # 添加血缘关系的节点和连线
  1021. if kinship_data and 'nodes' in kinship_data:
  1022. for node in kinship_data['nodes']:
  1023. node_id = node.get('id')
  1024. if node_id:
  1025. merged_nodes[node_id] = node
  1026. if kinship_data and 'lines' in kinship_data:
  1027. for line in kinship_data['lines']:
  1028. line_id = line.get('id')
  1029. if line_id:
  1030. merged_lines[line_id] = line
  1031. # 添加影响关系的节点和连线
  1032. if impact_data and 'nodes' in impact_data:
  1033. for node in impact_data['nodes']:
  1034. node_id = node.get('id')
  1035. if node_id:
  1036. merged_nodes[node_id] = node
  1037. if impact_data and 'lines' in impact_data:
  1038. for line in impact_data['lines']:
  1039. line_id = line.get('id')
  1040. if line_id:
  1041. merged_lines[line_id] = line
  1042. # 构建最终结果
  1043. result = {
  1044. "nodes": list(merged_nodes.values()),
  1045. "lines": list(merged_lines.values())
  1046. }
  1047. logger.info(f"成功获取完整关系图谱,ID: {nodeid}, 节点数: {len(merged_nodes)}, 关系数: {len(merged_lines)}")
  1048. return result
  1049. except Exception as e:
  1050. logger.error(f"获取数据模型完整关系图谱失败: {str(e)}")
  1051. return {"nodes": [], "lines": []}
  1052. # 更新数据模型
  1053. def data_model_edit(receiver):
  1054. """
  1055. 更新数据模型
  1056. Args:
  1057. receiver: 接收到的请求参数
  1058. Returns:
  1059. 更新结果
  1060. """
  1061. id = receiver.get('id')
  1062. name = receiver.get('name_zh')
  1063. name_en = receiver.get('name_en')
  1064. category = receiver.get('category')
  1065. describe = receiver.get('describe')
  1066. tag = receiver.get('tag')
  1067. frequency = receiver.get('frequency')
  1068. leader = receiver.get('leader')
  1069. organization = receiver.get('organization')
  1070. status = bool(receiver.get('status')) if receiver.get('status') is not None else None
  1071. meta_data = receiver.get('metaData', [])
  1072. # 更新数据模型节点 - 添加新的字段
  1073. query = """
  1074. MATCH (n:DataModel) WHERE id(n) = $id
  1075. SET n.name_zh = $name_zh,
  1076. n.name_en = $name_en,
  1077. n.category = $category,
  1078. n.describe = $describe,
  1079. n.frequency = $frequency,
  1080. n.leader = $leader,
  1081. n.organization = $organization,
  1082. n.status = $status,
  1083. n.create_time = $create_time
  1084. RETURN n
  1085. """
  1086. create_time = get_formatted_time()
  1087. with connect_graph().session() as session:
  1088. result = session.run(query,
  1089. id=id,
  1090. name_zh=name,
  1091. name_en=name_en,
  1092. category=category,
  1093. describe=describe,
  1094. frequency=frequency,
  1095. leader=leader,
  1096. organization=organization,
  1097. status=status,
  1098. create_time=create_time).data()
  1099. # 处理标签关系
  1100. if tag:
  1101. # 先删除所有标签关系
  1102. delete_query = """
  1103. MATCH (n:DataModel)-[r:LABEL]->() WHERE id(n) = $id
  1104. DELETE r
  1105. """
  1106. with connect_graph().session() as session:
  1107. session.run(delete_query, id=id)
  1108. # 再创建新的标签关系
  1109. tag_node = get_node_by_id('DataLabel', tag)
  1110. if tag_node:
  1111. model_node = get_node_by_id_no_label(id)
  1112. if model_node:
  1113. # 获取节点ID
  1114. model_id = model_node.id if hasattr(model_node, 'id') else model_node
  1115. tag_id = tag_node.id if hasattr(tag_node, 'id') else tag_node
  1116. # 直接使用Cypher查询检查关系是否存在
  1117. with connect_graph().session() as session:
  1118. rel_query = """
  1119. MATCH (a)-[r:LABEL]->(b)
  1120. WHERE id(a) = $start_id AND id(b) = $end_id
  1121. RETURN count(r) > 0 as exists
  1122. """
  1123. rel_result = session.run(rel_query,
  1124. start_id=int(model_id),
  1125. end_id=int(tag_id)).single()
  1126. # 如果关系不存在,则创建关系
  1127. if not (rel_result and rel_result["exists"]):
  1128. session.execute_write(
  1129. lambda tx: tx.run(
  1130. "MATCH (a), (b) WHERE id(a) = $a_id AND id(b) = $b_id CREATE (a)-[:LABEL]->(b)",
  1131. a_id=int(model_id), b_id=int(tag_id)
  1132. )
  1133. )
  1134. # 处理DataMeta节点关系更新
  1135. with connect_graph().session() as session:
  1136. # 先删除DataModel关联的所有DataMeta关系
  1137. delete_meta_query = """
  1138. MATCH (n:DataModel)-[r:INCLUDES]->(m:DataMeta)
  1139. WHERE id(n) = $id
  1140. DELETE r
  1141. """
  1142. session.run(delete_meta_query, id=id)
  1143. logger.info(f"已删除DataModel({id})的所有DataMeta关系")
  1144. # 根据上传的metaData数据是否有值来决定是否重新构建INCLUDES关系
  1145. if meta_data:
  1146. # 根据上传的metaData数据重新构建INCLUDES关系
  1147. for meta_item in meta_data:
  1148. meta_id = meta_item.get('id')
  1149. if meta_id:
  1150. try:
  1151. meta_id = int(meta_id)
  1152. # 验证DataMeta节点是否存在
  1153. check_meta_query = """
  1154. MATCH (m:DataMeta)
  1155. WHERE id(m) = $meta_id
  1156. RETURN m
  1157. """
  1158. meta_result = session.run(check_meta_query, meta_id=meta_id)
  1159. if meta_result.single():
  1160. # 创建INCLUDES关系
  1161. create_includes_query = """
  1162. MATCH (n:DataModel), (m:DataMeta)
  1163. WHERE id(n) = $model_id AND id(m) = $meta_id
  1164. CREATE (n)-[:INCLUDES]->(m)
  1165. RETURN n, m
  1166. """
  1167. session.run(create_includes_query, model_id=id, meta_id=meta_id)
  1168. logger.info(f"成功创建INCLUDES关系: DataModel({id}) -> DataMeta({meta_id})")
  1169. else:
  1170. logger.warning(f"DataMeta节点不存在,ID: {meta_id}")
  1171. except (ValueError, TypeError) as e:
  1172. logger.error(f"无效的meta_id: {meta_id}, 错误: {str(e)}")
  1173. else:
  1174. logger.info(f"meta_data为空,不需要重新创建INCLUDES关系,DataModel({id})将不关联任何DataMeta节点")
  1175. return {"message": "数据模型更新成功"}
  1176. def model_community(tag=None):
  1177. """
  1178. 查询DataModel的所有节点及DERIVED_FROM关系
  1179. Args:
  1180. tag: 可选的标签ID,如果指定则只查找有该标签的DataModel节点
  1181. Returns:
  1182. dict: 包含节点和连线信息的图谱数据,格式与model_kinship_graph相同
  1183. """
  1184. try:
  1185. with connect_graph().session() as session:
  1186. # 构建查询条件
  1187. if tag is not None:
  1188. # 确保tag参数是整数类型
  1189. try:
  1190. tag_id = int(tag)
  1191. except (ValueError, TypeError):
  1192. logger.warning(f"Invalid tag parameter: {tag}, expected integer")
  1193. return {"nodes": [], "lines": []}
  1194. # 有标签查询条件时,查询有指定标签的DataModel节点及其DERIVED_FROM关系
  1195. cypher = """
  1196. MATCH (dm:DataModel)-[:LABEL]->(t)
  1197. WHERE id(t) = $tag_id
  1198. WITH dm
  1199. MATCH path = (dm)-[:DERIVED_FROM*0..]->(target:DataModel)
  1200. RETURN path
  1201. UNION
  1202. MATCH (dm:DataModel)-[:LABEL]->(t)
  1203. WHERE id(t) = $tag_id
  1204. WITH dm
  1205. MATCH path = (source:DataModel)-[:DERIVED_FROM*0..]->(dm)
  1206. RETURN path
  1207. """
  1208. result = session.run(cypher, tag_id=tag_id)
  1209. else:
  1210. # 没有标签查询条件时,查询所有DataModel节点及其DERIVED_FROM关系
  1211. cypher = """
  1212. MATCH (dm:DataModel)
  1213. WITH dm
  1214. MATCH path = (dm)-[:DERIVED_FROM*0..]->(target:DataModel)
  1215. RETURN path
  1216. UNION
  1217. MATCH (dm:DataModel)
  1218. WITH dm
  1219. MATCH path = (source:DataModel)-[:DERIVED_FROM*0..]->(dm)
  1220. RETURN path
  1221. """
  1222. result = session.run(cypher)
  1223. # 收集节点和关系
  1224. nodes = {}
  1225. lines = {}
  1226. for record in result:
  1227. # 处理路径
  1228. path = record['path']
  1229. logger.debug(f"处理社区路径,长度: {len(path)}, 节点数: {len(path.nodes)}, 关系数: {len(path.relationships)}")
  1230. # 处理路径中的所有节点
  1231. for node in path.nodes:
  1232. node_id = int(node.id) # 直接转换为整数
  1233. if node_id not in nodes:
  1234. node_dict = serialize_node_properties(node)
  1235. node_dict["id"] = str(node_id)
  1236. node_dict["node_type"] = list(node.labels)[0] if node.labels else ""
  1237. nodes[node_id] = node_dict
  1238. logger.debug(f"添加社区节点: ID={node_id}, 标签={list(node.labels)}")
  1239. # 处理路径中的所有关系
  1240. for rel in path.relationships:
  1241. rel_id = int(rel.id) # 直接转换为整数
  1242. if rel_id not in lines:
  1243. rel_dict = {
  1244. "id": str(rel_id),
  1245. "from": str(int(rel.start_node.id)),
  1246. "to": str(int(rel.end_node.id)),
  1247. "text": rel.type
  1248. }
  1249. lines[rel_id] = rel_dict
  1250. logger.debug(f"添加社区关系: ID={rel_id}, 类型={rel.type}, 从{int(rel.start_node.id)}到{int(rel.end_node.id)}")
  1251. logger.info(f"成功获取数据模型社区图谱,标签ID: {tag}, 节点数: {len(nodes)}, 关系数: {len(lines)}")
  1252. return {
  1253. "nodes": list(nodes.values()),
  1254. "lines": list(lines.values())
  1255. }
  1256. except Exception as e:
  1257. logger.error(f"获取数据模型社区图谱失败: {str(e)}")
  1258. import traceback
  1259. logger.error(f"错误详情: {traceback.format_exc()}")
  1260. return {"nodes": [], "lines": []}
  1261. def model_search_list(model_id, page, page_size, name_en_filter=None,
  1262. name_zh_filter=None, category_filter=None, tag_filter=None):
  1263. """获取特定数据模型关联的元数据列表"""
  1264. try:
  1265. with connect_graph().session() as session:
  1266. # 确保model_id为整数
  1267. try:
  1268. model_id_int = int(model_id)
  1269. except (ValueError, TypeError):
  1270. logger.error(f"模型ID不是有效的整数: {model_id}")
  1271. return [], 0
  1272. # 基本匹配语句 - 支持DataMeta和Metadata标签
  1273. match_clause = """
  1274. MATCH (n:DataModel)-[:INCLUDES]->(m)
  1275. WHERE id(n) = $model_id
  1276. AND (m:DataMeta OR m:Metadata)
  1277. """
  1278. where_conditions = []
  1279. if name_en_filter:
  1280. where_conditions.append(f"m.name_en CONTAINS '{name_en_filter}'")
  1281. if name_zh_filter:
  1282. where_conditions.append(f"m.name_zh CONTAINS '{name_zh_filter}'")
  1283. if category_filter:
  1284. where_conditions.append(f"m.category = '{category_filter}'")
  1285. # 标签过滤需要额外的匹配
  1286. tag_match = ""
  1287. if tag_filter:
  1288. tag_match = "MATCH (m)-[:HAS_TAG]->(t:Tag) WHERE t.name_zh = $tag_filter"
  1289. where_clause = " AND " + " AND ".join(where_conditions) if where_conditions else ""
  1290. # 计算总数
  1291. count_cypher = f"""
  1292. {match_clause}{where_clause}
  1293. {tag_match}
  1294. RETURN count(m) as count
  1295. """
  1296. count_params = {"model_id": model_id_int}
  1297. if tag_filter:
  1298. count_params["tag_filter"] = tag_filter
  1299. count_result = session.run(count_cypher, **count_params)
  1300. total_count = count_result.single()["count"]
  1301. # 分页查询
  1302. skip = (page - 1) * page_size
  1303. cypher = f"""
  1304. {match_clause}{where_clause}
  1305. {tag_match}
  1306. RETURN m
  1307. ORDER BY m.name_zh
  1308. SKIP {skip} LIMIT {page_size}
  1309. """
  1310. result = session.run(cypher, **count_params)
  1311. # 格式化结果
  1312. metadata_list = []
  1313. for record in result:
  1314. meta = serialize_node_properties(record["m"])
  1315. meta["id"] = record["m"].id
  1316. metadata_list.append(meta)
  1317. logger.info(f"成功获取数据模型关联元数据,ID: {model_id_int}, 元数据数量: {total_count}")
  1318. return metadata_list, total_count
  1319. except Exception as e:
  1320. logger.error(f"获取数据模型关联的元数据列表失败: {str(e)}")
  1321. return [], 0