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