"""Word 与 Excel 原文件的安全、受限预览模型。 预览只返回浏览器绘制所需的结构化数据，不返回或执行 Office 包中的活动内容。 DOCX 的字符偏移与上传时的正文抽取规则保持一致，供前端定位当前切片。 """ from __future__ import annotations import io import re from typing import Any from docx import Document from docx.oxml.table import CT_Tbl from docx.oxml.text.paragraph import CT_P from docx.table import Table from docx.text.paragraph import Paragraph from openpyxl import load_workbook from app.modules.data_process.algorithms import ( _MAX_WORKBOOK_COLUMNS, _MAX_WORKBOOK_HEADER_SCAN_ROWS, _infer_xlsx_header_region, _normalize_spreadsheet_value, _rewrite_xlsx_workbook_relationships, _validate_office_archive, _xlsx_sheet_merge_ranges, normalize_text, ) MAX_DOCX_PREVIEW_BLOCKS = 2_000 MAX_XLSX_PREVIEW_ROWS = 200 def _docx_alignment(paragraph: Paragraph) -> str: value = paragraph.alignment return { 0: "left", 1: "center", 2: "right", 3: "justify", 4: "distribute", 5: "justify", 7: "justify", 8: "distribute", 9: "distribute", }.get(int(value) if value is not None else -1, "left") def _docx_heading_level(paragraph: Paragraph) -> int | None: style = paragraph.style if style is None: return None style_name = str(style.name or "") style_id = str(style.style_id or "") match = re.search(r"(?:heading|标题)\s*([1-6])", f"{style_name} {style_id}", re.IGNORECASE) return int(match.group(1)) if match else None def build_docx_preview(raw: bytes) -> dict[str, Any]: """把 DOCX 转为保留标题、段落和表格顺序的浏览器预览模型。""" _validate_office_archive(raw, "docx") try: document = Document(io.BytesIO(raw)) except Exception as exc: raise ValueError(f"invalid DOCX file: {exc}") from exc blocks: list[dict[str, Any]] = [] source_cursor = 0 has_source_content = False rendered_blocks = 0 truncated = False def source_range(value: str) -> tuple[str, int, int] | None: nonlocal source_cursor, has_source_content text = normalize_text(value) if not text: return None if has_source_content: source_cursor += 2 start = source_cursor source_cursor += len(text) has_source_content = True return text, start, source_cursor for child in document.element.body.iterchildren(): if rendered_blocks >= MAX_DOCX_PREVIEW_BLOCKS: truncated = True break if isinstance(child, CT_P): paragraph = Paragraph(child, document) located = source_range(paragraph.text) if located is None: continue text, start, end = located style_name = str(paragraph.style.name or "") if paragraph.style else "" blocks.append( { "type": "paragraph", "text": text, "style": style_name, "heading_level": _docx_heading_level(paragraph), "alignment": _docx_alignment(paragraph), "is_list": "list" in style_name.casefold() or "列表" in style_name, "source_start": start, "source_end": end, } ) rendered_blocks += 1 continue if not isinstance(child, CT_Tbl): continue table = Table(child, document) preview_rows: list[dict[str, Any]] = [] for row in table.rows: if rendered_blocks >= MAX_DOCX_PREVIEW_BLOCKS: truncated = True break cell_values = [normalize_text(cell.text) for cell in row.cells] located = source_range("\t".join(cell_values)) if located is None: continue _, start, end = located preview_rows.append( { "cells": cell_values, "source_start": start, "source_end": end, } ) rendered_blocks += 1 if preview_rows: blocks.append({"type": "table", "rows": preview_rows}) if truncated: break return { "format": "docx", "blocks": blocks, "truncated": truncated, } def build_xlsx_preview( raw: bytes, *, sheet_index: int = 0, offset: int = 0, limit: int = 100, ) -> dict[str, Any]: """按工作表分页返回 XLSX 的表头和记录网格。""" if sheet_index < 0 or offset < 0: raise ValueError("sheet_index and offset must be non-negative") if limit < 1 or limit > MAX_XLSX_PREVIEW_ROWS: raise ValueError( f"XLSX preview limit must be in [1, {MAX_XLSX_PREVIEW_ROWS}]" ) _validate_office_archive(raw, "xlsx") merged_by_sheet, normalized_targets = _xlsx_sheet_merge_ranges(raw) workbook_raw = ( _rewrite_xlsx_workbook_relationships(raw, normalized_targets) if normalized_targets else raw ) try: workbook = load_workbook( io.BytesIO(workbook_raw), read_only=True, data_only=True, keep_links=False, ) except Exception as exc: raise ValueError(f"invalid XLSX file: {exc}") from exc try: sheets = [ { "index": index, "name": worksheet.title, "state": worksheet.sheet_state, } for index, worksheet in enumerate(workbook.worksheets) ] if not sheets: raise ValueError("XLSX workbook contains no worksheets") if sheet_index >= len(sheets): raise ValueError("XLSX worksheet index is out of range") worksheet = workbook.worksheets[sheet_index] reset_dimensions = getattr(worksheet, "reset_dimensions", None) if callable(reset_dimensions): reset_dimensions() row_iterator = enumerate(worksheet.iter_rows(values_only=True), start=1) buffered_rows: dict[int, tuple[Any, ...]] = {} def normalized_values(row: tuple[Any, ...]) -> list[Any]: values = list(row) while values and values[-1] in {None, ""}: values.pop() if len(values) > _MAX_WORKBOOK_COLUMNS: raise ValueError( f"XLSX worksheet {worksheet.title!r} exceeds " f"{_MAX_WORKBOOK_COLUMNS} columns" ) return values for row_number, row in row_iterator: values = normalized_values(row) if not values or all(value in {None, ""} for value in values): continue buffered_rows[row_number] = tuple(values) if len(buffered_rows) >= _MAX_WORKBOOK_HEADER_SCAN_ROWS: break if not buffered_rows: return { "format": "xlsx", "sheets": sheets, "active_sheet": { "index": sheet_index, "name": worksheet.title, "columns": [], "rows": [], "offset": offset, "limit": limit, "has_more": False, }, } _, header_end_row, headers = _infer_xlsx_header_region( worksheet.title, buffered_rows, merged_by_sheet.get(worksheet.title, ()), ) preview_rows: list[dict[str, Any]] = [] record_index = 0 has_more = False def append_row(row_number: int, values: tuple[Any, ...] | list[Any]) -> bool: nonlocal record_index, has_more row_values = list(values) if len(row_values) > len(headers): raise ValueError( f"XLSX worksheet {worksheet.title!r} has a row wider than its header" ) row_values.extend([None] * (len(headers) - len(row_values))) record = { header: _normalize_spreadsheet_value(value) for header, value in zip(headers, row_values, strict=True) } if not any(value not in {"", None} for value in record.values()): return False current_index = record_index record_index += 1 if current_index < offset: return False if len(preview_rows) >= limit: has_more = True return True preview_rows.append( { "row_number": row_number, "record_index": current_index, "values": [record[header] for header in headers], "record": record, } ) return False for row_number, values in buffered_rows.items(): if row_number > header_end_row and append_row(row_number, values): break else: for row_number, row in row_iterator: values = normalized_values(row) if not values or all(value in {None, ""} for value in values): continue if append_row(row_number, values): break return { "format": "xlsx", "sheets": sheets, "active_sheet": { "index": sheet_index, "name": worksheet.title, "columns": headers, "rows": preview_rows, "offset": offset, "limit": limit, "has_more": has_more, }, } finally: workbook.close() __all__ = [ "MAX_DOCX_PREVIEW_BLOCKS", "MAX_XLSX_PREVIEW_ROWS", "build_docx_preview", "build_xlsx_preview", ]