diff --git a/etl/normalize/__init__.py b/etl/normalize/__init__.py new file mode 100644 index 0000000..0e1247f --- /dev/null +++ b/etl/normalize/__init__.py @@ -0,0 +1,5 @@ +"""Нормализация извлечённых позиций к справочнику услуг.""" + +from etl.normalize.matcher import Matcher + +__all__ = ["Matcher"] diff --git a/etl/normalize/embedding.py b/etl/normalize/embedding.py new file mode 100644 index 0000000..adcca25 --- /dev/null +++ b/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 diff --git a/etl/normalize/matcher.py b/etl/normalize/matcher.py new file mode 100644 index 0000000..d438e86 --- /dev/null +++ b/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] diff --git a/etl/pipeline.py b/etl/pipeline.py new file mode 100644 index 0000000..08e48ee --- /dev/null +++ b/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 diff --git a/etl/validate/__init__.py b/etl/validate/__init__.py new file mode 100644 index 0000000..278161b --- /dev/null +++ b/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"] diff --git a/etl/validate/rules.py b/etl/validate/rules.py new file mode 100644 index 0000000..698111f --- /dev/null +++ b/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 diff --git a/scripts/run_pipeline.py b/scripts/run_pipeline.py new file mode 100644 index 0000000..ece9119 --- /dev/null +++ b/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))