From 18a2cf0d2e259fa334d3e0906a9e6aba014c3f47 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 3 Jul 2026 06:04:32 +0500 Subject: [PATCH] =?UTF-8?q?feat(etl):=20=D0=B3=D0=B5=D0=BE-=D0=BF=D1=80?= =?UTF-8?q?=D0=BE=D1=85=D0=BE=D0=B4=20auto=5Fcore=5Fgeo=5Fv6=20(#2177=20?= =?UTF-8?q?=D0=BE=D1=81=D1=82=D0=B0=D1=82=D0=BE=D0=BA)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/core/config.py | 13 + backend/app/services/dadata_client.py | 135 +++++ .../app/services/etl/objective_backfill.py | 506 +++++++++++++++++- backend/tests/services/test_dadata_client.py | 92 ++++ .../tests/services/test_objective_backfill.py | 385 +++++++++++++ 5 files changed, 1125 insertions(+), 6 deletions(-) create mode 100644 backend/app/services/dadata_client.py create mode 100644 backend/tests/services/test_dadata_client.py diff --git a/backend/app/core/config.py b/backend/app/core/config.py index d45d0ab9..0bd88705 100644 --- a/backend/app/core/config.py +++ b/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() diff --git a/backend/app/services/dadata_client.py b/backend/app/services/dadata_client.py new file mode 100644 index 00000000..c74d2aeb --- /dev/null +++ b/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) diff --git a/backend/app/services/etl/objective_backfill.py b/backend/app/services/etl/objective_backfill.py index 10e06dd9..e678c5d9 100644 --- a/backend/app/services/etl/objective_backfill.py +++ b/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) - logger.info("core-pass result: %s", result) + if args.core_pass: + 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 diff --git a/backend/tests/services/test_dadata_client.py b/backend/tests/services/test_dadata_client.py new file mode 100644 index 00000000..5cbac20e --- /dev/null +++ b/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 diff --git a/backend/tests/services/test_objective_backfill.py b/backend/tests/services/test_objective_backfill.py index 6a77cf21..22551611 100644 --- a/backend/tests/services/test_objective_backfill.py +++ b/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()