from pathlib import Path from typing import List, Tuple, Dict, Any, Optional, Literal import re import json import logging import sys import pandas as pd # 尝试使用统一的日志系统 try: from ..utils.logging_config import get_logger logger = get_logger("pdf_converter_v2.utils.table_extractor") except (ImportError, ValueError): # 如果无法导入，使用标准 logging logger = logging.getLogger(__name__) # 确保日志系统正确配置 if not logger.handlers: handler = logging.StreamHandler(sys.stderr) handler.setFormatter(logging.Formatter('%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d - %(message)s')) logger.addHandler(handler) logger.setLevel(logging.INFO) try: import pdfplumber _PDFPLUMBER_AVAILABLE = True except ImportError as e: _PDFPLUMBER_AVAILABLE = False _PDFPLUMBER_IMPORT_ERROR = str(e) except Exception as e: # 捕获其他可能的导入错误 _PDFPLUMBER_AVAILABLE = False _PDFPLUMBER_IMPORT_ERROR = str(e) else: _PDFPLUMBER_IMPORT_ERROR = None # 文档类型 -> 表头规则 TABLE_TYPE_RULES: Dict[str, List[dict]] = { # 结算报告类 "settlementReport": [ { "name": "审定结算汇总表", "keywords": ["序号", "审计内容", "送审金额（含税）", "审定金额（含税）", "审定金额（不含税）", "增减金额", "备注"], "match_mode": "all", }, { "name": "合同执行情况", "keywords": ["施工单位", "中标通知书金额", "中标通知书编号", "合同金额", "结算送审金额", "差额"], "match_mode": "all", }, { "name": "赔偿合同", "keywords": ["合同对方", "赔偿事项", "合同金额", "结算送审金额", "差额"], "match_mode": "all", }, { "name": "物资采购合同1", "keywords": ["物料名称", "合同数量", "施工图数量", "单价（不含税）", "差额"], "match_mode": "all", }, { "name": "物资采购合同2", "keywords": ["物料名称", "合同金额（不含税）", "入账金额", "差额", "备注"], "match_mode": "all", }, { "name": "其他服务类合同", "keywords": ["服务商", "中标通知书", "合同金额", "送审金额", "结算金额"], "match_mode": "all", }, ], # 初设评审类 "designReview": [ { "name": "初设评审的概算投资", "keywords": ["序号", "工程名称", "建设规模", "静态投资", "其中：建设场地征用及清理费", "动态投资"], "match_mode": "all", }, { "name": "初设评审的概算投资明细", "keywords": ["序号", "工程或费用名称", "建筑工程费", "设备购置费", "安装工程费", "其他费用", "合计"], "match_mode": "all", }, { "name": "初设评审的概算投资费用", "keywords": ["序号", "工程或费用名称", "费用金额", "各项占静态投资", "单位投资"], "match_mode": "all", }, ], # 决算报告类 "finalAccount": [ { "name": "单项工程投资完成情况", "keywords": ["费用项目", "概算金额", "决算金额", "审定金额", "增值税额", "超", "节"], "match_mode": "all", }, ], } EXCLUDE_RULES: List[str] = [] # 是否启用跨页合并 ENABLE_MERGE_CROSS_PAGE_TABLES: bool = True def extract_table_title_from_page(page, table_bbox: tuple, max_lines: int = 10) -> str: """ 从 PDF 页面中提取表格上方的文本作为表格标题。 优先查找包含工程名称的行（如 "XXX变电站新建工程总概算表"）。 Args: page: pdfplumber page 对象 table_bbox: 表格的边界框 (x0, top, x1, bottom) max_lines: 最多向上搜索的行数 Returns: str: 表格标题，如果未找到则返回空字符串 """ if not table_bbox: return "" table_top = table_bbox[1] # 表格顶部 y 坐标 page_width = page.width # 在表格上方区域搜索文本（向上搜索到页面顶部） search_top = 0 # 从页面顶部开始搜索 search_bottom = table_top - 5 # 留一点边距 if search_bottom <= search_top: return "" # 提取表格上方区域的文本 crop_box = (0, search_top, page_width, search_bottom) try: cropped = page.within_bbox(crop_box) text = cropped.extract_text() or "" except Exception: return "" if not text.strip(): return "" # 取所有行 lines = [line.strip() for line in text.strip().split('\n') if line.strip()] if not lines: return "" # 排除的关键词（这些不是工程名称） exclude_keywords = ["工程规模", "金额单位", "建设规模", "附表", "附件"] def is_valid_title(line: str) -> bool: """检查是否是有效的工程名称标题""" # 排除包含排除关键词的行 if any(kw in line for kw in exclude_keywords): return False # 必须包含工程相关关键词 if not any(kw in line for kw in ["工程", "概算表", "估算表"]): return False return True # 优先查找包含 "总概算表" 的行（最可能是工程名称） for line in reversed(lines): if "总概算表" in line and is_valid_title(line): # 进一步检查是否包含具体工程类型 if any(kw in line for kw in ["变电站", "线路", "间隔", "kV", "KV", "千伏"]): return line.strip() # 次优先：查找包含 "概算表" 但不是 "总概算表" 的行 for line in reversed(lines): if "概算表" in line and "总概算表" not in line and is_valid_title(line): if any(kw in line for kw in ["变电站", "线路", "间隔", "kV", "KV", "千伏"]): return line.strip() # 再次优先：查找包含 "变电站"/"线路"/"间隔" + "工程" 的行 for line in reversed(lines): if is_valid_title(line): if any(kw in line for kw in ["变电站", "线路", "间隔"]) and "工程" in line: return line.strip() # 查找包含电压等级 + "工程" 的行 for line in reversed(lines): if is_valid_title(line): if any(kw in line for kw in ["kV", "KV", "千伏", "kv"]) and "工程" in line: return line.strip() # 最后：返回任何包含工程关键词的有效行 for line in reversed(lines): if is_valid_title(line): return line.strip() return "" def extract_tables_with_pdfplumber( pdf_path: str, pages: str = "all", extract_titles: bool = False, ) -> List[Tuple[int, pd.DataFrame, tuple, str]]: """ 使用 pdfplumber 提取 PDF 中的表格。 Args: pdf_path: PDF 文件路径 pages: 页码范围，如 "all" 或 "1-5,7,9-10" extract_titles: 是否提取表格标题 Returns: List[Tuple[int, pd.DataFrame, tuple, str]]: [(页码, DataFrame, bbox, title), ...] 如果 extract_titles=False，title 为空字符串 """ # 运行时再次尝试导入，因为模块加载时可能失败但运行时环境可能不同 global _PDFPLUMBER_AVAILABLE, pdfplumber, _PDFPLUMBER_IMPORT_ERROR if not _PDFPLUMBER_AVAILABLE: try: import pdfplumber _PDFPLUMBER_AVAILABLE = True _PDFPLUMBER_IMPORT_ERROR = None except ImportError as e: error_msg = f"pdfplumber 库未安装，无法提取表格（请安装 pdfplumber）" if _PDFPLUMBER_IMPORT_ERROR: error_msg += f"\n模块加载时导入错误: {_PDFPLUMBER_IMPORT_ERROR}" error_msg += f"\n运行时导入错误: {e}" raise RuntimeError(error_msg) except Exception as e: error_msg = f"pdfplumber 库导入失败: {e}" if _PDFPLUMBER_IMPORT_ERROR: error_msg += f"\n模块加载时导入错误: {_PDFPLUMBER_IMPORT_ERROR}" raise RuntimeError(error_msg) tables_data: List[Tuple[int, pd.DataFrame, tuple, str]] = [] with pdfplumber.open(pdf_path) as pdf: total_pages = len(pdf.pages) logger.info(f"[pdfplumber] 打开 PDF，共 {total_pages} 页") # 确定要处理的页面 if pages == "all": pages_to_process = pdf.pages logger.info(f"[pdfplumber] 处理所有页面: 1-{total_pages}") else: # 解析页面范围（如 "1-5,7,9-10"） page_numbers: List[int] = [] for part in pages.split(","): part = part.strip() if not part: continue if "-" in part: start, end = map(int, part.split("-")) page_numbers.extend(range(start, end + 1)) else: page_numbers.append(int(part)) pages_to_process = [ pdf.pages[i - 1] for i in page_numbers if 0 < i <= len(pdf.pages) ] logger.info(f"[pdfplumber] 处理指定页面: {page_numbers}") # 提取每一页的表格 for page in pages_to_process: page_num = page.page_number table_settings = { "vertical_strategy": "lines", "horizontal_strategy": "lines", "intersection_tolerance": 3, "min_words_vertical": 1, "min_words_horizontal": 1, } tables = page.find_tables(table_settings=table_settings) page_table_count = 0 for table in tables: table_data = table.extract() if table_data and len(table_data) > 0: df = pd.DataFrame(table_data) bbox = table.bbox # (x0, top, x1, bottom) # 提取表格标题 title = "" if extract_titles: title = extract_table_title_from_page(page, bbox) if title: logger.debug(f"[pdfplumber] 页面 {page_num}: 表格标题: {title}") tables_data.append((page_num, df, bbox, title)) page_table_count += 1 if page_table_count > 0: logger.debug(f"[pdfplumber] 页面 {page_num}: 提取到 {page_table_count} 个表格") logger.info(f"[pdfplumber] 提取完成: 共提取到 {len(tables_data)} 个表格") return tables_data def check_table_header(table_df: pd.DataFrame, rule: dict) -> Tuple[bool, str]: """ 检查表格是否匹配指定的表头规则（固定匹配：必须包含所有关键词） 处理表头换行（多行表头）并对关键字做去空格匹配。 """ if table_df.empty: return False, "" rule_name = rule.get("name", "未知规则") if rule_name in EXCLUDE_RULES: return False, "" # 默认从第一行开始 start_row = 0 # 合并前几行作为完整表头（通常表头可能占 1-3 行） header_rows_to_check = min(3, len(table_df) - start_row) header_text_parts: List[str] = [] for row_idx in range(start_row, start_row + header_rows_to_check): if row_idx >= len(table_df): break row = table_df.iloc[row_idx].astype(str).str.strip() for cell in row: cell_text = str(cell).strip() if cell_text and cell_text.lower() not in ["nan", "none", ""]: # 将单元格内的换行符替换为空格（处理 xlsx 中的换行） cell_text = cell_text.replace("\n", " ").replace("\r", " ") header_text_parts.append(cell_text) # 合并所有表头文本 header_text = " ".join(header_text_parts) # 规范空白：多个空格合并 header_text = re.sub(r"\s+", " ", header_text).strip() # 去掉所有空白，用于处理“中标通知\n书金额”这类情况 header_text_no_space = re.sub(r"\s+", "", header_text) keywords = rule.get("keywords", []) match_mode = rule.get("match_mode", "all") if not keywords: return False, "" if match_mode == "all": all_match = True for keyword in keywords: keyword_no_space = re.sub(r"\s+", "", keyword) if keyword in header_text or keyword_no_space in header_text_no_space: continue all_match = False break if all_match: return True, rule_name elif match_mode == "any": for keyword in keywords: keyword_no_space = re.sub(r"\s+", "", keyword) if keyword in header_text or keyword_no_space in header_text_no_space: return True, rule_name return False, "" def is_likely_header_only(table_df: pd.DataFrame, min_data_rows: int = 1) -> bool: """ 判断表格是否可能只包含表头（数据行很少）。 """ if table_df.empty: return True header_rows = min(3, len(table_df)) return len(table_df) <= header_rows + min_data_rows def has_similar_structure(table1_df: pd.DataFrame, table2_df: pd.DataFrame, tolerance: int = 1) -> bool: """ 判断两个表格是否有相似的结构（列数接近）。 """ if table1_df.empty or table2_df.empty: return False cols1 = len(table1_df.columns) cols2 = len(table2_df.columns) return abs(cols1 - cols2) <= tolerance def _merge_all_tables( tables: List[Tuple[int, pd.DataFrame, int]], ) -> List[Tuple[int, pd.DataFrame, int]]: """ 合并所有跨页表格（不进行过滤）： - 只处理"表头在当前页，内容在下一页"的典型情况； - 严格限制只合并相邻页，且列结构相似。 """ if not ENABLE_MERGE_CROSS_PAGE_TABLES or not tables: return tables # page -> [(idx, df)] page_map: Dict[int, List[Tuple[int, pd.DataFrame]]] = {} for orig_idx, df, page in tables: page_map.setdefault(page, []).append((orig_idx, df)) merged: List[Tuple[int, pd.DataFrame, int]] = [] processed: set[int] = set() sorted_pages = sorted(page_map.keys()) for page in sorted_pages: current_list = page_map[page] for orig_idx, df in current_list: if orig_idx in processed: continue current_df = df did_merge = False # 情况：当前表格只有表头，尝试合并下一页 if is_likely_header_only(current_df): next_page = page + 1 if next_page in page_map: for next_orig_idx, next_df in page_map[next_page]: if next_orig_idx in processed: continue if not has_similar_structure(current_df, next_df): continue # 合并：保留当前页表头，拼接下一页数据 header_rows = min(3, len(current_df)) header_df = current_df.iloc[:header_rows].copy() next_data_df = next_df.copy() # 对齐列数 if len(header_df.columns) != len(next_data_df.columns): if len(next_data_df.columns) < len(header_df.columns): for _ in range(len(next_data_df.columns), len(header_df.columns)): next_data_df[len(next_data_df.columns)] = "" merged_df = pd.concat([header_df, next_data_df], ignore_index=True) merged.append((orig_idx, merged_df, page)) processed.add(orig_idx) processed.add(next_orig_idx) did_merge = True break if not did_merge and orig_idx not in processed: merged.append((orig_idx, current_df, page)) processed.add(orig_idx) return merged def _merge_cross_page_tables( tables: List[Tuple[int, pd.DataFrame, str, int, str]], header_rules: List[dict], ) -> List[Tuple[int, pd.DataFrame, str, int, str]]: """ 简化版跨页合并逻辑（用于已匹配规则的表格）： - 只处理"表头在当前页，内容在下一页"的典型情况； - 严格限制只合并相邻页，且列结构相似； - 如果下一页第一行看起来像新的表头，则不合并。 元组格式: (orig_idx, df, rule_name, page, title) """ if not ENABLE_MERGE_CROSS_PAGE_TABLES or not tables: return tables # page -> [(idx, df, rule_name, title)] page_map: Dict[int, List[Tuple[int, pd.DataFrame, str, str]]] = {} for orig_idx, df, rule_name, page, title in tables: page_map.setdefault(page, []).append((orig_idx, df, rule_name, title)) merged: List[Tuple[int, pd.DataFrame, str, int, str]] = [] processed: set[int] = set() sorted_pages = sorted(page_map.keys()) for page in sorted_pages: current_list = page_map[page] for orig_idx, df, rule_name, title in current_list: if orig_idx in processed: continue current_df = df did_merge = False # 情况：当前表格只有表头，尝试合并下一页 if is_likely_header_only(current_df): next_page = page + 1 if next_page in page_map: for next_orig_idx, next_df, next_rule_name, next_title in page_map[next_page]: if next_orig_idx in processed: continue if rule_name and next_rule_name and rule_name != next_rule_name: continue if not has_similar_structure(current_df, next_df): continue # 判断下一页第一行是否像表头 next_first_row_text = "" if not next_df.empty: next_first_row_text = " ".join( next_df.iloc[0].astype(str).str.strip().tolist() ) keyword_count = 0 if header_rules and rule_name: for rule in header_rules: if rule.get("name") == rule_name: kws = rule.get("keywords", []) keyword_count = sum( 1 for kw in kws if kw in next_first_row_text ) break # 如果下一页第一行包含较多关键词，认为是新表头，不合并 if keyword_count >= 2: continue # 合并：保留当前页表头，拼接下一页数据 header_rows = min(3, len(current_df)) header_df = current_df.iloc[:header_rows].copy() next_data_df = next_df.copy() # 对齐列数 if len(header_df.columns) != len(next_data_df.columns): if len(next_data_df.columns) < len(header_df.columns): for _ in range(len(next_data_df.columns), len(header_df.columns)): next_data_df[len(next_data_df.columns)] = "" merged_df = pd.concat([header_df, next_data_df], ignore_index=True) merged.append((orig_idx, merged_df, rule_name, page, title)) processed.add(orig_idx) processed.add(next_orig_idx) did_merge = True break if not did_merge and orig_idx not in processed: merged.append((orig_idx, current_df, rule_name, page, title)) processed.add(orig_idx) return merged def _fix_broken_cells(table_df: pd.DataFrame, header_row_count: int = 1) -> pd.DataFrame: """ 修复被错误分割的单元格（一个单元格的内容被识别成多行） 检测规则： 1. 如果某一行的前N列有内容，但后面的列大部分为空（超过50%） 2. 且上一行对应列有内容，则认为当前行是上一行的延续，需要合并 Args: table_df: 表格DataFrame header_row_count: 表头行数，跳过表头不处理 Returns: pd.DataFrame: 修复后的表格 """ if table_df.empty or len(table_df) <= header_row_count: return table_df df = table_df.copy() rows_to_remove = [] # 从表头后开始检查 for i in range(header_row_count, len(df)): if i <= header_row_count: continue # 跳过第一行数据 # 获取当前行 current_row = df.iloc[i] prev_row = df.iloc[i-1] # 统计有内容的列和空列 non_empty_cols = [] empty_cols = [] for j in range(len(df.columns)): val = current_row.iloc[j] val_str = str(val).strip() is_empty = (val is None or pd.isna(val) or not val_str or val_str.lower() in ['nan', 'none', '', '0', '0.0']) if is_empty: empty_cols.append(j) else: non_empty_cols.append(j) # 如果没有非空列，跳过 if not non_empty_cols: continue # 计算空列比例 empty_ratio = len(empty_cols) / len(df.columns) if len(df.columns) > 0 else 0 # 如果空列超过50%，可能是被截断的行 if empty_ratio > 0.5: # 检查上一行对应位置是否有内容 can_merge = True merge_cols = [] for col_idx in non_empty_cols: prev_val = prev_row.iloc[col_idx] prev_val_str = str(prev_val).strip() curr_val = current_row.iloc[col_idx] curr_val_str = str(curr_val).strip() # 如果上一行对应列有内容，可以合并 if prev_val_str and prev_val_str.lower() not in ['nan', 'none', '', '0', '0.0']: merge_cols.append(col_idx) else: # 如果上一行对应列为空，但当前行有内容，可能是新行，不合并 # 但如果只有前几列有内容，且都是文本（不是数字），可能是延续 if col_idx < len(df.columns) * 0.5: # 前50%的列 # 检查是否是文本（不是纯数字） if not curr_val_str.replace('.', '').replace('-', '').isdigit(): merge_cols.append(col_idx) else: can_merge = False break else: can_merge = False break if can_merge and merge_cols: # 合并每个非空列到上一行对应的列 for col_idx in merge_cols: prev_val = str(prev_row.iloc[col_idx]).strip() curr_val = str(current_row.iloc[col_idx]).strip() # 合并（移除换行符） merged_val = prev_val + curr_val.replace('\n', '').replace('\r', '') df.iloc[i-1, col_idx] = merged_val rows_to_remove.append(i) logger.debug(f"[跨行合并] 合并行 {i} 到行 {i-1}，列: {merge_cols}") # 删除已合并的行 if rows_to_remove: df = df.drop(rows_to_remove).reset_index(drop=True) logger.info(f"[跨行合并] 修复了 {len(rows_to_remove)} 个被错误分割的单元格（跨行）") return df def _format_header_text(cell_val: str) -> str: """ 格式化表头文本：移除换行符和多余空格，用于匹配。 Args: cell_val: 原始单元格值 Returns: 格式化后的文本（移除所有空格，用于精确匹配） """ if not cell_val or str(cell_val).lower() in ['nan', 'none', '']: return "" # 清理换行符，直接移除 text = str(cell_val).strip().replace('\n', '').replace('\r', '') # 移除所有空格，用于匹配 text = re.sub(r'\s+', '', text) return text def _detect_header_rows(df: pd.DataFrame, header_row_idx: int, header_keywords: List[str]) -> int: """ 智能检测表头行数，避免把数据行当作表头。 Args: df: 数据框 header_row_idx: 已识别的表头行索引 header_keywords: 表头关键词列表 Returns: 表头行数（从 header_row_idx 开始） """ header_rows_to_check = 1 # 检查后续行是否是表头的延续 for i in range(header_row_idx + 1, min(header_row_idx + 3, len(df))): row = df.iloc[i] row_text = " ".join(row.astype(str).str.strip().tolist()) # 检查是否包含表头关键词 keyword_count = sum(1 for kw in header_keywords if kw in row_text) # 如果包含关键词，进一步判断 if keyword_count >= 2: # 至少包含2个关键词才可能是表头 # 检查是否主要是数字（如果是，则不是表头） numeric_count = sum(1 for cell in row if str(cell).strip().replace('.', '').replace('-', '').isdigit()) numeric_ratio = numeric_count / len(row) if len(row) > 0 else 0 # 检查是否包含明显的公司名称、人名等（如果是，则不是表头） has_company_name = any( '公司' in str(cell) or '有限' in str(cell) or '股份' in str(cell) or '工程' in str(cell) or '集团' in str(cell) or '局' in str(cell) for cell in row ) # 如果数字占比小于30%且不包含公司名称，可能是表头 if numeric_ratio < 0.3 and not has_company_name: header_rows_to_check += 1 else: # 包含公司名称或数字占比高，是数据行，停止 break else: # 关键词太少，不是表头，停止 break return header_rows_to_check def parse_settlement_summary_table(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 解析"审定结算汇总表"，提取数据并生成 JSON 格式。 返回格式: [{ "No": int, # 序号 "name": str, # 项目名称（审计内容） "settledVerifiedTaxExclusiveInvestment": float, # 结算审定不含税投资（元，两位小数） "settledVerifiedTaxInclusiveInvestment": float, # 结算审定含税投资（元，两位小数） }, ...] """ if df.empty: return [] # 尝试识别表头行（通常在前几行） header_row_idx = None for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) # 检查是否包含表头关键词 if any(kw in row_text for kw in ["序号", "审计内容", "审定金额（含税）", "审定金额（不含税）"]): header_row_idx = i break if header_row_idx is None: # 如果没有找到明确的表头，假设第一行是表头 header_row_idx = 0 # 先修复跨行单元格（在识别表头之前） df = _fix_broken_cells(df, header_row_count=header_row_idx + 1) # 合并前几行作为表头（处理多行表头的情况） header_keywords = ["序号", "审计内容", "送审金额", "审定金额", "增减金额", "备注"] header_rows_to_check = _detect_header_rows(df, header_row_idx, header_keywords) header_texts = [] # 每列的合并文本 num_cols = len(df.columns) # 打印原始表头行（用于调试） logger.info(f"[审定结算汇总表] 原始表头行 (从第 {header_row_idx + 1} 行开始，共检查 {header_rows_to_check} 行):") for row_idx in range(header_row_idx, min(header_row_idx + header_rows_to_check, len(df))): row_data = [str(df.iloc[row_idx, col_idx]).strip() for col_idx in range(num_cols)] logger.info(f"[审定结算汇总表] 第 {row_idx + 1} 行: {row_data}") for col_idx in range(num_cols): col_text_parts = [] for row_idx in range(header_row_idx, header_row_idx + header_rows_to_check): if row_idx < len(df): cell_val = str(df.iloc[row_idx, col_idx]).strip() if cell_val and cell_val.lower() not in ['nan', 'none', '']: # 格式化表头文本（移除换行符和空格） formatted_text = _format_header_text(cell_val) if formatted_text: col_text_parts.append(formatted_text) # 合并该列的所有表头文本（已经是格式化后的，无空格） merged_text = ''.join(col_text_parts) header_texts.append(merged_text) logger.info(f"[审定结算汇总表] 合并后的表头文本（已格式化）: {header_texts}") col_no = None # 序号列 col_name = None # 审计内容列（项目名称） col_tax_exclusive = None # 审定金额（不含税）列 col_tax_inclusive = None # 审定金额（含税）列 for idx, header_text in enumerate(header_texts): cell_lower = header_text.lower() if "序号" in header_text or "no" in cell_lower: col_no = idx elif "审计内容" in header_text or "项目名称" in header_text or "name" in cell_lower: col_name = idx elif "审定金额（不含税）" in header_text or ("不含税" in header_text and "审定" in header_text): col_tax_exclusive = idx elif "审定金额（含税）" in header_text or ("审定金额" in header_text and "含税" in header_text): col_tax_inclusive = idx # 如果关键列未找到，尝试通过位置推断 if col_no is None: col_no = 0 if col_name is None: col_name = 1 # 如果金额列未找到，尝试从后往前找（通常金额列在表格右侧） if col_tax_exclusive is None or col_tax_inclusive is None: for idx in range(len(header_texts) - 1, -1, -1): header_text = header_texts[idx] if "不含税" in header_text and col_tax_exclusive is None: col_tax_exclusive = idx elif "含税" in header_text and "审定" in header_text and col_tax_inclusive is None: col_tax_inclusive = idx if col_tax_exclusive is not None and col_tax_inclusive is not None: break logger.info(f"[审定结算汇总表] 列识别: 序号={col_no}, 项目名称={col_name}, 不含税={col_tax_exclusive}, 含税={col_tax_inclusive}") # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) result = [] def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None name_val = row.iloc[col_name] if col_name is not None and col_name < len(row) else None tax_exclusive_val = row.iloc[col_tax_exclusive] if col_tax_exclusive is not None and col_tax_exclusive < len(row) else None tax_inclusive_val = row.iloc[col_tax_inclusive] if col_tax_inclusive is not None and col_tax_inclusive < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): pass # 跳过序号为空的行（这些通常是合计、其中等说明行） if not no_str or pd.isna(no_val): continue # 解析项目名称（审计内容），清理换行符 name = str(name_val).strip() if name_val is not None and not pd.isna(name_val) else "" # 清理换行符，替换为空格 name = name.replace('\n', '').replace('\r', '') # 直接移除换行符，不替换为空格 # 清理多余空格 name = re.sub(r'\s+', ' ', name).strip() # 跳过空行 if not name or name == "": continue # 判断是否为合计行（合计行需要跳过） is_total = any(kw in name for kw in ["合计", "总计", "总计", "合计金额"]) if is_total: continue # 解析投资金额 settled_verified_tax_exclusive = parse_number(tax_exclusive_val) settled_verified_tax_inclusive = parse_number(tax_inclusive_val) # 添加到结果 result.append({ "No": no if no is not None else idx + 1, "name": name, "settledVerifiedTaxExclusiveInvestment": settled_verified_tax_exclusive, "settledVerifiedTaxInclusiveInvestment": settled_verified_tax_inclusive, }) return result def parse_contract_execution_table(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 解析"合同执行情况"表，提取数据并生成 JSON 格式。 返回格式: [{ "No": int, # 序号 "constructionUnit": str, # 施工单位 "bidNoticeAmount": float, # 中标通知书金额（元，两位小数） "bidNoticeNo": str, # 中标通知书编号 "contractAmount": float, # 合同金额（元，两位小数） "settlementSubmittedAmount": float, # 结算送审金额（元，两位小数） "differenceAmount": float, # 差额（元，两位小数） }, ...] """ if df.empty: return [] # 尝试识别表头行（通常在前几行） header_row_idx = None for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) # 检查是否包含表头关键词 if any(kw in row_text for kw in ["序号", "施工单位", "中标通知书金额", "中标通知书编号", "合同金额", "结算送审金额", "差额"]): header_row_idx = i break if header_row_idx is None: # 如果没有找到明确的表头，假设第一行是表头 header_row_idx = 0 # 先修复跨行单元格（在识别表头之前） df = _fix_broken_cells(df, header_row_count=header_row_idx + 1) # 合并前几行作为表头（处理多行表头的情况） # 智能判断表头行数：表头行通常不包含纯数字（除了序号列） header_rows_to_check = 1 for i in range(header_row_idx + 1, min(header_row_idx + 3, len(df))): row = df.iloc[i] # 检查这一行是否像表头（包含关键词且不全是数字） row_text = " ".join(row.astype(str).str.strip().tolist()) has_keywords = any(kw in row_text for kw in ["施工单位", "中标", "合同", "结算", "送审", "差额"]) # 如果包含关键词且看起来不像数据行（不全是数字），则可能是表头的一部分 if has_keywords: # 检查是否主要是数字（如果是，则不是表头） numeric_count = sum(1 for cell in row if str(cell).strip().replace('.', '').replace('-', '').isdigit()) if numeric_count < len(row) * 0.5: # 如果数字占比小于50%，可能是表头 header_rows_to_check += 1 else: break else: break header_texts = [] # 每列的合并文本 num_cols = len(df.columns) # 打印原始表头行（用于调试） logger.info(f"[合同执行情况] 原始表头行 (从第 {header_row_idx + 1} 行开始，共检查 {header_rows_to_check} 行):") for row_idx in range(header_row_idx, min(header_row_idx + header_rows_to_check, len(df))): row_data = [str(df.iloc[row_idx, col_idx]).strip() for col_idx in range(num_cols)] logger.info(f"[合同执行情况] 第 {row_idx + 1} 行: {row_data}") for col_idx in range(num_cols): col_text_parts = [] for row_idx in range(header_row_idx, header_row_idx + header_rows_to_check): if row_idx < len(df): cell_val = str(df.iloc[row_idx, col_idx]).strip() if cell_val and cell_val.lower() not in ['nan', 'none', '']: # 格式化表头文本（移除换行符和空格） formatted_text = _format_header_text(cell_val) if formatted_text: col_text_parts.append(formatted_text) # 合并该列的所有表头文本（已经是格式化后的，无空格） merged_text = ''.join(col_text_parts) header_texts.append(merged_text) logger.info(f"[合同执行情况] 合并后的表头文本（已格式化）: {header_texts}") col_no = None # 序号列 col_construction_unit = None # 施工单位列 col_bid_notice_amount = None # 中标通知书金额列 col_bid_notice_no = None # 中标通知书编号列 col_contract_amount = None # 合同金额列 col_settlement_submitted = None # 结算送审金额列 col_difference = None # 差额列 for idx, header_text in enumerate(header_texts): cell_lower = header_text.lower() # header_text 已经是格式化后的（无空格），直接使用 if "序号" in header_text or "no" in cell_lower: col_no = idx elif "施工单位" in header_text: col_construction_unit = idx elif "中标" in header_text and "金额" in header_text: col_bid_notice_amount = idx elif "中标" in header_text and "编号" in header_text: col_bid_notice_no = idx elif ("合同金额" in header_text or ("合同" in header_text and "金额" in header_text and "结算" not in header_text and "送审" not in header_text)): col_contract_amount = idx elif ("结算送审金额" in header_text or ("送审" in header_text and "金额" in header_text) or ("结算" in header_text and "送审" in header_text)): col_settlement_submitted = idx elif "差额" in header_text: col_difference = idx # 如果关键列未找到，尝试通过位置推断 if col_no is None: col_no = 0 if col_construction_unit is None: col_construction_unit = 1 # 如果金额列未找到，尝试从后往前找（通常金额列在表格右侧） # 同时检查列名中是否包含关键词的部分匹配 if col_bid_notice_amount is None: for idx, header_text in enumerate(header_texts): if "中标" in header_text and "金额" in header_text: col_bid_notice_amount = idx break if col_bid_notice_no is None: for idx, header_text in enumerate(header_texts): if "中标" in header_text and "编号" in header_text: col_bid_notice_no = idx break if col_contract_amount is None: # 尝试更灵活的匹配：合同金额可能在"合同"和"金额"分开的列中 for idx, header_text in enumerate(header_texts): # header_text 已经是格式化后的（无空格），直接使用 # 检查是否包含"合同"和"金额"，且不包含"结算"、"送审"等 if ("合同" in header_text and "金额" in header_text and "结算" not in header_text and "送审" not in header_text): col_contract_amount = idx break if col_settlement_submitted is None: # 尝试更灵活的匹配：结算送审金额可能在"结算"、"送审"、"金额"分开的列中 for idx, header_text in enumerate(header_texts): # header_text 已经是格式化后的（无空格），直接使用 # 检查是否包含"送审"和"金额"，或者"结算"和"送审" if (("送审" in header_text and "金额" in header_text) or ("结算" in header_text and "送审" in header_text) or "结算送审金额" in header_text): col_settlement_submitted = idx break if col_difference is None: for idx, header_text in enumerate(header_texts): if "差额" in header_text: col_difference = idx break logger.info(f"[合同执行情况] 列识别: 序号={col_no}, 施工单位={col_construction_unit}, " f"中标金额={col_bid_notice_amount}, 中标编号={col_bid_notice_no}, " f"合同金额={col_contract_amount}, 送审金额={col_settlement_submitted}, 差额={col_difference}") # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) result = [] def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None construction_unit_val = row.iloc[col_construction_unit] if col_construction_unit is not None and col_construction_unit < len(row) else None bid_notice_amount_val = row.iloc[col_bid_notice_amount] if col_bid_notice_amount is not None and col_bid_notice_amount < len(row) else None bid_notice_no_val = row.iloc[col_bid_notice_no] if col_bid_notice_no is not None and col_bid_notice_no < len(row) else None contract_amount_val = row.iloc[col_contract_amount] if col_contract_amount is not None and col_contract_amount < len(row) else None settlement_submitted_val = row.iloc[col_settlement_submitted] if col_settlement_submitted is not None and col_settlement_submitted < len(row) else None difference_val = row.iloc[col_difference] if col_difference is not None and col_difference < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): pass # 跳过序号为空的行（这些通常是合计、其中等说明行） if not no_str or pd.isna(no_val): continue # 解析施工单位，清理换行符 construction_unit = str(construction_unit_val).strip() if construction_unit_val is not None and not pd.isna(construction_unit_val) else "" # 清理换行符，直接移除 construction_unit = construction_unit.replace('\n', '').replace('\r', '') # 清理多余空格 construction_unit = re.sub(r'\s+', ' ', construction_unit).strip() # 跳过空行 if not construction_unit or construction_unit == "": continue # 判断是否为合计行（合计行需要跳过） is_total = any(kw in construction_unit for kw in ["合计", "总计", "总计", "合计金额"]) if is_total: continue # 解析中标通知书编号，清理换行符 bid_notice_no = str(bid_notice_no_val).strip() if bid_notice_no_val is not None and not pd.isna(bid_notice_no_val) else "" bid_notice_no = bid_notice_no.replace('\n', '').replace('\r', '') bid_notice_no = re.sub(r'\s+', ' ', bid_notice_no).strip() # 解析金额（保留两位小数） bid_notice_amount = parse_number(bid_notice_amount_val) contract_amount = parse_number(contract_amount_val) settlement_submitted_amount = parse_number(settlement_submitted_val) difference_amount = parse_number(difference_val) # 添加到结果 result.append({ "No": no if no is not None else idx + 1, "constructionUnit": construction_unit, "bidNoticeAmount": bid_notice_amount, "bidNoticeNo": bid_notice_no, "contractAmount": contract_amount, "settlementSubmittedAmount": settlement_submitted_amount, "differenceAmount": difference_amount, }) return result def parse_compensation_contract_table(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 解析"赔偿合同"表，提取数据并生成 JSON 格式。 返回格式: [{ "No": int, # 序号 "contractCounterpartyName": str, # 合同对方名称 "compensationItem": str, # 赔偿事项 "contractAmount": float, # 合同金额（元，两位小数） "settlementSubmittedAmount": float, # 结算送审金额（元，两位小数） "differenceAmount": float, # 差额（元，两位小数） }, ...] """ if df.empty: return [] # 尝试识别表头行（通常在前几行） header_row_idx = None for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) # 检查是否包含表头关键词 if any(kw in row_text for kw in ["序号", "合同对方", "赔偿事项", "合同金额", "结算送审金额", "差额"]): header_row_idx = i break if header_row_idx is None: # 如果没有找到明确的表头，假设第一行是表头 header_row_idx = 0 # 先修复跨行单元格（在识别表头之前） df = _fix_broken_cells(df, header_row_count=header_row_idx + 1) # 合并前几行作为表头（处理多行表头的情况） header_keywords = ["序号", "合同对方", "赔偿事项", "合同金额", "结算送审金额", "差额"] header_rows_to_check = _detect_header_rows(df, header_row_idx, header_keywords) header_texts = [] # 每列的合并文本 num_cols = len(df.columns) # 打印原始表头行（用于调试） logger.info(f"[赔偿合同] 原始表头行 (从第 {header_row_idx + 1} 行开始，共检查 {header_rows_to_check} 行):") for row_idx in range(header_row_idx, min(header_row_idx + header_rows_to_check, len(df))): row_data = [str(df.iloc[row_idx, col_idx]).strip() for col_idx in range(num_cols)] logger.info(f"[赔偿合同] 第 {row_idx + 1} 行: {row_data}") for col_idx in range(num_cols): col_text_parts = [] for row_idx in range(header_row_idx, header_row_idx + header_rows_to_check): if row_idx < len(df): cell_val = str(df.iloc[row_idx, col_idx]).strip() if cell_val and cell_val.lower() not in ['nan', 'none', '']: # 格式化表头文本（移除换行符和空格） formatted_text = _format_header_text(cell_val) if formatted_text: col_text_parts.append(formatted_text) # 合并该列的所有表头文本（已经是格式化后的，无空格） merged_text = ''.join(col_text_parts) header_texts.append(merged_text) logger.info(f"[赔偿合同] 合并后的表头文本（已格式化）: {header_texts}") col_no = None # 序号列 col_counterparty_name = None # 合同对方名称列 col_compensation_item = None # 赔偿事项列 col_contract_amount = None # 合同金额列 col_settlement_submitted = None # 结算送审金额列 col_difference = None # 差额列 for idx, header_text in enumerate(header_texts): cell_lower = header_text.lower() if "序号" in header_text or "no" in cell_lower: col_no = idx elif "合同对方" in header_text: col_counterparty_name = idx elif "赔偿事项" in header_text: col_compensation_item = idx elif "合同金额" in header_text and "结算" not in header_text: col_contract_amount = idx elif "结算送审金额" in header_text or ("送审金额" in header_text and "结算" in header_text): col_settlement_submitted = idx elif "差额" in header_text: col_difference = idx # 如果关键列未找到，尝试通过位置推断 if col_no is None: col_no = 0 if col_counterparty_name is None: col_counterparty_name = 1 if col_compensation_item is None: col_compensation_item = 2 # 如果金额列未找到，尝试从后往前找（通常金额列在表格右侧） if col_contract_amount is None or col_settlement_submitted is None or col_difference is None: for idx in range(len(header_texts) - 1, -1, -1): header_text = header_texts[idx] if "差额" in header_text and col_difference is None: col_difference = idx elif "送审" in header_text and "金额" in header_text and col_settlement_submitted is None: col_settlement_submitted = idx elif "合同金额" in header_text and col_contract_amount is None: col_contract_amount = idx if col_contract_amount is not None and col_settlement_submitted is not None and col_difference is not None: break logger.info(f"[赔偿合同] 列识别: 序号={col_no}, 合同对方={col_counterparty_name}, " f"赔偿事项={col_compensation_item}, 合同金额={col_contract_amount}, " f"送审金额={col_settlement_submitted}, 差额={col_difference}") # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) result = [] def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None counterparty_name_val = row.iloc[col_counterparty_name] if col_counterparty_name is not None and col_counterparty_name < len(row) else None compensation_item_val = row.iloc[col_compensation_item] if col_compensation_item is not None and col_compensation_item < len(row) else None contract_amount_val = row.iloc[col_contract_amount] if col_contract_amount is not None and col_contract_amount < len(row) else None settlement_submitted_val = row.iloc[col_settlement_submitted] if col_settlement_submitted is not None and col_settlement_submitted < len(row) else None difference_val = row.iloc[col_difference] if col_difference is not None and col_difference < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): pass # 跳过序号为空的行（这些通常是合计、其中等说明行） if not no_str or pd.isna(no_val): continue # 解析合同对方名称，清理换行符 counterparty_name = str(counterparty_name_val).strip() if counterparty_name_val is not None and not pd.isna(counterparty_name_val) else "" counterparty_name = counterparty_name.replace('\n', '').replace('\r', '') counterparty_name = re.sub(r'\s+', ' ', counterparty_name).strip() # 跳过空行 if not counterparty_name or counterparty_name == "": continue # 判断是否为合计行（合计行需要跳过） is_total = any(kw in counterparty_name for kw in ["合计", "总计", "总计", "合计金额"]) if is_total: continue # 解析赔偿事项，清理换行符 compensation_item = str(compensation_item_val).strip() if compensation_item_val is not None and not pd.isna(compensation_item_val) else "" compensation_item = compensation_item.replace('\n', '').replace('\r', '') compensation_item = re.sub(r'\s+', ' ', compensation_item).strip() # 解析金额（保留两位小数） contract_amount = parse_number(contract_amount_val) settlement_submitted_amount = parse_number(settlement_submitted_val) difference_amount = parse_number(difference_val) # 添加到结果 result.append({ "No": no if no is not None else idx + 1, "contractCounterpartyName": counterparty_name, "compensationItem": compensation_item, "contractAmount": contract_amount, "settlementSubmittedAmount": settlement_submitted_amount, "differenceAmount": difference_amount, }) return result def parse_material_purchase_contract1_table(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 解析"物资采购合同1"表，提取数据并生成 JSON 格式。 返回格式: [{ "No": int, # 序号 "materialName": str, # 物料名称 "contractQuantity": float, # 合同数量 "drawingQuantity": float, # 施工图数量 "unitPriceExcludingTax": float, # 单价（不含税）（元，两位小数） "differenceAmount": float, # 差额（元，两位小数） }, ...] """ if df.empty: return [] # 尝试识别表头行（通常在前几行） header_row_idx = None for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) # 检查是否包含表头关键词 if any(kw in row_text for kw in ["序号", "物料名称", "合同数量", "施工图数量", "单价（不含税）", "差额"]): header_row_idx = i break if header_row_idx is None: # 如果没有找到明确的表头，假设第一行是表头 header_row_idx = 0 # 先修复跨行单元格（在识别表头之前） df = _fix_broken_cells(df, header_row_count=header_row_idx + 1) # 合并前几行作为表头（处理多行表头的情况） header_keywords = ["序号", "物料名称", "合同数量", "施工图数量", "单价", "差额"] header_rows_to_check = _detect_header_rows(df, header_row_idx, header_keywords) header_texts = [] # 每列的合并文本 num_cols = len(df.columns) # 打印原始表头行（用于调试） logger.info(f"[物资采购合同1] 原始表头行 (从第 {header_row_idx + 1} 行开始，共检查 {header_rows_to_check} 行):") for row_idx in range(header_row_idx, min(header_row_idx + header_rows_to_check, len(df))): row_data = [str(df.iloc[row_idx, col_idx]).strip() for col_idx in range(num_cols)] logger.info(f"[物资采购合同1] 第 {row_idx + 1} 行: {row_data}") for col_idx in range(num_cols): col_text_parts = [] for row_idx in range(header_row_idx, header_row_idx + header_rows_to_check): if row_idx < len(df): cell_val = str(df.iloc[row_idx, col_idx]).strip() if cell_val and cell_val.lower() not in ['nan', 'none', '']: # 格式化表头文本（移除换行符和空格） formatted_text = _format_header_text(cell_val) if formatted_text: col_text_parts.append(formatted_text) # 合并该列的所有表头文本（已经是格式化后的，无空格） merged_text = ''.join(col_text_parts) header_texts.append(merged_text) logger.info(f"[物资采购合同1] 合并后的表头文本（已格式化）: {header_texts}") col_no = None # 序号列 col_material_name = None # 物料名称列 col_contract_quantity = None # 合同数量列 col_drawing_quantity = None # 施工图数量列 col_unit_price = None # 单价（不含税）列 col_difference = None # 差额列 for idx, header_text in enumerate(header_texts): cell_lower = header_text.lower() if "序号" in header_text or "no" in cell_lower: col_no = idx elif "物料名称" in header_text: col_material_name = idx elif "合同数量" in header_text: col_contract_quantity = idx elif "施工图数量" in header_text: col_drawing_quantity = idx elif "单价" in header_text and "不含税" in header_text: col_unit_price = idx elif "差额" in header_text: col_difference = idx # 如果关键列未找到，尝试通过位置推断 if col_no is None: col_no = 0 if col_material_name is None: col_material_name = 1 # 如果数量列未找到，尝试查找 if col_contract_quantity is None: for idx, header_text in enumerate(header_texts): if "合同" in header_text and "数量" in header_text: col_contract_quantity = idx break if col_drawing_quantity is None: for idx, header_text in enumerate(header_texts): if "施工图" in header_text and "数量" in header_text: col_drawing_quantity = idx break # 如果金额列未找到，尝试从后往前找（通常金额列在表格右侧） if col_unit_price is None: for idx, header_text in enumerate(header_texts): if "单价" in header_text: col_unit_price = idx break if col_difference is None: for idx in range(len(header_texts) - 1, -1, -1): if "差额" in header_texts[idx]: col_difference = idx break logger.info(f"[物资采购合同1] 列识别: 序号={col_no}, 物料名称={col_material_name}, " f"合同数量={col_contract_quantity}, 施工图数量={col_drawing_quantity}, " f"单价={col_unit_price}, 差额={col_difference}") # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) result = [] def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None material_name_val = row.iloc[col_material_name] if col_material_name is not None and col_material_name < len(row) else None contract_quantity_val = row.iloc[col_contract_quantity] if col_contract_quantity is not None and col_contract_quantity < len(row) else None drawing_quantity_val = row.iloc[col_drawing_quantity] if col_drawing_quantity is not None and col_drawing_quantity < len(row) else None unit_price_val = row.iloc[col_unit_price] if col_unit_price is not None and col_unit_price < len(row) else None difference_val = row.iloc[col_difference] if col_difference is not None and col_difference < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): pass # 跳过序号为空的行（这些通常是合计、其中等说明行） if not no_str or pd.isna(no_val): continue # 解析物料名称，清理换行符 material_name = str(material_name_val).strip() if material_name_val is not None and not pd.isna(material_name_val) else "" material_name = material_name.replace('\n', '').replace('\r', '') material_name = re.sub(r'\s+', ' ', material_name).strip() # 跳过空行 if not material_name or material_name == "": continue # 判断是否为合计行（合计行需要跳过） is_total = any(kw in material_name for kw in ["合计", "总计", "总计", "合计金额"]) if is_total: continue # 解析数量（合同数量和施工图数量可能是整数或小数） contract_quantity = parse_number(contract_quantity_val) drawing_quantity = parse_number(drawing_quantity_val) # 解析金额（保留两位小数） unit_price_excluding_tax = parse_number(unit_price_val) difference_amount = parse_number(difference_val) # 添加到结果 result.append({ "No": no if no is not None else idx + 1, "materialName": material_name, "contractQuantity": contract_quantity, "drawingQuantity": drawing_quantity, "unitPriceExcludingTax": unit_price_excluding_tax, "differenceAmount": difference_amount, }) return result def parse_material_purchase_contract2_table(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 解析"物资采购合同2"表，提取数据并生成 JSON 格式。 返回格式: [{ "No": int, # 序号 "materialName": str, # 物料名称 "contractAmount": float, # 合同金额（元，两位小数） "bookedAmount": float, # 入账金额（元，两位小数） "differenceAmount": float, # 差额（元，两位小数） "remark": str, # 备注 }, ...] """ if df.empty: return [] # 尝试识别表头行（通常在前几行） header_row_idx = None for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) # 检查是否包含表头关键词 if any(kw in row_text for kw in ["序号", "物料名称", "合同金额（不含税）", "入账金额", "差额", "备注"]): header_row_idx = i break if header_row_idx is None: # 如果没有找到明确的表头，假设第一行是表头 header_row_idx = 0 # 先修复跨行单元格（在识别表头之前） df = _fix_broken_cells(df, header_row_count=header_row_idx + 1) # 合并前几行作为表头（处理多行表头的情况） header_keywords = ["序号", "物料名称", "合同金额", "入账金额", "差额", "备注"] header_rows_to_check = _detect_header_rows(df, header_row_idx, header_keywords) header_texts = [] # 每列的合并文本 num_cols = len(df.columns) # 打印原始表头行（用于调试） logger.info(f"[物资采购合同2] 原始表头行 (从第 {header_row_idx + 1} 行开始，共检查 {header_rows_to_check} 行):") for row_idx in range(header_row_idx, min(header_row_idx + header_rows_to_check, len(df))): row_data = [str(df.iloc[row_idx, col_idx]).strip() for col_idx in range(num_cols)] logger.info(f"[物资采购合同2] 第 {row_idx + 1} 行: {row_data}") for col_idx in range(num_cols): col_text_parts = [] for row_idx in range(header_row_idx, header_row_idx + header_rows_to_check): if row_idx < len(df): cell_val = str(df.iloc[row_idx, col_idx]).strip() if cell_val and cell_val.lower() not in ['nan', 'none', '']: # 格式化表头文本（移除换行符和空格） formatted_text = _format_header_text(cell_val) if formatted_text: col_text_parts.append(formatted_text) # 合并该列的所有表头文本（已经是格式化后的，无空格） merged_text = ''.join(col_text_parts) header_texts.append(merged_text) logger.info(f"[物资采购合同2] 合并后的表头文本（已格式化）: {header_texts}") col_no = None # 序号列 col_material_name = None # 物料名称列 col_contract_amount = None # 合同金额（不含税）列 col_booked_amount = None # 入账金额列 col_difference = None # 差额列 col_remark = None # 备注列 for idx, header_text in enumerate(header_texts): cell_lower = header_text.lower() if "序号" in header_text or "no" in cell_lower: col_no = idx elif "物料名称" in header_text: col_material_name = idx elif "合同金额" in header_text and "不含税" in header_text: col_contract_amount = idx elif "入账金额" in header_text: col_booked_amount = idx elif "差额" in header_text: col_difference = idx elif "备注" in header_text: col_remark = idx # 如果关键列未找到，尝试通过位置推断 if col_no is None: col_no = 0 if col_material_name is None: col_material_name = 1 # 如果金额列未找到，尝试查找 if col_contract_amount is None: for idx, header_text in enumerate(header_texts): if "合同金额" in header_text: col_contract_amount = idx break if col_booked_amount is None: for idx, header_text in enumerate(header_texts): if "入账" in header_text and "金额" in header_text: col_booked_amount = idx break # 如果差额列未找到，尝试从后往前找（通常差额列在表格右侧） if col_difference is None: for idx in range(len(header_texts) - 1, -1, -1): if "差额" in header_texts[idx]: col_difference = idx break # 备注列通常在最后 if col_remark is None: for idx in range(len(header_texts) - 1, -1, -1): if "备注" in header_texts[idx]: col_remark = idx break logger.info(f"[物资采购合同2] 列识别: 序号={col_no}, 物料名称={col_material_name}, " f"合同金额={col_contract_amount}, 入账金额={col_booked_amount}, " f"差额={col_difference}, 备注={col_remark}") # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) result = [] def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None material_name_val = row.iloc[col_material_name] if col_material_name is not None and col_material_name < len(row) else None contract_amount_val = row.iloc[col_contract_amount] if col_contract_amount is not None and col_contract_amount < len(row) else None booked_amount_val = row.iloc[col_booked_amount] if col_booked_amount is not None and col_booked_amount < len(row) else None difference_val = row.iloc[col_difference] if col_difference is not None and col_difference < len(row) else None remark_val = row.iloc[col_remark] if col_remark is not None and col_remark < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): pass # 跳过序号为空的行（这些通常是合计、其中等说明行） if not no_str or pd.isna(no_val): continue # 解析物料名称，清理换行符 material_name = str(material_name_val).strip() if material_name_val is not None and not pd.isna(material_name_val) else "" material_name = material_name.replace('\n', '').replace('\r', '') material_name = re.sub(r'\s+', ' ', material_name).strip() # 跳过空行 if not material_name or material_name == "": continue # 判断是否为合计行（合计行需要跳过） is_total = any(kw in material_name for kw in ["合计", "总计", "总计", "合计金额"]) if is_total: continue # 解析金额（保留两位小数） contract_amount = parse_number(contract_amount_val) booked_amount = parse_number(booked_amount_val) difference_amount = parse_number(difference_val) # 解析备注，清理换行符 remark = str(remark_val).strip() if remark_val is not None and not pd.isna(remark_val) else "" remark = remark.replace('\n', '').replace('\r', '') remark = re.sub(r'\s+', ' ', remark).strip() # 添加到结果 result.append({ "No": no if no is not None else idx + 1, "materialName": material_name, "contractAmount": contract_amount, "bookedAmount": booked_amount, "differenceAmount": difference_amount, "remark": remark, }) return result def parse_other_service_contract_table(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 解析"其他服务类合同"表，提取数据并生成 JSON 格式。 返回格式: [{ "No": int, # 序号 "serviceProvider": str, # 服务商 "bidNotice": str, # 中标通知书 "contractAmount": float, # 合同金额（元，两位小数） "submittedAmount": float, # 送审金额（元，两位小数） "settlementAmount": float, # 结算金额（元，两位小数） }, ...] """ if df.empty: return [] # 尝试识别表头行（通常在前几行） header_row_idx = None for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) # 检查是否包含表头关键词 if any(kw in row_text for kw in ["序号", "服务商", "中标通知书", "合同金额", "送审金额", "结算金额"]): header_row_idx = i break if header_row_idx is None: # 如果没有找到明确的表头，假设第一行是表头 header_row_idx = 0 # 先修复跨行单元格（在识别表头之前） df = _fix_broken_cells(df, header_row_count=header_row_idx + 1) # 合并前几行作为表头（处理多行表头的情况） header_keywords = ["序号", "服务商", "中标通知书", "合同金额", "送审金额", "结算金额"] header_rows_to_check = _detect_header_rows(df, header_row_idx, header_keywords) header_texts = [] # 每列的合并文本 num_cols = len(df.columns) # 打印原始表头行（用于调试） logger.info(f"[其他服务类合同] 原始表头行 (从第 {header_row_idx + 1} 行开始，共检查 {header_rows_to_check} 行):") for row_idx in range(header_row_idx, min(header_row_idx + header_rows_to_check, len(df))): row_data = [str(df.iloc[row_idx, col_idx]).strip() for col_idx in range(num_cols)] logger.info(f"[其他服务类合同] 第 {row_idx + 1} 行: {row_data}") for col_idx in range(num_cols): col_text_parts = [] for row_idx in range(header_row_idx, header_row_idx + header_rows_to_check): if row_idx < len(df): cell_val = str(df.iloc[row_idx, col_idx]).strip() if cell_val and cell_val.lower() not in ['nan', 'none', '']: # 格式化表头文本（移除换行符和空格） formatted_text = _format_header_text(cell_val) if formatted_text: col_text_parts.append(formatted_text) # 合并该列的所有表头文本（已经是格式化后的，无空格） merged_text = ''.join(col_text_parts) header_texts.append(merged_text) logger.info(f"[其他服务类合同] 合并后的表头文本（已格式化）: {header_texts}") col_no = None # 序号列 col_service_provider = None # 服务商列 col_bid_notice = None # 中标通知书列 col_contract_amount = None # 合同金额列 col_submitted_amount = None # 送审金额列 col_settlement_amount = None # 结算金额列 for idx, header_text in enumerate(header_texts): cell_lower = header_text.lower() if "序号" in header_text or "no" in cell_lower: col_no = idx elif "服务商" in header_text: col_service_provider = idx elif "中标通知书" in header_text: col_bid_notice = idx elif "合同金额" in header_text and "送审" not in header_text and "结算" not in header_text: col_contract_amount = idx elif "送审金额" in header_text: col_submitted_amount = idx elif "结算金额" in header_text: col_settlement_amount = idx # 如果关键列未找到，尝试通过位置推断 if col_no is None: col_no = 0 if col_service_provider is None: col_service_provider = 1 # 如果金额列未找到，尝试查找 if col_contract_amount is None: for idx, header_text in enumerate(header_texts): if "合同" in header_text and "金额" in header_text: col_contract_amount = idx break if col_submitted_amount is None: for idx, header_text in enumerate(header_texts): if "送审" in header_text and "金额" in header_text: col_submitted_amount = idx break if col_settlement_amount is None: for idx, header_text in enumerate(header_texts): if "结算" in header_text and "金额" in header_text: col_settlement_amount = idx break # 如果中标通知书列未找到，尝试查找 if col_bid_notice is None: for idx, header_text in enumerate(header_texts): if "中标" in header_text and "通知" in header_text: col_bid_notice = idx break logger.info(f"[其他服务类合同] 列识别: 序号={col_no}, 服务商={col_service_provider}, " f"中标通知书={col_bid_notice}, 合同金额={col_contract_amount}, " f"送审金额={col_submitted_amount}, 结算金额={col_settlement_amount}") # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) result = [] def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None service_provider_val = row.iloc[col_service_provider] if col_service_provider is not None and col_service_provider < len(row) else None bid_notice_val = row.iloc[col_bid_notice] if col_bid_notice is not None and col_bid_notice < len(row) else None contract_amount_val = row.iloc[col_contract_amount] if col_contract_amount is not None and col_contract_amount < len(row) else None submitted_amount_val = row.iloc[col_submitted_amount] if col_submitted_amount is not None and col_submitted_amount < len(row) else None settlement_amount_val = row.iloc[col_settlement_amount] if col_settlement_amount is not None and col_settlement_amount < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): pass # 跳过序号为空的行（这些通常是合计、其中等说明行） if not no_str or pd.isna(no_val): continue # 解析服务商，清理换行符 service_provider = str(service_provider_val).strip() if service_provider_val is not None and not pd.isna(service_provider_val) else "" service_provider = service_provider.replace('\n', '').replace('\r', '') service_provider = re.sub(r'\s+', ' ', service_provider).strip() # 跳过空行 if not service_provider or service_provider == "": continue # 判断是否为合计行（合计行需要跳过） is_total = any(kw in service_provider for kw in ["合计", "总计", "总计", "合计金额"]) if is_total: continue # 解析中标通知书，清理换行符 bid_notice = str(bid_notice_val).strip() if bid_notice_val is not None and not pd.isna(bid_notice_val) else "" bid_notice = bid_notice.replace('\n', '').replace('\r', '') bid_notice = re.sub(r'\s+', ' ', bid_notice).strip() # 解析金额（保留两位小数） contract_amount = parse_number(contract_amount_val) submitted_amount = parse_number(submitted_amount_val) settlement_amount = parse_number(settlement_amount_val) # 添加到结果 result.append({ "No": no if no is not None else idx + 1, "serviceProvider": service_provider, "bidNotice": bid_notice, "contractAmount": contract_amount, "submittedAmount": submitted_amount, "settlementAmount": settlement_amount, }) return result def parse_design_review_table(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 解析 designReview 类型的表格，提取数据并生成结构化的 JSON 格式。 返回格式: [{ "name": str, # 大类名称（如"变电工程"、"线路工程"） "Level": 0, # 大类层级 "staticInvestment": float, # 静态投资总计 "dynamicInvestment": float, # 动态投资总计 "items": [ # 子项列表 { "No": int, # 序号 "name": str, # 工程名称 "Level": 1, # 子项层级 "staticInvestment": float, # 静态投资 "dynamicInvestment": float, # 动态投资 }, ... ] }, ...] """ if df.empty: return [] # 中文数字映射（用于识别序号格式） CHINESE_NUMBERS = { '一': 1, '二': 2, '三': 3, '四': 4, '五': 5, '六': 6, '七': 7, '八': 8, '九': 9, '十': 10, '十一': 11, '十二': 12, '十三': 13, '十四': 14, '十五': 15, '十六': 16, '十七': 17, '十八': 18, '十九': 19, '二十': 20 } def is_category_by_serial(no_str: str) -> bool: """ 通过序号格式判断是否为大类 大类：序号为中文数字，如"一"、"二"、"三"（不含括号） 子类：序号为带括号的中文数字，如"（一）"、"（二）"、"（三）" """ if not no_str: return False no_str = str(no_str).strip() # 如果包含括号，是子类 if '（' in no_str or '(' in no_str or '）' in no_str or ')' in no_str: return False # 检查是否是中文数字（大类） # 移除可能的空格和标点 cleaned = no_str.replace(' ', '').replace('、', '').replace('.', '').replace('。', '') # 检查是否以中文数字开头 for chinese_num in CHINESE_NUMBERS.keys(): if cleaned.startswith(chinese_num): return True return False # 尝试识别表头行（通常在前几行） header_row_idx = None for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) # 检查是否包含表头关键词 if any(kw in row_text for kw in ["序号", "工程名称", "建设规模", "静态投资", "动态投资"]): header_row_idx = i break if header_row_idx is None: # 如果没有找到明确的表头，假设第一行是表头 header_row_idx = 0 # 从表头行识别列索引 header_row = df.iloc[header_row_idx].astype(str).str.strip() col_no = None # 序号列 col_name = None # 工程名称列 col_scale = None # 建设规模列（用于判断层级） col_static = None # 静态投资列 col_dynamic = None # 动态投资列 for idx, cell in enumerate(header_row): cell_lower = cell.lower() if "序号" in cell or "no" in cell_lower: col_no = idx elif "工程名称" in cell or "name" in cell_lower: col_name = idx elif "建设规模" in cell or "规模" in cell: col_scale = idx elif "静态投资" in cell or "static" in cell_lower: col_static = idx elif "动态投资" in cell or "dynamic" in cell_lower: col_dynamic = idx # 如果关键列未找到，尝试通过位置推断（通常顺序：序号、工程名称、建设规模、静态投资、动态投资） if col_no is None: col_no = 0 if col_name is None: col_name = 1 if col_static is None: # 从后往前找 for idx in range(len(df.columns) - 1, -1, -1): if "投资" in str(df.iloc[header_row_idx, idx]) or "元" in str(df.iloc[header_row_idx, idx]): if col_dynamic is None: col_dynamic = idx elif col_static is None: col_static = idx break # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str # 先解析所有行 all_items = [] for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None name_val = row.iloc[col_name] if col_name is not None and col_name < len(row) else None static_val = row.iloc[col_static] if col_static is not None and col_static < len(row) else None dynamic_val = row.iloc[col_dynamic] if col_dynamic is not None and col_dynamic < len(row) else None # 解析序号（保留原始字符串） no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: # 先尝试解析为数字 try: no = int(float(no_str)) except (ValueError, TypeError): # 如果不是数字，尝试解析中文数字 cleaned = no_str.replace(' ', '').replace('、', '').replace('.', '').replace('。', '') # 移除括号 cleaned_no_brackets = cleaned.replace('（', '').replace('(', '').replace('）', '').replace(')', '') if cleaned_no_brackets in CHINESE_NUMBERS: no = CHINESE_NUMBERS[cleaned_no_brackets] else: # 尝试匹配中文数字前缀 for chinese_num, num_val in CHINESE_NUMBERS.items(): if cleaned_no_brackets.startswith(chinese_num): no = num_val break # 跳过序号为空的行（这些通常是"其中："等说明行） if not no_str or pd.isna(no_val): continue # 解析工程名称 name = str(name_val).strip() if name_val is not None and not pd.isna(name_val) else "" # 跳过空行 if not name or name == "": continue # 判断是否为合计行（合计行需要跳过） is_total = any(kw in name for kw in ["合计", "总计", "总计", "合计金额"]) if is_total: continue # 解析投资金额 static_investment = parse_number(static_val) dynamic_investment = parse_number(dynamic_val) # 判断是否为大类：通过序号格式识别 is_category = is_category_by_serial(no_str) all_items.append({ "No": no if no is not None else idx + 1, "name": name, "isCategory": is_category, "staticInvestment": static_investment, "dynamicInvestment": dynamic_investment, }) # 构建层级结构 result = [] current_category = None for item in all_items: if item["isCategory"]: # 如果遇到新的大类，先保存之前的大类（如果有） if current_category is not None: result.append(current_category) # 创建新的大类 current_category = { "name": item["name"], "Level": 0, "staticInvestment": item["staticInvestment"], "dynamicInvestment": item["dynamicInvestment"], "items": [] } else: # 子项，添加到当前大类 if current_category is not None: current_category["items"].append({ "No": item["No"], "name": item["name"], "Level": 1, "staticInvestment": item["staticInvestment"], "dynamicInvestment": item["dynamicInvestment"], }) else: # 如果没有大类，作为独立项（不应该发生，但容错处理） result.append({ "name": item["name"], "Level": 1, "staticInvestment": item["staticInvestment"], "dynamicInvestment": item["dynamicInvestment"], "items": [] }) # 保存最后一个大类 if current_category is not None: result.append(current_category) return result def parse_design_review_detail_table(df: pd.DataFrame, table_title: str) -> List[Dict[str, Any]]: """ 解析 designReview 类型的概算投资明细表格（规则2）。 表头格式： 序号 | 工程或费用名称 | 建筑工程费 | 设备购置费 | 安装工程费 | 其他费用 | 合计 | ... Args: df: 表格 DataFrame table_title: 表格标题（如"周村 220kV 变电站新建工程总概算表"），用于提取工程名称 返回格式: [{ "No": int, # 序号 "Level": int, # 明细等级 "name": str, # 单项工程名称（从标题提取，如"周村220KV变电站新建工程"） "projectOrExpenseName": str, # 工程或费用名称 "constructionProjectCost": float, # 建筑工程费（元） "equipmentPurchaseCost": float, # 设备购置费（元） "installationProjectCost": float, # 安装工程费（元） "otherExpenses": float, # 其他费用（元） }, ...] """ if df.empty: return [] # 从标题中提取工程名称 # 标题格式如："周村 220kV 变电站新建工程总概算表" -> "周村220kV变电站新建工程" project_name = table_title if project_name: # 移除"总概算表"、"概算表"、"估算表"等后缀 project_name = re.sub(r'(总概算表|概算表|估算表|汇总表)$', '', project_name) # 移除多余空格 project_name = re.sub(r'\s+', '', project_name) # 中文数字映射（用于判断层级） CHINESE_NUMBERS = { '一': 1, '二': 2, '三': 3, '四': 4, '五': 5, '六': 6, '七': 7, '八': 8, '九': 9, '十': 10, '十一': 11, '十二': 12, '十三': 13, '十四': 14, '十五': 15, '十六': 16, '十七': 17, '十八': 18, '十九': 19, '二十': 20 } # 特殊的"其中："项，应作为 Level 1（独立大类）而非子项 SPECIAL_LEVEL1_ITEMS = [ "可抵扣固定资产增值税额", ] def determine_level(no_str: str, expense_name: str = "") -> int: """ 根据序号格式和费用名称判断层级： - 中文数字（一、二、三）：Level 1（大类） - 阿拉伯数字（1、2、3）：Level 1（大类） - 带括号的数字（(1)、（一））：Level 2（子项） - 以"其中："开头的行：Level 2（子项），但特殊项除外 """ no_str = str(no_str).strip() if no_str else "" expense_name = str(expense_name).strip() if expense_name else "" # 移除"其中："前缀用于判断 name_without_prefix = expense_name if expense_name.startswith("其中："): name_without_prefix = expense_name[3:] elif expense_name.startswith("其中:"): name_without_prefix = expense_name[3:] # 检查是否为特殊的 Level 1 项 for special_item in SPECIAL_LEVEL1_ITEMS: if special_item in name_without_prefix: return 1 # 以"其中："或"其中:"开头的 -> Level 2（子项） if expense_name.startswith("其中：") or expense_name.startswith("其中:"): return 2 # 序号为空的情况，默认 Level 1 if not no_str: return 1 # 带括号的 -> Level 2（子项） if '（' in no_str or '(' in no_str or '）' in no_str or ')' in no_str: return 2 # 中文数字（一、二、三等）-> Level 1（大类） cleaned = no_str.replace(' ', '').replace('、', '').replace('.', '').replace('。', '') for chinese_num in CHINESE_NUMBERS.keys(): if cleaned == chinese_num or cleaned.startswith(chinese_num): return 1 # 阿拉伯数字（1、2、3等）-> Level 1（大类） if re.match(r'^\d+\.?$', cleaned): return 1 return 1 # 默认 def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除常见的非数字字符（保留小数点、负号） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str # 识别表头行（可能有多行表头，需要扫描多行来找到列索引） # 先找到包含 "序号" 的行作为起始行 header_start_idx = None for i in range(min(5, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) row_text_no_space = row_text.replace(" ", "") if "序号" in row_text_no_space: header_start_idx = i break if header_start_idx is None: header_start_idx = 0 logger.warning(f"[表格解析] 未找到包含'序号'的表头行，使用第一行作为表头") # 扫描表头区域（可能是多行表头），收集所有列的关键词 col_no = None col_name = None col_construction = None col_equipment = None col_installation = None col_other = None # 打印前几行表头用于调试 logger.debug(f"[表格解析] DataFrame 形状: {df.shape}") for i in range(min(3, len(df))): row_vals = df.iloc[i].astype(str).tolist() logger.debug(f"[表格解析] 第{i}行: {row_vals}") # 扫描前几行表头（多行表头情况） header_end_idx = header_start_idx for i in range(header_start_idx, min(header_start_idx + 3, len(df))): row = df.iloc[i].astype(str).str.strip() for idx, cell in enumerate(row): # 移除所有空格、换行符等 cell_no_space = re.sub(r'\s+', '', cell) if ("序号" in cell_no_space or cell_no_space == "No") and col_no is None: col_no = idx elif ("工程或费用名称" in cell_no_space or "费用名称" in cell_no_space) and col_name is None: col_name = idx # 增强列名匹配 - 使用更宽松的匹配 elif col_construction is None and any(kw in cell_no_space for kw in ["建筑工程费", "建筑工程", "建筑"]): col_construction = idx elif col_equipment is None and any(kw in cell_no_space for kw in ["设备购置费", "设备购置", "设备"]): col_equipment = idx elif col_installation is None and any(kw in cell_no_space for kw in ["安装工程费", "安装工程", "安装"]): col_installation = idx elif col_other is None and "其他费用" in cell_no_space: col_other = idx # 检查这一行是否像数据行（第一列是数字或中文数字） first_cell = row.iloc[0] if len(row) > 0 else "" first_cell_clean = re.sub(r'\s+', '', first_cell).replace("、", "") chinese_nums = ['一', '二', '三', '四', '五', '六', '七', '八', '九', '十'] if first_cell_clean and (first_cell_clean.isdigit() or first_cell_clean in chinese_nums): # 这一行可能是数据行，表头到此结束 header_end_idx = i - 1 break header_end_idx = i # 如果关键列未找到，尝试位置推断 if col_no is None: col_no = 0 if col_name is None: col_name = 1 # 如果费用列未找到，尝试按位置推断 # 假设顺序：序号(0)、名称(1)、建筑(2)、设备(3)、安装(4)、其他(5)、合计(6)... logger.info(f"[表格解析] 列数: {len(df.columns)}, 当前列索引: 建筑={col_construction}, 设备={col_equipment}, 安装={col_installation}, 其他={col_other}") if col_construction is None or col_equipment is None or col_installation is None or col_other is None: num_cols = len(df.columns) if num_cols >= 7: # 按位置推断 if col_construction is None: col_construction = 2 if col_equipment is None: col_equipment = 3 if col_installation is None: col_installation = 4 if col_other is None: col_other = 5 logger.info(f"[表格解析] 按位置推断列索引: 建筑={col_construction}, 设备={col_equipment}, 安装={col_installation}, 其他={col_other}") else: logger.warning(f"[表格解析] 列数不足({num_cols}<7)，无法按位置推断费用列") logger.info(f"[表格解析] 列索引: 序号={col_no}, 名称={col_name}, 建筑={col_construction}, 设备={col_equipment}, 安装={col_installation}, 其他={col_other}") logger.info(f"[表格解析] 表头范围: 行 {header_start_idx} - {header_end_idx}") # 从数据行开始解析（跳过表头行） data_rows = df.iloc[header_end_idx + 1:].reset_index(drop=True) result = [] for idx, row in data_rows.iterrows(): # 跳过空行 if row.isna().all(): continue # 提取各列数据 no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None name_val = row.iloc[col_name] if col_name is not None and col_name < len(row) else None construction_val = row.iloc[col_construction] if col_construction is not None and col_construction < len(row) else None equipment_val = row.iloc[col_equipment] if col_equipment is not None and col_equipment < len(row) else None installation_val = row.iloc[col_installation] if col_installation is not None and col_installation < len(row) else None other_val = row.iloc[col_other] if col_other is not None and col_other < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): # 尝试中文数字 cleaned = no_str.replace(' ', '').replace('、', '').replace('.', '').replace('。', '') cleaned_no_brackets = cleaned.replace('（', '').replace('(', '').replace('）', '').replace(')', '') if cleaned_no_brackets in CHINESE_NUMBERS: no = CHINESE_NUMBERS[cleaned_no_brackets] # 解析工程或费用名称（提前解析，用于判断是否为"其中："行） expense_name = str(name_val).strip() if name_val is not None and not pd.isna(name_val) else "" # 跳过空行 if not expense_name: continue # 判断是否为"其中："开头的行（这类行序号通常为空，但需要保留） is_sub_item = expense_name.startswith("其中：") or expense_name.startswith("其中:") # 跳过序号为空或无效的行（但"其中："行除外） if (not no_str or pd.isna(no_val)) and not is_sub_item: continue # 跳过合计行 if any(kw in expense_name for kw in ["合计", "总计", "小计"]): continue # 判断层级（传入费用名称用于判断"其中："） level = determine_level(no_str, expense_name) # 去除"其中："前缀 clean_expense_name = expense_name if expense_name.startswith("其中："): clean_expense_name = expense_name[3:] elif expense_name.startswith("其中:"): clean_expense_name = expense_name[3:] # 解析费用金额 construction_cost = parse_number(construction_val) equipment_cost = parse_number(equipment_val) installation_cost = parse_number(installation_val) other_cost = parse_number(other_val) result.append({ "No": no if no is not None else idx + 1, "Level": level, "name": project_name, # 从标题提取的工程名称 "projectOrExpenseName": clean_expense_name, "constructionProjectCost": construction_cost, "equipmentPurchaseCost": equipment_cost, "installationProjectCost": installation_cost, "otherExpenses": other_cost, }) logger.info(f"[表格解析] 解析完成: 工程名称={project_name}, 共 {len(result)} 条数据") return result def parse_design_review_cost_table(df: pd.DataFrame, table_title: str) -> List[Dict[str, Any]]: """ 解析 designReview 类型的概算投资费用表格（规则3）。 表头格式： 序号 | 工程或费用名称 | 费用金额 | 各项占静态投资% | 单位投资万元/km Args: df: 表格 DataFrame table_title: 表格标题（如"周村 220kV 变电站新建工程总概算表"），用于提取工程名称 返回格式: [{ "No": int, # 序号 "Level": int, # 明细等级 "name": str, # 单项工程名称（从标题提取） "projectOrExpenseName": str, # 工程或费用名称 "cost": float, # 费用金额（元） }, ...] """ if df.empty: return [] # 从标题中提取工程名称 project_name = table_title if project_name: # 移除"总概算表"、"概算表"、"估算表"等后缀 project_name = re.sub(r'(总概算表|概算表|估算表|汇总表)$', '', project_name) # 移除多余空格 project_name = re.sub(r'\s+', '', project_name) # 中文数字映射（用于判断层级） CHINESE_NUMBERS = { '一': 1, '二': 2, '三': 3, '四': 4, '五': 5, '六': 6, '七': 7, '八': 8, '九': 9, '十': 10, '十一': 11, '十二': 12, '十三': 13, '十四': 14, '十五': 15, '十六': 16, '十七': 17, '十八': 18, '十九': 19, '二十': 20 } # 特殊的"其中："项，应作为 Level 1（独立大类）而非子项 SPECIAL_LEVEL1_ITEMS = [ "可抵扣固定资产增值税额", ] def determine_level(no_str: str, expense_name: str = "") -> int: """ 根据序号格式和费用名称判断层级： - 中文数字（一、二、三）：Level 1（大类） - 阿拉伯数字（1、2、3）：Level 1（大类） - 带括号的数字（(1)、（一））：Level 2（子项） - 以"其中："开头的行：Level 2（子项），但特殊项除外 """ no_str = str(no_str).strip() if no_str else "" expense_name = str(expense_name).strip() if expense_name else "" # 移除"其中："前缀用于判断 name_without_prefix = expense_name if expense_name.startswith("其中："): name_without_prefix = expense_name[3:] elif expense_name.startswith("其中:"): name_without_prefix = expense_name[3:] # 检查是否为特殊的 Level 1 项 for special_item in SPECIAL_LEVEL1_ITEMS: if special_item in name_without_prefix: return 1 # 以"其中："或"其中:"开头的 -> Level 2（子项） if expense_name.startswith("其中：") or expense_name.startswith("其中:"): return 2 # 序号为空的情况，默认 Level 1 if not no_str: return 1 # 带括号的 -> Level 2（子项） if '（' in no_str or '(' in no_str or '）' in no_str or ')' in no_str: return 2 # 中文数字（一、二、三等）-> Level 1（大类） cleaned = no_str.replace(' ', '').replace('、', '').replace('.', '').replace('。', '') for chinese_num in CHINESE_NUMBERS.keys(): if cleaned == chinese_num or cleaned.startswith(chinese_num): return 1 # 阿拉伯数字（1、2、3等）-> Level 1（大类） if re.match(r'^\d+\.?$', cleaned): return 1 return 1 def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str # 识别表头行 header_row_idx = None for i in range(min(5, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) row_text_no_space = row_text.replace(" ", "") if "工程或费用名称" in row_text_no_space and "费用金额" in row_text_no_space: header_row_idx = i break elif "序号" in row_text and "费用金额" in row_text: header_row_idx = i break if header_row_idx is None: header_row_idx = 0 logger.warning(f"[表格解析] 未找到明确表头，使用第一行作为表头") # 从表头行识别列索引 header_row = df.iloc[header_row_idx].astype(str).str.strip() col_no = None col_name = None col_cost = None for idx, cell in enumerate(header_row): cell_no_space = cell.replace(" ", "") if "序号" in cell_no_space or cell_no_space == "No": col_no = idx elif "工程或费用名称" in cell_no_space or "费用名称" in cell_no_space: col_name = idx elif "费用金额" in cell_no_space: col_cost = idx if col_no is None: col_no = 0 if col_name is None: col_name = 1 if col_cost is None: col_cost = 2 logger.info(f"[表格解析] 列索引: 序号={col_no}, 名称={col_name}, 费用金额={col_cost}") # 从数据行开始解析 data_rows = df.iloc[header_row_idx + 1:].reset_index(drop=True) result = [] for idx, row in data_rows.iterrows(): if row.isna().all(): continue no_val = row.iloc[col_no] if col_no is not None and col_no < len(row) else None name_val = row.iloc[col_name] if col_name is not None and col_name < len(row) else None cost_val = row.iloc[col_cost] if col_cost is not None and col_cost < len(row) else None # 解析序号 no = None no_str = "" if no_val is not None and not pd.isna(no_val): no_str = str(no_val).strip() if no_str: try: no = int(float(no_str)) except (ValueError, TypeError): cleaned = no_str.replace(' ', '').replace('、', '').replace('.', '').replace('。', '') cleaned_no_brackets = cleaned.replace('（', '').replace('(', '').replace('）', '').replace(')', '') if cleaned_no_brackets in CHINESE_NUMBERS: no = CHINESE_NUMBERS[cleaned_no_brackets] # 解析工程或费用名称（提前解析，用于判断是否为"其中："行） expense_name = str(name_val).strip() if name_val is not None and not pd.isna(name_val) else "" if not expense_name: continue # 判断是否为"其中："开头的行（这类行序号通常为空，但需要保留） is_sub_item = expense_name.startswith("其中：") or expense_name.startswith("其中:") # 跳过序号为空或无效的行（但"其中："行除外） if (not no_str or pd.isna(no_val)) and not is_sub_item: continue if any(kw in expense_name for kw in ["合计", "总计", "小计"]): continue # 判断层级（传入费用名称用于判断"其中："） level = determine_level(no_str, expense_name) # 去除"其中："前缀 clean_expense_name = expense_name if expense_name.startswith("其中："): clean_expense_name = expense_name[3:] elif expense_name.startswith("其中:"): clean_expense_name = expense_name[3:] cost = parse_number(cost_val) result.append({ "No": no if no is not None else idx + 1, "Level": level, "name": project_name, "projectOrExpenseName": clean_expense_name, "cost": cost, }) logger.info(f"[表格解析] 解析完成: 工程名称={project_name}, 共 {len(result)} 条数据") return result def _group_items_by_name(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ 将平铺的项目列表按 name 字段分组，并按 Level 嵌套： - Level 1 的项作为大类 - Level 2 的项作为大类的子项（放入 items 中） 输入: [ {"No": "一", "Level": 1, "name": "工程A", "projectOrExpenseName": "主辅生产工程", ...}, {"No": "(一)", "Level": 2, "name": "工程A", "projectOrExpenseName": "主要生产工程", ...}, {"No": "(二)", "Level": 2, "name": "工程A", "projectOrExpenseName": "辅助生产工程", ...}, {"No": "二", "Level": 1, "name": "工程A", "projectOrExpenseName": "其他费用", ...}, {"No": "", "Level": 2, "name": "工程A", "projectOrExpenseName": "其中：建设场地征用", ...}, ] 输出: [ { "name": "工程A", "items": [ { "No": "一", "Level": 1, "projectOrExpenseName": "主辅生产工程", ..., "items": [ {"No": "(一)", "Level": 2, "projectOrExpenseName": "主要生产工程", ...}, {"No": "(二)", "Level": 2, "projectOrExpenseName": "辅助生产工程", ...}, ] }, { "No": "二", "Level": 1, "projectOrExpenseName": "其他费用", ..., "items": [ {"No": "", "Level": 2, "projectOrExpenseName": "其中：建设场地征用", ...}, ] } ] } ] """ if not items: return [] from collections import OrderedDict # 第一步：按工程名称（name）分组 grouped_by_name: OrderedDict[str, List[Dict[str, Any]]] = OrderedDict() for item in items: name = item.get("name", "未知工程") if name not in grouped_by_name: grouped_by_name[name] = [] # 复制 item 并移除 name 字段 item_copy = {k: v for k, v in item.items() if k != "name"} grouped_by_name[name].append(item_copy) # 第二步：在每个工程组内，按 Level 建立父子关系 result = [] for name, group_items in grouped_by_name.items(): nested_items = [] current_parent = None for item in group_items: level = item.get("Level", 1) if level == 1: # Level 1 是大类，创建新的父项 item_with_children = dict(item) item_with_children["items"] = [] nested_items.append(item_with_children) current_parent = item_with_children elif level == 2: # Level 2 是子项，放入当前父项的 items 中 if current_parent is not None: # 移除 Level 字段（子项统一在父项下，不需要重复标识） child_item = {k: v for k, v in item.items()} current_parent["items"].append(child_item) else: # 没有父项，作为独立项处理 nested_items.append(item) else: # 其他 Level，作为独立项 nested_items.append(item) result.append({ "name": name, "items": nested_items }) return result def parse_final_account_table(df: pd.DataFrame, project_name: str, project_no: int) -> List[Dict[str, Any]]: """ 解析 finalAccount 类型的单项工程投资完成情况表格。 表格结构： 费用项目 | 概算金额 | 决算金额(审定金额-不含税) | 增值税额 | 超(-)节(+)支金额 | 超(-)节(+)支率 Args: df: 表格 DataFrame project_name: 项目名称（从标题提取，如"周村220kV输变电工程变电站新建工程"） project_no: 项目序号（如1、2、3、4） 返回格式: [{ "No": int, # 序号（项目序号） "name": str, # 项目名称（审计内容） "feeName": str, # 费用项目 "estimatedCost": str, # 概算金额 "approvedFinalAccountExcludingVat": str, # 决算金额审定不含税 "vatAmount": str, # 增值税额 "costVariance": str, # 超节支金额 "varianceRate": str, # 超节支率 }, ...] """ if df.empty: return [] def parse_number(value: Any) -> str: """解析数字，返回字符串保留原始精度""" if pd.isna(value): return "0" value_str = str(value).strip() # 移除非数字字符（保留负号和小数点） value_str = re.sub(r'[^\d.\-]', '', value_str) if not value_str or value_str == '-': return "0" return value_str def parse_rate(value: Any) -> str: """解析百分比，返回字符串""" if pd.isna(value): return "0%" value_str = str(value).strip() # 移除非数字字符（保留负号、小数点和百分号） if '%' not in value_str: # 提取数字部分并添加百分号 num_str = re.sub(r'[^\d.\-]', '', value_str) if num_str and num_str != '-': return f"{num_str}%" return "0%" return value_str # 识别表头行 header_row_idx = None for i in range(min(5, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) row_text_no_space = row_text.replace(" ", "") # 匹配表头特征 if "费用项目" in row_text_no_space and "概算金额" in row_text_no_space: header_row_idx = i break elif "概算金额" in row_text_no_space and "决算金额" in row_text_no_space: header_row_idx = i break if header_row_idx is None: # 检查是否有多行表头（如"决算金额"跨行） for i in range(min(3, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) if "1" in row_text and "2" in row_text and "3" in row_text: # 这是列序号行，表头在前面 header_row_idx = i break if header_row_idx is None: header_row_idx = 0 logger.warning(f"[表格解析] 未找到明确表头，使用第一行作为表头") # 从表头行识别列索引 header_row = df.iloc[header_row_idx].astype(str).str.strip() col_fee_name = None # 费用项目 col_estimated = None # 概算金额 col_approved = None # 审定金额(不含税) col_vat = None # 增值税额 col_variance = None # 超节支金额 col_variance_rate = None # 超节支率 for idx, cell in enumerate(header_row): cell_no_space = cell.replace(" ", "") if "费用项目" in cell_no_space: col_fee_name = idx elif "概算金额" in cell_no_space or cell_no_space == "1": col_estimated = idx elif "审定金额" in cell_no_space or "不含税" in cell_no_space or cell_no_space == "2": col_approved = idx elif "增值税额" in cell_no_space or "增值税" in cell_no_space or cell_no_space == "3": col_vat = idx elif "超" in cell_no_space and "节" in cell_no_space and "金额" in cell_no_space: col_variance = idx elif cell_no_space == "4=1-2-3": col_variance = idx elif "超" in cell_no_space and "节" in cell_no_space and "率" in cell_no_space: col_variance_rate = idx elif cell_no_space == "5=4/1": col_variance_rate = idx # 如果无法识别列，使用默认索引（根据OCR输出的表格结构） if col_fee_name is None: col_fee_name = 0 if col_estimated is None: col_estimated = 1 if col_approved is None: col_approved = 2 if col_vat is None: col_vat = 3 if col_variance is None: col_variance = 4 if col_variance_rate is None: col_variance_rate = 5 logger.info(f"[表格解析] 列索引: 费用项目={col_fee_name}, 概算金额={col_estimated}, " f"审定不含税={col_approved}, 增值税={col_vat}, 超节支金额={col_variance}, 超节支率={col_variance_rate}") # 跳过表头行（可能有多行表头） # 找到数据开始行（第一个包含"建筑安装"或"设备购置"的行） data_start_idx = header_row_idx + 1 for i in range(header_row_idx + 1, min(header_row_idx + 5, len(df))): row_text = " ".join(df.iloc[i].astype(str).str.strip().tolist()) if "建筑安装" in row_text or "设备购置" in row_text or "其他费用" in row_text: data_start_idx = i break # 跳过列序号行（如 "1 | 2 | 3 | 4=1-2-3 | 5=4/1"） if re.match(r'^[\d\s=\-/]+$', row_text.replace(" ", "")): continue # 从数据行开始解析 data_rows = df.iloc[data_start_idx:].reset_index(drop=True) result = [] for idx, row in data_rows.iterrows(): if row.isna().all(): continue fee_name = row.iloc[col_fee_name] if col_fee_name < len(row) else None estimated = row.iloc[col_estimated] if col_estimated < len(row) else None approved = row.iloc[col_approved] if col_approved < len(row) else None vat = row.iloc[col_vat] if col_vat < len(row) else None variance = row.iloc[col_variance] if col_variance < len(row) else None variance_rate = row.iloc[col_variance_rate] if col_variance_rate < len(row) else None # 解析费用项目名称 fee_name_str = str(fee_name).strip() if fee_name is not None and not pd.isna(fee_name) else "" if not fee_name_str: continue # 跳过合计行 if any(kw in fee_name_str for kw in ["合计", "总计", "小计"]): continue # 只保留主要费用项目：建筑安装工程、设备购置、其他费用 valid_fee_names = ["建筑安装工程", "建筑安装", "设备购置", "其他费用"] is_valid = any(kw in fee_name_str for kw in valid_fee_names) if not is_valid: continue result.append({ "No": project_no, "name": project_name, "feeName": fee_name_str, "estimatedCost": parse_number(estimated), "approvedFinalAccountExcludingVat": parse_number(approved), "vatAmount": parse_number(vat), "costVariance": parse_number(variance), "varianceRate": parse_rate(variance_rate), }) logger.info(f"[表格解析] 解析完成: 项目名称={project_name}, 共 {len(result)} 条数据") return result def parse_settlement_report_tables( merged_tables: List[Tuple[int, pd.DataFrame, str, int]] ) -> Dict[str, List[Dict[str, Any]]]: """ 解析 settlementReport 类型的所有表格，按表名组织返回。 返回格式: { "审定结算汇总表": [...], "合同执行情况": [], "赔偿合同": [], "物资采购合同1": [], "物资采购合同2": [], "其他服务类合同": [], } """ result = { "审定结算汇总表": [], "合同执行情况": [], "赔偿合同": [], "物资采购合同1": [], "物资采购合同2": [], "其他服务类合同": [], } for orig_idx, df, rule_name, page in merged_tables: try: logger.info(f"[表格解析] 开始解析表格: {rule_name} (页面 {page}, 行数: {len(df)})") if rule_name == "审定结算汇总表": parsed_data = parse_settlement_summary_table(df) if parsed_data: result[rule_name] = parsed_data logger.info(f"[表格解析] {rule_name}: 解析成功，共 {len(parsed_data)} 条数据") elif rule_name == "合同执行情况": parsed_data = parse_contract_execution_table(df) if parsed_data: result[rule_name] = parsed_data logger.info(f"[表格解析] {rule_name}: 解析成功，共 {len(parsed_data)} 条数据") elif rule_name == "赔偿合同": parsed_data = parse_compensation_contract_table(df) if parsed_data: result[rule_name] = parsed_data logger.info(f"[表格解析] {rule_name}: 解析成功，共 {len(parsed_data)} 条数据") elif rule_name == "物资采购合同1": parsed_data = parse_material_purchase_contract1_table(df) if parsed_data: result[rule_name] = parsed_data logger.info(f"[表格解析] {rule_name}: 解析成功，共 {len(parsed_data)} 条数据") elif rule_name == "物资采购合同2": parsed_data = parse_material_purchase_contract2_table(df) if parsed_data: result[rule_name] = parsed_data logger.info(f"[表格解析] {rule_name}: 解析成功，共 {len(parsed_data)} 条数据") elif rule_name == "其他服务类合同": parsed_data = parse_other_service_contract_table(df) if parsed_data: result[rule_name] = parsed_data logger.info(f"[表格解析] {rule_name}: 解析成功，共 {len(parsed_data)} 条数据") else: logger.warning(f"[表格解析] 未知的表格类型: {rule_name}") except Exception as e: # 如果解析失败，记录错误但不影响其他表格 logger.warning(f"[表格解析] 解析 {rule_name} 表格失败: {e}", exc_info=True) return result def extract_and_filter_tables_for_pdf( pdf_path: str, base_output_dir: str, doc_type: Literal["settlementReport", "designReview", "finalAccount"], ) -> Dict[str, Any]: """ 从指定 PDF 提取所有表格 + 合并后的表格 + 筛选后的表格，全部落盘。 返回结构: { "tables_root": str, "extracted_dir": str, "merged_dir": str, "filtered_dir": str, "all_tables": [ {"page": int, "index_on_page": int, "excel_path": str} ], "merged_tables": [ {"page": int, "index_on_page": int, "excel_path": str} ], "filtered_tables": [ {"page": int, "index_on_page": int, "rule_name": str, "excel_path": str} ], } """ pdf_path_obj = Path(pdf_path) base_output = Path(base_output_dir) tables_root = base_output / "tables" extracted_dir = tables_root / "extracted_tables" merged_dir = tables_root / "merged_tables" filtered_dir = tables_root / "filtered_tables" extracted_dir.mkdir(parents=True, exist_ok=True) merged_dir.mkdir(parents=True, exist_ok=True) filtered_dir.mkdir(parents=True, exist_ok=True) # 立即输出日志，确保日志系统正常工作 logger.info(f"[表格提取] ========== 开始处理 PDF ==========") logger.info(f"[表格提取] PDF 路径: {pdf_path}") logger.info(f"[表格提取] 文档类型: {doc_type}") logger.info(f"[表格提取] PDF 文件存在: {Path(pdf_path).exists()}") if Path(pdf_path).exists(): logger.info(f"[表格提取] PDF 文件大小: {Path(pdf_path).stat().st_size} bytes") logger.info(f"[表格提取] 输出目录: {base_output_dir}") # 1. 使用 pdfplumber 从 PDF 提取所有表格（不限制页数） # 对于 designReview 类型，启用标题提取（用于识别多个概算表） extract_titles = (doc_type == "designReview") logger.info("[表格提取] 步骤1: 使用 pdfplumber 提取所有表格...") tables_data = extract_tables_with_pdfplumber(str(pdf_path_obj), pages="all", extract_titles=extract_titles) logger.info(f"[表格提取] 步骤1完成: 共提取到 {len(tables_data)} 个表格") # 2. 保存所有原始表格为 xlsx，命名: table_page{page}_{index}.xlsx logger.info("[表格提取] 步骤2: 保存所有原始表格到 extracted_tables...") page_table_count: Dict[int, int] = {} all_tables_meta: List[Dict[str, Any]] = [] # 存储表格标题映射（用于后续解析） table_titles: Dict[int, str] = {} # orig_idx -> title # 给每个表格一个全局索引，方便后续合并/去重 all_tables: List[Tuple[int, pd.DataFrame, int]] = [] # (orig_idx, df, page) for orig_idx, (page, df, _bbox, title) in enumerate(tables_data): # 保存标题映射 if title: table_titles[orig_idx] = title page_table_count[page] = page_table_count.get(page, 0) + 1 idx_on_page = page_table_count[page] excel_path = extracted_dir / f"table_page{page}_{idx_on_page}.xlsx" df.to_excel(str(excel_path), index=False, header=False) all_tables_meta.append( { "page": page, "index_on_page": idx_on_page, "excel_path": str(excel_path), } ) all_tables.append((orig_idx, df.copy(), page)) logger.info(f"[表格提取] 步骤2完成: 已保存 {len(all_tables)} 个原始表格") # 3. 合并所有跨页表格（不进行过滤），保存到 merged_tables logger.info("[表格提取] 步骤3: 合并跨页表格...") merged_all_tables = _merge_all_tables(all_tables) logger.info(f"[表格提取] 步骤3完成: 从 {len(all_tables)} 个原始表格合并为 {len(merged_all_tables)} 个表格") merged_page_table_count: Dict[int, int] = {} merged_meta: List[Dict[str, Any]] = [] logger.info("[表格提取] 步骤3.1: 保存合并后的表格到 merged_tables...") for merged_idx, (orig_idx, df, page) in enumerate(merged_all_tables): merged_page_table_count[page] = merged_page_table_count.get(page, 0) + 1 idx_on_page = merged_page_table_count[page] excel_path = merged_dir / f"table_{merged_idx + 1}.xlsx" df.to_excel(str(excel_path), index=False, header=False) merged_meta.append( { "page": page, "index_on_page": idx_on_page, "excel_path": str(excel_path), } ) logger.info(f"[表格提取] 步骤3.1完成: 已保存 {len(merged_meta)} 个合并后的表格") # 4. 根据 doc_type 选择对应的表头规则 logger.info(f"[表格提取] 步骤4: 加载表头规则，文档类型: {doc_type}") header_rules = TABLE_TYPE_RULES.get(doc_type, []) logger.info(f"[表格提取] 步骤4完成: 找到 {len(header_rules)} 个表头规则") # 如果没有规则，直接返回（保留 extracted_tables 和 merged_tables） if not header_rules: logger.warning("[表格提取] 未找到表头规则，跳过筛选步骤") return { "tables_root": str(tables_root), "extracted_dir": str(extracted_dir), "merged_dir": str(merged_dir), "filtered_dir": str(filtered_dir), "all_tables": all_tables_meta, "merged_tables": merged_meta, "filtered_tables": [], } # 5. 从合并后的表格中过滤出匹配规则的表格 logger.info("[表格提取] 步骤5: 从合并后的表格中筛选匹配规则的表格...") # 增加 title 字段: (orig_idx, df, rule_name, page, title) matched_for_merge: List[Tuple[int, pd.DataFrame, str, int, str]] = [] for merged_idx, (orig_idx, df, page) in enumerate(merged_all_tables): rule_name: Optional[str] = None for rule in header_rules: is_match, rn = check_table_header(df, rule) if is_match: rule_name = rn # 获取表格标题 title = table_titles.get(orig_idx, "") logger.info(f"[表格提取] 表格 {merged_idx + 1} (页面 {page}) 匹配规则: {rule_name}, 标题: {title}") break if rule_name: title = table_titles.get(orig_idx, "") matched_for_merge.append((orig_idx, df.copy(), rule_name, page, title)) logger.info(f"[表格提取] 步骤5完成: 共匹配到 {len(matched_for_merge)} 个表格") # 如果没有匹配到表格，直接返回（保留 extracted_tables 和 merged_tables） if not matched_for_merge: logger.warning("[表格提取] 未匹配到任何表格，跳过后续处理") return { "tables_root": str(tables_root), "extracted_dir": str(extracted_dir), "merged_dir": str(merged_dir), "filtered_dir": str(filtered_dir), "all_tables": all_tables_meta, "merged_tables": merged_meta, "filtered_tables": [], } # 6. 对已匹配规则的表格再次进行跨页合并（处理规则匹配后的特殊情况） logger.info("[表格提取] 步骤6: 对已匹配规则的表格进行跨页合并...") merged_tables = _merge_cross_page_tables(matched_for_merge, header_rules) logger.info(f"[表格提取] 步骤6完成: 跨页合并后剩余 {len(merged_tables)} 个表格") # 7. 保存筛选+合并后的表格到 filtered_dir，命名仍然按 page + 序号 logger.info("[表格提取] 步骤7: 保存筛选+合并后的表格到 filtered_tables...") filtered_page_table_count: Dict[int, int] = {} filtered_meta: List[Dict[str, Any]] = [] parsed_data = None for orig_idx, df, rule_name, page, title in merged_tables: filtered_page_table_count[page] = filtered_page_table_count.get(page, 0) + 1 idx_on_page = filtered_page_table_count[page] excel_path = filtered_dir / f"table_page{page}_{idx_on_page}.xlsx" df.to_excel(str(excel_path), index=False, header=False) filtered_meta.append( { "page": page, "index_on_page": idx_on_page, "rule_name": rule_name, "title": title, "excel_path": str(excel_path), } ) logger.info(f"[表格提取] 步骤7完成: 已保存 {len(filtered_meta)} 个筛选后的表格") # 8. 根据文档类型解析表格数据 logger.info(f"[表格提取] 步骤8: 解析表格数据，文档类型: {doc_type}...") logger.info(f"[表格提取] 待解析的表格列表: {[(rule_name, page, title) for _, _, rule_name, page, title in merged_tables]}") if doc_type == "designReview": # 对于 designReview 类型，返回类似 settlementReport 的结构 # 按规则类型分组： # - 初设评审的概算投资（规则1）: 嵌套结构 # - 初设评审的概算投资明细（规则2）: 平铺结构，多个表格 # - 初设评审的概算投资费用（规则3）: 平铺结构，多个表格 parsed_data = { "初设评审的概算投资": [], "初设评审的概算投资明细": [], "初设评审的概算投资费用": [], } for orig_idx, df, rule_name, page, title in merged_tables: if rule_name == "初设评审的概算投资": # 规则1：嵌套结构 try: logger.info(f"[表格提取] 解析 designReview 表格(规则1): {rule_name} (页面 {page}, 行数: {len(df)})") summary_data = parse_design_review_table(df) if summary_data: parsed_data["初设评审的概算投资"] = summary_data logger.info(f"[表格提取] 解析完成: 共 {len(summary_data)} 条数据") except Exception as e: logger.warning(f"[表格提取] 解析 designReview 表格(规则1)失败: {e}", exc_info=True) elif rule_name == "初设评审的概算投资明细": # 规则2：使用标题作为工程名称 try: logger.info(f"[表格提取] 解析 designReview 表格(规则2): {rule_name} (页面 {page}, 标题: {title}, 行数: {len(df)})") detail_data = parse_design_review_detail_table(df, title) if detail_data: parsed_data["初设评审的概算投资明细"].extend(detail_data) logger.info(f"[表格提取] 解析完成: 共 {len(detail_data)} 条数据") except Exception as e: logger.warning(f"[表格提取] 解析 designReview 表格(规则2)失败: {e}", exc_info=True) elif rule_name == "初设评审的概算投资费用": # 规则3：使用标题作为工程名称 try: logger.info(f"[表格提取] 解析 designReview 表格(规则3): {rule_name} (页面 {page}, 标题: {title}, 行数: {len(df)})") cost_data = parse_design_review_cost_table(df, title) if cost_data: parsed_data["初设评审的概算投资费用"].extend(cost_data) logger.info(f"[表格提取] 解析完成: 共 {len(cost_data)} 条数据") except Exception as e: logger.warning(f"[表格提取] 解析 designReview 表格(规则3)失败: {e}", exc_info=True) # 将规则2和规则3的平铺结果按工程名称分组为嵌套结构 for rule_key in ["初设评审的概算投资明细", "初设评审的概算投资费用"]: if rule_key in parsed_data and parsed_data[rule_key]: flat_items = parsed_data[rule_key] grouped = _group_items_by_name(flat_items) parsed_data[rule_key] = grouped # 统计解析结果 logger.info(f"[表格提取] 解析结果统计:") total_records = 0 for table_type, table_data in parsed_data.items(): if table_data: if isinstance(table_data, list): # 嵌套结构，统计所有 items record_count = sum(len(item.get("items", [])) for item in table_data) if table_data and isinstance(table_data[0], dict) and "items" in table_data[0] else len(table_data) else: record_count = len(table_data) total_records += record_count logger.info(f"[表格提取] - {table_type}: {len(table_data)} 个工程，共 {record_count} 条明细") else: logger.info(f"[表格提取] - {table_type}: 未匹配到数据") logger.info(f"[表格提取] 总计: {total_records} 条数据") elif doc_type == "settlementReport": # 对于 settlementReport 类型，解析所有匹配的表格，按表名组织 try: logger.info(f"[表格提取] 解析 settlementReport 表格，共 {len(merged_tables)} 个表格") # 转换为4元组格式以兼容现有的 parse_settlement_report_tables 函数 tables_4tuple = [(orig_idx, df, rule_name, page) for orig_idx, df, rule_name, page, title in merged_tables] parsed_data = parse_settlement_report_tables(tables_4tuple) # 统计每个表的解析结果 logger.info(f"[表格提取] 解析结果统计:") total_records = 0 for table_name, table_data in parsed_data.items(): if table_data: record_count = len(table_data) total_records += record_count logger.info(f"[表格提取] - {table_name}: {record_count} 条数据") else: logger.info(f"[表格提取] - {table_name}: 未匹配到数据") logger.info(f"[表格提取] 总计: {total_records} 条数据") except Exception as e: logger.warning(f"[表格提取] 解析 settlementReport 表格失败: {e}", exc_info=True) logger.info("[表格提取] 步骤8完成: 表格数据解析完成") result = { "tables_root": str(tables_root), "extracted_dir": str(extracted_dir), "merged_dir": str(merged_dir), "filtered_dir": str(filtered_dir), "all_tables": all_tables_meta, "merged_tables": merged_meta, "filtered_tables": filtered_meta, } # 添加解析后的 JSON 数据 if parsed_data is not None: result["parsed_data"] = parsed_data result["parsed_data_json"] = json.dumps(parsed_data, ensure_ascii=False, indent=2) logger.info("[表格提取] JSON 数据已生成") logger.info(f"[表格提取] 处理完成: 原始表格 {len(all_tables_meta)} 个, 合并后 {len(merged_meta)} 个, 筛选后 {len(filtered_meta)} 个") return result