Files
PTSRaiAgent/app/core/privacy.py
ElebobaandClaude Opus 5.5 b4731fb1d6
CI / ruff · pytest (pull_request) Successful in 22s
chore(core): stage 1 contracts — dialog, API schemas + fixtures, integrations, monitoring, modules
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>
2026-10-03 15:22:25 +03:00

444 lines
19 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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]