fix(ptica): гео-проход не тратит геокод на занятых кандидатов и не «подтверждает» их (#2464)

Разбирая неоднозначные ядра, find_geo_matches геокодил ВСЕХ objective-кандидатов,
включая тех, чьё имя уже занято в objective_complex_mapping. Записать такое имя
нельзя в принципе: apply_geo_matches вставляет с
ON CONFLICT (objective_complex_name, objective_group) DO NOTHING, а _TAKEN_NAMES_SQL
выбирает ровно по этому ключу.

Последствия на боевом пути:
- занятый кандидат мог оказаться единственным в радиусе и уходил в confirmed —
  прогон рапортовал подтверждение, за которым запись молча не делала ничего;
- занятый рядом со свободным давал «двое в радиусе» → ambiguous_multi, хотя
  выбор был единственным;
- квота DaData (лимит 200 вызовов за прогон) тратилась на заведомо непишущихся.

Замер на проде 19.08: 930 несопоставленных domrf ЕКБ-объектов, из них 14
неоднозначных; 28 слотов кандидатов, 14 занятых; 3 из 6 разных адресов к геокоду
принадлежали занятым. После отсева у всех 14 остаётся ровно один кандидат.

Заодно причина отказа перестаёт врать: при пустом in_radius это 'too_far', а не
'ambiguous_multi'; когда заняты все кандидаты — отдельная 'all_candidates_taken'.
После отсева случай «остался один кандидат» становится частым, и старая метка
читалась бы оператором как факт неоднозначности.

backend/app/services/etl/objective_backfill.py

@@ -364,12 +364,15 @@ class CoreMatchReport:
     ambiguous — >1 objective-кандидатов по core → в отчёт, разрешение вручную/гео.
     skipped_taken — objective_complex_name уже занят в mapping (UNIQUE-констрейнт;
                     его domrf-группа уже покрыта — дубли не нужны).
+    taken_names — сами занятые имена. Нужны гео-проходу (#2464): он разбирает
+                  ambiguous по ВСЕМ кандидатам ядра, а занятого записать нельзя.
     """
 
     tier_a: list[CoreMatch] = field(default_factory=list)
     tier_b: list[CoreMatch] = field(default_factory=list)
     ambiguous: list[CoreMatch] = field(default_factory=list)
     skipped_taken: list[CoreMatch] = field(default_factory=list)
+    taken_names: set[str] = field(default_factory=set)
 
     def counts(self) -> dict[str, int]:
         return {
@@ -456,6 +459,7 @@ def find_core_matches(db: Session) -> CoreMatchReport:
     taken_names: set[str] = {
         str(r[0]) for r in db.execute(_TAKEN_NAMES_SQL, {"group": OBJECTIVE_GROUP}).all()
     }
+    report.taken_names = taken_names
 
     # domrf-сторона: несопоставленные ЕКБ, latest snapshot per obj_id
     for row in db.execute(_DOMRF_UNMAPPED_SQL).all():
@@ -696,7 +700,8 @@ class GeoMatch:
 @dataclass
 class GeoReject:
     """Отклонённый гео-кандидат (для отчёта). reason: 'no_address' |
-    'no_geocode' | 'too_far' | 'ambiguous_multi' | 'call_limit'.
+    'no_geocode' | 'too_far' | 'ambiguous_multi' | 'partial_geocode' |
+    'all_candidates_taken' | 'call_limit'.
 
     distance_m None когда дистанцию посчитать не удалось (нет адреса/геокода/
     координат domrf).
@@ -820,6 +825,7 @@ def find_geo_matches(db: Session, *, max_distance_m: float = GEO_MAX_DISTANCE_M)
 
     tier_b = core_report.tier_b
     ambiguous = core_report.ambiguous
+    taken_names = core_report.taken_names
     if not tier_b and not ambiguous:
         logger.info("find_geo_matches: нет tier_b/ambiguous кандидатов — nothing to do")
         return report
@@ -890,7 +896,19 @@ def find_geo_matches(db: Session, *, max_distance_m: float = GEO_MAX_DISTANCE_M)
         if domrf_pt is None:
             report.rejected.append(_geo_reject(m, "ambiguous", "no_geocode"))
             continue
-        candidates = objective_by_core.get(m.core, [])
+        # Занятые objective-имена отсеиваем ДО геокода. Записать такое имя
+        # нельзя в принципе: apply_geo_matches вставляет с
+        # ON CONFLICT (objective_complex_name, objective_group) DO NOTHING, а
+        # _TAKEN_NAMES_SQL выбирает ровно по этому ключу. Раньше занятый кандидат
+        # мог оказаться единственным в радиусе и уходил в confirmed — прогон
+        # рапортовал подтверждение, которого запись затем молча не делала.
+        # Замер на проде 19.08: 14 неоднозначных строк (из 930 несопоставленных),
+        # 28 слотов кандидатов, из них 14 занятых; 3 адреса из 6 к геокоду —
+        # занятых. После отсева у всех 14 остаётся ровно один кандидат.
+        candidates = [c for c in objective_by_core.get(m.core, []) if c[0] not in taken_names]
+        if not candidates:
+            report.rejected.append(_geo_reject(m, "ambiguous", "all_candidates_taken"))
+            continue
         in_radius: list[tuple[str, int | None, str, float]] = []
         any_geocoded = False
         geocoded_count = 0
@@ -934,9 +952,12 @@ def find_geo_matches(db: Session, *, max_distance_m: float = GEO_MAX_DISTANCE_M)
             reason = "call_limit" if report.call_limit_hit else "no_geocode"
             report.rejected.append(_geo_reject(m, "ambiguous", reason))
         else:
-            # 0 в радиусе, или >1 в радиусе → остаётся ambiguous
+            # Отделяем «никто не близко» от «близко несколько». После отсева
+            # занятых кандидат часто остаётся один, и метка ambiguous_multi при
+            # пустом in_radius была бы прямой неправдой в отчёте оператору.
             nearest = min((d for *_, d in in_radius), default=None)
-            report.rejected.append(_geo_reject(m, "ambiguous", "ambiguous_multi", nearest))
+            reason = "ambiguous_multi" if in_radius else "too_far"
+            report.rejected.append(_geo_reject(m, "ambiguous", reason, nearest))
 
     logger.info(
         "find_geo_matches: %s call_limit_hit=%s",

backend/tests/services/test_objective_backfill.py

@@ -850,3 +850,105 @@ def test_apply_geo_matches_on_conflict_counts_skip() -> None:
     assert result["inserted"] == 0
     assert result["conflict_skipped"] == 1
     mock_db.commit.assert_called_once()
+
+
+# ── ambiguous: занятые кандидаты отсеиваются до геокода (#2464) ──────────────
+#
+# Записать занятое objective-имя нельзя: apply_geo_matches вставляет с
+# ON CONFLICT (objective_complex_name, objective_group) DO NOTHING, а
+# _TAKEN_NAMES_SQL выбирает ровно по этому ключу. Поэтому геокод занятого — трата
+# квоты DaData, а его попадание в confirmed — отчёт о подтверждении, за которым
+# нет записи.
+
+
+def test_geo_ambiguous_ignores_taken_candidate_and_resolves() -> None:
+    """Занятый кандидат ближе свободного → раньше оба были «в радиусе» и случай
+    оставался ambiguous. Занятый не участвует → свободный резолвится."""
+    taken_lat, taken_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 100.0)
+    free_lat, free_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)],
+    )
+
+    geocoded: list[str] = []
+
+    def _fake_geocode(addr: str) -> tuple[float, float]:
+        geocoded.append(addr)
+        return (taken_lat, taken_lon) if "Занятая" in addr else (free_lat, free_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
+    assert report.confirmed[0].objective_project_name == 'Бутик-квартал "Меридиан"'
+    # квота DaData не тратится на кандидата, которого запись всё равно отбросит
+    assert not any("Занятая" in a for a in geocoded)
+
+
+def test_geo_ambiguous_taken_candidate_not_confirmed_when_only_one_near() -> None:
+    """Занятый в радиусе, свободный далеко: раньше подтверждался ЗАНЯТЫЙ и прогон
+    рапортовал confirmed при нуле записей. Теперь честный reject 'too_far'."""
+    taken_lat, taken_lon = _offset_coords(_DOMRF_LAT, _DOMRF_LON, 100.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]:
+        return (taken_lat, taken_lon) if "Занятая" in addr else (far_lat, far_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
+    # 'ambiguous_multi' здесь был бы неправдой: близко не «несколько», а никто
+    assert report.rejected[0].reason == "too_far"
+
+
+def test_geo_ambiguous_all_candidates_taken_rejects_without_geocode() -> None:
+    """Все кандидаты ядра заняты → отдельная причина отказа и ни одного вызова
+    DaData (раньше геокодились оба и случай выглядел как ambiguous_multi)."""
+    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)],
+    )
+
+    with patch(_CLEAN_ADDRESS_PATH, side_effect=AssertionError("DaData не должна вызываться")):
+        report = find_geo_matches(mock_db, max_distance_m=500)
+
+    assert report.confirmed == []
+    assert len(report.rejected) == 1
+    assert report.rejected[0].reason == "all_candidates_taken"
+    assert report.dadata_calls == 0