# coding:utf8 import re from typing import Dict, Any from BiddingKG.dl.template_extract.abstract_template import AbstractTemplate, table2list import os import json class DealInfoTemplate(AbstractTemplate): """ 成交信息提取模板(适配包含标包、成交人等字段的表格) """ def __init__(self): super().__init__( template_id=os.path.abspath(__file__), priority=3 ) def check_call_timing(self, preprocessed_data: Dict[str, Any]) -> bool: """检查是否符合模板调用条件""" if "表格列表" not in preprocessed_data: return False tables = preprocessed_data["表格列表"] if len(tables) < 1: return False try: datas = tables[0] # 验证表头列数(目标表格为3列) if len(datas[0]) != 3: print("表头数量异常;", len(datas[0])) return False # 表头内容正则匹配(泛化类似表达) header_patterns = [ '序号|编号', # 第1列:序号(泛化编号表达) '标包(名称)?|标段(名称)?|包(名称)?|项目(名称)?', # 第2列:标包(泛化项目、标段等表达) '(成交|中标|中选)(人|单位|供应商)(名称)?' # 第3列:成交人(泛化中标人等表达) ] # 逐列验证表头 for i, pattern in enumerate(header_patterns): if not re.fullmatch(pattern, datas[0][i]): print('表头未验证: ', pattern, datas[0][i]) return False # 确保表格有数据行 if len(datas) < 2: return False self.header = datas[0] self.data = datas[1:] return True except Exception as e: print(f"表格验证错误: {e}") return False def extract(self, preprocessed_data: Dict[str, Any]) -> Dict[str, Any]: """提取表格中的中标信息等内容""" extract_result = { "招标信息": [], "中标信息": [], "候选人信息": [] } for row in self.data: # 跳过列数异常的行 if len(row) != 3: continue # 解析每行数据(对应3列结构) seq, package, winner = row # 整理中标信息(必须包含中标人) if winner: # 确保中标人信息存在 bid_info = { "标包": package, "中标人": winner } extract_result["中标信息"].append(bid_info) # 招标信息无招标人或预算,置空 # 候选人信息无候选人及排名,不添加 return extract_result # 使用示例 if __name__ == "__main__": from lxml import etree # 读取并解析数据文件 with open("data.json", "r", encoding="utf8") as f: preprocessed_data = json.load(f) # 初始化模板并提取信息 template = DealInfoTemplate() if template.check_call_timing(preprocessed_data): result = template.extract(preprocessed_data) print("提取结果:", json.dumps(result, ensure_ascii=False, indent=2)) else: print("当前数据不满足模板调用条件")