ruxia
958 lines
31 KiB
Python
958 lines
31 KiB
Python
from neo4j import GraphDatabase
|
|
from project import *
|
|
import atexit
|
|
|
|
|
|
class ProjectTookiItNeo4j(ProjectTookiIt):
|
|
"""
|
|
基于Neo4j数据库的项目类实现
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""
|
|
初始化Neo4j连接
|
|
|
|
Args:
|
|
uri (str): Neo4j数据库URI
|
|
user (str): 用户名
|
|
password (str): 密码
|
|
"""
|
|
uri = "bolt://172.20.0.145:7687"
|
|
user = "neo4j"
|
|
password = "password"
|
|
|
|
super().__init__()
|
|
self.driver = GraphDatabase.driver(uri, auth=(user, password))
|
|
self.session = self.driver.session()
|
|
|
|
# 初始化其他必要的数据结构
|
|
self.material_equipment_dict = {} # 材机字典，键为ID
|
|
self.fee_templates = {} # 取费表模板字典，键为ID
|
|
self.fee_schedules = {} # 费用表字典，键为ID
|
|
self.project_properties = {} # 工程属性字典
|
|
|
|
def close(self):
|
|
"""
|
|
关闭数据库连接
|
|
"""
|
|
if self.session:
|
|
self.session.close()
|
|
if self.driver:
|
|
self.driver.close()
|
|
|
|
# 通用节点查询方法
|
|
def get_node_by_path(self, path, node_labels=None):
|
|
"""
|
|
通过路径获取节点对象
|
|
|
|
Args:
|
|
path (str): 以'/'分隔的多级节点路径
|
|
node_labels (list): 节点标签列表，用于过滤结果
|
|
|
|
Returns:
|
|
dict|None: 节点数据，如果路径不存在返回None
|
|
"""
|
|
if not path:
|
|
return None
|
|
|
|
# 分割路径为各个部分
|
|
path_parts = path.split("/")
|
|
|
|
# 构建查询
|
|
if len(path_parts) == 1:
|
|
# 只有一级路径，直接查询
|
|
if node_labels:
|
|
labels_str = ":" + "|:".join(node_labels)
|
|
query = f"""
|
|
MATCH (n{labels_str})
|
|
WHERE n.name = $name
|
|
RETURN n LIMIT 1
|
|
"""
|
|
else:
|
|
query = """
|
|
MATCH (n)
|
|
WHERE n.name = $name
|
|
RETURN n LIMIT 1
|
|
"""
|
|
params = {"name": path_parts[0]}
|
|
else:
|
|
# 多级路径，构建路径查询
|
|
last_part = path_parts[-1]
|
|
if node_labels:
|
|
labels_str = ":" + "|:".join(node_labels)
|
|
query = f"""
|
|
MATCH path = (root)-[*]->(target{labels_str})
|
|
WHERE target.name = $last_part
|
|
RETURN target as n LIMIT 1
|
|
"""
|
|
else:
|
|
query = """
|
|
MATCH path = (root)-[*]->(target)
|
|
WHERE target.name = $last_part
|
|
RETURN target as n LIMIT 1
|
|
"""
|
|
params = {"last_part": last_part}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
record = result.single()
|
|
|
|
if not record:
|
|
return None
|
|
|
|
return record["n"]
|
|
except Exception as e:
|
|
print(f"获取节点对象时出错: {e}")
|
|
return None
|
|
|
|
# 项目划分查询方法
|
|
def get_division_item_by_path(self, path):
|
|
"""
|
|
通过路径获取项目划分对象
|
|
"""
|
|
node_data = self.get_node_by_path(path, ["ProjectDivisionItem"])
|
|
if not node_data:
|
|
return None
|
|
|
|
item = ProjectDivisionItem()
|
|
for key, value in node_data.items():
|
|
if hasattr(item, key):
|
|
setattr(item, key, value)
|
|
|
|
return item
|
|
|
|
def get_division_node_by_parent_and_name(self, parent_path, partial_name):
|
|
"""
|
|
通过父节点路径和模糊节点名称获取项目划分对象，包括子节点
|
|
|
|
Args:
|
|
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
|
partial_name (str): 目标节点的模糊或不完整名称
|
|
|
|
Returns:
|
|
list: 包含所有匹配项目划分节点的列表，如果没有匹配返回空列表
|
|
"""
|
|
if not partial_name:
|
|
return []
|
|
|
|
# 使用通用方法获取父节点
|
|
parent_node_data = self.get_node_by_path(parent_path, ["ProjectDivisionItem"])
|
|
parent_node_id = parent_node_data["id"] if parent_node_data and "id" in parent_node_data else None
|
|
|
|
# 构建查询，根据是否有父节点ID调整查询条件，使用递归关系查询
|
|
if parent_node_id:
|
|
query = """
|
|
MATCH (p)-[:CONTAINS|HAS|RELATED_TO*]-(n:ProjectDivisionItem)
|
|
WHERE p.id = $parent_id AND n.name CONTAINS $partial_name
|
|
RETURN n LIMIT 50
|
|
"""
|
|
params = {"parent_id": parent_node_id, "partial_name": partial_name}
|
|
else:
|
|
query = """
|
|
MATCH (n:ProjectDivisionItem)
|
|
WHERE n.name CONTAINS $partial_name
|
|
RETURN n LIMIT 50
|
|
"""
|
|
params = {"partial_name": partial_name}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
items = []
|
|
for record in result:
|
|
node_data = record["n"]
|
|
item = ProjectDivisionItem()
|
|
for key, value in node_data.items():
|
|
if hasattr(item, key):
|
|
setattr(item, key, value)
|
|
items.append(item)
|
|
|
|
return items
|
|
except Exception as e:
|
|
print(f"通过父节点路径和模糊名称获取项目划分对象时出错: {e}")
|
|
return []
|
|
|
|
# 工程量查询方法
|
|
def get_quantities_by_paths(self, paths_str):
|
|
"""
|
|
获取指定项目路径下的工程量对象
|
|
|
|
Args:
|
|
paths_str (str): 以'/'分隔的多级节点路径
|
|
|
|
Returns:
|
|
ProjectQuantity|None: 对应的工程量对象，如果路径不存在返回None，成功找到返回工程量对象
|
|
"""
|
|
if not paths_str:
|
|
return None
|
|
|
|
# 使用通用方法获取节点，考虑所有可能的工程量类型
|
|
node_data = self.get_node_by_path(paths_str, ["ProjectQuantity", "Quota", "MainMaterial", "Equipment"])
|
|
|
|
if not node_data:
|
|
return None
|
|
|
|
# 根据节点标签或类型属性创建对应类型的对象
|
|
quantity = self._create_quantity_object(node_data)
|
|
|
|
# 填充属性
|
|
for key, value in node_data.items():
|
|
if hasattr(quantity, key):
|
|
setattr(quantity, key, value)
|
|
|
|
return quantity
|
|
|
|
def get_quantities_node_by_parent_and_code(self, parent_path, quantity_type=None, code=None):
|
|
"""
|
|
通过父节点路径和编码获取工程量对象（定额），包括子节点
|
|
|
|
Args:
|
|
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
|
quantity_type (str): 工程量类型('定额'、'主材'、'设备'或None表示所有类型)
|
|
code(str): 工程量编码，以'/'分隔的多个编码
|
|
|
|
Returns:
|
|
list: 包含所有匹配节点的列表，如果没有匹配返回空列表
|
|
"""
|
|
if not code:
|
|
return []
|
|
|
|
# 使用通用方法获取父节点
|
|
parent_node_data = self.get_node_by_path(parent_path)
|
|
|
|
# 从路径中获取父节点名称
|
|
path_parts = parent_path.split("/")
|
|
parent_name = path_parts[-1]
|
|
|
|
# 处理编码，可能有多个编码用/分隔
|
|
code_parts = code.split("/")
|
|
code_conditions = []
|
|
for code_part in code_parts:
|
|
if code_part:
|
|
|
|
code_conditions.append(f"q.编码 = '{code_part}'")
|
|
|
|
if not code_conditions:
|
|
return []
|
|
|
|
code_query = " OR ".join(code_conditions)
|
|
|
|
# 根据工程量类型确定标签
|
|
node_labels = []
|
|
if quantity_type == "定额":
|
|
node_labels = ["ProjectQuantity", "Quota"]
|
|
elif quantity_type == "主材":
|
|
node_labels = ["ProjectQuantity", "MainMaterial"]
|
|
elif quantity_type == "设备":
|
|
node_labels = ["ProjectQuantity", "Equipment"]
|
|
else:
|
|
node_labels = ["ProjectQuantity"]
|
|
|
|
# 构建标签字符串
|
|
labels_str = ":" + ":".join(node_labels) if node_labels else ""
|
|
|
|
# 使用name属性进行匹配
|
|
query = f"""
|
|
MATCH (p)-[*1..5]->(q{labels_str})
|
|
WHERE p.name = $parent_name AND ({code_query})
|
|
RETURN q
|
|
LIMIT 10
|
|
"""
|
|
params = {"parent_name": parent_name}
|
|
|
|
try:
|
|
result = self.session.run(query, params)
|
|
quantities = []
|
|
|
|
for record in result:
|
|
node_data = record["q"]
|
|
quantity = self._create_quantity_object(node_data, quantity_type)
|
|
|
|
# 将节点属性赋值到对象
|
|
for key, value in node_data.items():
|
|
setattr(quantity, key, value)
|
|
|
|
# 转换为字典
|
|
if hasattr(quantity, "to_dict"):
|
|
quantities.append(quantity.to_dict())
|
|
else:
|
|
# 如果没有 to_dict 方法，就用 vars() 动态获取属性
|
|
quantities.append(vars(quantity))
|
|
|
|
return quantities
|
|
except Exception as e:
|
|
print(f"通过编码获取工程量对象时出错: {e}")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return []
|
|
|
|
def get_quantities_node_by_parent_and_name(self, parent_path, partial_name, quantity_type=None):
|
|
"""
|
|
通过父节点路径、模糊节点名称和类型获取工程量对象（主材或者设备），包括子节点
|
|
|
|
Args:
|
|
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
|
partial_name (str): 目标节点的模糊或不完整名称
|
|
quantity_type (str): 工程量类型('定额'、'主材'、'设备'或None表示所有类型)
|
|
|
|
Returns:
|
|
list: 包含所有匹配节点的列表，如果没有匹配返回空列表
|
|
"""
|
|
if not partial_name:
|
|
return []
|
|
|
|
# 调试输出
|
|
print(f"搜索父路径: {parent_path}")
|
|
print(f"搜索名称: {partial_name}")
|
|
print(f"搜索类型: {quantity_type}")
|
|
|
|
# 使用通用方法获取父节点
|
|
parent_node_data = self.get_node_by_path(parent_path)
|
|
|
|
if parent_node_data:
|
|
print(f"找到父节点: {parent_node_data}")
|
|
else:
|
|
print("未找到父节点")
|
|
|
|
parent_node_id = parent_node_data["id"] if parent_node_data and "id" in parent_node_data else None
|
|
|
|
# 根据工程量类型确定标签和类型条件
|
|
node_labels = []
|
|
type_condition = ""
|
|
|
|
if quantity_type == "定额":
|
|
node_labels = ["ProjectQuantity", "Quota"]
|
|
type_condition = "q.类型 = '0'"
|
|
elif quantity_type == "主材":
|
|
node_labels = ["ProjectQuantity", "MainMaterial"]
|
|
type_condition = "q.类型 = '1'"
|
|
elif quantity_type == "设备":
|
|
node_labels = ["ProjectQuantity", "Equipment"]
|
|
type_condition = "q.类型 = '5'"
|
|
else:
|
|
node_labels = ["ProjectQuantity"]
|
|
|
|
# 构建标签字符串
|
|
labels_str = ":" + ":".join(node_labels) if node_labels else ""
|
|
|
|
# 扩展关系类型
|
|
relationship_types = "CONTAINS|HAS|RELATED_TO|USES|BELONGS_TO"
|
|
|
|
# 构建查询 - 使用递归关系查询
|
|
if parent_node_id:
|
|
query = f"""
|
|
MATCH (p)-[:{relationship_types}*1..10]->(q{labels_str})
|
|
WHERE p.id = $parent_id AND q.name CONTAINS $partial_name
|
|
{f'AND {type_condition}' if type_condition else ''}
|
|
RETURN q LIMIT 50
|
|
"""
|
|
params = {"parent_id": parent_node_id, "partial_name": partial_name}
|
|
print(f"执行查询: {query}")
|
|
print(f"参数: {params}")
|
|
else:
|
|
query = f"""
|
|
MATCH (q{labels_str})
|
|
WHERE q.name CONTAINS $partial_name
|
|
{f'AND {type_condition}' if type_condition else ''}
|
|
RETURN q LIMIT 50
|
|
"""
|
|
params = {"partial_name": partial_name}
|
|
print(f"执行全局查询: {query}")
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
quantities = []
|
|
for record in result:
|
|
node_data = record["q"]
|
|
print(f"找到节点: {node_data}")
|
|
|
|
# 创建对应类型的对象
|
|
quantity = self._create_quantity_object(node_data, quantity_type)
|
|
|
|
# 填充属性
|
|
for key, value in node_data.items():
|
|
if hasattr(quantity, key):
|
|
setattr(quantity, key, value)
|
|
|
|
quantities.append(quantity)
|
|
|
|
print(f"查询结果数量: {len(quantities)}")
|
|
return quantities
|
|
except Exception as e:
|
|
print(f"通过名称获取工程量对象时出错: {e}")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return []
|
|
|
|
# 辅助方法，用于根据节点数据创建对应类型的工程量对象
|
|
def _create_quantity_object(self, node_data, quantity_type=None):
|
|
"""
|
|
根据节点数据创建对应类型的工程量对象
|
|
|
|
Args:
|
|
node_data (dict): 节点数据
|
|
quantity_type (str): 工程量类型('定额'、'主材'、'设备'或None)
|
|
|
|
Returns:
|
|
ProjectQuantity: 创建的工程量对象
|
|
"""
|
|
# 如果指定了类型，直接创建对应类型的对象
|
|
if quantity_type == "定额":
|
|
return Ration()
|
|
elif quantity_type == "主材":
|
|
return Material()
|
|
elif quantity_type == "设备":
|
|
return Equipment()
|
|
|
|
# 如果没有指定类型，尝试通过节点属性或标签判断
|
|
if "类型" in node_data:
|
|
if node_data["类型"] == "0":
|
|
return Ration()
|
|
elif node_data["类型"] == "1":
|
|
return Material()
|
|
elif node_data["类型"] == "5":
|
|
return Equipment()
|
|
|
|
# 通过标签判断
|
|
labels = list(node_data.labels) if hasattr(node_data, "labels") else []
|
|
if "Quota" in labels:
|
|
return Ration()
|
|
elif "MainMaterial" in labels:
|
|
return Material()
|
|
elif "Equipment" in labels:
|
|
return Equipment()
|
|
|
|
# 默认返回基类对象
|
|
return ProjectQuantity()
|
|
|
|
# 材机查询方法实现
|
|
def get_material_equipment_by_path(self, paths_str):
|
|
"""
|
|
通过路径获取材机对象
|
|
|
|
Args:
|
|
paths_str (str): 以'/'分隔的多级项目划分名称路径
|
|
|
|
Returns:
|
|
list: 包含所有匹配的材机对象的列表
|
|
"""
|
|
if not paths_str:
|
|
return []
|
|
|
|
# 使用通用方法获取节点
|
|
node_data = self.get_node_by_path(paths_str)
|
|
node_id = node_data["id"] if node_data and "id" in node_data else None
|
|
|
|
if not node_id:
|
|
return []
|
|
|
|
# 查询与该节点关联的所有材机对象
|
|
query = """
|
|
MATCH (p)-[r]-(m)
|
|
WHERE p.id = $node_id AND (m:MaterialOrEquipment OR m:Material OR m:Equipment)
|
|
RETURN m LIMIT 50
|
|
"""
|
|
params = {"node_id": node_id}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
materials = []
|
|
for record in result:
|
|
material = self._create_material_object(record["m"])
|
|
materials.append(material)
|
|
|
|
# 更新缓存
|
|
if hasattr(material, "id") and material.id:
|
|
self.material_equipment_dict[material.id] = material
|
|
|
|
return materials
|
|
except Exception as e:
|
|
print(f"通过路径获取材机对象时出错: {e}")
|
|
return []
|
|
|
|
def get_material_equipment_by_parent_and_name(self, parent_path, partial_name):
|
|
"""
|
|
通过父节点路径和模糊名称获取材机对象
|
|
|
|
Args:
|
|
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
|
partial_name (str): 目标节点的模糊或不完整名称
|
|
|
|
Returns:
|
|
list: 包含所有匹配的材机对象的列表
|
|
"""
|
|
if not partial_name:
|
|
return []
|
|
|
|
# 使用通用方法获取父节点
|
|
parent_node_data = self.get_node_by_path(parent_path)
|
|
parent_node_id = parent_node_data["id"] if parent_node_data and "id" in parent_node_data else None
|
|
|
|
# 构建查询，根据是否有父节点ID调整查询条件
|
|
if parent_node_id:
|
|
# 如果找到了父节点，查找与父节点有关系的材机节点
|
|
query = """
|
|
MATCH (p)-[:CONTAINS|HAS|USES|RELATED_TO]-(m)
|
|
WHERE p.id = $parent_id AND m.name CONTAINS $partial_name
|
|
AND (m:MaterialOrEquipment OR m:Material OR m:Equipment)
|
|
RETURN m LIMIT 20
|
|
"""
|
|
params = {"parent_id": parent_node_id, "partial_name": partial_name}
|
|
else:
|
|
# 如果没有找到父节点或没有提供父节点路径，只按名称查询
|
|
query = """
|
|
MATCH (m)
|
|
WHERE m.name CONTAINS $partial_name
|
|
AND (m:MaterialOrEquipment OR m:Material OR m:Equipment)
|
|
RETURN m LIMIT 20
|
|
"""
|
|
params = {"partial_name": partial_name}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
materials = []
|
|
for record in result:
|
|
material = self._create_material_object(record["m"])
|
|
materials.append(material)
|
|
|
|
# 更新缓存
|
|
if hasattr(material, "id") and material.id:
|
|
self.material_equipment_dict[material.id] = material
|
|
|
|
return materials
|
|
except Exception as e:
|
|
print(f"通过父节点路径和模糊名称获取材机对象时出错: {e}")
|
|
return []
|
|
|
|
# 辅助方法，用于创建材机对象并填充属性
|
|
def _create_material_object(self, node_data):
|
|
"""
|
|
根据节点数据创建材机对象并填充属性
|
|
|
|
Args:
|
|
node_data (dict): 节点数据
|
|
|
|
Returns:
|
|
MaterialOrEquipment: 创建的材机对象
|
|
"""
|
|
material = MaterialOrEquipment()
|
|
|
|
# 填充属性
|
|
for key, value in node_data.items():
|
|
if hasattr(material, key):
|
|
setattr(material, key, value)
|
|
|
|
return material
|
|
|
|
# 取费表模板查询方法实现
|
|
def get_fee_template_by_path(self, paths_str):
|
|
"""
|
|
通过路径获取取费表模板
|
|
|
|
Args:
|
|
paths_str (str): 以'/'分隔的多级项目划分名称路径
|
|
|
|
Returns:
|
|
list: 包含所有匹配的取费表模板对象的列表
|
|
"""
|
|
if not paths_str:
|
|
return []
|
|
|
|
# 使用通用方法获取节点
|
|
node_data = self.get_node_by_path(paths_str)
|
|
node_id = node_data["id"] if node_data and "id" in node_data else None
|
|
|
|
if not node_id:
|
|
return []
|
|
|
|
# 查询与该节点关联的所有取费表模板
|
|
query = """
|
|
MATCH (p)-[r]-(t)
|
|
WHERE p.id = $node_id AND (t:FeeTableTemplate OR t.type = 'FeeTableTemplate')
|
|
RETURN t LIMIT 20
|
|
"""
|
|
params = {"node_id": node_id}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
templates = []
|
|
for record in result:
|
|
template = self._create_fee_template_object(record["t"])
|
|
templates.append(template)
|
|
|
|
return templates
|
|
except Exception as e:
|
|
print(f"通过路径获取取费表模板时出错: {e}")
|
|
return []
|
|
|
|
def get_fee_template_by_parent_and_name(self, parent_path, partial_name):
|
|
"""
|
|
通过父节点路径和模糊名称获取取费表模板
|
|
|
|
Args:
|
|
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
|
partial_name (str): 目标节点的模糊或不完整名称
|
|
|
|
Returns:
|
|
list: 包含所有匹配的取费表模板对象的列表
|
|
"""
|
|
if not partial_name:
|
|
return []
|
|
|
|
# 使用通用方法获取父节点
|
|
parent_node_data = self.get_node_by_path(parent_path)
|
|
parent_node_id = parent_node_data["id"] if parent_node_data and "id" in parent_node_data else None
|
|
|
|
# 构建查询，根据是否有父节点ID调整查询条件
|
|
if parent_node_id:
|
|
query = """
|
|
MATCH (p)-[:CONTAINS|HAS|USES|RELATED_TO]-(t)
|
|
WHERE p.id = $parent_id AND t.name CONTAINS $partial_name
|
|
AND (t:FeeTableTemplate OR t.type = 'FeeTableTemplate')
|
|
RETURN t LIMIT 20
|
|
"""
|
|
params = {"parent_id": parent_node_id, "partial_name": partial_name}
|
|
else:
|
|
query = """
|
|
MATCH (t)
|
|
WHERE t.name CONTAINS $partial_name
|
|
AND (t:FeeTableTemplate OR t.type = 'FeeTableTemplate')
|
|
RETURN t LIMIT 20
|
|
"""
|
|
params = {"partial_name": partial_name}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
templates = []
|
|
for record in result:
|
|
template = self._create_fee_template_object(record["t"])
|
|
templates.append(template)
|
|
|
|
return templates
|
|
except Exception as e:
|
|
print(f"通过父节点路径和模糊名称获取取费表模板时出错: {e}")
|
|
return []
|
|
|
|
# 辅助方法，用于创建取费表模板对象并填充属性
|
|
def _create_fee_template_object(self, node_data):
|
|
"""
|
|
根据节点数据创建取费表模板对象并填充属性
|
|
|
|
Args:
|
|
node_data (dict): 节点数据
|
|
|
|
Returns:
|
|
FeeTableTemplateItem: 创建的取费表模板对象
|
|
"""
|
|
template = FeeTableTemplateItem()
|
|
|
|
# 填充属性
|
|
for key, value in node_data.items():
|
|
if hasattr(template, key):
|
|
setattr(template, key, value)
|
|
|
|
# 更新缓存
|
|
if hasattr(template, "OutlayID") and template.OutlayID:
|
|
self.fee_templates[template.OutlayID] = template
|
|
|
|
return template
|
|
|
|
# 费用表查询方法实现
|
|
def get_fee_schedule_on_auxiliary_expense_table(self, table_name, fee_name, fee):
|
|
"""
|
|
在辅助费用表中查找费用
|
|
|
|
Args:
|
|
table_name (str): 费用表名称
|
|
fee_name (str): 要查找的费用名称
|
|
fee (str): 匹配的费用值属性名
|
|
|
|
Returns:
|
|
str: 匹配到的费用名称节点对应的费用值
|
|
"""
|
|
if not table_name or not fee_name or not fee:
|
|
return None
|
|
|
|
# 构建查询，查找辅助费用表中的特定费用 - 使用任意关系类型
|
|
query = """
|
|
MATCH (t:FeeScheduleItem)-[r]->(f:Fee)
|
|
WHERE t.name = $table_name AND f.name = $fee_name
|
|
RETURN f LIMIT 1
|
|
"""
|
|
params = {"table_name": table_name, "fee_name": fee_name}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
all_records = result.data()
|
|
|
|
if len(all_records) > 0:
|
|
fee_node = all_records[0]["f"]
|
|
value = fee_node.get(fee)
|
|
if value is not None:
|
|
return value
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"在辅助费用表中查找费用时出错: {e}")
|
|
return None
|
|
|
|
def get_fee_schedule_on_other_expense_table(self, table_name, fee_name, fee):
|
|
"""
|
|
在其它费用表中查找费用
|
|
|
|
Args:
|
|
table_name (str): 费用表名称
|
|
fee_name (str): 要查找的费用名称
|
|
fee (str): 匹配的费用值属性名
|
|
|
|
Returns:
|
|
str: 匹配到的费用名称节点对应的费用值
|
|
"""
|
|
if not table_name or not fee_name or not fee:
|
|
return None
|
|
|
|
# 构建查询，查找其它费用表中的特定费用 - 使用任意关系类型
|
|
query = """
|
|
MATCH (t:FeeScheduleItem)-[r]->(f:Fee)
|
|
WHERE t.name = $table_name AND f.name = $fee_name
|
|
RETURN f LIMIT 1
|
|
"""
|
|
params = {"table_name": table_name, "fee_name": fee_name}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
all_records = result.data()
|
|
|
|
if len(all_records) > 0:
|
|
fee_node = all_records[0]["f"]
|
|
value = fee_node.get(fee)
|
|
if value is not None:
|
|
return value
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"在其它费用表中查找费用时出错: {e}")
|
|
return None
|
|
|
|
def get_fee_schedule_on_land_acquisition_fee_table_table(self, table_name, fee_name, fee):
|
|
"""
|
|
在土地征用费表中查找费用
|
|
|
|
Args:
|
|
table_name (str): 费用表名称
|
|
fee_name (str): 要查找的费用名称
|
|
fee (str): 匹配的费用值属性名
|
|
|
|
Returns:
|
|
str: 匹配到的费用名称节点对应的费用值
|
|
"""
|
|
if not table_name or not fee_name or not fee:
|
|
return None
|
|
|
|
# 构建查询，查找土地征用费表中的特定费用 - 使用任意关系类型
|
|
query = """
|
|
MATCH (t:FeeScheduleItem)-[r]->(f:Fee)
|
|
WHERE t.name = $table_name AND f.name = $fee_name
|
|
RETURN f LIMIT 1
|
|
"""
|
|
params = {"table_name": table_name, "fee_name": fee_name}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
all_records = result.data()
|
|
|
|
if len(all_records) > 0:
|
|
fee_node = all_records[0]["f"]
|
|
value = fee_node.get(fee)
|
|
if value is not None:
|
|
return value
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"在土地征用费表中查找费用时出错: {e}")
|
|
return None
|
|
|
|
def get_fee_schedule_on_installation_price_difference_table(self, table_name, fee_name, fee):
|
|
"""
|
|
在安装价差表中查找费用
|
|
|
|
Args:
|
|
table_name (str): 费用表名称
|
|
fee_name (str): 要查找的费用名称
|
|
fee (str): 匹配的费用值属性名
|
|
|
|
Returns:
|
|
str: 匹配到的费用名称节点对应的费用值
|
|
"""
|
|
if not table_name or not fee_name or not fee:
|
|
return None
|
|
|
|
# 构建查询，查找安装价差表中的特定费用 - 使用任意关系类型
|
|
query = """
|
|
MATCH (t:FeeScheduleItem)-[r]->(f:Fee)
|
|
WHERE t.name = $table_name AND f.name = $fee_name
|
|
RETURN f LIMIT 1
|
|
"""
|
|
params = {"table_name": table_name, "fee_name": fee_name}
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
|
|
all_records = result.data()
|
|
|
|
if len(all_records) > 0:
|
|
fee_node = all_records[0]["f"]
|
|
value = fee_node.get(fee)
|
|
if value is not None:
|
|
return value
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"在安装价差表中查找费用时出错: {e}")
|
|
return None
|
|
|
|
def get_fee_schedule_on_Engineering_Cost_table(self, table_name, fee_name, fee):
|
|
"""
|
|
在工程费用表中查找费用
|
|
|
|
Args:
|
|
table_name (str): 费用表名称
|
|
fee_name (str): 要查找的费用名称
|
|
fee (str): 匹配的费用值属性名
|
|
|
|
Returns:
|
|
str: 匹配到的费用名称节点对应的费用值
|
|
"""
|
|
if not table_name or not fee_name or not fee:
|
|
return None
|
|
|
|
# 调试输出
|
|
print(f"查询费用表: {table_name}")
|
|
print(f"查询费用名称: {fee_name}")
|
|
print(f"查询费用属性: {fee}")
|
|
|
|
# 构建查询，使用递归关系查询，并扩展关系类型
|
|
query = """
|
|
MATCH (t:FeeScheduleItem)-[r*1..5]->(f:Fee)
|
|
WHERE t.name = $table_name AND f.name = $fee_name
|
|
RETURN f LIMIT 10
|
|
"""
|
|
params = {"table_name": table_name, "fee_name": fee_name}
|
|
|
|
print(f"执行查询: {query}")
|
|
print(f"参数: {params}")
|
|
|
|
try:
|
|
result = self.session.run(query, **params)
|
|
all_records = result.data()
|
|
|
|
print(f"查询结果数量: {len(all_records)}")
|
|
|
|
if len(all_records) > 0:
|
|
fee_node = all_records[0]["f"]
|
|
print(f"找到费用节点: {fee_node}")
|
|
|
|
# 获取节点的所有属性，用于调试
|
|
for key, value in fee_node.items():
|
|
print(f"属性: {key} = {value}")
|
|
|
|
value = fee_node.get(fee)
|
|
if value is not None:
|
|
print(f"找到费用值: {value}")
|
|
return value
|
|
else:
|
|
print(f"节点中没有属性: {fee}")
|
|
|
|
# 如果没有找到，尝试另一种查询方式，使用CONTAINS进行模糊匹配
|
|
print("尝试使用模糊匹配...")
|
|
backup_query = """
|
|
MATCH (t:FeeScheduleItem)-[r*1..5]->(f:Fee)
|
|
WHERE t.name CONTAINS $table_name AND f.name CONTAINS $fee_name
|
|
RETURN f LIMIT 10
|
|
"""
|
|
backup_result = self.session.run(backup_query, **params)
|
|
backup_records = backup_result.data()
|
|
|
|
print(f"模糊匹配结果数量: {len(backup_records)}")
|
|
|
|
if len(backup_records) > 0:
|
|
fee_node = backup_records[0]["f"]
|
|
print(f"找到费用节点(模糊匹配): {fee_node}")
|
|
|
|
# 获取节点的所有属性，用于调试
|
|
for key, value in fee_node.items():
|
|
print(f"属性: {key} = {value}")
|
|
|
|
value = fee_node.get(fee)
|
|
if value is not None:
|
|
print(f"找到费用值: {value}")
|
|
return value
|
|
else:
|
|
print(f"节点中没有属性: {fee}")
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"在工程费用表中查找费用时出错: {e}")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return None
|
|
|
|
|
|
class ProjectBuilder:
|
|
"""
|
|
项目构建器
|
|
描述: 用于构建项目对象的构建器
|
|
"""
|
|
|
|
_instance = None
|
|
|
|
@staticmethod
|
|
def build():
|
|
"""
|
|
构建并返回项目实例
|
|
|
|
Returns:
|
|
ProjectTookiItNeo4j: 创建的项目实例
|
|
"""
|
|
# 如果已经有实例，先关闭它
|
|
if ProjectBuilder._instance is not None:
|
|
ProjectBuilder._instance.close()
|
|
|
|
# 创建新实例
|
|
|
|
ProjectBuilder._instance = ProjectTookiItNeo4j()
|
|
|
|
return ProjectBuilder._instance
|
|
|
|
@staticmethod
|
|
def close():
|
|
"""
|
|
关闭当前项目实例的连接
|
|
"""
|
|
if ProjectBuilder._instance is not None:
|
|
ProjectBuilder._instance.close()
|
|
ProjectBuilder._instance = None
|
|
|
|
|
|
# 注册退出处理函数，确保程序退出时自动关闭连接
|
|
atexit.register(ProjectBuilder.close)
|
|
|
|
|
|
# project = ProjectBuilder.build()
|
|
# result = project.get_quantities_node_by_parent_and_code(
|
|
# "工程数据/安装工程/安装/架空输电线路本体工程/基础工程", "定额", "YX2-1/YX2-2/YX2-3/YX2-6/YX2-7"
|
|
# )
|
|
|
|
# print(result)
|