"""Review filter helpers for the Scrub replacement workflow. The filters are intentionally presentation-oriented. They help the user focus on specific rows, but they should not silently remove hidden rows from the export workflow unless the UI explicitly implements safe merge-back behaviour. """ from __future__ import annotations from typing import Any, Iterable, Mapping FILTER_SHOW_ALL = "Toon alles" FILTER_NEEDS_REVIEW = "Alleen controle nodig" FILTER_LEGAL_REFERENCES = "Alleen juridische referenties" FILTER_NAMES_ADDRESSES = "Alleen namen/adressen" FILTER_LOW_CONFIDENCE = "Alleen lage zekerheid" REVIEW_FILTER_OPTIONS = [ FILTER_SHOW_ALL, FILTER_NEEDS_REVIEW, FILTER_LEGAL_REFERENCES, FILTER_NAMES_ADDRESSES, FILTER_LOW_CONFIDENCE, ] LEGAL_REFERENCE_ENTITY_TYPES = { "NL_LEGAL_CASE_NUMBER", "NL_ROLNUMMER", "NL_REKESTNUMMER", "NL_PARKETNUMMER", "NL_DOSSIER_NUMBER", "NL_CLIENT_NUMBER", "NL_CJIB_NUMBER", "NL_POLICE_REPORT_NUMBER", "NL_INSURANCE_CLAIM_NUMBER", "NL_INCIDENT_NUMBER", "NL_CLAIM_NUMBER", "NL_OTHER_REFERENCE", "NL_CLIENT_REFERENCE", "NL_CASE_REFERENCE", "NL_INTERNAL_REFERENCE", "NL_CONTEXTUAL_REFERENCE", "NL_INVOICE_NUMBER", "NL_ORDER_OR_CONTRACT_NUMBER", "NL_SCHOOL_REFERENCE", "NL_CHILD_PROTECTION_REFERENCE", "NL_EMPLOYMENT_REFERENCE", "NL_INSURANCE_REFERENCE", "NL_HEALTHCARE_REFERENCE", "NL_POLICE_REFERENCE", "NL_IMMIGRATION_REFERENCE", "NL_MUNICIPAL_REFERENCE", "NL_REAL_ESTATE_REFERENCE", "NL_VEHICLE_REFERENCE", "NL_OBJECT_REFERENCE", "NL_SUSPICIOUS_REFERENCE_CANDIDATE", "NL_POSSIBLE_LICENSE_PLATE", "NL_ECLI", "NL_KVK_NUMBER", "NL_VAT_NUMBER", "NL_BIG_NUMBER", } NAME_ADDRESS_ENTITY_TYPES = { "PERSON", "NL_LEGAL_PARTY_NAME", "LOCATION", "NL_ADDRESS", "NL_POSTCODE", "ORGANIZATION", "NL_COURT_OR_AUTHORITY", } def _cell(row: Mapping[str, Any], key: str, default: Any = "") -> Any: value = row.get(key, default) return default if value is None else value def _normalised_entity_type(row: Mapping[str, Any]) -> str: return str(_cell(row, "entity_type", "")).strip().upper() def _normalised_status(row: Mapping[str, Any]) -> str: status = str(_cell(row, "review_status", "")).strip().lower() label = str(_cell(row, "review_status_label", "")).strip().lower() if status: return status if label == "controle nodig": return "needs_review" if label == "automatisch vervangen": return "auto_detected" if label == "handmatig toegevoegd": return "manual" if label == "onthouden vervanging": return "remembered" return "" def _score(row: Mapping[str, Any]) -> float | None: raw = _cell(row, "score", None) if raw in (None, ""): return None try: return float(raw) except Exception: return None def row_matches_review_filter(row: Mapping[str, Any], filter_label: str) -> bool: if filter_label == FILTER_SHOW_ALL: return True if filter_label == FILTER_NEEDS_REVIEW: return _normalised_status(row) == "needs_review" entity_type = _normalised_entity_type(row) if filter_label == FILTER_LEGAL_REFERENCES: return entity_type in LEGAL_REFERENCE_ENTITY_TYPES if filter_label == FILTER_NAMES_ADDRESSES: return entity_type in NAME_ADDRESS_ENTITY_TYPES if filter_label == FILTER_LOW_CONFIDENCE: confidence = str(_cell(row, "confidence", "")).strip().lower() score = _score(row) return confidence == "laag" or (score is not None and score < 0.60) return True def filter_review_records(records: Iterable[Mapping[str, Any]], filter_label: str) -> list[Mapping[str, Any]]: return [row for row in records if row_matches_review_filter(row, filter_label)] def filter_review_dataframe(df, filter_label: str): """Filter a pandas-like DataFrame without importing pandas at module import. Tests use the record-level function. The Streamlit app can use this helper with a pandas DataFrame. """ if filter_label == FILTER_SHOW_ALL: return df if df is None or len(df) == 0: return df mask = [row_matches_review_filter(row, filter_label) for _, row in df.iterrows()] return df.loc[mask].reset_index(drop=True)