CI / ruff · pytest (pull_request) Successful in 22s
Freeze interfaces between lanes A (bot), B (backend), C (dashboard) so three developers work in parallel without touching each other's files: dialog Event/Reply + engine registry with echo stub, integration protocols with in-memory defaults, MonitoringService with noop stub, /api/v1 schemas with fixture-backed routes, frozen main.py wiring, config/modules.yaml + due_scales(), DAR-5, tzdata, Andrey trajectory, lanes in WORKFLOW, ADR-010. Refs #34 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
444 lines
19 KiB
Python
444 lines
19 KiB
Python
"""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"(?<![\w@.])@[A-Za-z0-9_]{3,32}\b")
|
||
|
||
# 16-digit bank card, contiguous or in groups of 4.
|
||
_CARD = re.compile(r"(?<!\d)\d{4}(?:[ \-]?\d{4}){3}(?!\d)")
|
||
# SNILS in its official layout: 123-456-789 01 (dashes or spaces).
|
||
_SNILS_FMT = re.compile(r"(?<!\d)\d{3}[ \-]\d{3}[ \-]\d{3}[ \-]\d{2}(?!\d)")
|
||
# Russian phones: +7 / 8 followed by 10 digits with any separators, e.g. +7 (916) 123-45-67, 8-800-2000-122.
|
||
_PHONE_FULL = re.compile(r"(?<![\d\w+])(?:\+\s?7|8)(?:[ \-()]{0,3}\d){10}(?!\d)")
|
||
# Mobile number without the country prefix: 916 123-45-67, (916)1234567.
|
||
_PHONE_MOBILE = re.compile(r"(?<![\d\w])\(?9\d{2}\)?[ \-]?\d{3}[ \-]?\d{2}[ \-]?\d{2}(?!\d)")
|
||
# Local 7-digit number with mandatory dashes (123-45-67); dashes keep "4-6 часов" safe.
|
||
_PHONE_LOCAL = re.compile(r"(?<![\d\w\-])\d{3}-\d{2}-\d{2}(?![\d\-])")
|
||
# Passport: "серия 4510 номер 123456" or a bare "45 10 123456" / "4510 123456".
|
||
_PASSPORT_KW = re.compile(r"серия\s*:?\s*\d{2}\s?\d{2}\s*,?\s*(?:(?:№|номер|n)\s*:?\s*)?\d{6}(?!\d)", re.IGNORECASE)
|
||
_PASSPORT_BARE = re.compile(r"(?<!\d)\d{2}\s?\d{2}\s(?:№\s*)?\d{6}(?!\d)")
|
||
_SNILS_PLAIN = re.compile(r"(?<!\d)\d{11}(?!\d)")
|
||
_INN = re.compile(r"(?<!\d)(?:\d{12}|\d{10})(?!\d)")
|
||
|
||
_MONTHS = "января|февраля|марта|апреля|мая|июня|июля|августа|сентября|октября|ноября|декабря"
|
||
_DAY = r"(?:0?[1-9]|[12]\d|3[01])"
|
||
_MON = r"(?:0?[1-9]|1[0-2])"
|
||
_DATES = [
|
||
# 12.03.1985, 1/3/1985, 12-03-1985 — a full date with a 4-digit year.
|
||
re.compile(rf"(?<![\d.]){_DAY}[./\-]{_MON}[./\-](?:19|20)\d{{2}}(?!\d)"),
|
||
# 12.03.85 — dots only, so ranges and fractions stay intact.
|
||
re.compile(r"(?<![\d.])(?:0[1-9]|[12]\d|3[01])\.(?:0[1-9]|1[0-2])\.\d{2}(?![\d.])"),
|
||
# ISO 1985-03-12.
|
||
re.compile(r"(?<!\d)(?:19|20)\d{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12]\d|3[01])(?!\d)"),
|
||
# 12 марта 1985 (г.|года). A day+month without a year ("12 марта") is not identifying and kept.
|
||
re.compile(rf"(?<!\d){_DAY}\s+(?:{_MONTHS})\s+(?:19|20)\d{{2}}(?:\s*(?:г\.|года\b|г\b))?", re.IGNORECASE),
|
||
# 1985 г.р. / 1985 года рождения.
|
||
re.compile(r"(?<!\d)(?:19|20)\d{2}\s*(?:г\.\s*р\.?|г\.р\.?|года\s+рождения)", re.IGNORECASE),
|
||
]
|
||
# "родился в 1985", "дата рождения: 1985" — replace only the year.
|
||
_BIRTH_YEAR = re.compile(r"((?:родил(?:ся|ась)|рождения)\s*:?\s*(?:в\s+)?)((?:19|20)\d{2})(?!\d)", re.IGNORECASE)
|
||
|
||
_STREET_TYPE = (
|
||
r"(?:ул\.|улица|улице|улицу|улицы|пр-т|пр-кт|просп\.|проспект|проспекте|пер\.|переулок|переулке"
|
||
r"|б-р|бульвар|бульваре|ш\.|шоссе|наб\.|набережная|набережной|пл\.|площадь|площади|мкр\.?|микрорайон"
|
||
r"|проезд|тупик)"
|
||
)
|
||
# Street name must start with a capital or a number ("ул. 8 Марта", "1-я Тверская"): this is what keeps
|
||
# "на улице страшно" or "проезд оплатил" untouched.
|
||
_STREET_NAME = r"(?:\d+-?(?:я|й|го|ая|ий)?\s+)?[А-ЯЁ][А-ЯЁа-яё\-]+(?:\s+[А-ЯЁ][а-яё\-]+)?"
|
||
_CITY = r"(?:(?<![\d\w.])(?<!\d\s)(?:г\.|город)\s*[А-ЯЁ][а-яё\-]+,?\s*)?"
|
||
_HOUSE_TAIL = (
|
||
r"(?:\s*,?\s*(?:(?:д\.|дом)\s*)?\d+[а-яА-Я]?(?:/\d+)?)?"
|
||
r"(?:\s*,?\s*(?:корп\.|корпус|к\.|стр\.|строение)\s*\d+)?"
|
||
r"(?:\s*,?\s*(?:кв\.|квартира|кв)\s*\d+)?"
|
||
)
|
||
_ADDRESSES = [
|
||
re.compile(rf"{_CITY}(?<![\w\-]){_STREET_TYPE}\s*{_STREET_NAME}{_HOUSE_TAIL}"),
|
||
# "Тверская улица, 5", "Ленинский проспект".
|
||
re.compile(
|
||
rf"{_CITY}\b[А-ЯЁ][а-яё]+(?:ая|ий|ый|ой)\s+(?:улица|проспект|переулок|бульвар|шоссе|набережная|площадь)"
|
||
rf"{_HOUSE_TAIL}"
|
||
),
|
||
# A bare "д. 5, кв. 12" / "дом 5". `(?<![\w.])` keeps "и т.д. 5" safe.
|
||
re.compile(
|
||
r"(?<![\w.])(?:д\.|дом)\s*№?\s*\d+[а-я]?(?:/\d+)?(?:\s*,?\s*(?:корп\.|к\.|стр\.)\s*\d+)?"
|
||
r"(?:\s*,?\s*(?:кв\.|квартира)\s*\d+)?(?!\d)"
|
||
),
|
||
re.compile(r"(?<![\w.])(?:кв\.|квартира)\s*№?\s*\d+(?!\d)"),
|
||
# "г. Москва" on its own; "(?<!\d\s)" keeps "в 2010 г. Мне стало хуже" safe.
|
||
re.compile(r"(?<![\d\w.])(?<!\d\s)г\.\s*[А-ЯЁ][а-яё]{2,}(?:-[А-ЯЁ][а-яё]+)?"),
|
||
]
|
||
|
||
# Surname + two initials ("Смирнова Е. А.") or initials + surname ("Е.А. Смирнова").
|
||
_INITIALS = [
|
||
re.compile(r"\b[А-ЯЁ][а-яё]+(?:-[А-ЯЁ][а-яё]+)?\s+[А-ЯЁ]\.\s?[А-ЯЁ]\.(?!\w)"),
|
||
re.compile(r"(?<![\w.])[А-ЯЁ]\.\s?[А-ЯЁ]\.\s?[А-ЯЁ][а-яё]+(?:-[А-ЯЁ][а-яё]+)?"),
|
||
]
|
||
|
||
|
||
def _snils_checksum_ok(digits: str) -> 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]
|