BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/DX000002/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 PowerBidTemplate(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]
            # 验证表头列数（目标表格为11列）
            if len(datas[0]) != 11:
                print("表头数量异常；", len(datas[0]))
                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} vs {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 = {
            "招标信息": [],
            "中标信息": [],
            "候选人信息": []
        }

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

            # 解析每行数据（按列顺序对应）
            (seq, subject, subject_budget, builder, section_no,
             section_name, section_budget, limit_price, _, _, _) = row

            # 补充金额单位（从对应表头提取）
            # 标的预算单位补充
            if re.search('([万亿]?元)', self.header[2]) and not re.search('([万亿]?元)', subject_budget):
                unit = re.search('([万亿]?元)', self.header[2]).group(1)
                subject_budget += unit

            # 标段预算单位补充
            if re.search('([万亿]?元)', self.header[6]) and not re.search('([万亿]?元)', section_budget):
                unit = re.search('([万亿]?元)', self.header[6]).group(1)
                section_budget += unit

            # 最高限价单位补充
            if re.search('([万亿]?元)', self.header[7]) and not re.search('([万亿]?元)', limit_price):
                unit = re.search('([万亿]?元)', self.header[7]).group(1)
                # 处理"/"等无数据标识
                limit_price = limit_price if limit_price != "/" else ""
                if limit_price:
                    limit_price += unit

            # 整理招标信息（取标段预算作为主要预算）
            tender_info = {
                "标的": subject.strip(),
                "标包": section_name.strip(),
                "包号": section_no.strip(),
                "招标人": builder.strip(),
                "预算": section_budget.strip()
            }
            # 补充限价（非空时）
            if limit_price and limit_price != "/":
                tender_info["限价"] = limit_price.strip()

            # 验证招标信息有效性（包含招标人或预算）
            if tender_info["招标人"] or tender_info["预算"]:
                extract_result["招标信息"].append(tender_info)

        return extract_result


# 使用示例
if __name__ == "__main__":
    from lxml import etree

    # 读取并解析数据文件
    with open("data.json", "r", encoding="utf8") as f:
        preprocessed_data = json.load(f)

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