# coding:utf8 import re from typing import Dict, Any from BiddingKG.dl.template_extract.abstract_template import AbstractTemplate, table2list import os import json class AdaptedPurchaseTemplate(AbstractTemplate): """ 适配data.json表格结构的信息提取模板 提取项目信息、招标信息、中标信息、候选人信息及产品信息 """ 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: print("无表格数据") return False try: # 验证供应商表格表头(主要数据表格) target_header_patterns = [ r'序号', # 第1列:序号 r'供应商名称|供应商', # 第2列:供应商名称 r'候选供应商|中标候选人|投标人类型', # 第3列:候选状态 r'已报物料|投标内容|标的', # 第4列:已报物料 r'交货时间|投标时间|有效期' # 第5列:时间信息 ] # 查找符合条件的主要表格 self.main_tables = [] for table in tables: if len(table) < 1: continue header = table[0] if len(header) != 5: continue match = True for i, pattern in enumerate(target_header_patterns): if not re.search(pattern, header[i]): # 使用search增强匹配灵活性 match = False break if match and len(table) >= 2: # 确保有数据行 self.main_tables.append(table) # 至少需要一个主要表格 return len(self.main_tables) > 0 except Exception as e: print(f"表格验证错误: {e}") return False def extract(self, preprocessed_data: Dict[str, Any]) -> Dict[str, Any]: """提取表格中的相关信息及字段""" extract_result = { "招标信息": [], "中标信息": [], "候选人信息": [], "产品信息": [] } # 处理主要表格数据 for table in self.main_tables: header = table[0] for row in table[1:]: # 跳过表头行 if len(row) != 5: continue # 跳过列数异常的行 序号, 供应商名称, 候选状态, 已报物料, 交货时间 = [cell.strip() for cell in row] # 处理中标信息(仅预成交供应商作为中标人) if re.fullmatch(r'预成交供应商', 候选状态): if 供应商名称: extract_result["中标信息"].append({"中标人": 供应商名称}) # 处理候选人信息(明确排除备选供应商,此处逻辑可根据实际需求扩展其他候选人类型) # 按需求:备选供应商不作为候选人,故不添加相关逻辑 # 过滤不符合条件的字段 # 招标信息必须包含招标人或预算 if not extract_result["招标信息"] or all( not (info.get("招标人") or info.get("预算")) for info in extract_result["招标信息"] ): extract_result["招标信息"] = [] # 中标信息必须包含中标人 extract_result["中标信息"] = [ info for info in extract_result["中标信息"] if info.get("中标人") ] # 候选人信息必须包含候选人和排名(当前逻辑下候选人信息应为空) extract_result["候选人信息"] = [ info for info in extract_result["候选人信息"] if info.get("候选人") and info.get("排名") ] # 有候选人信息则清空中标信息(当前逻辑下不影响) if extract_result["候选人信息"]: extract_result["中标信息"] = [] # 移除无数据的字段 if not extract_result["招标信息"]: del extract_result["招标信息"] if not extract_result["中标信息"]: del extract_result["中标信息"] if not extract_result["候选人信息"]: del extract_result["候选人信息"] if not extract_result["产品信息"]: del extract_result["产品信息"] return extract_result # 使用示例 if __name__ == "__main__": # 读取数据文件 with open("data.json", "r", encoding="utf8") as f: preprocessed_data = json.load(f) # 初始化模板并提取信息 template = AdaptedPurchaseTemplate() if template.check_call_timing(preprocessed_data): result = template.extract(preprocessed_data) print("提取结果:", json.dumps(result, ensure_ascii=False, indent=2)) else: print("当前数据不满足模板调用条件")