from __future__ import annotations import json import os import subprocess from collections import Counter from pathlib import Path from fastapi import HTTPException from docx_text import get_strict_invariant_hash, normalize_visible_text from models import ApplyResult, ResolveStrategy, ReviewFinding from comment_target_resolver import resolve_findings_against_docx, resolve_single_target, FindingResolution from logger import setup_logger logger = setup_logger(__name__) ROOT_DIR = Path(__file__).resolve().parent.parent COMMENTER_PROJECT = ROOT_DIR / "tools" / "docx-commenter" / "docx-commenter.csproj" def _hash_text(input_text: str) -> str: import hashlib return hashlib.sha256(input_text.encode("utf-8")).hexdigest()[:16] def _comment_ops(findings: list[ReviewFinding], resolution_map: dict | None = None) -> dict: return { "author": os.environ.get("MIKE_COMMENT_AUTHOR", "Tabularrr"), "comments": [ { "id": finding.id, "model_block_id": finding.model_block_id, "comment_key": finding.comment_key, "paragraph_id": finding.start_paragraph_id, "end_paragraph_id": finding.end_paragraph_id, "target_text": finding.target_text, "anchor_mode": finding.anchor_mode, "before_anchor": finding.before_anchor, "after_anchor": finding.after_anchor, "comment": finding.comment, "stable_id": finding.stable_id, "native_para_id": finding.native_para_id, "text_hash": finding.text_hash, "xml_path": finding.xml_path, "type": finding.op_type, "new_text": finding.new_text, "insert_text": finding.insert_text, "position": finding.position, "rewrite_mode": finding.rewrite_mode, "old_span": finding.old_span, "new_span": finding.new_span, "bookmark_name": finding.bookmark_name, } for finding in findings ], } def _run_docx_commenter( input_docx: Path, output_docx: Path, ops_path: Path, report_path: Path, dry_run: bool = False, ) -> list[dict]: if not COMMENTER_PROJECT.exists(): raise HTTPException( status_code=500, detail=f"DOCX commenter project is missing: {COMMENTER_PROJECT}", ) command = [ "dotnet", "run", "--project", str(COMMENTER_PROJECT), "--", "--input", str(input_docx), "--ops", str(ops_path), "--output", str(output_docx), "--report", str(report_path), ] if dry_run: command.extend(["--dry-run", "true"]) logger.info(f"Running docx-commenter with command: {' '.join(command)}") completed = subprocess.run( command, cwd=str(ROOT_DIR), text=True, capture_output=True, timeout=120 ) if completed.returncode != 0: logger.error(f"DOCX commenter failed. Stdout: {completed.stdout}, Stderr: {completed.stderr}") raise HTTPException( status_code=500, detail=f"DOCX commenter failed: {completed.stderr or completed.stdout}", ) if not report_path.exists(): logger.error("DOCX commenter did not create an apply report.") raise HTTPException( status_code=500, detail="DOCX commenter did not create an apply report." ) logger.info("DOCX commenter completed successfully.") raw_report = json.loads(report_path.read_text(encoding="utf-8")) raw_results = raw_report.get("results") if isinstance(raw_report, dict) else [] if not isinstance(raw_results, list): return [] return raw_results def apply_docx_comments( *, input_docx: Path, output_docx: Path, findings: list[ReviewFinding], work_dir: Path, allow_paragraph_fallback: bool = False, track_changes: bool = False, ) -> list[ApplyResult]: from docx_text import extract_docx_paragraphs from comment_target_resolver import generate_editable_subspan_candidates logger.info(f"Applying docx comments. Input: {input_docx}, Output: {output_docx}, Total Findings: {len(findings)}, Track Changes: {track_changes}") work_dir.mkdir(parents=True, exist_ok=True) output_docx.parent.mkdir(parents=True, exist_ok=True) paragraphs = extract_docx_paragraphs(input_docx) para_map = {p.id: p.text for p in paragraphs} exact_findings = [finding for finding in findings if finding.anchor_mode == "exact"] exact_resolution_results = resolve_findings_against_docx( input_docx, exact_findings, allow_paragraph_fallback=allow_paragraph_fallback, ) if exact_findings else [] exact_resolution_by_id = {r.finding.id: r for r in exact_resolution_results} resolution_results: list[FindingResolution] = [] for finding in findings: if finding.anchor_mode == "exact": resolved = exact_resolution_by_id.get(finding.id) if resolved is not None: resolution_results.append(resolved) continue resolution_results.append(FindingResolution( finding=finding, original_target_text=finding.target_text, original_before_anchor=finding.before_anchor, original_after_anchor=finding.after_anchor, strategy="paragraph_anchor" if finding.anchor_mode == "paragraph" else "not_resolved", confidence=1.0 if finding.anchor_mode == "paragraph" else 0.0, was_resolved=finding.anchor_mode == "paragraph", )) resolved_findings = [r.finding for r in resolution_results] ops_path = work_dir / "comment_ops_dryrun1.json" report_path = work_dir / "comment_report_dryrun1.json" ops_path.write_text( json.dumps(_comment_ops(resolved_findings), indent=2, ensure_ascii=False), encoding="utf-8", ) dry_run_results = _run_docx_commenter(input_docx, output_docx, ops_path, report_path, dry_run=True) needs_subspan = [] for res in dry_run_results: code = res.get("failure_code") if code in ("crosses_complex_container", "no_editable_runs"): needs_subspan.append(res.get("id")) if needs_subspan: candidate_ops = [] candidate_mapping = {} for r_finding in resolved_findings: if r_finding.id in needs_subspan and r_finding.anchor_mode == "exact": para_text = para_map.get(r_finding.start_paragraph_id, "") if not para_text: continue candidates = generate_editable_subspan_candidates(para_text, r_finding.target_text) for i, cand in enumerate(candidates): cand_id = f"{r_finding.id}_cand_{i}" candidate_mapping[cand_id] = (r_finding.id, cand) candidate_ops.append({ "id": cand_id, "paragraph_id": r_finding.start_paragraph_id, "end_paragraph_id": r_finding.end_paragraph_id, "target_text": cand, "anchor_mode": "exact", "before_anchor": "", "after_anchor": "", "comment": "dry run", }) if candidate_ops: ops_path2 = work_dir / "comment_ops_dryrun2.json" report_path2 = work_dir / "comment_report_dryrun2.json" ops_path2.write_text( json.dumps({"author": "Tabularrr", "comments": candidate_ops}, indent=2, ensure_ascii=False), encoding="utf-8", ) dry_run_results2 = _run_docx_commenter(input_docx, output_docx, ops_path2, report_path2, dry_run=True) successful_candidates = {} for res in dry_run_results2: if res.get("status") == "applied" or res.get("failure_code") == "editable": cand_id = res.get("id") if cand_id in candidate_mapping: finding_id, cand_text = candidate_mapping[cand_id] if finding_id not in successful_candidates: successful_candidates[finding_id] = cand_text for r in resolution_results: if r.finding.id in successful_candidates: r.finding = r.finding.model_copy(update={ "target_text": successful_candidates[r.finding.id], "before_anchor": "", "after_anchor": "", }) r.strategy = "editable_subspan" r.was_resolved = True resolved_findings = [r.finding for r in resolution_results] resolution_map: dict[str, dict] = {} for r in resolution_results: if r.was_resolved or r.strategy in ("editable_subspan", "paragraph_anchor"): resolution_map[r.finding.id] = { "original_target_text": r.original_target_text[:200], "resolved_target_text": r.finding.target_text[:200], "strategy": r.strategy, "confidence": r.confidence, } ops_path = work_dir / "comment_ops.json" report_path = work_dir / "comment_report.json" ops_path.write_text( json.dumps(_comment_ops(resolved_findings), indent=2, ensure_ascii=False), encoding="utf-8", ) raw_results = _run_docx_commenter(input_docx, output_docx, ops_path, report_path, dry_run=False) if not track_changes: try: before_hash = get_strict_invariant_hash(input_docx) after_hash = get_strict_invariant_hash(output_docx) if before_hash != after_hash: before = normalize_visible_text(input_docx) after = normalize_visible_text(output_docx) error_msg = "SAFETY VIOLATION: Document body text or structure was modified during a comments-only review." for i in range(min(len(before.paragraphs), len(after.paragraphs))): b = before.paragraphs[i] a = after.paragraphs[i] if b.text != a.text: error_msg += f"\n- Mismatch at {b.id}:\n EXPECTED: \"{b.text}\"\n ACTUAL: \"{a.text}\"" break if len(before.paragraphs) != len(after.paragraphs): error_msg += f"\n- Paragraph count changed: {len(before.paragraphs)} -> {len(after.paragraphs)}" elif before_hash != after_hash and before.text_hash == after.text_hash: error_msg += "\n- Structural/Invisible change detected: Tracked changes, tables, or formatting were modified even though visible paragraph text remains the same." error_msg += "\n\nREASON: 'Comments-only' mode strictly forbids ANY document modifications (body text, tables, redlines, etc). You must ONLY add comments." error_msg += "\nACTION: If you intended to suggest a change, put it in the 'comment' field. Do not attempt to edit the contract text." raise HTTPException(status_code=422, detail=error_msg) except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=f"Invariant check failed: {str(e)}") return [ ApplyResult( id=item.get("id", ""), status=item.get("status", "not_applied"), message=item.get("message", ""), paragraph_id=item.get("paragraph_id"), match_count=item.get("match_count", 0), failure_code=item.get("failure_code"), model_block_id=item.get("model_block_id"), writer_paragraph_id=item.get("writer_paragraph_id"), native_para_id=item.get("native_para_id"), stable_id=item.get("stable_id"), xml_path=item.get("xml_path"), comment_key=item.get("comment_key"), expected_text_hash=item.get("expected_text_hash"), actual_text_hash=item.get("actual_text_hash"), original_target_text=resolution_map.get(item.get("id", ""), {}).get("original_target_text"), resolved_target_text=resolution_map.get(item.get("id", ""), {}).get("resolved_target_text"), resolve_strategy=_build_strategy(item, resolution_map), resolve_confidence=_build_confidence(item, resolution_map), ) for item in raw_results ] def _build_strategy(item: dict, resolution_map: dict) -> ResolveStrategy | None: fid = item.get("id", "") entry = resolution_map.get(fid) if entry: return entry.get("strategy", "whitespace_normalized") if item.get("status") == "applied": return "exact" return "not_resolved" def _build_confidence(item: dict, resolution_map: dict) -> float | None: fid = item.get("id", "") entry = resolution_map.get(fid) if entry: if item.get("status") == "applied": return entry.get("confidence", 0.95) return 0.0 if item.get("status") == "applied": return 1.0 return None