Files
langchain_projectagent/project.py
T

617 lines
24 KiB
Python

"""
软件知识图谱类定义
根据Ontology_Layer.txt文件中的知识图谱信息创建
"""
from abc import ABC, abstractmethod
import json
class ProjectToolkit(ABC):
"""
项目类(抽象基类)
描述: 代表整个项目结构的顶层容器
"""
def __init__(self):
self.project_division_set = ProjectDivisionItem() # 项目划分集对象
# 项目划分查询方法
@abstractmethod
def get_division_by_name(self, name):
"""
通过名称获取项目划分对象
Args:
path (str): 项目划分节点的名称
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回的项目划分对象数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表,如果未找到节点,则包含父节点下所有ProjectDivisionItem节点
Note:
当路径为空字符串时,会返回None
"""
pass
@abstractmethod
def get_division_item_by_path(self, path):
"""
通过路径获取项目划分对象
Args:
path (str): 项目划分节点的路径,以'/'分隔的多级节点路径
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回的项目划分对象数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表,如果未找到节点,则包含父节点下所有ProjectDivisionItem节点
Note:
当路径为空字符串时,会返回None
"""
pass
@abstractmethod
def get_division_node_by_parent_and_name(self, parent_path, partial_name):
"""
通过父节点路径和模糊节点名称获取项目划分对象,包括子节点
Args:
parent_path (str): 父节点的路径,以'/'分隔的多级节点路径
partial_name (str): 目标节点的模糊或不完整名称
Returns:
Tuple[str, List[Dict[str, Any]], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (List[Dict[str, Any]]): 成功时返回的数据列表
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表
"""
pass
# 工程量查询方法
@abstractmethod
def get_quantities_by_paths(self, paths_str):
"""
获取指定项目路径下的工程量对象
Args:
paths_str (str): 以'/'分隔的多级节点路径
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]:
- status: 'success'或'error'
- data: 成功时返回的节点数据字典,失败时为{}
- error: 错误信息,成功时为''
- helper_info: 辅助信息列表(当目标节点不存在时返回父节点下的所有ProjectQuantity节点名称)
"""
pass
@abstractmethod
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:
Tuple[str, List[Dict[str, Any]], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (List[Dict[str, Any]]): 成功时返回的数据列表
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表
"""
pass
@abstractmethod
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:
Tuple[str, List[Dict[str, Any]], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (List[Dict[str, Any]]): 成功时返回的数据列表
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表
"""
pass
# 材机查询方法
@abstractmethod
def get_material_equipment_by_path(self, paths_str):
"""
通过路径获取材机对象(MaterialOrEquipment 类型节点)
Args:
path (str): 项目划分节点的路径,以'/'分隔的多级节点路径
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回的节点属性数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 如果未找到目标节点,则包含父节点下所有 MaterialOrEquipment 节点名称
"""
pass
@abstractmethod
def get_material_equipment_by_parent_and_name(self, parent_path, partial_name):
"""
通过父节点路径和模糊名称获取材机对象
Args:
parent_path (str): 父节点的路径,以'/'分隔的多级节点路径
partial_name (str): 目标节点的模糊或不完整名称
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回的节点属性数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 如果未找到目标节点,则包含父节点下所有 MaterialOrEquipment 节点名称
"""
pass
# 取费查询方法
@abstractmethod
def get_fee_template_by_path(self, paths_str):
"""
通过路径获取取费节点
Args:
path (str): 项目划分节点的路径,以'/'分隔的多级节点路径
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回的节点属性数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 如果未找到目标节点,则包含父节点下所有 FeeCollection 节点名称
"""
pass
@abstractmethod
def get_fee_template_by_parent_and_name(self, parent_path, partial_name):
"""
通过父节点路径和节点名称获取取费节点
Args:
parent_path (str): 父节点的路径,以'/'分隔的多级节点路径
node_name (str): 目标节点的名称,例如:直接费
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回的节点属性数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 如果未找到父节点,则包含父节点的父节点下所有节点名称
- 如果未找到目标节点,则包含父节点下所有 FeeCollection 节点名称
- 否则为空
"""
pass
# 费用表查询方法
@abstractmethod
def get_fee_schedule_on_auxiliary_expense_table(self, table_name, fee_name, fee: str):
"""
在辅助费用表中查找费用
Args:
table_name (str): 费用表名称
fee_name (str): 要查找的费用名称(可能在多级子节点中)
fee_attribute (str): 费用值属性名
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回包含属性值的数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 第一步失败:返回"工程数据"节点下所有子节点名称
- 第二步失败:返回费用表节点下所有子节点名称(递归查找)
- 第三步失败:返回费用节点的所有可用属性名称
- 否则为空
"""
pass
@abstractmethod
def get_fee_schedule_on_other_expense_table(self, table_name, fee_name, fee):
"""
在其它费用表中查找费用
Args:
table_name (str): 费用表名称
fee_name (str): 要查找的费用名称(可能在多级子节点中)
fee_attribute (str): 费用值属性名
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回包含属性值的数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 第一步失败:返回"工程数据"节点下所有子节点名称
- 第二步失败:返回费用表节点下所有子节点名称(递归查找)
- 第三步失败:返回费用节点的所有可用属性名称
- 否则为空
"""
pass
@abstractmethod
def get_fee_schedule_on_land_acquisition_fee_table_table(self, table_name, fee_name, fee):
"""
在其中:土地征用费表中查找费用
Args:
table_name (str): 费用表名称
fee_name (str): 要查找的费用名称(可能在多级子节点中)
fee_attribute (str): 费用值属性名
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回包含属性值的数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 第一步失败:返回"工程数据"节点下所有子节点名称
- 第二步失败:返回费用表节点下所有子节点名称(递归查找)
- 第三步失败:返回费用节点的所有可用属性名称
- 否则为空
"""
pass
@abstractmethod
def get_fee_schedule_on_installation_price_difference_table(self, table_name, fee_name, fee):
"""
在安装价差表中查找费用
Args:
table_name (str): 费用表名称
fee_name (str): 要查找的费用名称(可能在多级子节点中)
fee_attribute (str): 费用值属性名
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回包含属性值的数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 第一步失败:返回"工程数据"节点下所有子节点名称
- 第二步失败:返回费用表节点下所有子节点名称(递归查找)
- 第三步失败:返回费用节点的所有可用属性名称
- 否则为空
"""
pass
@abstractmethod
def get_fee_schedule_on_Engineering_Cost_table(self, table_name, fee_name, fee):
"""
在工程费用表中查找费用
Args:
table_name (str): 费用表名称
fee_name (str): 要查找的费用名称(可能在多级子节点中)
fee_attribute (str): 费用值属性名
Returns:
Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
- status (str): 状态,'success'或'error'
- data (Dict[str, Any]): 成功时返回包含属性值的数据
- error (str): 错误时的错误信息
- helper_info (List[Any]): 辅助信息列表:
- 第一步失败:返回"工程数据"节点下所有子节点名称
- 第二步失败:返回费用表节点下所有子节点名称(递归查找)
- 第三步失败:返回费用节点的所有可用属性名称
- 否则为空
"""
pass
class ProjectDivisionSet:
"""
项目划分项
描述: 代表项目结构中的层级条目,具有自身的属性,并且可以包含ProjectDivisionTree(项目划分树)
"""
def __init__(self):
self.name = None # xsd:string
class ProjectDivisionTree:
"""
项目划分项
描述: 代表项目结构中的层级条目,具有自身的属性,并且可以包含ProjectDivisionItem(项目划分项)
"""
def __init__(self):
self.name = None # xsd:string
class ProjectDivisionItem:
"""
项目划分项
描述: 代表项目结构中的层级条目,具有自身的属性,并且可以包含子项目划分项或详细工作项
JSON对应: ProjectDivisionSet数组中的对象,或ProjectDivisionItem的children数组中type为"项目划分"的对象
"""
def __init__(self):
self.GUID = None # xsd:string
self.id = None # xsd:string
self.name = None # xsd:string
self.代码 = None # xsd:string (可选)
self.费率 = None # xsd:string
self.单位 = None # xsd:string (可选)
self.取费表id = None # xsd:string
self.颜色标记 = None # xsd:string
self.取费表 = None # xsd:string
self.合价含税 = None # xsd:string (可选)
self.type = None # xsd:string
self.专业类型 = None # xsd:string
self.资源库列表 = None # xsd:list (可选)
self.notCheck = None # xsd:string (可选)
class ProjectQuantity:
"""
工程量
描述: 代表项目划分项(ProjectDivisionItem)下的具体工作单元或物料项,是定额、主材、设备的父类型
"""
def __init__(self):
self.id = None # xsd:string
self.类型 = None # xsd:string ("0"为定额,"1"为主材,"5"为设备)
self.name = None # xsd:string
self.编码 = None # xsd:string
self.单位 = None # xsd:string
self.数量 = None # xsd:string
self.资源库名称 = None # xsd:string
self.投标数量 = None # xsd:string (可选)
self.投标单价 = None # xsd:string (可选)
self.特征段 = None # xsd:string (可选)
self.关联父级量 = None # xsd:string (可选)
self.颜色标记 = None # xsd:string (可选)
self.单价不含税 = None
self.cost_set = None # xsd:CostSet
class Ration(ProjectQuantity):
"""
定额
描述: 代表一种标准的工程量条目,通常包含详细的工、料、机消耗标准
"""
def __init__(self):
super().__init__()
self.计算式 = None # xsd:string
self.中标计算式 = None # xsd:string
self.人工费 = None # xsd:string
self.机械费 = None # xsd:string
self.甲供材料费不含税 = None # xsd:string
self.材料费 = None # xsd:string
self.定额系数 = None # xsd:string
self.人工系数 = None # xsd:string
self.材料系数 = None # xsd:string
self.机械系数 = None # xsd:string
self.定额范围 = None # xsd:string
self.定额章节名称 = None # xsd:string
self.费用类型 = None # xsd:string
self.甲供材料费含税 = None # xsd:string
self.投标合价 = None # xsd:string
self.其中甲供材料费 = None # xsd:string
self.合价不含税 = None # xsd:string
self.基价 = None # xsd:string
self.所属定额库 = None # xsd:string
def __str__(self):
"""返回定额的字符串表示"""
attrs = {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
return json.dumps(attrs, ensure_ascii=False, indent=2)
class Material(ProjectQuantity):
"""
主材
描述: 代表工程中使用的主要材料
"""
def __init__(self):
super().__init__()
self.规格型号 = None # xsd:string
self.损耗率 = None # xsd:string
self.供货方 = None # xsd:string
self.集中配送 = None # xsd:string (可选)
self.单重 = None # xsd:string (可选)
self.市场价不含税 = None # xsd:string
self.市场价含税 = None # xsd:string
self.单价含税 = None # xsd:string
self.结算市场价不含税 = None # xsd:string (可选)
self.结算市场价含税 = None # xsd:string (可选)
self.基准价不含税 = None # xsd:string (可选)
self.基准价含税 = None # xsd:string (可选)
self.费用类型 = None # xsd:string
self.增值税率 = None # xsd:string (可选)
self.合价含税 = None # xsd:string
self.合价不含税 = None # xsd:string
self.线重 = None # xsd:string (可选)
self.制造长度 = None # xsd:string (可选)
self.截面积 = None # xsd:string (可选)
def __str__(self):
"""返回材料的字符串表示"""
attrs = {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
return json.dumps(attrs, ensure_ascii=False, indent=2)
class Equipment(ProjectQuantity):
"""
设备
描述: 代表工程中安装或使用的设备
"""
def __init__(self):
super().__init__()
self.规格型号 = None # xsd:string
self.供货方 = None # xsd:string
self.运杂费率 = None # xsd:string (可选)
self.单价含税 = None # xsd:string
self.设备类型 = None # xsd:string (可选)
self.增值税率 = None # xsd:string (可选)
self.合价含税 = None # xsd:string
self.合价不含税 = None # xsd:string
class MaterialOrEquipment:
"""
材机
描述: 代表DetailedWorkItem中所列出的具体材料、人工或机械设备及其详细信息
"""
def __init__(self):
self.id = None # xsd:string
self.编码 = None # xsd:string
self.名称 = None # xsd:string
self.单位 = None # xsd:string
self.类型 = None # xsd:string
self.供货方 = None # xsd:string
self.预算价不含税 = None # xsd:string
self.市场价不含税 = None # xsd:string
self.预算价含税 = None # xsd:string
self.市场价含税 = None # xsd:string
self.结算预算价不含税 = None # xsd:string
self.结算市场价不含税 = None # xsd:string
self.结算预算价含税 = None # xsd:string
self.结算市场价含税 = None # xsd:string
self.暂估价 = None # xsd:string
self.拆分 = None # xsd:string
self.全口径市场价不含税 = None # xsd:string
self.全口径市场价含税 = None # xsd:string
self.商品砼 = None # xsd:string
self.数量 = None # xsd:string
self.是否未计价 = None # xsd:string
class FeeTableTemplateItem:
"""
取费表模板项
描述:
"""
def __init__(self):
self.name = None # xsd:string
self.OutlayID = None # xsd:string
self.type = None # xsd:string
self.profession = None # xsd:string
class FeeCollection:
"""
取费
描述:
"""
def __init__(self):
self.name = None # xsd:string
self.serialNumber = None # xsd:string
self.code = None # xsd:string
self.base = None # xsd:string
self.rate = None # xsd:string
class FeeScheduleItem:
"""
费用表项
描述:
"""
def __init__(self):
self.name = None # xsd:string
class Fee:
"""
取费
描述:
"""
def __init__(self):
self.name = None # xsd:string
self.serialNumber = None # xsd:string
self.code = None # xsd:string
self.rate = None # xsd:string
self.amount = None # xsd:string
self.输出 = None # xsd:string (可选)
self.取费基数 = None # xsd:string (可选)
self.编制依据 = None # xsd:string (可选)
self.编码 = None # xsd:string (可选)
self.备注 = None # xsd:string (可选)
self.表一显示 = None # xsd:string (可选)
self.报表输出 = None # xsd:string (可选)
self.建安合计费 = None # xsd:string (可选)
self.合计费 = None # xsd:string (可选)
self.占总计 = None # xsd:string (可选)
self.建筑费 = None # xsd:string (可选)
self.安装费 = None # xsd:string (可选)
self.设备费 = None # xsd:string (可选)
self.其他费 = None # xsd:string (可选)
self.施工费 = None # xsd:string (可选)
self.单位投资 = None # xsd:string (可选)
class ProjectBuilder:
# 存储注册的工具类
_registry = None
_config = {}
"""项目工具工厂类"""
def __init__(self):
pass
@classmethod
def register(cls, toolkit_class: type, config: dict):
"""
注册工具类到工厂
参数:
toolkit_class: 继承自ProjectToolkit的具体工具类
"""
if not issubclass(toolkit_class, ProjectToolkit):
raise TypeError(f"{toolkit_class.__name__} 必须继承自 ProjectToolkit")
cls._config = config
cls._registry = toolkit_class
@classmethod
def build(cls) -> ProjectToolkit:
"""
创建工具实例
参数:
返回:
实例化的工具对象
"""
if cls._registry is None:
raise KeyError(f"未注册的类,请先注册类")
return cls._registry(cls._config)