diff --git a/docs/legacy-indic-encoding.md b/docs/legacy-indic-encoding.md new file mode 100644 index 00000000..76043855 --- /dev/null +++ b/docs/legacy-indic-encoding.md @@ -0,0 +1,85 @@ +# Legacy Indic encoding conversion + +OpenMed can convert Devanagari text stored as ISCII-1991 bytes or as a +caller-described, ASCII-remapped legacy font before script routing and PII +detection. Conversion is deterministic and local; it makes no network calls +and does not log or persist source text. + +## ISCII conversion and offsets + +Pass raw bytes when source-byte offsets matter: + +```python +from openmed.processing import iscii_to_unicode, unicode_to_iscii + +source = bytes.fromhex("cf cc e1 d5 20 d5 cf e8 cc da") +conversion = iscii_to_unicode(source) + +assert conversion.text == "रमेश शर्मा" +assert unicode_to_iscii(conversion.text) == source + +start = conversion.text.index("रमेश शर्मा") +original_start, original_end = conversion.to_original_span( + start, + start + len("रमेश शर्मा"), +) +assert source[original_start:original_end] == source +``` + +The converter handles the ISCII Devanagari attribute sequence, standard +nukta combinations, double danda, and the explicit and soft halant sequences. +Output is placed in logical Unicode order, normalized to NFC, and aligned to +the original byte stream. A Latin clinical note does not meet the conservative +ISCII detection gate and is returned unchanged by `convert_legacy_encoding`. + +## User-supplied legacy-font maps + +OpenMed intentionally bundles no proprietary legacy-font mapping data. Supply +a JSON or YAML file for a font whose mapping you are licensed to use: + +```json +{ + "name": "hospital-archive-font", + "provenance": "licensed by the data owner", + "mapping": { + "0x41": "र", + "B": "ा", + "67": "म" + } +} +``` + +Keys may be byte integers, single-byte characters, decimal byte strings, or +hexadecimal byte strings. Values are Unicode strings and may expand a single +legacy glyph into multiple Unicode characters. Mapping files are limited to +1 MiB and each replacement to 64 Unicode code points to bound untrusted input +and expansion costs. + +```python +from openmed.processing import LegacyFontMap, convert_legacy_encoding + +font_map = LegacyFontMap.from_file("hospital-archive-font.json") +conversion = convert_legacy_encoding( + b"ABC", + encoding="legacy-font", + legacy_font_map=font_map, +) +assert conversion.text == "राम" +``` + +Auto-detection converts only dense runs with at least three mapped bytes and +at least two resulting Devanagari letters. Use +`encoding="legacy-font"` when a known, very short run is below that deliberately +conservative threshold. + +## Provenance and limitations + +The built-in ISCII assignments follow the public Indian standard +IS 13194:1991 and the Unicode Consortium's published ISCII semantics. The +mapping table was constructed independently for OpenMed and is distributed +under Apache-2.0. Caller-supplied legacy-font maps retain their own provenance +and license; OpenMed does not redistribute them. + +Vedic stress extension sequences can be decoded, but complete Vedic Sanskrit +accent and contextual invisible-letter fidelity is outside the byte-lossless +guarantee. OCR and image processing are also outside this text-only feature. diff --git a/mkdocs.yml b/mkdocs.yml index 81369d4c..d7f3e614 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -54,6 +54,7 @@ nav: - Configuration Profiles: profiles.md - Detector Plugin SDK: plugin-sdk.md - Batch Processing: batch-processing.md + - Legacy Indic Encoding: legacy-indic-encoding.md - Performance Profiling: profiling.md - Low-resource Benchmark: benchmarks/low-resource.md - Zero-shot NER How-to: zero-shot-howto.md diff --git a/openmed/core/pipeline.py b/openmed/core/pipeline.py index 7678b8d2..cc16737c 100644 --- a/openmed/core/pipeline.py +++ b/openmed/core/pipeline.py @@ -906,6 +906,17 @@ def _normalize_code_mixed_tags( return tuple(normalized_tags) def stage1_normalize(self, text: str) -> NormalizedDocument: + from ..processing.legacy_encoding import convert_legacy_encoding + + legacy_conversion = convert_legacy_encoding(text) + legacy_text = legacy_conversion.text + if legacy_conversion.encoding == "unicode": + legacy_origins = tuple((index, index + 1) for index in range(len(text))) + else: + # Auto-detected ISCII reaches this API as a Latin-1 surrogate + # string, so original byte and source-character offsets coincide. + legacy_origins = legacy_conversion.offset_map.converted_to_original_spans + repair_encoding, encoding_repair_metadata = _encoding_repairer() # Encoding repair is an optional (ftfy) capability. When it is # unavailable the repairer is the identity function, so calling it once @@ -920,7 +931,9 @@ def stage1_normalize(self, text: str) -> NormalizedDocument: normalized_to_original_span: list[tuple[int, int]] = [] normalized_length = 0 - for start, end, segment, is_whitespace in _iter_normalization_segments(text): + for start, end, segment, is_whitespace in _iter_normalization_segments( + legacy_text + ): normalized_start = normalized_length if is_whitespace: normalized_segment = " " @@ -935,14 +948,17 @@ def stage1_normalize(self, text: str) -> NormalizedDocument: if not normalized_segment: continue - for index in range(start, end): + source_spans = legacy_origins[start:end] + source_start = min(span[0] for span in source_spans) + source_end = max(span[1] for span in source_spans) + for index in range(source_start, source_end): original_to_normalized[index] = normalized_start normalized_parts.append(normalized_segment) normalized_length += len(normalized_segment) for _ in normalized_segment: - normalized_to_original.append(start) - normalized_to_original_span.append((start, end)) + normalized_to_original.append(source_start) + normalized_to_original_span.append((source_start, source_end)) normalized_text = "".join(normalized_parts) offset_map = OffsetMap( @@ -960,6 +976,15 @@ def stage1_normalize(self, text: str) -> NormalizedDocument: "original_length": len(text), "normalized_length": len(normalized_text), "encoding_repair": encoding_repair_metadata, + "legacy_encoding": { + "encoding": legacy_conversion.encoding, + "changed": legacy_conversion.changed, + "converted_bytes": ( + sum(ord(char) >= 0x80 for char in text) + if legacy_conversion.encoding == "iscii" + else 0 + ), + }, }, ) target_script = getattr(self.config, "chinese_target_script", None) diff --git a/openmed/core/script_detect.py b/openmed/core/script_detect.py index a4e91f64..785e60f6 100644 --- a/openmed/core/script_detect.py +++ b/openmed/core/script_detect.py @@ -23,10 +23,13 @@ from dataclasses import dataclass from enum import Enum from pathlib import Path -from typing import NoReturn +from typing import TYPE_CHECKING, NoReturn from .language_pack_catalog import SCRIPT_LANGUAGE_HINTS +if TYPE_CHECKING: + from ..processing.legacy_encoding import LegacyFontMap + UNKNOWN_SCRIPT = "Unknown" logger = logging.getLogger(__name__) @@ -519,6 +522,8 @@ class DetectionNormalization: folded_native_digits: int = 0 indic_changes: int = 0 indic_scripts: tuple[str, ...] = () + converted_legacy_bytes: int = 0 + legacy_encoding: str = "unicode" scripts: tuple[str, ...] = () mixed_script: bool = False chinese_variant_normalized: bool = False @@ -534,6 +539,7 @@ def changed(self) -> bool: or self.folded_confusables > 0 or self.folded_native_digits > 0 or self.indic_changes > 0 + or self.converted_legacy_bytes > 0 or self.chinese_variant_normalized ) @@ -565,6 +571,8 @@ def to_metadata(self) -> dict[str, object]: "folded_native_digits": self.folded_native_digits, "indic_changes": self.indic_changes, "indic_scripts": list(self.indic_scripts), + "converted_legacy_bytes": self.converted_legacy_bytes, + "legacy_encoding": self.legacy_encoding, "mixed_script": self.mixed_script, "opencc_available": self.opencc_available, "removed_zero_width": self.removed_zero_width, @@ -1146,6 +1154,93 @@ def normalize_for_pii_detection( *, width_convention: str = "cjk", chinese_target_script: str | None = None, + legacy_font_map: LegacyFontMap | None = None, +) -> DetectionNormalization: + """Convert legacy Indic text, then apply offset-safe Unicode defenses. + + Likely ISCII-1991 input and caller-mapped legacy-font runs are converted + before script routing. The resulting offsets are composed back to the + original source-byte coordinate system; ordinary Unicode text follows the + existing normalization path unchanged. + """ + + from ..processing.legacy_encoding import convert_legacy_encoding + + legacy_conversion = convert_legacy_encoding( + text, + legacy_font_map=legacy_font_map, + ) + normalized = _normalize_unicode_for_pii_detection( + legacy_conversion.text, + width_convention=width_convention, + chinese_target_script=chinese_target_script, + ) + if legacy_conversion.encoding == "unicode": + return normalized + + legacy_origins = legacy_conversion.offset_map.converted_to_original_spans + starts: list[int] = [] + ends: list[int] = [] + for start, end in zip(normalized.offset_starts, normalized.offset_ends): + source_start, source_end = _source_span_for_legacy_range( + legacy_origins, + start, + end, + ) + starts.append(source_start) + ends.append(source_end) + + converted_sources: set[int] = set() + converted_by_span: dict[tuple[int, int], list[str]] = {} + for char, span in zip(legacy_conversion.text, legacy_origins): + converted_by_span.setdefault(span, []).append(char) + for (start, end), converted_chars in converted_by_span.items(): + if text[start:end] != "".join(converted_chars): + converted_sources.update(range(start, end)) + + return DetectionNormalization( + text=normalized.text, + original_length=len(text), + offset_starts=tuple(starts), + offset_ends=tuple(ends), + removed_zero_width=normalized.removed_zero_width, + stripped_combining_marks=normalized.stripped_combining_marks, + folded_confusables=normalized.folded_confusables, + folded_native_digits=normalized.folded_native_digits, + indic_changes=normalized.indic_changes, + indic_scripts=normalized.indic_scripts, + converted_legacy_bytes=len(converted_sources), + legacy_encoding=legacy_conversion.encoding, + scripts=normalized.scripts, + mixed_script=normalized.mixed_script, + chinese_variant_normalized=normalized.chinese_variant_normalized, + chinese_target_script=normalized.chinese_target_script, + opencc_available=normalized.opencc_available, + ) + + +def _source_span_for_legacy_range( + origins: tuple[tuple[int, int], ...], + start: int, + end: int, +) -> tuple[int, int]: + spans = origins[start:end] + if spans: + return min(span[0] for span in spans), max(span[1] for span in spans) + if start < len(origins): + anchor = origins[start][0] + elif origins: + anchor = origins[-1][1] + else: + anchor = 0 + return anchor, anchor + + +def _normalize_unicode_for_pii_detection( + text: str, + *, + width_convention: str = "cjk", + chinese_target_script: str | None = None, ) -> DetectionNormalization: """Fold adversarial Unicode artifacts while preserving offset remapping. @@ -1240,11 +1335,11 @@ def normalize_for_pii_detection( and original_start > 0 and _script_for_char(text[original_start - 1]) == "Ethiopic" ) - if ( - category == "Mn" - and _script_for_char(char) not in INDIC_SCRIPTS - and not attached_ethiopic_mark - ): + attached_indic_mark = category == "Mn" and _is_attached_indic_mark( + digit_folding.text, + index, + ) + if category == "Mn" and not attached_indic_mark and not attached_ethiopic_mark: stripped_combining_marks += 1 continue @@ -1519,6 +1614,35 @@ def _script_for_char(char: str) -> str | None: return None +def _is_attached_indic_mark(text: str, index: int) -> bool: + char = text[index] + if not _is_indic_codepoint(char) or index == 0: + return False + previous = text[index - 1] + return _is_indic_codepoint(previous) and unicodedata.category(previous)[0] in { + "L", + "M", + } + + +def _is_indic_codepoint(char: str) -> bool: + codepoint = ord(char) + return any( + start <= codepoint <= end + for start, end in ( + (0x0900, 0x097F), + (0x0980, 0x09FF), + (0x0A00, 0x0A7F), + (0x0A80, 0x0AFF), + (0x0B00, 0x0B7F), + (0x0B80, 0x0BFF), + (0x0C00, 0x0C7F), + (0x0C80, 0x0CFF), + (0x0D00, 0x0D7F), + ) + ) + + def _expand_detection_window( text: str, core_start: int, diff --git a/openmed/processing/__init__.py b/openmed/processing/__init__.py index a3402c64..6d1df7a1 100644 --- a/openmed/processing/__init__.py +++ b/openmed/processing/__init__.py @@ -50,6 +50,16 @@ deidentify_stream, replay, ) +from .legacy_encoding import ( + ISCII_MAPPING_PROVENANCE, + ConversionOffsetMap, + LegacyConversion, + LegacyFontMap, + convert_legacy_encoding, + detect_legacy_encoding, + iscii_to_unicode, + unicode_to_iscii, +) from .object_storage import ( ObjectProgressCallback, ObjectStorageBatchResult, @@ -64,6 +74,7 @@ IndicNormalization, IndicNormalizer, TextProcessor, + normalize_indic_text, postprocess_text, preprocess_text, ) @@ -128,6 +139,15 @@ "INDIC_SCRIPTS", "IndicNormalization", "IndicNormalizer", + "normalize_indic_text", + "ConversionOffsetMap", + "ISCII_MAPPING_PROVENANCE", + "LegacyConversion", + "LegacyFontMap", + "convert_legacy_encoding", + "detect_legacy_encoding", + "iscii_to_unicode", + "unicode_to_iscii", "preprocess_text", "postprocess_text", "TokenizationHelper", diff --git a/openmed/processing/legacy_encoding.py b/openmed/processing/legacy_encoding.py new file mode 100644 index 00000000..7547c0bf --- /dev/null +++ b/openmed/processing/legacy_encoding.py @@ -0,0 +1,924 @@ +"""Convert legacy Devanagari encodings to Unicode with byte offsets. + +The ISCII assignments in this module follow the public Indian standard +IS 13194:1991 and the Unicode Consortium's published description of ISCII +semantics. The tables were constructed for OpenMed from those public code +assignments; they are original project data and are distributed under +OpenMed's Apache-2.0 license. + +No proprietary legacy-font table is bundled. :class:`LegacyFontMap` loads a +caller-supplied JSON or YAML data file instead. This keeps font-specific +licensing and provenance with the user who is entitled to the source font or +mapping data. + +Vedic stress marks are decoded when their standard ISCII extension sequences +are encountered, but full Vedic round-trip coverage is outside this module's +lossless guarantee. The guarantee covers the standard non-Vedic Devanagari +repertoire represented below, excluding contextual ``INV`` presentation +controls. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType +from typing import Literal + +import yaml + +from .text import normalize_indic_text + +LegacyEncoding = Literal["unicode", "iscii", "legacy-font"] +ErrorMode = Literal["strict", "replace"] + +ISCII_MAPPING_PROVENANCE = ( + "IS 13194:1991 public-standard assignments; OpenMed-original table, Apache-2.0" +) + +_ISCII_ATR = 0xEF +_ISCII_EXT = 0xF0 +_ISCII_DEVANAGARI = 0x42 +_ISCII_INV = 0xD9 +_ISCII_HALANT = 0xE8 +_ISCII_NUKTA = 0xE9 + +# One-byte ISCII-1991 Devanagari assignments. ASCII remains ASCII and is not +# duplicated here. Context-sensitive two-byte forms are listed separately. +_ISCII_TO_UNICODE: dict[int, str] = { + 0xA0: "\u00a0", + 0xA1: "\u0901", + 0xA2: "\u0902", + 0xA3: "\u0903", + 0xA4: "\u0905", + 0xA5: "\u0906", + 0xA6: "\u0907", + 0xA7: "\u0908", + 0xA8: "\u0909", + 0xA9: "\u090a", + 0xAA: "\u090b", + 0xAB: "\u090e", + 0xAC: "\u090f", + 0xAD: "\u0910", + 0xAE: "\u090d", + 0xAF: "\u0912", + 0xB0: "\u0913", + 0xB1: "\u0914", + 0xB2: "\u0911", + 0xB3: "\u0915", + 0xB4: "\u0916", + 0xB5: "\u0917", + 0xB6: "\u0918", + 0xB7: "\u0919", + 0xB8: "\u091a", + 0xB9: "\u091b", + 0xBA: "\u091c", + 0xBB: "\u091d", + 0xBC: "\u091e", + 0xBD: "\u091f", + 0xBE: "\u0920", + 0xBF: "\u0921", + 0xC0: "\u0922", + 0xC1: "\u0923", + 0xC2: "\u0924", + 0xC3: "\u0925", + 0xC4: "\u0926", + 0xC5: "\u0927", + 0xC6: "\u0928", + 0xC7: "\u0929", + 0xC8: "\u092a", + 0xC9: "\u092b", + 0xCA: "\u092c", + 0xCB: "\u092d", + 0xCC: "\u092e", + 0xCD: "\u092f", + 0xCE: "\u095f", + 0xCF: "\u0930", + 0xD0: "\u0931", + 0xD1: "\u0932", + 0xD2: "\u0933", + 0xD3: "\u0934", + 0xD4: "\u0935", + 0xD5: "\u0936", + 0xD6: "\u0937", + 0xD7: "\u0938", + 0xD8: "\u0939", + 0xD9: "\u200d", + 0xDA: "\u093e", + 0xDB: "\u093f", + 0xDC: "\u0940", + 0xDD: "\u0941", + 0xDE: "\u0942", + 0xDF: "\u0943", + 0xE0: "\u0946", + 0xE1: "\u0947", + 0xE2: "\u0948", + 0xE3: "\u0945", + 0xE4: "\u094a", + 0xE5: "\u094b", + 0xE6: "\u094c", + 0xE7: "\u0949", + 0xE8: "\u094d", + 0xE9: "\u093c", + 0xEA: "\u0964", + 0xF1: "\u0966", + 0xF2: "\u0967", + 0xF3: "\u0968", + 0xF4: "\u0969", + 0xF5: "\u096a", + 0xF6: "\u096b", + 0xF7: "\u096c", + 0xF8: "\u096d", + 0xF9: "\u096e", + 0xFA: "\u096f", +} + +_ISCII_SEQUENCE_TO_UNICODE: dict[bytes, str] = { + b"\xa4\xe0": "\u0904", + b"\xa6\xe9": "\u090c", + b"\xa1\xe9": "\u0950", + b"\xaa\xe9": "\u0960", + b"\xa7\xe9": "\u0961", + b"\xb3\xe9": "\u0958", + b"\xb4\xe9": "\u0959", + b"\xb5\xe9": "\u095a", + b"\xba\xe9": "\u095b", + b"\xbf\xe9": "\u095c", + b"\xc0\xe9": "\u095d", + b"\xc9\xe9": "\u095e", + b"\xdb\xe9": "\u0962", + b"\xdc\xe9": "\u0963", + b"\xdf\xe9": "\u0944", + b"\xea\xe9": "\u093d", + b"\xea\xea": "\u0965", +} + +_ISCII_EXTENSION_TO_UNICODE = { + 0xB5: "\u0951", # Vedic: Udatta (documented limitation). + 0xB8: "\u0952", # Vedic: Anudatta (documented limitation). + 0xBF: "\u0970", # Devanagari abbreviation sign. +} + +_UNICODE_SPECIAL_TO_ISCII: dict[str, bytes] = { + "\u0904": b"\xa4\xe0", + "\u090c": b"\xa6\xe9", + "\u093d": b"\xea\xe9", + "\u0944": b"\xdf\xe9", + "\u0950": b"\xa1\xe9", + "\u0951": b"\xf0\xb5", + "\u0952": b"\xf0\xb8", + "\u0960": b"\xaa\xe9", + "\u0961": b"\xa7\xe9", + "\u0962": b"\xdb\xe9", + "\u0963": b"\xdc\xe9", + "\u0965": b"\xea\xea", + "\u0970": b"\xf0\xbf", +} + +# U+095F has a canonical decomposition to YA + NUKTA under NFC, while ISCII +# assigns the letter its own byte. Prefer that canonical byte when encoding. +_UNICODE_SEQUENCE_TO_ISCII = {"\u092f\u093c": b"\xce"} + +_UNICODE_TO_ISCII = { + unicode_char: iscii_byte + for iscii_byte, unicode_char in _ISCII_TO_UNICODE.items() + if iscii_byte != _ISCII_INV +} + +_ISCII_CORE_LETTERS = frozenset(range(0xB3, 0xD9)) +_ISCII_UNDEFINED = frozenset(range(0xEB, 0xEF)) | frozenset(range(0xFB, 0x100)) +_LEGACY_GAP_BYTES = frozenset(b" \t\r\n-_/.,:;()[]{}") +_MAX_FONT_MAP_FILE_BYTES = 1024 * 1024 +_MAX_FONT_MAP_VALUE_CHARS = 64 +_MAX_FONT_MAP_NAME_CHARS = 256 +_MAX_FONT_MAP_PROVENANCE_CHARS = 1024 +_MIN_AUTO_ISCII_CORE_LETTERS = 4 +_MIN_AUTO_ISCII_DENSITY = 0.6 + + +@dataclass(frozen=True) +class ConversionOffsetMap: + """Map converted character spans to original byte spans and back.""" + + original_length: int + converted_to_original_spans: tuple[tuple[int, int], ...] + original_to_converted: tuple[int | None, ...] + + def to_original_span(self, start: int, end: int) -> tuple[int, int]: + """Map converted ``[start, end)`` offsets to source-byte offsets.""" + + length = len(self.converted_to_original_spans) + if not (0 <= start <= end <= length): + raise ValueError("span must satisfy 0 <= start <= end <= len(text)") + if start == end: + if start < length: + anchor = self.converted_to_original_spans[start][0] + elif length: + anchor = self.converted_to_original_spans[-1][1] + else: + anchor = 0 + return anchor, anchor + spans = self.converted_to_original_spans[start:end] + return min(item[0] for item in spans), max(item[1] for item in spans) + + def to_converted_span(self, start: int, end: int) -> tuple[int, int]: + """Map source-byte ``[start, end)`` offsets to converted offsets.""" + + if not (0 <= start <= end <= self.original_length): + raise ValueError("span must satisfy 0 <= start <= end <= original_length") + mapped = [ + index + for index, (source_start, source_end) in enumerate( + self.converted_to_original_spans + ) + if source_start < end and source_end > start + ] + if mapped: + return min(mapped), max(mapped) + 1 + cursor = start + while cursor < self.original_length: + value = self.original_to_converted[cursor] + if value is not None: + return value, value + cursor += 1 + terminal = len(self.converted_to_original_spans) + return terminal, terminal + + +@dataclass(frozen=True) +class LegacyConversion: + """Unicode conversion result and alignment to the original byte stream.""" + + text: str + original: bytes + encoding: LegacyEncoding + offset_map: ConversionOffsetMap + changed: bool + + def to_original_span(self, start: int, end: int) -> tuple[int, int]: + """Map a converted span to the original byte stream.""" + + return self.offset_map.to_original_span(start, end) + + +@dataclass(frozen=True) +class LegacyFontMap: + """Caller-supplied byte-to-Unicode mapping for one legacy font.""" + + name: str + mapping: Mapping[int, str] + provenance: str = "user-supplied" + + def __post_init__(self) -> None: + """Validate and freeze the mapping.""" + + if not isinstance(self.name, str) or not self.name.strip(): + raise ValueError("legacy font map name must not be empty") + if len(self.name) > _MAX_FONT_MAP_NAME_CHARS: + raise ValueError("legacy font map name is too long") + if not isinstance(self.provenance, str): + raise ValueError("legacy font map provenance must be text") + if len(self.provenance) > _MAX_FONT_MAP_PROVENANCE_CHARS: + raise ValueError("legacy font map provenance is too long") + validated: dict[int, str] = {} + for key, value in self.mapping.items(): + if isinstance(key, bool) or not isinstance(key, int) or not 0 <= key <= 255: + raise ValueError("legacy font map keys must be byte values 0..255") + if not isinstance(value, str) or not value: + raise ValueError("legacy font map values must be non-empty strings") + if len(value) > _MAX_FONT_MAP_VALUE_CHARS: + raise ValueError( + "legacy font map values may contain at most 64 characters" + ) + validated[key] = value + if not validated: + raise ValueError("legacy font map must contain at least one mapping") + object.__setattr__(self, "mapping", MappingProxyType(validated)) + + @classmethod + def from_file(cls, path: str | Path) -> "LegacyFontMap": + """Load a JSON or YAML legacy-font map. + + The file may contain a top-level ``mapping`` object plus optional + ``name`` and ``provenance`` fields, or it may be the mapping object + itself. Keys may be integers, one-byte characters, decimal strings, or + hexadecimal strings such as ``"0x66"``. + + Args: + path: JSON, YAML, or YML mapping file. + + Returns: + Validated immutable legacy-font map. + """ + + source = Path(path) + suffix = source.suffix.lower() + if suffix not in {".json", ".yaml", ".yml"}: + raise ValueError("legacy font map must be JSON, YAML, or YML") + with source.open("rb") as handle: + raw_payload = handle.read(_MAX_FONT_MAP_FILE_BYTES + 1) + if len(raw_payload) > _MAX_FONT_MAP_FILE_BYTES: + raise ValueError("legacy font map file exceeds the 1 MiB limit") + source_text = raw_payload.decode("utf-8") + if suffix == ".json": + payload = json.loads(source_text) + else: + payload = yaml.safe_load(source_text) + if not isinstance(payload, Mapping): + raise ValueError("legacy font map file must contain an object") + + raw_mapping = payload.get("mapping", payload) + if not isinstance(raw_mapping, Mapping): + raise ValueError("legacy font map 'mapping' must be an object") + mapping: dict[int, str] = {} + for key, value in raw_mapping.items(): + byte_key = _parse_map_key(key) + if byte_key in mapping: + raise ValueError(f"duplicate legacy font map byte: {byte_key}") + if not isinstance(value, str): + raise ValueError("legacy font map values must be strings") + mapping[byte_key] = value + name = payload.get("name", source.stem) + provenance = payload.get("provenance", f"user-supplied:{source.name}") + if not isinstance(name, str): + raise ValueError("legacy font map name must be text") + if not isinstance(provenance, str): + raise ValueError("legacy font map provenance must be text") + return cls(name=name, mapping=mapping, provenance=provenance) + + +def detect_legacy_encoding( + data: bytes | str, + *, + legacy_font_map: LegacyFontMap | None = None, +) -> LegacyEncoding: + """Conservatively detect ISCII or a supplied ASCII-remapped font. + + A Devanagari ISCII attribute sequence is definitive. Otherwise at least + two valid high bytes, including a core ISCII consonant, are required. A + legacy-font candidate needs a dense run of at least three mapped bytes and + must produce Devanagari letters. These gates intentionally prefer a false + negative over corrupting an ordinary Latin clinical note. + """ + + raw = _coerce_legacy_bytes(data) + if raw is None: + return "unicode" + if _is_non_ascii_utf8(raw): + return "unicode" + if raw.startswith(b"\xef\x42"): + return "iscii" + + content_bytes = [byte for byte in raw if byte not in _LEGACY_GAP_BYTES] + high_bytes = [byte for byte in content_bytes if byte >= 0xA0] + valid_high = [byte for byte in high_bytes if byte in _ISCII_TO_UNICODE] + if ( + len(valid_high) >= _MIN_AUTO_ISCII_CORE_LETTERS + and len(valid_high) / max(1, len(high_bytes)) >= 0.8 + and len(valid_high) / max(1, len(content_bytes)) >= _MIN_AUTO_ISCII_DENSITY + and sum(byte in _ISCII_CORE_LETTERS for byte in valid_high) + >= _MIN_AUTO_ISCII_CORE_LETTERS + and not any(0x80 <= byte <= 0x9F for byte in raw) + and not any(byte in _ISCII_UNDEFINED for byte in high_bytes) + ): + return "iscii" + + if legacy_font_map is not None and _legacy_font_candidate_runs( + raw, legacy_font_map + ): + return "legacy-font" + return "unicode" + + +def iscii_to_unicode( + data: bytes, + *, + errors: ErrorMode = "strict", +) -> LegacyConversion: + """Convert ISCII-1991 Devanagari bytes to NFC Unicode. + + Contextual nukta forms, explicit/soft virama controls, Devanagari script + attributes, double danda, and supported extension sequences are handled. + Every output character retains the source-byte span that produced it. + """ + + _validate_errors(errors) + chars: list[str] = [] + origins: list[tuple[int, int]] = [] + index = 0 + while index < len(data): + byte = data[index] + if byte < 0x80: + chars.append(chr(byte)) + origins.append((index, index + 1)) + index += 1 + continue + + if byte == _ISCII_ATR: + if index + 1 >= len(data): + index = _decode_error(data, index, errors, chars, origins) + continue + if data[index + 1] != _ISCII_DEVANAGARI: + index = _decode_error(data, index, errors, chars, origins, length=2) + continue + index += 2 + continue + + if byte == _ISCII_EXT: + if index + 1 >= len(data): + index = _decode_error(data, index, errors, chars, origins) + continue + extension = _ISCII_EXTENSION_TO_UNICODE.get(data[index + 1]) + if extension is None: + index = _decode_error(data, index, errors, chars, origins, length=2) + continue + chars.append(extension) + origins.append((index, index + 2)) + index += 2 + continue + + pair = data[index : index + 2] + if pair == b"\xe8\xe8": + chars.extend(("\u094d", "\u200c")) + origins.extend(((index, index + 1), (index + 1, index + 2))) + index += 2 + continue + if pair == b"\xe8\xe9": + chars.extend(("\u094d", "\u200d")) + origins.extend(((index, index + 1), (index + 1, index + 2))) + index += 2 + continue + sequence_char = _ISCII_SEQUENCE_TO_UNICODE.get(pair) + if sequence_char is not None: + chars.append(sequence_char) + origins.append((index, index + 2)) + index += 2 + continue + + unicode_char = _ISCII_TO_UNICODE.get(byte) + if unicode_char is None: + index = _decode_error(data, index, errors, chars, origins) + continue + chars.append(unicode_char) + origins.append((index, index + 1)) + index += 1 + + normalization = normalize_indic_text("".join(chars), char_origins=tuple(origins)) + offset_map = _build_offset_map(len(data), normalization.char_origins) + return LegacyConversion( + text=normalization.text, + original=bytes(data), + encoding="iscii", + offset_map=offset_map, + changed=True, + ) + + +def unicode_to_iscii( + text: str, + *, + errors: ErrorMode = "strict", + include_atr: bool = False, +) -> bytes: + """Encode Unicode Devanagari to ISCII-1991 bytes. + + Args: + text: Unicode text containing ASCII and supported Devanagari. + errors: ``"strict"`` raises :class:`UnicodeEncodeError`; ``"replace"`` + emits ASCII ``?`` for unsupported code points. + include_atr: Prefix the Devanagari script attribute ``EF 42``. + + Returns: + ISCII byte stream. + """ + + _validate_errors(errors) + normalized = normalize_indic_text(text).text + output = bytearray(b"\xef\x42" if include_atr else b"") + previous_char = "" + index = 0 + while index < len(normalized): + sequence = next( + ( + (unicode_sequence, iscii_sequence) + for unicode_sequence, iscii_sequence in _UNICODE_SEQUENCE_TO_ISCII.items() + if normalized.startswith(unicode_sequence, index) + ), + None, + ) + if sequence is not None: + unicode_sequence, iscii_sequence = sequence + output.extend(iscii_sequence) + previous_char = unicode_sequence[-1] + index += len(unicode_sequence) + continue + + char = normalized[index] + codepoint = ord(char) + if codepoint < 0x80: + output.append(codepoint) + elif char == "\u200c" and previous_char == "\u094d": + output.append(_ISCII_HALANT) + elif char == "\u200d": + output.append(_ISCII_NUKTA if previous_char == "\u094d" else _ISCII_INV) + elif char in _UNICODE_SPECIAL_TO_ISCII: + output.extend(_UNICODE_SPECIAL_TO_ISCII[char]) + elif char in _UNICODE_TO_ISCII: + output.append(_UNICODE_TO_ISCII[char]) + elif errors == "replace": + output.append(ord("?")) + else: + raise UnicodeEncodeError( + "iscii-dev", + normalized, + index, + index + 1, + "character is not representable in ISCII Devanagari", + ) + previous_char = char + index += 1 + return bytes(output) + + +def convert_legacy_encoding( + data: bytes | str, + *, + encoding: Literal["auto", "unicode", "iscii", "legacy-font"] = "auto", + legacy_font_map: LegacyFontMap | None = None, + errors: ErrorMode = "strict", +) -> LegacyConversion: + """Convert detected ISCII or caller-mapped legacy-font runs to Unicode. + + ``bytes`` are preferred whenever original-stream byte offsets matter. A + ``str`` containing Latin-1 code points is accepted for compatibility with + systems that decoded legacy bytes without conversion; its code-point and + original-byte offsets are then identical. + """ + + _validate_errors(errors) + if encoding not in {"auto", "unicode", "iscii", "legacy-font"}: + raise ValueError(f"unknown legacy encoding: {encoding!r}") + + raw = _coerce_legacy_bytes(data) + if raw is None: + if encoding in {"iscii", "legacy-font"}: + raise ValueError("legacy conversion requires bytes or Latin-1 text") + assert isinstance(data, str) + return _unicode_identity(data) + + detected: LegacyEncoding + if encoding == "auto": + detected = detect_legacy_encoding(raw, legacy_font_map=legacy_font_map) + else: + detected = encoding + + if detected == "iscii": + return iscii_to_unicode(raw, errors=errors) + if detected == "legacy-font": + if legacy_font_map is None: + raise ValueError("legacy-font conversion requires legacy_font_map") + return _convert_legacy_font( + raw, + legacy_font_map, + auto_detected=encoding == "auto", + errors=errors, + ) + return _decode_unicode(raw, original=data, errors=errors) + + +def _convert_legacy_font( + data: bytes, + font_map: LegacyFontMap, + *, + auto_detected: bool, + errors: ErrorMode, +) -> LegacyConversion: + candidate_ranges = ( + _legacy_font_candidate_runs(data, font_map) + if auto_detected + else ((0, len(data)),) + ) + chars: list[str] = [] + origins: list[tuple[int, int]] = [] + range_index = 0 + active = candidate_ranges[range_index] if candidate_ranges else None + + for index, byte in enumerate(data): + while active is not None and index >= active[1]: + range_index += 1 + active = ( + candidate_ranges[range_index] + if range_index < len(candidate_ranges) + else None + ) + in_candidate = active is not None and active[0] <= index < active[1] + replacement = font_map.mapping.get(byte) if in_candidate else None + if replacement is None: + if byte < 0x80: + replacement = chr(byte) + elif errors == "replace": + replacement = "\ufffd" + else: + raise UnicodeDecodeError( + f"legacy-font:{font_map.name}", + data, + index, + index + 1, + "unmapped non-ASCII byte", + ) + chars.extend(replacement) + origins.extend((index, index + 1) for _ in replacement) + + normalization = normalize_indic_text("".join(chars), char_origins=tuple(origins)) + return LegacyConversion( + text=normalization.text, + original=data, + encoding="legacy-font", + offset_map=_build_offset_map(len(data), normalization.char_origins), + changed=True, + ) + + +def _legacy_font_candidate_runs( + data: bytes, + font_map: LegacyFontMap, +) -> tuple[tuple[int, int], ...]: + mapped_positions = [ + index for index, byte in enumerate(data) if byte in font_map.mapping + ] + if not mapped_positions: + return () + + groups: list[list[int]] = [[mapped_positions[0]]] + for position in mapped_positions[1:]: + previous = groups[-1][-1] + gap = data[previous + 1 : position] + if len(gap) <= 2 and all(byte in _LEGACY_GAP_BYTES for byte in gap): + groups[-1].append(position) + else: + groups.append([position]) + + candidates: list[tuple[int, int]] = [] + for positions in groups: + start, end = positions[0], positions[-1] + 1 + mapped_count = len(positions) + if mapped_count < 3 or len({data[index] for index in positions}) < 2: + continue + output = "".join(font_map.mapping[data[index]] for index in positions) + devanagari_letters = sum(0x0904 <= ord(char) <= 0x0939 for char in output) + alphanumeric = sum(chr(byte).isalnum() for byte in data[start:end]) + coverage = mapped_count / max(1, alphanumeric) + if ( + devanagari_letters >= 2 + and coverage >= 0.7 + and _looks_suspicious_legacy_cluster(data[start:end]) + ): + candidates.append((start, end)) + return tuple(candidates) + + +def _decode_unicode( + data: bytes, + *, + original: bytes | str, + errors: ErrorMode, +) -> LegacyConversion: + if isinstance(original, str): + text = original + origins = tuple((index, index + 1) for index in range(len(text))) + original_bytes = data + else: + original_bytes = data + if errors == "strict": + text = data.decode("utf-8") + origins_list: list[tuple[int, int]] = [] + cursor = 0 + for char in text: + width = len(char.encode("utf-8")) + origins_list.append((cursor, cursor + width)) + cursor += width + origins = tuple(origins_list) + else: + text, origins = _decode_utf8_replacing_invalid(data) + return LegacyConversion( + text=text, + original=original_bytes, + encoding="unicode", + offset_map=_build_offset_map(len(original_bytes), origins), + changed=False, + ) + + +def _unicode_identity(text: str) -> LegacyConversion: + data = text.encode("utf-8") + origins: list[tuple[int, int]] = [] + cursor = 0 + for char in text: + width = len(char.encode("utf-8")) + origins.append((cursor, cursor + width)) + cursor += width + return LegacyConversion( + text=text, + original=data, + encoding="unicode", + offset_map=_build_offset_map(len(data), tuple(origins)), + changed=False, + ) + + +def _decode_utf8_replacing_invalid( + data: bytes, +) -> tuple[str, tuple[tuple[int, int], ...]]: + """Decode UTF-8 with replacement characters aligned to consumed bytes.""" + + chars: list[str] = [] + origins: list[tuple[int, int]] = [] + cursor = 0 + while cursor < len(data): + first = data[cursor] + if first < 0x80: + chars.append(chr(first)) + origins.append((cursor, cursor + 1)) + cursor += 1 + continue + + if 0xC2 <= first <= 0xDF: + width, second_min, second_max = 2, 0x80, 0xBF + elif first == 0xE0: + width, second_min, second_max = 3, 0xA0, 0xBF + elif 0xE1 <= first <= 0xEC or 0xEE <= first <= 0xEF: + width, second_min, second_max = 3, 0x80, 0xBF + elif first == 0xED: + width, second_min, second_max = 3, 0x80, 0x9F + elif first == 0xF0: + width, second_min, second_max = 4, 0x90, 0xBF + elif 0xF1 <= first <= 0xF3: + width, second_min, second_max = 4, 0x80, 0xBF + elif first == 0xF4: + width, second_min, second_max = 4, 0x80, 0x8F + else: + chars.append("\ufffd") + origins.append((cursor, cursor + 1)) + cursor += 1 + continue + + valid_width = 1 + while valid_width < width and cursor + valid_width < len(data): + byte = data[cursor + valid_width] + lower = second_min if valid_width == 1 else 0x80 + upper = second_max if valid_width == 1 else 0xBF + if not lower <= byte <= upper: + break + valid_width += 1 + + if valid_width == width: + end = cursor + width + chars.append(data[cursor:end].decode("utf-8")) + origins.append((cursor, end)) + cursor = end + continue + + chars.append("\ufffd") + origins.append((cursor, cursor + valid_width)) + cursor += valid_width + return "".join(chars), tuple(origins) + + +def _build_offset_map( + original_length: int, + origins: tuple[tuple[int, int], ...], +) -> ConversionOffsetMap: + original_to_converted: list[int | None] = [None] * original_length + for converted_index, (start, end) in enumerate(origins): + for original_index in range(start, end): + if original_to_converted[original_index] is None: + original_to_converted[original_index] = converted_index + return ConversionOffsetMap( + original_length=original_length, + converted_to_original_spans=origins, + original_to_converted=tuple(original_to_converted), + ) + + +def _decode_error( + data: bytes, + index: int, + errors: ErrorMode, + chars: list[str], + origins: list[tuple[int, int]], + *, + length: int = 1, +) -> int: + end = min(len(data), index + length) + if errors == "strict": + raise UnicodeDecodeError( + "iscii-dev", + data, + index, + end, + "invalid or unsupported ISCII sequence", + ) + chars.append("\ufffd") + origins.append((index, end)) + return end + + +def _coerce_legacy_bytes(data: bytes | str) -> bytes | None: + if isinstance(data, bytes): + return data + if not isinstance(data, str): + raise TypeError("legacy input must be bytes or str") + try: + return data.encode("latin-1") + except UnicodeEncodeError: + return None + + +def _is_non_ascii_utf8(data: bytes) -> bool: + if not any(byte >= 0x80 for byte in data): + return False + try: + decoded = data.decode("utf-8") + except UnicodeDecodeError: + return False + return any(ord(char) >= 0x80 for char in decoded) + + +def _looks_suspicious_legacy_cluster(data: bytes) -> bool: + if any(byte >= 0x80 for byte in data): + return True + text = data.decode("ascii") + if any(char in "`~[]{}\\|#$%^&*_+=/<>" for char in text): + return True + + words = [word for word in _ascii_words(text) if len(word) >= 4] + if not words: + return False + suspicious = 0 + for word in words: + interior_case_change = any( + left.islower() and right.isupper() for left, right in zip(word, word[1:]) + ) + vowel_ratio = sum(char.lower() in "aeiou" for char in word) / len(word) + if interior_case_change or vowel_ratio < 0.2: + suspicious += 1 + return suspicious / len(words) >= 0.6 + + +def _ascii_words(text: str) -> tuple[str, ...]: + words: list[str] = [] + current: list[str] = [] + for char in text: + if char.isalpha(): + current.append(char) + elif current: + words.append("".join(current)) + current = [] + if current: + words.append("".join(current)) + return tuple(words) + + +def _parse_map_key(key: object) -> int: + if isinstance(key, bool): + raise ValueError("legacy font map keys must identify one byte") + if isinstance(key, int): + value = key + elif isinstance(key, str): + if len(key) == 1: + value = ord(key) + elif key.lower().startswith("0x"): + if len(key) > 4: + raise ValueError(f"invalid legacy font map key: {key!r}") + value = int(key, 16) + elif key.isdecimal(): + if len(key) > 3: + raise ValueError(f"invalid legacy font map key: {key!r}") + value = int(key, 10) + else: + raise ValueError(f"invalid legacy font map key: {key!r}") + else: + raise ValueError("legacy font map keys must identify one byte") + if not 0 <= value <= 255: + raise ValueError("legacy font map keys must be byte values 0..255") + return value + + +def _validate_errors(errors: str) -> None: + if errors not in {"strict", "replace"}: + raise ValueError("errors must be 'strict' or 'replace'") + + +__all__ = [ + "ConversionOffsetMap", + "ISCII_MAPPING_PROVENANCE", + "LegacyConversion", + "LegacyEncoding", + "LegacyFontMap", + "convert_legacy_encoding", + "detect_legacy_encoding", + "iscii_to_unicode", + "unicode_to_iscii", +] diff --git a/openmed/processing/text.py b/openmed/processing/text.py index 38f2d4d3..3bc5a492 100644 --- a/openmed/processing/text.py +++ b/openmed/processing/text.py @@ -692,6 +692,155 @@ def _normalize_vowel_endings( return output, changes +_DEVANAGARI_PREBASE_I = "\u093f" +_DEVANAGARI_NUKTA = "\u093c" +_DEVANAGARI_VIRAMA = "\u094d" +_INDIC_JOIN_CONTROLS = frozenset({"\u200c", "\u200d"}) + + +@dataclass(frozen=True) +class LegacyIndicNormalization: + """Logically ordered NFC text with a caller-supplied source alignment.""" + + text: str + char_origins: tuple[tuple[int, int], ...] + + def to_original_span(self, start: int, end: int) -> tuple[int, int]: + """Map a normalized span to the supplied source coordinate system.""" + + if not (0 <= start <= end <= len(self.text)): + raise ValueError("span must satisfy 0 <= start <= end <= len(text)") + if start == end: + if start < len(self.char_origins): + anchor = self.char_origins[start][0] + elif self.char_origins: + anchor = self.char_origins[-1][1] + else: + anchor = 0 + return anchor, anchor + spans = self.char_origins[start:end] + return min(span[0] for span in spans), max(span[1] for span in spans) + + +def normalize_indic_text( + text: str, + *, + char_origins: tuple[tuple[int, int], ...] | None = None, +) -> LegacyIndicNormalization: + """Normalize legacy Devanagari ordering and return stable NFC text. + + ISCII stores dependent matras in logical order, while visual-order legacy + fonts can place the vowel sign I before its consonant cluster. This narrow + compatibility normalizer reorders those glyph streams without applying the + broader spelling canonicalizations performed by :class:`IndicNormalizer`, + so ISCII round trips remain byte-identical. + """ + + if char_origins is None: + char_origins = tuple((index, index + 1) for index in range(len(text))) + if len(char_origins) != len(text): + raise ValueError("char_origins must contain one span per input character") + + units: list[tuple[str, tuple[int, int]]] = [] + for char, origin in zip(text, char_origins): + for normalized_char in unicodedata.normalize("NFC", char): + units.append((normalized_char, origin)) + + reordered: list[tuple[str, tuple[int, int]]] = [] + index = 0 + while index < len(units): + char = units[index][0] + if ( + char == _DEVANAGARI_PREBASE_I + and index + 1 < len(units) + and _is_devanagari_consonant(units[index + 1][0]) + ): + cluster_end = _devanagari_cluster_end(units, index + 1) + reordered.extend(units[index + 1 : cluster_end]) + reordered.append(units[index]) + index = cluster_end + continue + reordered.append(units[index]) + index += 1 + + # Unicode canonical ordering requires nukta (CCC 7) before virama (CCC 9). + index = 1 + while index < len(reordered): + if ( + reordered[index][0] == _DEVANAGARI_NUKTA + and reordered[index - 1][0] == _DEVANAGARI_VIRAMA + ): + reordered[index - 1], reordered[index] = ( + reordered[index], + reordered[index - 1], + ) + index += 1 + + normalized_units = _normalize_legacy_unit_groups(reordered) + return LegacyIndicNormalization( + text="".join(char for char, _ in normalized_units), + char_origins=tuple(origin for _, origin in normalized_units), + ) + + +def _is_devanagari_consonant(char: str) -> bool: + codepoint = ord(char) + return 0x0915 <= codepoint <= 0x0939 or 0x0978 <= codepoint <= 0x097F + + +def _devanagari_cluster_end( + units: list[tuple[str, tuple[int, int]]], + start: int, +) -> int: + index = start + 1 + if index < len(units) and units[index][0] == _DEVANAGARI_NUKTA: + index += 1 + while index < len(units) and units[index][0] == _DEVANAGARI_VIRAMA: + index += 1 + if index < len(units) and units[index][0] in _INDIC_JOIN_CONTROLS: + index += 1 + if index >= len(units) or not _is_devanagari_consonant(units[index][0]): + break + index += 1 + if index < len(units) and units[index][0] == _DEVANAGARI_NUKTA: + index += 1 + return index + + +def _normalize_legacy_unit_groups( + units: list[tuple[str, tuple[int, int]]], +) -> list[tuple[str, tuple[int, int]]]: + normalized: list[tuple[str, tuple[int, int]]] = [] + group: list[tuple[str, tuple[int, int]]] = [] + + def flush() -> None: + if not group: + return + source_text = "".join(char for char, _ in group) + nfc_text = unicodedata.normalize("NFC", source_text) + if nfc_text == source_text: + normalized.extend(group) + else: + combined = ( + min(origin[0] for _, origin in group), + max(origin[1] for _, origin in group), + ) + normalized.extend((char, combined) for char in nfc_text) + group.clear() + + for unit in units: + char = unit[0] + if ( + group + and unicodedata.combining(char) == 0 + and char not in _INDIC_JOIN_CONTROLS + ): + flush() + group.append(unit) + flush() + return normalized + + class TextProcessor: """Handles text preprocessing and cleaning for medical text analysis.""" diff --git a/tests/unit/processing/test_legacy_encoding.py b/tests/unit/processing/test_legacy_encoding.py new file mode 100644 index 00000000..f756ef75 --- /dev/null +++ b/tests/unit/processing/test_legacy_encoding.py @@ -0,0 +1,278 @@ +"""Acceptance gates for ISCII and caller-supplied legacy-font conversion.""" + +from __future__ import annotations + +import json +import unicodedata + +import pytest + +from openmed.core.pipeline import Pipeline +from openmed.core.script_detect import normalize_for_pii_detection +from openmed.processing.legacy_encoding import ( + ISCII_MAPPING_PROVENANCE, + LegacyFontMap, + convert_legacy_encoding, + detect_legacy_encoding, + iscii_to_unicode, + unicode_to_iscii, +) +from openmed.processing.text import normalize_indic_text + +# Synthetic non-Vedic Hindi fixture: "रमेश शर्मा" in ISCII-1991. +_ISCII_NAME = bytes.fromhex("cf cc e1 d5 20 d5 cf e8 cc da") + + +def _synthetic_font_map() -> LegacyFontMap: + # Visual-order "f" is the pre-base I glyph; S is a virama glyph. + return LegacyFontMap( + name="synthetic-devanagari", + mapping={ + ord("f"): "ि", + ord("k"): "क", + ord("S"): "्", + ord("x"): "ष", + }, + provenance="synthetic-test-fixture", + ) + + +def test_non_vedic_iscii_round_trips_byte_identically(): + converted = iscii_to_unicode(_ISCII_NAME) + + assert converted.text == "रमेश शर्मा" + assert unicodedata.normalize("NFC", converted.text) == converted.text + assert unicode_to_iscii(converted.text) == _ISCII_NAME + + +def test_prefixed_devanagari_attribute_is_auto_detected(): + source = bytes.fromhex("ef 42") + _ISCII_NAME + + assert detect_legacy_encoding(source) == "iscii" + converted = convert_legacy_encoding(source) + + assert converted.text == "रमेश शर्मा" + assert converted.to_original_span(0, len(converted.text)) == (2, len(source)) + + +def test_nukta_and_virama_sequences_are_logically_ordered_and_lossless(): + # QA + virama: KA + NUKTA + HALANT in ISCII. + source = bytes.fromhex("b3 e9 e8") + converted = iscii_to_unicode(source) + + assert converted.text == "क़्" + assert [unicodedata.combining(char) for char in converted.text[1:]] == [7, 9] + assert unicode_to_iscii(converted.text) == source + + +@pytest.mark.parametrize( + ("source", "join_control"), + [ + (bytes.fromhex("b3 e8 e8"), "\u200c"), + (bytes.fromhex("b3 e8 e9"), "\u200d"), + ], +) +def test_contextual_virama_controls_round_trip(source, join_control): + converted = iscii_to_unicode(source) + + assert converted.text == f"क्{join_control}" + assert unicode_to_iscii(converted.text) == source + + +def test_visual_order_legacy_font_run_becomes_idempotent_unicode(): + source = b"fkSx" + font_map = _synthetic_font_map() + + assert detect_legacy_encoding(source, legacy_font_map=font_map) == "legacy-font" + converted = convert_legacy_encoding(source, legacy_font_map=font_map) + + assert converted.text == "क्षि" + assert normalize_indic_text(converted.text).text == converted.text + assert unicodedata.normalize("NFC", converted.text) == converted.text + + +def test_conversion_offset_map_returns_original_phi_byte_span(): + font_map = LegacyFontMap( + name="synthetic-name-glyphs", + mapping={ + ord("A"): "राम", + ord("B"): " ", + ord("C"): "शर्मा", + }, + ) + source = b"ID=ABC;" + converted = convert_legacy_encoding( + source, + encoding="legacy-font", + legacy_font_map=font_map, + ) + start = converted.text.index("राम शर्मा") + end = start + len("राम शर्मा") + + original_start, original_end = converted.to_original_span(start, end) + + assert source[original_start:original_end] == b"ABC" + assert converted.offset_map.to_converted_span(3, 6) == (3, end) + + +def test_pure_latin_clinical_note_is_not_auto_converted(): + note = "Patient John Smith visited clinic for BP follow-up." + font_map = _synthetic_font_map() + + assert detect_legacy_encoding(note, legacy_font_map=font_map) == "unicode" + converted = convert_legacy_encoding(note, legacy_font_map=font_map) + + assert converted.text == note + assert converted.encoding == "unicode" + assert not converted.changed + + +def test_pure_latin_note_is_unchanged_with_dense_user_map(): + note = "Patient John Smith visited clinic for BP follow-up." + dense_map = LegacyFontMap( + name="dense-synthetic-map", + mapping={ + ord(char): "क" + for char in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + }, + ) + + assert detect_legacy_encoding(note, legacy_font_map=dense_map) == "unicode" + assert convert_legacy_encoding(note, legacy_font_map=dense_map).text == note + + +def test_valid_utf8_devanagari_bytes_are_not_misdetected_as_iscii(): + source = "रमेश शर्मा".encode() + + assert detect_legacy_encoding(source) == "unicode" + converted = convert_legacy_encoding(source) + + assert converted.text == "रमेश शर्मा" + assert converted.encoding == "unicode" + + +def test_c1_control_bytes_prevent_false_iscii_detection(): + text = "\x80\xa0\xb3" + + assert detect_legacy_encoding(text) == "unicode" + assert convert_legacy_encoding(text).text == text + + +@pytest.mark.parametrize( + "text", + ["À côté", "ÀÉÎ", "Crème brûlée", "José", "Patient naïB cell"], +) +def test_latin1_clinical_text_is_not_misdetected_as_iscii(text): + assert detect_legacy_encoding(text) == "unicode" + assert convert_legacy_encoding(text).text == text + + +def test_detection_normalizer_routes_iscii_and_maps_span_to_source(): + # Latin-1 is the compatibility representation used when an application has + # already placed the raw legacy bytes in a Python string. + original = _ISCII_NAME.decode("latin-1") + normalized = normalize_for_pii_detection(original) + + assert normalized.text == "रमेश शर्मा" + assert normalized.legacy_encoding == "iscii" + assert normalized.converted_legacy_bytes == len(_ISCII_NAME) - 1 + assert normalized.remap_span(0, len(normalized.text)) == (0, len(original)) + assert normalized.to_metadata()["legacy_encoding"] == "iscii" + + +def test_pipeline_converts_iscii_before_script_routing(): + original = _ISCII_NAME.decode("latin-1") + pipeline = Pipeline(lang="auto") + + document = pipeline.stage1_normalize(original) + route = pipeline.stage2_language_script(document.normalized_text) + + assert document.normalized_text == "रमेश शर्मा" + assert document.offset_map.normalized_span_to_original_offsets( + 0, len(document.normalized_text) + ) == (0, len(original)) + assert document.metadata["legacy_encoding"] == { + "encoding": "iscii", + "changed": True, + "converted_bytes": len(_ISCII_NAME) - 1, + } + assert route.script == "Devanagari" + assert route.lang == "hi" + + +def test_detection_normalizer_preserves_attached_indic_marks(): + normalized = normalize_for_pii_detection("क़्") + + assert normalized.text == "क़्" + assert normalized.stripped_combining_marks == 0 + assert not normalized.changed + + +def test_detection_normalizer_strips_standalone_indic_mark(): + normalized = normalize_for_pii_detection("़Patient") + + assert normalized.text == "Patient" + assert normalized.stripped_combining_marks == 1 + + +def test_user_mapping_file_loads_without_bundled_font_data(tmp_path): + path = tmp_path / "custom-font.json" + path.write_text( + json.dumps( + { + "name": "hospital-archive-font", + "provenance": "licensed by data owner", + "mapping": {"0x41": "र", "B": "ा", "67": "म"}, + } + ), + encoding="utf-8", + ) + + font_map = LegacyFontMap.from_file(path) + + assert font_map.name == "hospital-archive-font" + assert dict(font_map.mapping) == {0x41: "र", ord("B"): "ा", 67: "म"} + assert font_map.provenance == "licensed by data owner" + + +def test_invalid_iscii_and_invalid_mapping_are_rejected(tmp_path): + with pytest.raises(UnicodeDecodeError): + iscii_to_unicode(b"\xff") + + path = tmp_path / "bad.txt" + path.write_text("{}", encoding="utf-8") + with pytest.raises(ValueError, match="JSON, YAML, or YML"): + LegacyFontMap.from_file(path) + + +def test_legacy_map_file_and_expansion_are_bounded(tmp_path): + oversized = tmp_path / "oversized.json" + oversized.write_bytes(b" " * (1024 * 1024 + 1)) + + with pytest.raises(ValueError, match="1 MiB"): + LegacyFontMap.from_file(oversized) + with pytest.raises(ValueError, match="at most 64"): + LegacyFontMap(name="expanding", mapping={65: "क" * 65}) + + +def test_unicode_replacement_offsets_track_consumed_invalid_bytes(): + source = b"A\xe2\x82B\xff" + + converted = convert_legacy_encoding( + source, + encoding="unicode", + errors="replace", + ) + + assert converted.text == source.decode("utf-8", errors="replace") + assert converted.offset_map.converted_to_original_spans == ( + (0, 1), + (1, 3), + (3, 4), + (4, 5), + ) + + +def test_mapping_provenance_documents_public_standard_and_license(): + assert "IS 13194:1991" in ISCII_MAPPING_PROVENANCE + assert "Apache-2.0" in ISCII_MAPPING_PROVENANCE