aibridze/document_intelligence

1"""2Extraction Validator — post-extraction quality checks.3 4Validates extracted data against the source document text:5  1. Party names exist in the source text6  2. Dates are valid formats and exist in source7  3. Amounts/numbers are plausible8  4. Critical fields are non-empty9  5. Computes per-field and overall semantic confidence10"""11 12import re13import logging14from typing import Dict, Any, List, Tuple15from datetime import datetime16 17from app.models.extraction_schemas import ContractType18 19logger = logging.getLogger("trident_poc")20 21 22# Fields that MUST be non-empty for a valid extraction23REQUIRED_FIELDS = {24    ContractType.NDA: {25        "parties.disclosing_party": "Disclosing party name",26        "parties.receiving_party": "Receiving party name",27        "purpose": "Purpose of the NDA",28    },29    ContractType.SERVICE_PROVIDER: {30        "parties.company": "Company name",31        "parties.service_provider": "Service provider name",32        "scope_of_work": "Scope of work",33    },34    ContractType.SUPPLY_CUM_SERVICE: {35        "parties.company": "Company name",36        "parties.service_provider": "Service provider name",37        "project_details.project_name": "Project name",38        "pricing.total_value": "Total value",39    },40    ContractType.CONSULTANCY: {41        "parties.company": "Company name",42        "parties.consultant": "Consultant name",43        "scope": "Scope of consultancy",44        "consideration.amount": "Consultancy fee",45    },46    ContractType.CUSTOMER_SUPPLY: {47        "parties.buyer": "Buyer name",48        "parties.supplier": "Supplier name",49    },50}51 52 53class ExtractionValidator:54    """Validates extraction results against source text."""55 56    def validate(57        self,58        extraction: dict,59        source_text: str,60        contract_type: ContractType,61    ) -> Dict[str, Any]:62        """63        Run all validation checks and return a comprehensive report.64 65        Returns:66            {67                "overall_confidence": float (0-1),68                "field_fill_rate": float (0-1),69                "text_match_rate": float (0-1),70                "required_fields_rate": float (0-1),71                "issues": [{"field": str, "issue": str, "severity": str}],72                "field_scores": {"field_path": float},73            }74        """75        source_lower = source_text.lower()76        issues = []77        field_scores = {}78 79        # 1. Field fill rate80        total_fields, filled_fields = self._count_fields(extraction)81        field_fill_rate = filled_fields / max(total_fields, 1)82 83        # 2. Required fields check84        required = REQUIRED_FIELDS.get(contract_type, {})85        required_filled = 086        for path, label in required.items():87            value = self._get_nested(extraction, path)88            if value and str(value).strip():89                required_filled += 190                field_scores[path] = 1.091            else:92                issues.append({93                    "field": path,94                    "issue": f"Required field '{label}' is empty",95                    "severity": "high",96                })97                field_scores[path] = 0.098 99        required_rate = required_filled / max(len(required), 1)100 101        # 3. Text match validation — verify extracted values exist in source102        match_checks = 0103        match_hits = 0104        text_fields = self._extract_text_values(extraction)105 106        for field_path, value in text_fields:107            if not value or len(str(value)) < 3:108                continue109 110            match_checks += 1111            val_str = str(value).lower().strip()112 113            # Check if the core of the value exists in source text114            # For names: check the main words115            # For amounts: check the number part116            if self._value_exists_in_text(val_str, source_lower):117                match_hits += 1118                field_scores[field_path] = field_scores.get(field_path, 0) + 1.0119            else:120                # Partial match — check key words121                words = [w for w in val_str.split() if len(w) > 3]122                word_matches = sum(1 for w in words if w in source_lower)123                partial_score = word_matches / max(len(words), 1)124 125                if partial_score >= 0.5:126                    match_hits += 0.7127                    field_scores[field_path] = field_scores.get(field_path, 0) + 0.7128                else:129                    issues.append({130                        "field": field_path,131                        "issue": f"Value '{str(value)[:50]}...' not found in source text",132                        "severity": "medium",133                    })134                    field_scores[field_path] = field_scores.get(field_path, 0) + 0.2135 136        text_match_rate = match_hits / max(match_checks, 1)137 138        # 4. Date format validation139        date_fields = self._find_date_fields(extraction)140        for path, value in date_fields:141            if value and not self._is_valid_date(str(value)):142                issues.append({143                    "field": path,144                    "issue": f"Invalid date format: '{value}'",145                    "severity": "low",146                })147 148        # 5. Overall confidence (weighted average)149        overall = (150            field_fill_rate * 0.35 +        # 35% weight on field completeness151            required_rate * 0.35 +            # 35% weight on critical fields152            text_match_rate * 0.30            # 30% weight on source text verification153        )154 155        return {156            "overall_confidence": round(overall, 3),157            "field_fill_rate": round(field_fill_rate, 3),158            "text_match_rate": round(text_match_rate, 3),159            "required_fields_rate": round(required_rate, 3),160            "total_fields": total_fields,161            "filled_fields": filled_fields,162            "issues_count": len(issues),163            "issues": issues[:20],  # Cap at 20 to avoid noise164            "field_scores": field_scores,165        }166 167    # ─── Field Counting ───────────────────────────────────────────168 169    def _count_fields(170        self,171        data: dict,172        _depth: int = 0,173        _skip: set = None,174    ) -> Tuple[int, int]:175        """Count total and filled fields (first 3 levels of nesting)."""176        if _skip is None:177            _skip = {"_extraction_meta", "handwritten_data", "error"}178        if _depth > 3:179            return (0, 0)180 181        total = 0182        filled = 0183 184        for key, value in data.items():185            if key in _skip:186                continue187 188            if isinstance(value, dict):189                sub_total, sub_filled = self._count_fields(value, _depth + 1, _skip)190                total += sub_total191                filled += sub_filled192            elif isinstance(value, list):193                total += 1194                if len(value) > 0:195                    filled += 1196            else:197                total += 1198                if value is not None and value != "":199                    filled += 1200 201        return (total, filled)202 203    # ─── Text Matching ────────────────────────────────────────────204 205    def _extract_text_values(206        self,207        data: dict,208        prefix: str = "",209        _depth: int = 0,210    ) -> List[Tuple[str, str]]:211        """Extract all string values with their field paths."""212        if _depth > 4:213            return []214 215        results = []216        skip = {"_extraction_meta", "handwritten_data", "error", "unstructured_handwriting"}217 218        for key, value in data.items():219            if key in skip:220                continue221            path = f"{prefix}.{key}" if prefix else key222 223            if isinstance(value, dict):224                results.extend(self._extract_text_values(value, path, _depth + 1))225            elif isinstance(value, list):226                for i, item in enumerate(value):227                    if isinstance(item, dict):228                        results.extend(229                            self._extract_text_values(item, f"{path}[{i}]", _depth + 1)230                        )231                    elif isinstance(item, str) and len(item) > 3:232                        results.append((f"{path}[{i}]", item))233            elif isinstance(value, str) and len(value) > 3:234                results.append((path, value))235 236        return results237 238    @staticmethod239    def _value_exists_in_text(value: str, source_lower: str) -> bool:240        """Check if a value (or its essential parts) exist in the source text."""241        # Direct containment check242        if value in source_lower:243            return True244 245        # For amounts: strip currency symbols and check the number246        amount_match = re.search(r'[\d,]+\.?\d*', value)247        if amount_match:248            number = amount_match.group().replace(",", "")249            if number in source_lower.replace(",", ""):250                return True251 252        # For names: check if all significant words appear253        words = [w for w in value.split() if len(w) > 2 and w.isalpha()]254        if words and all(w in source_lower for w in words):255            return True256 257        return False258 259    # ─── Date Validation ──────────────────────────────────────────260 261    def _find_date_fields(self, data: dict, prefix: str = "") -> List[Tuple[str, str]]:262        """Find all fields likely containing dates."""263        date_fields = []264        date_keywords = {"date", "start", "end", "effective", "expiry", "execution", "termination"}265 266        for key, value in data.items():267            path = f"{prefix}.{key}" if prefix else key268 269            if isinstance(value, dict):270                date_fields.extend(self._find_date_fields(value, path))271            elif isinstance(value, str) and any(kw in key.lower() for kw in date_keywords):272                date_fields.append((path, value))273 274        return date_fields275 276    @staticmethod277    def _is_valid_date(value: str) -> bool:278        """Check if a string looks like a valid date."""279        if not value or value.lower() in ("null", "none", "n/a"):280            return True  # Null is acceptable, not an error281 282        # Common date patterns283        patterns = [284            r'\d{1,2}[/\-\.]\d{1,2}[/\-\.]\d{2,4}',          # DD/MM/YYYY285            r'\d{4}[/\-\.]\d{1,2}[/\-\.]\d{1,2}',            # YYYY/MM/DD286            r'\d{1,2}\s+\w+\s+\d{4}',                         # 07 March 2026287            r'\w+\s+\d{1,2},?\s+\d{4}',                       # March 07, 2026288            r'\d{1,2}(?:st|nd|rd|th)\s+\w+\s+\d{4}',         # 7th March 2026289            r'\d{1,2}(?:st|nd|rd|th)\s+day\s+of\s+\w+',      # 7th day of March290            r'\w+\s+\d{1,2}(?:st|nd|rd|th)?,?\s+\d{4}',      # November 1, 2025291        ]292 293        for pattern in patterns:294            if re.search(pattern, value, re.IGNORECASE):295                return True296 297        return False298 299    # ─── Helpers ──────────────────────────────────────────────────300 301    @staticmethod302    def _get_nested(data: dict, path: str) -> Any:303        """Get a nested dict value by dot-separated path."""304        keys = path.split(".")305        current = data306        for key in keys:307            if isinstance(current, dict):308                current = current.get(key)309            else:310                return None311        return current