feat(etl): гео-проход auto_core_geo_v6 (#2177 остаток)

backend/app/core/config.py

@@ -346,5 +346,18 @@ class Settings(BaseSettings):
     # Сколько ретраев на 429/5xx до деградации в fallback (циркуит-брейкер-lite).
     llm_max_retries: int = 2
 
+    # ── DaData /clean/address геокод (objective_backfill geo-pass, #2177) ──────
+    # Токен + секрет для DaData /clean/address (обогащение адреса → geo_lat/geo_lon).
+    # На gendesign-проде уже заведены в окружении контейнера (проверено 2026-07-03).
+    # Оба нужны для clean_address (в отличие от suggest, которому хватает токена).
+    # None/пусто → dadata_client.clean_address graceful-возвращает None (geo-pass
+    # тогда reject'ит всё «нет геокода», не падает). ENV: DADATA_API_TOKEN /
+    # DADATA_API_SECRET. Литерала ключа в коде/тестах нет.
+    dadata_api_token: str | None = None
+    dadata_api_secret: str | None = None
+    # Таймаут одного HTTP-вызова к DaData (сек). Воркер geo-pass не должен висеть
+    # на недоступном сервисе. ENV: DADATA_TIMEOUT_S.
+    dadata_timeout_s: float = 8.0
+
 
 settings = Settings()

backend/app/services/dadata_client.py

@@ -0,0 +1,135 @@
+"""Минимальный синхронный DaData-клиент — /clean/address → (lat, lon).
+
+Нужен ТОЛЬКО для geo-pass в objective_backfill (#2177): адрес objective-проекта
+→ координаты для haversine-сверки с domrf-точкой. Не тащим tradein-модуль
+(`tradein-mvp/.../dadata.py`) импортом — это другое приложение с другим settings.
+
+Клиент СИНХРОННЫЙ (httpx.Client): geo-pass гоняется в CLI / sync-Session, без
+event-loop. Авторизация /clean/address требует ОБА: token + secret (в отличие от
+/suggest, которому хватает токена). Оба берутся из settings (env DADATA_API_TOKEN /
+DADATA_API_SECRET).
+
+Возвращает None (не raise) на любой сбой — нет credentials, пустой/короткий адрес,
+сетевая/HTTP-ошибка, 429 quota, невалидный JSON, адрес не распознан. Вызывающая
+сторона (geo-pass) трактует None как «нет геокода» → reject-кандидат, проход не
+падает.
+
+Docs: https://dadata.ru/api/clean/address/
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+DADATA_CLEAN_URL = "https://cleaner.dadata.ru/api/v1/clean/address"
+
+
+def _credentials() -> tuple[str, str] | None:
+    """(token, secret) если ОБА заданы непусто, иначе None (graceful disable)."""
+    token = (settings.dadata_api_token or "").strip()
+    secret = (settings.dadata_api_secret or "").strip()
+    if not token or not secret:
+        return None
+    return token, secret
+
+
+def _coerce_float(value: Any) -> float | None:
+    """DaData отдаёт geo_lat/geo_lon строкой — coerce в float безопасно."""
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def clean_address(address: str) -> tuple[float, float] | None:
+    """Геокодировать один адрес через DaData /clean/address → (lat, lon).
+
+    Args:
+        address: свободный текст адреса (objective_lots.address).
+
+    Returns:
+        (geo_lat, geo_lon) как float-пара, либо None если:
+          - credentials не заданы в ENV (graceful disable);
+          - адрес пустой / короче 3 символов;
+          - сетевая / HTTP-ошибка (timeout, 5xx, 429 quota, 401/403 auth, 4xx);
+          - невалидный JSON / пустой массив / не dict;
+          - DaData не вернул geo_lat/geo_lon (адрес не геокодирован).
+    """
+    if not address or len(address.strip()) < 3:
+        return None
+
+    creds = _credentials()
+    if creds is None:
+        logger.debug("dadata_client: credentials не заданы — skip геокод")
+        return None
+
+    token, secret = creds
+    headers = {
+        "Authorization": f"Token {token}",
+        "X-Secret": secret,
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    body = [address.strip()]
+
+    try:
+        with httpx.Client(timeout=settings.dadata_timeout_s) as client:
+            response = client.post(DADATA_CLEAN_URL, headers=headers, json=body)
+    except (httpx.TimeoutException, httpx.NetworkError) as exc:
+        logger.warning("dadata_client: network error для %r: %s", address[:60], exc)
+        return None
+    except Exception as exc:
+        # defensive: любой прочий httpx/клиент-сбой → None, geo-pass reject'ит,
+        # НЕ роняем весь проход. Логируем на warning (не глотаем молча).
+        logger.warning("dadata_client: unexpected client error для %r: %s", address[:60], exc)
+        return None
+
+    status = response.status_code
+    if status == 429:
+        logger.warning("dadata_client: HTTP 429 — quota exceeded")
+        return None
+    if status in (401, 403):
+        logger.error(
+            "dadata_client: HTTP %d — auth rejected. Проверь DADATA_API_TOKEN/SECRET.",
+            status,
+        )
+        return None
+    if status >= 400:
+        logger.warning("dadata_client: HTTP %d для %r", status, address[:60])
+        return None
+
+    try:
+        payload = response.json()
+    except ValueError as exc:
+        logger.warning("dadata_client: невалидный JSON: %s", exc)
+        return None
+
+    if not isinstance(payload, list) or not payload:
+        logger.info("dadata_client: пустой ответ для %r", address[:60])
+        return None
+
+    item = payload[0]
+    if not isinstance(item, dict):
+        logger.warning("dadata_client: первый элемент не dict (%s)", type(item).__name__)
+        return None
+
+    lat = _coerce_float(item.get("geo_lat"))
+    lon = _coerce_float(item.get("geo_lon"))
+    if lat is None or lon is None:
+        logger.info(
+            "dadata_client: нет координат (qc_geo=%s) для %r",
+            item.get("qc_geo"),
+            address[:60],
+        )
+        return None
+
+    return (lat, lon)

backend/app/services/etl/objective_backfill.py

@@ -21,6 +21,13 @@ match_method history:
                        vs «Малевич»). Ядро-нормализация схлопывает обёртки → exact-core match;
                        dev-подтверждение (пересечение токенов / подстрока) отсеивает
                        одноимённые чужие проекты (Объектив = только ЕКБ). #2177 шаг 2.
+  'auto_core_geo_v6' — гео-разрешение остатка core-pass (tier_b / ambiguous) через
+                       DaData-геокод. Адрес objective-проекта (mode objective_lots.
+                       address per project) → clean_address → (lat, lon); haversine
+                       до domrf-точки (domrf_kn_objects.latitude/longitude, latest
+                       snapshot). tier_b: дистанция ≤ порога → confirm (гео вместо
+                       dev). ambiguous: геокодим ВСЕХ кандидатов ядра, confirm только
+                       если РОВНО ОДИН в радиусе. score 0.80, is_reviewed=false. #2177.
   'manual'         — ручная корректура
 """
 
@@ -28,6 +35,7 @@ from __future__ import annotations
 
 import argparse
 import logging
+import math
 import re
 import sys
 from dataclasses import dataclass, field
@@ -35,6 +43,7 @@ from dataclasses import dataclass, field
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services.dadata_client import clean_address
 from app.services.site_finder.best_layouts import normalize_complex_name
 
 logger = logging.getLogger(__name__)
@@ -637,19 +646,498 @@ def run_core_pass(*, apply: bool = False) -> dict[str, int]:
         db.close()
 
 
+# ════════════════════════════════════════════════════════════════════════════
+# Geo-pass (#2177) — auto_core_geo_v6
+# ════════════════════════════════════════════════════════════════════════════
+#
+# Разрешает ОСТАТОК core-pass (tier_b + ambiguous) географией: адрес objective-
+# проекта → DaData-геокод → haversine до domrf-точки. tier_b подтверждается
+# близостью (гео вместо dev-confirm); ambiguous — только если РОВНО ОДИН из
+# кандидатов ядра в радиусе (иначе остаётся ambiguous).
+
+GEO_MATCH_METHOD = "auto_core_geo_v6"
+GEO_MATCH_SCORE = 0.80
+GEO_MAX_DISTANCE_M = 500.0
+# Бережный потолок вызовов DaData за один прогон (МАКС-тариф, но не жжём квоту
+# на прогон-цикле). Превышение → стоп с warning, недобранные кандидаты reject'ятся
+# как «без геокода».
+GEO_DADATA_CALL_LIMIT = 200
+_EARTH_RADIUS_M = 6_371_000.0
+
+
+def _haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    """Расстояние (метры) между двумя точками по формуле гаверсинуса."""
+    p1 = math.radians(lat1)
+    p2 = math.radians(lat2)
+    dphi = math.radians(lat2 - lat1)
+    dlambda = math.radians(lon2 - lon1)
+    a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlambda / 2) ** 2
+    return 2 * _EARTH_RADIUS_M * math.asin(min(1.0, math.sqrt(a)))
+
+
+@dataclass
+class GeoMatch:
+    """Один подтверждённый гео-матч domrf-объекта к objective-проекту.
+
+    origin — 'tier_b' | 'ambiguous' (из какого класса core-pass пришёл).
+    distance_m — гаверсинус адрес-проекта ↔ domrf-точки (метры).
+    """
+
+    domrf_obj_id: int
+    domrf_comm_name: str
+    core: str
+    objective_project_name: str
+    objective_project_id: int | None
+    address: str
+    distance_m: float
+    origin: str
+
+
+@dataclass
+class GeoReject:
+    """Отклонённый гео-кандидат (для отчёта). reason: 'no_address' |
+    'no_geocode' | 'too_far' | 'ambiguous_multi' | 'call_limit'.
+
+    distance_m None когда дистанцию посчитать не удалось (нет адреса/геокода/
+    координат domrf).
+    """
+
+    domrf_obj_id: int
+    domrf_comm_name: str
+    objective_project_name: str
+    origin: str
+    reason: str
+    distance_m: float | None = None
+
+
+@dataclass
+class GeoMatchReport:
+    """Итог geo-pass.
+
+    confirmed — подтверждённые гео-матчи (пишутся в apply_geo_matches).
+    rejected  — кандидаты без адреса/геокода/слишком далёкие/оставшиеся ambiguous.
+    dadata_calls — сколько раз реально дёрнули DaData (для контроля квоты).
+    call_limit_hit — True если упёрлись в GEO_DADATA_CALL_LIMIT (часть кандидатов
+                     не обработана — см. rejected c reason='call_limit').
+    """
+
+    confirmed: list[GeoMatch] = field(default_factory=list)
+    rejected: list[GeoReject] = field(default_factory=list)
+    dadata_calls: int = 0
+    call_limit_hit: bool = False
+
+    def counts(self) -> dict[str, int]:
+        return {
+            "confirmed": len(self.confirmed),
+            "rejected": len(self.rejected),
+            "dadata_calls": self.dadata_calls,
+        }
+
+
+# ── SQL: mode (самый частый непустой) адрес objective_lots per project_name ────
+# Адресов на проект много (по корпусам) — берём mode: GROUP BY (project_name,
+# address) → COUNT DESC, DISTINCT ON (project_name) отбирает верхнюю строку.
+# NULLIF(trim,'') отсекает пустые адреса до агрегации.
+_OBJECTIVE_ADDRESS_SQL = text("""
+    SELECT DISTINCT ON (project_name)
+        project_name, address
+    FROM (
+        SELECT
+            ol.project_name AS project_name,
+            NULLIF(trim(ol.address), '') AS address,
+            COUNT(*) AS cnt
+        FROM objective_lots ol
+        WHERE ol.project_name IS NOT NULL
+          AND ol.project_name <> ''
+          AND NULLIF(trim(ol.address), '') IS NOT NULL
+        GROUP BY ol.project_name, NULLIF(trim(ol.address), '')
+    ) agg
+    ORDER BY project_name, cnt DESC, address
+""")
+
+# ── SQL: latest-snapshot координаты domrf per obj_id (для набора obj_id) ───────
+# DISTINCT ON (obj_id) + snapshot_date DESC — берём свежую точку. Фильтр по
+# конкретным obj_id (кандидаты geo-pass), NULL-координаты отбрасываются.
+_DOMRF_COORDS_SQL = text("""
+    SELECT DISTINCT ON (o.obj_id)
+        o.obj_id, o.latitude, o.longitude
+    FROM domrf_kn_objects o
+    WHERE o.obj_id = ANY(:obj_ids)
+      AND o.latitude IS NOT NULL
+      AND o.longitude IS NOT NULL
+    ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST
+""")
+
+
+def _load_project_addresses(db: Session) -> dict[str, str]:
+    """project_name → mode-адрес (самый частый непустой) из objective_lots."""
+    out: dict[str, str] = {}
+    for row in db.execute(_OBJECTIVE_ADDRESS_SQL).all():
+        name = str(row[0])
+        addr = row[1]
+        if addr:
+            out[name] = str(addr)
+    return out
+
+
+def _load_domrf_coords(db: Session, obj_ids: list[int]) -> dict[int, tuple[float, float]]:
+    """obj_id → (lat, lon) latest snapshot для заданного набора obj_id."""
+    if not obj_ids:
+        return {}
+    out: dict[int, tuple[float, float]] = {}
+    for row in db.execute(_DOMRF_COORDS_SQL, {"obj_ids": obj_ids}).all():
+        lat = float(row[1])
+        lon = float(row[2])
+        out[int(row[0])] = (lat, lon)
+    return out
+
+
+def find_geo_matches(db: Session, *, max_distance_m: float = GEO_MAX_DISTANCE_M) -> GeoMatchReport:
+    """Разрешить остаток core-pass (tier_b + ambiguous) через DaData-геокод.
+
+    Алгоритм:
+      1. find_core_matches → берём report.tier_b и report.ambiguous.
+      2. Для каждого кандидата: адрес проекта (mode objective_lots.address) →
+         clean_address → (lat, lon); haversine до domrf-точки.
+         • tier_b: дистанция ≤ max_distance_m → confirm (гео вместо dev);
+         • ambiguous: геокодим ВСЕХ кандидатов ядра; confirm только если РОВНО
+           ОДИН в радиусе (иначе остаётся ambiguous);
+         • нет адреса / DaData null / дистанция больше → reject c дистанцией.
+      3. Кэш геокода в памяти прогона (адреса повторяются). Лимит вызовов DaData
+         ≤ GEO_DADATA_CALL_LIMIT — при превышении стоп с warning, остаток reject.
+
+    Args:
+        db: SQLAlchemy sync Session.
+        max_distance_m: радиус подтверждения (метры), default 500.
+
+    Returns:
+        GeoMatchReport (confirmed / rejected / dadata_calls / call_limit_hit).
+
+    Реального обращения к БД / DaData в тестах нет — оба мокаются по паттерну файла.
+    """
+    core_report = find_core_matches(db)
+    report = GeoMatchReport()
+
+    tier_b = core_report.tier_b
+    ambiguous = core_report.ambiguous
+    if not tier_b and not ambiguous:
+        logger.info("find_geo_matches: нет tier_b/ambiguous кандидатов — nothing to do")
+        return report
+
+    addresses = _load_project_addresses(db)
+
+    # ── ambiguous: нужны ВСЕ objective-кандидаты ядра, не только первый ────────
+    # core-report.ambiguous хранит по одному CoreMatch на domrf-obj_id (первый
+    # кандидат). Восстанавливаем полный список project_name'ов ядра из objective-
+    # стороны, чтобы геокодить всех и выбрать единственного в радиусе.
+    objective_by_core = _load_objective_by_core(db)
+
+    # obj_id'ы, для которых нужны координаты domrf (tier_b + ambiguous)
+    needed_obj_ids = [m.domrf_obj_id for m in tier_b] + [m.domrf_obj_id for m in ambiguous]
+    domrf_coords = _load_domrf_coords(db, needed_obj_ids)
+
+    # Кэш геокода в памяти прогона: address → (lat, lon) | None (None = «пробовали,
+    # DaData не дал»; кэшируем и его, чтобы не жечь квоту на повтор одного адреса).
+    geo_cache: dict[str, tuple[float, float] | None] = {}
+
+    def _geocode(addr: str) -> tuple[float, float] | None:
+        """Геокод с кэшем + лимитом вызовов. None при отсутствии/лимите."""
+        if addr in geo_cache:
+            return geo_cache[addr]
+        if report.dadata_calls >= GEO_DADATA_CALL_LIMIT:
+            report.call_limit_hit = True
+            return None
+        report.dadata_calls += 1
+        coords = clean_address(addr)
+        geo_cache[addr] = coords
+        return coords
+
+    # ── tier_b: 1 кандидат, подтверждаем близостью ────────────────────────────
+    for m in tier_b:
+        domrf_pt = domrf_coords.get(m.domrf_obj_id)
+        addr = addresses.get(m.objective_project_name)
+        if not addr:
+            report.rejected.append(_geo_reject(m, "tier_b", "no_address"))
+            continue
+        if domrf_pt is None:
+            # у domrf нет координат — дистанцию не посчитать; не жжём квоту на геокод
+            report.rejected.append(_geo_reject(m, "tier_b", "no_geocode"))
+            continue
+        if report.dadata_calls >= GEO_DADATA_CALL_LIMIT and addr not in geo_cache:
+            report.call_limit_hit = True
+            report.rejected.append(_geo_reject(m, "tier_b", "call_limit"))
+            continue
+        coords = _geocode(addr)
+        if coords is None:
+            report.rejected.append(_geo_reject(m, "tier_b", "no_geocode"))
+            continue
+        dist = _haversine_m(coords[0], coords[1], domrf_pt[0], domrf_pt[1])
+        if dist <= max_distance_m:
+            report.confirmed.append(_geo_confirm(m, addr, dist, "tier_b"))
+        else:
+            report.rejected.append(_geo_reject(m, "tier_b", "too_far", dist))
+
+    # ── ambiguous: геокодим всех кандидатов ядра, confirm если РОВНО ОДИН близко ─
+    for m in ambiguous:
+        domrf_pt = domrf_coords.get(m.domrf_obj_id)
+        if domrf_pt is None:
+            report.rejected.append(_geo_reject(m, "ambiguous", "no_geocode"))
+            continue
+        candidates = objective_by_core.get(m.core, [])
+        in_radius: list[tuple[str, int | None, str, float]] = []
+        any_geocoded = False
+        geocoded_count = 0
+        for proj_name, _devs, proj_id in candidates:
+            addr = addresses.get(proj_name)
+            if not addr:
+                continue
+            if report.dadata_calls >= GEO_DADATA_CALL_LIMIT and addr not in geo_cache:
+                report.call_limit_hit = True
+                continue
+            coords = _geocode(addr)
+            if coords is None:
+                continue
+            any_geocoded = True
+            geocoded_count += 1
+            dist = _haversine_m(coords[0], coords[1], domrf_pt[0], domrf_pt[1])
+            if dist <= max_distance_m:
+                in_radius.append((proj_name, proj_id, addr, dist))
+
+        # Ревью 2026-07-03: «ровно один в радиусе» доверяем ТОЛЬКО когда
+        # геокодились ВСЕ кандидаты ядра. Иначе «один» — артефакт отсутствия
+        # данных (сосед без адреса/геокода мог быть ближе) → слепой confirm
+        # запрещён, остаётся честный reject partial_geocode.
+        if len(in_radius) == 1 and geocoded_count < len(candidates):
+            report.rejected.append(_geo_reject(m, "ambiguous", "partial_geocode", in_radius[0][3]))
+        elif len(in_radius) == 1:
+            proj_name, proj_id, addr, dist = in_radius[0]
+            report.confirmed.append(
+                GeoMatch(
+                    domrf_obj_id=m.domrf_obj_id,
+                    domrf_comm_name=m.domrf_comm_name,
+                    core=m.core,
+                    objective_project_name=proj_name,
+                    objective_project_id=proj_id,
+                    address=addr,
+                    distance_m=dist,
+                    origin="ambiguous",
+                )
+            )
+        elif not any_geocoded:
+            reason = "call_limit" if report.call_limit_hit else "no_geocode"
+            report.rejected.append(_geo_reject(m, "ambiguous", reason))
+        else:
+            # 0 в радиусе, или >1 в радиусе → остаётся ambiguous
+            nearest = min((d for *_, d in in_radius), default=None)
+            report.rejected.append(_geo_reject(m, "ambiguous", "ambiguous_multi", nearest))
+
+    logger.info(
+        "find_geo_matches: %s call_limit_hit=%s",
+        report.counts(),
+        report.call_limit_hit,
+    )
+    return report
+
+
+def _load_objective_by_core(db: Session) -> dict[str, list[tuple[str, set[str], int | None]]]:
+    """core → [(project_name, {devs}, project_id), ...] (как в find_core_matches)."""
+    out: dict[str, list[tuple[str, set[str], int | None]]] = {}
+    for row in db.execute(_OBJECTIVE_PROJECTS_SQL).all():
+        project_name = str(row[0])
+        developers = {str(d) for d in (row[1] or []) if d}
+        project_id = int(row[2]) if row[2] is not None else None
+        core = normalize_complex_name(project_name)
+        if not core:
+            continue
+        out.setdefault(core, []).append((project_name, developers, project_id))
+    return out
+
+
+def _geo_confirm(m: CoreMatch, address: str, distance_m: float, origin: str) -> GeoMatch:
+    return GeoMatch(
+        domrf_obj_id=m.domrf_obj_id,
+        domrf_comm_name=m.domrf_comm_name,
+        core=m.core,
+        objective_project_name=m.objective_project_name,
+        objective_project_id=m.objective_project_id,
+        address=address,
+        distance_m=distance_m,
+        origin=origin,
+    )
+
+
+def _geo_reject(
+    m: CoreMatch, origin: str, reason: str, distance_m: float | None = None
+) -> GeoReject:
+    return GeoReject(
+        domrf_obj_id=m.domrf_obj_id,
+        domrf_comm_name=m.domrf_comm_name,
+        objective_project_name=m.objective_project_name,
+        origin=origin,
+        reason=reason,
+        distance_m=distance_m,
+    )
+
+
+def apply_geo_matches(
+    db: Session,
+    report: GeoMatchReport,
+    *,
+    dry_run: bool = True,
+) -> dict[str, int]:
+    """Вставить report.confirmed в objective_complex_mapping (auto_core_geo_v6).
+
+    Та же per-row SAVEPOINT + ON CONFLICT (objective_complex_name, objective_group)
+    DO NOTHING механика, что в apply_core_matches. Каждая строка:
+    match_method='auto_core_geo_v6', match_score=0.80, is_reviewed=false,
+    note с дистанцией в метрах и origin (tier_b/ambiguous).
+
+    Args:
+        db: SQLAlchemy sync Session.
+        report: результат find_geo_matches().
+        dry_run: True (default) — только counts + первые 30 пар с дистанциями,
+                 БЕЗ insert.
+
+    Returns:
+        dict: inserted, conflict_skipped, error_skipped, confirmed_total.
+    """
+    counts = report.counts()
+    logger.info(
+        "apply_geo_matches (dry_run=%s): confirmed=%d rejected=%d dadata_calls=%d",
+        dry_run,
+        counts["confirmed"],
+        counts["rejected"],
+        counts["dadata_calls"],
+    )
+    for gm in report.confirmed[:30]:
+        logger.info(
+            "  confirm[%s]: obj_id=%s core=%r domrf=%r → objective=%r (%.0f м, addr=%r)",
+            gm.origin,
+            gm.domrf_obj_id,
+            gm.core,
+            gm.domrf_comm_name,
+            gm.objective_project_name,
+            gm.distance_m,
+            gm.address[:60],
+        )
+
+    if dry_run:
+        return {
+            "inserted": 0,
+            "conflict_skipped": 0,
+            "error_skipped": 0,
+            "confirmed_total": counts["confirmed"],
+        }
+
+    inserted = 0
+    conflict_skipped = 0
+    error_skipped = 0
+    for gm in report.confirmed:
+        note = f"#2177 geo-pass ({gm.origin}): {gm.distance_m:.0f} м to domrf point"
+        try:
+            with db.begin_nested():
+                result = db.execute(
+                    text(
+                        """
+                        INSERT INTO objective_complex_mapping
+                            (objective_complex_name, objective_project_id, domrf_obj_id,
+                             objective_group, match_method, match_score, is_reviewed, note)
+                        VALUES (
+                            CAST(:name AS text),
+                            CAST(:project_id AS bigint),
+                            CAST(:obj_id AS bigint),
+                            CAST(:group AS text),
+                            CAST(:method AS text),
+                            CAST(:score AS numeric),
+                            CAST(:reviewed AS boolean),
+                            CAST(:note AS text)
+                        )
+                        ON CONFLICT (objective_complex_name, objective_group) DO NOTHING
+                        """
+                    ),
+                    {
+                        "name": gm.objective_project_name,
+                        "project_id": gm.objective_project_id,
+                        "obj_id": gm.domrf_obj_id,
+                        "group": OBJECTIVE_GROUP,
+                        "method": GEO_MATCH_METHOD,
+                        "score": GEO_MATCH_SCORE,
+                        "reviewed": False,
+                        "note": note,
+                    },
+                )
+            if result.rowcount > 0:
+                inserted += 1
+            else:
+                conflict_skipped += 1
+        except Exception as e:
+            logger.warning(
+                "geo insert failed для %s ↔ %s: %s",
+                gm.domrf_comm_name,
+                gm.objective_project_name,
+                e,
+            )
+            error_skipped += 1
+
+    db.commit()
+    logger.info(
+        "apply_geo_matches done: inserted=%d conflict_skipped=%d error_skipped=%d",
+        inserted,
+        conflict_skipped,
+        error_skipped,
+    )
+    return {
+        "inserted": inserted,
+        "conflict_skipped": conflict_skipped,
+        "error_skipped": error_skipped,
+        "confirmed_total": counts["confirmed"],
+    }
+
+
+def run_geo_pass(
+    *, apply: bool = False, max_distance_m: float = GEO_MAX_DISTANCE_M
+) -> dict[str, int]:
+    """CLI-обёртка geo-pass: открыть Session, найти гео-матчи, (опц.) применить.
+
+    Дефолт — dry-run. Реальный insert только при apply=True.
+    """
+    from app.core.db import SessionLocal
+
+    db = SessionLocal()
+    try:
+        report = find_geo_matches(db, max_distance_m=max_distance_m)
+        return apply_geo_matches(db, report, dry_run=not apply)
+    finally:
+        db.close()
+
+
 def _main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(
-        description="objective_backfill core-pass (#2177 step2, auto_core_dev_v5)."
+        description="objective_backfill passes (#2177): core-pass (auto_core_dev_v5), "
+        "geo-pass (auto_core_geo_v6)."
     )
     parser.add_argument(
         "--core-pass",
         action="store_true",
-        help="Запустить core-name + dev-confirm проход (иначе no-op).",
+        help="Запустить core-name + dev-confirm проход (auto_core_dev_v5).",
+    )
+    parser.add_argument(
+        "--geo-pass",
+        action="store_true",
+        help="Запустить DaData-гео проход по tier_b/ambiguous остатку core-pass "
+        "(auto_core_geo_v6).",
+    )
+    parser.add_argument(
+        "--max-distance-m",
+        type=float,
+        default=GEO_MAX_DISTANCE_M,
+        help="Радиус подтверждения гео-матча в метрах (geo-pass; "
+        f"default {GEO_MAX_DISTANCE_M:.0f}).",
     )
     parser.add_argument(
         "--apply",
         action="store_true",
-        help="Реально писать tier_a в БД (по умолчанию — dry-run).",
+        help="Реально писать матчи в БД (по умолчанию — dry-run).",
     )
     args = parser.parse_args(argv)
 
@@ -658,12 +1146,18 @@ def _main(argv: list[str] | None = None) -> int:
         format="%(asctime)s %(levelname)s %(name)s: %(message)s",
     )
 
-    if not args.core_pass:
+    if not args.core_pass and not args.geo_pass:
         parser.print_help()
         return 0
 
-    result = run_core_pass(apply=args.apply)
+    if args.core_pass:
-    logger.info("core-pass result: %s", result)
+        result = run_core_pass(apply=args.apply)
+        logger.info("core-pass result: %s", result)
+
+    if args.geo_pass:
+        geo_result = run_geo_pass(apply=args.apply, max_distance_m=args.max_distance_m)
+        logger.info("geo-pass result: %s", geo_result)
+
     return 0
 
 

backend/tests/services/test_dadata_client.py

@@ -0,0 +1,92 @@
+"""Тесты для dadata_client.clean_address (#2177 geo-pass).
+
+httpx.Client замокан — реальная сеть/DaData не дёргается. settings-credentials
+патчатся через monkeypatch (литерала ключа нигде нет).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import httpx
+import pytest
+
+from app.services import dadata_client
+from app.services.dadata_client import clean_address
+
+
+@pytest.fixture(autouse=True)
+def _creds(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Задаём фейковые credentials, чтобы clean_address не отваливался на guard."""
+    monkeypatch.setattr(dadata_client.settings, "dadata_api_token", "test-token")
+    monkeypatch.setattr(dadata_client.settings, "dadata_api_secret", "test-secret")
+
+
+def _mock_client(response: MagicMock) -> MagicMock:
+    """httpx.Client() context-manager, post() → response."""
+    client = MagicMock()
+    client.__enter__ = MagicMock(return_value=client)
+    client.__exit__ = MagicMock(return_value=False)
+    client.post.return_value = response
+    return client
+
+
+def _resp(status: int, json_body: Any) -> MagicMock:
+    resp = MagicMock()
+    resp.status_code = status
+    resp.json.return_value = json_body
+    resp.text = ""
+    return resp
+
+
+def test_clean_address_returns_coords() -> None:
+    """Успешный ответ с geo_lat/geo_lon → (lat, lon) float-пара."""
+    resp = _resp(200, [{"geo_lat": "56.8389", "geo_lon": "60.6057", "qc_geo": 0}])
+    with patch("httpx.Client", return_value=_mock_client(resp)):
+        result = clean_address("г Екатеринбург, ул Малышева, 125")
+    assert result == (56.8389, 60.6057)
+
+
+def test_clean_address_no_coords_returns_none() -> None:
+    """Ответ без geo_lat/geo_lon (адрес не геокодирован) → None."""
+    resp = _resp(200, [{"qc_geo": 5, "result": None}])
+    with patch("httpx.Client", return_value=_mock_client(resp)):
+        assert clean_address("нераспознаваемо") is None
+
+
+def test_clean_address_empty_array_returns_none() -> None:
+    """Пустой массив → None."""
+    resp = _resp(200, [])
+    with patch("httpx.Client", return_value=_mock_client(resp)):
+        assert clean_address("г Екатеринбург, ул X, 1") is None
+
+
+def test_clean_address_short_input_returns_none() -> None:
+    """Короткий/пустой адрес → None без сетевого вызова."""
+    assert clean_address("") is None
+    assert clean_address("ab") is None
+
+
+def test_clean_address_no_credentials_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Нет credentials → graceful None (geo-pass reject'ит, не падает)."""
+    monkeypatch.setattr(dadata_client.settings, "dadata_api_token", None)
+    monkeypatch.setattr(dadata_client.settings, "dadata_api_secret", None)
+    assert clean_address("г Екатеринбург, ул Малышева, 125") is None
+
+
+def test_clean_address_429_returns_none() -> None:
+    """HTTP 429 quota → None."""
+    resp = _resp(429, {})
+    with patch("httpx.Client", return_value=_mock_client(resp)):
+        assert clean_address("г Екатеринбург, ул X, 1") is None
+
+
+def test_clean_address_timeout_returns_none() -> None:
+    """Сетевой timeout → None, не пробрасываем исключение."""
+    client = MagicMock()
+    client.__enter__ = MagicMock(return_value=client)
+    client.__exit__ = MagicMock(return_value=False)
+    client.post.side_effect = httpx.TimeoutException("timed out")
+    with patch("httpx.Client", return_value=client):
+        assert clean_address("г Екатеринбург, ул X, 1") is None

backend/tests/services/test_objective_backfill.py

@@ -12,14 +12,21 @@ from app.services.etl.objective_backfill import (
     AUTO_ACCEPT_THRESHOLD,
     CORE_MATCH_METHOD,
     CORE_MATCH_SCORE,
+    GEO_DADATA_CALL_LIMIT,
+    GEO_MATCH_METHOD,
+    GEO_MATCH_SCORE,
     REVIEW_THRESHOLD,
     CoreMatchReport,
+    GeoMatchReport,
     MatchCandidate,
     _dev_confirms,
+    _haversine_m,
     _norm_dev,
     apply_core_matches,
+    apply_geo_matches,
     auto_apply_matches,
     find_core_matches,
+    find_geo_matches,
     find_match_candidates,
     trigger_mv_refresh,
 )
@@ -440,3 +447,381 @@ def test_apply_core_matches_ignores_non_tier_a() -> None:
     mock_db.execute.assert_not_called()
     # commit всё равно вызывается (пустой tier_a — no-op транзакция)
     mock_db.commit.assert_called_once()
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Geo-pass (#2177) — auto_core_geo_v6
+# ═════════════════════════════════════════════════════════════════════════════
+
+# Патч-путь: clean_address импортирован в namespace objective_backfill.
+_CLEAN_ADDRESS_PATH = "app.services.etl.objective_backfill.clean_address"
+
+# Опорная domrf-точка для тестов (центр ЕКБ, произвольная).
+_DOMRF_LAT = 56.8389
+_DOMRF_LON = 60.6057
+
+
+def _offset_coords(base_lat: float, base_lon: float, meters_north: float) -> tuple[float, float]:
+    """Точка в meters_north к северу от base (для управляемой дистанции)."""
+    # 1° широты ≈ 111_320 м. Сдвиг только по широте — долготу оставляем.
+    return (base_lat + meters_north / 111_320.0, base_lon)
+
+
+def _make_geo_db(
+    objective_rows: list[tuple[str, list[str] | None, int | None]],
+    taken_names: list[str],
+    domrf_rows: list[tuple[int, str, str | None]],
+    address_rows: list[tuple[str, str]],
+    coord_rows: list[tuple[int, float, float]],
+) -> MagicMock:
+    """Mock Session для find_geo_matches — 6 db.execute().all() по порядку:
+
+    find_core_matches: 1) _OBJECTIVE_PROJECTS_SQL, 2) _TAKEN_NAMES_SQL,
+                       3) _DOMRF_UNMAPPED_SQL
+    geo:               4) _OBJECTIVE_ADDRESS_SQL (адреса per project),
+                       5) _OBJECTIVE_PROJECTS_SQL (повтор, _load_objective_by_core),
+                       6) _DOMRF_COORDS_SQL (координаты domrf per obj_id).
+    """
+
+    def _result(rows: list[Any]) -> MagicMock:
+        r = MagicMock()
+        r.all.return_value = rows
+        return r
+
+    mock_db = MagicMock()
+    mock_db.execute.side_effect = [
+        _result(objective_rows),
+        _result([(n,) for n in taken_names]),
+        _result(domrf_rows),
+        _result(address_rows),
+        _result(objective_rows),
+        _result(coord_rows),
+    ]
+    return mock_db
+
+
+# ── haversine ────────────────────────────────────────────────────────────────
+
+
+def test_haversine_zero_and_known_distance() -> None:
+    """haversine: одна точка → 0; сдвиг 120 м → ≈120 м (±5 м)."""
+    assert _haversine_m(_DOMRF_LAT, _DOMRF_LON, _DOMRF_LAT, _DOMRF_LON) == 0.0
+    lat2, lon2 = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 120.0)
+    dist = _haversine_m(_DOMRF_LAT, _DOMRF_LON, lat2, lon2)
+    assert abs(dist - 120.0) < 5.0
+
+
+# ── tier_b confirm / reject по дистанции ─────────────────────────────────────
+
+
+def test_geo_tier_b_confirm_within_radius() -> None:
+    """tier_b, адрес в 120 м от domrf → confirm."""
+    addr_lat, addr_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 120.0)
+    mock_db = _make_geo_db(
+        objective_rows=[("Арсенал", ["Брозэкс"], 7002)],
+        taken_names=[],
+        domrf_rows=[(202, 'ЖК "Арсенал"', "ООО СЗ Арсенал")],  # dev не совпал → tier_b
+        address_rows=[("Арсенал", "г Екатеринбург, ул Арсенальная, 1")],
+        coord_rows=[(202, _DOMRF_LAT, _DOMRF_LON)],
+    )
+
+    with patch(_CLEAN_ADDRESS_PATH, return_value=(addr_lat, addr_lon)) as mock_geo:
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert len(report.confirmed) == 1
+    gm = report.confirmed[0]
+    assert gm.domrf_obj_id == 202
+    assert gm.objective_project_name == "Арсенал"
+    assert gm.origin == "tier_b"
+    assert abs(gm.distance_m - 120.0) < 5.0
+    assert report.rejected == []
+    assert report.dadata_calls == 1
+    mock_geo.assert_called_once()
+
+
+def test_geo_tier_b_reject_too_far() -> None:
+    """tier_b, адрес в 900 м → reject с дистанцией (за пределами 500 м)."""
+    addr_lat, addr_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 900.0)
+    mock_db = _make_geo_db(
+        objective_rows=[("Арсенал", ["Брозэкс"], 7002)],
+        taken_names=[],
+        domrf_rows=[(202, 'ЖК "Арсенал"', "ООО СЗ Арсенал")],
+        address_rows=[("Арсенал", "г Екатеринбург, ул Дальняя, 99")],
+        coord_rows=[(202, _DOMRF_LAT, _DOMRF_LON)],
+    )
+
+    with patch(_CLEAN_ADDRESS_PATH, return_value=(addr_lat, addr_lon)):
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert report.confirmed == []
+    assert len(report.rejected) == 1
+    rej = report.rejected[0]
+    assert rej.reason == "too_far"
+    assert rej.distance_m is not None and abs(rej.distance_m - 900.0) < 10.0
+
+
+# ── ambiguous: резолв на ближнего / остаётся при двух в радиусе ──────────────
+
+
+def test_geo_ambiguous_resolves_to_nearest() -> None:
+    """ambiguous, 2 кандидата ядра: один в 150 м, другой в 5 км → резолв на ближнего."""
+    near_lat, near_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 150.0)
+    far_lat, far_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 5000.0)
+
+    mock_db = _make_geo_db(
+        objective_rows=[
+            ("Меридиан", ["Девелопер"], 7003),
+            ("Меридиан ", ["Другой"], 7004),  # то же ядро «меридиан»
+        ],
+        taken_names=[],
+        domrf_rows=[(303, 'ЖК "Меридиан"', "ООО Девелопер")],
+        address_rows=[
+            ("Меридиан", "г Екатеринбург, ул Близкая, 1"),
+            ("Меридиан ", "г Екатеринбург, ул Далёкая, 2"),
+        ],
+        coord_rows=[(303, _DOMRF_LAT, _DOMRF_LON)],
+    )
+
+    def _fake_geocode(addr: str) -> tuple[float, float]:
+        if "Близкая" in addr:
+            return (near_lat, near_lon)
+        return (far_lat, far_lon)
+
+    with patch(_CLEAN_ADDRESS_PATH, side_effect=_fake_geocode):
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert len(report.confirmed) == 1
+    gm = report.confirmed[0]
+    assert gm.objective_project_name == "Меридиан"  # ближний
+    assert gm.objective_project_id == 7003
+    assert gm.origin == "ambiguous"
+    assert abs(gm.distance_m - 150.0) < 10.0
+
+
+def test_geo_ambiguous_partial_geocode_does_not_blind_confirm() -> None:
+    """ambiguous, один кандидат геокодится в 100 м, второй НЕ геокодится вовсе →
+    НЕ confirm (reject partial_geocode): «один в радиусе» — артефакт отсутствия
+    данных, сосед мог быть ближе (ревью 2026-07-03)."""
+    near_lat, near_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 100.0)
+
+    mock_db = _make_geo_db(
+        objective_rows=[
+            ("Меридиан", ["Девелопер"], 7003),
+            ("Меридиан ", ["Другой"], 7004),  # то же ядро
+        ],
+        taken_names=[],
+        domrf_rows=[(303, 'ЖК "Меридиан"', "ООО Девелопер")],
+        address_rows=[
+            ("Меридиан", "г Екатеринбург, ул Близкая, 1"),
+            ("Меридиан ", "г Екатеринбург, ул Негеокодимая, 2"),
+        ],
+        coord_rows=[(303, _DOMRF_LAT, _DOMRF_LON)],
+    )
+
+    def _fake_geocode(addr: str) -> tuple[float, float] | None:
+        if "Близкая" in addr:
+            return (near_lat, near_lon)
+        return None  # сосед не геокодится
+
+    with patch(_CLEAN_ADDRESS_PATH, side_effect=_fake_geocode):
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert report.confirmed == []
+    reasons = {r.reason for r in report.rejected}
+    assert "partial_geocode" in reasons
+
+
+def test_geo_ambiguous_two_in_radius_stays_ambiguous() -> None:
+    """ambiguous, оба кандидата в радиусе → НЕ резолвим, остаётся ambiguous."""
+    a_lat, a_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 100.0)
+    b_lat, b_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 200.0)
+
+    mock_db = _make_geo_db(
+        objective_rows=[
+            ("Меридиан", ["Девелопер"], 7003),
+            ("Меридиан ", ["Другой"], 7004),
+        ],
+        taken_names=[],
+        domrf_rows=[(303, 'ЖК "Меридиан"', "ООО Девелопер")],
+        address_rows=[
+            ("Меридиан", "г Екатеринбург, ул А, 1"),
+            ("Меридиан ", "г Екатеринбург, ул Б, 2"),
+        ],
+        coord_rows=[(303, _DOMRF_LAT, _DOMRF_LON)],
+    )
+
+    def _fake_geocode(addr: str) -> tuple[float, float]:
+        return (a_lat, a_lon) if "ул А" in addr else (b_lat, b_lon)
+
+    with patch(_CLEAN_ADDRESS_PATH, side_effect=_fake_geocode):
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert report.confirmed == []
+    assert len(report.rejected) == 1
+    assert report.rejected[0].reason == "ambiguous_multi"
+
+
+# ── no address / no geocode → reject, не падаем ──────────────────────────────
+
+
+def test_geo_no_address_rejects_without_geocode_call() -> None:
+    """tier_b без адреса → reject 'no_address', DaData не дёргается."""
+    mock_db = _make_geo_db(
+        objective_rows=[("Арсенал", ["Брозэкс"], 7002)],
+        taken_names=[],
+        domrf_rows=[(202, 'ЖК "Арсенал"', "ООО СЗ Арсенал")],
+        address_rows=[],  # адреса нет
+        coord_rows=[(202, _DOMRF_LAT, _DOMRF_LON)],
+    )
+
+    with patch(_CLEAN_ADDRESS_PATH, return_value=(1.0, 2.0)) as mock_geo:
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert report.confirmed == []
+    assert len(report.rejected) == 1
+    assert report.rejected[0].reason == "no_address"
+    assert report.dadata_calls == 0
+    mock_geo.assert_not_called()
+
+
+def test_geo_null_geocode_rejects_no_crash() -> None:
+    """tier_b, DaData вернул None → reject 'no_geocode', не падаем."""
+    mock_db = _make_geo_db(
+        objective_rows=[("Арсенал", ["Брозэкс"], 7002)],
+        taken_names=[],
+        domrf_rows=[(202, 'ЖК "Арсенал"', "ООО СЗ Арсенал")],
+        address_rows=[("Арсенал", "нераспознаваемый адрес")],
+        coord_rows=[(202, _DOMRF_LAT, _DOMRF_LON)],
+    )
+
+    with patch(_CLEAN_ADDRESS_PATH, return_value=None):
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert report.confirmed == []
+    assert len(report.rejected) == 1
+    assert report.rejected[0].reason == "no_geocode"
+    assert report.dadata_calls == 1
+
+
+# ── лимит вызовов DaData соблюдается ─────────────────────────────────────────
+
+
+def test_geo_call_limit_respected() -> None:
+    """При GEO_DADATA_CALL_LIMIT+2 tier_b с уникальными адресами делаем ровно LIMIT
+    вызовов, остаток reject c call_limit_hit=True."""
+    n = GEO_DADATA_CALL_LIMIT + 2
+    objective_rows = [(f"Проект{i}", [f"Dev{i}"], 8000 + i) for i in range(n)]
+    domrf_rows = [(9000 + i, f'ЖК "Проект{i}"', f"ООО Иное{i}") for i in range(n)]
+    address_rows = [(f"Проект{i}", f"г Екатеринбург, ул {i}, дом {i}") for i in range(n)]
+    coord_rows = [(9000 + i, _DOMRF_LAT, _DOMRF_LON) for i in range(n)]
+
+    mock_db = _make_geo_db(
+        objective_rows=objective_rows,
+        taken_names=[],
+        domrf_rows=domrf_rows,
+        address_rows=address_rows,
+        coord_rows=coord_rows,
+    )
+
+    call_counter = {"n": 0}
+
+    def _fake_geocode(addr: str) -> tuple[float, float]:
+        call_counter["n"] += 1
+        return (_DOMRF_LAT, _DOMRF_LON)  # всё в 0 м → confirm
+
+    with patch(_CLEAN_ADDRESS_PATH, side_effect=_fake_geocode):
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert report.dadata_calls == GEO_DADATA_CALL_LIMIT
+    assert call_counter["n"] == GEO_DADATA_CALL_LIMIT
+    assert report.call_limit_hit is True
+    # 2 кандидата за лимитом → reject 'call_limit'
+    call_limit_rejects = [r for r in report.rejected if r.reason == "call_limit"]
+    assert len(call_limit_rejects) == 2
+
+
+# ── apply_geo_matches: dry-run / apply ───────────────────────────────────────
+
+
+def _confirmed_geo_report() -> GeoMatchReport:
+    """Отчёт с одним confirmed tier_b гео-матчем (150 м)."""
+    addr_lat, addr_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 150.0)
+    mock_db = _make_geo_db(
+        objective_rows=[("Арсенал", ["Брозэкс"], 7002)],
+        taken_names=[],
+        domrf_rows=[(202, 'ЖК "Арсенал"', "ООО СЗ Арсенал")],
+        address_rows=[("Арсенал", "г Екатеринбург, ул Арсенальная, 1")],
+        coord_rows=[(202, _DOMRF_LAT, _DOMRF_LON)],
+    )
+    with patch(_CLEAN_ADDRESS_PATH, return_value=(addr_lat, addr_lon)):
+        return find_geo_matches(mock_db, max_distance_m=500)
+
+
+def test_apply_geo_matches_dry_run_no_write() -> None:
+    """dry_run=True → только counts, execute/commit не вызываются."""
+    report = _confirmed_geo_report()
+    assert len(report.confirmed) == 1
+
+    mock_db = MagicMock()
+    result = apply_geo_matches(mock_db, report, dry_run=True)
+
+    assert result["inserted"] == 0
+    assert result["confirmed_total"] == 1
+    mock_db.execute.assert_not_called()
+    mock_db.commit.assert_not_called()
+
+
+def test_apply_geo_matches_writes_v6_score_note() -> None:
+    """apply пишет confirmed с method=auto_core_geo_v6, score=0.80, note-дистанцией."""
+    report = _confirmed_geo_report()
+
+    savepoint_cm = MagicMock()
+    savepoint_cm.__enter__ = MagicMock(return_value=None)
+    savepoint_cm.__exit__ = MagicMock(return_value=False)
+
+    exec_result = MagicMock()
+    exec_result.rowcount = 1
+
+    mock_db = MagicMock()
+    mock_db.begin_nested.return_value = savepoint_cm
+    mock_db.execute.return_value = exec_result
+
+    result = apply_geo_matches(mock_db, report, dry_run=False)
+
+    assert result["inserted"] == 1
+    assert result["conflict_skipped"] == 0
+    mock_db.commit.assert_called_once()
+
+    params = mock_db.execute.call_args[0][1]
+    assert params["method"] == GEO_MATCH_METHOD
+    assert params["score"] == GEO_MATCH_SCORE
+    assert params["reviewed"] is False
+    assert params["group"] == "Екатеринбург"
+    assert params["name"] == "Арсенал"
+    assert params["obj_id"] == 202
+    assert params["project_id"] == 7002
+    # note несёт дистанцию в метрах + origin
+    assert "м" in params["note"]
+    assert "tier_b" in params["note"]
+
+
+def test_apply_geo_matches_on_conflict_counts_skip() -> None:
+    """ON CONFLICT DO NOTHING (rowcount=0) → conflict_skipped, не падает."""
+    report = _confirmed_geo_report()
+
+    savepoint_cm = MagicMock()
+    savepoint_cm.__enter__ = MagicMock(return_value=None)
+    savepoint_cm.__exit__ = MagicMock(return_value=False)
+
+    exec_result = MagicMock()
+    exec_result.rowcount = 0
+
+    mock_db = MagicMock()
+    mock_db.begin_nested.return_value = savepoint_cm
+    mock_db.execute.return_value = exec_result
+
+    result = apply_geo_matches(mock_db, report, dry_run=False)
+
+    assert result["inserted"] == 0
+    assert result["conflict_skipped"] == 1
+    mock_db.commit.assert_called_once()