BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/XX7523/template2.py

# coding:utf8
import re
from typing import Dict, Any
from BiddingKG.dl.template_extract.abstract_template import AbstractTemplate
import os
import json


class CustomTransactionTemplate(AbstractTemplate):
    """
    适配data.json表格结构（供应商/商品/货号/规格/单价/数量/总价）的成交信息提取模板
    提取产品信息、中标信息，严格遵循返回格式要求
    """

    def __init__(self):
        super().__init__(
            template_id=os.path.abspath(__file__),
            priority=3
        )
        self.table_header = None  # 表格表头
        self.table_data = None  # 表格数据行

    def check_call_timing(self, preprocessed_data: Dict[str, Any]) -> bool:
        """
        检查是否符合模板调用条件：匹配表格结构和核心表头（按data.json实际表头位置）
        :param preprocessed_data: 预处理后的输入数据（包含表格列表）
        :return: 是否符合调用条件
        """
        # 检查表格列表是否存在且有效
        if "表格列表" not in preprocessed_data:
            return False
        tables = preprocessed_data["表格列表"]
        if len(tables) < 1 or not isinstance(tables[0], list) or len(tables[0]) < 2:
            return False

        try:
            # 提取第一个表格的表头和数据
            target_table = tables[0]
            self.table_header = target_table[0]
            self.table_data = target_table[1:]

            # 定义核心表头的泛化正则匹配规则（按data.json实际表头位置）
            # 位置0:供应商 位置1:商品名称 位置2:货号 位置3:规格 位置4:单价 位置5:数量 位置6:总价
            header_patterns = [
                r'供应商|供应商名称|成交供应商|中标供应商|供货单位',  # 第0列：供应商
                r'商品名称|产品名称|货品名称|物料名称|品名',  # 第1列：商品名称
                r'货号|物料编码|产品编码|商品编号',  # 第2列：货号
                r'规格|型号|规格型号|技术参数',  # 第3列：规格
                r'单价|单价金额|单价（元）|单价（￥）',  # 第4列：单价
                r'数量|数量值|采购数量|供货数量',  # 第5列：数量
                r'总价|总金额|合计金额|金额总计'  # 第6列：总价
            ]

            # 验证表头匹配（按位置校验，必须全部匹配）
            for idx, pattern in enumerate(header_patterns):
                if idx >= len(self.table_header):
                    return False
                header_str = str(self.table_header[idx]).strip()
                if not re.fullmatch(pattern, header_str, re.IGNORECASE):
                    print(f"表头位置{idx}匹配失败: 预期{pattern}，实际{header_str}")
                    return False

            return True
        except Exception as e:
            print(f"校验表格失败: {e}")
            return False

    def extract(self, preprocessed_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        提取表格数据，按指定格式返回（过滤空值、校验字段有效性）
        :param preprocessed_data: 预处理后的输入数据
        :return: 提取结果（符合指定JSON格式）
        """
        # 初始化返回结果结构
        extract_result = {
            "项目编号": "",
            "项目名称": "",
            "招标信息": [],
            "中标信息": [],
            "候选人信息": [],
            "产品信息": []
        }

        # 提取核心数据（按data.json表头位置）
        for row in self.table_data:
            if len(row) != len(self.table_header):
                continue  # 跳过长度不匹配的行

            # 按表头位置提取字段（先做基础清洗）
            supplier_name = str(row[0]).strip()  # 供应商（中标人）
            product_name = str(row[1]).strip()  # 商品名称
            product_spec = str(row[3]).strip()  # 规格
            unit_price = str(row[4]).strip()  # 单价
            quantity = str(row[5]).strip()  # 数量
            total_price = str(row[6]).strip()  # 总价

            # ========== 1. 提取中标信息（必须包含中标人） ==========
            if supplier_name:
                # 处理单价/总价的单位补充（优先从表头提取，无则默认加"元"）
                # 补充单价单位
                price_header = str(self.table_header[4]).strip()
                unit_match = re.search(r'([万亿美欧日]?元|￥)', price_header, re.IGNORECASE)
                if unit_match and unit_price and not re.search(unit_match.group(1), unit_price, re.IGNORECASE):
                    unit_price += unit_match.group(1).replace("￥", "元")
                elif unit_price and not re.search(r'[万亿美欧日]?元|￥', unit_price, re.IGNORECASE):
                    unit_price += "元"

                # 补充总价单位
                total_header = str(self.table_header[6]).strip()
                total_unit_match = re.search(r'([万亿美欧日]?元|￥)', total_header, re.IGNORECASE)
                if total_unit_match and total_price and not re.search(total_unit_match.group(1), total_price,
                                                                      re.IGNORECASE):
                    total_price += total_unit_match.group(1).replace("￥", "元")
                elif total_price and not re.search(r'[万亿美欧日]?元|￥', total_price, re.IGNORECASE):
                    total_price += "元"

                # 构建中标信息（中标价取总价，无总价则取单价*数量）
                bid_price = total_price if total_price else (f"{float(unit_price.replace('元', '')) * float(quantity)}元"
                                                             if unit_price and quantity and unit_price.replace('元',
                                                                                                               '').replace(
                    '.', '').isdigit() and quantity.replace('.', '').isdigit() else "")

                bid_info = {
                    "标的": "",
                    "标包": "",
                    "包号": "",
                    "中标人": supplier_name,
                    "中标价": bid_price,
                    "联系人": "",
                    "电话": "",
                    "服务时间": "",
                    "地址": ""
                }
                # 过滤空值字段
                bid_info_filtered = {k: v for k, v in bid_info.items() if v}
                if bid_info_filtered:  # 确保非空才添加
                    extract_result["中标信息"].append(bid_info_filtered)

            # ========== 2. 提取产品信息（必须包含产品+至少一个其他要素） ==========
            if product_name:
                product_info = {
                    "产品": product_name,
                    "品目": "",
                    "品牌": "",
                    "规格": product_spec,
                    "数量": quantity,
                    "单位": "",
                    "单价": unit_price,
                    "总价": total_price
                }
                # 过滤空值字段
                product_info_filtered = {k: v for k, v in product_info.items() if v}
                # 验证：必须包含产品 + 至少一个其他要素
                if len(product_info_filtered) >= 2:
                    extract_result["产品信息"].append(product_info_filtered)

        # ========== 3. 处理其他字段（无数据则置空） ==========
        # 招标信息：无招标人/预算，置空
        extract_result["招标信息"] = []
        # 候选人信息：无候选人+排名，置空
        extract_result["候选人信息"] = []

        # ========== 4. 过滤顶层空值字段 ==========
        extract_result = {k: v for k, v in extract_result.items() if v or isinstance(v, list)}

        return extract_result


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

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