Spaces:
Sleeping
Sleeping
File size: 6,025 Bytes
06671aa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | from __future__ import annotations
import hashlib
import zipfile
from pathlib import Path
from typing import Iterable
from xml.etree import ElementTree as ET
from fastapi import HTTPException
from models import ParagraphText, BodyTextInvariant
from logger import setup_logger
logger = setup_logger(__name__)
NS = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main",
"w14": "http://schemas.microsoft.com/office/word/2010/wordml"}
W_TEXT = f"{{{NS['w']}}}t"
W_TABLE = f"{{{NS['w']}}}tbl"
W_PARAGRAPH = f"{{{NS['w']}}}p"
W_DELETED = f"{{{NS['w']}}}del"
W14_PARAID = f"{{{NS['w14']}}}paraId"
def _iter_paragraphs(node: ET.Element, *, inside_table: bool = False, inside_deleted: bool = False) -> Iterable[tuple[ET.Element, bool, bool]]:
tag = node.tag
next_inside_table = inside_table or tag == W_TABLE
next_inside_deleted = inside_deleted or tag == W_DELETED
if tag == W_PARAGRAPH:
yield node, next_inside_table, next_inside_deleted
for child in list(node):
yield from _iter_paragraphs(child, inside_table=next_inside_table, inside_deleted=next_inside_deleted)
def _text_from(node: ET.Element, *, inside_deleted: bool = False) -> str:
if inside_deleted or node.tag == W_DELETED:
return ""
chunks: list[str] = []
if node.tag == W_TEXT and node.text:
chunks.append(node.text)
for child in list(node):
chunks.append(_text_from(child, inside_deleted=False))
return "".join(chunks)
def _normalize_display_text(text: str) -> str:
text = text.replace("\u00a0", " ")
return text.strip()
def extract_docx_paragraphs(path: Path) -> list[ParagraphText]:
try:
with zipfile.ZipFile(path) as archive:
xml = archive.read("word/document.xml")
except KeyError as exc:
raise HTTPException(status_code=400, detail="DOCX does not contain word/document.xml.") from exc
except zipfile.BadZipFile as exc:
raise HTTPException(status_code=400, detail="Uploaded file is not a valid DOCX package.") from exc
root = ET.fromstring(xml)
paragraphs: list[ParagraphText] = []
for para, inside_table, inside_deleted in _iter_paragraphs(root):
if inside_deleted:
continue
text = _normalize_display_text(_text_from(para))
if not text:
continue
paragraph_id = f"P_{len(paragraphs) + 1:04d}"
pPr = para.find(f"{{{NS['w']}}}pPr")
native_para_id = None
if pPr is not None:
pid_elem = pPr.find(W14_PARAID)
if pid_elem is not None:
native_para_id = pid_elem.get(f"{{{NS['w14']}}}val")
paragraphs.append(ParagraphText(id=paragraph_id, text=text, native_para_id=native_para_id))
return paragraphs
def render_review_text(paragraphs: list[ParagraphText]) -> str:
return "\n\n".join(f"[[{para.id}]]\n{para.text}" for para in paragraphs)
def normalize_visible_text(path: Path) -> BodyTextInvariant:
paragraphs = extract_docx_paragraphs(path)
combined = "\n".join(f"{p.id}:{p.text}" for p in paragraphs)
text_hash = hashlib.sha256(combined.encode("utf-8")).hexdigest()
return BodyTextInvariant(paragraphs=paragraphs, text_hash=text_hash)
def get_strict_invariant_hash(path: Path) -> str:
"""
Returns a hash of the document body XML after stripping ALL comment markers.
This ensures that NOT ONLY visible text is preserved, but also:
- Table structures and content
- Existing tracked changes (w:ins, w:del)
- Headers/footers (if we checked those files, but we'll stick to document.xml for now)
- Formatting and styles
"""
try:
with zipfile.ZipFile(path) as archive:
xml_bytes = archive.read("word/document.xml")
except Exception:
return "missing"
# We use a regex-based approach or ET to strip markers.
# Since we want to be strict about the REMAINDER of the XML, we strip:
# <w:commentRangeStart ... />
# <w:commentRangeEnd ... />
# <w:commentReference ... />
# Using ET for safety
root = ET.fromstring(xml_bytes)
# Markers to remove
to_remove = [
f"{{{NS['w']}}}commentRangeStart",
f"{{{NS['w']}}}commentRangeEnd",
f"{{{NS['w']}}}commentReference",
]
W_RUN = f"{{{NS['w']}}}r"
def strip_comments(node: ET.Element):
# We must iterate in reverse to safely remove children
for i in range(len(node) - 1, -1, -1):
child = node[i]
if child.tag in to_remove:
node.remove(child)
elif child.tag == W_RUN:
# If this is a run, check if it ONLY contains markers we're removing (and maybe rPr)
strip_comments(child)
# After stripping children of the run, if no children left (or only rPr), remove the run
remaining = [c for c in list(child) if not c.tag.endswith("rPr")]
if not remaining and not (child.text and child.text.strip()):
node.remove(child)
else:
strip_comments(child)
strip_comments(root)
# Also remove any @_w:rsidR and @_w:rsidRPr attributes which might change
# when Word or a tool touches a paragraph even if content doesn't change.
def strip_rsid(node: ET.Element):
to_del = [k for k in node.attrib if "rsid" in k]
for k in to_del:
del node.attrib[k]
for child in node:
strip_rsid(child)
strip_rsid(root)
# We now hash the CONCATENATED TEXT of the entire stripped body.
# This is much more robust than hashing the XML string because it allows
# for structural changes like run-splitting (which the .NET commenter does
# for precise anchoring) while still ensuring that every single visible
# character (including inside tables) remains identical.
full_text = "".join(root.itertext())
return hashlib.sha256(full_text.encode("utf-8")).hexdigest()
|