330 lines
14 KiB
Python
330 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""使用 Python 标准库将 DOCX 的常见内容转换为 Markdown。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import re
|
||
import sys
|
||
import zipfile
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path, PurePosixPath
|
||
from xml.etree import ElementTree as ET
|
||
|
||
|
||
NS = {
|
||
"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main",
|
||
"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
|
||
"a": "http://schemas.openxmlformats.org/drawingml/2006/main",
|
||
"pr": "http://schemas.openxmlformats.org/package/2006/relationships",
|
||
}
|
||
|
||
|
||
def qname(prefix: str, local: str) -> str:
|
||
"""生成 ElementTree 使用的完整命名空间标签。"""
|
||
|
||
return f"{{{NS[prefix]}}}{local}"
|
||
|
||
|
||
def escape_markdown(text: str) -> str:
|
||
"""转义会破坏普通行结构的 Markdown 字符。"""
|
||
|
||
return text.replace("\\", "\\\\").replace("|", "\\|")
|
||
|
||
|
||
@dataclass
|
||
class Result:
|
||
"""记录转换产物与可向用户展示的统计信息。"""
|
||
|
||
output_file: Path
|
||
images_dir: Path
|
||
paragraphs: int = 0
|
||
tables: int = 0
|
||
images: int = 0
|
||
warnings: list[str] = field(default_factory=list)
|
||
|
||
|
||
class DocxConverter:
|
||
"""读取 DOCX 包并转换常见 OOXML 块级元素。"""
|
||
|
||
def __init__(self, source: Path, output_dir: Path, force: bool = False) -> None:
|
||
self.source = source
|
||
self.output_dir = output_dir
|
||
self.force = force
|
||
self.output_file = output_dir / f"{source.stem}.md"
|
||
self.images_dir = output_dir / "images"
|
||
self.relationships: dict[str, tuple[str, str]] = {}
|
||
self.heading_styles: dict[str, int] = {}
|
||
self.number_formats: dict[str, str] = {}
|
||
self.image_names: dict[str, str] = {}
|
||
self.result = Result(self.output_file, self.images_dir)
|
||
|
||
def convert(self) -> Result:
|
||
"""验证目标、解析文档并写入 Markdown 与图片。"""
|
||
|
||
if self.output_dir.exists() and not self.force:
|
||
raise FileExistsError(f"输出目录已存在：{self.output_dir}")
|
||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
with zipfile.ZipFile(self.source) as archive:
|
||
names = set(archive.namelist())
|
||
if "word/document.xml" not in names:
|
||
raise ValueError("文件不是有效的 DOCX：缺少 word/document.xml")
|
||
|
||
self._load_relationships(archive, names)
|
||
self._load_styles(archive, names)
|
||
self._load_numbering(archive, names)
|
||
self._detect_unsupported_parts(names)
|
||
|
||
root = ET.fromstring(archive.read("word/document.xml"))
|
||
body = root.find("w:body", NS)
|
||
if body is None:
|
||
raise ValueError("文件不是有效的 DOCX：缺少正文节点")
|
||
|
||
blocks: list[str] = []
|
||
for child in body:
|
||
if child.tag == qname("w", "p"):
|
||
rendered = self._render_paragraph(child, archive)
|
||
if rendered:
|
||
blocks.append(rendered)
|
||
self.result.paragraphs += 1
|
||
elif child.tag == qname("w", "tbl"):
|
||
rendered = self._render_table(child, archive)
|
||
if rendered:
|
||
blocks.append(rendered)
|
||
self.result.tables += 1
|
||
|
||
markdown = "\n\n".join(blocks).strip() + "\n"
|
||
self.output_file.write_text(markdown, encoding="utf-8")
|
||
return self.result
|
||
|
||
def _load_relationships(self, archive: zipfile.ZipFile, names: set[str]) -> None:
|
||
"""读取超链接和媒体关系，后续按关系 ID 解析目标。"""
|
||
|
||
path = "word/_rels/document.xml.rels"
|
||
if path not in names:
|
||
return
|
||
root = ET.fromstring(archive.read(path))
|
||
for rel in root.findall("pr:Relationship", NS):
|
||
rel_id = rel.get("Id")
|
||
target = rel.get("Target")
|
||
rel_type = rel.get("Type", "").rsplit("/", 1)[-1]
|
||
if rel_id and target:
|
||
self.relationships[rel_id] = (rel_type, target)
|
||
|
||
def _load_styles(self, archive: zipfile.ZipFile, names: set[str]) -> None:
|
||
"""识别标题样式 ID；同时兼容英文 Heading 与中文标题名称。"""
|
||
|
||
if "word/styles.xml" not in names:
|
||
return
|
||
root = ET.fromstring(archive.read("word/styles.xml"))
|
||
for style in root.findall("w:style", NS):
|
||
if style.get(qname("w", "type")) != "paragraph":
|
||
continue
|
||
style_id = style.get(qname("w", "styleId"), "")
|
||
name_node = style.find("w:name", NS)
|
||
style_name = name_node.get(qname("w", "val"), "") if name_node is not None else ""
|
||
match = re.search(r"(?:heading|标题)\s*([1-6])", style_name, re.IGNORECASE)
|
||
if not match:
|
||
match = re.search(r"heading([1-6])", style_id, re.IGNORECASE)
|
||
if match:
|
||
self.heading_styles[style_id] = int(match.group(1))
|
||
|
||
def _load_numbering(self, archive: zipfile.ZipFile, names: set[str]) -> None:
|
||
"""建立编号实例到列表类型的基础映射。"""
|
||
|
||
if "word/numbering.xml" not in names:
|
||
return
|
||
root = ET.fromstring(archive.read("word/numbering.xml"))
|
||
abstract_formats: dict[str, str] = {}
|
||
for abstract in root.findall("w:abstractNum", NS):
|
||
abstract_id = abstract.get(qname("w", "abstractNumId"), "")
|
||
level = abstract.find("w:lvl", NS)
|
||
fmt = level.find("w:numFmt", NS) if level is not None else None
|
||
if fmt is not None:
|
||
abstract_formats[abstract_id] = fmt.get(qname("w", "val"), "bullet")
|
||
|
||
for number in root.findall("w:num", NS):
|
||
number_id = number.get(qname("w", "numId"), "")
|
||
abstract = number.find("w:abstractNumId", NS)
|
||
if abstract is not None:
|
||
abstract_id = abstract.get(qname("w", "val"), "")
|
||
self.number_formats[number_id] = abstract_formats.get(abstract_id, "bullet")
|
||
|
||
def _detect_unsupported_parts(self, names: set[str]) -> None:
|
||
"""对存在但当前不能可靠转换的部件给出明确警告。"""
|
||
|
||
checks = {
|
||
"word/comments.xml": "文档包含批注，当前转换器不会输出批注内容",
|
||
"word/footnotes.xml": "文档包含脚注，当前转换器不会输出脚注内容",
|
||
"word/endnotes.xml": "文档包含尾注，当前转换器不会输出尾注内容",
|
||
}
|
||
for path, warning in checks.items():
|
||
if path in names:
|
||
self.result.warnings.append(warning)
|
||
if any(name.startswith("word/header") for name in names):
|
||
self.result.warnings.append("文档包含页眉，当前转换器不会输出页眉内容")
|
||
if any(name.startswith("word/footer") for name in names):
|
||
self.result.warnings.append("文档包含页脚，当前转换器不会输出页脚内容")
|
||
if any(name.startswith("word/embeddings/") for name in names):
|
||
self.result.warnings.append("文档包含内嵌附件，当前转换器不会提取附件")
|
||
|
||
def _render_paragraph(self, paragraph: ET.Element, archive: zipfile.ZipFile) -> str:
|
||
"""转换段落、超链接、基础行内格式和段落内图片。"""
|
||
|
||
parts: list[str] = []
|
||
for child in paragraph:
|
||
if child.tag == qname("w", "r"):
|
||
parts.append(self._render_run(child, archive))
|
||
elif child.tag == qname("w", "hyperlink"):
|
||
text = "".join(self._render_run(run, archive) for run in child.findall("w:r", NS))
|
||
rel_id = child.get(qname("r", "id"))
|
||
relation = self.relationships.get(rel_id or "")
|
||
if text and relation and relation[0] == "hyperlink":
|
||
parts.append(f"[{text}]({relation[1]})")
|
||
else:
|
||
parts.append(text)
|
||
|
||
text = "".join(parts).strip()
|
||
if not text:
|
||
return ""
|
||
|
||
properties = paragraph.find("w:pPr", NS)
|
||
if properties is not None:
|
||
style = properties.find("w:pStyle", NS)
|
||
style_id = style.get(qname("w", "val"), "") if style is not None else ""
|
||
if style_id in self.heading_styles:
|
||
return f"{'#' * self.heading_styles[style_id]} {text}"
|
||
|
||
numbering = properties.find("w:numPr", NS)
|
||
if numbering is not None:
|
||
level = numbering.find("w:ilvl", NS)
|
||
number = numbering.find("w:numId", NS)
|
||
depth = int(level.get(qname("w", "val"), "0")) if level is not None else 0
|
||
number_id = number.get(qname("w", "val"), "") if number is not None else ""
|
||
marker = "-" if self.number_formats.get(number_id, "bullet") == "bullet" else "1."
|
||
return f"{' ' * depth}{marker} {text}"
|
||
|
||
return text
|
||
|
||
def _render_run(self, run: ET.Element, archive: zipfile.ZipFile) -> str:
|
||
"""转换单个文字区段，并在当前位置追加图片引用。"""
|
||
|
||
chunks: list[str] = []
|
||
for child in run:
|
||
if child.tag == qname("w", "t"):
|
||
chunks.append(child.text or "")
|
||
elif child.tag in {qname("w", "br"), qname("w", "cr")}:
|
||
chunks.append(" \n")
|
||
elif child.tag == qname("w", "tab"):
|
||
chunks.append(" ")
|
||
elif child.tag == qname("w", "drawing"):
|
||
chunks.extend(self._render_images(child, archive))
|
||
|
||
text = "".join(chunks)
|
||
if not text:
|
||
return ""
|
||
|
||
properties = run.find("w:rPr", NS)
|
||
if properties is not None:
|
||
if properties.find("w:b", NS) is not None:
|
||
text = f"**{text}**"
|
||
if properties.find("w:i", NS) is not None:
|
||
text = f"*{text}*"
|
||
return text
|
||
|
||
def _render_images(self, drawing: ET.Element, archive: zipfile.ZipFile) -> list[str]:
|
||
"""提取 drawing 关系指向的包内媒体，使用内容哈希稳定命名。"""
|
||
|
||
rendered: list[str] = []
|
||
for blip in drawing.findall(".//a:blip", NS):
|
||
rel_id = blip.get(qname("r", "embed"))
|
||
relation = self.relationships.get(rel_id or "")
|
||
if not relation or relation[0] != "image":
|
||
continue
|
||
target = PurePosixPath("word") / PurePosixPath(relation[1])
|
||
normalized = str(PurePosixPath(*[part for part in target.parts if part not in {".", ".."}]))
|
||
try:
|
||
data = archive.read(normalized)
|
||
except KeyError:
|
||
self.result.warnings.append(f"图片关系无法读取：{relation[1]}")
|
||
continue
|
||
if normalized not in self.image_names:
|
||
suffix = Path(normalized).suffix.lower() or ".bin"
|
||
filename = f"image-{hashlib.sha256(data).hexdigest()[:12]}{suffix}"
|
||
self.images_dir.mkdir(parents=True, exist_ok=True)
|
||
(self.images_dir / filename).write_bytes(data)
|
||
self.image_names[normalized] = filename
|
||
self.result.images += 1
|
||
rendered.append(f"")
|
||
return rendered
|
||
|
||
def _render_table(self, table: ET.Element, archive: zipfile.ZipFile) -> str:
|
||
"""将普通表格转换为 Markdown；复杂合并关系以警告提示。"""
|
||
|
||
rows: list[list[str]] = []
|
||
merged = False
|
||
for row in table.findall("w:tr", NS):
|
||
cells: list[str] = []
|
||
for cell in row.findall("w:tc", NS):
|
||
if cell.find(".//w:gridSpan", NS) is not None or cell.find(".//w:vMerge", NS) is not None:
|
||
merged = True
|
||
paragraphs = [self._render_paragraph(p, archive) for p in cell.findall("w:p", NS)]
|
||
content = "<br>".join(part for part in paragraphs if part)
|
||
cells.append(escape_markdown(content))
|
||
rows.append(cells)
|
||
|
||
if not rows:
|
||
return ""
|
||
width = max(len(row) for row in rows)
|
||
padded = [row + [""] * (width - len(row)) for row in rows]
|
||
header = padded[0]
|
||
lines = [f"| {' | '.join(header)} |", f"| {' | '.join(['---'] * width)} |"]
|
||
lines.extend(f"| {' | '.join(row)} |" for row in padded[1:])
|
||
if merged:
|
||
self.result.warnings.append("文档包含合并单元格，Markdown 表格可能无法保持原结构")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||
"""解析命令行参数。"""
|
||
|
||
parser = argparse.ArgumentParser(description="将 DOCX 的常见内容转换为 Markdown")
|
||
parser.add_argument("input", type=Path, help="输入 .docx 文件")
|
||
parser.add_argument("--output-dir", type=Path, help="输出目录，默认位于输入文件旁的同名目录")
|
||
parser.add_argument("--force", action="store_true", help="允许覆盖同名生成文件，但不删除其他文件")
|
||
return parser.parse_args(argv)
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
"""命令行入口，失败时返回非零状态。"""
|
||
|
||
args = parse_args(argv)
|
||
source = args.input.resolve()
|
||
if not source.is_file():
|
||
print(f"错误：输入文件不存在：{source}", file=sys.stderr)
|
||
return 2
|
||
if source.suffix.lower() != ".docx":
|
||
print("错误：只支持 .docx 文件", file=sys.stderr)
|
||
return 2
|
||
|
||
output_dir = (args.output_dir or source.with_suffix("")).resolve()
|
||
try:
|
||
result = DocxConverter(source, output_dir, args.force).convert()
|
||
except (FileExistsError, ValueError, zipfile.BadZipFile, OSError) as error:
|
||
print(f"错误：{error}", file=sys.stderr)
|
||
return 1
|
||
|
||
print(f"Markdown：{result.output_file}")
|
||
print(f"段落：{result.paragraphs}，表格：{result.tables}，图片：{result.images}")
|
||
if result.warnings:
|
||
print("警告：")
|
||
for warning in dict.fromkeys(result.warnings):
|
||
print(f"- {warning}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|