Merge pull request 'fix(tradein/geocode): прошить city_hint в deals-скрипт + развести счётчики гейта (#2603)' (#2655) from fix/2603-geocode-city-hint-tails into main

tradein-mvp/backend/app/api/v1/admin.py

@@ -72,7 +72,7 @@ from app.services import cian_session as cian_session_svc
 from app.services import domclick_session as domclick_session_svc
 from app.services import proxy_rotation as proxy_rotation_svc
 from app.services import scrape_runs as runs_mod
-from app.services.geocoder import geocode
+from app.services.geocoder import geocode, known_city_hint
 from app.services.proxy_pool import clear_source_bans
 from app.services.scheduler import has_running_run
 from app.services.scraper_adapters import (
@@ -319,7 +319,14 @@ async def geocode_missing(
         # развёртки/импорта. Прокидываем как city_hint, а не полагаемся на то, что
         # геокодер угадает город по тексту address (голый "ул. Победы, 30" без
         # города в тексте иначе уходит в Екатеринбург).
-        city = row.get("city")
+        #
+        # known_city_hint (#2603) — гейт по словарю городов области: при
+        # target="deals" сюда приходит росреестровое поле, в хвосте которого
+        # лежат не-города («Бессонова», «Билейский рыбопитомник»), а мусорный
+        # хинт закрывает EKB-локальные тиры и уезжает префиксом в запрос
+        # провайдеру, т.е. вреднее отсутствия хинта. Общий хелпер, тот же, что у
+        # scripts/geocode_deals_nominatim.py и tasks/geocode_missing.py.
+        city = known_city_hint(row.get("city"))
         result = await geocode(clean, db, city_hint=city)
         if result is None:
             # Помечаем что пробовали — иначе ретрай на каждом cron.

tradein-mvp/backend/app/services/geocoder.py

@@ -156,6 +156,41 @@ SVERDLOVSK_OBLAST_CITIES = frozenset(
         # без district-префикса ложно ушёл бы в non-EKB gate.
     }
 )
+
+
+def known_city_hint(value: str | None) -> str | None:
+    """`value` как city_hint, если это узнаваемое имя города региона 66, иначе None.
+
+    Для callers, которые берут город из КОЛОНКИ БД и передают его в `geocode()`
+    (#2603): `deals.city` — росреестровое поле, заполнено на 100%, но в хвосте
+    распределения лежит мусор («Бессонова», «Бердюгина», «Билейский
+    рыбопитомник» — улицы/урочища, попавшие в поле города). Мусорный хинт хуже
+    отсутствия хинта: любой не-ЕКБ `city_hint` жёстко закрывает EKB-локальные
+    тиры (`_ekb_local_tiers_allowed`) И подставляется префиксом в запрос
+    провайдеру (`_resolve_city_for_geocode`) — «Бессонова, Бессонова 10»
+    провайдер не резолвит вовсе.
+
+    Словарь — `SVERDLOVSK_OBLAST_CITIES`, тот же, на котором стоят городские
+    гейты самого геокодера. Отдельного списка сознательно НЕ заводим: город,
+    отсутствующий в этом наборе, и так обрабатывается геокодером как незнакомый
+    (`_names_non_ekb_city` его не увидит, `estimator._resolve_target_city` не
+    резолвит) — т.е. новый город области в любом случае добавляется СЮДА, и
+    гейт хинта не создаёт новой связности.
+
+    Цена решения (осознанная): легитимный, но не перечисленный населённый пункт
+    («Реж», «Арамиль», сёла/посёлки) хинта не получит и вернётся к поведению «по
+    тексту адреса» — то же, что было до прошивки хинта, без регрессии.
+
+    Пользовательский ввод (`/geocode/lookup`, `/geocode/suggest`,
+    `TradeInEstimateInput.city_hint`) сюда НЕ заворачиваем: там город назвал
+    человек, и молча его игнорировать нельзя — для произвольной строки
+    fail-closed отрабатывает `_ekb_local_tiers_allowed` (#2580/#2589).
+    """
+    if not value:
+        return None
+    return value if " ".join(value.lower().split()) in SVERDLOVSK_OBLAST_CITIES else None
+
+
 # Значение для DaData-констрейнта `locations: [{"region": ...}]`.
 # ВАЖНО: DaData хранит имя региона БЕЗ типа — `region="Свердловская"`,
 # `region_type="обл"` (тип лежит в отдельных полях `region_type` /

tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py

@@ -85,7 +85,17 @@ class BackfillCoordsResult:
     updated: int = 0  # реально обновлено (UPDATE rowcount)
     no_address: int = 0  # listing.address IS NULL / не распарсился
     no_match: int = 0  # адрес распарсился, но в реестре здания нет
-    skipped_non_ekb: int = 0  # non-ЕКБ гейт: колонка city (#2594) ИЛИ текст адреса (#2583)
+    skipped_non_ekb: int = 0  # non-ЕКБ гейт ВСЕГО: колонка city (#2594) ИЛИ текст (#2583)
+    # Подмножество skipped_non_ekb — только те, кого отсёк гейт по КОЛОНКЕ city
+    # (#2603). Зачем отдельный счётчик: колоночный гейт стоит ПЕРЕД парсером
+    # адреса, поэтому по мере раскатки областных развёрток (#2598) строки, которые
+    # сейчас падают в no_address (город неизвестен, адрес не парсится), начнут
+    # перетекать в skipped_non_ekb — и общий счётчик поменяет смысл ровно тогда,
+    # когда по нему хотят валидировать раскатку. Разность
+    # skipped_non_ekb - skipped_non_ekb_by_column = вклад ТЕКСТОВОГО гейта, т.е.
+    # старая метрика #2583 остаётся вычислимой. Обратная совместимость:
+    # skipped_non_ekb продолжает означать то же, что и раньше (гейт целиком).
+    skipped_non_ekb_by_column: int = 0
     errors: int = 0  # исключения при обработке отдельной записи
     duration_sec: float = field(default=0.0)
 
@@ -97,6 +107,7 @@ class BackfillCoordsResult:
             "no_address": self.no_address,
             "no_match": self.no_match,
             "skipped_non_ekb": self.skipped_non_ekb,
+            "skipped_non_ekb_by_column": self.skipped_non_ekb_by_column,
             "errors": self.errors,
             "duration_sec": int(self.duration_sec),
         }
@@ -227,6 +238,10 @@ def backfill_coords_from_geoportal(
                     city: str | None = row.get("city")
                     if city is not None and city != "Екатеринбург":
                         res.skipped_non_ekb += 1
+                        # Отдельный срез (#2603) — общий skipped_non_ekb смешивает
+                        # колоночный и текстовый гейты, а по мере раскатки #2598
+                        # колоночный будет забирать строки из no_address.
+                        res.skipped_non_ekb_by_column += 1
                         continue
 
                     # Текстовый гейт (#2583, H3) — fallback для листингов, у которых
@@ -313,13 +328,15 @@ def backfill_coords_from_geoportal(
 
     logger.info(
         "backfill_coords: DONE — candidates=%d matched=%d updated=%d "
-        "no_address=%d no_match=%d skipped_non_ekb=%d errors=%d duration=%.1fs",
+        "no_address=%d no_match=%d skipped_non_ekb=%d (by_column=%d) "
+        "errors=%d duration=%.1fs",
         res.candidates,
         res.matched,
         res.updated,
         res.no_address,
         res.no_match,
         res.skipped_non_ekb,
+        res.skipped_non_ekb_by_column,
         res.errors,
         res.duration_sec,
     )
@@ -369,7 +386,8 @@ def run_geoportal_coords_backfill(
         runs_mod.mark_done(db, run_id, counters)
         logger.info(
             "run_geoportal_coords_backfill: run_id=%d DONE candidates=%d matched=%d "
-            "updated=%d no_address=%d no_match=%d skipped_non_ekb=%d errors=%d duration=%.1fs",
+            "updated=%d no_address=%d no_match=%d skipped_non_ekb=%d (by_column=%d) "
+            "errors=%d duration=%.1fs",
             run_id,
             res.candidates,
             res.matched,
@@ -377,6 +395,7 @@ def run_geoportal_coords_backfill(
             res.no_address,
             res.no_match,
             res.skipped_non_ekb,
+            res.skipped_non_ekb_by_column,
             res.errors,
             res.duration_sec,
         )

tradein-mvp/backend/app/tasks/geocode_missing.py

@@ -41,7 +41,7 @@ from sqlalchemy.orm import Session
 
 from app.services import scrape_runs as runs_mod
 from app.services.estimator import _geocode_is_coarse
-from app.services.geocoder import geocode
+from app.services.geocoder import geocode, known_city_hint
 
 logger = logging.getLogger(__name__)
 
@@ -77,7 +77,8 @@ async def geocode_missing_listings(
        не попадут в выдачу пользователю)
 
     2. Для каждой пары (address, city):
-       - geocode(address, db, city_hint=city) — auto-cache (hit или miss)
+       - geocode(address, db, city_hint=known_city_hint(city)) — auto-cache
+         (hit или miss); хинт гейтится словарём городов области (#2603)
        - Если есть результат: UPDATE listings SET lat, lon
          WHERE address = :addr AND city IS NOT DISTINCT FROM :city AND lat IS NULL
          (IS NOT DISTINCT FROM, а не `=` — стандартная SQL NULL-семантика: `city = NULL`
@@ -160,7 +161,14 @@ async def geocode_missing_listings(
         result.addresses_processed += 1
 
         try:
-            geo = await geocode(address, db, city_hint=city)
+            # known_city_hint (#2603) — общий гейт по словарю городов области для
+            # всех DB-колоночных callers. Для listings.city он сегодня no-op
+            # (скрапер пишет только шесть кураторских имён из
+            # scraper_kit CITY_DISPLAY_NAMES, все они есть в словаре), но держит
+            # инвариант единым с deals-путями, где колонка росреестровая и в
+            # хвосте лежит мусор. Сырой `city` ниже остаётся ключом группы для
+            # UPDATE — гейт влияет только на подсказку геокодеру.
+            geo = await geocode(address, db, city_hint=known_city_hint(city))
         except Exception as exc:
             logger.warning("geocode_missing: geocode raised for '%s': %s", address[:60], exc)
             result.addresses_failed += 1
@@ -223,9 +231,13 @@ async def geocode_missing_listings(
         result.addresses_geocoded += 1
 
         if dry_run:
+            # city в логе (#2603) — с #2594 это часть ключа группы: без него две
+            # строки dry-run с одинаковым текстом адреса неотличимы друг от друга.
             logger.info(
-                "geocode_missing[dry]: '%s' → (%.5f, %.5f) provider=%s would update %d listings",
+                "geocode_missing[dry]: '%s' city=%r → (%.5f, %.5f) provider=%s "
+                "would update %d listings",
                 address[:60],
+                city,
                 geo.lat,
                 geo.lon,
                 geo.provider,

tradein-mvp/backend/scripts/geocode_deals_nominatim.py

@@ -18,12 +18,31 @@ result into `geocode_cache`. We do NOT call Nominatim directly — that keeps a
 single source of truth for provider order, ЕКБ bbox filtering, and the 1 req/sec
 policy.
 
-Why dedup by address (not one call per row)
+Why dedup by (address, city) — not one call per row
--------------------------------------------
+---------------------------------------------------
-`deals.address` is street-only ('Екатеринбург, <street>'), so thousands of rows
+`deals.address` is street-only ('<city>, <street>'), so thousands of rows share
-share the same address. The geocoder caches by normalized address, but we also
+the same address. The geocoder caches by normalized address, but we also GROUP
-GROUP BY address up front so the real call count is driven by DISTINCT streets,
+BY up front so the real call count is driven by DISTINCT streets, not the
-not the ~6,951 row backlog. One geocode call → UPDATE every deal on that street.
+~6,951 row backlog. One geocode call → UPDATE every deal on that street.
+
+Grouping is by the PAIR (address, city), not by address alone (#2603, same
+shape as #2601 fixed in app/tasks/geocode_missing.py): the same address text in
+two different cities must not collapse into one geocode call, and the UPDATE
+must not spill onto the other city's rows — hence `city IS NOT DISTINCT FROM`
+(plain `=` never matches a NULL city, so a NULL-city group would update nothing).
+
+City hint (and why it is filtered)
+----------------------------------
+`deals.city` comes from Rosreestr (migration 177) and is filled on 100% of the
+rows, so a hint is available for every group — but its long tail holds values
+that are not cities at all ('Бессонова', 'Бердюгина', 'Билейский рыбопитомник').
+A non-EKB hint is a HARD signal inside the geocoder: it closes the EKB-only
+local tiers (`_ekb_local_tiers_allowed`) and gets prefixed into the provider
+query, so a junk hint makes the result strictly WORSE than no hint at all.
+We therefore pass the hint only for values that are recognised cities of oblast
+66 — `geocoder.known_city_hint`, shared with the other DB-column callers
+(`app/api/v1/admin.py`, `app/tasks/geocode_missing.py`); everything else
+degrades to the previous behaviour (no hint, address text only).
 
 Street-level precision is accepted: the estimator's comparable search uses a
 1000-2000 m radius, so a street-level point lands every deal on that street in
@@ -89,12 +108,13 @@ try:
     from app.services.geocoder import (  # type: ignore[import-not-found]
         GeocodeResult,
         geocode,
+        known_city_hint,
     )
 except ImportError:  # pragma: no cover
     import sys
 
     sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-    from app.services.geocoder import GeocodeResult, geocode
+    from app.services.geocoder import GeocodeResult, geocode, known_city_hint
 
 logging.basicConfig(
     level=logging.INFO,
@@ -122,17 +142,23 @@ _LOG_EVERY = 25
 
 @dataclass
 class AddressGroup:
-    """One distinct deals.address and how many lat-IS-NULL rows share it."""
+    """One distinct (deals.address, deals.city) pair + its lat-IS-NULL row count.
+
+    `city` is the raw Rosreestr value (may be NULL for non-rosreestr sources, and
+    may be junk — see `geocoder.known_city_hint`), kept verbatim so the UPDATE
+    can target exactly the rows this group came from.
+    """
 
     address: str
     deals_count: int
+    city: str | None = None
 
 
 @dataclass
 class Stats:
     """Final-summary counters.
 
-    - processed     distinct addresses fed to the geocoder this run.
+    - processed     distinct (address, city) groups fed to the geocoder this run.
     - geocoded      addresses the geocoder resolved to coords.
     - geocode_failed addresses the geocoder returned None for (still stamped).
     - skipped       distinct addresses skipped before any geocode call
@@ -156,10 +182,8 @@ class Stats:
 # ---------------------------------------------------------------------------
 
 
-def _select_pending_addresses(
+def _select_pending_addresses(db: Session, *, limit: int, stale_days: int) -> list[AddressGroup]:
-    db: Session, *, limit: int, stale_days: int
+    """Distinct (address, city) pairs still needing coords — resume-safe set.
-) -> list[AddressGroup]:
-    """Distinct deals.address still needing coords — resume-safe candidate set.
 
     Combines the `deals_geocode_pending_idx` partial index predicate
     (`lat IS NULL`) with a staleness filter so failed/un-geocodable addresses
@@ -167,17 +191,25 @@ def _select_pending_addresses(
 
     The GROUP BY collapses the ~6,951-row backlog into its distinct streets;
     `:limit` caps the deal ROWS fanned out, computed from the running SUM of
-    per-address counts so a single huge street can't blow past the cap. We
+    per-group counts so a single huge street can't blow past the cap. We order
-    order by occurrence DESC (biggest ROI per geocode call first) then address
+    by occurrence DESC (biggest ROI per geocode call first) then address, city
-    for a deterministic resume order.
+    for a deterministic resume order (the window ORDER BY must match the outer
+    one, otherwise the running_rows cap slices a different ordering).
+
+    GROUP BY address, city — NOT address alone (#2603, the shape #2601 fixed in
+    app/tasks/geocode_missing.py): the same street name in two cities is two
+    geocode calls with two different hints, not one call whose result lands on
+    both. SQL groups NULL cities together, so a NULL-city group stays its own
+    group rather than merging into an arbitrary city.
     """
-    rows = db.execute(
+    rows = (
+        db.execute(
             text(
-            "SELECT address, deals_count FROM ("
+                "SELECT address, city, deals_count FROM ("
-            "  SELECT address, "
+                "  SELECT address, city, "
                 "         COUNT(*) AS deals_count, "
                 "         SUM(COUNT(*)) OVER ("
-            "             ORDER BY COUNT(*) DESC, address ASC"
+                "             ORDER BY COUNT(*) DESC, address ASC, city ASC NULLS FIRST"
                 "         ) AS running_rows "
                 "  FROM deals "
                 "  WHERE lat IS NULL "
@@ -186,25 +218,41 @@ def _select_pending_addresses(
                 "    AND (geocode_tried_at IS NULL "
                 "         OR geocode_tried_at < NOW() "
                 "            - make_interval(days => CAST(:stale_days AS int))) "
-            "  GROUP BY address "
+                "  GROUP BY address, city "
                 ") g "
                 "WHERE running_rows - deals_count < CAST(:limit AS int) "
-            "ORDER BY deals_count DESC, address ASC"
+                "ORDER BY deals_count DESC, address ASC, city ASC NULLS FIRST"
             ),
             {"limit": limit, "stale_days": stale_days},
-    ).mappings().all()
+        )
-    return [AddressGroup(address=r["address"], deals_count=r["deals_count"]) for r in rows]
+        .mappings()
+        .all()
+    )
+    return [
+        AddressGroup(
+            address=r["address"],
+            city=r.get("city"),
+            deals_count=r["deals_count"],
+        )
+        for r in rows
+    ]
 
 
 def _count_pending_total(db: Session, *, stale_days: int) -> tuple[int, int]:
-    """Full backlog: (distinct addresses, total rows) eligible this pass.
+    """Full backlog: (distinct (address, city) pairs, total rows) this pass.
 
     Denominators for the dry-run projection — counts every lat-IS-NULL deal
-    that passes the staleness filter, ignoring --limit.
+    that passes the staleness filter, ignoring --limit. Counts PAIRS, matching
+    what `_select_pending_addresses` actually feeds the geocoder (#2603) —
+    counting distinct addresses here would understate the denominator and the
+    projection would read above 100%. COALESCE(city, '') keeps a NULL-city pair
+    countable: `count()` skips NULL inputs, and a bare row expression with a
+    NULL field invites exactly that argument — the empty string can't collide
+    with a real city name, so the pair count stays honest either way.
     """
     row = db.execute(
         text(
-            "SELECT COUNT(DISTINCT address) AS streets, COUNT(*) AS rows "
+            "SELECT COUNT(DISTINCT (address, COALESCE(city, ''))) AS streets, COUNT(*) AS rows "
             "FROM deals "
             "WHERE lat IS NULL "
             "  AND address IS NOT NULL "
@@ -225,8 +273,10 @@ def _count_pending_total(db: Session, *, stale_days: int) -> tuple[int, int]:
 # ---------------------------------------------------------------------------
 
 
-def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float) -> int:
+def _update_deals_geocoded(
-    """UPDATE every lat-IS-NULL deal on `address`; geom auto-fills via trigger.
+    db: Session, *, address: str, city: str | None, lat: float, lon: float
+) -> int:
+    """UPDATE every lat-IS-NULL deal on (address, city); geom via trigger.
 
     The `deals_set_geom_trg` BEFORE UPDATE OF lat, lon trigger
     (002_core_tables.sql, reuses listings_set_geom()) populates geom from the
@@ -234,6 +284,12 @@ def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float)
     so the row drops out of the candidate set. The `AND lat IS NULL` guard
     keeps this idempotent and avoids clobbering rows another pass already set.
 
+    `city IS NOT DISTINCT FROM` (not `=`) scopes the write to the group the
+    geocode was made for: plain `=` is never true for a NULL city, so a
+    NULL-city group would update zero rows and re-run forever; `IS NOT DISTINCT
+    FROM` treats NULL=NULL as a match while staying strict for a real city, so
+    the same street text in another city keeps its own coords (#2603).
+
     Returns the number of deal rows updated.
     """
     result = db.execute(
@@ -243,14 +299,15 @@ def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float)
             "       lon = CAST(:lon AS double precision), "
             "       geocode_tried_at = NOW() "
             " WHERE address = CAST(:addr AS text) "
+            "   AND city IS NOT DISTINCT FROM CAST(:city AS text) "
             "   AND lat IS NULL"
         ),
-        {"addr": address, "lat": lat, "lon": lon},
+        {"addr": address, "city": city, "lat": lat, "lon": lon},
     )
     return result.rowcount
 
 
-def _mark_deals_tried(db: Session, *, address: str) -> int:
+def _mark_deals_tried(db: Session, *, address: str, city: str | None) -> int:
     """Stamp `geocode_tried_at = NOW()` WITHOUT touching lat/lon (geocode miss).
 
     Critical for resume: an address the geocoder can't resolve must still drop
@@ -259,6 +316,9 @@ def _mark_deals_tried(db: Session, *, address: str) -> int:
     it for `stale_days`. The `AND lat IS NULL` guard means a concurrent success
     can't be downgraded.
 
+    Scoped to the (address, city) pair for the same reason as the coords write
+    — a miss in one city must not defer the other city's retry (#2603).
+
     Returns the number of deal rows stamped.
     """
     result = db.execute(
@@ -266,9 +326,10 @@ def _mark_deals_tried(db: Session, *, address: str) -> int:
             "UPDATE deals "
             "   SET geocode_tried_at = NOW() "
             " WHERE address = CAST(:addr AS text) "
+            "   AND city IS NOT DISTINCT FROM CAST(:city AS text) "
             "   AND lat IS NULL"
         ),
-        {"addr": address},
+        {"addr": address, "city": city},
     )
     return result.rowcount
 
@@ -285,7 +346,7 @@ async def _run_backfill(
     batch: str,
     dry_run: bool,
 ) -> Stats:
-    """For each distinct address: geocode once, then UPDATE all its deals.
+    """For each (address, city) pair: geocode once, then UPDATE all its deals.
 
     Per-address SAVEPOINT (`db.begin_nested()`) so one bad UPDATE can't abort
     the batch (backend.md SAVEPOINT rule). Per-address commit on success → a
@@ -301,6 +362,12 @@ async def _run_backfill(
 
     for i, group in enumerate(groups, start=1):
         address = group.address
+        city = group.city
+        # Only a recognised oblast-66 city is fed to the geocoder; junk Rosreestr
+        # values degrade to None (geocoder.known_city_hint — shared with the
+        # other DB-column callers). The raw `city` is still used for the UPDATE
+        # scope — it identifies the group either way.
+        hint = known_city_hint(city)
 
         # The geocoder itself rejects <3 chars, but skip here too so the dry-run
         # report and counters stay honest (no phantom "processed" address).
@@ -310,13 +377,14 @@ async def _run_backfill(
 
         result: GeocodeResult | None = None
         try:
-            result = await geocode(address, db)
+            result = await geocode(address, db, city_hint=hint)
         except Exception as exc:  # defensive — one geocode error must not kill batch
             stats.geocode_failed += 1
             stats.processed += 1
             logger.warning(
-                "geocode raised for addr=%r (%d deals): %s",
+                "geocode raised for addr=%r city=%r (%d deals): %s",
                 address[:60],
+                city,
                 group.deals_count,
                 exc,
             )
@@ -331,18 +399,21 @@ async def _run_backfill(
             stats.processed += 1
             if dry_run:
                 logger.info(
-                    "DRY-RUN addr=%r (%d deals) → NOT FOUND (would stamp tried)",
+                    "DRY-RUN addr=%r city=%r (%d deals) → NOT FOUND (would stamp tried)",
                     address[:60],
+                    city,
                     group.deals_count,
                 )
             else:
                 try:
                     with db.begin_nested():
-                        _mark_deals_tried(db, address=address)
+                        _mark_deals_tried(db, address=address, city=city)
                     db.commit()
                 except Exception as exc:  # defensive — isolate one bad UPDATE
                     db.rollback()
-                    logger.warning("mark_tried failed for addr=%r: %s", address[:60], exc)
+                    logger.warning(
+                        "mark_tried failed for addr=%r city=%r: %s", address[:60], city, exc
+                    )
             _maybe_log_progress(i, groups, batch, stats)
             continue
 
@@ -356,8 +427,9 @@ async def _run_backfill(
 
         if dry_run:
             logger.info(
-                "DRY-RUN addr=%r → (%.5f, %.5f) provider=%s would update %d deals",
+                "DRY-RUN addr=%r city=%r → (%.5f, %.5f) provider=%s would update %d deals",
                 address[:60],
+                city,
                 result.lat,
                 result.lon,
                 result.provider,
@@ -369,7 +441,7 @@ async def _run_backfill(
         try:
             with db.begin_nested():
                 n = _update_deals_geocoded(
-                    db, address=address, lat=result.lat, lon=result.lon
+                    db, address=address, city=city, lat=result.lat, lon=result.lon
                 )
             # Per-address commit so resume picks up exactly where we crashed.
             db.commit()
@@ -378,7 +450,7 @@ async def _run_backfill(
             db.rollback()
             # The geocode itself succeeded (and is cached); only the write
             # failed. Count the address as geocoded but log the write failure.
-            logger.warning("db_write failed for addr=%r: %s", address[:60], exc)
+            logger.warning("db_write failed for addr=%r city=%r: %s", address[:60], city, exc)
 
         _maybe_log_progress(i, groups, batch, stats)
 
@@ -513,12 +585,10 @@ async def main(argv: list[str] | None = None) -> int:
 
     db = SessionLocal()
     try:
-        groups = _select_pending_addresses(
+        groups = _select_pending_addresses(db, limit=args.limit, stale_days=args.stale_days)
-            db, limit=args.limit, stale_days=args.stale_days
-        )
         total_rows = sum(g.deals_count for g in groups)
         logger.info(
-            "loaded %d distinct addresses (%d deal rows) needing coords",
+            "loaded %d distinct (address, city) groups (%d deal rows) needing coords",
             len(groups),
             total_rows,
         )
@@ -533,9 +603,7 @@ async def main(argv: list[str] | None = None) -> int:
         stats = await _run_backfill(db, groups, batch=args.batch, dry_run=args.dry_run)
 
         if args.dry_run:
-            total_streets, backlog_rows = _count_pending_total(
+            total_streets, backlog_rows = _count_pending_total(db, stale_days=args.stale_days)
-                db, stale_days=args.stale_days
-            )
             _report_dry_run(
                 stats,
                 total_streets=total_streets,

tradein-mvp/backend/tests/scripts/test_geocode_deals_nominatim.py

@@ -12,6 +12,10 @@ Coverage (per the issue's test plan):
       batch.
     - dedup: distinct addresses drive geocode call count (one call per address).
     - main() wiring: SessionLocal, --limit bind, returns geocoded count.
+    - #2603: grouping by the (address, city) PAIR, city_hint filtered through
+      the shared `geocoder.known_city_hint` (junk Rosreestr city → no hint; the
+      helper's own cases live in tests/test_geocoder_city_hint.py), and writes
+      scoped with `city IS NOT DISTINCT FROM`.
 
 No real Postgres. `geocode()` is async → patched with AsyncMock (mirrors
 tests/tasks/test_geocode_missing.py). The Session is a MagicMock that routes
@@ -93,7 +97,7 @@ def _make_db_mock(
                 tried_updates.append(dict(params) if params else {})
             result.rowcount = update_rowcount
             return result
-        if "COUNT(DISTINCT address)" in sql_str:
+        if "COUNT(DISTINCT" in sql_str:
             result.first.return_value = (total_streets, total_rows)
             return result
         if "FROM deals" in sql_str:
@@ -176,13 +180,16 @@ async def test_mark_deals_tried_sql_stamps_tried_at_only():
     """The mark-tried writer sets geocode_tried_at and nothing else (no lat/lon)."""
     db = MagicMock()
     db.execute.return_value = MagicMock(rowcount=4)
-    n = _mark_deals_tried(db, address="Екатеринбург, X")
+    n = _mark_deals_tried(db, address="Екатеринбург, X", city="Екатеринбург")
     args, _kw = db.execute.call_args
     sql_str = str(args[0])
     assert "UPDATE deals" in sql_str
     assert "geocode_tried_at = NOW()" in sql_str
     assert "SET lat" not in sql_str
     assert "lat IS NULL" in sql_str  # idempotency / no-clobber guard
+    # #2603: a miss in one city must not defer the other city's retry.
+    assert "city IS NOT DISTINCT FROM CAST(:city AS text)" in sql_str
+    assert args[1]["city"] == "Екатеринбург"
     assert n == 4
 
 
@@ -312,7 +319,9 @@ async def test_run_backfill_db_write_failure_isolated_to_address():
 def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom():
     db = MagicMock()
     db.execute.return_value = MagicMock(rowcount=7)
-    n = _update_deals_geocoded(db, address="Екатеринбург, Y", lat=56.1, lon=60.2)
+    n = _update_deals_geocoded(
+        db, address="Екатеринбург, Y", city="Екатеринбург", lat=56.1, lon=60.2
+    )
     args, _kw = db.execute.call_args
     sql_str = str(args[0])
     binds = args[1]
@@ -321,10 +330,28 @@ def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom():
     # geom must NOT be set manually — the deals_set_geom_trg trigger fills it.
     assert "geom" not in sql_str
     assert "lat IS NULL" in sql_str  # no-clobber guard
-    assert binds == {"addr": "Екатеринбург, Y", "lat": 56.1, "lon": 60.2}
+    # #2603: the write is scoped to the (address, city) group it was made for.
+    assert "city IS NOT DISTINCT FROM CAST(:city AS text)" in sql_str
+    assert binds == {
+        "addr": "Екатеринбург, Y",
+        "city": "Екатеринбург",
+        "lat": 56.1,
+        "lon": 60.2,
+    }
     assert n == 7
 
 
+def test_update_deals_geocoded_null_city_group_uses_null_safe_predicate():
+    """A NULL-city group must still update its rows: plain `city = NULL` is never
+    true, so `IS NOT DISTINCT FROM` is what keeps those rows reachable."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock(rowcount=2)
+    _update_deals_geocoded(db, address="Тестовая, 1", city=None, lat=56.1, lon=60.2)
+    args, _kw = db.execute.call_args
+    assert "city IS NOT DISTINCT FROM CAST(:city AS text)" in str(args[0])
+    assert args[1]["city"] is None
+
+
 # ---------------------------------------------------------------------------
 # resume filter — candidate SQL carries lat IS NULL + staleness predicate
 # ---------------------------------------------------------------------------
@@ -333,23 +360,19 @@ def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom():
 def test_select_pending_addresses_filters_null_and_stale():
     """The candidate query must combine lat IS NULL with the staleness window
     and bind --stale-days, so already-tried-recently rows are excluded."""
-    db, _, _ = _make_db_mock(
+    db, _, _ = _make_db_mock(address_rows=[{"address": "Екатеринбург, Z", "deals_count": 3}])
-        address_rows=[{"address": "Екатеринбург, Z", "deals_count": 3}]
-    )
 
     groups = _select_pending_addresses(db, limit=2000, stale_days=30)
 
     assert groups == [AddressGroup(address="Екатеринбург, Z", deals_count=3)]
     # Inspect the SQL + binds of the SELECT.
-    select_call = next(
+    select_call = next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))
-        c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
-    )
     sql_str = str(select_call[0][0])
     binds = select_call[0][1]
     assert "lat IS NULL" in sql_str
     assert "geocode_tried_at IS NULL" in sql_str
     assert "make_interval(days => CAST(:stale_days AS int))" in sql_str
-    assert "GROUP BY address" in sql_str
+    assert "GROUP BY address, city" in sql_str  # #2603 — pair, not address alone
     assert binds["stale_days"] == 30
     assert binds["limit"] == 2000
 
@@ -357,9 +380,7 @@ def test_select_pending_addresses_filters_null_and_stale():
 def test_select_pending_addresses_passes_custom_stale_days():
     db, _, _ = _make_db_mock(address_rows=[])
     _select_pending_addresses(db, limit=500, stale_days=7)
-    select_call = next(
+    select_call = next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))
-        c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
-    )
     binds = select_call[0][1]
     assert binds["stale_days"] == 7
     assert binds["limit"] == 500
@@ -419,9 +440,7 @@ async def test_main_respects_limit_bind():
     ):
         await main(["--limit", "50"])
 
-    select_call = next(
+    select_call = next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))
-        c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
-    )
     assert select_call[0][1]["limit"] == 50
 
 
@@ -440,6 +459,97 @@ async def test_main_no_pending_returns_zero():
     db.close.assert_called_once()
 
 
+# ---------------------------------------------------------------------------
+# #2603 — (address, city) pair grouping + filtered city_hint
+# ---------------------------------------------------------------------------
+
+
+def test_select_pending_addresses_returns_city_in_group():
+    """The candidate SELECT reads `city` and carries it into the group."""
+    db, _, _ = _make_db_mock(
+        address_rows=[
+            {"address": "Нижний Тагил, Победы", "city": "Нижний Тагил", "deals_count": 3},
+            {"address": "Тестовая, 1", "city": None, "deals_count": 1},
+        ]
+    )
+
+    groups = _select_pending_addresses(db, limit=2000, stale_days=30)
+
+    assert groups == [
+        AddressGroup(address="Нижний Тагил, Победы", deals_count=3, city="Нижний Тагил"),
+        AddressGroup(address="Тестовая, 1", deals_count=1, city=None),
+    ]
+    sql_str = str(next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))[0][0])
+    assert "SELECT address, city, deals_count" in sql_str
+
+
+async def test_run_backfill_passes_known_city_as_hint():
+    groups = [AddressGroup(address="Победы, 30", deals_count=2, city="Нижний Тагил")]
+    db, coord_updates, _ = _make_db_mock(update_rowcount=2)
+
+    with patch(
+        _GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")
+    ) as mock_geo:
+        await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    mock_geo.assert_called_once_with("Победы, 30", db, city_hint="Нижний Тагил")
+    assert coord_updates[0]["city"] == "Нижний Тагил"
+
+
+async def test_run_backfill_junk_city_geocodes_without_hint_but_scopes_update():
+    """Junk city → geocode WITHOUT a hint (old behaviour), yet the UPDATE is still
+    scoped to that exact group so it can't spill onto another city's rows."""
+    groups = [AddressGroup(address="Бессонова, 11", deals_count=1, city="Бессонова")]
+    db, coord_updates, _ = _make_db_mock(update_rowcount=1)
+
+    with patch(
+        _GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")
+    ) as mock_geo:
+        await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    mock_geo.assert_called_once_with("Бессонова, 11", db, city_hint=None)
+    assert coord_updates[0]["city"] == "Бессонова"
+
+
+async def test_run_backfill_same_address_two_cities_independent():
+    """Key #2603 scenario: identical address text in two cities → two geocode
+    calls with their own hints, and two UPDATEs scoped to their own city."""
+    groups = [
+        AddressGroup(address="Победы, 30", deals_count=4, city="Екатеринбург"),
+        AddressGroup(address="Победы, 30", deals_count=2, city="Нижний Тагил"),
+    ]
+    db, coord_updates, _ = _make_db_mock(update_rowcount=1)
+
+    with patch(
+        _GEOCODE_PATH,
+        new_callable=AsyncMock,
+        side_effect=[_result("nominatim"), _result("nominatim", lat=57.910, lon=59.985)],
+    ) as mock_geo:
+        stats = await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    assert mock_geo.call_count == 2
+    assert [c.kwargs["city_hint"] for c in mock_geo.call_args_list] == [
+        "Екатеринбург",
+        "Нижний Тагил",
+    ]
+    assert stats.geocoded == 2
+    assert [(u["addr"], u["city"]) for u in coord_updates] == [
+        ("Победы, 30", "Екатеринбург"),
+        ("Победы, 30", "Нижний Тагил"),
+    ]
+
+
+async def test_run_backfill_miss_marks_only_its_own_city():
+    groups = [AddressGroup(address="Победы, 30", deals_count=1, city="Нижний Тагил")]
+    db, coord_updates, tried_updates = _make_db_mock()
+
+    with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=None):
+        await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    assert coord_updates == []
+    assert tried_updates == [{"addr": "Победы, 30", "city": "Нижний Тагил"}]
+
+
 # ---------------------------------------------------------------------------
 # Stats dataclass
 # ---------------------------------------------------------------------------

tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py

@@ -359,6 +359,51 @@ def test_city_column_null_falls_back_to_text_gate() -> None:
     mock_geo.assert_not_called()
 
 
+# ── наблюдаемость гейта: отдельный срез по колонке (#2603) ───────────────────
+
+
+def test_skipped_non_ekb_by_column_counts_only_column_gate() -> None:
+    """Колоночный и текстовый гейты различимы: by_column считает ТОЛЬКО первый.
+
+    Общий skipped_non_ekb остаётся суммой обоих (обратная совместимость), но по
+    мере раскатки областных развёрток (#2598) колоночный гейт начнёт забирать
+    строки из no_address — без отдельного среза общий счётчик поменял бы смысл
+    ровно тогда, когда по нему валидируют раскатку.
+    """
+    rows = [
+        # 1. Гейт по колонке: город в колонке, в тексте адреса города НЕТ.
+        {"id": 300, "address": "ул. Победы, 30", "city": "Нижний Тагил"},
+        # 2. Текстовый гейт: колонка пуста, город назван в тексте.
+        {"id": 301, "address": "г. Нижний Тагил, проспект Ленина, 1", "city": None},
+    ]
+    db = _make_db([rows, []])
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
+            return_value=_HIT,
+        ),
+        patch("app.tasks.backfill_listings_coords_geoportal._parse_street_house"),
+    ):
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert res.candidates == 2
+    assert res.skipped_non_ekb == 2  # оба гейта, как и раньше
+    assert res.skipped_non_ekb_by_column == 1  # только колоночный
+    # Вклад текстового гейта вычисляется разностью — старая метрика #2583 жива.
+    assert res.skipped_non_ekb - res.skipped_non_ekb_by_column == 1
+
+
+def test_to_counters_exposes_skipped_non_ekb_by_column() -> None:
+    """Новый счётчик уходит в scrape_runs.counters, старый ключ на месте."""
+    counters = BackfillCoordsResult(skipped_non_ekb=7, skipped_non_ekb_by_column=5).to_counters()
+
+    assert counters["skipped_non_ekb"] == 7
+    assert counters["skipped_non_ekb_by_column"] == 5
+    # Дефолт нулевой — прогон без колоночных скипов пишет 0, а не теряет ключ.
+    assert BackfillCoordsResult().to_counters()["skipped_non_ekb_by_column"] == 0
+
+
 # ── idempotency ───────────────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/tasks/test_geocode_missing.py

@@ -761,11 +761,17 @@ def test_admin_geocode_missing_post_dry_run_endpoint_exists() -> None:
 
 
 @pytest.mark.asyncio
-async def test_admin_geocode_missing_passes_city_hint() -> None:
+@pytest.mark.parametrize("target", ["listings", "deals"])
+async def test_admin_geocode_missing_passes_city_hint(target: str) -> None:
     """POST /admin/geocode-missing читает city из SELECT и передаёт как city_hint.
 
     Раньше endpoint читал только row["address"] и звал geocode(clean, db) без
     города — голый тагильский адрес без города в тексте уходил в Екатеринбург.
+
+    Оба target'а (#2603): колонка city есть и в listings (миграция 196), и в
+    deals (миграция 177) — SELECT + прокидывание хинта общие для обеих веток,
+    расходится только extra_filter, поэтому deals обязан покрываться тем же
+    контрактом.
     """
     from app.api.v1 import admin as admin_module
 
@@ -792,11 +798,89 @@ async def test_admin_geocode_missing_passes_city_hint() -> None:
         new_callable=AsyncMock,
         return_value=geo,
     ) as mock_geo:
-        result = await admin_module.geocode_missing(db, limit=100, target="listings")
+        result = await admin_module.geocode_missing(
+            db,
+            limit=100,
+            target=target,  # type: ignore[arg-type]
+        )
 
     mock_geo.assert_called_once_with("ул. Победы, 30", db, city_hint="Нижний Тагил")
     assert result["geocoded"] == 1
     assert result["skipped"] == 0
+    # SELECT адресован именно запрошенной таблице (обе несут колонку city).
+    assert f"FROM {target}" in str(db.execute.call_args_list[0][0][0])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("target", ["listings", "deals"])
+async def test_admin_geocode_missing_drops_junk_city_hint(target: str) -> None:
+    """Мусорный город из колонки НЕ уходит в city_hint (#2603).
+
+    `deals.city` росреестровое: в хвосте «Бессонова», «Билейский рыбопитомник» —
+    улицы/урочища, а не города. Хинт из такого значения закрывает EKB-локальные
+    тиры и уезжает префиксом в запрос провайдеру, т.е. хуже отсутствия хинта.
+    Гейт общий с двумя другими DB-колоночными callers (geocoder.known_city_hint).
+    """
+    from app.api.v1 import admin as admin_module
+
+    rows = [{"id": 56, "address": "ул. Бессонова, 11", "city": "Бессонова"}]
+
+    db = MagicMock()
+    select_result = MagicMock()
+    select_result.mappings.return_value.all.return_value = rows
+    update_result = MagicMock()
+    remaining_result = MagicMock()
+    remaining_result.scalar.return_value = 0
+    db.execute.side_effect = [select_result, update_result, remaining_result]
+
+    geo = GeocodeResult(
+        lat=56.838,
+        lon=60.605,
+        full_address="Екатеринбург, ул. Бессонова, 11",
+        provider="nominatim",  # type: ignore[arg-type]
+        confidence="exact",
+    )
+
+    with patch(
+        "app.api.v1.admin.geocode",
+        new_callable=AsyncMock,
+        return_value=geo,
+    ) as mock_geo:
+        await admin_module.geocode_missing(
+            db,
+            limit=100,
+            target=target,  # type: ignore[arg-type]
+        )
+
+    mock_geo.assert_called_once_with("ул. Бессонова, 11", db, city_hint=None)
+
+
+@pytest.mark.asyncio
+async def test_geocode_missing_listings_drops_junk_city_hint() -> None:
+    """Тот же гейт в ночной задаче: мусорный city → city_hint=None (#2603).
+
+    Для listings.city это сегодня no-op (скрапер пишет шесть кураторских имён),
+    но инвариант «в geocode() уходит только словарный город» держим единым для
+    всех трёх DB-колоночных callers, чтобы седьмой не пришлось чинить заново.
+    Сырой city при этом остаётся ключом группы для UPDATE.
+    """
+    rows = [{"address": "ул. Бессонова, 11", "city": "Бессонова", "listings_count": 2}]
+    db = _mock_db_rows(rows)
+
+    with patch(
+        "app.tasks.geocode_missing.geocode",
+        new_callable=AsyncMock,
+        return_value=_make_geocode_result(),
+    ) as mock_geo:
+        await geocode_missing_listings(db, batch_size=10)
+
+    mock_geo.assert_called_once_with("ул. Бессонова, 11", db, city_hint=None)
+    update_binds = [
+        c[0][1]
+        for c in db.execute.call_args_list
+        if "UPDATE listings" in str(c[0][0]) and "SET lat" in str(c[0][0])
+    ]
+    assert update_binds[0]["city"] == "Бессонова"  # группа UPDATE — по сырому городу
 
 
 @pytest.mark.asyncio

tradein-mvp/backend/tests/test_geocoder_city_hint.py

@@ -38,9 +38,37 @@ from app.services.geocoder import (
     _nominatim_suggest,
     _resolve_city_for_geocode,
     geocode,
+    known_city_hint,
     suggest,
 )
 
+# ── known_city_hint (#2603) ──────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        # Известные города области — значение проходит КАК ЕСТЬ (регистр/пробелы
+        # нормализует сам геокодер, caller'ы не должны его причёсывать).
+        ("Екатеринбург", "Екатеринбург"),
+        ("Нижний Тагил", "Нижний Тагил"),
+        ("нижний  тагил", "нижний  тагил"),
+        ("Каменск-Уральский", "Каменск-Уральский"),
+        # Мусор из хвоста росреестрового deals.city — это улицы/урочища, а не
+        # города; хинт из них закрыл бы EKB-тиры и уехал префиксом в запрос.
+        ("Бессонова", None),
+        ("Бердюгина", None),
+        ("Билейский рыбопитомник", None),
+        # Пусто — хинта нет.
+        (None, None),
+        ("", None),
+        ("   ", None),
+    ],
+)
+def test_known_city_hint(value: str | None, expected: str | None) -> None:
+    assert known_city_hint(value) == expected
+
+
 # ── _resolve_city_for_geocode ────────────────────────────────────────────────