#!/usr/bin/env python3 """Compare every CZOA export row with the saved DOCX's raw OOXML. This implementation deliberately does not use CZOA code or fixture manifests. It reads both inputs without modifying them. Exit code 0 means all supported fields match; exit code 1 means a discrepancy or incomplete source metadata. """ import argparse import hashlib import sys from collections import Counter from pathlib import Path from xml.etree import ElementTree as ET from zipfile import ZipFile W = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}" W14 = "{http://schemas.microsoft.com/office/word/2010/wordml}" W15 = "{http://schemas.microsoft.com/office/word/2012/wordml}" CID = "{http://schemas.microsoft.com/office/word/2016/wordml/cid}" HEADERS = ( "Comment ID", "Thread ID", "Type", "Parent Comment ID", "Status", "Author", "Initials", "Date (UTC)", "Document area", "Nearest heading", "Paragraph", "Referenced text", "Comment text", ) def digest(path): h = hashlib.sha256() with open(path, "rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def paragraph_text(p): return "".join((node.text or "") if node.tag == W + "t" else "\t" if node.tag == W + "tab" else "\n" for node in p.iter() if node.tag in (W + "t", W + "tab", W + "br")) def read_docx(path): issues = [] with ZipFile(path) as archive: if len(archive.namelist()) != len(set(archive.namelist())): raise ValueError("Duplicate ZIP member names") def xml(part, required=True): try: return ET.fromstring(archive.read(part)) except KeyError: if required: raise ValueError(f"Missing DOCX part: {part}") from None issues.append(f"Missing metadata part: {part}") return None comments = xml("word/comments.xml") extensions = xml("word/commentsExtended.xml", required=False) identifiers = xml("word/commentsIds.xml", required=False) document = xml("word/document.xml") ext = {} if extensions is not None: for item in extensions.findall(W15 + "commentEx"): key = item.get(W15 + "paraId") if key in ext: issues.append(f"Duplicate extension paraId: {key}") ext[key] = item durable = {} if identifiers is not None: for item in identifiers.findall(CID + "commentId"): key = item.get(CID + "paraId") if key in durable: issues.append(f"Duplicate durable paraId: {key}") durable[key] = item.get(CID + "durableId") by_id, by_para = {}, {} for comment in comments.findall(W + "comment"): cid = comment.get(W + "id") ps = comment.findall(W + "p") if not ps: issues.append(f"Comment {cid} has no paragraph") continue para_id = ps[-1].get(W14 + "paraId") if cid in by_id or para_id in by_para: issues.append(f"Duplicate comment ID or paraId: {cid}, {para_id}") entry = { "Comment ID": cid, "Author": comment.get(W + "author") or "", "Initials": comment.get(W + "initials") or "", "Date (UTC)": comment.get(W + "date") or "", "Comment text": "\n".join(paragraph_text(p) for p in ps), "para_id": para_id, } by_id[cid] = entry by_para[para_id] = entry # Number body paragraphs, including paragraphs nested in tables. body = document.find(W + "body") if body is None: raise ValueError("No main document body") anchor_text = {cid: [] for cid in by_id} anchor_context = {} active = set() heading = "" for number, p in enumerate(body.iter(W + "p"), 1): style = p.find("./" + W + "pPr/" + W + "pStyle") if style is not None and (style.get(W + "val") or "").lower().startswith("heading"): heading = paragraph_text(p) for node in p.iter(): if node.tag == W + "commentRangeStart": cid = node.get(W + "id") if cid not in by_id: issues.append(f"Anchor start {cid} has no comment record") else: active.add(cid) anchor_context.setdefault(cid, (heading, str(number))) elif node.tag == W + "commentRangeEnd": cid = node.get(W + "id") if cid not in active: issues.append(f"Anchor end {cid} has no active start") active.discard(cid) elif node.tag == W + "t": for cid in active: anchor_text[cid].append(node.text or "") elif node.tag == W + "tab": for cid in active: anchor_text[cid].append("\t") if active: issues.append(f"Unclosed anchor ranges: {sorted(active)}") expected = {} for cid, entry in by_id.items(): para_id = entry["para_id"] extension = ext.get(para_id) if extension is None: issues.append(f"Comment {cid} lacks extension status/parent metadata") continue parent_para = extension.get(W15 + "paraIdParent") parent = by_para.get(parent_para) if parent_para else None if parent_para and parent is None: issues.append(f"Comment {cid} references missing parent paraId {parent_para}") continue root = parent or entry root_id = root["Comment ID"] if root_id not in anchor_context: issues.append(f"Comment {cid} has no anchored root comment {root_id}") continue thread_id = durable.get(root["para_id"]) if thread_id is None: issues.append(f"Comment {cid} lacks root durable thread ID") continue # Thread resolution belongs to the root, so replies inherit its status. root_extension = ext.get(root["para_id"]) if root_extension is None or root_extension.get(W15 + "done") not in ("0", "1"): issues.append(f"Comment {cid} lacks a known thread resolution state") continue nearest_heading, paragraph = anchor_context[root_id] expected[cid] = { **{key: entry[key] for key in ("Comment ID", "Author", "Initials", "Date (UTC)", "Comment text")}, "Thread ID": thread_id, "Type": "Reply" if parent else "Comment", "Parent Comment ID": root_id if parent else "", "Status": "Resolved" if root_extension.get(W15 + "done") == "1" else "Open", "Document area": "Main document", "Nearest heading": nearest_heading, "Paragraph": paragraph, "Referenced text": "".join(anchor_text[root_id]), } return expected, issues, len(by_id) def read_xlsx(path): """Read the export with only Python's standard library. This keeps the verifier runnable with the macOS system Python and avoids trusting a spreadsheet library to coerce cell types for us. """ issues = [] NS = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}" REL = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}" PKGREL = "{http://schemas.openxmlformats.org/package/2006/relationships}" def column_number(reference): letters = "".join(ch for ch in reference if ch.isalpha()).upper() number = 0 for letter in letters: number = number * 26 + ord(letter) - 64 return number with ZipFile(path) as archive: workbook = ET.fromstring(archive.read("xl/workbook.xml")) sheets = workbook.findall("./" + NS + "sheets/" + NS + "sheet") if len(sheets) != 1: issues.append(f"Expected one export sheet, found {len(sheets)}") if not sheets: raise ValueError("XLSX contains no worksheet") rel_id = sheets[0].get(REL + "id") rels = ET.fromstring(archive.read("xl/_rels/workbook.xml.rels")) target = None for rel in rels.findall(PKGREL + "Relationship"): if rel.get("Id") == rel_id: target = rel.get("Target") break if not target: raise ValueError("XLSX worksheet relationship is missing") sheet_path = target.lstrip("/") if not sheet_path.startswith("xl/"): sheet_path = "xl/" + sheet_path shared = [] if "xl/sharedStrings.xml" in archive.namelist(): shared_root = ET.fromstring(archive.read("xl/sharedStrings.xml")) for item in shared_root.findall(NS + "si"): shared.append("".join(node.text or "" for node in item.iter(NS + "t"))) sheet = ET.fromstring(archive.read(sheet_path)) rows = {} max_column = 0 for row in sheet.findall(".//" + NS + "row"): row_no = int(row.get("r") or 0) values = {} for cell in row.findall(NS + "c"): reference = cell.get("r") or "" col = column_number(reference) max_column = max(max_column, col) if cell.find(NS + "f") is not None: issues.append(f"Executable XLSX formula at row {row_no}, column {col}") cell_type = cell.get("t") if cell_type == "inlineStr": value = "".join(node.text or "" for node in cell.iter(NS + "t")) else: value_node = cell.find(NS + "v") raw = value_node.text if value_node is not None and value_node.text is not None else "" if cell_type == "s" and raw: value = shared[int(raw)] else: value = raw values[col] = value rows[row_no] = values headers = tuple(rows.get(1, {}).get(col, "") for col in range(1, 14)) if headers != HEADERS: raise ValueError(f"Unexpected XLSX headers: {headers!r}") if max_column != len(HEADERS): issues.append(f"Unexpected XLSX column count: {max_column}") actual = {} for row_no in range(2, max(rows, default=1) + 1): cells = [rows.get(row_no, {}).get(col, "") for col in range(1, 14)] if not any(cells): issues.append(f"Blank XLSX row {row_no}") continue cid = str(cells[0]) if cid in actual: issues.append(f"Duplicate XLSX comment ID {cid} at row {row_no}") actual[cid] = {key: str(cell) if cell is not None else "" for key, cell in zip(HEADERS, cells)} return actual, issues def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("docx", type=Path) parser.add_argument("xlsx", type=Path) args = parser.parse_args() try: expected, issues, source_count = read_docx(args.docx) actual, sheet_issues = read_xlsx(args.xlsx) issues.extend(sheet_issues) except (ValueError, OSError, ET.ParseError) as exc: print(f"FAIL: {exc}", file=sys.stderr) return 1 print(f"DOCX SHA256: {digest(args.docx)}") print(f"XLSX SHA256: {digest(args.xlsx)}") print(f"DOCX comment records: {source_count}; XLSX rows: {len(actual)}") print(f"DOCX types: {dict(Counter(x['Type'] for x in expected.values()))}") print(f"DOCX status: {dict(Counter(x['Status'] for x in expected.values()))}") for cid in sorted(expected.keys() - actual.keys(), key=lambda x: int(x)): issues.append(f"Missing XLSX comment ID {cid}") for cid in sorted(actual.keys() - expected.keys(), key=lambda x: int(x) if x.isdigit() else -1): issues.append(f"Extra XLSX comment ID {cid}") fields_compared = 0 for cid in sorted(expected.keys() & actual.keys(), key=lambda x: int(x)): for field in HEADERS: fields_compared += 1 if expected[cid][field] != actual[cid][field]: issues.append(f"ID {cid}, {field}: DOCX={expected[cid][field]!r}, XLSX={actual[cid][field]!r}") print(f"Fields compared: {fields_compared}") if issues: print(f"FAIL: {len(issues)} discrepancies") for issue in issues[:50]: print(" -", issue) if len(issues) > 50: print(f" - ... {len(issues) - 50} more") return 1 print("PASS: all source comments and all 13 exported fields match; no formulas") return 0 if __name__ == "__main__": sys.exit(main())