BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/30306/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 CustomDealInfoTemplate(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:
            return False

        try:
            # 取第一个表格进行验证
            datas = tables[0]
            # 验证表头列数
            if len(datas[0]) != 8:
                print("表头数量异常；", len(datas[0]))
                return False

            # 表头内容正则匹配（泛化表达）
            header_patterns = [
                '需求编号|项目编号',  # 第1列：需求编号/项目编号
                '(设备|材料)名称/编号|标的名称|产品名称',  # 第2列：设备名称/编号
                '成交候选人排名|排名|名次',  # 第3列：成交候选人排名
                '公司名称|供应商名称|单位名称',  # 第4列：公司名称
                '投标(设备|材料)数量|数量',  # 第5列：投标设备数量
                '最终报价（不含税）|报价|价格|金额',  # 第6列：最终报价
                '选择(设备|材料)',  # 第7列：选择设备（用于判断中标）
                '评审情况|评审结果'  # 第8列：评审情况
            ]

            # 逐列验证表头
            for i, pattern in enumerate(header_patterns):
                if not re.search(pattern, datas[0][i], re.IGNORECASE):
                    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 = {
            "项目编号": "",
            "项目名称": "",
            "招标信息": [],
            "中标信息": [],
            "候选人信息": [],
            "产品信息": []
        }

        # 提取项目编号（取第一行的需求编号）
        if self.data:
            extract_result["项目编号"] = self.data[0][0]

        # 用于存储已处理的标的信息，避免重复
        processed_subjects = set()

        for row in self.data:
            # 跳过列数异常的行
            if len(row) != 8:
                continue

            # 解析每行数据
            (req_id, device_name, rank, company, quantity, price, selected, review) = row

            # 候选人信息（必须包含候选人和排名）
            if company and rank:
                if rank == '1' and selected == "0": # 选择设备不为0的才算中标
                    rank = '5'
                elif rank != '1' and selected != "0":
                    rank = '1'
                candidate_info = {
                    "标的": device_name,
                    "候选人": company,
                    "投标价": f"{price}元" if price else "",  # 补充单位
                    "排名": rank,
                    "选择数量": selected
                }
                # 过滤空值字段
                candidate_info = {k: v for k, v in candidate_info.items() if v}
                extract_result["候选人信息"].append(candidate_info)

        # 招标信息：由于没有招标人或预算信息，置空
        extract_result["招标信息"] = []

        return extract_result


# 使用示例
if __name__ == "__main__":
    # 读取并解析数据文件
    with open("data.json", "r", encoding="utf8") as f:
        preprocessed_data = json.load(f)

    # 初始化模板并提取信息
    template = CustomDealInfoTemplate()
    if template.check_call_timing(preprocessed_data):
        result = template.extract(preprocessed_data)
        print("提取结果：", json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print("当前数据不满足模板调用条件")