Spaces:
Sleeping
Sleeping
| 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() | |