BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/shared_template/template3.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 PurchaseTransactionTemplate(AbstractTemplate):
    """
    采购成交信息提取模板（适配物料采购表格场景）
    适配包含中标供应商、物料明细的表格提取
    """

    def __init__(self):
        super().__init__(
            template_id=os.path.abspath(__file__),
            priority=3
        )
        self.header = None  # 物料采购表格表头
        self.data = None  # 物料采购表格数据

    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:
            # 验证物料采购表格（唯一表格）
            purchase_table = tables[0]
            if len(purchase_table) < 2:  # 至少包含表头+1行数据
                return False

            # 物料采购表头正则匹配（按实际位置泛化）
            purchase_header_patterns = [
                r'序号',  # 位置0
                r'中标供应商|成交供应商|中标单位|成交单位',  # 位置1
                r'物料编码|物资编码|材料编码|货品编码',  # 位置2
                r'物料描述|物资描述|材料名称|货品名称|产品名称',  # 位置3
                r'规格型号|规格|型号|技术参数',  # 位置4
                r'计量单位|单位|数量单位',  # 位置5
                r'采购数量|计划数量|需求数量',  # 位置6
                r'中标数量|成交数量|供货数量',  # 位置7
                r'成交单价|中标单价|单价|报价',  # 位置8
                r'到货日期|交货日期|交付日期|到货时间',  # 位置9
                r'交货地点|到货地点|交付地址|交货地址'  # 位置10
            ]

            # 校验每个表头位置的匹配度
            for i, pattern in enumerate(purchase_header_patterns):
                if i >= len(purchase_table[0]):  # 兼容表头长度不足的情况
                    return False
                if not re.fullmatch(pattern, purchase_table[0][i], re.IGNORECASE):
                    # print(f'物料采购表头未验证: 位置{i} 预期[{pattern}] 实际[{purchase_table[0][i]}]')
                    return False

            # 赋值表格数据
            self.header = purchase_table[0]
            self.data = purchase_table[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) < 11:  # 确保行数据长度匹配表头
                continue

            # 按表头位置解析字段
            seq = row[0]
            supplier_name = row[1]
            material_code = row[2]
            product_name = row[3]
            spec = row[4]
            unit = row[5]
            purchase_quantity = row[6]
            bid_quantity = row[7]
            unit_price = row[8]
            arrival_date = row[9]
            delivery_address = row[10]

            # 中标信息必须包含中标人
            if not supplier_name:
                continue

            # 补充单价单位（元）
            price_unit = "元"
            if re.search(r'[万亿美欧日]?元', self.header[8], re.IGNORECASE):
                price_unit = re.search(r'[万亿美欧日]?元', self.header[8], re.IGNORECASE).group(0)
            if unit_price and not re.search('[万亿美欧日]?元', unit_price):
                unit_price += price_unit

            # 组装中标信息（过滤空值）
            bid_info = {
                "标的": product_name,
                "中标人": supplier_name,
                "中标价": unit_price,
                # "地址": delivery_address,
                "服务时间": arrival_date  # 到货日期作为服务/交付时间
            }
            # 过滤空值字段
            bid_info = {k: v for k, v in bid_info.items() if v}
            extract_result["中标信息"].append(bid_info)

        # 提取产品信息（物料采购表格）
        product_list = []
        for row in self.data:
            if len(row) < 11:
                continue

            # 按表头位置解析字段
            seq = row[0]
            supplier_name = row[1]
            material_code = row[2]
            product_name = row[3]
            spec = row[4]
            unit = row[5]
            purchase_quantity = row[6]
            bid_quantity = row[7]
            unit_price = row[8]
            arrival_date = row[9]
            delivery_address = row[10]

            # 补充单价单位
            price_unit = "元"
            if re.search(r'[万亿美欧日]?元', self.header[8], re.IGNORECASE):
                price_unit = re.search(r'[万亿美欧日]?元', self.header[8], re.IGNORECASE).group(0)
            if unit_price and not re.search('[万亿美欧日]?元', unit_price):
                unit_price += price_unit

            # 组装产品信息
            product_info = {
                "产品": product_name,
                "规格": spec,
                "数量": bid_quantity,  # 中标数量作为实际供货数量
                "单位": unit,
                "单价": unit_price
            }

            # 产品信息必须包含产品及至少一个其他要素
            other_fields = [product_info["规格"], product_info["数量"],
                            product_info["单位"], product_info["单价"]]
            if product_info["产品"] and any(other_fields):
                # 过滤空值字段
                product_info = {k: v for k, v in product_info.items() if v}
                product_list.append(product_info)

        extract_result["产品信息"] = product_list

        # 招标信息校验：无招标人/预算则置空
        extract_result["招标信息"] = []

        # 候选人信息：无候选人及排名则置空
        extract_result["候选人信息"] = []

        # 整体过滤空值字段（顶层）
        extract_result = {k: v for k, v in extract_result.items() if v or v == []}

        return extract_result


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

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