simplify_redlines.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. """Simplify tracked changes by merging adjacent w:ins or w:del elements.
  2. Merges adjacent <w:ins> elements from the same author into a single element.
  3. Same for <w:del> elements. This makes heavily-redlined documents easier to
  4. work with by reducing the number of tracked change wrappers.
  5. Rules:
  6. - Only merges w:ins with w:ins, w:del with w:del (same element type)
  7. - Only merges if same author (ignores timestamp differences)
  8. - Only merges if truly adjacent (only whitespace between them)
  9. """
  10. import xml.etree.ElementTree as ET
  11. import zipfile
  12. from pathlib import Path
  13. import defusedxml.minidom
  14. WORD_NS = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
  15. def simplify_redlines(input_dir: str) -> tuple[int, str]:
  16. doc_xml = Path(input_dir) / "word" / "document.xml"
  17. if not doc_xml.exists():
  18. return 0, f"Error: {doc_xml} not found"
  19. try:
  20. dom = defusedxml.minidom.parseString(doc_xml.read_text(encoding="utf-8"))
  21. root = dom.documentElement
  22. merge_count = 0
  23. containers = _find_elements(root, "p") + _find_elements(root, "tc")
  24. for container in containers:
  25. merge_count += _merge_tracked_changes_in(container, "ins")
  26. merge_count += _merge_tracked_changes_in(container, "del")
  27. doc_xml.write_bytes(dom.toxml(encoding="UTF-8"))
  28. return merge_count, f"Simplified {merge_count} tracked changes"
  29. except Exception as e:
  30. return 0, f"Error: {e}"
  31. def _merge_tracked_changes_in(container, tag: str) -> int:
  32. merge_count = 0
  33. tracked = [
  34. child
  35. for child in container.childNodes
  36. if child.nodeType == child.ELEMENT_NODE and _is_element(child, tag)
  37. ]
  38. if len(tracked) < 2:
  39. return 0
  40. i = 0
  41. while i < len(tracked) - 1:
  42. curr = tracked[i]
  43. next_elem = tracked[i + 1]
  44. if _can_merge_tracked(curr, next_elem):
  45. _merge_tracked_content(curr, next_elem)
  46. container.removeChild(next_elem)
  47. tracked.pop(i + 1)
  48. merge_count += 1
  49. else:
  50. i += 1
  51. return merge_count
  52. def _is_element(node, tag: str) -> bool:
  53. name = node.localName or node.tagName
  54. return name == tag or name.endswith(f":{tag}")
  55. def _get_author(elem) -> str:
  56. author = elem.getAttribute("w:author")
  57. if not author:
  58. for attr in elem.attributes.values():
  59. if attr.localName == "author" or attr.name.endswith(":author"):
  60. return attr.value
  61. return author
  62. def _can_merge_tracked(elem1, elem2) -> bool:
  63. if _get_author(elem1) != _get_author(elem2):
  64. return False
  65. node = elem1.nextSibling
  66. while node and node != elem2:
  67. if node.nodeType == node.ELEMENT_NODE:
  68. return False
  69. if node.nodeType == node.TEXT_NODE and node.data.strip():
  70. return False
  71. node = node.nextSibling
  72. return True
  73. def _merge_tracked_content(target, source):
  74. while source.firstChild:
  75. child = source.firstChild
  76. source.removeChild(child)
  77. target.appendChild(child)
  78. def _find_elements(root, tag: str) -> list:
  79. results = []
  80. def traverse(node):
  81. if node.nodeType == node.ELEMENT_NODE:
  82. name = node.localName or node.tagName
  83. if name == tag or name.endswith(f":{tag}"):
  84. results.append(node)
  85. for child in node.childNodes:
  86. traverse(child)
  87. traverse(root)
  88. return results
  89. def get_tracked_change_authors(doc_xml_path: Path) -> dict[str, int]:
  90. if not doc_xml_path.exists():
  91. return {}
  92. try:
  93. tree = ET.parse(doc_xml_path)
  94. root = tree.getroot()
  95. except ET.ParseError:
  96. return {}
  97. namespaces = {"w": WORD_NS}
  98. author_attr = f"{{{WORD_NS}}}author"
  99. authors: dict[str, int] = {}
  100. for tag in ["ins", "del"]:
  101. for elem in root.findall(f".//w:{tag}", namespaces):
  102. author = elem.get(author_attr)
  103. if author:
  104. authors[author] = authors.get(author, 0) + 1
  105. return authors
  106. def _get_authors_from_docx(docx_path: Path) -> dict[str, int]:
  107. try:
  108. with zipfile.ZipFile(docx_path, "r") as zf:
  109. if "word/document.xml" not in zf.namelist():
  110. return {}
  111. with zf.open("word/document.xml") as f:
  112. tree = ET.parse(f)
  113. root = tree.getroot()
  114. namespaces = {"w": WORD_NS}
  115. author_attr = f"{{{WORD_NS}}}author"
  116. authors: dict[str, int] = {}
  117. for tag in ["ins", "del"]:
  118. for elem in root.findall(f".//w:{tag}", namespaces):
  119. author = elem.get(author_attr)
  120. if author:
  121. authors[author] = authors.get(author, 0) + 1
  122. return authors
  123. except (zipfile.BadZipFile, ET.ParseError):
  124. return {}
  125. def infer_author(modified_dir: Path, original_docx: Path, default: str = "Claude") -> str:
  126. modified_xml = modified_dir / "word" / "document.xml"
  127. modified_authors = get_tracked_change_authors(modified_xml)
  128. if not modified_authors:
  129. return default
  130. original_authors = _get_authors_from_docx(original_docx)
  131. new_changes: dict[str, int] = {}
  132. for author, count in modified_authors.items():
  133. original_count = original_authors.get(author, 0)
  134. diff = count - original_count
  135. if diff > 0:
  136. new_changes[author] = diff
  137. if not new_changes:
  138. return default
  139. if len(new_changes) == 1:
  140. return next(iter(new_changes))
  141. raise ValueError(
  142. f"Multiple authors added new changes: {new_changes}. "
  143. "Cannot infer which author to validate."
  144. )