BIDI_EXTRACTION_AI/BiddingKG/dl/template_extract/templates/XX0182/template3.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
        )
        self.main_table = None  # 第一个表格，存储所有需提取信息

    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:
            print("表格数量异常，至少需要1个表格")
            return False

        try:
            self.main_table = tables[0]  # 只处理第一个表格

            # 验证表格关键表头（按data.json实际位置泛化匹配）
            header_patterns = [
                (0, 0, r'项目编号'),  # 第1行第1列
                (0, 2, r'项目名称|工程名称'),  # 第1行第3列
                (1, 0, r'经办人单位|采购单位|招标人单位'),  # 第2行第1列
                (1, 2, r'经办人|联系人'),  # 第2行第3列
                (2, 0, r'预算金额|招标金额'),  # 第3行第1列
                (2, 2, r'成交金额|中标金额'),  # 第3行第3列
                (3, 0, r'成交供应商|中标人'),  # 第4行第1列
                (5, 0, r'中标人地址|联系地址'),  # 第6行第1列
                (5, 2, r'采购单位|招标人|采购人'),  # 第6行第3列
                (8, 0, r'供应商联系手机|中标人电话|联系方式'),  # 第9行第1列
            ]

            # 验证表头
            for row, col, pattern in header_patterns:
                if row >= len(self.main_table) or col >= len(self.main_table[row]):
                    return False
                header_text = self.main_table[row][col].strip()
                if not re.search(pattern, header_text, re.IGNORECASE):
                    print(f"表头不匹配: {header_text} 不符合 {pattern}")
                    return False

            return True
        except Exception as e:
            print(f"表格验证错误: {e}")
            return False

    def extract(self, preprocessed_data: Dict[str, Any]) -> Dict[str, Any]:
        """提取表格中的各类信息"""
        result = {}

        # 提取项目基本信息
        result["项目编号"] = self._get_value(self.main_table, 0, 1)  # 第1行第2列
        result["项目名称"] = self._get_value(self.main_table, 0, 3)  # 第1行第4列

        # 提取候选人信息（无候选人信息）
        candidate_info_list = self._extract_candidate_info()
        if candidate_info_list:
            result["候选人信息"] = candidate_info_list
        else:
            # 无候选人信息时提取中标信息
            winner_info = self._extract_winner_info()
            if winner_info:
                result["中标信息"] = [winner_info]

        # 提取招标信息
        tender_info = self._extract_tender_info()
        if tender_info:
            result["招标信息"] = [tender_info]

        # 过滤空值字段
        return {k: v for k, v in result.items() if v or v == []}

    def _get_value(self, table: list, row: int, col: int) -> str:
        """获取表格指定位置的值"""
        if row < len(table) and col < len(table[row]):
            return table[row][col].strip()
        return ""

    def _format_amount(self, value: str, header_row: int, header_col: int) -> str:
        """格式化金额，补充单位"""
        if not value:
            return ""
        # 检查值中是否已包含单位
        if re.search(r'[万亿美欧日]?元|￥|人民币|美元|欧元', value):
            return value
        # 从表头获取单位补充
        header_text = self._get_value(self.main_table, header_row, header_col)
        if re.search(r'[万亿美欧日]?元|￥|人民币|美元|欧元', header_text):
            unit = re.search(r'[万亿美欧日]?元|￥|人民币|美元|欧元', header_text).group(0)
            return f"{value}{unit}"
        return value

    def _extract_tender_info(self) -> Dict[str, Any]:
        """提取招标信息（招标人相关信息）"""
        tender = {
            "标的": self._get_value(self.main_table, 0, 3),  # 项目名称作为标的
            "招标人": self._get_value(self.main_table, 5, 3),  # 采购单位（第6行第3列）
            "预算": self._format_amount(self._get_value(self.main_table, 2, 1), 2, 0),  # 预算金额
            "联系人": self._get_value(self.main_table, 1, 3),  # 经办人（招标人联系人）
        }
        # 必须包含招标人或预算
        if not tender.get("招标人") and not tender.get("预算"):
            return {}
        # 过滤空值
        return {k: v for k, v in tender.items() if v}

    def _extract_winner_info(self) -> Dict[str, Any]:
        """提取中标信息（中标人相关信息）"""
        winner = {
            "标的": self._get_value(self.main_table, 0, 3),  # 项目名称作为标的
            "中标人": self._get_value(self.main_table, 3, 1),  # 成交供应商（中标人）
            "中标价": self._format_amount(self._get_value(self.main_table, 2, 3), 2, 2),  # 成交金额
            "电话": self._get_value(self.main_table, 8, 1),  # 供应商联系手机
            "地址": self._get_value(self.main_table, 5, 1),  # 联系地址（中标人地址）
        }
        # 必须包含中标人
        if not winner.get("中标人"):
            return {}
        # 过滤空值
        return {k: v for k, v in winner.items() if v}

    def _extract_candidate_info(self) -> list:
        """提取候选人信息（表格中无相关信息）"""
        return []  # 表格中无候选人及排名相关信息


# 使用示例
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("当前数据不满足模板调用条件")