lanyuanxiaoyao
1aea561277
- 新增 scripts/readers/_utils.py 作为 Reader 内部共享工具模块 - 将 parse_with_markitdown 等函数从 core/markdown.py 迁移到 _utils.py - 函数重命名：parse_with_xxx → parse_via_xxx，_unstructured_elements_to_markdown → convert_unstructured_to_markdown - 更新 17 个 Reader 实现文件的 import 路径 - 从 core/__init__.py 移除已迁移函数的导出 - 新增测试文件 tests/test_readers/test_utils.py - 新增 spec 文档 openspec/specs/reader-internal-utils/spec.md 这次重构明确了模块边界：core/ 提供公共 API，readers/_utils.py 提供 Reader 内部工具
171 lines
6.1 KiB
Python
171 lines
6.1 KiB
Python
"""使用 XML 原生解析 PPTX 文件"""
|
|
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
import zipfile
|
|
from typing import Any, List, Optional, Tuple
|
|
|
|
from scripts.readers._utils import build_markdown_table, flush_list_stack
|
|
|
|
|
|
def parse(file_path: str) -> Tuple[Optional[str], Optional[str]]:
|
|
"""使用 XML 原生解析 PPTX 文件"""
|
|
pptx_namespace = {
|
|
"a": "http://schemas.openxmlformats.org/drawingml/2006/main",
|
|
"p": "http://schemas.openxmlformats.org/presentationml/2006/main",
|
|
"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
|
|
}
|
|
|
|
def extract_text_with_formatting(text_elem: Any, namespaces: dict) -> str:
|
|
result = []
|
|
runs = text_elem.findall(".//a:r", namespaces=namespaces)
|
|
for run in runs:
|
|
t_elem = run.find(".//a:t", namespaces=namespaces)
|
|
if t_elem is None or not t_elem.text:
|
|
continue
|
|
|
|
text = t_elem.text
|
|
|
|
rPr = run.find(".//a:rPr", namespaces=namespaces)
|
|
is_bold = False
|
|
is_italic = False
|
|
|
|
if rPr is not None:
|
|
is_bold = rPr.find(".//a:b", namespaces=namespaces) is not None
|
|
is_italic = rPr.find(".//a:i", namespaces=namespaces) is not None
|
|
|
|
if is_bold and is_italic:
|
|
text = f"***{text}***"
|
|
elif is_bold:
|
|
text = f"**{text}**"
|
|
elif is_italic:
|
|
text = f"*{text}*"
|
|
|
|
result.append(text)
|
|
|
|
return "".join(result).strip() if result else ""
|
|
|
|
def convert_table_to_md(table_elem: Any, namespaces: dict) -> str:
|
|
rows = table_elem.findall(".//a:tr", namespaces=namespaces)
|
|
if not rows:
|
|
return ""
|
|
|
|
rows_data = []
|
|
for row in rows:
|
|
cells = row.findall(".//a:tc", namespaces=namespaces)
|
|
row_data = []
|
|
for cell in cells:
|
|
cell_text = extract_text_with_formatting(cell, namespaces)
|
|
if cell_text:
|
|
cell_text = cell_text.replace("\n", " ").replace("\r", "")
|
|
row_data.append(cell_text if cell_text else "")
|
|
rows_data.append(row_data)
|
|
return build_markdown_table(rows_data)
|
|
|
|
def is_list_item(p_elem: Any, namespaces: dict) -> Tuple[bool, bool]:
|
|
if p_elem is None:
|
|
return False, False
|
|
|
|
pPr = p_elem.find(".//a:pPr", namespaces=namespaces)
|
|
if pPr is None:
|
|
return False, False
|
|
|
|
buChar = pPr.find(".//a:buChar", namespaces=namespaces)
|
|
if buChar is not None:
|
|
return True, False
|
|
|
|
buAutoNum = pPr.find(".//a:buAutoNum", namespaces=namespaces)
|
|
if buAutoNum is not None:
|
|
return True, True
|
|
|
|
return False, False
|
|
|
|
def get_indent_level(p_elem: Any, namespaces: dict) -> int:
|
|
if p_elem is None:
|
|
return 0
|
|
|
|
pPr = p_elem.find(".//a:pPr", namespaces=namespaces)
|
|
if pPr is None:
|
|
return 0
|
|
|
|
lvl = pPr.get("lvl")
|
|
return int(lvl) if lvl else 0
|
|
|
|
try:
|
|
md_content = []
|
|
|
|
with zipfile.ZipFile(file_path) as zip_file:
|
|
slide_files = [
|
|
f
|
|
for f in zip_file.namelist()
|
|
if re.match(r"ppt/slides/slide\d+\.xml$", f)
|
|
]
|
|
slide_files.sort(
|
|
key=lambda f: int(re.search(r"slide(\d+)\.xml$", f).group(1))
|
|
)
|
|
|
|
for slide_idx, slide_file in enumerate(slide_files, 1):
|
|
md_content.append("\n## Slide {}\n".format(slide_idx))
|
|
|
|
with zip_file.open(slide_file) as slide_xml:
|
|
slide_root = ET.parse(slide_xml).getroot()
|
|
|
|
tx_bodies = slide_root.findall(
|
|
".//p:sp/p:txBody", namespaces=pptx_namespace
|
|
)
|
|
|
|
tables = slide_root.findall(".//a:tbl", namespaces=pptx_namespace)
|
|
for table in tables:
|
|
table_md = convert_table_to_md(table, pptx_namespace)
|
|
if table_md:
|
|
md_content.append(table_md)
|
|
|
|
for tx_body in tx_bodies:
|
|
paragraphs = tx_body.findall(
|
|
".//a:p", namespaces=pptx_namespace
|
|
)
|
|
list_stack = []
|
|
|
|
for para in paragraphs:
|
|
is_list, is_ordered = is_list_item(para, pptx_namespace)
|
|
|
|
if is_list:
|
|
level = get_indent_level(para, pptx_namespace)
|
|
|
|
while len(list_stack) <= level:
|
|
list_stack.append("")
|
|
|
|
text = extract_text_with_formatting(
|
|
para, pptx_namespace
|
|
)
|
|
if text:
|
|
marker = "1. " if is_ordered else "- "
|
|
indent = " " * level
|
|
list_stack[level] = f"{indent}{marker}{text}"
|
|
|
|
for i in range(len(list_stack)):
|
|
if list_stack[i]:
|
|
md_content.append(list_stack[i] + "\n")
|
|
list_stack[i] = ""
|
|
else:
|
|
if list_stack:
|
|
flush_list_stack(list_stack, md_content)
|
|
|
|
text = extract_text_with_formatting(
|
|
para, pptx_namespace
|
|
)
|
|
if text:
|
|
md_content.append(f"{text}\n")
|
|
|
|
if list_stack:
|
|
flush_list_stack(list_stack, md_content)
|
|
|
|
md_content.append("---\n")
|
|
|
|
content = "\n".join(md_content)
|
|
if not content.strip():
|
|
return None, "文档为空"
|
|
return content, None
|
|
except Exception as e:
|
|
return None, f"XML 解析失败: {str(e)}"
|