feat(etl): нормализация к справочнику (код→точное→эмбеддинги→fuzzy), валидация и конвейер в SQLite

etl/normalize/__init__.py

@@ -0,0 +1,5 @@
+"""Нормализация извлечённых позиций к справочнику услуг."""
+
+from etl.normalize.matcher import Matcher
+
+__all__ = ["Matcher"]

etl/normalize/embedding.py

@@ -0,0 +1,72 @@
+"""Эмбеддинги через Gemini API (SDK google-genai) — без локального torch.
+
+Лёгкая зависимость вместо sentence-transformers: контейнер на маленьком VPS не
+раздувается и не держит модель в RAM. Размерность 768 выбрана, чтобы вектор лез в
+HNSW-индекс pgvector. Требуется переменная окружения GEMINI_API_KEY.
+"""
+
+from __future__ import annotations
+
+import os
+
+import numpy as np
+
+EMBED_DIM = 768
+
+
+class GeminiEmbedder:
+    """Обёртка над Gemini-эмбеддингами с интерфейсом `.encode()` (как у sentence-transformers)."""
+
+    def __init__(
+        self,
+        model: str = "gemini-embedding-001",
+        dim: int = EMBED_DIM,
+        api_key: str | None = None,
+        batch: int = 100,
+    ):
+        from google import genai
+
+        self._client = genai.Client(api_key=api_key or os.environ["GEMINI_API_KEY"])
+        self.model = model
+        self.dim = dim
+        self.batch = batch
+
+    def encode(
+        self,
+        texts,
+        normalize_embeddings: bool = True,
+        task_type: str = "RETRIEVAL_DOCUMENT",
+        show_progress_bar: bool = False,
+    ):
+        """Векторизовать список строк. Возвращает numpy-массив (n, dim) float32.
+
+        Для справочника используем RETRIEVAL_DOCUMENT, для запросов —
+        RETRIEVAL_QUERY: так несвязанные названия расходятся сильнее, чем при
+        SEMANTIC_SIMILARITY, и порог отсечения работает надёжнее.
+        """
+        import time
+
+        from google.genai import types
+
+        items = [t if t and t.strip() else " " for t in texts]
+        config = types.EmbedContentConfig(task_type=task_type, output_dimensionality=self.dim)
+        vectors: list[list[float]] = []
+        for start in range(0, len(items), self.batch):
+            chunk = items[start : start + self.batch]
+            for attempt in range(8):
+                try:
+                    response = self._client.models.embed_content(
+                        model=self.model, contents=chunk, config=config
+                    )
+                    vectors.extend(embedding.values for embedding in response.embeddings)
+                    break
+                except Exception as exc:  # пауза и повтор при превышении квоты (429)
+                    if ("RESOURCE_EXHAUSTED" in str(exc) or "429" in str(exc)) and attempt < 7:
+                        time.sleep(20)
+                        continue
+                    raise
+
+        array = np.array(vectors, dtype=np.float32)
+        if normalize_embeddings:
+            array = array / np.clip(np.linalg.norm(array, axis=1, keepdims=True), 1e-9, None)
+        return array

etl/normalize/matcher.py

@@ -0,0 +1,135 @@
+"""Сопоставление извлечённых позиций с эталонным справочником услуг.
+
+Каскад по убыванию надёжности:
+  1. код тарификатора — точное совпадение, если клиника его указала;
+  2. точное совпадение нормализованного названия;
+  3. эмбеддинги (многоязычная модель) — семантическая близость названий;
+  4. нечёткое сравнение (RapidFuzz) как запасной сигнал;
+  5. иначе — очередь ручной разметки (unmatched).
+
+В справочнике нет синонимов, поэтому шаги 1 и 3 несут основную нагрузку. Коды у
+клиник встречаются с лишним хвостом («A02.020.000.2»), поэтому сравниваем по
+канонической части кода тарификатора.
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from rapidfuzz import fuzz, process  # noqa: E402
+
+from contracts.models import MatchResult  # noqa: E402
+from etl.dictionary import Service, normalize_name  # noqa: E402
+
+# Каноническая часть кода тарификатора: буква + три группы цифр.
+_CODE_CORE_RE = re.compile(r"[A-ZА-Я]\d{2}\.\d{3}\.\d{3}")
+
+
+class Matcher:
+    """Готовит индексы справочника и сопоставляет с ним сырые названия услуг."""
+
+    def __init__(
+        self,
+        services: list[Service],
+        embedder=None,
+        emb_threshold: float = 0.70,
+        fuzzy_threshold: int = 88,
+        dict_emb=None,
+    ):
+        self.services = services
+        self.by_code = {s.tarificator_code: s for s in services if s.tarificator_code}
+        self.by_norm: dict[str, Service] = {}
+        for s in services:
+            self.by_norm.setdefault(s.name_norm, s)
+        self.names_norm = [s.name_norm for s in services]
+        self.emb_threshold = emb_threshold
+        self.fuzzy_threshold = fuzzy_threshold
+        self.embedder = embedder
+        # Готовые эмбеддинги справочника (dict_emb) переиспользуются между прогонами:
+        # при догрузке нового прайса не нужно заново векторизовать тысячи услуг.
+        self.dict_emb = dict_emb
+        if embedder is not None and self.dict_emb is None:
+            self.dict_emb = embedder.encode(
+                [s.name_ru for s in services],
+                normalize_embeddings=True,
+                task_type="RETRIEVAL_DOCUMENT",
+            )
+
+    def _match_by_code(self, code: str | None) -> Service | None:
+        if not code:
+            return None
+        m = _CODE_CORE_RE.search(code)
+        return self.by_code.get(m.group(0)) if m else None
+
+    def _match_by_fuzzy(self, name_norm: str) -> MatchResult:
+        best = process.extractOne(name_norm, self.names_norm, scorer=fuzz.token_set_ratio)
+        if best and best[1] >= self.fuzzy_threshold:
+            return MatchResult(
+                service_id=self.services[best[2]].service_id,
+                method="fuzzy",
+                confidence=best[1] / 100,
+            )
+        return MatchResult()
+
+    def match_batch(self, names: list[str], codes: list[str | None]) -> list[MatchResult]:
+        """Сопоставить пачку позиций. Эмбеддинги считаются разом — так быстрее."""
+        results: list[MatchResult | None] = [None] * len(names)
+        pending_idx: list[int] = []
+        pending_norm: list[str] = []
+
+        for i, (name, code) in enumerate(zip(names, codes, strict=True)):
+            service = self._match_by_code(code)
+            if service:
+                results[i] = MatchResult(
+                    service_id=service.service_id, method="code", confidence=1.0
+                )
+                continue
+            name_norm = normalize_name(name)
+            exact = self.by_norm.get(name_norm)
+            if exact:
+                results[i] = MatchResult(
+                    service_id=exact.service_id, method="exact", confidence=1.0
+                )
+                continue
+            pending_idx.append(i)
+            pending_norm.append(name_norm)
+
+        if pending_norm and self.embedder is not None and self.dict_emb is not None:
+            # Дедуп: одинаковые названия эмбеддим один раз (экономит запросы и квоту API).
+            pending_names = [names[i] for i in pending_idx]
+            unique: dict[str, int] = {}
+            for nm in pending_names:
+                unique.setdefault(nm, len(unique))
+            query = self.embedder.encode(
+                list(unique), normalize_embeddings=True, task_type="RETRIEVAL_QUERY"
+            )
+            sims = query @ self.dict_emb.T  # косинус по нормированным векторам
+            best_idx = sims.argmax(axis=1)
+            best_score = sims.max(axis=1)
+            for k, i in enumerate(pending_idx):
+                u = unique[pending_names[k]]
+                suggested = self.services[int(best_idx[u])].service_id
+                score = float(best_score[u])
+                if score >= self.emb_threshold:
+                    results[i] = MatchResult(
+                        service_id=suggested,
+                        method="embedding",
+                        confidence=score,
+                        suggested_service_id=suggested,
+                        suggested_score=score,
+                    )
+                else:
+                    # Не дотянул до порога — в очередь, но кандидата сохраняем оператору.
+                    fallback = self._match_by_fuzzy(pending_norm[k])
+                    fallback.suggested_service_id = suggested
+                    fallback.suggested_score = score
+                    results[i] = fallback
+        else:
+            for k, i in enumerate(pending_idx):
+                results[i] = self._match_by_fuzzy(pending_norm[k])
+
+        return [r if r is not None else MatchResult() for r in results]

etl/pipeline.py

@@ -0,0 +1,255 @@
+"""Конвейер обработки архива: извлечение → нормализация → валидация → SQLite.
+
+Для MVP данные складываются в SQLite (разворачивается без Docker), а боевая схема
+PostgreSQL + pgvector лежит в db/migrations. Здесь же — дедупликация партнёров,
+версионирование цен (последний прайс активен, старые архивируются) и сбор отчёта о
+качестве для дашборда и сдачи.
+
+Два входа: `run()` собирает базу с нуля (полная пересборка), `ingest()` догружает
+новые прайсы в уже собранную базу (append) — на нём держится приём ZIP через интерфейс.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import sys
+import uuid
+from collections import Counter
+from datetime import date
+from pathlib import Path
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from etl.dictionary import load_dictionary, normalize_name
+from etl.extractors import extract
+from etl.normalize import Matcher
+from etl.validate import validate_item
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS service (
+    service_id TEXT PRIMARY KEY, specialty TEXT, name_ru TEXT, name_norm TEXT, tarificator_code TEXT
+);
+CREATE TABLE IF NOT EXISTS partner (
+    partner_id TEXT PRIMARY KEY, name TEXT, name_norm TEXT UNIQUE, city TEXT
+);
+CREATE TABLE IF NOT EXISTS price_document (
+    doc_id TEXT PRIMARY KEY, partner_id TEXT, file_name TEXT, file_format TEXT,
+    effective_date TEXT, parse_status TEXT, rows_count INTEGER
+);
+CREATE TABLE IF NOT EXISTS price_item (
+    item_id INTEGER PRIMARY KEY AUTOINCREMENT, doc_id TEXT, partner_id TEXT,
+    service_name_raw TEXT, service_code_source TEXT, service_id TEXT,
+    prices TEXT, price_resident REAL, price_nonresident REAL, unit TEXT,
+    effective_date TEXT, map_method TEXT, map_confidence REAL, status TEXT, is_active INTEGER,
+    suggested_service_id TEXT, suggested_score REAL
+);
+-- Синонимы, выученные при ручной верификации (для дообучения нормализации).
+CREATE TABLE IF NOT EXISTS learned_synonym (name_norm TEXT PRIMARY KEY, service_id TEXT);
+CREATE INDEX IF NOT EXISTS price_item_service ON price_item(service_id);
+CREATE INDEX IF NOT EXISTS price_item_partner ON price_item(partner_id);
+"""
+
+_INSERT_ITEM = """INSERT INTO price_item
+    (doc_id, partner_id, service_name_raw, service_code_source, service_id, prices,
+     price_resident, price_nonresident, unit, effective_date, map_method, map_confidence,
+     status, is_active, suggested_service_id, suggested_score)
+    VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,1,?,?)"""
+
+
+def _emb_path(db_path: str) -> Path:
+    """Путь к кэшу эмбеддингов справочника рядом с базой."""
+    return Path(db_path).parent / "dict_emb.npy"
+
+
+def _open(db_path: str) -> sqlite3.Connection:
+    # Полная пересборка: удаляем файл, чтобы схема всегда создавалась свежей.
+    Path(db_path).unlink(missing_ok=True)
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    conn.executescript(SCHEMA)
+    return conn
+
+
+def _load_services(conn, services) -> None:
+    """Залить справочник услуг, если таблица ещё пуста."""
+    if conn.execute("SELECT 1 FROM service LIMIT 1").fetchone():
+        return
+    conn.executemany(
+        "INSERT INTO service VALUES (?,?,?,?,?)",
+        [(s.service_id, s.specialty, s.name_ru, s.name_norm, s.tarificator_code) for s in services],
+    )
+
+
+def _process_documents(conn, data_dir, matcher, partners, today, use_vision) -> Counter:
+    """Извлечь файлы из каталога, сопоставить со справочником и записать позиции.
+
+    Мутирует `conn` и словарь `partners` (норм. имя → partner_id, чтобы один и тот же
+    партнёр из разных файлов не задвоился). Возвращает счётчик для отчёта о качестве.
+    """
+    report: Counter = Counter()
+    staged: list[tuple] = []  # (doc_id, partner_id, raw, code, prices, unit, eff_iso)
+
+    for path in sorted(Path(data_dir).glob("*")):
+        if path.is_dir():
+            continue
+        doc = extract(str(path), use_vision=use_vision)
+        partner_name = doc.partner_name or path.stem
+        partner_norm = normalize_name(partner_name)
+        partner_id = partners.get(partner_norm)
+        if partner_id is None:
+            partner_id = str(uuid.uuid4())
+            partners[partner_norm] = partner_id
+            conn.execute(
+                "INSERT INTO partner VALUES (?,?,?,?)",
+                (partner_id, partner_name, partner_norm, None),
+            )
+            report["new_partners"] += 1
+
+        doc_id = str(uuid.uuid4())
+        eff_iso = doc.effective_date.isoformat() if doc.effective_date else None
+        conn.execute(
+            "INSERT INTO price_document VALUES (?,?,?,?,?,?,?)",
+            (doc_id, partner_id, doc.file_name, doc.file_format, eff_iso, "done", len(doc.rows)),
+        )
+        report["documents"] += 1
+        for row in doc.rows:
+            staged.append(
+                (
+                    doc_id,
+                    partner_id,
+                    row.service_name_raw,
+                    row.service_code_source,
+                    row.prices,
+                    row.unit,
+                    eff_iso,
+                )
+            )
+
+    # Нормализация одной пачкой (эмбеддинги считаются разом — быстрее и дешевле).
+    matches = matcher.match_batch([s[2] for s in staged], [s[3] for s in staged])
+    for (doc_id, partner_id, raw, code, prices, unit, eff_iso), match in zip(
+        staged, matches, strict=True
+    ):
+        status, _flags = validate_item(raw, prices, effective_date=None, today=today)
+        if match.service_id:
+            report["matched"] += 1
+        report[f"method_{match.method}"] += 1
+        conn.execute(
+            _INSERT_ITEM,
+            (
+                doc_id,
+                partner_id,
+                raw,
+                code,
+                match.service_id,
+                json.dumps(prices, ensure_ascii=False),
+                prices.get("resident"),
+                prices.get("nonresident"),
+                unit,
+                eff_iso,
+                match.method,
+                round(match.confidence, 3),
+                status,
+                match.suggested_service_id,
+                round(match.suggested_score, 3) if match.suggested_score else None,
+            ),
+        )
+        report["items"] += 1
+    return report
+
+
+def _apply_versioning(conn) -> None:
+    """У партнёра по той же услуге более свежий прайс вытесняет старые (is_active = 0)."""
+    conn.execute(
+        """UPDATE price_item SET is_active = 0
+           WHERE effective_date IS NOT NULL AND EXISTS (
+               SELECT 1 FROM price_item b
+               WHERE b.partner_id = price_item.partner_id
+                 AND b.service_name_raw = price_item.service_name_raw
+                 AND b.effective_date > price_item.effective_date)"""
+    )
+
+
+def _summary(report: Counter) -> dict:
+    total = report["items"]
+    return {
+        "documents": report["documents"],
+        "partners": report["new_partners"],
+        "items": total,
+        "auto_matched": report["matched"],
+        "auto_matched_pct": round(100 * report["matched"] / total, 1) if total else 0.0,
+        "unmatched": total - report["matched"],
+        "by_method": {
+            method: report[f"method_{method}"]
+            for method in ("code", "exact", "embedding", "fuzzy", None)
+        },
+    }
+
+
+def run(
+    data_dir: str,
+    db_path: str,
+    dict_path: str,
+    embedder=None,
+    use_vision: bool = False,
+    today: date | None = None,
+) -> dict:
+    """Собрать базу с нуля по всему архиву и вернуть отчёт о качестве."""
+    today = today or date.today()
+    services = load_dictionary(dict_path)
+    matcher = Matcher(services, embedder=embedder)
+    # Кэшируем эмбеддинги справочника: при последующих догрузках их не пересчитываем.
+    if matcher.dict_emb is not None:
+        np.save(_emb_path(db_path), matcher.dict_emb)
+
+    conn = _open(db_path)
+    _load_services(conn, services)
+    partners: dict[str, str] = {}
+    report = _process_documents(conn, data_dir, matcher, partners, today, use_vision)
+    _apply_versioning(conn)
+    conn.commit()
+    summary = _summary(report)
+    conn.close()
+    return summary
+
+
+def ingest(
+    data_dir: str,
+    db_path: str,
+    dict_path: str,
+    embedder=None,
+    use_vision: bool | None = None,
+    today: date | None = None,
+) -> dict:
+    """Догрузить новые прайсы из каталога в существующую базу (append, без пересборки).
+
+    На этом держится приём ZIP через интерфейс: справочник и его эмбеддинги уже готовы,
+    поэтому считается только новый файл. Партнёры дедуплицируются с уже загруженными,
+    версионирование вытесняет устаревшие прайсы той же клиники.
+    """
+    today = today or date.today()
+    services = load_dictionary(dict_path)
+    emb_path = _emb_path(db_path)
+    dict_emb = np.load(emb_path) if (embedder is not None and emb_path.exists()) else None
+    matcher = Matcher(services, embedder=embedder, dict_emb=dict_emb)
+    if use_vision is None:
+        use_vision = embedder is not None  # есть ключ Gemini — разрешаем распознавание сканов
+
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    conn.executescript(SCHEMA)
+    _load_services(conn, services)
+    # Уже загруженные партнёры — чтобы повторная выгрузка той же клиники не задвоила её.
+    partners = {
+        row["name_norm"]: row["partner_id"]
+        for row in conn.execute("SELECT partner_id, name_norm FROM partner")
+    }
+    report = _process_documents(conn, data_dir, matcher, partners, today, use_vision)
+    _apply_versioning(conn)
+    conn.commit()
+    summary = _summary(report)
+    conn.close()
+    return summary

etl/validate/__init__.py

@@ -0,0 +1,5 @@
+"""Валидация позиций прайса по правилам ТЗ §4.4."""
+
+from etl.validate.rules import Flag, to_kzt, validate_item
+
+__all__ = ["Flag", "to_kzt", "validate_item"]

etl/validate/rules.py

@@ -0,0 +1,103 @@
+"""Валидация позиций прайса — правила ТЗ §4.4.
+
+Каждое правило возвращает флаги; по их уровню определяется статус позиции:
+ошибка → error, предупреждение → needs_review, иначе → done. Конвертация валют и
+сравнение с прошлой версией (детектор аномалий) тоже здесь.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date
+
+# Ориентировочные курсы к тенге. В проде — курс Нацбанка на дату прайса.
+RATES_TO_KZT: dict[str, float] = {"KZT": 1.0, "USD": 480.0, "RUB": 5.3}
+
+ANOMALY_RATIO = 0.5  # изменение цены более чем на 50% к прошлой версии — аномалия
+
+
+@dataclass
+class Flag:
+    """Замечание валидации по одной позиции."""
+
+    level: str  # info / warning / error
+    code: str
+    message: str
+
+
+def validate_name(name: str | None) -> list[Flag]:
+    if name and name.strip():
+        return []
+    return [Flag("error", "empty_name", "пустое название услуги")]
+
+
+def validate_prices(prices: dict[str, float]) -> list[Flag]:
+    flags: list[Flag] = []
+    for tier, value in prices.items():
+        if value is None or value <= 0:
+            flags.append(
+                Flag("warning", "price_nonpositive", f"тариф «{tier}»: цена не положительна")
+            )
+    resident, nonresident = prices.get("resident"), prices.get("nonresident")
+    if resident and nonresident and nonresident < resident:
+        flags.append(
+            Flag("warning", "nonresident_below_resident", "цена нерезидента ниже цены резидента")
+        )
+    return flags
+
+
+def validate_date(effective_date: date | None, today: date | None = None) -> list[Flag]:
+    if effective_date and today and effective_date > today:
+        return [Flag("warning", "future_date", "дата прайса в будущем")]
+    return []
+
+
+def validate_anomaly(new_price: float | None, previous_price: float | None) -> list[Flag]:
+    if (
+        previous_price
+        and new_price
+        and abs(new_price - previous_price) / previous_price > ANOMALY_RATIO
+    ):
+        return [
+            Flag(
+                "warning",
+                "price_anomaly",
+                f"цена изменилась более чем на {int(ANOMALY_RATIO * 100)}% "
+                f"(было {previous_price:g}, стало {new_price:g})",
+            )
+        ]
+    return []
+
+
+def to_kzt(amount: float, currency: str, rates: dict[str, float] | None = None) -> float:
+    """Перевести сумму в тенге по курсу (оригинал сохраняется отдельно)."""
+    rates = rates or RATES_TO_KZT
+    return amount * rates.get(currency.upper(), 1.0)
+
+
+def validate_item(
+    name: str | None,
+    prices: dict[str, float],
+    *,
+    currency: str = "KZT",
+    effective_date: date | None = None,
+    previous_resident: float | None = None,
+    today: date | None = None,
+) -> tuple[str, list[Flag]]:
+    """Проверить позицию и вернуть (статус, флаги).
+
+    Статус: error — нельзя сохранять как есть; needs_review — на ручную проверку;
+    done — чисто.
+    """
+    flags: list[Flag] = []
+    flags += validate_name(name)
+    flags += validate_prices(prices)
+    flags += validate_date(effective_date, today)
+    if previous_resident is not None:
+        flags += validate_anomaly(prices.get("resident"), previous_resident)
+
+    if any(f.level == "error" for f in flags):
+        return "error", flags
+    if any(f.level == "warning" for f in flags):
+        return "needs_review", flags
+    return "done", flags

scripts/run_pipeline.py

@@ -0,0 +1,28 @@
+"""Запуск конвейера на архиве: data/raw → SQLite + отчёт о качестве.
+
+Флаг --vision включает добор трудных PDF через Gemini Vision (медленно).
+"""
+
+import json
+import sys
+from datetime import date
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO))
+
+from etl.normalize.embedding import GeminiEmbedder  # noqa: E402
+from etl.pipeline import run  # noqa: E402
+
+summary = run(
+    data_dir=str(REPO / "data/raw"),
+    db_path=str(REPO / "data/medarchive.db"),
+    dict_path=str(REPO / "data/reference/dictionary.xlsx"),
+    embedder=GeminiEmbedder(),
+    use_vision="--vision" in sys.argv,
+    today=date(2026, 6, 26),
+)
+(REPO / "data/quality_report.json").write_text(
+    json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8"
+)
+print(json.dumps(summary, ensure_ascii=False, indent=2))