""" 表格结构提取器（V2 新增） 负责： 1. 从招标 PDF 提取完整表格（单元格级，含合并信息）→ List[ExtractedTable] 2. 从模板 DOCX 提取所有表格结构（含占位符检测）→ List[ExtractedTable] 3. 从参考投标文件提取段落信息，用于标题层级推断 设计哲学：逻辑提取为主，LLM 按需辅助。 """ from __future__ import annotations import logging import os import re from typing import Dict, List, Optional, Set, Tuple from docx import Document as DocxDocument from models import ExtractedTable, TableCell, SkeletonParagraph logger = logging.getLogger(__name__) # “投标报价分项明细表”在参考投书中不是一张表，而是由 # “三、投标报价分项明细表”等文字和 明细1..明细8、附表1 多张物理表组成的完整片段。 # 提取时必须按原文顺序保留整段文字和所有表，写成一个单独的 DOCX 片段。 _BID_DETAIL_GROUP_NAME = "投标报价分项明细表" _BID_DETAIL_ARTIFACT_NAME = f"{_BID_DETAIL_GROUP_NAME}.docx" _BID_DETAIL_SECTION_RE = re.compile( r"(明细\s*[一二三四五六七八九十百\d]+\s*[：:]\s*[^|]{0,80})" ) _BID_DETAIL_APPENDIX_RE = re.compile( r"(附表\s*\d*\s*[：:]\s*[^|]{0,80})" ) _REFERENCE_ARTIFACT_RE = re.compile(r"^\d{2,}_.*\.docx$", re.IGNORECASE) def _bid_detail_section_label(table: ExtractedTable) -> str: """从表格标题/说明中取出“明细N：...”或“附表N：...”分节标签。""" text = " | ".join( value.strip() for value in ( getattr(table, "caption", "") or "", getattr(table, "title_hint", "") or "", ) if value and value.strip() ) if not text: return "" for pattern in (_BID_DETAIL_SECTION_RE, _BID_DETAIL_APPENDIX_RE): match = pattern.search(text) if match: return match.group(1).strip() return "" def _is_bid_detail_group_start(table: ExtractedTable) -> bool: """判断某张参考表是否可作为“投标报价分项明细表”合并序列的起点。""" text = " ".join( str(getattr(table, attr, "") or "") for attr in ("title_hint", "caption") ) return ( "投标报价分项明细表" in text or "报价分项明细表" in text or "投标报价分项明细" in text ) def _find_bid_detail_group( tables: List[ExtractedTable], ) -> Tuple[Optional[int], Optional[int]]: """定位“投标报价分项明细表”连续物理表区间。""" if not tables: return None, None start_index = next( ( index for index, table in enumerate(tables) if _is_bid_detail_group_start(table) ), None, ) if start_index is None: return None, None end_index = start_index + 1 while end_index < len(tables) and _bid_detail_section_label(tables[end_index]): end_index += 1 return start_index, end_index def _build_reference_fragment_impl( docx_path: str, start_idx: int, end_idx: int, ) -> List[bytes]: """流式截取参考 DOCX 正文中一段连续“段落 + 表格”的 OOXML 字节。""" import zipfile import xml.etree.ElementTree as ET w = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}" items: List[bytes] = [] with zipfile.ZipFile(docx_path) as archive: with archive.open("word/document.xml") as xml_stream: stack: List[str] = [] body_depth: Optional[int] = None child_index = -1 for event, node in ET.iterparse( xml_stream, events=("start", "end") ): if event == "start": stack.append(node.tag) if node.tag == f"{w}body": body_depth = len(stack) continue is_body_child = ( body_depth is not None and len(stack) == body_depth + 1 ) if is_body_child and node.tag in (f"{w}p", f"{w}tbl"): child_index += 1 if start_idx <= child_index <= end_idx: items.append(ET.tostring(node, encoding="utf-8")) node.clear() if child_index > end_idx: break if node.tag == f"{w}body": body_depth = None stack.pop() return items def _build_reference_fragment( docx_path: str, start_idx: int, end_idx: int, ) -> List[bytes]: """隔离执行参考 DOCX 片段截取，原生崩溃时安全返回空列表。""" result = _run_reference_xml_worker( "fragment", docx_path, start_idx, end_idx ) return result if isinstance(result, list) else [] def _build_bid_detail_fragment( tables: List[ExtractedTable], docx_path: str, ) -> List[ExtractedTable]: """把参考投书中“投标报价分项明细表”的文字 + 多张物理表保留为完整片段。""" if not tables or not docx_path or not os.path.isfile(docx_path): return tables start_index, end_index = _find_bid_detail_group(tables) if start_index is None or end_index is None: return tables group = tables[start_index:end_index] first = group[0] stream = _build_reference_doc_stream(docx_path) table_stream_positions = [ index for index, item in enumerate(stream) if item.get("kind") == "tbl" ] if not table_stream_positions: return tables first_table_idx = int(getattr(first, "para_idx", -1)) if not (0 <= first_table_idx < len(table_stream_positions)): return tables first_stream = table_stream_positions[first_table_idx] # 从第一张明细表往前找“三、投标报价分项明细表”标题，把它一并纳入片段。 start_stream = first_stream for index in range(first_stream - 1, max(0, first_stream - 10) - 1, -1): text = str(stream[index].get("text", "") or "") if "投标报价分项明细表" in text or "报价分项明细表" in text: start_stream = index break last_table_idx = int(getattr(group[-1], "para_idx", -1)) if not (0 <= last_table_idx < len(table_stream_positions)): return tables last_stream = table_stream_positions[last_table_idx] fragment_xml = _build_reference_fragment( docx_path, start_stream, last_stream ) if not fragment_xml: return tables fragment = ExtractedTable( table_id=f"{getattr(first, 'table_id', '') or 'REF-BID-DETAIL'}-FRAGMENT", source_file=docx_path, source_type="reference_docx", para_idx=first_table_idx, rows=0, cols=0, cells=[], caption=_BID_DETAIL_GROUP_NAME, title_hint=_BID_DETAIL_GROUP_NAME, table_type="bid_detail_table", ) fragment.is_fragment = True fragment.fragment_xml = fragment_xml fragment.fragment_source_indices = [ int(getattr(table, "para_idx", -1)) for table in group ] logger.info( f"参考投书“{_BID_DETAIL_GROUP_NAME}”片段: " f"{len(fragment_xml)} 个段落/表格，来源物理表索引 " f"{fragment.fragment_source_indices}" ) return [ *tables[:start_index], fragment, *tables[end_index:], ] def _write_fragment_docx(fragment_xml: List[bytes], path: str) -> str: """把“文字 + 多张表”片段原样写入单个 DOCX。""" os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) doc = DocxDocument() from docx.oxml import parse_xml for xml_bytes in fragment_xml or []: try: element = parse_xml(xml_bytes) doc.element.body.insert_element_before(element, "w:sectPr") except Exception as exc: logger.warning(f"片段元素写入失败，跳过: {exc}") doc.save(path) return os.path.abspath(path) # ============================================================ # 招标表格分类（供提示词+代码方式提取结果回填分类） # # 说明：招标 PDF 的表格/函/表提取已统一走 document_parser 的 # extract_items_from_tender（LLM 识别 + pdf_table_to_docx 结构匹配）， # 本模块不再提供独立的纯结构提取入口，仅保留模板 DOCX 表格提取、 # 参考投标骨架段落提取，以及招标表格分类等共享逻辑。 # ============================================================ # 招标表格分类关键词（按优先级排序：响应表 > 评分 > 格式函件 > 参考格式） _TENDER_TABLE_TYPE_KEYWORDS: Dict[str, List[str]] = { "scoring": [ "评分", "分值", "评审项目", "评审内容", "评标标准", "评分办法", "综合评分", "打分", ], "qualification_response": [ "资格条件", "资格要求", "资格性审查", "具备条件说明", "投标人资格", "符合性审查", ], "substantive_response": [ "实质性要求", "实质性响应", "带★", "★号", "★", "必须满足", "不可偏离", ], "bid_form": [ "投标函", "开标一览表", "法定代表人授权", "法定代表人证明", "法定代表人", "投标总价", "授权委托书", "报价表", ], "reference_format": [ "类似项目", "项目负责人", "人员配备", "项目经理", "简历表", "项目业绩", "业绩", ], } def _classify_tender_table( cells: List[List[TableCell]], title_hint: str = "", ) -> str: """基于单元格内容 + 表格上方标题自动分类招标表格类型 分类策略： 1. 表格标题优先：标题明确含"实质性要求响应表"/"资格条件响应表"时 直接判定（表头可能是通用的"项目内容/具备的条件说明（要求）"， 无法区分两类响应表，必须依赖标题） 2. 关键词兜底：评分 > 实质性 > 资格 > 格式函件 > 参考格式 > other （"实质性要求"优先级高于"具备的条件说明"，避免实质性响应表 被资格类关键词抢走——两表表头结构相同） """ all_text = " ".join( cell.text for row in cells for cell in row if cell.text.strip() ) hint_text = title_hint.replace(" ", "").replace("\n", "") or "" # ---- 1. 标题优先判定 ---- if hint_text: if "实质性要求响应表" in hint_text: return "substantive_response" if "资格条件响应表" in hint_text: return "qualification_response" if "客观分评审因素响应情况表" in hint_text: return "scoring" if "开标一览表" in hint_text: return "bid_open_table" if "投标报价汇总表" in hint_text or "报价汇总表" in hint_text: return "bid_summary_table" if "报价明细表" in hint_text or "报价分项明细表" in hint_text: return "bid_detail_table" if "投标函" in hint_text: return "bid_form" # ---- 2. 关键词兜底（实质性在资格之前：两表表头结构相同，内容均含 # "实质性要求/资格条件"字样，先匹配实质性的专属词） ---- priority_order = [ "scoring", "substantive_response", "qualification_response", "bid_form", "reference_format", ] for table_type in priority_order: for kw in _TENDER_TABLE_TYPE_KEYWORDS.get(table_type, []): if kw in all_text: return table_type return "other" # ============================================================ # 2. 模板 DOCX → ExtractedTable # ============================================================ def extract_tables_from_template_docx(docx_path: str) -> List[ExtractedTable]: """从模板 DOCX 中提取所有表格（含占位符标记） 使用 python-docx 遍历表格，保留： - 原始格式（字号、加粗、对齐） - 合并单元格信息 - %%...%% 占位符检测 Args: docx_path: 模板 DOCX 路径 Returns: ExtractedTable 列表 """ if not os.path.exists(docx_path): logger.warning(f"模板文件不存在: {docx_path}") return [] doc = DocxDocument(docx_path) extracted: List[ExtractedTable] = [] for table_idx, table in enumerate(doc.tables): cells, grid_cols = _extract_docx_table_grid(table) rows = len(cells) cols = grid_cols if rows == 0 or cols == 0: continue has_placeholders = any( re.search(r"%%(.+?)%%", cell.text) for row in cells for cell in row ) # 提取表格标题（表格前一两个段落的文本） caption = _extract_table_caption_from_xml(doc, table_idx) extracted.append(ExtractedTable( table_id=f"TEMPLATE-T{len(extracted) + 1:02d}", source_file=docx_path, source_type="template_docx", para_idx=table_idx, rows=rows, cols=cols, cells=cells, caption=caption, has_placeholders=has_placeholders, table_type=_classify_template_table(cells, caption), )) logger.info( f"模板表格提取完成: {len(extracted)} 个表格 " f"（来源: {os.path.basename(docx_path)}），" f"{sum(1 for t in extracted if t.has_placeholders)} 个含占位符" ) return extracted def _extract_docx_table_grid(table) -> Tuple[List[List[TableCell]], int]: """从 python-docx Table 提取完整单元格网格（保留合并单元格结构）。 原实现直接使用 ``table.rows[i].cells``，python-docx 会把合并单元格 在每行重复展开，导致 gridSpan/vMerge 的合并结构丢失（开标一览表 “报价总价（大写）：”跨 2 列、“人民币…元整”跨 3 列被拆开）。 本实现按 w:tbl 的 XML 原始结构逐行重建紧凑网格： - 普通单元格占 1 个网格列； - gridSpan 单元格占用 colspan 个网格列，仅在起始格保留文本， 其余网格位置填入空白 continuation 格； - vMerge 合并列：起始行记录 rowspan，后续行在该列填入 is_merged_continuation 空白格（不再重复文本）； - 返回的行数是物理行数（含仅作为 vMerge 延续的空行）， grid_cols 为表格网格总列数。 Returns: (cells, grid_cols) cells[row][col] 为 TableCell；合并起始格携带 colspan/rowspan， 被合并覆盖的网格位置为 blank continuation 格。 """ from docx.oxml.ns import qn tbl = getattr(table, "_element", None) if tbl is None: return [], 0 grid_cols = 0 grid = tbl.tblGrid if grid is not None: grid_cols = len(grid.findall(qn("w:gridCol"))) rows = tbl.findall(qn("w:tr")) if not rows: return [], grid_cols # 先扫一遍所有行，统计每行实际占用的网格列数（取最大值） max_cols = grid_cols if max_cols <= 0: for tr in rows: row_cols = 0 for tc in tr.findall(qn("w:tc")): tc_pr = tc.find(qn("w:tcPr")) colspan = 1 if tc_pr is not None: gs = tc_pr.find(qn("w:gridSpan")) if gs is not None: try: colspan = max(1, int(gs.get(qn("w:val")) or 1)) except (TypeError, ValueError): colspan = 1 row_cols += colspan max_cols = max(max_cols, row_cols) grid_cols = max_cols cells: List[List[TableCell]] = [] # 当前从上一行延续下来的纵向合并：grid_col -> 起始格信息 active_vmerge: Dict[int, TableCell] = {} for tr in rows: row_cells: List[TableCell] = [None] * grid_cols continued_origins: set = set() # 1) 纵向合并延续：被上一行 vMerge 覆盖的网格列先填 continuation 格 # （真实 XML 行若含 vMerge continue 会再次覆盖同一位置，无冲突； # 若某合并未延续，处理后会被移出 active_vmerge） for c, origin in active_vmerge.items(): if c >= grid_cols: continue for k in range(origin.colspan): cc = c + k if cc < grid_cols and row_cells[cc] is None: row_cells[cc] = TableCell( text="", row=len(cells), col=cc, rowspan=1, colspan=1, is_merged_origin=False, is_merged_continuation=True, ) # 2) 处理本行实际 w:tc（放在下一个空网格位置） col = 0 used_vmerge_cols: set = set() for tc in tr.findall(qn("w:tc")): tc_pr = tc.find(qn("w:tcPr")) colspan = 1 vmerge_val = "" if tc_pr is not None: gs = tc_pr.find(qn("w:gridSpan")) if gs is not None: try: colspan = max(1, int(gs.get(qn("w:val")) or 1)) except (TypeError, ValueError): colspan = 1 vm = tc_pr.find(qn("w:vMerge")) if vm is not None: vmerge_val = vm.get(qn("w:val")) or "continue" if vmerge_val == "continue": # 延续格必须对应到上一行 active_vmerge 预填的 continuation 格， # 不能因该格已被预填而跳到下一列，否则纵向合并会错位。 target_col = None for c in sorted(active_vmerge.keys()): if c in used_vmerge_cols or c >= grid_cols: continue if ( row_cells[c] is not None and row_cells[c].is_merged_continuation ): target_col = c break if target_col is None: while col < grid_cols and row_cells[col] is not None: col += 1 target_col = col if target_col < grid_cols: used_vmerge_cols.add(target_col) continued_origins.add(target_col) if row_cells[target_col] is None: row_cells[target_col] = TableCell( text="", row=len(cells), col=target_col, rowspan=1, colspan=1, is_merged_origin=False, is_merged_continuation=True, ) col = target_col + colspan continue while col < grid_cols and row_cells[col] is not None: col += 1 if col >= grid_cols: break text = "".join( node.text or "" for node in tc.iter(qn("w:t")) ) font_size, bold, alignment = _extract_tc_format(tc) # 防御：网格列数小于 colspan 和时截断，避免越界 if col + colspan > grid_cols: colspan = max(1, grid_cols - col) else: is_origin = vmerge_val in ("", "restart") # 起始行行号未知：rowspan 先记 1，扫完所有行后回填 cell = TableCell( text=text, row=len(cells), col=col, rowspan=1, colspan=colspan, is_merged_origin=(colspan > 1 or vmerge_val == "restart"), is_merged_continuation=False, font_size=font_size, bold=bold, alignment=alignment, ) for k in range(colspan): if k == 0: row_cells[col] = cell else: row_cells[col + k] = TableCell( text="", row=len(cells), col=col + k, rowspan=1, colspan=1, is_merged_origin=False, is_merged_continuation=True, ) if vmerge_val == "restart": active_vmerge[col] = cell col += colspan # 3) 本行未延续的纵向合并结束，移出 active_vmerge（后续行不再占位） for c in [c for c in active_vmerge if c not in continued_origins]: del active_vmerge[c] # 补齐行尾未覆盖的网格列（防御性） for c in range(grid_cols): if row_cells[c] is None: row_cells[c] = TableCell( text="", row=len(cells), col=c, rowspan=1, colspan=1, is_merged_origin=False, is_merged_continuation=True, ) cells.append(row_cells) # 回填 vMerge 起始格的实际行数：按起始格的 colspan 范围统计连续 # continuation 行数。这样既支持普通纵向合并，也支持“横向+纵向” # 混合合并，不把合并类型写死。 if cells: for ri, row in enumerate(cells): for ci, cell in enumerate(row): if not cell.is_merged_origin or cell.rowspan != 1: continue colspan = max(1, int(cell.colspan or 1)) rowspan = 1 for rr in range(ri + 1, len(cells)): all_cont = True for cc in range(ci, ci + colspan): if cc >= grid_cols: all_cont = False break target = cells[rr][cc] if not (target.is_merged_continuation or target.text == ""): all_cont = False break if not all_cont: break rowspan += 1 if rowspan > 1: cell.rowspan = rowspan cell.is_merged_origin = True return cells, grid_cols def _extract_tc_format(tc) -> Tuple[Optional[float], bool, str]: """从 w:tc 提取字号（pt）/加粗/段落对齐，尽力而为，失败返回默认值。""" from docx.oxml.ns import qn font_size: Optional[float] = None bold = False alignment = "" try: for r_node in tc.iter(qn("w:r")): r_pr = r_node.find(qn("w:rPr")) if r_pr is None: continue sz = r_pr.find(qn("w:sz")) if sz is not None and font_size is None: try: font_size = int(sz.get(qn("w:val")) or 0) / 2.0 except (TypeError, ValueError): pass b_node = r_pr.find(qn("w:b")) if b_node is not None: val = b_node.get(qn("w:val")) if val is None or val in ("1", "true", "on"): bold = True if not alignment: for p_node in tc.iter(qn("w:p")): p_pr = p_node.find(qn("w:pPr")) if p_pr is None: continue jc = p_pr.find(qn("w:jc")) if jc is not None: val = jc.get(qn("w:val")) or "" if val in ("left", "center", "right", "both", "distribute"): alignment = "center" if val == "distribute" else val break except Exception: pass return font_size, bold, alignment def _extract_table_caption_from_xml(doc: DocxDocument, table_idx: int) -> str: """从文档 XML 结构中提取表格前面的标题/说明文字（段落）""" try: body = doc.element.body table_elements = body.findall( "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}tbl" ) if table_idx >= len(table_elements): return "" tbl = table_elements[table_idx] prev = tbl.getprevious() recent = [] while prev is not None and len(recent) < 3: tag = prev.tag.split("}")[-1] if "}" in prev.tag else prev.tag if tag != "p": break para_text = "".join( node.text or "" for node in prev.iter() if node.tag.endswith("}t") and node.text ) if para_text.strip(): recent.append(para_text.strip()) else: break prev = prev.getprevious() caption = " | ".join(reversed(recent)).strip() return caption[:300] if caption else "" except Exception: return "" def _classify_template_table(cells: List[List[TableCell]], caption: str) -> str: """基于模板表格内容分类""" all_text = caption + " " + " ".join( cell.text for row in cells for cell in row if cell.text.strip() ) if any(kw in all_text for kw in ["开标一览表"]): return "bid_open_table" if any(kw in all_text for kw in ["报价汇总表"]): return "bid_summary_table" if any(kw in all_text for kw in ["报价明细表", "报价分项明细表"]): return "bid_detail_table" if any(kw in all_text for kw in ["投标函"]): return "bid_form" if any(kw in all_text for kw in ["资格条件", "资格要求", "资格性"]): return "qualification_response" if any(kw in all_text for kw in ["实质性要求", "实质性响应", "偏离"]): return "substantive_response" if any(kw in all_text for kw in ["类似项目", "业绩", "项目负责人", "项目经理"]): return "reference_format" if any(kw in all_text for kw in ["报价", "一览表", "汇总"]): return "bid_form" return "other" # ============================================================ # 参考投书表格 LLM 增强提取（V3 新增） # 结构提取之外，用 LLM 补充表格名称、所属章节、用途、表头项目信息字段， # 使 Step 4/6 在"招标/模板均无对应表"时能按名称从参考投书中取用表格。 # 处理单元：一张表 + 所在页（跨页表格整表为一个单元）+ 前文上下文， # 单元之间并发调用 LLM。 # ============================================================ _REFERENCE_TABLE_ENRICH_PROMPT = """你是投标文件表格识别专家。下方是参考投标文件中提取出的若干表格（每个表格给出： 编号、表格前后文提示、行列数、前两行内容预览）。请对每个表格输出其补充信息。 输出严格 JSON： {"tables": [ {"table_id": "与输入编号一致", "name": "表格正式名称（如：开标一览表、投标报价汇总表、资格条件响应表）", "chapter": "所属章节（如：第二章：投标报价）", "purpose": "用途，如：报价/响应/清单/证明/考核表", "header_fields": ["项目名称：...", "项目编号：...", "包号：..."], "column_notes": "各列含义简要说明", "can_reuse": true, "notes": "备注（含需要人工确认的事项）"} ]} 要求： 1. name 必须是标准表格名称，不含序号前缀（如"一、""（二）"）； 2. 表格前后文包含"开标一览表"等标题时，name 取该标题； 3. header_fields 提取表格上方的"项目名称：""项目编号：""包号："等项目信息行（原样）； 4. 没有把握的字段填空字符串，不得编造；can_reuse 默认 true； 5. 只输出 JSON，不输出任何其他文字。""" def _build_reference_table_preview(tables: List[ExtractedTable]) -> str: """把结构表格压缩成 LLM 可读的预览文本""" lines = [] for idx, t in enumerate(tables): caption = (getattr(t, "caption", "") or "").strip() title_hint = (getattr(t, "title_hint", "") or "").strip() ctx = " / ".join(x for x in (title_hint, caption) if x) rows = getattr(t, "rows", 0) or (len(t.cells) if t.cells else 0) cols = getattr(t, "cols", 0) if not cols and t.cells: cols = max(len(r) for r in t.cells) if t.cells else 0 preview_rows = [] for r in (t.cells or [])[:2]: cells = [c.text.replace("\n", " ")[:16] for c in r] preview_rows.append(" | ".join(cells)) lines.append( f"[{idx}] 上下文: {ctx[:120]}\n" f" 规格: {rows} 行 x {cols} 列\n" f" 预览:\n" + "\n".join(f" {r}" for r in preview_rows) ) return "\n\n".join(lines) # 无分页标记的 DOCX，按字符流估算页数（中文文档每页约 1000 字符） _CHARS_PER_PAGE_ESTIMATE = 1000 def _build_reference_doc_stream(docx_path: str) -> List[dict]: """在隔离子进程中构建参考 DOCX 正文流。 即使 ``ElementTree.iterparse`` 使用流式读取，Windows 上同一主进程先后 加载 PDF/ONNX/python-docx 后，Expat 仍可能直接触发 0xC0000005，无法由 Python ``try/except`` 捕获。隔离后原生崩溃只损失可选的表格上下文，不会 终止 Step6；失败时返回空流，LLM 仍可依据表格标题和单元格内容识别。 """ result = _run_reference_xml_worker("stream", docx_path) return result if isinstance(result, list) else [] def _build_reference_doc_stream_impl(docx_path: str) -> List[dict]: """用标准库 ET 流式解析 document.xml，按正文顺序输出段落/表格流。 返回 [{kind: "p"|"tbl", text: str, char_start: int, char_end: int}, ...] - 段落：拼接该段全部 w:t 文本； - 表格：拼接全部单元格文本（跨页表格仍是一个 w:tbl 元素，天然完整）。 char_start/char_end 用于估算表格所在页。 """ import zipfile import xml.etree.ElementTree as ET w = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}" stream = [] pos = 0 # 184 MB 级参考投书的 document.xml 解压后可能达到数百 MB。 # ``z.read`` + ``ET.fromstring`` 会同时保留 XML bytes 和完整元素树， # Windows/PyCharm 下可能在 Python 抛出 MemoryError 前先由 Expat 触发 # 0xC0000005。改用 ZipExtFile + iterparse，只保留当前 body 直属块。 with zipfile.ZipFile(docx_path) as z: with z.open("word/document.xml") as xml_stream: stack = [] body_depth = None for event, node in ET.iterparse( xml_stream, events=("start", "end") ): if event == "start": stack.append(node.tag) if node.tag == f"{w}body": body_depth = len(stack) continue is_body_child = ( body_depth is not None and len(stack) == body_depth + 1 ) if is_body_child and node.tag in (f"{w}p", f"{w}tbl"): texts = [t.text or "" for t in node.iter(f"{w}t")] if node.tag == f"{w}p": text = "".join(texts).strip() kind = "p" span = len(text) else: text = " ".join( value.strip() for value in texts if value.strip() ) kind = "tbl" span = max(len(text), 1) stream.append({ "kind": kind, "text": text, "char_start": pos, "char_end": pos + span, }) pos += span + 1 node.clear() if node.tag == f"{w}body": body_depth = None stack.pop() return stream def _run_reference_xml_worker( mode: str, docx_path: str, *extra_args, ): """隔离执行大型 DOCX XML 解析；原生崩溃时安全返回 ``None``。""" import pickle import subprocess import sys import tempfile worker = os.path.join( os.path.dirname(os.path.abspath(__file__)), "_reference_xml_worker.py", ) fd, result_path = tempfile.mkstemp(prefix="reference_xml_", suffix=".pkl") os.close(fd) try: command = [ sys.executable, worker, mode, os.path.abspath(docx_path), result_path, *[str(arg) for arg in extra_args], ] try: completed = subprocess.run( command, capture_output=True, timeout=900, ) except subprocess.TimeoutExpired: logger.warning(f"参考 DOCX XML 子进程超时: mode={mode}") return None if completed.returncode != 0: logger.warning( "参考 DOCX XML 子进程异常，使用安全降级: " f"mode={mode}, rc={completed.returncode}" ) return None if not os.path.isfile(result_path) or os.path.getsize(result_path) == 0: logger.warning(f"参考 DOCX XML 子进程未产生结果: mode={mode}") return None with open(result_path, "rb") as file: return pickle.load(file) except Exception as exc: logger.warning(f"参考 DOCX XML 隔离解析失败: mode={mode}, {exc}") return None finally: try: os.unlink(result_path) except OSError: pass def _estimate_page_range(char_start: int, char_end: int) -> Tuple[int, int]: """按字符流估算页码范围（近似值，供人工定位用）""" p1 = char_start // _CHARS_PER_PAGE_ESTIMATE + 1 p2 = max(p1, char_end // _CHARS_PER_PAGE_ESTIMATE + 1) return p1, p2 def _collect_table_context( stream: List[dict], tbl_idx: int, ) -> Tuple[int, str, List[str]]: """收集表格前文上下文（所属章节标题、项目信息行）与项目信息字段行""" # 找到第 tbl_idx 张表格在流中的位置 seen = -1 pos = -1 for i, item in enumerate(stream): if item["kind"] == "tbl": seen += 1 if seen == tbl_idx: pos = i break if pos < 0: return -1, "", [] context_lines = [] header_fields = [] for j in range(pos - 1, max(-1, pos - 6), -1): item = stream[j] if item["kind"] != "p": continue text = item["text"] if not text: continue context_lines.insert(0, text) if re.match(r"^(项目名称|项目编号|招标编号|招标项目编号|包号|包件号|投标人名称)[：:]", text): header_fields.append(text) # 遇到章/节标题即停止向上收集 if re.match(r"^第\s*[一二三四五六七八九十百\d]+\s*章", text) or \ re.match(r"^[一二三四五六七八九十百]+、", text): break return pos, "\n".join(context_lines[:5]), header_fields _REFERENCE_TABLE_UNIT_PROMPT = """你是投标文件表格识别专家。下面是一张从参考投标文件中提取的表格（含 文档页码估算、前文上下文、表格行列数与内容预览）。请输出该表格的补充信息。 输出严格 JSON： {"name": "表格正式名称（如：开标一览表、投标报价汇总表、资格条件响应表，不含序号前缀）", "chapter": "所属章节（如：第二章：投标报价）", "purpose": "用途，如：报价/响应/清单/证明/考核表", "header_fields": ["项目名称：...", "项目编号：...", "包号：..."], "column_notes": "各列含义简要说明", "can_reuse": true, "notes": "备注（含需要人工确认的事项）"} 要求： 1. name 优先取前文上下文中的表格标题（如"一、开标一览表"→"开标一览表"）； 2. header_fields 输出表格上方的"项目名称：""项目编号：""包号："等项目信息行（原样）； 3. 没有把握的字段填空字符串，不得编造；can_reuse 默认 true； 4. 只输出 JSON，不输出任何其他文字。""" def _llm_enrich_one_table( tbl_idx: int, table: ExtractedTable, docx_path: str, stream: List[dict], ) -> Optional[dict]: """单个表格单元的 LLM 增强（供线程池并发调用）""" from llm_client import LLMClient stream_pos, context, header_fields = _collect_table_context(stream, tbl_idx) rows = len(table.cells) if table.cells else table.rows cols = table.cols if not cols and table.cells: cols = max(len(r) for r in table.cells) if table.cells else 0 if stream_pos >= 0: p1, p2 = _estimate_page_range( stream[stream_pos]["char_start"], stream[stream_pos]["char_end"], ) else: p1, p2 = 1, 1 # 预览：前 8 行 + 行数（跨页表格整表为单元，行数完整） preview_rows = [] for r in (table.cells or [])[:8]: preview_rows.append( " | ".join(c.text.replace("\n", " ")[:20] for c in r) ) if rows > 8: preview_rows.append(f"...（共 {rows} 行）") user_prompt = ( f"## 表格单元 {tbl_idx + 1}\n" f"文档页码（估算）: 第 {p1}~{p2} 页\n" f"前文上下文:\n{context or '（无）'}\n" f"规格: {rows} 行 x {cols} 列\n" f"内容预览:\n" + "\n".join(f" {r}" for r in preview_rows) ) llm = LLMClient() result = llm.extract_json( system_prompt=_REFERENCE_TABLE_UNIT_PROMPT, user_prompt=user_prompt, max_tokens=4096, ) if not isinstance(result, dict): return None if not header_fields and isinstance(result.get("header_fields"), list): # 用前文扫描到的项目信息行兜底 result["header_fields"] = header_fields or result["header_fields"] return result def enrich_reference_tables_with_llm( tables: List[ExtractedTable], llm=None, docx_path: str = "", ) -> List[ExtractedTable]: """用 LLM 补充参考投书表格的名称/所属章节/用途/表头项目信息字段。 LLM 失败时原样返回结构表格（不影响主流程）。 """ if not tables: return tables if not tables: return tables # 构建文档流（页码估算 + 前文上下文） stream = _build_reference_doc_stream(docx_path) if docx_path else [] if len(stream) < len(tables): # 流缺失时退化为简单估算 stream = [ {"kind": "tbl", "char_start": i * 1000, "char_end": (i + 1) * 1000} for i in range(len(tables)) ] from concurrent.futures import ThreadPoolExecutor, as_completed max_workers = min(5, len(tables)) metas = {} with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = { executor.submit( _llm_enrich_one_table, idx, table, docx_path, stream ): idx for idx, table in enumerate(tables) } for future in as_completed(futures): idx = futures[future] try: result = future.result() except Exception as e: logger.warning( f"参考投书表格 LLM 增强 [{idx}] 失败（保留结构表格）: {e}" ) continue if isinstance(result, dict): metas[idx] = result if not metas: logger.warning("参考投书表格 LLM 增强全部失败，保留结构表格") return tables updated = 0 for idx, t in enumerate(tables): meta = metas.get(idx) if not meta: continue name = (meta.get("name") or "").strip() chapter = (meta.get("chapter") or "").strip() header_fields = meta.get("header_fields") or [] if name: t.title_hint = name if not t.caption: t.caption = name # 用表名重新分类（开标一览表/报价汇总表/报价明细表等） new_type = _classify_tender_table( t.cells or [], title_hint=name ) if new_type != "other": t.table_type = new_type if chapter and chapter not in (t.caption or ""): t.caption = f"{chapter} {t.caption or name}".strip() if isinstance(header_fields, list): t.header_fields = [ str(h).strip() for h in header_fields if str(h).strip() ] updated += 1 if updated: logger.info(f"参考投书表格 LLM 增强: {updated} 张表格补充名称/章节/字段信息") return tables def write_reference_tables_to_docx( tables: List[ExtractedTable], output_dir: str, ) -> int: """把参考投书表格逐张写成独立 DOCX，便于人工核对。 参考投书可能包含横纵交错的大型合并表。使用 python-docx 逐格 ``table.cell(...).merge(...)`` 重建这类表时，会反复展开逻辑网格；在 Windows/PyCharm 中已观察到 lxml 在垃圾回收阶段直接以 0xC0000005 崩溃。这里优先流式读取来源 DOCX 的原生 ``w:tbl`` XML 并整表克隆， 同时也能保留 tblPr/tblGrid、单元格宽度、边框和合并属性。 """ if not tables or not output_dir: return 0 os.makedirs(output_dir, exist_ok=True) _clean_reference_table_artifacts(output_dir) source_table_xml: Dict[str, List[bytes]] = {} for source_file in { os.path.abspath(t.source_file) for t in tables if getattr(t, "source_file", "") and not getattr(t, "is_fragment", False) and os.path.isfile(t.source_file) }: try: source_table_xml[source_file] = _stream_source_table_xml(source_file) except Exception as exc: logger.warning( f"参考投书原生表格 XML 读取失败，将使用无合并兜底: " f"{os.path.basename(source_file)} ({exc})" ) written = 0 for idx, t in enumerate(tables, 1): name = (getattr(t, "title_hint", "") or getattr(t, "caption", "") or f"表格{idx}").strip() if bool(getattr(t, "is_fragment", False)): path = os.path.join(output_dir, _BID_DETAIL_ARTIFACT_NAME) _write_fragment_docx( list(getattr(t, "fragment_xml", []) or []), path ) t.fragment_docx_path = os.path.abspath(path) t.artifact_path = os.path.abspath(path) written += 1 logger.info( f" 参考投书“{_BID_DETAIL_GROUP_NAME}”片段已落盘: {path}" ) continue else: safe = re.sub(r'[<>:"/\\|?*]+', "_", name)[:50] or f"table_{idx}" path = os.path.join(output_dir, f"{idx:02d}_{safe}.docx") doc = DocxDocument() doc.add_heading(f"{idx}. {name}", level=1) if getattr(t, "caption", ""): doc.add_paragraph(str(t.caption)) source_file = os.path.abspath(t.source_file) if t.source_file else "" source_index = int(getattr(t, "para_idx", -1)) xml_items = source_table_xml.get(source_file, []) if 0 <= source_index < len(xml_items): from docx.oxml import parse_xml source_tbl = parse_xml(xml_items[source_index]) doc.element.body.insert_element_before(source_tbl, "w:sectPr") else: _append_unmerged_reference_table(doc, t) doc.save(path) t.artifact_path = os.path.abspath(path) written += 1 logger.info(f"参考投书表格已落盘: {written} 个 DOCX -> {output_dir}") return written def _clean_reference_table_artifacts(output_dir: str) -> int: """清理参考表旧产物，避免重跑后同序号不同命名文件重复累积。""" if not output_dir or not os.path.isdir(output_dir): return 0 removed = 0 for name in os.listdir(output_dir): if name == _BID_DETAIL_ARTIFACT_NAME or _REFERENCE_ARTIFACT_RE.match(name): path = os.path.join(output_dir, name) if os.path.isfile(path): try: os.remove(path) removed += 1 except OSError as exc: logger.warning(f"清理参考表旧产物失败: {name} ({exc})") if removed: logger.info(f"已清理参考表旧产物: {removed} 个文件 -> {output_dir}") return removed def _stream_source_table_xml(docx_path: str) -> List[bytes]: """隔离读取来源表格 XML；失败时由调用方使用无合并安全兜底。""" result = _run_reference_xml_worker("tables", docx_path) return result if isinstance(result, list) else [] def _stream_source_table_xml_impl(docx_path: str) -> List[bytes]: """以恒定内存读取正文直属表格 XML，避免再次构造完整 lxml 文档树。""" import zipfile import xml.etree.ElementTree as ET w = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}" items: List[bytes] = [] with zipfile.ZipFile(docx_path) as archive: with archive.open("word/document.xml") as xml_stream: stack: List[str] = [] body_depth: Optional[int] = None for event, node in ET.iterparse(xml_stream, events=("start", "end")): if event == "start": stack.append(node.tag) if node.tag == f"{w}body": body_depth = len(stack) continue if ( body_depth is not None and len(stack) == body_depth + 1 and node.tag == f"{w}tbl" ): items.append(ET.tostring(node, encoding="utf-8")) node.clear() if node.tag == f"{w}body": body_depth = None stack.pop() return items def _append_unmerged_reference_table(doc, table_data: ExtractedTable) -> None: """来源 XML 不可用时的安全兜底：保留文字网格，不执行危险的合并。""" rows = getattr(table_data, "cells", None) or [] if not rows: doc.add_paragraph("（无表格数据）") return cols = getattr(table_data, "cols", 0) or max(len(row) for row in rows) table = doc.add_table(rows=len(rows), cols=cols) for ri, row in enumerate(rows): for ci, cell_data in enumerate(row[:cols]): if isinstance(cell_data, TableCell): if cell_data.is_merged_continuation: continue text = cell_data.text or "" else: text = str(cell_data) table.rows[ri].cells[ci].text = text def extract_reference_tables_with_llm( docx_path: str, output_dir: str = "", ) -> List[ExtractedTable]: """参考投书表格统一提取入口：结构提取 + LLM 补充名称/章节/用途/表头字段。 提供 output_dir 时，把每张表写成独立 DOCX 便于人工核对。 """ tables = extract_tables_from_template_docx(docx_path) if not tables: return tables for table in tables: table.source_type = "reference_docx" tables = enrich_reference_tables_with_llm(tables, docx_path=docx_path) tables = _build_bid_detail_fragment(tables, docx_path) if output_dir: write_reference_tables_to_docx(tables, output_dir) return tables # ============================================================ # 3. 参考投标文件 → 段落 diff → 骨架段落 # ============================================================ def extract_skeleton_paragraphs( reference_path: str, template_path: str = "", similarity_threshold: float = 0.75, reference_texts=None, template_texts=None, ) -> List[SkeletonParagraph]: """从参考投标文件中提取骨架段落（用于推断模板标题层级） 工作流程： 1. 读取参考投标文件的所有段落 2. 如果提供了模板路径，与模板段落做 diff 比对 3. 相似度 ≥ similarity_threshold 的段落标记为"骨架段落" 4. 根据字号和格式推断标题层级（Heading 1/2/3） 骨架段落的含义： 模板缺少 Word 标题样式（Heading 1/2/3），但参考投标文件中有。 通过比对两者相似的段落，可以反推模板中各段落应该使用什么层级的标题。 Args: reference_path: 参考投标文件路径（PDF/DOCX） template_path: 模板 DOCX 路径（可选，用于 diff 比对） similarity_threshold: 相似度阈值（默认 0.75，即 75% 相似 = 骨架段落） reference_texts: 参考文件正文段落文本（可选，传入则不再重复打开大文件） template_texts: 模板正文段落文本（可选，传入则不再重复打开大文件） Returns: SkeletonParagraph 列表 """ if not os.path.exists(reference_path): logger.warning(f"参考投标文件不存在: {reference_path}") return [] # 读取参考文件段落（优先使用调用方已读好的文本，避免重复打开大文件） if reference_texts is not None: ref_paras = [ {"text": t.strip(), "font_size": None, "bold": False} for t in reference_texts if t.strip() ] else: ref_paras = _read_paragraphs(reference_path) if not ref_paras: return [] # 读取模板段落（如提供了模板路径） template_paras: List[str] = [] if template_path and os.path.exists(template_path): if template_texts is not None: template_paras = [ {"text": t.strip(), "font_size": None, "bold": False} for t in template_texts if t.strip() ] else: template_paras = _read_paragraphs(template_path) # 模板段落前缀索引（与 template_parser._diff_skeleton 同策略）： # 只对同前缀段落做相似度比对，避免 O(N×M) 全量比对——既大幅提速， # 也避免海量字符串分配触发大文件解析后的内存损坏（0xC0000005）。 template_index: Dict[str, List[str]] = {} for tp in template_paras: t = tp["text"].strip() if t and len(t) >= 4: key = _norm_prefix(t[:20]) template_index.setdefault(key, []).append(t) skeletons: List[SkeletonParagraph] = [] for i, para in enumerate(ref_paras): text = para["text"].strip() if not text or len(text) < 4: continue # 跳过太短的段落 # 与模板段落做 diff（如果提供了模板） best_similarity = 0.0 if template_paras: key = _norm_prefix(text[:20]) for t in template_index.get(key, []): sim = _jaccard_similarity(text, t) if sim > best_similarity: best_similarity = sim # 推断标题层级 inferred_level = _infer_heading_level(para) skeletons.append(SkeletonParagraph( para_idx=i, text=text, similarity=best_similarity if template_paras else 1.0, inferred_level=inferred_level, font_size=para.get("font_size"), bold=para.get("bold", False), is_heading=(inferred_level > 0), )) # 统计 heading_count = sum(1 for s in skeletons if s.is_heading) high_sim_count = sum( 1 for s in skeletons if s.similarity >= similarity_threshold ) logger.info( f"骨架段落提取完成: {len(skeletons)} 个段落, " f"{heading_count} 个标题段落, " f"{high_sim_count} 个高相似度（≥{similarity_threshold:.0%}）骨架段落" ) return skeletons def _norm_prefix(s: str) -> str: """归一化前缀：数字→#，用于段落前缀索引""" return re.sub(r"\d+", "#", (s or "").strip()) def _read_paragraphs(file_path: str) -> List[Dict]: """读取文件的段落列表（含格式信息） 支持 PDF（通过 pdfplumber）和 DOCX（通过 python-docx）。 返回格式：[{"text": str, "font_size": float|None, "bold": bool}, ...] """ ext = os.path.splitext(file_path)[1].lower() paragraphs: List[Dict] = [] if ext == ".pdf": try: import pdfplumber with pdfplumber.open(file_path) as pdf: for page in pdf.pages: text = page.extract_text() if text: for line in text.split("\n"): line = line.strip() if line: paragraphs.append({ "text": line, "font_size": None, "bold": False, }) except Exception as e: logger.warning(f"PDF 段落读取失败: {e}") elif ext in (".docx", ".doc"): # 大文件（≥20MB，如 104MB 模板 / 112MB 参考标书）用标准库 ET 读取， # 避免 python-docx/lxml 反复解析大文件导致的原生崩溃（0xC0000005）。 # ET 路径不提供字号/加粗，标题层级推断回退到正则规则。 if os.path.getsize(file_path) >= 20 * 1024 * 1024: try: from doc_reader.reader import read_docx_paragraph_texts_et for text in read_docx_paragraph_texts_et(file_path): if text.strip(): paragraphs.append({ "text": text.strip(), "font_size": None, "bold": False, }) return paragraphs except Exception as e: logger.warning(f"DOCX 段落读取失败(ET): {e}") return paragraphs try: doc = DocxDocument(file_path) for i, para in enumerate(doc.paragraphs): text = para.text.strip() if not text: continue bold = False runs = para.runs if para.runs else [] if runs and all(r.bold for r in runs if r.text.strip()): bold = True # 注意：不要访问 para.style（python-docx 1.2.0 在 Python 3.13 下 # 遍历样式枚举会崩溃）；style 字段下游并未使用，只保留 text/字号/加粗 paragraphs.append({ "text": text, "font_size": _get_para_font_size(para), "bold": bold, }) except Exception as e: logger.warning(f"DOCX 段落读取失败: {e}") return paragraphs def _get_para_font_size(para) -> Optional[float]: """获取段落的首个有效字号""" for run in (para.runs or []): if run.font.size and run.text.strip(): return run.font.size.pt return None def _jaccard_similarity(text1: str, text2: str) -> float: """计算两个文本的 Jaccard 相似度（基于字符 2-gram） 用于段落级别的快速比对，不依赖分词。 """ def _bigrams(s: str) -> Set[str]: s = s.replace(" ", "").replace("\n", "") return {s[i:i + 2] for i in range(len(s) - 1)} b1 = _bigrams(text1) b2 = _bigrams(text2) if not b1 or not b2: return 0.0 intersection = len(b1 & b2) union = len(b1 | b2) return intersection / union if union > 0 else 0.0 def _infer_heading_level(para_info: Dict) -> int: """根据段落特征推断标题层级 推断规则（优先级从高到低）： 1. 格式特征：字号 + 加粗组合判断 2. 内容特征：中文序号模式判断 3. 长度特征：短文本更可能是标题 Returns: 0 = 正文, 1 = 章(Heading 1), 2 = 节(Heading 2), 3 = 小节(Heading 3) """ text = para_info.get("text", "").strip() font_size = para_info.get("font_size") bold = para_info.get("bold", False) # 规则 1：由格式推断 if font_size: if font_size >= 16 and bold: return 1 # 大字号加粗 = 章标题 if font_size >= 14 and bold: return 2 # 中字号加粗 = 节标题 if font_size >= 12 and bold: return 3 # 小字号加粗 = 小节标题 # 规则 2：由中文序号模式推断 if re.match(r"^第[一二三四五六七八九十\d]+章", text): return 1 if re.match(r"^[一二三四五六七八九十]+[、，,]", text) and len(text) < 30: return 2 if re.match(r"^[（(][一二三四五六七八九十]+[）)]", text) and len(text) < 30: return 3 # 规则 3：由长度和加粗推断（标题通常较短） if bold and len(text) < 20: return 2 if bold and len(text) < 40: return 3 return 0 # ============================================================ # 4. 批量提取：一步完成全部表格/段落提取 # ============================================================ def extract_all( template_docx_path: str = "", reference_bid_path: str = "", ) -> Dict: """批量提取：模板表格 + 参考投标骨架段落 Args: template_docx_path: 模板 DOCX 路径 reference_bid_path: 参考投标文件路径 Returns: { "template_tables": List[ExtractedTable], "skeleton_paragraphs": List[SkeletonParagraph], } """ result: Dict = { "template_tables": [], "skeleton_paragraphs": [], } if template_docx_path: result["template_tables"] = extract_tables_from_template_docx(template_docx_path) if reference_bid_path: result["skeleton_paragraphs"] = extract_skeleton_paragraphs( reference_path=reference_bid_path, template_path=template_docx_path, ) return result __all__ = [ "extract_tables_from_template_docx", "extract_skeleton_paragraphs", "extract_all", ]