BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/DX007098/template1.py

# 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("当前数据不满足模板调用条件")