BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/shared_template/template5.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]
            # 表头列数需匹配目标表格（4列）
            if len(header) != 4:
                # print(f"表头列数异常：{len(header)}列，预期4列")
                return False

            # 表头内容正则匹配（泛化类似表达）
            header_patterns = [
                r'供应商名称|中标单位名称',  # 第1列：供应商名称
                r'供应商地址|中标单位地址',  # 第2列：供应商地址
                r'中标（成交）金额|成交价|中标价|成交金额',  # 第3列：金额
                r'评审总得分|评审得分|综合得分|评分'  # 第4列：评审得分
            ]

            # 逐列验证表头
            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 = {
            "项目编号": "",
            "项目名称": "",
            "招标信息": [],
            "中标信息": [],
            "候选人信息": [],
            "产品信息": []
        }

        # 过滤空值字段的辅助函数
        def filter_empty_fields(d: Dict[str, Any]) -> Dict[str, Any]:
            return {k: v for k, v in d.items() if v and v.strip()}

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

            # 解析每行数据
            supplier_name, supplier_addr, amount, score = row

            # 补充金额单位（如果表头有单位而内容没有）
            if re.search(r'[万亿美欧日]?元', self.header[2]) and not re.search(r'[万亿美欧日]?元', amount):
                amount += re.search(r'[万亿美欧日]?元', self.header[2]).group(0)

            # ******** 中标信息提取 ********
            # 中标信息必须包含中标人
            if supplier_name:
                bid_win_info = {
                    "标的": "",
                    "标包": "",
                    "包号": "",
                    "中标人": supplier_name,
                    "中标价": amount,
                    "联系人": "",
                    "电话": "",
                    "服务时间": "",
                    "地址": supplier_addr
                }
                filtered_bid_win = filter_empty_fields(bid_win_info)
                if filtered_bid_win:  # 确保有有效字段
                    extract_result["中标信息"].append(filtered_bid_win)

            # ******** 候选人信息提取（无排名则不提取）********
            # 本表格无排名字段，故候选人信息为空
            # ******** 招标信息提取（需包含招标人或预算，本表格无相关字段，故置空）********
            # ******** 产品信息提取（无相关字段，故置空）********

        # 过滤空列表和空值字段
        final_result = {}
        for key, value in extract_result.items():
            if isinstance(value, list):
                if value:  # 列表非空时保留
                    filtered_list = [filter_empty_fields(item) for item in value if filter_empty_fields(item)]
                    if filtered_list:
                        final_result[key] = filtered_list
            else:
                if value and value.strip():  # 非空字符串保留
                    final_result[key] = value

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