Clause Classification Systems

import re
import logging
import uuid
from typing import List, Dict, Optional
from pydantic import BaseModel, Field, ValidationError, field_validator
from enum import Enum

# Configure audit logging for compliance tracking and operational visibility
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s"
)
logger = logging.getLogger("lease_classifier")

class ClauseType(str, Enum):
    RENT_ESCALATION = "rent_escalation"
    CAM_CHARGES = "cam_charges"
    RENEWAL_OPTION = "renewal_option"
    ASSIGNMENT_SUBLET = "assignment_sublet"
    DEFAULT_REMEDIES = "default_remedies"
    UNKNOWN = "unknown"

class ClassifiedClause(BaseModel):
    clause_id: str = Field(..., description="Deterministic UUID for audit tracking")
    clause_type: ClauseType
    raw_text: str = Field(..., min_length=10, description="Original extracted clause text")
    confidence: float = Field(..., ge=0.0, le=1.0, description="Classification confidence score")
    metadata: Dict[str, str] = Field(default_factory=dict)

    @field_validator("confidence")
    @classmethod
    def enforce_confidence_precision(cls, v: float) -> float:
        return round(v, 4)

class ClauseClassifier:
    CONFIDENCE_THRESHOLD = 0.75
    FALLBACK_ROUTING = ClauseType.DEFAULT_REMEDIES

    # Pre-compiled patterns for O(1) lookup and thread-safe execution
    _PATTERNS: Dict[ClauseType, re.Pattern] = {
        ClauseType.RENT_ESCALATION: re.compile(
            r"(?:rent|base\s+rent|annual\s+rent|minimum\s+rent).{0,60}(?:increase|escalat|adjust|CPI|indexation|step-up)",
            re.IGNORECASE | re.DOTALL
        ),
        ClauseType.CAM_CHARGES: re.compile(
            r"(?:common\s+area\s+maintenance|CAM|operating\s+expenses|OPEX|proportionate\s+share).{0,60}(?:charge|reimburse|allocate|expense|reconciliation)",
            re.IGNORECASE | re.DOTALL
        ),
        ClauseType.RENEWAL_OPTION: re.compile(
            r"(?:renew|extend|option\s+to\s+renew|extension\s+term|successive\s+term).{0,60}(?:notice|exercise|deadline|election)",
            re.IGNORECASE | re.DOTALL
        ),
        ClauseType.ASSIGNMENT_SUBLET: re.compile(
            r"(?:assign|sublet|transfer|consent\s+required|landlord\s+approval).{0,60}(?:right|prohibited|condition|restriction)",
            re.IGNORECASE | re.DOTALL
        ),
    }

    @classmethod
    def classify(cls, clause_text: str, clause_id: Optional[str] = None, model_confidence: Optional[float] = None) -> ClassifiedClause:
        """
        Classify a lease clause using hybrid rule-based and probabilistic routing.
        Returns a validated Pydantic model ready for downstream serialization.
        """
        target_id = clause_id or str(uuid.uuid4())
        matched_type = cls._rule_match(clause_text)
        rule_confidence = cls._calculate_rule_confidence(clause_text, matched_type)

        # Hybrid confidence: prioritize external ML output if provided, else use rule confidence
        final_confidence = model_confidence if model_confidence is not None else rule_confidence

        # Enforce threshold; route to fallback if below operational confidence
        if final_confidence < cls.CONFIDENCE_THRESHOLD:
            logger.warning(f"Low confidence ({final_confidence:.4f}) for clause {target_id}. Routing to {cls.FALLBACK_ROUTING.value}.")
            final_type = cls.FALLBACK_ROUTING
            final_confidence = max(final_confidence, 0.50)  # Floor for audit visibility
        else:
            final_type = matched_type if matched_type else ClauseType.UNKNOWN

        try:
            return ClassifiedClause(
                clause_id=target_id,
                clause_type=final_type,
                raw_text=clause_text.strip(),
                confidence=final_confidence,
                metadata={
                    "routing_method": "hybrid",
                    "threshold_applied": str(cls.CONFIDENCE_THRESHOLD),
                    "pattern_matched": matched_type.value if matched_type else "none"
                }
            )
        except ValidationError as e:
            logger.error(f"Schema validation failed for clause {target_id}: {e}")
            raise

    @classmethod
    def _rule_match(cls, text: str) -> Optional[ClauseType]:
        for clause_type, pattern in cls._PATTERNS.items():
            if pattern.search(text):
                return clause_type
        return None

    @classmethod
    def _calculate_rule_confidence(cls, text: str, matched_type: Optional[ClauseType]) -> float:
        """
        Heuristic confidence scoring based on pattern match density and keyword proximity.
        In production, replace with calibrated model probabilities from NLP pipelines.
        """
        if not matched_type:
            return 0.0
        pattern = cls._PATTERNS[matched_type]
        matches = pattern.findall(text)
        if not matches:
            return 0.0
        # Base confidence scales with match count, capped at 0.95 to reserve headroom for ML fusion
        return min(0.95, 0.65 + (len(matches) * 0.15))

# Execution Example for Lease Abstraction Workflows
if __name__ == "__main__":
    sample_clauses = [
        "Base Rent shall increase by 3.0% annually on each anniversary of the Commencement Date.",
        "Tenant shall pay its proportionate share of Common Area Maintenance charges as calculated per the annual reconciliation statement.",
        "Landlord grants Tenant the option to renew this Lease for one additional five-year term, provided written notice is delivered 180 days prior to expiration.",
        "Tenant shall not assign or sublet the Premises without Landlord's prior written consent, which shall not be unreasonably withheld."
    ]

    for idx, text in enumerate(sample_clauses, start=1):
        result = ClauseClassifier.classify(text, clause_id=f"LEASE-2024-CL-{idx:03d}")
        print(f"[{result.clause_type.value.upper()}] (Conf: {result.confidence:.2f}) -> {result.clause_id}")