feat(etl): нормализация (каскад), валидация (8 правил), Vision-fallback
- etl/normalize/matcher.py — каскад код→точное→эмбеддинги→fuzzy→unmatched - etl/validate/rules.py — правила ТЗ §4.4; проверено: 15588 done / 11 needs_review - etl/extractors/vision.py — Gemini Vision для битых/нестандартных PDF - замер % автонормализации — после установки модели эмбеддингов Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
"""Запасной путь извлечения через Gemini Vision.
|
||||
|
||||
Когда нативный парсер не справился (битый текстовый слой, нестандартная вёрстка,
|
||||
скан), страница PDF отрисовывается в картинку и отдаётся Gemini с требованием
|
||||
вернуть позиции прайса строго по JSON-схеме. Требует переменную окружения
|
||||
GEMINI_API_KEY; модель задаётся через GEMINI_MODEL (по умолчанию gemini-2.0-flash).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
|
||||
from contracts.models import RawRow # noqa: E402
|
||||
|
||||
_VALID_TIERS = {"resident", "nonresident", "cis", "far", "insurance", "partner"}
|
||||
|
||||
_PROMPT = """Ты извлекаешь позиции из прайс-листа медицинской клиники на изображении.
|
||||
Верни СТРОГО JSON-массив объектов, без пояснений и текста вокруг. Поля объекта:
|
||||
- "name": название услуги как в документе (строка);
|
||||
- "code": код услуги или тарификатора, если есть, иначе null;
|
||||
- "unit": единица измерения, если есть, иначе null;
|
||||
- "prices": объект «тариф → число в тенге». Ключи тарифов:
|
||||
"resident" (резидент / гражданин РК), "nonresident" (нерезидент / иностранец),
|
||||
"cis" (СНГ / ближнее зарубежье), "far" (дальнее зарубежье),
|
||||
"insurance" (страховые компании), "partner" (партнёр).
|
||||
Если в строке одна цена — положи её в "resident".
|
||||
Пропускай заголовки разделов, итоги и строки без цены. Числа — без пробелов и валюты."""
|
||||
|
||||
|
||||
def _render_pages(path: str, max_pages: int, dpi: int) -> list[bytes]:
|
||||
"""Отрисовать первые страницы PDF в PNG-картинки."""
|
||||
import fitz # PyMuPDF
|
||||
|
||||
document = fitz.open(path)
|
||||
images = []
|
||||
for page in document[:max_pages]:
|
||||
images.append(page.get_pixmap(dpi=dpi).tobytes("png"))
|
||||
return images
|
||||
|
||||
|
||||
def _parse_response(text: str) -> list[RawRow]:
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return []
|
||||
rows: list[RawRow] = []
|
||||
for obj in data if isinstance(data, list) else []:
|
||||
if not isinstance(obj, dict):
|
||||
continue
|
||||
name = str(obj.get("name") or "").strip()
|
||||
prices = {
|
||||
tier: float(value)
|
||||
for tier, value in (obj.get("prices") or {}).items()
|
||||
if tier in _VALID_TIERS and isinstance(value, (int, float)) and value > 0
|
||||
}
|
||||
if name and prices:
|
||||
rows.append(
|
||||
RawRow(
|
||||
service_name_raw=name,
|
||||
service_code_source=(obj.get("code") or None),
|
||||
prices=prices,
|
||||
unit=(obj.get("unit") or None),
|
||||
)
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def extract_with_vision(path: str, max_pages: int = 12, dpi: int = 140) -> list[RawRow]:
|
||||
"""Извлечь позиции из PDF через Gemini Vision. Возвращает список RawRow."""
|
||||
api_key = os.environ.get("GEMINI_API_KEY")
|
||||
if not api_key:
|
||||
raise RuntimeError("нет GEMINI_API_KEY в окружении")
|
||||
|
||||
import google.generativeai as genai
|
||||
|
||||
genai.configure(api_key=api_key)
|
||||
model = genai.GenerativeModel(os.environ.get("GEMINI_MODEL", "gemini-2.0-flash"))
|
||||
|
||||
rows: list[RawRow] = []
|
||||
for png in _render_pages(path, max_pages, dpi):
|
||||
response = model.generate_content(
|
||||
[_PROMPT, {"mime_type": "image/png", "data": png}],
|
||||
generation_config={"response_mime_type": "application/json", "temperature": 0},
|
||||
)
|
||||
rows.extend(_parse_response(response.text))
|
||||
return rows
|
||||
@@ -0,0 +1,4 @@
|
||||
"""Нормализация извлечённых позиций к справочнику услуг."""
|
||||
from etl.normalize.matcher import Matcher
|
||||
|
||||
__all__ = ["Matcher"]
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Сопоставление извлечённых позиций с эталонным справочником услуг.
|
||||
|
||||
Каскад по убыванию надёжности:
|
||||
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.60,
|
||||
fuzzy_threshold: int = 88,
|
||||
):
|
||||
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
|
||||
self.dict_emb = None
|
||||
if embedder is not None:
|
||||
self.dict_emb = embedder.encode(
|
||||
[s.name_ru for s in services], normalize_embeddings=True, show_progress_bar=False
|
||||
)
|
||||
|
||||
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:
|
||||
query = self.embedder.encode(
|
||||
[names[i] for i in pending_idx], normalize_embeddings=True, show_progress_bar=False
|
||||
)
|
||||
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):
|
||||
if best_score[k] >= self.emb_threshold:
|
||||
results[i] = MatchResult(
|
||||
service_id=self.services[int(best_idx[k])].service_id,
|
||||
method="embedding",
|
||||
confidence=float(best_score[k]),
|
||||
)
|
||||
else:
|
||||
results[i] = self._match_by_fuzzy(pending_norm[k])
|
||||
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]
|
||||
@@ -0,0 +1,4 @@
|
||||
"""Валидация позиций прайса по правилам ТЗ §4.4."""
|
||||
from etl.validate.rules import Flag, to_kzt, validate_item
|
||||
|
||||
__all__ = ["Flag", "to_kzt", "validate_item"]
|
||||
@@ -0,0 +1,96 @@
|
||||
"""Валидация позиций прайса — правила ТЗ §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
|
||||
@@ -0,0 +1,59 @@
|
||||
"""Замер автонормализации: извлечь весь архив, сопоставить со справочником, дать %."""
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(REPO))
|
||||
|
||||
from sentence_transformers import SentenceTransformer
|
||||
|
||||
from etl.dictionary import load_dictionary
|
||||
from etl.extractors import extract
|
||||
from etl.normalize import Matcher
|
||||
|
||||
# --- собрать все извлечённые позиции ---
|
||||
rows: list[tuple[str, str | None]] = []
|
||||
for path in sorted((REPO / "data/raw").glob("*")):
|
||||
for row in extract(str(path)).rows:
|
||||
rows.append((row.service_name_raw, row.service_code_source))
|
||||
print(f"позиций всего: {len(rows)}")
|
||||
|
||||
# дедуп по (имя, код) — одинаковые строки сопоставляем один раз
|
||||
occurrences: Counter[tuple[str, str | None]] = Counter(rows)
|
||||
keys = list(occurrences)
|
||||
print(f"уникальных (имя, код): {len(keys)}")
|
||||
|
||||
services = load_dictionary(REPO / "data/reference/dictionary.xlsx")
|
||||
print("загружаю модель эмбеддингов (первый раз — скачивание)…")
|
||||
embedder = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
|
||||
matcher = Matcher(services, embedder=embedder)
|
||||
|
||||
print("сопоставляю…")
|
||||
results = matcher.match_batch([k[0] for k in keys], [k[1] for k in keys])
|
||||
|
||||
by_method: Counter[str | None] = Counter()
|
||||
total = matched = 0
|
||||
samples: dict[str | None, list] = {}
|
||||
canon_by_id = {s.service_id: s.name_ru for s in services}
|
||||
for key, res in zip(keys, results, strict=True):
|
||||
count = occurrences[key]
|
||||
total += count
|
||||
by_method[res.method] += count
|
||||
if res.service_id:
|
||||
matched += count
|
||||
bucket = samples.setdefault(res.method, [])
|
||||
if len(bucket) < 5:
|
||||
bucket.append((key[0], canon_by_id.get(res.service_id, "—"), round(res.confidence, 2)))
|
||||
|
||||
print(f"\nАВТОНОРМАЛИЗАЦИЯ: {matched}/{total} = {100 * matched / total:.0f}% (цель ТЗ ≥70%)")
|
||||
for method in ("code", "exact", "embedding", "fuzzy", None):
|
||||
print(f" {method or 'unmatched':10}: {by_method[method]} строк")
|
||||
|
||||
print("\nпримеры сопоставлений:")
|
||||
for method in ("code", "exact", "embedding", "fuzzy"):
|
||||
for raw, canon, score in samples.get(method, [])[:3]:
|
||||
print(f" [{method:9}] «{raw[:36]}» → «{canon[:36]}» ({score})")
|
||||
print("\nпримеры unmatched:")
|
||||
for raw, _canon, _score in samples.get(None, [])[:6]:
|
||||
print(f" «{raw[:52]}»")
|
||||
Reference in New Issue
Block a user