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

            # 表头内容正则匹配（泛化类似表达）
            header_patterns = [
                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(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 = {"候选人信息": []}

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

            # 解析每行数据
            seq, subject, package, ratio, rank, candidate, price, quality, period, qualification = row

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

            # 整理候选人信息
            candidate_info = {
                "标的": subject,
                "标包": package,
                "候选人": candidate,
                "投标价": price,
                "排名": rank
            }

            extract_result["候选人信息"].append(candidate_info)

        # 按照要求，有候选人信息时不返回中标信息
        # 检查招标信息是否满足条件（包含招标人或预算），此处表格无相关信息，故不添加
        return extract_result


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

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