#!/usr/bin/env python3 # -*- coding: utf-8 -*- """skill-case-get —— .docx / .pptx 解包抽取器(纯标准库) 本机没有 python-docx / python-pptx,Node 也没有对应库; .docx / .pptx 本质是 zip,因此用标准库 `zipfile` + `xml.etree.ElementTree` 解包解析。 抽取内容: - 文字段落(按文档/幻灯片顺序,保持阅读序) - 表格文本(.docx 的 w:tbl,逐行拼接) - 内嵌媒体(word/media/*、ppt/media/*、xl/media/*)落地到 --media-out - 超链接(document.xml 的 w:hyperlink r:id → rels 目标;slide 的 a:hlinkClick) - .pptx 的幻灯片切换顺序(presentation.xml 的 sldIdLst + rels) - .pptx 图解/备注:p:sp 的 a:t 文本(标题、正文、备注页) - .docx 页眉页脚(可选,--no-headers 关闭) 用法: python3 office_extract.py --input [--media-out ] [--out ] python3 office_extract.py --input a.docx --input b.pptx --out bundle.json python3 office_extract.py --selftest 输出:一份 JSON(stdout 或 --out),结构见 `result` 组装处。 不写日志、不落任何 PII 到 stdout 之外的位置;媒体文件按原字节落地到 media-out。 """ import argparse import json import os import re import sys import zipfile import xml.etree.ElementTree as ET NS = { # WordprocessingML "w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main", "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships", "wp": "http://schemas.openxmlformats.org/drawingml/2006/wordprocessingDrawing", "a": "http://schemas.openxmlformats.org/drawingml/2006/main", # PresentationML "p": "http://schemas.openxmlformats.org/presentationml/2006/main", # package relationships "pr": "http://schemas.openxmlformats.org/package/2006/relationships", "ct": "http://schemas.openxmlformats.org/package/2006/content-types", } DOCX_EXT = {".docx", ".docm"} PPTX_EXT = {".pptx", ".pptm"} MEDIA_PREFIXES = ("word/media/", "ppt/media/", "xl/media/", "media/") IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".tiff", ".emf", ".wmf"} VIDEO_EXTS = {".mp4", ".mov", ".m4v", ".avi", ".webm", ".mkv", ".wmv"} AUDIO_EXTS = {".mp3", ".m4a", ".wav", ".aac", ".flac", ".ogg", ".wma"} REL_NS = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}" PR_NS = "{http://schemas.openxmlformats.org/package/2006/relationships}" def qname(prefix, local): return "{%s}%s" % (NS[prefix], local) def safe_relpath(name): """把 zip 内条目名规整成安全的相对路径(防 zip slip)。 正确处理 `..`:向上弹一层(而不是直接丢弃),这样 `ppt/slides/../media/a.png` → `ppt/media/a.png`; 越出根的 `..` 一律忽略,路径不可能逃出包外。 """ clean = name.replace("\\", "/").lstrip("/") parts = [] for segment in clean.split("/"): if not segment or segment == ".": continue if segment == "..": if parts: parts.pop() continue parts.append(segment) return "/".join(parts) def local_name(tag): return tag.split("}", 1)[1] if "}" in tag else tag # --------------------------------------------------------------------------- # relationships # --------------------------------------------------------------------------- def parse_rels(zf, rels_path): """解析 _rels 部件 → {rId: {target, type, mode}}。""" out = {} try: raw = zf.read(rels_path) except (KeyError, OSError): return out try: root = ET.fromstring(raw) except ET.ParseError: return out for rel in root: rid = rel.get("Id") or "" target = rel.get("Target") or "" if not rid: continue out[rid] = { "id": rid, "target": target, "type": (rel.get("Type") or "").rsplit("/", 1)[-1], "mode": rel.get("TargetMode") or "Internal", } return out def rel_target_to_part(rels_path, target): """把相对 Target 规整成 zip 内条目路径。""" if target.startswith("/"): return safe_relpath(target) base = posix_dirname(posix_dirname(rels_path)) # _rels 的上一级 return safe_relpath(posix_join(base, target)) def posix_dirname(p): return p.rsplit("/", 1)[0] if "/" in p else "" def posix_join(a, b): if not a: return safe_relpath(b) return safe_relpath(a + "/" + b) # --------------------------------------------------------------------------- # text helpers # --------------------------------------------------------------------------- def collect_text_runs(element, run_tag, text_tag): """按阅读序收集某个节点下的全部文本片段。""" parts = [] for node in element.iter(): if node.tag == run_tag or local_name(node.tag) == local_name(text_tag): if node.tag == text_tag and node.text: parts.append(node.text) return "".join(parts) def paragraph_text(p_node): """一个 w:p / a:p 的文本。""" parts = [] for node in p_node.iter(): ln = local_name(node.tag) if ln == "t" and node.text: parts.append(node.text) elif ln == "tab": parts.append("\t") elif ln in ("br", "cr"): parts.append("\n") return "".join(parts).strip() def is_heading(p_node): """w:p 是否带 Heading 样式。""" for node in p_node.iter(qname("w", "pStyle")): val = node.get(qname("w", "val")) or "" if val.lower().startswith("heading") or val in ("Title", "标题"): return True return False # --------------------------------------------------------------------------- # docx # --------------------------------------------------------------------------- def extract_docx(zf, media_out, include_headers=True): result = { "kind": "document", "format": "docx", "paragraphs": [], "outline": [], "tables": [], "hyperlinks": [], "headings": [], "media": [], "embeddedObjects": [], "notes": [], } doc_part = "word/document.xml" if doc_part not in zf.namelist(): for name in zf.namelist(): if name.startswith("word/") and name.endswith("document.xml"): doc_part = name break if doc_part not in zf.namelist(): result["error"] = "DOCX_MISSING_DOCUMENT_XML" return result rels = parse_rels(zf, "word/_rels/document.xml.rels") root = ET.fromstring(zf.read(doc_part)) body = root.find(qname("w", "body")) if body is None: body = root def walk_blocks(container, scope): for child in container: ln = local_name(child.tag) if ln == "p": text = paragraph_text(child) if text: entry = {"index": len(result["paragraphs"]), "scope": scope, "text": text} if is_heading(child): entry["heading"] = True result["headings"].append(text) result["paragraphs"].append(entry) result["outline"].append({"level": 1 if entry.get("heading") else 0, "text": text}) # 段落里的图片(drawing / pict) for blip in child.iter(): if local_name(blip.tag) == "blip": rid = blip.get(REL_NS + "embed") or blip.get(REL_NS + "link") if rid and rid in rels: result["media"].append({ "rId": rid, "part": rel_target_to_part("word/_rels/document.xml.rels", rels[rid]["target"]), "anchor": "inline", "nearbyText": text[:80], }) elif ln == "tbl": rows = [] for tr in child.findall(qname("w", "tr")): cells = [] for tc in tr.findall(qname("w", "tc")): cells.append(" ".join(paragraph_text(p) for p in tc.findall(qname("w", "p"))).strip()) rows.append(cells) if rows: result["tables"].append({"scope": scope, "rows": rows, "rowCount": len(rows)}) walk_blocks(child, scope) elif ln in ("sdt", "sdtContent", "tc"): walk_blocks(child, scope) walk_blocks(body, "body") # 超链接:w:hyperlink r:id → rels for link in root.iter(qname("w", "hyperlink")): rid = link.get(REL_NS + "id") or link.get(qname("r", "id")) or "" target = rels.get(rid, {}).get("target", "") text = paragraph_text(link) if target or text: result["hyperlinks"].append({"text": text, "target": target, "scope": "body"}) # 内嵌 OLE / 附件对象 for obj in root.iter(qname("w", "object")): for node in obj.iter(): if local_name(node.tag) == "OLEObject": rid = node.get(REL_NS + "id") or node.get(qname("r", "id")) or "" if rid in rels: result["embeddedObjects"].append({"rId": rid, "part": rel_target_to_part("word/_rels/document.xml.rels", rels[rid]["target"]), "progId": node.get("ProgID") or ""}) if include_headers: for name in sorted(zf.namelist()): if re.match(r"^word/(header|footer)\d*\.xml$", name): try: hroot = ET.fromstring(zf.read(name)) except ET.ParseError: continue texts = [paragraph_text(p) for p in hroot.iter(qname("w", "p"))] texts = [t for t in texts if t] if texts: result["notes"].append({"scope": name, "text": "\n".join(texts)}) result["media"] = attach_media(zf, result["media"], media_out) return result # --------------------------------------------------------------------------- # pptx # --------------------------------------------------------------------------- def slide_sort_key(name): m = re.search(r"slide(\d+)\.xml$", name) return int(m.group(1)) if m else 10 ** 6 def extract_pptx(zf, media_out): result = { "kind": "presentation", "format": "pptx", "slides": [], "paragraphs": [], "outline": [], "tables": [], "hyperlinks": [], "headings": [], "media": [], "embeddedObjects": [], "notes": [], "slideCount": 0, } pres_rels = parse_rels(zf, "ppt/_rels/presentation.xml.rels") ordered_parts = [] # 首选:presentation.xml 的 sldIdLst(真实放映顺序) if "ppt/presentation.xml" in zf.namelist(): try: proot = ET.fromstring(zf.read("ppt/presentation.xml")) for sld in proot.iter(qname("p", "sldId")): rid = sld.get(REL_NS + "id") or sld.get(qname("r", "id")) or "" if rid in pres_rels: ordered_parts.append(rel_target_to_part("ppt/_rels/presentation.xml.rels", pres_rels[rid]["target"])) except ET.ParseError: pass if not ordered_parts: ordered_parts = sorted( [n for n in zf.namelist() if re.match(r"^ppt/slides/slide\d+\.xml$", n)], key=slide_sort_key, ) for idx, part in enumerate(ordered_parts, start=1): if part not in zf.namelist(): continue try: sroot = ET.fromstring(zf.read(part)) except ET.ParseError: continue rels = parse_rels(zf, posix_join(posix_dirname(part), "_rels/" + part.rsplit("/", 1)[-1] + ".rels")) title = "" body_lines = [] slide_texts = [] pic_count = 0 for sp in sroot.iter(qname("p", "sp")): ph = None for node in sp.iter(qname("p", "ph")): ph = node.get("type") or "body" break texts = [] for p_node in sp.iter(qname("a", "p")): t = paragraph_text(p_node) if t: texts.append(t) if not texts: continue if ph == "title" or ph == "ctrTitle" or (not title and ph is None and len(texts) == 1 and len(texts[0]) < 60): if not title: title = texts[0] result["headings"].append(texts[0]) continue body_lines.extend(texts) # 表格 for tbl in sroot.iter(qname("a", "tbl")): rows = [] for tr in tbl.iter(qname("a", "tr")): cells = [] for tc in tr.iter(qname("a", "tc")): cell_text = " ".join(paragraph_text(p) for p in tc.iter(qname("a", "p"))).strip() cells.append(cell_text) if cells: rows.append(cells) if rows: result["tables"].append({"scope": part, "slide": idx, "rows": rows, "rowCount": len(rows)}) # 图片 for blip in sroot.iter(qname("a", "blip")): rid = blip.get(REL_NS + "embed") or blip.get(REL_NS + "link") if rid and rid in rels: pic_count += 1 result["media"].append({ "rId": rid, "part": rel_target_to_part(posix_join(posix_dirname(part), "_rels/" + part.rsplit("/", 1)[-1] + ".rels"), rels[rid]["target"]), "anchor": "slide", "slide": idx, "nearbyText": title or (body_lines[0] if body_lines else ""), }) # 超链接 for link in sroot.iter(qname("a", "hlinkClick")): rid = link.get(REL_NS + "id") if rid and rid in rels: result["hyperlinks"].append({"text": title, "target": rels[rid]["target"], "scope": part, "slide": idx}) # 备注页 notes_part = posix_join(posix_dirname(part), "notesSlides/notesSlide%d.xml" % idx) notes_text = "" if notes_part in zf.namelist(): try: nroot = ET.fromstring(zf.read(notes_part)) notes_text = "\n".join(t for t in (paragraph_text(p) for p in nroot.iter(qname("a", "p"))) if t) except ET.ParseError: notes_text = "" if notes_text: result["notes"].append({"scope": notes_part, "slide": idx, "text": notes_text}) slide_entry = { "index": idx, "part": part, "title": title, "texts": body_lines, "text": "\n".join([title] + body_lines).strip(), "notes": notes_text, "imageCount": pic_count, } result["slides"].append(slide_entry) for line in slide_entry["text"].split("\n"): if line.strip(): result["paragraphs"].append({"index": len(result["paragraphs"]), "scope": part, "slide": idx, "text": line.strip()}) result["outline"].append({"level": 1 if line.strip() == title else 0, "text": line.strip()}) # 内嵌文件(ppt/embeddings/*) for name in zf.namelist(): if name.startswith("ppt/embeddings/") and not name.endswith("/"): result["embeddedObjects"].append({"part": safe_relpath(name), "progId": ""}) result["slideCount"] = len(result["slides"]) result["media"] = attach_media(zf, result["media"], media_out) return result # --------------------------------------------------------------------------- # media # --------------------------------------------------------------------------- def classify_media(part): ext = os.path.splitext(part)[1].lower() if ext in IMAGE_EXTS or ext == "": return "image" if ext in VIDEO_EXTS: return "video" if ext in AUDIO_EXTS: return "audio" return "other" def attach_media(zf, media_entries, media_out): """把媒体条目去重、补上 kind/size,并按需落地到 media_out。""" by_part = {} order = [] for entry in media_entries: part = safe_relpath(entry.get("part") or "") if not part: continue if part not in by_part: by_part[part] = {**entry, "part": part, "kind": classify_media(part), "occurrences": 0} order.append(part) by_part[part]["occurrences"] += 1 out = [] for part in order: entry = by_part[part] entry["size"] = zf.getinfo(part).file_size if part in zf.namelist() else 0 entry["localPath"] = "" if media_out and part in zf.namelist(): safe_name = part.replace("/", "__") target = os.path.join(media_out, safe_name) os.makedirs(media_out, exist_ok=True) with zf.open(part) as src, open(target, "wb") as dst: dst.write(src.read()) entry["localPath"] = os.path.abspath(target) out.append(entry) return out # --------------------------------------------------------------------------- # entry # --------------------------------------------------------------------------- def package_kind(path): ext = os.path.splitext(path)[1].lower() if ext in DOCX_EXT: return "docx" if ext in PPTX_EXT: return "pptx" # 扩展名不认识时,看 package 内容 try: with zipfile.ZipFile(path) as zf: names = zf.namelist() if any(n.startswith("word/") for n in names): return "docx" if any(n.startswith("ppt/") for n in names): return "pptx" except (zipfile.BadZipFile, OSError): pass return "" def extract_one(path, media_out=None, include_headers=True): kind = package_kind(path) base = { "input": os.path.abspath(path), "fileName": os.path.basename(path), "byteSize": os.path.getsize(path) if os.path.exists(path) else 0, } if not kind: return {**base, "ok": False, "error": "UNSUPPORTED_PACKAGE", "message": "不是 .docx/.pptx 包(zip 内无 word/ 或 ppt/)"} try: with zipfile.ZipFile(path) as zf: if kind == "docx": payload = extract_docx(zf, media_out, include_headers=include_headers) else: payload = extract_pptx(zf, media_out) except zipfile.BadZipFile: return {**base, "ok": False, "error": "BAD_ZIP", "message": "文件损坏或不是有效 zip 包"} except ET.ParseError as exc: return {**base, "ok": False, "error": "BAD_XML", "message": "XML 解析失败:%s" % exc} return {**base, "ok": True, **payload} def selftest(): """自检:现场构造一个最小 docx + pptx,解包并断言关键结果。""" import io import tempfile doc_xml = ( '' '' "" '帝国理工 计算机科学 辅导案例' "学生考前冲刺,最终提分 18 分" "" '查看反馈' "AB" "" ) % (NS["w"], NS["r"], NS["wp"], NS["a"]) doc_rels = ( '' '' '' '' "" ) % NS["pr"] slide_xml = ( '' '' '案例合集' "正文一" '' "" ) % (NS["p"], NS["a"], NS["r"]) slide_rels = ( '' '' '' "" ) % NS["pr"] pres_xml = ( '' '' '' ) % (NS["p"], NS["r"]) pres_rels = ( '' '' '' "" ) % NS["pr"] png_bytes = bytes.fromhex( "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c489" "0000000a49444154789c6300010000050001" "0d0a2db40000000049454e44ae426082" ) failures = [] with tempfile.TemporaryDirectory() as tmp: docx_path = os.path.join(tmp, "case.docx") with zipfile.ZipFile(docx_path, "w") as zf: zf.writestr("[Content_Types].xml", "") zf.writestr("word/document.xml", doc_xml) zf.writestr("word/_rels/document.xml.rels", doc_rels) zf.writestr("word/media/image1.png", png_bytes) res = extract_one(docx_path, media_out=os.path.join(tmp, "media")) if not res.get("ok"): failures.append("docx 解包失败:%s" % res.get("error")) else: texts = [p["text"] for p in res["paragraphs"]] if "帝国理工 计算机科学 辅导案例" not in texts: failures.append("docx 标题段落缺失") if not any("提分 18 分" in t for t in texts): failures.append("docx 跨 run 文本拼接失败") if not res["headings"]: failures.append("docx 标题识别失败") if not res["tables"] or res["tables"][0]["rows"][0] != ["A", "B"]: failures.append("docx 表格抽取失败") if not res["hyperlinks"] or res["hyperlinks"][0]["target"] != "https://example.invalid/case": failures.append("docx 超链接抽取失败") media = [m for m in res["media"] if m["kind"] == "image"] if not media: failures.append("docx 内嵌图片未识别") elif not os.path.exists(media[0]["localPath"]): failures.append("docx 内嵌图片未落地") pptx_path = os.path.join(tmp, "deck.pptx") with zipfile.ZipFile(pptx_path, "w") as zf: zf.writestr("[Content_Types].xml", "") zf.writestr("ppt/presentation.xml", pres_xml) zf.writestr("ppt/_rels/presentation.xml.rels", pres_rels) zf.writestr("ppt/slides/slide1.xml", slide_xml) zf.writestr("ppt/slides/_rels/slide1.xml.rels", slide_rels) zf.writestr("ppt/media/image1.png", png_bytes) res = extract_one(pptx_path, media_out=os.path.join(tmp, "media2")) if not res.get("ok"): failures.append("pptx 解包失败:%s" % res.get("error")) else: if res.get("slideCount") != 1: failures.append("pptx 页数不对:%s" % res.get("slideCount")) if not res["slides"] or res["slides"][0]["title"] != "案例合集": failures.append("pptx 标题占位符识别失败") if not any("正文一" in t for t in res["slides"][0]["texts"]): failures.append("pptx 正文抽取失败") if not [m for m in res["media"] if m["kind"] == "image"]: failures.append("pptx 图片未识别") elif res["media"][0]["part"] != "ppt/media/image1.png": failures.append("pptx 图片路径未正确解析 ..:%s" % res["media"][0]["part"]) elif not os.path.exists(res["media"][0]["localPath"]): failures.append("pptx 内嵌图片未落地") if failures: print(json.dumps({"ok": False, "failures": failures}, ensure_ascii=False, indent=2)) return 1 print(json.dumps({"ok": True, "checked": ["docx", "pptx"], "message": "office 解包自检通过"}, ensure_ascii=False, indent=2)) return 0 def main(argv=None): parser = argparse.ArgumentParser(description="skill-case-get .docx/.pptx 解包抽取器(标准库实现)") parser.add_argument("--input", action="append", default=[], help="待解包的 .docx/.pptx(可重复)") parser.add_argument("--media-out", default="", help="内嵌媒体落地目录") parser.add_argument("--out", default="", help="结果 JSON 输出路径(默认 stdout)") parser.add_argument("--no-headers", action="store_true", help=".docx 不解页眉页脚") parser.add_argument("--selftest", action="store_true", help="运行内置自检") args = parser.parse_args(argv) if args.selftest: return selftest() if not args.input: parser.error("至少需要一个 --input") results = [extract_one(p, args.media_out or None, include_headers=not args.no_headers) for p in args.input] payload = { "ok": all(r.get("ok") for r in results), "generator": "skill-case-get/office_extract.py", "count": len(results), "results": results, } text = json.dumps(payload, ensure_ascii=False, indent=2) if args.out: with open(args.out, "w", encoding="utf-8") as fh: fh.write(text + "\n") else: sys.stdout.write(text + "\n") return 0 if payload["ok"] else 2 if __name__ == "__main__": sys.exit(main())