BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/DX000002/template6.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 CustomTemplate(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]
            # 验证表头列数（data.json中为11列）
            if len(datas[0]) != 11:
                print(f"表头数量异常；实际{len(datas[0])}列，预期11列")
                return False

            # 表头内容正则匹配（按实际位置泛化）
            header_patterns = [
                '序号',  # 第1列：序号
                '标的(名称)?',  # 第2列：标的名称
                '最大中标数量',  # 第3列：最大中标数量
                '划分包数',  # 第4列：划分包数
                '标包(序号|号)?',  # 第5列：标包序号
                '标包名称',  # 第6列：标包名称
                '标包金额（万元）',  # 第7列：标包金额
                '最高限价（万元）',  # 第8列：最高限价
                '是否特殊包',  # 第9列：是否特殊包
                '响应保证金',  # 第10列：响应保证金
                '备注',  # 第11列：备注
            ]

            # 逐列验证表头
            for i, pattern in enumerate(header_patterns):
                if not re.fullmatch(pattern, datas[0][i]):
                    print(f'表头未验证：位置{i+1}，预期{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][1]

        # 提取招标信息
        for row in self.data:
            # 跳过列数异常的行
            if len(row) != 11:
                continue

            # 解析每行数据（按实际列位置）
            seq, subject, max_win_num, package_count, package_num, package_name, \
                budget, limited_price, is_special, deposit, remark = row

            # 补充金额单位（表头已明确万元）
            if budget and not re.search('万元', budget):
                budget += '万元'
            if limited_price and not re.search('万元', limited_price):
                limited_price += '万元'

            # 构建招标信息条目
            tender_info = {
                "标的": subject,
                "标包": package_name,
                "预算": budget,
                "限价": limited_price
            }

            # 检查招标信息有效性（必须包含预算）
            if "预算" in tender_info:
                extract_result["招标信息"].append(tender_info)

        # 处理无有效招标信息的情况
        if not extract_result["招标信息"]:
            extract_result["招标信息"] = []

        return extract_result


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

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