"""PII scrubbing and pseudonymisation (issue #2, guardrail #1). Why this module exists ---------------------- Messages from patients are health data — a special category of personal data under 152-FZ art. 10 — and are covered by medical secrecy (323-FZ art. 13). Guardrail #1: **only text that passed `scrub()` may reach the LLM** (`app/llm/gateway.py` accepts `ScrubbedText` only), and only scrubbed text is stored in `messages.text_scrubbed`. Direct identifiers live solely in the identity DB; the clinical DB is linked to it by `pseudo_id()` — an HMAC token that cannot be reversed without `PSEUDO_KEY`. Design ------ `scrub()` is a deterministic rule-based pipeline (no ML, works offline, explainable to the jury): ordered regex passes replace structured identifiers with bracketed placeholders, then a token pass finds person names using a dictionary (`config/privacy/names_ru.yaml`) plus morphology heuristics. The order matters: longer / more specific number formats go first so that, e.g., a card number is not partially eaten by the phone pattern. Placeholders are Latin and bracketed, so no later pass (which looks for digits or Cyrillic) can match them — this is what makes the function idempotent; on top of that the pipeline is iterated to a fixed point. The trade-off is deliberate: over-scrubbing a city or a capitalised surname-like word is cheap, leaking an identifier to an external model is not. At the same time ordinary clinical text (doses, scale scores, durations, times, lowercase complaints) must survive intact, because the LLM extracts symptom data from it — this is covered by false-positive tests in `tests/test_privacy.py`. """ from __future__ import annotations import hashlib import hmac import re from collections.abc import Callable from dataclasses import dataclass from functools import lru_cache from pathlib import Path import yaml from pydantic import BaseModel from app.config import get_settings from app.core.types import ScrubbedText # --------------------------------------------------------------------------------------------------- # Regex passes for structured identifiers # --------------------------------------------------------------------------------------------------- _SOCIAL = re.compile( r"(?:https?://)?(?:www\.)?" r"(?:t\.me|telegram\.me|vk\.com|vk\.ru|ok\.ru|instagram\.com|facebook\.com|max\.ru)/[^\s]+", re.IGNORECASE, ) _URL = re.compile( r"(?:https?://|www\.)[^\s]+" r"|\b[a-z0-9][a-z0-9-]*(?:\.[a-z0-9-]+)*\.(?:ru|com|рф|org|net|su|io|me|info)\b(?:/[^\s]*)?", re.IGNORECASE, ) _EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+") _HANDLE = re.compile(r"(? bool: """Official SNILS control number: weighted sum of the first 9 digits (weights 9..1) mod 101.""" total = sum(int(d) * w for d, w in zip(digits[:9], range(9, 0, -1), strict=True)) control = total % 101 if control == 100: control = 0 return control == int(digits[9:]) def _phone_or_snils(m: re.Match[str]) -> str: # A contiguous 11-digit "8XXXXXXXXXX" could also be a SNILS; the checksum disambiguates. raw = m.group(0) if raw.isdigit() and len(raw) == 11 and _snils_checksum_ok(raw): return "[SNILS]" return "[PHONE]" def _url_repl(placeholder: str) -> Callable[[re.Match[str]], str]: def repl(m: re.Match[str]) -> str: # Keep sentence punctuation that a greedy `\S+` swallowed: "см. t.me/x." → "см. [HANDLE]." raw = m.group(0) stripped = raw.rstrip(".,;:!?)»\"'") return placeholder + raw[len(stripped) :] return repl _REGEX_PASSES: list[tuple[re.Pattern[str], str | Callable[[re.Match[str]], str]]] = [ (_SOCIAL, _url_repl("[HANDLE]")), (_EMAIL, "[EMAIL]"), (_URL, _url_repl("[URL]")), (_HANDLE, "[HANDLE]"), (_CARD, "[CARD]"), (_SNILS_FMT, "[SNILS]"), (_PHONE_FULL, _phone_or_snils), (_PHONE_MOBILE, "[PHONE]"), (_PHONE_LOCAL, "[PHONE]"), (_PASSPORT_KW, "[PASSPORT]"), (_PASSPORT_BARE, "[PASSPORT]"), (_SNILS_PLAIN, "[SNILS]"), (_INN, "[INN]"), *((p, "[DATE]") for p in _DATES), (_BIRTH_YEAR, lambda m: m.group(1) + "[DATE]"), *((p, "[ADDRESS]") for p in _ADDRESSES), *((p, "[NAME]") for p in _INITIALS), ] # --------------------------------------------------------------------------------------------------- # Person names: dictionary + morphology # --------------------------------------------------------------------------------------------------- class NameLexicon(BaseModel): """Schema of `config/privacy/names_ru.yaml` (values may be extended by the clinical zone).""" names: list[str] ambiguous_names: list[str] = [] ambiguous_forms: list[str] = [] surname_stopwords: list[str] = [] @dataclass(frozen=True) class _NameIndex: forms: frozenset[str] # every known name form, normalised (lowercase, ё→е) ambiguous: frozenset[str] # forms that are also ordinary Russian words surname_stop: frozenset[str] def _norm(word: str) -> str: return word.lower().replace("ё", "е") _HUSHING = set("жшщчц") _VELAR_HUSH = set("гкхжшщч") def _declensions(name: str) -> set[str]: """Generate singular case forms of a Russian first name by its ending. Covers the regular paradigms (Иван, Алексей, Дмитрий, Игорь, Ирина, Мария, Таня, Саша); irregular ones (fleeting vowels: Павел→Павла) are listed explicitly in the YAML. """ n = _norm(name) forms = {n} if n.endswith("ий"): stem = n[:-2] forms |= {stem + e for e in ("ия", "ию", "ием", "ии")} elif n.endswith(("ей", "ай", "ой")): stem = n[:-1] forms |= {stem + e for e in ("я", "ю", "ем", "е")} elif n.endswith("ия"): stem = n[:-1] forms |= {stem + e for e in ("и", "ю", "ей")} elif n.endswith("а"): stem = n[:-1] gen = "и" if stem[-1] in _VELAR_HUSH else "ы" ins = "ей" if stem[-1] in _HUSHING else "ой" forms |= {stem + gen, stem + "е", stem + "у", stem + ins} elif n.endswith("я"): stem = n[:-1] forms |= {stem + e for e in ("и", "е", "ю", "ей")} elif n.endswith("ь"): stem = n[:-1] forms |= {stem + e for e in ("я", "ю", "ем", "е", "и")} else: ins = "ем" if n[-1] in _HUSHING else "ом" forms |= {n + "а", n + "у", n + ins, n + "е"} return forms def _lexicon_path() -> Path: return get_settings().config_dir / "privacy" / "names_ru.yaml" @lru_cache def _name_index() -> _NameIndex: lex = NameLexicon.model_validate(yaml.safe_load(_lexicon_path().read_text(encoding="utf-8"))) forms: set[str] = set() ambiguous = {_norm(f) for f in lex.ambiguous_forms} ambiguous_names = {_norm(n) for n in lex.ambiguous_names} for line in lex.names: head, *extra = line.split() name_forms = _declensions(head) | {_norm(e) for e in extra} forms |= name_forms if _norm(head) in ambiguous_names: ambiguous |= name_forms stop = frozenset(_norm(w) for w in lex.surname_stopwords) return _NameIndex(frozenset(forms), frozenset(ambiguous), stop) _TOKEN = re.compile(r"[А-ЯЁа-яё]+(?:-[А-ЯЁа-яё]+)*") _CAPITALISED = re.compile(r"^[А-ЯЁ][а-яё]+(?:-[А-ЯЁ][а-яё]+)*$") # Patronymics: -ович/-евич/-ьич (+ case endings), -овна/-евна/-ична (+ case endings). _PATRONYMIC = re.compile( r"^[а-яё]+(?:(?:(?:ов|ев|ёв)ич|ьич|инич)(?:а|у|ем|е)?|(?:(?:ов|ев|ёв)н|ичн)(?:а|ы|е|у|ой))$" ) # Strong surname suffixes: -ов/-ев/-ёв, -ский/-цкий, -енко with case endings. _SURNAME_STRONG = re.compile( r"^[а-яё]{2,}(?:(?:ов|ев|ёв)(?:а|у|ым|ой|ою|е|ых)?|(?:ск|цк)(?:ий|ая|ого|ому|им|ой|ую|ом)|енко)$" ) # Weak: -ин/-ын are also drug names (Феварин, Глицин), so they count only next to other name evidence. _SURNAME_WEAK = re.compile(r"^[а-яё]{2,}(?:ин|ын)(?:а|у|ым|ой|е)?$") _TRIGGERS = frozenset({"зовут", "звать", "имя", "фамилия", "фио", "отчество"}) _MAX_NAME_TOKENS = 4 @dataclass class _Tok: text: str start: int end: int low: str # normalised cap: bool initial: bool # starts a sentence def _sentence_initial(text: str, pos: int) -> bool: i = pos - 1 while i >= 0 and text[i] in " \t«\"'(—–-": i -= 1 return i < 0 or text[i] in ".!?…\n" def _is_surname(tok: _Tok, idx: _NameIndex, *, weak: bool) -> bool: if tok.low in idx.surname_stop: return False return bool(_SURNAME_STRONG.match(tok.low) or (weak and _SURNAME_WEAK.match(tok.low))) def _is_patronymic(tok: _Tok) -> bool: return tok.cap and bool(_PATRONYMIC.match(tok.low)) def _dict_hit(tok: _Tok, idx: _NameIndex) -> tuple[bool, bool]: """(is a known name form, is ambiguous). Hyphenated names match if any part is a known name.""" if tok.low in idx.forms: return True, tok.low in idx.ambiguous parts = tok.low.split("-") if len(parts) > 1 and tok.cap and any(p in idx.forms and p not in idx.ambiguous for p in parts): return True, False return False, False def _name_spans(text: str) -> list[tuple[int, int]]: idx = _name_index() toks = [ _Tok(m.group(0), m.start(), m.end(), _norm(m.group(0)), m.group(0)[0].isupper(), False) for m in _TOKEN.finditer(text) ] for t in toks: t.initial = _sentence_initial(text, t.start) def adjacent(a: _Tok, b: _Tok) -> bool: gap = text[a.end : b.start] return gap.isspace() and "\n" not in gap def after_trigger(i: int) -> bool: if i == 0 or toks[i - 1].low not in _TRIGGERS: return False return re.fullmatch(r"\s*[:—–-]?\s*", text[toks[i - 1].end : toks[i].start]) is not None def strong_neighbour(i: int) -> bool: if i + 1 < len(toks) and adjacent(toks[i], toks[i + 1]): nxt = toks[i + 1] return nxt.cap and (_is_patronymic(nxt) or _is_surname(nxt, idx, weak=True)) return False def is_name_anchor(i: int, t: _Tok) -> bool: known, ambiguous = _dict_hit(t, idx) if after_trigger(i): # "меня зовут Алексей" / "меня зовут алексей" (lowercase only if it is a known name). return bool(_CAPITALISED.match(t.text)) or known if known and not ambiguous: return t.cap or len(t.low) >= 3 if known and ambiguous: # "Вера" mid-sentence or before a patronymic is a name; sentence-initial "Вера в себя" is not. return t.cap and (not t.initial or strong_neighbour(i)) return _is_patronymic(t) anchors: set[int] = set() for i, t in enumerate(toks): cap_word = bool(_CAPITALISED.match(t.text)) if is_name_anchor(i, t): anchors.add(i) elif cap_word and not t.initial and _is_surname(t, idx, weak=False): # Mid-sentence capitalised -ов/-ская word: "был у Петрова". May over-scrub a city (Ростов). anchors.add(i) elif ( cap_word and not t.initial and i + 1 < len(toks) and adjacent(t, toks[i + 1]) and _CAPITALISED.match(toks[i + 1].text) ): # Two capitalised words mid-sentence: "встретил Петрова Ивана". anchors.update({i, i + 1}) spans: list[tuple[int, int]] = [] covered: set[int] = set() for i in sorted(anchors): if i in covered: continue lo = hi = i lower_anchor = not toks[i].cap # Extend right over the rest of the full name (surname, patronymic). while hi + 1 < len(toks) and hi - lo + 1 < _MAX_NAME_TOKENS and adjacent(toks[hi], toks[hi + 1]): nxt = toks[hi + 1] capitalised = nxt.cap and _CAPITALISED.match(nxt.text) # "алексей петров" typed in lowercase: extend only over a clearly surname-like word. lower_tail = lower_anchor and (_is_surname(nxt, idx, weak=False) or _PATRONYMIC.match(nxt.low)) if capitalised or (hi + 1) in anchors or lower_tail: hi += 1 else: break # Extend left only over a surname-looking word ("Иванов Иван"), never over "Сегодня Иван". while lo > 0 and hi - lo + 1 < _MAX_NAME_TOKENS and adjacent(toks[lo - 1], toks[lo]): prv = toks[lo - 1] if prv.cap and _CAPITALISED.match(prv.text) and _is_surname(prv, idx, weak=True): lo -= 1 else: break covered.update(range(lo, hi + 1)) spans.append((toks[lo].start, toks[hi].end)) return spans def _scrub_names(text: str) -> str: out: list[str] = [] pos = 0 for start, end in _name_spans(text): out.append(text[pos:start]) out.append("[NAME]") pos = end out.append(text[pos:]) # "[NAME] [NAME]" produced by separate spans is one person in practice. return re.sub(r"\[NAME\](?:[ \t]+\[NAME\])+", "[NAME]", "".join(out)) def _scrub_once(text: str) -> str: for pattern, repl in _REGEX_PASSES: text = pattern.sub(repl, text) return _scrub_names(text) _MAX_PASSES = 5 def scrub(text: str) -> ScrubbedText: """Replace personal data (phones, emails, SNILS, names, addresses...) with placeholders. Placeholders: [PHONE] [EMAIL] [SNILS] [PASSPORT] [INN] [CARD] [DATE] [ADDRESS] [NAME] [URL] [HANDLE]. Idempotent: `scrub(scrub(x)) == scrub(x)`. """ # Removing one identifier can expose another pattern boundary; iterate to a fixed point. for _ in range(_MAX_PASSES): new = _scrub_once(text) if new == text: break text = new return ScrubbedText(text) def pseudo_id(snils: str, key: str | None = None) -> str: """HMAC-SHA256(PSEUDO_KEY, normalised SNILS) as 32 hex chars. The token is the only link between the identity and clinical databases. SNILS is normalised to 11 digits (spaces and dashes removed); anything else raises `ValueError`. The checksum is not validated: demo data is synthetic. """ digits = re.sub(r"[\s\-]", "", snils) if len(digits) != 11 or not digits.isascii() or not digits.isdigit(): raise ValueError("SNILS must contain exactly 11 digits") secret = (key if key is not None else get_settings().pseudo_key).encode("utf-8") return hmac.new(secret, digits.encode("ascii"), hashlib.sha256).hexdigest()[:32]