""" ClauseGuard — Contract Comparison Engine v3.0 ═════════════════════════════════════════════ FIXED in v3.0: • Semantic similarity using sentence embeddings (when available) • Better clause type detection with legal taxonomy • Improved diff visualization • Fallback to SequenceMatcher when embeddings unavailable """ import re from difflib import SequenceMatcher from collections import defaultdict # Try to load sentence-transformers for semantic comparison _HAS_EMBEDDINGS = False _embedder = None try: from sentence_transformers import SentenceTransformer, util _HAS_EMBEDDINGS = True except ImportError: pass def _load_embedder(): global _embedder if _HAS_EMBEDDINGS and _embedder is None: try: _embedder = SentenceTransformer("all-MiniLM-L6-v2") print("[ClauseGuard] Sentence embeddings loaded for comparison") except Exception as e: print(f"[ClauseGuard] Embeddings not available: {e}") def _normalize_clause(text): """Normalize clause text for comparison.""" text = text.lower() text = re.sub(r'[^a-z0-9\s]', ' ', text) text = re.sub(r'\s+', ' ', text).strip() return text def _clause_similarity(a, b): """Compute similarity using semantic embeddings or string matching.""" if _embedder is not None: try: emb_a = _embedder.encode(a[:512], convert_to_tensor=True) emb_b = _embedder.encode(b[:512], convert_to_tensor=True) sim = util.cos_sim(emb_a, emb_b).item() return max(0, min(1, sim)) except Exception: pass # Fallback to string matching return SequenceMatcher(None, _normalize_clause(a), _normalize_clause(b)).ratio() def _extract_clause_type(clause_text): """Clause type detection with legal taxonomy.""" text_lower = clause_text.lower() type_keywords = { "governing law": ["govern", "law of", "jurisdiction of", "applicable law"], "termination": ["terminat", "cancel", "expir"], "indemnification": ["indemnif", "hold harmless", "defend and indemnify"], "confidentiality": ["confidential", "non-disclosure", "nda", "proprietary"], "liability": ["liability", "liable", "damages", "limitation of"], "payment": ["payment", "fee", "price", "compensat", "invoice", "remit"], "intellectual property": ["intellectual property", "ip rights", "copyright", "patent", "trademark"], "warranty": ["warrant", "guarantee", "representation"], "force majeure": ["force majeure", "act of god", "beyond control"], "arbitration": ["arbitrat", "mediation", "dispute resolution"], "assignment": ["assign", "transfer of rights"], "non-compete": ["non-compete", "not compete", "competition"], "renewal": ["renew", "extend", "automatic renewal"], "effective date": ["effective date", "commencement"], "insurance": ["insurance", "coverage", "policy of insurance"], "audit": ["audit", "inspection", "examination of records"], "data protection": ["data protection", "privacy", "personal data", "gdpr", "ccpa"], "notice": ["notice", "notification", "written notice"], } for ctype, keywords in type_keywords.items(): if any(kw in text_lower for kw in keywords): return ctype return "general" def compare_contracts(text_a, text_b, clauses_a=None, clauses_b=None): """Compare two contracts with semantic similarity.""" if not text_a or not text_b: return {"error": "Both contracts required"} # Try to load embedder _load_embedder() # Split into clauses if not provided if clauses_a is None: clauses_a = _split_clauses(text_a) if clauses_b is None: clauses_b = _split_clauses(text_b) # Fix 9: Detect contract types and flag cross-domain comparisons _CONTRACT_TYPE_KEYWORDS = { "employment": ["employee", "employer", "salary", "compensation", "benefits", "vacation", "severance", "at-will"], "lease": ["landlord", "tenant", "rent", "premises", "lease", "occupancy", "security deposit", "eviction"], "service": ["service provider", "customer", "SLA", "deliverables", "statement of work", "SOW"], "nda": ["confidential", "non-disclosure", "disclosing party", "receiving party"], "saas": ["subscription", "SaaS", "cloud", "uptime", "API", "data processing"], "purchase": ["buyer", "seller", "purchase order", "goods", "shipment", "delivery"], } def _detect_contract_type(text): text_lower = text.lower() scores = {} for ctype, keywords in _CONTRACT_TYPE_KEYWORDS.items(): scores[ctype] = sum(1 for kw in keywords if kw.lower() in text_lower) best = max(scores, key=scores.get) return best if scores[best] >= 2 else "general" type_a = _detect_contract_type(text_a) type_b = _detect_contract_type(text_b) is_cross_domain = type_a != type_b and type_a != "general" and type_b != "general" # Build clause type maps type_map_a = defaultdict(list) type_map_b = defaultdict(list) for c in clauses_a: type_map_a[_extract_clause_type(c)].append(c) for c in clauses_b: type_map_b[_extract_clause_type(c)].append(c) # Find matches matched_a = set() matched_b = set() modified = [] # Fix 10: Raise thresholds to reject false "modified" matches SIMILARITY_THRESHOLD = 0.75 # was 0.70 — too many false matches MODIFIED_THRESHOLD = 0.55 # was 0.40 — "Good Reason" ≠ "Force Majeure" for i, ca in enumerate(clauses_a): best_sim = 0 best_j = -1 for j, cb in enumerate(clauses_b): if j in matched_b: continue sim = _clause_similarity(ca, cb) if sim > best_sim: best_sim = sim best_j = j if best_sim >= SIMILARITY_THRESHOLD: matched_a.add(i) matched_b.add(best_j) if best_sim < 0.95: modified.append({ "type": "modified", "similarity": round(best_sim, 3), "clause_a": ca[:200], "clause_b": clauses_b[best_j][:200], "clause_type": _extract_clause_type(ca), }) elif best_sim >= MODIFIED_THRESHOLD: matched_a.add(i) if best_j >= 0: matched_b.add(best_j) modified.append({ "type": "partial", "similarity": round(best_sim, 3), "clause_a": ca[:200], "clause_b": clauses_b[best_j][:200] if best_j >= 0 else "", "clause_type": _extract_clause_type(ca), }) removed = [clauses_a[i] for i in range(len(clauses_a)) if i not in matched_a] added = [clauses_b[j] for j in range(len(clauses_b)) if j not in matched_b] # Compute alignment score total_pairs = max(len(clauses_a), len(clauses_b)) if total_pairs > 0: alignment = len(matched_a) / total_pairs else: alignment = 0.0 # Risk delta: compare risk keywords with context risk_keywords = ["unlimited", "unilateral", "waive", "arbitration", "indemnif", "not liable", "no warranty", "sole discretion", "terminate", "non-compete", "liquidated damages", "uncapped"] risk_a = sum(1 for kw in risk_keywords if kw in text_a.lower()) risk_b = sum(1 for kw in risk_keywords if kw in text_b.lower()) if risk_a > risk_b + 2: risk_delta = "Contract A is significantly riskier" risk_winner = "B" elif risk_b > risk_a + 2: risk_delta = "Contract B is significantly riskier" risk_winner = "A" elif risk_a > risk_b: risk_delta = "Contract A is slightly riskier" risk_winner = "B" elif risk_b > risk_a: risk_delta = "Contract B is slightly riskier" risk_winner = "A" else: risk_delta = "Similar risk profiles" risk_winner = "tie" # Fix 9: Cross-domain warning if is_cross_domain: risk_delta = f"Cross-domain comparison ({type_a} vs {type_b}) — risk delta not meaningful across different contract types" risk_winner = "cross-domain" comparison_method = "semantic (sentence embeddings)" if _embedder is not None else "lexical (string matching)" return { "alignment_score": round(alignment, 3), "contract_a_clauses": len(clauses_a), "contract_b_clauses": len(clauses_b), "contract_a_type": type_a, "contract_b_type": type_b, "is_cross_domain": is_cross_domain, "added_clauses": [{"text": c[:200], "type": _extract_clause_type(c)} for c in added[:50]], "removed_clauses": [{"text": c[:200], "type": _extract_clause_type(c)} for c in removed[:50]], "modified_clauses": modified[:50], "risk_delta": risk_delta, "risk_winner": risk_winner, "comparison_method": comparison_method, "type_map_a": {k: len(v) for k, v in type_map_a.items()}, "type_map_b": {k: len(v) for k, v in type_map_b.items()}, } def _split_clauses(text): """Split text into clauses.""" text = re.sub(r'\n{3,}', '\n\n', text.strip()) # Try section-based splitting first section_splits = re.split( r'(?:\n\n)(?=\d+[.)]\s|\([a-z]\)\s|(?:Section|Article|Clause)\s+\d+)', text ) if len(section_splits) >= 3: return [p.strip() for p in section_splits if len(p.strip()) > 30] # Fallback to paragraph/sentence splitting parts = re.split( r'(?<=[.!?])\s+(?=[A-Z0-9(])|(?:\n\n)', text ) return [p.strip() for p in parts if len(p.strip()) > 30] def render_comparison_html(result): """Render comparison results as HTML for Gradio.""" if "error" in result: return f'

{result["error"]}

' method = result.get("comparison_method", "unknown") method_badge = f'
Comparison method: {method}
' html = f'''
{method_badge}
{result["contract_a_clauses"]}
Clauses in Contract A
{result["contract_b_clauses"]}
Clauses in Contract B
{result["alignment_score"]*100:.1f}%
Alignment Score
⚖️ {result["risk_delta"]}
''' # Modified clauses if result["modified_clauses"]: html += '

📝 Modified Clauses

' for m in result["modified_clauses"][:20]: html += f'''
{m["clause_type"].upper()} · Similarity: {m["similarity"]*100:.0f}%
{m["clause_a"][:150]}...
{m["clause_b"][:150]}...
''' html += '
' # Added clauses if result["added_clauses"]: html += '

➕ Added in Contract B

' for a in result["added_clauses"][:15]: html += f'
{a["type"].upper()} · {a["text"][:150]}...
' html += '
' # Removed clauses if result["removed_clauses"]: html += '

➖ Removed from Contract A

' for r in result["removed_clauses"][:15]: html += f'
{r["type"].upper()} · {r["text"][:150]}...
' html += '
' html += '
' return html