Python批量解析并提取Word文档的工程化实践

import zipfile
from lxml import etree

NS = {
    'w': 'http://schemas.openxmlformats.org/wordprocessingml/2006/main',
    'r': 'http://schemas.openxmlformats.org/officeDocument/2006/relationships',
}

def read_docx_xml(path):
    """读取 docx 中的 document.xml 并返回根节点"""
    with zipfile.ZipFile(path, 'r') as z:
        with z.open('word/document.xml') as f:
            tree = etree.parse(f)
    return tree.getroot()

def iter_block_items(root):
    """按文档顺序产出段落和表格，保持原始顺序"""
    body = root.find('w:body', NS)
    for child in body:
        tag = etree.QName(child).localname
        if tag == 'p':
            yield ('paragraph', child)
        elif tag == 'tbl':
            yield ('table', child)

def get_paragraph_text(p_elem):
    """提取段落纯文本，拼接所有 w:t 节点"""
    texts = p_elem.findall('.//w:t', NS)
    return ''.join(t.text or '' for t in texts)

# 使用示例
root = read_docx_xml('题目参考.docx')
for kind, elem in iter_block_items(root):
    if kind == 'paragraph':
        txt = get_paragraph_text(elem)
        if txt.strip():
            print(f'[P] {txt}')
    else:
        print('[TABLE] 发现表格，行数待统计')

def parse_table(tbl_elem):
    """解析表格为二维列表，处理 gridSpan 和 vMerge"""
    rows = []
    for tr in tbl_elem.findall('w:tr', NS):
        row = []
        for tc in tr.findall('w:tc', NS):
            # 水平合并列数
            tcPr = tc.find('w:tcPr', NS)
            grid_span = 1
            v_merge = None
            if tcPr is not None:
                gs = tcPr.find('w:gridSpan', NS)
                if gs is not None:
                    grid_span = int(gs.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val'))
                vm = tcPr.find('w:vMerge', NS)
                if vm is not None:
                    v_merge = vm.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val') or 'continue'
            # 提取单元格内所有段落文本
            cell_text = ' '.join(
                get_paragraph_text(p) for p in tc.findall('w:p', NS)
            ).strip()
            row.append({
                'text': cell_text,
                'grid_span': grid_span,
                'v_merge': v_merge,
            })
        rows.append(row)
    return rows

def build_numbering_map(zip_path):
    """构建 numId -> abstractNumId -> lvl 定义的映射"""
    with zipfile.ZipFile(zip_path, 'r') as z:
        with z.open('word/numbering.xml') as f:
            root = etree.parse(f).getroot()
    num_map = {}
    for num in root.findall('w:num', NS):
        num_id = num.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}numId')
        abstract_ref = num.find('w:abstractNumId', NS)
        if abstract_ref is not None:
            num_map[num_id] = abstract_ref.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val')
    abstract_map = {}
    for an in root.findall('w:abstractNum', NS):
        an_id = an.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}abstractNumId')
        lvls = {}
        for lvl in an.findall('w:lvl', NS):
            ilvl = lvl.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}ilvl')
            start = lvl.find('w:start', NS)
            lvl_text = lvl.find('w:lvlText', NS)
            lvls[ilvl] = {
                'start': int(start.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val')) if start is not None else 1,
                'text': lvl_text.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val') if lvl_text is not None else '',
            }
        abstract_map[an_id] = lvls
    return num_map, abstract_map

def resolve_bold(run_elem, styles_root):
    """解析 run 是否加粗，考虑样式继承"""
    # 直接属性
    rPr = run_elem.find('w:rPr', NS)
    if rPr is not None:
        b = rPr.find('w:b', NS)
        if b is not None:
            val = b.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val')
            return val not in ('0', 'false')
        rStyle = rPr.find('w:rStyle', NS)
        if rStyle is not None:
            style_id = rStyle.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val')
            # 在 styles.xml 中查找该样式的 w:b
            for style in styles_root.findall('w:style', NS):
                if style.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}styleId') == style_id:
                    sb = style.find('.//w:b', NS)
                    if sb is not None:
                        return sb.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val') not in ('0', 'false')
    return False  # 默认不加粗

def extract_images(docx_path, output_dir):
    """提取所有图片并返回 rId -> 文件路径 的映射"""
    import os
    os.makedirs(output_dir, exist_ok=True)
    rel_map = {}
    with zipfile.ZipFile(docx_path, 'r') as z:
        # 解析关系文件
        with z.open('word/_rels/document.xml.rels') as f:
            rels = etree.parse(f).getroot()
        for rel in rels:
            rid = rel.get('Id')
            target = rel.get('Target')
            if 'media' in target:
                rel_map[rid] = target
        # 提取媒体文件
        for name in z.namelist():
            if name.startswith('word/media/'):
                data = z.read(name)
                out_path = os.path.join(output_dir, os.path.basename(name))
                with open(out_path, 'wb') as img_f:
                    img_f.write(data)
    return rel_map

from concurrent.futures import ProcessPoolExecutor, as_completed
import glob, json, os

def parse_one_docx(path):
    """解析单个 docx，返回结构化字典"""
    root = read_docx_xml(path)
    blocks = []
    for kind, elem in iter_block_items(root):
        if kind == 'paragraph':
            txt = get_paragraph_text(elem)
            if txt.strip():
                blocks.append({'type': 'p', 'text': txt})
        else:
            table_data = parse_table(elem)
            blocks.append({'type': 'table', 'data': table_data})
    return {'file': os.path.basename(path), 'blocks': blocks}

def batch_parse(pattern, workers=4):
    files = glob.glob(pattern)
    results = []
    with ProcessPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(parse_one_docx, f): f for f in files}
        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as e:
                print(f'解析失败: {futures[future]}, 错误: {e}')
    return results

# 使用
all_data = batch_parse('data/*.docx', workers=8)
with open('parsed_output.json', 'w', encoding='utf-8') as f:
    json.dump(all_data, f, ensure_ascii=False, indent=2)

def normalize_block(kind, elem, numbering_state):
    """将原始 XML 节点转为统一 schema"""
    if kind == 'paragraph':
        text = get_paragraph_text(elem)
        num_info = extract_numbering(elem, numbering_state)
        return {
            'type': 'paragraph',
            'text': text,
            'numbering': num_info,  # {'numId': ..., 'ilvl': ..., 'display': '1.1'}
        }
    else:
        rows = parse_table(elem)
        return {
            'type': 'table',
            'rows': [[cell['text'] for cell in row] for row in rows],
            'merges': [
                {'row': ri, 'col': ci, 'grid_span': cell['grid_span'], 'v_merge': cell['v_merge']}
                for ri, row in enumerate(rows) for ci, cell in enumerate(row)
                if cell['grid_span'] > 1 or cell['v_merge']
            ],
        }

def collect_style_stats(root):
    """统计文档中所有段落样式的使用频率"""
    from collections import Counter
    counter = Counter()
    for p in root.findall('.//w:p', NS):
        pPr = p.find('w:pPr', NS)
        if pPr is not None:
            pStyle = pPr.find('w:pStyle', NS)
            if pStyle is not None:
                style_id = pStyle.get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val')
                counter[style_id] += 1
    return counter

# 输出示例：{'Heading1': 12, 'QuestionStem': 47, 'OptionA': 188, ...}