BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/shared_template/template4.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 BidInfoExtractTemplate(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]
            # 至少需要表头行和一行数据
            if len(datas) < 2:
                return False

            header = datas[0]
            # 表头列数需匹配目标表格（9列）
            if len(header) != 9:
                # print(f"表头列数异常：{len(header)}列，预期9列")
                return False

            # 表头内容正则匹配（泛化类似表达）
            header_patterns = [
                r'分包名称|标包名称|标段名称|包号|标段号',  # 第1列：分包名称
                r'供应商名称|投标人名称|中标人名称|候选人名称',  # 第2列：供应商名称
                r'报价金额|投标报价|参选价|投标价(（万?元）)?',  # 第3列：报价金额
                r'成交金额|中标金额|成交价(（万?元）)?',  # 第4列：成交金额
                r'实际成交金额|最终成交金额|实际中标金额(（万?元）)?',  # 第5列：实际成交金额
                r'评审方式|评审类型|评标方式',  # 第6列：评审方式
                r'评审结果|中标结果|成交结果',  # 第7列：评审结果
                r'需求日期|采购日期|招标日期|服务日期',  # 第8列：需求日期
                r'成交/未成交原因|中标/未中标原因|成交原因|未成交原因'  # 第9列：成交/未成交原因
            ]

            # 逐列验证表头
            for i, pattern in enumerate(header_patterns):
                if not re.fullmatch(pattern, header[i]):
                    # print(f'表头第{i+1}列未匹配：{pattern} vs {header[i]}')
                    return False

            self.header = header
            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) != 9:
                continue

            # 解析每行数据（按data.json表头顺序）
            package_name, supplier_name, quote_amount, deal_amount, actual_deal_amount, review_method, review_result, demand_date, deal_reason = row

            # 处理金额单位补充
            # 报价金额单位补充
            if re.search(r'[万亿美欧日]?元', self.header[2]) and not re.search(r'[万亿美欧日]?元', quote_amount):
                quote_amount += re.search(r'[万亿美欧日]?元', self.header[2]).group(0)
            # 成交金额单位补充
            if re.search(r'[万亿美欧日]?元', self.header[3]) and not re.search(r'[万亿美欧日]?元', deal_amount):
                deal_amount += re.search(r'[万亿美欧日]?元', self.header[3]).group(0)
            # 实际成交金额单位补充
            if re.search(r'[万亿美欧日]?元', self.header[4]) and not re.search(r'[万亿美欧日]?元', actual_deal_amount):
                actual_deal_amount += re.search(r'[万亿美欧日]?元', self.header[4]).group(0)

            # 判断是否为成交/中标结果，优先提取候选人信息（有排名逻辑，此处根据成交原因提取排名）
            rank = ""
            # 从成交原因中提取排名（如"第1中选人"）
            rank_match = re.search(r'第(\d+)中选人', deal_reason)
            if rank_match:
                rank = rank_match.group(1)

            if rank and supplier_name:
                # 提取候选人信息（满足候选人+排名条件）
                candidate_info = {
                    "标的": "",  # 表格无标的信息，置空
                    "标包": package_name,
                    "包号": "",  # 表格无包号信息，置空
                    "候选人": supplier_name,
                    "投标价": quote_amount,
                    "排名": rank,
                    "联系人": "",  # 表格无候选人联系人信息，置空
                    "电话": "",  # 表格无候选人电话信息，置空
                    "服务时间": demand_date,  # 用需求日期替代服务时间
                    "地址": ""  # 表格无候选人地址信息，置空
                }
                # 过滤空值字段
                candidate_info = {k: v for k, v in candidate_info.items() if v}
                extract_result["候选人信息"].append(candidate_info)
            elif review_result == "成交" and supplier_name:
                # 提取中标信息（满足中标人条件）
                bid_win_info = {
                    "标的": "",  # 表格无标的信息，置空
                    "标包": package_name,
                    "包号": "",  # 表格无包号信息，置空
                    "中标人": supplier_name,
                    "中标价": actual_deal_amount,
                    "联系人": "",  # 表格无中标人联系人信息，置空
                    "电话": "",  # 表格无中标人电话信息，置空
                    "服务时间": demand_date,  # 用需求日期替代服务时间
                    "地址": ""  # 表格无中标人地址信息，置空
                }
                # 过滤空值字段
                bid_win_info = {k: v for k, v in bid_win_info.items() if v}
                extract_result["中标信息"].append(bid_win_info)

            # 招标信息：表格无招标人/预算信息，故置空
            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["中标信息"]
        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 = BidInfoExtractTemplate()
    if template.check_call_timing(preprocessed_data):
        result = template.extract(preprocessed_data)
        print("提取结果：", json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print("当前数据不满足模板调用条件")