Spaces:
Running
Running
| """Pure Scrub Key model helpers for SolidPrivacy Scrub. | |
| A Scrub Key is a local mapping between original values and scrubbed | |
| placeholders. The helpers in this module deliberately avoid UI integration, | |
| cloud processing, file-system persistence and time-based side effects. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from collections.abc import Iterable, Mapping | |
| from typing import Any | |
| SCRUB_KEY_SCHEMA = "solidprivacy.scrub_key" | |
| SCRUB_KEY_SCHEMA_VERSION = "1.0" | |
| REQUIRED_ITEM_FIELDS = ( | |
| "original_value", | |
| "placeholder", | |
| "entity_type", | |
| "type_label", | |
| "source", | |
| "review_status", | |
| "include_state", | |
| "timestamp", | |
| ) | |
| TRUE_VALUES = {"1", "true", "yes", "ja", "y", "on", "checked", "aangevinkt", "included", "meenemen"} | |
| FALSE_VALUES = {"0", "false", "no", "nee", "n", "off", "unchecked", "uit", "excluded", "niet meenemen"} | |
| FIELD_ALIASES: dict[str, tuple[str, ...]] = { | |
| "original_value": ( | |
| "original_value", | |
| "original", | |
| "found_text", | |
| "text", | |
| "Gevonden tekst", | |
| "gevonden_tekst", | |
| ), | |
| "placeholder": ( | |
| "placeholder", | |
| "replacement", | |
| "replace_with", | |
| "scrubbed_value", | |
| "Vervangen door", | |
| "vervangen_door", | |
| ), | |
| "entity_type": ( | |
| "entity_type", | |
| "entity", | |
| "type", | |
| "Entity Type", | |
| "Entiteitstype", | |
| ), | |
| "type_label": ( | |
| "type_label", | |
| "user_facing_type_label", | |
| "display_label", | |
| "Type gegeven", | |
| "type_gegeven", | |
| ), | |
| "source": ( | |
| "source", | |
| "Bron", | |
| "bron", | |
| ), | |
| "review_status": ( | |
| "review_status", | |
| "review_status_label", | |
| "status", | |
| "Status", | |
| ), | |
| "include": ( | |
| "include", | |
| "included", | |
| "include_state", | |
| "Meenemen", | |
| "meenemen", | |
| ), | |
| "timestamp": ( | |
| "timestamp", | |
| "created_at", | |
| "reviewed_at", | |
| "updated_at", | |
| "Tijdstip", | |
| "tijdstip", | |
| ), | |
| "document_label": ( | |
| "document_label", | |
| "project_label", | |
| "dossier_label", | |
| "Document", | |
| "Project", | |
| "Dossier", | |
| ), | |
| } | |
| def _normalise_rows(rows: Any) -> list[dict[str, Any]]: | |
| """Return review rows as dictionaries without importing pandas.""" | |
| if rows is None: | |
| return [] | |
| if hasattr(rows, "to_dict"): | |
| try: | |
| records = rows.to_dict(orient="records") | |
| return [dict(row) for row in records] | |
| except TypeError: | |
| pass | |
| if isinstance(rows, Mapping): | |
| return [dict(rows)] | |
| if isinstance(rows, Iterable) and not isinstance(rows, (str, bytes)): | |
| return [dict(row) for row in rows if isinstance(row, Mapping)] | |
| return [] | |
| def _text(value: Any) -> str: | |
| return str(value or "").strip() | |
| def _lower(value: Any) -> str: | |
| return _text(value).lower() | |
| def _first_text(row: Mapping[str, Any], field_name: str) -> str: | |
| for alias in FIELD_ALIASES[field_name]: | |
| if alias in row: | |
| return _text(row.get(alias)) | |
| return "" | |
| def _include_state(row: Mapping[str, Any]) -> str: | |
| raw_value = None | |
| for alias in FIELD_ALIASES["include"]: | |
| if alias in row: | |
| raw_value = row.get(alias) | |
| break | |
| if isinstance(raw_value, bool): | |
| return "included" if raw_value else "excluded" | |
| if raw_value is None: | |
| return "excluded" | |
| if isinstance(raw_value, (int, float)): | |
| return "included" if raw_value != 0 else "excluded" | |
| normalised = _lower(raw_value) | |
| if normalised in TRUE_VALUES: | |
| return "included" | |
| if normalised in FALSE_VALUES: | |
| return "excluded" | |
| return "excluded" | |
| def build_scrub_key(rows: Any, document_label: str | None = None) -> dict[str, Any]: | |
| """Build a deterministic Scrub Key from reviewed replacement rows. | |
| Excluded rows are omitted by design. This keeps the first pure model aligned | |
| with current export semantics: only rows selected by the reviewer become part | |
| of the reversible mapping. | |
| """ | |
| records = _normalise_rows(rows) | |
| fallback_document_label = _text(document_label) or None | |
| items: list[dict[str, Any]] = [] | |
| for row in records: | |
| include_state = _include_state(row) | |
| if include_state != "included": | |
| continue | |
| item_document_label = _first_text(row, "document_label") or fallback_document_label | |
| item = { | |
| "original_value": _first_text(row, "original_value"), | |
| "placeholder": _first_text(row, "placeholder"), | |
| "entity_type": _first_text(row, "entity_type"), | |
| "type_label": _first_text(row, "type_label"), | |
| "source": _first_text(row, "source"), | |
| "review_status": _first_text(row, "review_status"), | |
| "include_state": include_state, | |
| "timestamp": _first_text(row, "timestamp"), | |
| "document_label": item_document_label, | |
| } | |
| items.append(item) | |
| return { | |
| "schema": SCRUB_KEY_SCHEMA, | |
| "schema_version": SCRUB_KEY_SCHEMA_VERSION, | |
| "workflow": "Scrub → Review → Scrub Key → AI → Reinsert → Export → Audit", | |
| "privacy_model": "pseudonymization_not_full_anonymization", | |
| "reversible": True, | |
| "storage_policy": "local_only_protect_key", | |
| "external_ai_policy": "do_not_share_key_unless_explicitly_intended_and_allowed", | |
| "excluded_rows_policy": "omitted", | |
| "document_label": fallback_document_label, | |
| "item_count": len(items), | |
| "items": items, | |
| } | |
| def scrub_key_to_json(scrub_key: Mapping[str, Any]) -> str: | |
| """Serialize a Scrub Key to stable, human-readable JSON.""" | |
| return json.dumps(dict(scrub_key), ensure_ascii=False, indent=2, sort_keys=True) | |
| def scrub_key_from_json(text: str) -> dict[str, Any]: | |
| """Load a Scrub Key JSON string into a dictionary.""" | |
| loaded = json.loads(text) | |
| if not isinstance(loaded, dict): | |
| raise ValueError("Scrub Key JSON must contain an object at the top level.") | |
| return loaded | |
| def validate_scrub_key(scrub_key: Any) -> list[str]: | |
| """Return validation messages for a Scrub Key. | |
| An empty list means the key is structurally valid. Validation is deliberately | |
| conservative and does not attempt to verify whether values are real PII. | |
| """ | |
| issues: list[str] = [] | |
| if not isinstance(scrub_key, Mapping): | |
| return ["Scrub Key must be a dictionary."] | |
| if scrub_key.get("schema") != SCRUB_KEY_SCHEMA: | |
| issues.append(f"Missing or invalid schema: expected {SCRUB_KEY_SCHEMA}.") | |
| if not scrub_key.get("schema_version"): | |
| issues.append("Missing schema_version.") | |
| if scrub_key.get("reversible") is not True: | |
| issues.append("Scrub Key must explicitly mark reversible=true.") | |
| if scrub_key.get("privacy_model") != "pseudonymization_not_full_anonymization": | |
| issues.append("Scrub Key must explicitly mark pseudonymization, not full anonymization.") | |
| if scrub_key.get("excluded_rows_policy") != "omitted": | |
| issues.append("Scrub Key must state excluded_rows_policy=omitted for this model version.") | |
| items = scrub_key.get("items") | |
| if not isinstance(items, list): | |
| issues.append("Missing or invalid items list.") | |
| return issues | |
| if scrub_key.get("item_count") != len(items): | |
| issues.append("item_count does not match the number of items.") | |
| for index, item in enumerate(items): | |
| if not isinstance(item, Mapping): | |
| issues.append(f"Item {index} must be a dictionary.") | |
| continue | |
| for field_name in REQUIRED_ITEM_FIELDS: | |
| if field_name not in item: | |
| issues.append(f"Item {index} missing required field: {field_name}.") | |
| elif _text(item.get(field_name)) == "": | |
| issues.append(f"Item {index} has empty required field: {field_name}.") | |
| if item.get("include_state") not in {"included", "excluded"}: | |
| issues.append(f"Item {index} has invalid include_state.") | |
| return issues | |