BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/00198/template1.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 BidInfoExtractTemplate(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]
            # 至少需要标题行、列头行、一行数据（共3行）
            if len(datas) < 3:
                print(f"表格行数异常：{len(datas)}行，预期至少3行")
                return False

            # ========== 1. 匹配第一行标题行 ==========
            title_row = datas[0]
            # 标题行泛化匹配规则：所有列均为“中标单位信息”相关表达（支持重复）
            title_pattern = r'中标单位信息|中标信息|中标结果|中标公示'
            title_match = all([re.fullmatch(title_pattern, cell) for cell in title_row])
            if not title_match:
                print(f"第一行标题行未匹配：{title_row}")
                return False

            # ========== 2. 匹配第二行列头行 ==========
            header_row = datas[1]
            # 列头行需为7列
            if len(header_row) != 7:
                print(f"第二行列头行列数异常：{len(header_row)}列，预期7列")
                return False

            # 列头行正则匹配规则（泛化类似表达）
            header_patterns = [
                r'序号|编号',  # 第1列：序号
                r'单位名称|公司名称|中标单位名称',  # 第2列：单位名称
                r'项目经理|项目负责人|联系人',  # 第3列：项目经理
                r'资格等级|资质等级|资质级别',  # 第4列：资格等级
                r'资格证书编号|资质证书编号|证书编号',  # 第5列：资格证书编号
                r'中标价（万元）|中标价|成交价|中标金额(（万?元）)?',  # 第6列：中标价
                r'中标工期（天）|工期|中标工期|服务期|交货期(（天）)?'  # 第7列：中标工期
            ]

            # 逐列验证列头行
            for i, pattern in enumerate(header_patterns):
                if not re.fullmatch(pattern, header_row[i]):
                    print(f'第二行列头行第{i+1}列未匹配：{pattern} vs {header_row[i]}')
                    return False

            # 保存表头和数据行（数据行从第三行开始）
            self.title_header = title_row
            self.col_header = header_row
            self.data = datas[2:]
            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) != 7:
                continue

            # 解析每行数据
            seq, company_name, project_manager, qualification_level, cert_no, price, period = row

            # 补充中标价单位（如果表头有单位而内容没有）
            if price:  # 仅当价格非空时补充单位
                price_pattern = re.search(r'[万亿美欧日]?元', self.col_header[5])
                if price_pattern and not re.search(r'[万亿美欧日]?元', price):
                    price += price_pattern.group(0)
            if period:
                period_pattern = re.search(r'日|天|年|月|周|星期', self.col_header[6])
                if period_pattern and not re.search(r'日|天|年|月|周|星期', period):
                    period += period_pattern.group(0)

            # 整理中标信息
            bid_info = {
                "标的": "",
                "标包": "",
                "包号": "",
                "中标人": company_name,
                "中标价": price,
                "联系人": project_manager,
                "电话": "",
                "服务时间": period,
                "地址": ""
            }

            # 过滤空值字段
            bid_info = {k: v for k, v in bid_info.items() if v}

            # 中标信息必须包含中标人，满足则添加
            if bid_info.get("中标人"):
                extract_result["中标信息"].append(bid_info)

        # 过滤空列表和空值字段
        extract_result = {k: v for k, v in extract_result.items() if v or (isinstance(v, list) and len(v) > 0)}

        return extract_result


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

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