from docx import Document import json from docx.oxml.table import CT_Tbl from docx.oxml.text.paragraph import CT_P from lxml import etree import os import zipfile RESULT_TYPE_TEXT = 'text' RESULT_TYPE_TABLE = 'table' def build_result(result_type, index, data): return { 'type': result_type, 'index': index, 'data': data } def build_catalog_result(index, depth, data): return { 'index': index, 'depth': depth, 'data': data } # 解析docx文件中的XML内容 def get_xml_content(docx_filename, xml_filename): with zipfile.ZipFile(docx_filename) as z: return z.read(xml_filename) def parse_paragraph(paragraph, index, namespaces): paragraph_text = paragraph.text.strip() if paragraph else '' if paragraph_text: return build_result(RESULT_TYPE_TEXT, index, paragraph_text) return None def parse_table(table, index): table_data = [] for row in table.rows: row_data = [cell.text for cell in row.cells] table_data.append(row_data) return build_result(RESULT_TYPE_TABLE, index, table_data) def parse_paragraph_element(paragraph_element, index, namespaces): paragraph_xml = etree.fromstring(paragraph_element.xml) paragraph_text = ''.join(paragraph_xml.xpath('//w:t/text()', namespaces=namespaces)).strip() if paragraph_text: return build_result(RESULT_TYPE_TEXT, index, paragraph_text) return None def parse_table_element(table_element, index, namespaces): table_xml = etree.fromstring(table_element.xml) table_data = [] for row in table_xml.xpath('//w:tr', namespaces=namespaces): row_data = [] for cell in row.xpath('./w:tc | ./w:sdt', namespaces=namespaces): cell_text = ''.join(cell.xpath('.//w:t/text()', namespaces=namespaces)).strip() grid_span_xpath = etree.XPath('.//w:tcPr/w:gridSpan/@w:val', namespaces=namespaces) grid_span = int(grid_span_xpath(cell)[0]) if grid_span_xpath(cell) else 1 if grid_span > 1: row_data.extend([cell_text] * grid_span) else: row_data.append(cell_text) table_data.append(row_data) return build_result(RESULT_TYPE_TABLE, index, table_data) def add_to_catalog(element_xml, index, catalog_content, namespaces, paragraph_text, heading_styles): p_element = etree.fromstring(element_xml) # outlineLvl = p_element.xpath('.//w:outlineLvl', namespaces=namespaces) # if outlineLvl: # level = int(outlineLvl[0].get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val')) # catalog_content.append(build_catalog_result(index, level, paragraph_text)) level = is_heading_paragraph(p_element, heading_styles, namespaces) if level != -1: catalog_content.append(build_catalog_result(index, level, paragraph_text)) # 检查段落是否为标题样式 def is_heading_paragraph(paragraph, heading_styles, namespaces): pPr = paragraph.find('.//w:pPr', namespaces=namespaces) if pPr is not None: pStyle = pPr.find('.//w:pStyle', namespaces=namespaces) pOutLineLvl = pPr.find('.//w:outlineLvl', namespaces=namespaces) if pStyle is not None: style_val = pStyle.get(f"{{{namespaces['w']}}}val") if style_val.isdigit(): return int(style_val) if pOutLineLvl is not None: outLineLvl_val = pOutLineLvl.get(f"{{{namespaces['w']}}}val") if outLineLvl_val.isdigit(): return int(outLineLvl_val) + 1 # if pStyle is not None and pStyle.get(ns['w'] + 'val') in heading_styles: # if style_val > 0: # return True return -1 def get_paragraph_text(paragraph_element, namespaces): paragraph_text = '' for run in paragraph_element.findall('.//w:r', namespaces=namespaces): for text in run.findall('.//w:t', namespaces=namespaces): paragraph_text += text.text if text.text is not None else '' return paragraph_text def add_to_catalog_paragraph(text, index, catalog_content, namespaces): # 添加段落到目录 catalog_content.append(build_catalog_result(index, 1, text)) # 假设默认级别为1 def parse_sdt_catalog(sdt_element, catalog_content, index, namespaces): sdt_content = sdt_element.find('.//w:sdtContent', namespaces=namespaces) if sdt_content is not None: for child in sdt_content: if child.tag.endswith('p'): # 内容控件中的段落 paragraph_text = get_paragraph_text(child, namespaces) if paragraph_text.strip(): # 检查文本是否为空 add_to_catalog_paragraph(paragraph_text, index, catalog_content, namespaces) index += 1 # 更新索引 elif child.tag.endswith('tbl'): # 内容控件中的表格 # 处理表格内容（如果需要） pass elif child.tag.endswith('sdt'): # 嵌套的内容控件 index = parse_sdt_catalog(child, catalog_content, index, namespaces) # 递归解析嵌套的内容控件 return index def parse_docx(docx_path): try: document = Document(docx_path) styles_xml = get_xml_content(docx_path, 'word/styles.xml') except Exception as e: print(f"Error loading document: {e}") return None, None doc_content = [] # 内容（文本+表格） catalog_content = [] # 目录 current_index = 1 # 维护全局的 index 变量 paragraph_index = 0 table_index = 0 # 获取整个文档的XML内容 xml_root = document.part.element namespaces = xml_root.nsmap # 获取所有标题样式 styles_root = etree.fromstring(styles_xml) heading_styles = set() for style in styles_root.xpath('//w:style', namespaces=namespaces): style_type = style.get(namespaces['w'] + 'type') if style_type == 'paragraph' and style.get(namespaces['w'] + 'styleId').startswith('Heading'): heading_styles.add(style.get(namespaces['w'] + 'styleId')) # 遍历文档中的所有元素 for i, element in enumerate(document.element.body): if isinstance(element, CT_P): # 段落 paragraph_result = parse_paragraph_element(element, current_index, namespaces) if paragraph_result: doc_content.append(paragraph_result) # 判断是否为目录，是就插入目录内容 paragraph = document.paragraphs[paragraph_index] add_to_catalog(paragraph._element.xml, current_index, catalog_content, namespaces, paragraph.text, heading_styles) current_index += 1 # 更新 index paragraph_index += 1 elif isinstance(element, CT_Tbl): # 表格 table_result = parse_table_element(element, current_index, namespaces) if table_result: doc_content.append(table_result) current_index += 1 # 更新 index table_index += 1 elif element.tag.endswith('sdt'): # 内容控件 current_index = parse_sdt(element, doc_content, current_index, namespaces, catalog_content, heading_styles) # 更新索引 return json.dumps(doc_content, indent=4, ensure_ascii=False), json.dumps(catalog_content, indent=4, ensure_ascii=False) def parse_sdt(sdt_element, doc_content, current_index, namespaces, catalog_content, heading_styles): sdtContent = sdt_element.find('.//w:sdtContent', namespaces=namespaces) if sdtContent is not None: for child in sdtContent: if child.tag.endswith('p'): # 内容控件中的段落 paragraph_text = '' for run in child.findall('.//w:r', namespaces=namespaces): for text in run.findall('.//w:t', namespaces=namespaces): paragraph_text += text.text if text.text is not None else '' if paragraph_text.strip(): # 检查文本是否为空 doc_content.append(build_result(RESULT_TYPE_TEXT, current_index, paragraph_text.strip())) # 判断是否为目录，是就插入目录内容 add_to_catalog(child.xml, current_index, catalog_content, namespaces, paragraph_text, heading_styles) current_index += 1 # 更新索引 elif child.tag.endswith('tbl'): # 内容控件中的表格 table_data = [] merged_cells = {} # 用于记录跨行单元格的信息 for row_idx, row in enumerate(child.findall('.//w:tr', namespaces=namespaces)): row_data = [] for col_idx, cell in enumerate(row.findall('.//w:tc', namespaces=namespaces)): cell_text = '' for run in cell.findall('.//w:r', namespaces=namespaces): for text in run.findall('.//w:t', namespaces=namespaces): cell_text += text.text if text.text is not None else '' # 检查单元格是否跨列 grid_span_xpath = etree.XPath('.//w:tcPr/w:gridSpan/@w:val', namespaces=namespaces) grid_span = int(grid_span_xpath(cell)[0]) if grid_span_xpath(cell) else 1 if grid_span > 1: row_data.extend([cell_text.strip()] * grid_span) else: row_data.append(cell_text.strip()) # 检查单元格是否跨行 v_merge_xpath = etree.XPath('.//w:tcPr/w:vMerge/@w:val', namespaces=namespaces) v_merge = v_merge_xpath(cell) if v_merge and v_merge[0] == 'restart': merged_cells[(row_idx, col_idx)] = (int(grid_span), 1) elif v_merge and v_merge[0] == 'continue': if (row_idx - 1, col_idx) in merged_cells: merged_cells[(row_idx - 1, col_idx)] = (merged_cells[(row_idx - 1, col_idx)][0], merged_cells[(row_idx - 1, col_idx)][1] + 1) # 跨行单元格不需要再次添加到 row_data 中 else: # 只有非跨行单元格才需要添加到 row_data 中 pass # 处理跨行单元格 for (r, c), (col_span, row_span) in list(merged_cells.items()): if r < row_idx: for i in range(row_span): if r + i == row_idx: row_data[c:c] = [row_data[c]] * (col_span - 1) break if r + row_span - 1 == row_idx: del merged_cells[(r, c)] table_data.append(row_data) if table_data: # 检查表格数据是否为空 doc_content.append(build_result(RESULT_TYPE_TABLE, current_index, table_data)) current_index += 1 # 更新索引 elif child.tag.endswith('sdt'): # 嵌套的内容控件 current_index = parse_sdt(child, doc_content, current_index, namespaces, catalog_content, heading_styles) # 递归解析嵌套的内容控件 return current_index # 返回更新后的索引 def split_text_table(json_data): # 分组 text_elements = [element for element in json_data if element['type'] == 'text'] table_elements = [element for element in json_data if element['type'] == 'table'] # 转换为JSON字符串 text_elements_json = json.dumps(text_elements, ensure_ascii=False, indent=4) table_elements_json = json.dumps(table_elements, ensure_ascii=False, indent=4) return text_elements_json, table_elements_json def append_to_file(file_path, text): try: with open(file_path, 'a', encoding='utf-8') as file: file.write(text + '\n') except Exception as e: print(f"Error writing to file: {e}") if __name__ == "__main__": current_directory = os.getcwd() docx_relative_path = '101.docx' file_relative_path = 'file\\docx\\test1.txt' docx_path = os.path.join(current_directory, docx_relative_path) file_path = os.path.join(current_directory, file_relative_path) try: parsed_content, catalog_content = parse_docx(docx_path) if parsed_content and catalog_content: json_parsed_content = json.loads(parsed_content) text_elements_json, table_elements_json = split_text_table(json_parsed_content) append_to_file(file_path, text_elements_json) append_to_file(file_path, table_elements_json) append_to_file(file_path, catalog_content) except Exception as e: print(f"Error parse_docx: {e}")