feat(день2): экран верификации с предложениями + обучение, детектор переплаты

- очередь верификации: предложенный кандидат (топ эмбеддинг), подтвердить/другое/пропустить
- POST /match пополняет синонимы (learned_synonym) — обучающаяся нормализация
- сравнение: медиана + разброс цен + отклонение клиники (переплата/выгода)
- эмбеддер: дедуп названий + ретрай при 429 (квота Gemini)
- развёрнуто на med.secondbrain.tools

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

etl/normalize/embedding.py

@@ -43,6 +43,8 @@ class GeminiEmbedder:
         RETRIEVAL_QUERY: так несвязанные названия расходятся сильнее, чем при
         SEMANTIC_SIMILARITY, и порог отсечения работает надёжнее.
         """
+        import time
+
         from google.genai import types
 
         items = [t if t and t.strip() else " " for t in texts]
@@ -50,8 +52,16 @@ class GeminiEmbedder:
         vectors: list[list[float]] = []
         for start in range(0, len(items), self.batch):
             chunk = items[start : start + self.batch]
-            response = self._client.models.embed_content(model=self.model, contents=chunk, config=config)
-            vectors.extend(embedding.values for embedding in response.embeddings)
+            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:

etl/normalize/matcher.py

@@ -89,23 +89,35 @@ class Matcher:
             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(
-                [names[i] for i in pending_idx],
-                normalize_embeddings=True,
-                task_type="RETRIEVAL_QUERY",
+                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):
-                if best_score[k] >= self.emb_threshold:
+                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=self.services[int(best_idx[k])].service_id,
+                        service_id=suggested,
                         method="embedding",
-                        confidence=float(best_score[k]),
+                        confidence=score,
+                        suggested_service_id=suggested,
+                        suggested_score=score,
                     )
                 else:
-                    results[i] = self._match_by_fuzzy(pending_norm[k])
+                    # Не дотянул до порога — в очередь, но кандидата сохраняем оператору.
+                    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])

etl/pipeline.py

@@ -37,18 +37,21 @@ 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
+    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);
 """
 
 
 def _open(db_path: str) -> sqlite3.Connection:
+    # Полная пересборка: удаляем файл, чтобы схема всегда создавалась свежей.
+    Path(db_path).unlink(missing_ok=True)
     conn = sqlite3.connect(db_path)
     conn.executescript(SCHEMA)
-    for table in ("price_item", "price_document", "partner", "service"):
-        conn.execute(f"DELETE FROM {table}")  # пересборка с нуля при каждом прогоне
     return conn
 
 
@@ -102,11 +105,12 @@ def run(data_dir: str, db_path: str, dict_path: str, embedder=None, use_vision:
             """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)
-               VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,1)""",
+                status, is_active, suggested_service_id, suggested_score)
+               VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,1,?,?)""",
             (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),
+             round(match.confidence, 3), status, match.suggested_service_id,
+             round(match.suggested_score, 3) if match.suggested_score else None),
         )
         report["items"] += 1