From 6e5800ef62fb276997644cb068ec5101d11e8cd9 Mon Sep 17 00:00:00 2001 From: Maziyar Panahi Date: Fri, 17 Jul 2026 01:35:41 +0200 Subject: [PATCH 1/2] feat: add SMS short-text de-identification --- docs/examples.md | 1 + docs/sms-short-text-deid.md | 161 ++++ examples/sms_deid_helpdesk_logs.py | 44 + mkdocs.yml | 1 + openmed/core/pii.py | 25 +- openmed/core/pipeline.py | 5 +- openmed/multimodal/__init__.py | 28 + openmed/multimodal/sms_messages.py | 865 ++++++++++++++++++ .../fixtures/sms/generic_messages.csv | 3 + .../fixtures/sms/rapidpro_messages.json | 33 + tests/unit/multimodal/test_sms_messages.py | 251 +++++ 11 files changed, 1408 insertions(+), 9 deletions(-) create mode 100644 docs/sms-short-text-deid.md create mode 100644 examples/sms_deid_helpdesk_logs.py create mode 100644 openmed/multimodal/sms_messages.py create mode 100644 tests/unit/multimodal/fixtures/sms/generic_messages.csv create mode 100644 tests/unit/multimodal/fixtures/sms/rapidpro_messages.json create mode 100644 tests/unit/multimodal/test_sms_messages.py diff --git a/docs/examples.md b/docs/examples.md index 97a30066f..cb228f184 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -70,6 +70,7 @@ content-security policy blocks the optional inline styling. | --- | --- | | `examples/pii_model_comparison.py` | Compares multiple PII models across shared sample text and summarizes extraction quality. | | `examples/pii_batch_processing.py` | Runs batch PII extraction and de-identification with `BatchProcessor(operation=...)`. | +| `examples/sms_deid_helpdesk_logs.py` | Redacts RapidPro-style JSON or generic CSV helpdesk exports with the `short_text` preset, contact pseudonymization, timestamp coarsening, and bounded batches. | | `examples/pii_multilingual_new_languages.py` | Exercises Dutch, Hindi, Telugu, Portuguese, Arabic, Japanese, and Turkish registry entries, locale-specific regex matches, and optional live extraction with the new public checkpoints. | | `examples/gradio_deid_app.py` | Interactive Gradio UI to paste synthetic text, pick a `mask`/`replace`/`hash` method, and view the de-identified output plus detected entities (optional `pip install gradio`). | | `examples/v16_policy_audit_release_gates.py` | Demonstrates v1.6 policy profiles, canonical spans, signed audit reports, review bundles, redaction previews, leakage heatmaps, and k-anonymity metrics without model downloads. | diff --git a/docs/sms-short-text-deid.md b/docs/sms-short-text-deid.md new file mode 100644 index 000000000..1512020a6 --- /dev/null +++ b/docs/sms-short-text-deid.md @@ -0,0 +1,161 @@ +# SMS short-text de-identification + +OpenMed's `short_text` preset is for clinical messages up to SMS scale, +including MomConnect-style helpdesk conversations and RapidPro flow/message +exports. It runs locally and combines the normal PII pipeline with deterministic +rules for message formats where sentence context is sparse or absent. + +RapidPro represents SMS addresses as contact URNs such as +`tel:+250788123123`, and a logical message can span more than one physical SMS. +The adapter therefore treats the message record—not a sentence—as the unit of +work. See RapidPro's documentation for its +[contact URN](https://rapidpro.github.io/rapidpro/docs/contacts/) and +[message](https://rapidpro.github.io/rapidpro/docs/msgs/) models. + +## What the preset changes + +The preset is named `short_text` and uses a lower high-recall threshold than +the default clinical-note path. It adds context-independent recognition for: + +- Kenyan, Ugandan, and South African E.164-style MSISDNs (`+254`, `+256`, + `+27`), spaced or punctuated variants, local `07xx` numbers, and 4–6 digit + shortcodes; +- contextual and common-shape national/medical identifiers; +- names following code-mixed maternal-health honorifics such as `mama`, + `maama`, `sisi`, `dada`, `mme`, and `ndugu`, including all-caps messages; +- common SMS and clinical abbreviations (`ANC`, `EDD`, `LMP`, `CHW`, `pls`, + `appt`, and similar) as allowed terms so a sparse context window does not turn + them into identifiers. + +Detection may normalize whitespace internally, but replacement offsets are +mapped to the original string. The preset never truncates, wraps, or otherwise +reflows source text. + +## One message + +Install the model-backed PII dependencies once, then keep the model cached for +offline use: + +```bash +uv pip install -e ".[hf]" +``` + +```python +from openmed.multimodal import deidentify_short_text + +message = " HABARI MAMA Amina Njeri call +254712345678 ANC " +result = deidentify_short_text(message) + +assert result.original_text == message +print(result.deidentified_text) +# " HABARI [PERSON] call [PHONE] ANC " +``` + +`deidentify_short_text` defaults to mask replacements and accepts the normal +OpenMed model, language, policy, loader, and mapping options. Code-mixed rules +remain language-independent; `lang` is still passed to the model-backed and +standard locale recognizers. + +## RapidPro-shaped JSON + +The JSON adapter accepts a top-level list, a single message record, or an object +whose record list is under `messages`, `results`, `runs`, `records`, or nested +`data`. Every string field named `text`, including flow-result `input.text`, is +de-identified. The surrounding object, row order, flow UUIDs, directions, and +other non-sensitive fields are retained. + +```python +import os + +from openmed.multimodal import redact_sms_json + +result = redact_sms_json( + "rapidpro-message-export.json", + "rapidpro-message-export.redacted.json", + batch_size=512, +) +print(result.summary.to_dict()) +``` + +Contact `urn`, `urns`, `contact_urn`, and string-valued `contact` fields are +replaced with HMAC-SHA256 pseudonyms. Nested contact `name`, `urn`, and +`address` values are pseudonymized too; contact UUIDs and flow UUIDs remain +intact. Repeated source values receive the same pseudonym within one run. + +By default a fresh random HMAC key is generated per run, so outputs from +different runs cannot be linked. Supply a secret key only when an authorized +workflow needs stable cross-run linkage: + +```python +result = redact_sms_json( + "messages.json", + "messages.redacted.json", + contact_hash_key=os.environ["OPENMED_SMS_CONTACT_HASH_KEY"], +) +``` + +Keep that key outside source control and logs. The summary contains only counts, +never contact values or message text. + +## Generic CSV and high-volume logs + +Generic CSV input must include `text` and at least one of `urn` or `contact`. +The expected RapidPro-compatible core columns are: + +```text +urn,contact,direction,text,sent_on +``` + +Extra columns are preserved in their original order. The CSV path reads and +writes incrementally, while `iter_redacted_sms_records` accepts any iterable of +mapping records. Both retain at most one source batch in memory and reuse +OpenMed's `BatchProcessor` for batched model inference. + +```python +from openmed.multimodal import redact_sms_csv + +result = redact_sms_csv( + "helpdesk.csv", + "helpdesk.redacted.csv", + batch_size=512, +) +assert result.summary.row_count > 0 +``` + +For database cursors or other record streams: + +```python +from openmed.multimodal import iter_redacted_sms_records + +for redacted_record in iter_redacted_sms_records(database_cursor, batch_size=512): + write_record(redacted_record) +``` + +The outer record batch may be larger than 100 rows; the existing batch utility +automatically uses its validated inference chunk size inside that bound. + +## Timestamp handling and verification + +ISO-like values in `sent_on`, `received_on`, `delivered_on`, `created_on`, +`modified_on`, and `timestamp` are coarsened to `YYYY-MM-DD`. Invalid or custom +timestamp strings are retained rather than guessed. + +Before an export crosses a trust boundary, verify known synthetic or +run-scoped source identifiers with `assert_redacted`: + +```python +from openmed.interop import assert_redacted + +assert_redacted(redacted_text, replacement_to_original_mapping) +``` + +Use only synthetic fixtures in tests and benchmarks. Do not log raw input rows, +model entities, or failed records; the adapter's error and summary paths are +designed to report counts and error types without message content. + +## Boundaries + +This adapter does not call a live RapidPro API, serve webhooks, classify message +intent, or process WhatsApp media. It handles exported text payloads only. JSON +documents are materialized to preserve arbitrary wrapper schemas; for exports +with millions of rows, prefer streaming CSV or `iter_redacted_sms_records`. diff --git a/examples/sms_deid_helpdesk_logs.py b/examples/sms_deid_helpdesk_logs.py new file mode 100644 index 000000000..a53934b05 --- /dev/null +++ b/examples/sms_deid_helpdesk_logs.py @@ -0,0 +1,44 @@ +#!/usr/bin/env python3 +"""De-identify a RapidPro-style JSON or generic CSV message export.""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path + +from openmed.multimodal import redact_sms_export + +HASH_KEY_ENV = "OPENMED_SMS_CONTACT_HASH_KEY" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Redact an SMS helpdesk export with the short_text preset." + ) + parser.add_argument("input", type=Path, help="Source .json or .csv export") + parser.add_argument("output", type=Path, help="Destination redacted export") + parser.add_argument( + "--batch-size", + type=int, + default=512, + help="Maximum source records retained per redaction batch", + ) + return parser + + +def main() -> int: + args = build_parser().parse_args() + result = redact_sms_export( + args.input, + args.output, + batch_size=args.batch_size, + contact_hash_key=os.environ.get(HASH_KEY_ENV), + ) + print(json.dumps(result.summary.to_dict(), sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/mkdocs.yml b/mkdocs.yml index fd96dfaae..2e7b77592 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -61,6 +61,7 @@ nav: - GPTQ Export: export-gptq.md - PII & De-identification: - De-identification API: api/deidentification.md + - SMS Short-Text De-identification: sms-short-text-deid.md - PII Anonymization: anonymization.md - Per-Language De-identification: languages.md - Smart Entity Merging: pii-smart-merging.md diff --git a/openmed/core/pii.py b/openmed/core/pii.py index c8dd72e7c..9e45be347 100644 --- a/openmed/core/pii.py +++ b/openmed/core/pii.py @@ -606,15 +606,16 @@ def _prepare_pii_text( lang: str, normalize_accents: Optional[bool], config: Optional[OpenMedConfig] = None, + preserve_whitespace: bool = False, ) -> _PreparedPIIText: - """Return original stripped text, inference text, and normalization flag.""" + """Return source text, inference text, and normalization metadata.""" do_normalize = ( normalize_accents if normalize_accents is not None else (lang in _ACCENT_NORMALIZE_LANGS) ) - original_text = text.strip() + original_text = text if preserve_whitespace else text.strip() width_convention = config.cjk_width_convention if config is not None else "cjk" detection_normalization = normalize_for_pii_detection( original_text, @@ -798,6 +799,7 @@ def _extract_pii_batch( custom_recognizer: Any = None, abdm: Optional[bool] = None, *, + preserve_whitespace: bool = False, locale: Optional[str] = None, loader: Optional["ModelLoader"] = None, privacy_filter_pipeline: Optional[Any] = None, @@ -811,6 +813,7 @@ def _extract_pii_batch( lang=lang, normalize_accents=normalize_accents, config=config, + preserve_whitespace=preserve_whitespace, ) for text in texts ] @@ -918,6 +921,7 @@ def extract_pii( max_cache_entries: int = 128, normalize_accents: Optional[bool] = None, *, + preserve_whitespace: bool = False, locale: Optional[str] = None, loader: Optional["ModelLoader"] = None, batch_size: Optional[int] = None, @@ -951,6 +955,8 @@ def extract_pii( names. Entity spans in the result reference the *original* (accented) text. ``None`` (default) auto-enables for languages in ``_ACCENT_NORMALIZE_LANGS`` (currently Spanish). + preserve_whitespace: Preserve leading and trailing source whitespace + so returned entity offsets refer to the exact input string. loader: Optional shared model loader to reuse warmed pipelines. batch_size: Optional backend inference batch size. num_workers: Optional backend inference worker count. @@ -1015,6 +1021,7 @@ def extract_pii( use_smart_merging=use_smart_merging, lang=lang, normalize_accents=normalize_accents, + preserve_whitespace=preserve_whitespace, locale=locale, loader=loader, custom_recognizer=custom_recognizer, @@ -1826,6 +1833,7 @@ def _deidentify_batch( lang: str = "en", normalize_accents: Optional[bool] = None, use_safety_sweep: bool = True, + preserve_whitespace: bool = False, custom_recognizer: Any = None, abdm: Optional[bool] = None, policy: Optional[str] = None, @@ -1847,16 +1855,17 @@ def _deidentify_batch( date_shift_max_days=date_shift_max_days, date_shift_secret=date_shift_secret, ) - stripped_texts = [text.strip() for text in texts] + source_texts = [text if preserve_whitespace else text.strip() for text in texts] recognizer = coerce_custom_recognizer(custom_recognizer) pii_results = _extract_pii_batch( - stripped_texts, + source_texts, model_name=model_name, confidence_threshold=confidence_threshold, config=config, use_smart_merging=use_smart_merging, lang=lang, normalize_accents=normalize_accents, + preserve_whitespace=preserve_whitespace, locale=locale, custom_recognizer=recognizer, abdm=abdm_mode_enabled( @@ -1872,14 +1881,14 @@ def _deidentify_batch( if use_safety_sweep: swept_results = [] - for stripped_text, pii_result in zip(stripped_texts, pii_results): + for source_text, pii_result in zip(source_texts, pii_results): pii_result, _ = _apply_safety_sweep_to_result( - stripped_text, + source_text, pii_result, lang=lang, locale=locale, ) - _suppress_custom_allowed_entities(stripped_text, pii_result, recognizer) + _suppress_custom_allowed_entities(source_text, pii_result, recognizer) swept_results.append(pii_result) pii_results = swept_results @@ -1906,7 +1915,7 @@ def _deidentify_batch( surrogate_vault=surrogate_vault, policy=policy or "hipaa_safe_harbor", ) - for text, pii_result in zip(stripped_texts, pii_results) + for text, pii_result in zip(source_texts, pii_results) ] diff --git a/openmed/core/pipeline.py b/openmed/core/pipeline.py index dfa4b099e..2bfa8836f 100644 --- a/openmed/core/pipeline.py +++ b/openmed/core/pipeline.py @@ -243,6 +243,7 @@ def __init__( lang: str = "en", normalize_accents: Optional[bool] = None, use_safety_sweep: bool = True, + preserve_whitespace: bool = False, loader: Any = None, privacy_filter_pipeline: Any = None, model_detector: ModelDetector | None = None, @@ -292,6 +293,7 @@ def __init__( self.lang = lang self.normalize_accents = normalize_accents self.use_safety_sweep = use_safety_sweep + self.preserve_whitespace = bool(preserve_whitespace) self.loader = loader self.privacy_filter_pipeline = privacy_filter_pipeline self.model_detector = model_detector @@ -354,7 +356,7 @@ def run( stage_name: 0.0 for stage_name in STAGE_NAMES } cascade_duration_ms: float | None = None - original_text = text.strip() + original_text = text if self.preserve_whitespace else text.strip() with _stage_timer(stage_durations_ms, STAGE_NAMES[0]): normalized = self.stage1_normalize(original_text) with _stage_timer(stage_durations_ms, STAGE_NAMES[1]): @@ -960,6 +962,7 @@ def stage5_fast_pii_model(self, text: str, route: LanguageRoute) -> Any: self.use_smart_merging, lang=route.lang, normalize_accents=self.normalize_accents, + preserve_whitespace=self.preserve_whitespace, loader=self.loader, batch_size=getattr(self.config, "batch_size", None), num_workers=getattr(self.config, "num_workers", None), diff --git a/openmed/multimodal/__init__.py b/openmed/multimodal/__init__.py index 41c2c3cd2..3870dc827 100644 --- a/openmed/multimodal/__init__.py +++ b/openmed/multimodal/__init__.py @@ -95,6 +95,21 @@ register_ocr_engine, run_doctr_ocr, ) +from .sms_messages import ( + DEFAULT_SMS_MODEL, + SHORT_TEXT, + SHORT_TEXT_PRESET, + RedactedSMSExport, + ShortTextPreset, + SMSRedactionSummary, + coarsen_timestamp, + deidentify_short_text, + iter_redacted_sms_records, + redact_sms_csv, + redact_sms_export, + redact_sms_json, + resolve_short_text_preset, +) from .tabular_csv import ( ColumnDecision, RedactedTable, @@ -170,6 +185,19 @@ "register_ocr_engine", "available_ocr_engines", "run_doctr_ocr", + "DEFAULT_SMS_MODEL", + "SHORT_TEXT", + "SHORT_TEXT_PRESET", + "RedactedSMSExport", + "SMSRedactionSummary", + "ShortTextPreset", + "coarsen_timestamp", + "deidentify_short_text", + "iter_redacted_sms_records", + "redact_sms_csv", + "redact_sms_export", + "redact_sms_json", + "resolve_short_text_preset", "ColumnDecision", "TableView", "RedactedTable", diff --git a/openmed/multimodal/sms_messages.py b/openmed/multimodal/sms_messages.py new file mode 100644 index 000000000..df83d6919 --- /dev/null +++ b/openmed/multimodal/sms_messages.py @@ -0,0 +1,865 @@ +"""Short-message de-identification and SMS export adapters. + +The ``short_text`` preset is deliberately deterministic around the identifiers +that dominate SMS-scale clinical data. It augments the normal OpenMed PII +pipeline with African MSISDN, shortcode, national-ID, and maternal-health +honorific rules while allowing common SMS/clinical abbreviations to remain +ordinary text. Export helpers preserve source schemas, pseudonymize contact +addresses, coarsen timestamps to calendar dates, and process message text in +bounded batches. +""" + +from __future__ import annotations + +import copy +import csv +import hashlib +import hmac +import io +import json +import os +import secrets +from collections.abc import Callable, Iterable, Iterator, Mapping, MutableMapping +from dataclasses import dataclass +from datetime import date, datetime +from pathlib import Path +from typing import Any, TextIO + +SHORT_TEXT = "short_text" +DEFAULT_SMS_MODEL = "OpenMed/OpenMed-PII-SuperClinical-Small-44M-v1" + +_NAME_TOKEN = r"(?:[A-Z][^\W\d_]{1,}(?:[-'’][A-Z]?[^\W\d_]*)?|[A-Z]{2,})" +_HONORIFIC_NAME_PATTERN = ( + r"(?i:\b(?:mama|maama|mami|sisi|dada|mme|mrs?|ms|dr|ndugu)\.?\s+)" + + _NAME_TOKEN + + r"(?:\s+" + + _NAME_TOKEN + + r")?" +) + +_SHORT_TEXT_DENY_PATTERNS: tuple[dict[str, Any], ...] = ( + { + "pattern": ( + r"(? Any: + """Build a fresh deterministic recognizer for this preset.""" + + from openmed.core.custom_recognizer import CustomRecognizer + + return CustomRecognizer( + deny_patterns=_SHORT_TEXT_DENY_PATTERNS, + allow_patterns=_SHORT_TEXT_ALLOW_PATTERNS, + case_sensitive=True, + ) + + +SHORT_TEXT_PRESET = ShortTextPreset() + + +@dataclass(frozen=True) +class SMSRedactionSummary: + """PHI-free counts collected while redacting one SMS export.""" + + row_count: int = 0 + message_count: int = 0 + redacted_message_count: int = 0 + pseudonymized_contact_count: int = 0 + coarsened_timestamp_count: int = 0 + batch_count: int = 0 + + def to_dict(self) -> dict[str, int]: + """Return a JSON-serializable, raw-value-free summary.""" + + return { + "row_count": self.row_count, + "message_count": self.message_count, + "redacted_message_count": self.redacted_message_count, + "pseudonymized_contact_count": self.pseudonymized_contact_count, + "coarsened_timestamp_count": self.coarsened_timestamp_count, + "batch_count": self.batch_count, + } + + +@dataclass(frozen=True) +class RedactedSMSExport: + """Materialized text or streamed path plus a PHI-free summary.""" + + format: str + summary: SMSRedactionSummary + text: str | None = None + output_path: Path | None = None + + +@dataclass +class _SummaryAccumulator: + row_count: int = 0 + message_count: int = 0 + redacted_message_count: int = 0 + coarsened_timestamp_count: int = 0 + batch_count: int = 0 + + def freeze(self, *, pseudonymized_contact_count: int) -> SMSRedactionSummary: + return SMSRedactionSummary( + row_count=self.row_count, + message_count=self.message_count, + redacted_message_count=self.redacted_message_count, + pseudonymized_contact_count=pseudonymized_contact_count, + coarsened_timestamp_count=self.coarsened_timestamp_count, + batch_count=self.batch_count, + ) + + +@dataclass +class _TextSlot: + container: MutableMapping[str, Any] | list[Any] + key: str | int + text: str + + +class _ContactHasher: + def __init__(self, key: str | bytes | None = None) -> None: + if key is None: + self._key = secrets.token_bytes(32) + elif isinstance(key, str): + self._key = key.encode("utf-8") + else: + self._key = bytes(key) + if not self._key: + raise ValueError("contact_hash_key must not be empty") + self._tokens: dict[str, str] = {} + + def pseudonymize(self, value: str) -> str: + if not value: + return value + token = self._tokens.get(value) + if token is None: + digest = hmac.new( + self._key, + value.encode("utf-8"), + hashlib.sha256, + ).hexdigest()[:24] + token = f"contact_sha256:{digest}" + self._tokens[value] = token + return token + + @property + def count(self) -> int: + return len(self._tokens) + + +def resolve_short_text_preset( + preset: str | ShortTextPreset = SHORT_TEXT, +) -> ShortTextPreset: + """Resolve the supported short-message preset. + + Args: + preset: The ``"short_text"`` name or a customized preset instance. + + Returns: + The resolved preset. + + Raises: + ValueError: If a string names an unsupported preset. + """ + + if isinstance(preset, ShortTextPreset): + return preset + if preset != SHORT_TEXT: + raise ValueError(f"unsupported SMS pipeline preset: {preset!r}") + return SHORT_TEXT_PRESET + + +def deidentify_short_text( + text: str, + *, + preset: str | ShortTextPreset = SHORT_TEXT, + method: str = "mask", + model_name: str = DEFAULT_SMS_MODEL, + lang: str = "en", + policy: Any = None, + keep_mapping: bool = False, + loader: Any = None, + model_detector: Callable[..., Any] | None = None, +) -> Any: + """De-identify one SMS-scale string without trimming or reflowing it. + + The input is never truncated. Whitespace normalization may be used for + detection internally, but detected offsets are mapped back to the original + string before replacements are applied. + + Args: + text: One message payload. + preset: The ``short_text`` preset or a customized preset instance. + method: OpenMed de-identification method. + model_name: PII model registry key or repository identifier. + lang: Language hint for model and locale-specific recognizers. + policy: Optional OpenMed de-identification policy. + keep_mapping: Whether to retain a reversible placeholder mapping. + loader: Optional warmed model loader. + model_detector: Optional detector seam for offline callers and tests. + + Returns: + An OpenMed de-identification result whose offsets refer to ``text``. + """ + + if not isinstance(text, str): + raise TypeError("text must be a string") + options = resolve_short_text_preset(preset) + + from openmed.core.pipeline import Pipeline + + pipeline = Pipeline( + model_name=model_name, + confidence_threshold=options.confidence_threshold, + use_smart_merging=options.use_smart_merging, + lang=lang, + use_safety_sweep=options.use_safety_sweep, + preserve_whitespace=options.preserve_whitespace, + loader=loader, + model_detector=model_detector, + policy=policy, + custom_recognizer=options.build_recognizer(), + ) + result = pipeline.run( + text, + method=method, + keep_mapping=keep_mapping, + ).deidentification_result + metadata = dict(getattr(result, "metadata", None) or {}) + metadata["pipeline_preset"] = options.name + metadata["source_length"] = len(text) + result.metadata = metadata + return result + + +def iter_redacted_sms_records( + records: Iterable[Mapping[str, Any]], + *, + preset: str | ShortTextPreset = SHORT_TEXT, + method: str = "mask", + model_name: str = DEFAULT_SMS_MODEL, + lang: str = "en", + policy: Any = None, + batch_size: int | None = None, + contact_hash_key: str | bytes | None = None, + text_redactor: TextRedactor | None = None, + loader: Any = None, +) -> "_SMSRecordIterator": + """Return a bounded-memory iterator over redacted message records. + + The default path reuses :class:`openmed.processing.batch.BatchProcessor` + and retains at most ``batch_size`` source records at a time. A + ``text_redactor`` can be injected for offline tests or an already-warmed + application-specific detector. + + Args: + records: Iterable of message record mappings. + preset: The ``short_text`` preset or a customized preset instance. + method: OpenMed de-identification method. + model_name: PII model registry key or repository identifier. + lang: Language hint for model and locale-specific recognizers. + policy: Optional OpenMed de-identification policy. + batch_size: Maximum source records retained per outer batch. + contact_hash_key: Optional HMAC key for stable cross-run pseudonyms. + text_redactor: Optional replacement for model-backed text redaction. + loader: Optional warmed model loader. + + Returns: + An iterator exposing a PHI-free ``summary`` property. + """ + + options = resolve_short_text_preset(preset) + resolved_batch_size = options.batch_size if batch_size is None else batch_size + if not isinstance(resolved_batch_size, int) or resolved_batch_size <= 0: + raise ValueError("batch_size must be positive") + accumulator = _SummaryAccumulator() + hasher = _ContactHasher(contact_hash_key) + processor = None + if text_redactor is None: + processor = _build_batch_processor( + options, + method=method, + model_name=model_name, + lang=lang, + policy=policy, + batch_size=resolved_batch_size, + loader=loader, + ) + iterator = _iter_redacted_record_batches( + records, + batch_size=resolved_batch_size, + processor=processor, + text_redactor=text_redactor, + hasher=hasher, + accumulator=accumulator, + ) + return _SMSRecordIterator(iterator, accumulator, hasher) + + +class _SMSRecordIterator: + def __init__( + self, + iterator: Iterator[dict[str, Any]], + accumulator: _SummaryAccumulator, + hasher: _ContactHasher, + ) -> None: + self._iterator = iterator + self._accumulator = accumulator + self._hasher = hasher + + def __iter__(self) -> "_SMSRecordIterator": + return self + + def __next__(self) -> dict[str, Any]: + return next(self._iterator) + + @property + def summary(self) -> SMSRedactionSummary: + """Return counts accumulated through the records consumed so far.""" + + return self._accumulator.freeze(pseudonymized_contact_count=self._hasher.count) + + +def redact_sms_json( + source: str | os.PathLike[str] | TextIO | Mapping[str, Any] | list[Any], + output: str | os.PathLike[str] | TextIO | None = None, + **kwargs: Any, +) -> RedactedSMSExport: + """Redact a RapidPro-shaped JSON message or flow-results export. + + Args: + source: JSON mapping/list, text, path, or readable text stream. + output: Optional destination path or writable text stream. + **kwargs: Options forwarded to :func:`iter_redacted_sms_records`. + + Returns: + Materialized output text when ``output`` is omitted, otherwise the + destination path when available, plus a PHI-free summary. + """ + + payload = copy.deepcopy(_load_json_payload(source)) + records = _locate_records(payload) + single_root_record = ( + isinstance(payload, MutableMapping) + and len(records) == 1 + and records[0] is payload + ) + iterator = iter_redacted_sms_records(records, **kwargs) + redacted_records = list(iterator) + if single_root_record: + payload.clear() + payload.update(redacted_records[0]) + else: + records[:] = redacted_records + text = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n" + output_path = _write_optional_text_output(output, text) + return RedactedSMSExport( + format="json", + summary=iterator.summary, + text=text if output is None else None, + output_path=output_path, + ) + + +def redact_sms_csv( + source: str | os.PathLike[str] | TextIO, + output: str | os.PathLike[str] | TextIO | None = None, + **kwargs: Any, +) -> RedactedSMSExport: + """Stream a generic SMS CSV while preserving columns and row order. + + Args: + source: CSV text, path, or readable text stream. + output: Optional destination path or writable text stream. + **kwargs: Options forwarded to :func:`iter_redacted_sms_records`. + + Returns: + Materialized output text when ``output`` is omitted, otherwise the + destination path when available, plus a PHI-free summary. + """ + + source_handle, close_source = _open_text_source(source) + output_handle, close_output, output_path = _open_text_output(output) + try: + reader = csv.DictReader(source_handle) + if not reader.fieldnames: + raise ValueError("SMS CSV must include a header row") + if "text" not in reader.fieldnames: + raise ValueError("SMS CSV must include a 'text' column") + if not ({"urn", "contact"} & set(reader.fieldnames)): + raise ValueError("SMS CSV must include an 'urn' or 'contact' column") + + writer = csv.DictWriter( + output_handle, + fieldnames=reader.fieldnames, + extrasaction="raise", + lineterminator="\n", + ) + writer.writeheader() + iterator = iter_redacted_sms_records(reader, **kwargs) + for row in iterator: + writer.writerow(row) + summary = iterator.summary + materialized = ( + output_handle.getvalue() if isinstance(output_handle, io.StringIO) else None + ) + finally: + if close_source: + source_handle.close() + if close_output: + output_handle.close() + + return RedactedSMSExport( + format="csv", + summary=summary, + text=materialized, + output_path=output_path, + ) + + +def redact_sms_export( + source: str | os.PathLike[str] | TextIO | Mapping[str, Any] | list[Any], + output: str | os.PathLike[str] | TextIO | None = None, + *, + format: str | None = None, + **kwargs: Any, +) -> RedactedSMSExport: + """Dispatch a JSON or CSV SMS export through the ``short_text`` preset. + + Args: + source: Export payload, text, path, or readable stream. + output: Optional destination path or writable text stream. + format: Explicit ``"json"`` or ``"csv"`` override. + **kwargs: Options forwarded to the selected export adapter. + + Returns: + The redacted export result and PHI-free summary. + """ + + resolved = (format or _infer_export_format(source)).lower() + if resolved == "json": + return redact_sms_json(source, output, **kwargs) + if resolved == "csv": + if isinstance(source, (Mapping, list)): + raise TypeError("CSV source must be text, a path, or a text stream") + return redact_sms_csv(source, output, **kwargs) + raise ValueError("format must be 'json' or 'csv'") + + +def coarsen_timestamp(value: Any) -> Any: + """Coarsen an ISO-like timestamp to ``YYYY-MM-DD`` when parseable. + + Args: + value: Date, datetime, ISO-like string, or value to leave untouched. + + Returns: + A calendar-date string for parseable timestamps, otherwise ``value``. + """ + + if isinstance(value, datetime): + return value.date().isoformat() + if isinstance(value, date): + return value.isoformat() + if not isinstance(value, str) or not value: + return value + + candidate = value.strip() + try: + parsed = datetime.fromisoformat(candidate.replace("Z", "+00:00")) + except ValueError: + try: + return date.fromisoformat(candidate).isoformat() + except ValueError: + return value + return parsed.date().isoformat() + + +def _build_batch_processor( + preset: ShortTextPreset, + *, + method: str, + model_name: str, + lang: str, + policy: Any, + batch_size: int, + loader: Any, +) -> Any: + from openmed.processing.batch import BatchProcessor + + return BatchProcessor( + model_name=model_name, + operation="deidentify", + batch_size=min(batch_size, 100), + confidence_threshold=preset.confidence_threshold, + continue_on_error=False, + loader=loader, + method=method, + lang=lang, + policy=policy, + use_smart_merging=preset.use_smart_merging, + use_safety_sweep=preset.use_safety_sweep, + preserve_whitespace=preset.preserve_whitespace, + custom_recognizer=preset.build_recognizer(), + ) + + +def _iter_redacted_record_batches( + records: Iterable[Mapping[str, Any]], + *, + batch_size: int, + processor: Any, + text_redactor: TextRedactor | None, + hasher: _ContactHasher, + accumulator: _SummaryAccumulator, +) -> Iterator[dict[str, Any]]: + batch: list[Mapping[str, Any]] = [] + for record in records: + if not isinstance(record, Mapping): + raise TypeError("SMS export records must be mappings") + batch.append(record) + if len(batch) >= batch_size: + yield from _redact_record_batch( + batch, + processor=processor, + text_redactor=text_redactor, + hasher=hasher, + accumulator=accumulator, + ) + batch = [] + if batch: + yield from _redact_record_batch( + batch, + processor=processor, + text_redactor=text_redactor, + hasher=hasher, + accumulator=accumulator, + ) + + +def _redact_record_batch( + records: list[Mapping[str, Any]], + *, + processor: Any, + text_redactor: TextRedactor | None, + hasher: _ContactHasher, + accumulator: _SummaryAccumulator, +) -> list[dict[str, Any]]: + copied = [copy.deepcopy(dict(record)) for record in records] + slots: list[_TextSlot] = [] + for record in copied: + _collect_text_slots(record, slots) + _pseudonymize_contacts(record, hasher) + accumulator.coarsened_timestamp_count += _coarsen_timestamps(record) + + nonempty_slots = [slot for slot in slots if slot.text] + if nonempty_slots: + if text_redactor is not None: + redacted_texts = [ + _coerce_redacted_text(text_redactor(slot.text)) + for slot in nonempty_slots + ] + else: + batch_result = processor.process_texts( + [slot.text for slot in nonempty_slots], + ids=[f"sms_{index}" for index in range(len(nonempty_slots))], + ) + if batch_result.failed_items: + raise RuntimeError("short_text batch de-identification failed") + redacted_texts = [ + _coerce_redacted_text(item.result) for item in batch_result.items + ] + for slot, redacted in zip(nonempty_slots, redacted_texts): + slot.container[slot.key] = redacted + if redacted != slot.text: + accumulator.redacted_message_count += 1 + + accumulator.row_count += len(copied) + accumulator.message_count += len(slots) + accumulator.batch_count += 1 + return copied + + +def _collect_text_slots(value: Any, slots: list[_TextSlot]) -> None: + if isinstance(value, MutableMapping): + for key, item in list(value.items()): + if str(key).lower() == "text" and isinstance(item, str): + slots.append(_TextSlot(value, key, item)) + elif isinstance(item, (MutableMapping, list)): + _collect_text_slots(item, slots) + elif isinstance(value, list): + for index, item in enumerate(value): + if isinstance(item, (MutableMapping, list)): + _collect_text_slots(item, slots) + + +def _pseudonymize_contacts(value: Any, hasher: _ContactHasher) -> None: + if isinstance(value, MutableMapping): + for key, item in list(value.items()): + normalized_key = str(key).lower() + if normalized_key in _CONTACT_VALUE_KEYS and isinstance(item, str): + value[key] = hasher.pseudonymize(item) + elif normalized_key in _CONTACT_LIST_KEYS and isinstance(item, list): + value[key] = [ + hasher.pseudonymize(entry) if isinstance(entry, str) else entry + for entry in item + ] + elif normalized_key in _CONTACT_CONTAINER_KEYS: + if isinstance(item, str): + value[key] = hasher.pseudonymize(item) + elif isinstance(item, MutableMapping): + _pseudonymize_contact_mapping(item, hasher) + elif isinstance(item, list): + _pseudonymize_contacts(item, hasher) + elif isinstance(item, (MutableMapping, list)): + _pseudonymize_contacts(item, hasher) + elif isinstance(value, list): + for item in value: + if isinstance(item, (MutableMapping, list)): + _pseudonymize_contacts(item, hasher) + + +def _pseudonymize_contact_mapping( + contact: MutableMapping[str, Any], + hasher: _ContactHasher, +) -> None: + for key, item in list(contact.items()): + normalized_key = str(key).lower() + if normalized_key not in _CONTACT_MAPPING_KEYS: + continue + if isinstance(item, str): + contact[key] = hasher.pseudonymize(item) + elif isinstance(item, list): + contact[key] = [ + hasher.pseudonymize(entry) if isinstance(entry, str) else entry + for entry in item + ] + + +def _coarsen_timestamps(value: Any) -> int: + changed = 0 + if isinstance(value, MutableMapping): + for key, item in list(value.items()): + if str(key).lower() in _TIMESTAMP_KEYS: + coarsened = coarsen_timestamp(item) + value[key] = coarsened + changed += int(coarsened != item) + elif isinstance(item, (MutableMapping, list)): + changed += _coarsen_timestamps(item) + elif isinstance(value, list): + for item in value: + if isinstance(item, (MutableMapping, list)): + changed += _coarsen_timestamps(item) + return changed + + +def _coerce_redacted_text(result: Any) -> str: + if hasattr(result, "deidentified_text"): + return str(result.deidentified_text) + if isinstance(result, str): + return result + raise TypeError("text redactor must return a string or de-identification result") + + +def _load_json_payload( + source: str | os.PathLike[str] | TextIO | Mapping[str, Any] | list[Any], +) -> Mapping[str, Any] | list[Any]: + if isinstance(source, Mapping): + return source + if isinstance(source, list): + return source + if hasattr(source, "read"): + payload = json.load(source) + elif isinstance(source, os.PathLike): + with Path(source).open("r", encoding="utf-8") as handle: + payload = json.load(handle) + elif isinstance(source, str): + path = Path(source) + if "\n" not in source and "\r" not in source and path.exists(): + with path.open("r", encoding="utf-8") as handle: + payload = json.load(handle) + else: + payload = json.loads(source) + else: + raise TypeError("JSON source must be a mapping, list, path, or text stream") + if not isinstance(payload, (Mapping, list)): + raise ValueError("SMS JSON root must be an object or array") + return payload + + +def _locate_records(payload: Mapping[str, Any] | list[Any]) -> list[Any]: + if isinstance(payload, list): + return payload + for key in _RECORD_COLLECTION_KEYS: + candidate = payload.get(key) + if isinstance(candidate, list): + return candidate + if isinstance(candidate, Mapping): + try: + return _locate_records(candidate) + except ValueError: + pass + if isinstance(payload, MutableMapping): + return [payload] + raise ValueError("SMS JSON object does not contain a supported record collection") + + +def _open_text_source( + source: str | os.PathLike[str] | TextIO, +) -> tuple[TextIO, bool]: + if hasattr(source, "read"): + return source, False + if isinstance(source, os.PathLike): + return Path(source).open("r", encoding="utf-8", newline=""), True + path = Path(source) + if "\n" not in source and "\r" not in source and path.exists(): + return path.open("r", encoding="utf-8", newline=""), True + return io.StringIO(source, newline=""), True + + +def _open_text_output( + output: str | os.PathLike[str] | TextIO | None, +) -> tuple[TextIO, bool, Path | None]: + if output is None: + return io.StringIO(newline=""), False, None + if hasattr(output, "write"): + return output, False, None + path = Path(output) + return path.open("w", encoding="utf-8", newline=""), True, path + + +def _write_optional_text_output( + output: str | os.PathLike[str] | TextIO | None, + text: str, +) -> Path | None: + if output is None: + return None + if hasattr(output, "write"): + output.write(text) + return None + path = Path(output) + path.write_text(text, encoding="utf-8") + return path + + +def _infer_export_format( + source: str | os.PathLike[str] | TextIO | Mapping[str, Any] | list[Any], +) -> str: + if isinstance(source, (Mapping, list)): + return "json" + if isinstance(source, os.PathLike): + suffix = Path(source).suffix.lower() + if suffix in {".json", ".csv"}: + return suffix[1:] + if isinstance(source, str): + path = Path(source) + if "\n" not in source and "\r" not in source and path.exists(): + suffix = path.suffix.lower() + if suffix in {".json", ".csv"}: + return suffix[1:] + return "json" if source.lstrip().startswith(("{", "[")) else "csv" + raise ValueError("format is required for streams without a recognizable suffix") + + +__all__ = [ + "DEFAULT_SMS_MODEL", + "SHORT_TEXT", + "SHORT_TEXT_PRESET", + "RedactedSMSExport", + "SMSRedactionSummary", + "ShortTextPreset", + "coarsen_timestamp", + "deidentify_short_text", + "iter_redacted_sms_records", + "redact_sms_csv", + "redact_sms_export", + "redact_sms_json", + "resolve_short_text_preset", +] diff --git a/tests/unit/multimodal/fixtures/sms/generic_messages.csv b/tests/unit/multimodal/fixtures/sms/generic_messages.csv new file mode 100644 index 000000000..76c41d04b --- /dev/null +++ b/tests/unit/multimodal/fixtures/sms/generic_messages.csv @@ -0,0 +1,3 @@ +urn,contact,direction,text,sent_on,flow_uuid +tel:+256772345678,tel:+256772345678,in,SAWUBONA SISI Nomsa Dlamini call 0712345678 ID: UG123456,2026-07-17T10:20:30Z,99999999-8888-7777-6666-555555555555 +tel:+256772345678,tel:+256772345678,out,PLS RPLY 12345,2026-07-17T10:21:31Z,99999999-8888-7777-6666-555555555555 diff --git a/tests/unit/multimodal/fixtures/sms/rapidpro_messages.json b/tests/unit/multimodal/fixtures/sms/rapidpro_messages.json new file mode 100644 index 000000000..d9ff52995 --- /dev/null +++ b/tests/unit/multimodal/fixtures/sms/rapidpro_messages.json @@ -0,0 +1,33 @@ +{ + "export": "messages", + "flow": { + "uuid": "11111111-2222-3333-4444-555555555555", + "name": "Synthetic maternal helpdesk" + }, + "messages": [ + { + "uuid": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeee1", + "urn": "tel:+254712345678", + "contact": { + "uuid": "contact-0001", + "name": "Amina Njeri", + "urn": "tel:+254712345678" + }, + "direction": "in", + "text": "HABARI MAMA Amina Njeri call +254712345678 ID: KE123456", + "sent_on": "2026-07-17T09:10:11Z" + }, + { + "uuid": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeee2", + "urn": "tel:+254712345678", + "contact": { + "uuid": "contact-0001", + "name": "Amina Njeri", + "urn": "tel:+254712345678" + }, + "direction": "out", + "text": "ANC appt confirmed", + "sent_on": "2026-07-17T09:11:12Z" + } + ] +} diff --git a/tests/unit/multimodal/test_sms_messages.py b/tests/unit/multimodal/test_sms_messages.py new file mode 100644 index 000000000..528ad59e9 --- /dev/null +++ b/tests/unit/multimodal/test_sms_messages.py @@ -0,0 +1,251 @@ +"""Tests for SMS-scale de-identification and export handling.""" + +from __future__ import annotations + +import csv +import io +import json +import math +from pathlib import Path + +import openmed +from openmed.core.pipeline import Pipeline +from openmed.interop import assert_redacted +from openmed.multimodal.sms_messages import ( + SHORT_TEXT, + deidentify_short_text, + iter_redacted_sms_records, + redact_sms_csv, + redact_sms_json, +) +from openmed.processing import BatchProcessor +from openmed.processing.outputs import PredictionResult + +FIXTURES = Path(__file__).parent / "fixtures" / "sms" + + +def _empty_detector(text: str, **_: object) -> PredictionResult: + return PredictionResult( + text=text, + entities=[], + model_name="synthetic-empty-detector", + timestamp="2026-01-01T00:00:00Z", + ) + + +def _expected_replacement_only_text(result: object) -> str: + original = result.original_text + cursor = 0 + parts: list[str] = [] + for entity in sorted(result.pii_entities, key=lambda item: item.start): + parts.append(original[cursor : entity.start]) + parts.append(entity.redacted_text or "") + cursor = entity.end + parts.append(original[cursor:]) + return "".join(parts) + + +def _synthetic_corpus( + size: int = 600, +) -> tuple[list[dict[str, str]], list[tuple[str, ...]]]: + names = ( + "Amina Njeri", + "Nomsa Dlamini", + "Achieng Otieno", + "Nabirye Namusoke", + "Thandi Mokoena", + ) + languages = ( + ("HELLO", "MAMA"), + ("HABARI", "MAMA"), + ("SAWUBONA", "SISI"), + ) + rows: list[dict[str, str]] = [] + seeded: list[tuple[str, ...]] = [] + for index in range(size): + greeting, honorific = languages[index % len(languages)] + name = names[index % len(names)] + country = ("254", "256", "27")[index % 3] + subscriber_length = 9 if country != "27" else 9 + subscriber = str(700000000 + index).zfill(subscriber_length) + phone = f"+{country}{subscriber}" + identifier = f"ID: OM{index:06d}" + message_date = f"2026-07-{(index % 28) + 1:02d}" + text = ( + f"{greeting} {honorific} {name} {identifier} " + f"CALL {phone} ANC {message_date} PLS" + ) + assert len(text) <= 320 + rows.append( + { + "urn": f"tel:{phone}", + "contact": f"tel:{phone}", + "direction": "in" if index % 2 == 0 else "out", + "text": text, + "sent_on": f"{message_date}T10:11:12Z", + } + ) + seeded.append((name, identifier, phone, message_date)) + return rows, seeded + + +def _recall(outputs: list[str], seeded: list[tuple[str, ...]]) -> float: + total = sum(len(values) for values in seeded) + redacted = sum( + value not in output + for output, values in zip(outputs, seeded) + for value in values + ) + return redacted / total + + +def test_short_text_preset_preserves_every_non_replacement_character(): + text = "\tSAWUBONA SISI Nomsa Dlamini\ncall 0712345678 ANC pls\n" + + result = deidentify_short_text(text, model_detector=_empty_detector) + + assert result.original_text == text + assert result.deidentified_text == _expected_replacement_only_text(result) + assert result.deidentified_text.startswith("\t") + assert result.deidentified_text.endswith(" ANC pls\n") + assert "Nomsa Dlamini" not in result.deidentified_text + assert "0712345678" not in result.deidentified_text + assert result.metadata["pipeline_preset"] == SHORT_TEXT + + +def test_default_pipeline_clinical_note_behavior_remains_unchanged(): + note = "Patient is stable with no identifiers." + + result = Pipeline(model_detector=_empty_detector).run(note) + + assert result.original_text == note + assert result.redacted_text == note + + +def test_short_text_corpus_has_higher_recall_and_no_seeded_phi_leaks(monkeypatch): + rows, seeded = _synthetic_corpus() + monkeypatch.setattr(openmed, "analyze_text", _empty_detector) + + iterator = iter_redacted_sms_records( + rows, + batch_size=128, + contact_hash_key="synthetic-test-key", + loader=object(), + ) + short_rows = list(iterator) + short_outputs = [row["text"] for row in short_rows] + + default_processor = BatchProcessor( + operation="deidentify", + batch_size=100, + continue_on_error=False, + loader=object(), + method="mask", + use_safety_sweep=True, + ) + default_result = default_processor.process_texts([row["text"] for row in rows]) + default_outputs = [item.result.deidentified_text for item in default_result.items] + + short_recall = _recall(short_outputs, seeded) + default_recall = _recall(default_outputs, seeded) + assert short_recall == 1.0 + assert short_recall > default_recall + assert iterator.summary.row_count == 600 + assert iterator.summary.batch_count == math.ceil(600 / 128) + + corpus_text = "\n".join(short_outputs) + original_mapping = { + f"seed_{row_index}_{value_index}": value + for row_index, values in enumerate(seeded) + for value_index, value in enumerate(values[:3]) + } + assert assert_redacted(corpus_text, original_mapping) == corpus_text + + +def test_rapidpro_json_round_trip_preserves_schema_and_flow_fields(monkeypatch): + monkeypatch.setattr(openmed, "analyze_text", _empty_detector) + + result = redact_sms_json( + FIXTURES / "rapidpro_messages.json", + batch_size=2, + contact_hash_key="round-trip-key", + loader=object(), + ) + payload = json.loads(result.text or "") + original = json.loads((FIXTURES / "rapidpro_messages.json").read_text()) + + assert set(payload) == set(original) + assert len(payload["messages"]) == len(original["messages"]) + assert payload["flow"] == original["flow"] + assert [row["direction"] for row in payload["messages"]] == ["in", "out"] + assert [row["uuid"] for row in payload["messages"]] == [ + row["uuid"] for row in original["messages"] + ] + assert payload["messages"][0]["urn"] == payload["messages"][1]["urn"] + assert payload["messages"][0]["contact"]["urn"] == payload["messages"][0]["urn"] + assert payload["messages"][0]["contact"]["uuid"] == "contact-0001" + assert payload["messages"][0]["sent_on"] == "2026-07-17" + assert "Amina Njeri" not in result.text + assert "+254712345678" not in result.text + assert result.summary.row_count == 2 + assert result.summary.message_count == 2 + + +def test_flow_result_input_text_is_redacted_without_changing_flow_uuid(monkeypatch): + monkeypatch.setattr(openmed, "analyze_text", _empty_detector) + payload = { + "results": [ + { + "flow": {"uuid": "flow-result-uuid"}, + "contact": {"uuid": "contact-uuid", "urn": "tel:+27821234567"}, + "input": {"text": "MAMA Thandi Mokoena ID: ZA123456"}, + "created_on": "2026-07-17T12:00:00+02:00", + } + ] + } + + result = redact_sms_json( + payload, + contact_hash_key="flow-result-key", + loader=object(), + ) + reparsed = json.loads(result.text or "") + + row = reparsed["results"][0] + assert row["flow"]["uuid"] == "flow-result-uuid" + assert row["contact"]["uuid"] == "contact-uuid" + assert row["created_on"] == "2026-07-17" + assert "Thandi Mokoena" not in row["input"]["text"] + assert "ZA123456" not in row["input"]["text"] + + +def test_generic_csv_round_trip_preserves_columns_rows_and_non_text_fields(monkeypatch): + monkeypatch.setattr(openmed, "analyze_text", _empty_detector) + + result = redact_sms_csv( + FIXTURES / "generic_messages.csv", + batch_size=1, + contact_hash_key="csv-round-trip-key", + loader=object(), + ) + rows = list(csv.DictReader(io.StringIO(result.text or ""))) + + assert len(rows) == 2 + assert list(rows[0]) == [ + "urn", + "contact", + "direction", + "text", + "sent_on", + "flow_uuid", + ] + assert [row["direction"] for row in rows] == ["in", "out"] + assert rows[0]["flow_uuid"] == rows[1]["flow_uuid"] + assert rows[0]["urn"] == rows[1]["urn"] + assert rows[0]["contact"] == rows[1]["contact"] + assert rows[0]["sent_on"] == "2026-07-17" + assert "Nomsa Dlamini" not in rows[0]["text"] + assert "0712345678" not in rows[0]["text"] + assert "12345" not in rows[1]["text"] + assert result.summary.row_count == 2 + assert result.summary.batch_count == 2 From 6a3317e2c6a5b59afe66380f176c93b867e2a4c0 Mon Sep 17 00:00:00 2001 From: Maziyar Panahi Date: Mon, 20 Jul 2026 11:41:47 +0200 Subject: [PATCH 2/2] fix: harden SMS export processing --- docs/sms-short-text-deid.md | 10 ++-- openmed/multimodal/sms_messages.py | 59 ++++++++++++---------- tests/unit/multimodal/test_sms_messages.py | 7 +-- 3 files changed, 43 insertions(+), 33 deletions(-) diff --git a/docs/sms-short-text-deid.md b/docs/sms-short-text-deid.md index 1512020a6..096d1f9cd 100644 --- a/docs/sms-short-text-deid.md +++ b/docs/sms-short-text-deid.md @@ -106,10 +106,12 @@ The expected RapidPro-compatible core columns are: urn,contact,direction,text,sent_on ``` -Extra columns are preserved in their original order. The CSV path reads and -writes incrementally, while `iter_redacted_sms_records` accepts any iterable of -mapping records. Both retain at most one source batch in memory and reuse -OpenMed's `BatchProcessor` for batched model inference. +Extra columns are preserved in their original order. When an output destination +is supplied, the CSV path reads and writes incrementally, while +`iter_redacted_sms_records` accepts any iterable of mapping records. Both keep +constant-size contact-hashing state, retain at most one source batch in memory, +and reuse OpenMed's `BatchProcessor` for batched model inference. Omitting the +CSV output destination materializes the result text for convenience. ```python from openmed.multimodal import redact_sms_csv diff --git a/openmed/multimodal/sms_messages.py b/openmed/multimodal/sms_messages.py index df83d6919..43b279c01 100644 --- a/openmed/multimodal/sms_messages.py +++ b/openmed/multimodal/sms_messages.py @@ -30,11 +30,11 @@ _NAME_TOKEN = r"(?:[A-Z][^\W\d_]{1,}(?:[-'’][A-Z]?[^\W\d_]*)?|[A-Z]{2,})" _HONORIFIC_NAME_PATTERN = ( - r"(?i:\b(?:mama|maama|mami|sisi|dada|mme|mrs?|ms|dr|ndugu)\.?\s+)" + r"(?i:\b(?:mama|maama|mami|sisi|dada|mme|mrs?|ms|dr|ndugu)\.?\s+" + _NAME_TOKEN + r"(?:\s+" + _NAME_TOKEN - + r")?" + + r")?)" ) _SHORT_TEXT_DENY_PATTERNS: tuple[dict[str, Any], ...] = ( @@ -62,7 +62,7 @@ { "pattern": ( r"(?i:\b(?:id|nid|mrn|patient\s*(?:id|no)|clinic\s*no)" - r"\s*[:#-]?\s*)[A-Z0-9][A-Z0-9/.-]{3,}" + r"\s*[:#-]?\s*[A-Z0-9][A-Z0-9/.-]{3,})" ), "label": "ID_NUM", "rule_id": "short_text_contextual_id", @@ -214,25 +214,22 @@ def __init__(self, key: str | bytes | None = None) -> None: self._key = bytes(key) if not self._key: raise ValueError("contact_hash_key must not be empty") - self._tokens: dict[str, str] = {} + self._count = 0 def pseudonymize(self, value: str) -> str: if not value: return value - token = self._tokens.get(value) - if token is None: - digest = hmac.new( - self._key, - value.encode("utf-8"), - hashlib.sha256, - ).hexdigest()[:24] - token = f"contact_sha256:{digest}" - self._tokens[value] = token - return token + digest = hmac.new( + self._key, + value.encode("utf-8"), + hashlib.sha256, + ).hexdigest()[:24] + self._count += 1 + return f"contact_sha256:{digest}" @property def count(self) -> int: - return len(self._tokens) + return self._count def resolve_short_text_preset( @@ -651,7 +648,9 @@ def _redact_record_batch( [slot.text for slot in nonempty_slots], ids=[f"sms_{index}" for index in range(len(nonempty_slots))], ) - if batch_result.failed_items: + if batch_result.failed_items or len(batch_result.items) != len( + nonempty_slots + ): raise RuntimeError("short_text batch de-identification failed") redacted_texts = [ _coerce_redacted_text(item.result) for item in batch_result.items @@ -761,9 +760,10 @@ def _load_json_payload( with Path(source).open("r", encoding="utf-8") as handle: payload = json.load(handle) elif isinstance(source, str): - path = Path(source) - if "\n" not in source and "\r" not in source and path.exists(): - with path.open("r", encoding="utf-8") as handle: + if source.lstrip().startswith(("{", "[")): + payload = json.loads(source) + elif _path_exists(source): + with Path(source).open("r", encoding="utf-8") as handle: payload = json.load(handle) else: payload = json.loads(source) @@ -798,9 +798,8 @@ def _open_text_source( return source, False if isinstance(source, os.PathLike): return Path(source).open("r", encoding="utf-8", newline=""), True - path = Path(source) - if "\n" not in source and "\r" not in source and path.exists(): - return path.open("r", encoding="utf-8", newline=""), True + if _path_exists(source): + return Path(source).open("r", encoding="utf-8", newline=""), True return io.StringIO(source, newline=""), True @@ -839,15 +838,23 @@ def _infer_export_format( if suffix in {".json", ".csv"}: return suffix[1:] if isinstance(source, str): - path = Path(source) - if "\n" not in source and "\r" not in source and path.exists(): - suffix = path.suffix.lower() + if source.lstrip().startswith(("{", "[")): + return "json" + if _path_exists(source): + suffix = Path(source).suffix.lower() if suffix in {".json", ".csv"}: return suffix[1:] - return "json" if source.lstrip().startswith(("{", "[")) else "csv" + return "csv" raise ValueError("format is required for streams without a recognizable suffix") +def _path_exists(value: str | os.PathLike[str]) -> bool: + try: + return Path(value).exists() + except OSError: + return False + + __all__ = [ "DEFAULT_SMS_MODEL", "SHORT_TEXT", diff --git a/tests/unit/multimodal/test_sms_messages.py b/tests/unit/multimodal/test_sms_messages.py index 528ad59e9..41f09d41b 100644 --- a/tests/unit/multimodal/test_sms_messages.py +++ b/tests/unit/multimodal/test_sms_messages.py @@ -100,7 +100,7 @@ def _recall(outputs: list[str], seeded: list[tuple[str, ...]]) -> float: def test_short_text_preset_preserves_every_non_replacement_character(): - text = "\tSAWUBONA SISI Nomsa Dlamini\ncall 0712345678 ANC pls\n" + text = "\tsawubona sisi nomsa dlamini\ncall 0712345678 id: ug123456 ANC pls\n" result = deidentify_short_text(text, model_detector=_empty_detector) @@ -108,8 +108,9 @@ def test_short_text_preset_preserves_every_non_replacement_character(): assert result.deidentified_text == _expected_replacement_only_text(result) assert result.deidentified_text.startswith("\t") assert result.deidentified_text.endswith(" ANC pls\n") - assert "Nomsa Dlamini" not in result.deidentified_text + assert "nomsa dlamini" not in result.deidentified_text assert "0712345678" not in result.deidentified_text + assert "ug123456" not in result.deidentified_text assert result.metadata["pipeline_preset"] == SHORT_TEXT @@ -205,7 +206,7 @@ def test_flow_result_input_text_is_redacted_without_changing_flow_uuid(monkeypat } result = redact_sms_json( - payload, + json.dumps(payload, separators=(",", ":")), contact_hash_key="flow-result-key", loader=object(), )