feat(msk): пред-геокод московского Авито — город из слага, регион из КЛАДР

У карточек Авито нет координат ни у одной из 50 335, а адрес — голая улица с
домом («Варшавское ш.,62к1»). Сбор шёл по `/moskva_i_mo/`, поэтому Москву от
области отделить было нечем: префиксный фильтр, работающий у Циана по округу,
здесь отбросил бы 100% строк. Наивный матч адресов к `houses(77)` даёт ровно 0
совпадений — дома лежат как «ЮАО, р-н Даниловский, проспект Андропова, 18».

Замер показал, что одного геокода мало: с констрейнтом «Москва + Московская»
дом находится у 92% адресов, но верхний кандидат DaData расходится с реальным
городом у 15% и почти всегда в пользу столицы. Недостающий сигнал лежал рядом и
бесплатно — слаг города в `source_url` (`avito.ru/moskva/...`), заполнен у 100%
карточек: 22 120 с `moskva`, остальное — подмосковные слаги.

Поэтому город берётся из слага и сужает констрейнт, а регион — из КЛАДР ответа,
не из слага: Новая Москва (Троицк, Щербинка, Коммунарка, Зеленоград) идёт своими
слагами, но это регион 77. Регион 77 пишется в `listings` сразу с координатами и
`geo_precision='house'`, поэтому строки попадают в radius-подбор аналогов без
ожидания `geocode_missing`. Регион 50 не пишется, а копится строкой в кэше
`msk_raw.avito_geocode` — до появления региона в реестре.

Кэш ключуется слагом и нормализованным адресом, отрицательные ответы тоже
кэшируются, так что повторный прогон внешний сервис не дёргает. `geocode_cache`
приложения не тронут — там другой ключ. `--allow-unfiltered` без `--geocode`
остаётся прежним аварийным режимом.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VQ8jqr4SFirX5tFLwdSrXh

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

@@ -349,6 +349,7 @@ async def suggest_addresses(
     limit: int = 8,
     city: str | None = "Екатеринбург",
     region: str | None = None,
+    regions: list[str] | None = None,
 ) -> list[DadataSuggestion]:
     """Автокомплит адресов через DaData /suggest/address.
 
@@ -367,6 +368,13 @@ async def suggest_addresses(
             БЕЗ типа («Свердловская», а тип — отдельно в `region_type`="обл").
             Передашь «Свердловская область» — совпадений не будет, и запрос
             вернёт ПУСТО без всякой ошибки (hard-filter, не boost).
+        regions: НЕСКОЛЬКО регионов разом — `locations` у DaData это список, и
+            элементы в нём складываются по ИЛИ. Нужно там, где один регион даёт
+            заведомо неверный ответ: адрес Авито по Москве и области («Юбилейная
+            ул.,20Б») лежит либо в 77, либо в 50, и констрейнт из одного региона
+            молча притягивает подмосковный дом к московской улице. Имеет
+            приоритет над `region`/`city`. Имена — так же БЕЗ типа («Москва»,
+            «Московская»).
 
     Returns:
         list[DadataSuggestion] — пустой список если:
@@ -394,7 +402,11 @@ async def suggest_addresses(
         "query": query.strip(),
         "count": max(1, min(int(limit), 20)),
     }
-    if region:
+    if regions:
+        # Список → ИЛИ по регионам (см. docstring). Пустые имена выбрасываем:
+        # `{"region": ""}` — не «любой регион», а гарантированный ноль хитов.
+        body["locations"] = [{"region": r} for r in regions if r and r.strip()]
+    elif region:
         # `locations` с полем region — уже жёсткий фильтр сам по себе (DaData
         # ограничивает выдачу этим регионом). `restrict_value` — ТОП-LEVEL параметр
         # body (не ключ внутри объекта locations) — здесь он был бы silent no-op,

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

@@ -13,14 +13,45 @@ bbox региона 77; без округа внутри bbox — 17 576 (это
 7 337. Отдельно 212 карточек с адресом вида «Екатеринбург (Cian)» — артефакт
 парсера, считаются своим счётчиком, чтобы не растворяться в «не Москва».
 
-Отбор Москвы (source=avito) НЕВОЗМОЖЕН по адресу: у Авито адрес — голая улица с
+Отбор Москвы (source=avito) по адресу НЕВОЗМОЖЕН: у Авито адрес — голая улица с
 домом («Варшавское ш.,62к1»), ни города, ни округа, и координат нет НИ У ОДНОЙ
-карточки. Поэтому:
+карточки (замер: lat/lon/cadastral_number/geo_precision пусты у всех 50 335).
-  * префиксный фильтр к Авито не применяется — он отбросил бы 100% строк;
+Поэтому для Авито работает ПРЕД-ГЕОКОД (`--geocode`), а не префиксный фильтр.
-  * запись Авито требует явного `--allow-unfiltered`: молча залить в регион 77
+
-    вперемешку Москву и область — хуже, чем не залить ничего;
+Два сигнала, и оба нужны — ни один по отдельности не годится.
-  * строки Авито лягут БЕЗ geom (lat/lon пусты) — они не попадут в radius-подбор
+
-    аналогов estimator'а, пока их не догеокодит `geocode_missing`.
+1. Город по версии самого Авито — слаг в `source_url`
+   (`avito.ru/<slug>/kvartiry/...`). Заполнен у 100% карточек, стоит 0 вызовов:
+   `moskva` — 21 841 карточка, остальное — муниципалитеты области (balashiha,
+   himki, podolsk, …). Это единственный ТОЧНЫЙ признак населённого пункта, но
+   регион по нему не выводится: Троицк/Щербинка/Коммунарка/Московский/
+   Зеленоград — свои слаги, а регион у них 77 (Новая Москва и ЗелАО).
+
+2. Координаты и регион — DaData /suggest/address с `locations`-констрейнтом.
+   Слаг сужает констрейнт (`moskva` → только регион «Москва», иначе — «Москва»
+   И «Московская» разом), но РЕГИОН БЕРЁТСЯ ИЗ ОТВЕТА (`region_kladr_id`), а не
+   из слага — иначе Новая Москва уехала бы в область.
+
+Почему не наоборот (только геокод, без слага). Замер на 200 случайных адресах
+Авито с констрейнтом «Москва + Московская»: дом с координатами нашёлся у 184
+(92%), но верхний кандидат разошёлся со слагом у 28 из 184 (15%) — и почти
+всегда в пользу Москвы («пр-т Мира,19» при слаге fryazino → «г Москва, пр-кт
+Мира, 19»). Голый адрес без города DaData тянет в столицу; 15% чужих домов в
+регионе 77 — ровно та ошибка, ради которой стоял `--allow-unfiltered`.
+С сужением по слагу (`moskva` → регион «Москва») резолв 77 из 82 (94%),
+qc_geo=0 у 97% найденных.
+
+Что куда едет:
+  * регион 77 → в `listings`, С координатами (geom есть сразу, radius-подбор
+    аналогов работает без ожидания `geocode_missing`);
+  * регион 50 → НЕ пишется, ждёт появления региона 50 в реестре; лежит не в
+    воздухе, а строкой в `msk_raw.avito_geocode` (region_code=50) — когда
+    регион появится, прогон по этой полке уже не потребует внешних вызовов;
+  * адрес не разрешился (ЖК без улицы, «Мкр-н имени В.Н. Махалина, 33») →
+    свой счётчик, карточка не пишется.
+
+`--allow-unfiltered` (без `--geocode`) остаётся прежним аварийным режимом: пишет
+Москву вперемешку с областью и БЕЗ geom. Молча он по-прежнему не срабатывает.
 
 Пересчёт `listing_segment` (пункт, ради которого нельзя копировать payload как
 есть). Кит ставит 'novostroyki' по одному лишь наличию `offer.newbuilding.id`,
@@ -43,15 +74,18 @@ source_id) считает сам кит (`ScrapedLot.compute_dedup_hash`), це
 Запуск:
     python -m app.tasks.msk_raw_import --dry-run
     python -m app.tasks.msk_raw_import --limit 500
-    python -m app.tasks.msk_raw_import --source avito --allow-unfiltered
+    python -m app.tasks.msk_raw_import --source avito --geocode --geocode-limit 9000
+    python -m app.tasks.msk_raw_import --source avito --allow-unfiltered  # аварийный
 """
 
 from __future__ import annotations
 
 import argparse
+import asyncio
 import logging
 import re
 from dataclasses import dataclass
+from urllib.parse import urlsplit
 
 from pydantic import ValidationError
 from scraper_kit.base import ScrapedLot, save_listings
@@ -59,14 +93,36 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.core.db import SessionLocal
+from app.services import dadata
+from app.services.geocoder import normalize_address
 from app.services.scraper_adapters import RealMatcherAdapter
 
 logger = logging.getLogger(__name__)
 
 MOSCOW_REGION_CODE = 77
 MOSCOW_CITY = "Москва"
+# Регион 50 (Московская область) в реестре `app.services.regions` ещё не заведён —
+# карточки области не пишутся, а откладываются (см. докстринг модуля).
+OBLAST_REGION_CODE = 50
 DEFAULT_BATCH_SIZE = 500
 
+# Слаг города в `source_url` Авито: `https://www.avito.ru/<slug>/kvartiry/...`.
+AVITO_MOSCOW_SLUG = "moskva"
+# Имена регионов для DaData-констрейнта `locations`. DaData хранит имя БЕЗ типа
+# («Москва», «Московская»), тип лежит отдельно в `region_type` — с типом
+# hard-фильтр молча схлопывает выдачу в ноль (та же грабля, что в
+# `geocoder.SVERDLOVSK_OBLAST_REGION`).
+DADATA_MOSCOW_REGION = "Москва"
+DADATA_OBLAST_REGION = "Московская"
+# Первые две цифры КЛАДР региона в ответе DaData → код региона.
+_KLADR_TO_REGION = {"77": MOSCOW_REGION_CODE, "50": OBLAST_REGION_CODE}
+# Дневной потолок внешних вызовов. Free tier DaData /suggest — 10 000/сутки на
+# аккаунт, и тот же аккаунт обслуживает автокомплит формы оценки; берём с запасом.
+DEFAULT_GEOCODE_LIMIT = 9000
+# Сколько запросов к DaData держим в полёте одновременно. Больше смысла нет:
+# упираемся не в нас, а в квоту.
+_GEOCODE_CONCURRENCY = 6
+
 # Префиксы административных округов Москвы — единственный признак города в адресе
 # карточки Циана (сам город в адрес не попадает).
 MOSCOW_OKRUGS = (
@@ -107,6 +163,233 @@ LIMIT :limit
 """
 
 
+# ── Пред-геокод Авито ────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True, slots=True)
+class GeoPoint:
+    """Разрешённая точка адреса Авито. `region_code` — из ответа, не из слага."""
+
+    lat: float
+    lon: float
+    region_code: int
+    full_address: str
+
+
+@dataclass
+class GeocodeBudget:
+    """Дневной потолок внешних вызовов на прогон.
+
+    Кончился — не падаем, а перестаём спрашивать: карточки без точки просто не
+    пишутся в этот заход, а на следующем подхватятся с того же места (уже
+    разрешённые адреса лежат в кэше и квоты не стоят).
+    """
+
+    remaining: int
+    spent: int = 0
+    exhausted: bool = False
+
+    def take(self, n: int) -> int:
+        """Сколько из `n` запросов позволено сделать сейчас."""
+        allowed = max(0, min(n, self.remaining))
+        if allowed < n:
+            self.exhausted = True
+        self.remaining -= allowed
+        self.spent += allowed
+        return allowed
+
+
+# Кэш пред-геокода. Живёт в `msk_raw` (рядом с сырьём, а не в прикладной схеме):
+# это свойство КОРПУСА, а не приложения, и переживает пересбор listings.
+# `geocode_cache` приложения сознательно не трогаем — там другой ключ (адрес +
+# city_hint) и другой TTL, а нам нужен слаг в ключе и код региона в значении.
+# Он же — «полка ожидания» для области: строки с region_code=50 никуда не
+# пишутся, но остаются разрешёнными, и когда регион 50 появится в реестре,
+# прогон по ним не потратит ни одного внешнего вызова.
+#
+# Строка с lat IS NULL — ОТРИЦАТЕЛЬНЫЙ результат («DaData дома не знает»), и он
+# тоже кэшируется: без этого каждый повторный прогон заново тратил бы квоту на
+# те же ~8% неразрешимых адресов (ЖК без улицы, «Мкр-н имени В.Н. Махалина,
+# 33»). Передумать можно руками:
+# DELETE FROM msk_raw.avito_geocode WHERE lat IS NULL.
+_GEO_CACHE_DDL = """
+CREATE TABLE IF NOT EXISTS msk_raw.avito_geocode (
+    cache_key    text PRIMARY KEY,
+    slug         text,
+    address      text NOT NULL,
+    lat          double precision,
+    lon          double precision,
+    region_code  smallint,
+    full_address text,
+    resolved_at  timestamptz NOT NULL DEFAULT NOW()
+)
+"""
+_GEO_CACHE_SELECT = """
+SELECT cache_key, lat, lon, region_code, full_address
+FROM msk_raw.avito_geocode
+WHERE cache_key = ANY(:keys)
+"""
+_GEO_CACHE_UPSERT = """
+INSERT INTO msk_raw.avito_geocode
+    (cache_key, slug, address, lat, lon, region_code, full_address)
+VALUES (:key, :slug, :address, :lat, :lon, :region, :full)
+ON CONFLICT (cache_key) DO UPDATE
+    SET lat = EXCLUDED.lat,
+        lon = EXCLUDED.lon,
+        region_code = EXCLUDED.region_code,
+        full_address = EXCLUDED.full_address,
+        resolved_at = NOW()
+"""
+
+
+def avito_city_slug(payload: dict) -> str | None:
+    """Слаг города из `source_url`: `avito.ru/<slug>/kvartiry/...` → `<slug>`.
+
+    Единственный признак населённого пункта, который Авито отдаёт честно и
+    даром. Регион из него НЕ выводится (см. докстринг модуля) — он лишь сужает
+    констрейнт геокодера.
+    """
+    url = payload.get("source_url")
+    if not isinstance(url, str) or not url:
+        return None
+    parts = [p for p in urlsplit(url).path.split("/") if p]
+    return parts[0] if parts else None
+
+
+def geo_cache_key(address: str, slug: str | None) -> str:
+    """Ключ кэша.
+
+    Слаг — ЧАСТЬ ключа, не украшение: один и тот же текст адреса встречается в
+    разных муниципалитетах (замер: 1110 адресов из 21 570 живут сразу под
+    несколькими слагами), и это РАЗНЫЕ дома.
+    """
+    return f"{slug or '-'}|{normalize_address(address)}"
+
+
+def _dadata_regions_for(slug: str | None) -> list[str]:
+    """Констрейнт `locations` по слагу.
+
+    Москва — только столица; иначе оба региона, потому что слаг может оказаться
+    Новой Москвой (Троицк/Щербинка/Коммунарка/Московский) или Зеленоградом — у
+    них свои слаги, а регион 77.
+    """
+    if slug == AVITO_MOSCOW_SLUG:
+        return [DADATA_MOSCOW_REGION]
+    return [DADATA_MOSCOW_REGION, DADATA_OBLAST_REGION]
+
+
+def _point_from_suggestions(suggestions: list) -> GeoPoint | None:
+    """Лучший ДОМ с координатами из выдачи DaData, иначе None.
+
+    Берём только house-level (`kind == 'house'`) с координатами: улица/город
+    дают точку в середине улицы или в центре НП — для radius-подбора аналогов
+    это хуже, чем отсутствие точки (сосед через квартал попадёт в выборку, а
+    настоящий сосед — нет). Регион — первые две цифры КЛАДР ответа.
+    """
+    for s in suggestions:
+        if s.kind != "house" or s.lat is None or s.lon is None:
+            continue
+        region = _KLADR_TO_REGION.get((s.kladr_id or "")[:2])
+        if region is None:
+            continue  # ни 77, ни 50 — констрейнт пробит, такой ответ не берём
+        return GeoPoint(
+            lat=float(s.lat),
+            lon=float(s.lon),
+            region_code=region,
+            full_address=s.unrestricted_value,
+        )
+    return None
+
+
+async def _geocode_many(items: list[tuple[str, str, str | None]]) -> dict[str, GeoPoint | None]:
+    """`[(cache_key, address, slug)]` → `{cache_key: GeoPoint | None}`.
+
+    Параллелим с потолком `_GEOCODE_CONCURRENCY` — упираемся в квоту, а не в нас.
+    Отказ DaData (сеть/429/401) выглядит как пустая выдача: `suggest_addresses`
+    гасит исключения сам и возвращает []. Такой адрес получит None — и он, как и
+    честное «дома не знаю», уедет в кэш отрицательным. Отсюда правило прогона:
+    увидел в итоге всплеск `не разрешён` — сначала проверь логи DaData, потом
+    чисти отрицательные строки кэша (SQL выше), иначе разовый 429 замолчит
+    адреса до ручной чистки.
+    """
+    sem = asyncio.Semaphore(_GEOCODE_CONCURRENCY)
+
+    async def one(key: str, address: str, slug: str | None) -> tuple[str, GeoPoint | None]:
+        async with sem:
+            found = await dadata.suggest_addresses(
+                address, limit=10, regions=_dadata_regions_for(slug)
+            )
+        return key, _point_from_suggestions(found)
+
+    done = await asyncio.gather(*(one(k, a, sl) for k, a, sl in items))
+    return dict(done)
+
+
+def _resolve_points(
+    db: Session,
+    rows: list,
+    budget: GeocodeBudget,
+    *,
+    dry_run: bool,
+) -> dict[str, GeoPoint | None]:
+    """Точки для всех адресов страницы: сначала кэш, остаток — у DaData.
+
+    Ключ дедуплицируется в пределах страницы: одна и та же связка слаг+адрес
+    (несколько квартир в одном доме — обычное дело, 50 335 карточек на 21 570
+    адресов) стоит ОДИН внешний вызов.
+    """
+    wanted: dict[str, tuple[str, str | None]] = {}
+    for row in rows:
+        payload = row["payload"] or {}
+        address = payload.get("address")
+        if not address or is_artifact_address(address):
+            continue
+        slug = avito_city_slug(payload)
+        wanted.setdefault(geo_cache_key(address, slug), (address, slug))
+    if not wanted:
+        return {}
+
+    points: dict[str, GeoPoint | None] = {}
+    cached = db.execute(text(_GEO_CACHE_SELECT), {"keys": list(wanted)}).mappings().all()
+    for row in cached:
+        points[row["cache_key"]] = (
+            None
+            if row["lat"] is None
+            else GeoPoint(
+                lat=row["lat"],
+                lon=row["lon"],
+                region_code=row["region_code"],
+                full_address=row["full_address"] or "",
+            )
+        )
+
+    misses = [(key, *wanted[key]) for key in wanted if key not in points]
+    if not misses:
+        return points
+    allowed = budget.take(len(misses))
+    if allowed == 0:
+        return points
+    fresh = asyncio.run(_geocode_many(misses[:allowed]))
+    points.update(fresh)
+    if dry_run:
+        return points  # замер не пишет даже кэш — прогон остаётся повторяемым
+    for key, point in fresh.items():
+        address, slug = wanted[key]
+        db.execute(
+            text(_GEO_CACHE_UPSERT),
+            {
+                "key": key,
+                "slug": slug,
+                "address": address,
+                "lat": None if point is None else point.lat,
+                "lon": None if point is None else point.lon,
+                "region": None if point is None else point.region_code,
+                "full": None if point is None else point.full_address,
+            },
+        )
+    return points
+
+
 @dataclass
 class ImportCounters:
     """Разбор прогона. Числа обязаны сходиться, см. `check()`."""
@@ -115,9 +398,18 @@ class ImportCounters:
     skipped_artifact: int = 0
     skipped_not_moscow: int = 0
     skipped_invalid: int = 0
+    # Пред-геокод Авито: два РАЗНЫХ исхода, и смешивать их нельзя. Область —
+    # адрес разрешён, дом реальный, просто регион 50 (ждёт реестра). Не
+    # разрешён — DaData дома не нашла ИЛИ кончился бюджет вызовов; всплеск
+    # этого счётчика читается как «проверь квоту», а не «в Москве стало меньше
+    # домов».
+    skipped_oblast: int = 0
+    skipped_ungeocoded: int = 0
     selected: int = 0
     inserted: int = 0
     updated: int = 0
+    geocode_calls: int = 0  # фактически потраченных внешних вызовов
+    geocode_budget_exhausted: bool = False
 
     @property
     def written(self) -> int:
@@ -139,6 +431,8 @@ class ImportCounters:
             == self.selected
             + self.skipped_artifact
             + self.skipped_not_moscow
+            + self.skipped_oblast
+            + self.skipped_ungeocoded
             + self.skipped_invalid
         )
 
@@ -204,28 +498,40 @@ def import_msk_raw(
     limit: int | None = None,
     dry_run: bool = False,
     allow_unfiltered: bool = False,
+    geocode: bool = False,
+    geocode_limit: int = DEFAULT_GEOCODE_LIMIT,
 ) -> ImportCounters:
     """Переливает сырьё `msk_raw` в `listings`. Коммит — на каждом батче."""
     view = SOURCE_VIEWS[source]
     counters = ImportCounters()
     matcher = RealMatcherAdapter()
+    budget = GeocodeBudget(remaining=max(0, geocode_limit))
 
     filter_by_okrug = source == "cian"
-    if not filter_by_okrug:
+    if geocode and filter_by_okrug:
-        # У Авито в адресе нет ни города, ни округа, и нет координат — отсечь
+        # У Циана город опознаётся префиксом округа даром и без ошибок —
-        # область нечем. Пишем только по явному разрешению.
+        # тратить на него внешнюю квоту незачем.
+        raise SystemExit(f"source={source}: --geocode нужен только для avito")
+    if not filter_by_okrug and not geocode:
+        # Без пред-геокода у Авито по-прежнему нечем отделить Москву от области:
+        # ни города в адресе, ни координат. Пишем только по явному разрешению.
         if not (dry_run or allow_unfiltered):
             raise SystemExit(
                 f"source={source}: адрес не содержит признака города, Москву от "
-                "области не отличить. Нужен --allow-unfiltered (или --dry-run)."
+                "области не отличить. Нужен --geocode (штатный путь), "
+                "--allow-unfiltered (аварийный) или --dry-run."
             )
         logger.warning(
-            "source=%s: фильтр по округу НЕ применяется (в адресе нет города); "
+            "source=%s: пред-геокод ВЫКЛЮЧЕН — фильтра по городу нет вовсе; "
-            "строки лягут без geom — координат нет ни у одной карточки",
+            "строки лягут без geom и вперемешку с областью",
             source,
         )
+    if geocode and not dry_run:
+        db.execute(text(_GEO_CACHE_DDL))
+        db.commit()
 
     for rows in _iter_pages(db, view, batch_size=batch_size, limit=limit):
+        points = _resolve_points(db, rows, budget, dry_run=dry_run) if geocode else {}
         lots: list[ScrapedLot] = []
         for row in rows:
             counters.read += 1
@@ -237,6 +543,24 @@ def import_msk_raw(
             if filter_by_okrug and not is_moscow_address(address):
                 counters.skipped_not_moscow += 1
                 continue
+            if geocode:
+                point = points.get(geo_cache_key(address or "", avito_city_slug(payload)))
+                if point is None:
+                    counters.skipped_ungeocoded += 1
+                    continue
+                if point.region_code != MOSCOW_REGION_CODE:
+                    counters.skipped_oblast += 1
+                    continue
+                # Координаты кладём в КОПИЮ payload'а: исходную строку сырья не
+                # трогаем, пересбор корпуса от этого не зависит. geom появляется
+                # сразу — карточка идёт в radius-подбор аналогов, не дожидаясь
+                # ночного `geocode_missing`.
+                payload = {
+                    **payload,
+                    "lat": point.lat,
+                    "lon": point.lon,
+                    "geo_precision": "house",
+                }
             try:
                 lots.append(build_lot(payload))
             except ValidationError as exc:
@@ -267,10 +591,20 @@ def import_msk_raw(
             counters.updated,
         )
 
+    counters.geocode_calls = budget.spent
+    counters.geocode_budget_exhausted = budget.exhausted
+    if budget.exhausted:
+        logger.warning(
+            "msk_raw %s: дневной бюджет геокода (%d) исчерпан — остаток корпуса "
+            "подхватит следующий прогон, разрешённые адреса уже в кэше",
+            source,
+            geocode_limit,
+        )
     logger.info(
         "msk_raw %s ИТОГ%s: прочитано=%d отобрано=%d записано=%d "
         "(new=%d upd=%d, писатель пропустил=%d) | пропущено: не Москва=%d "
-        "артефакт=%d невалидный payload=%d | сходится=%s",
+        "область(50)=%d не разрешён=%d артефакт=%d невалидный payload=%d | "
+        "геокод-вызовов=%d | сходится=%s",
         source,
         " (dry-run)" if dry_run else "",
         counters.read,
@@ -280,8 +614,11 @@ def import_msk_raw(
         counters.updated,
         counters.writer_skipped if not dry_run else 0,
         counters.skipped_not_moscow,
+        counters.skipped_oblast,
+        counters.skipped_ungeocoded,
         counters.skipped_artifact,
         counters.skipped_invalid,
+        counters.geocode_calls,
         counters.check(),
     )
     return counters
@@ -300,7 +637,20 @@ def main() -> None:
     parser.add_argument(
         "--allow-unfiltered",
         action="store_true",
-        help="разрешить запись источника без признака города в адресе (avito)",
+        help="АВАРИЙНЫЙ режим: писать avito без фильтра по городу и без geom",
+    )
+    parser.add_argument(
+        "--geocode",
+        action="store_true",
+        help="штатный путь для avito: пред-геокод адреса (слаг + DaData), "
+        "в listings уходит только регион 77, область откладывается",
+    )
+    parser.add_argument(
+        "--geocode-limit",
+        type=int,
+        default=DEFAULT_GEOCODE_LIMIT,
+        help=f"потолок внешних вызовов за прогон (по умолчанию {DEFAULT_GEOCODE_LIMIT}; "
+        "free tier DaData — 10 000/сутки на аккаунт, его же ест автокомплит формы)",
     )
     args = parser.parse_args()
 
@@ -313,6 +663,8 @@ def main() -> None:
             limit=args.limit,
             dry_run=args.dry_run,
             allow_unfiltered=args.allow_unfiltered,
+            geocode=args.geocode,
+            geocode_limit=args.geocode_limit,
         )
     finally:
         db.close()

tradein-mvp/backend/tests/test_msk_raw_import.py

@@ -113,7 +113,9 @@ class _FakeSession:
         self.rows = rows
         self.commits = 0
 
-    def execute(self, _sql: Any, params: dict[str, Any]) -> _FakeResult:
+    def execute(self, _sql: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        if not params or "after" not in params:
+            return _FakeResult([])  # DDL кэша пред-геокода — строк не отдаёт
         after, limit = params["after"], params["limit"]
         return _FakeResult([r for r in self.rows if r["id"] > after][:limit])
 
@@ -194,3 +196,168 @@ def test_avito_refuses_to_write_without_explicit_opt_in(captured: list[dict[str,
         _FakeSession(rows), source="avito", batch_size=10, allow_unfiltered=True
     )
     assert counters.selected == 1  # префиксный фильтр к Авито не применяется
+
+
+# ── пред-геокод Авито ────────────────────────────────────────────────────────
+#
+# Риски этой ветки, которые тут и караулятся:
+#   * регион возьмут из слага, а не из ответа — Новая Москва (Троицк/Щербинка/
+#     Коммунарка) уедет в область, хотя это регион 77;
+#   * область (регион 50) просочится в listings;
+#   * слаг выпадет из ключа кэша — адреса-тёзки из разных муниципалитетов
+#     («Центральная ул.,5») схлопнутся в один дом;
+#   * повторный прогон заново потратит квоту по уже разрешённым адресам.
+
+
+def _avito(**over: Any) -> dict[str, Any]:
+    base = _payload(
+        source="avito",
+        source_url="https://www.avito.ru/moskva/kvartiry/2-k._kvartira_56_m_56_et._46317006",
+        address="Варшавское ш.,62к1",
+    )
+    base.update(over)
+    return base
+
+
+class _Sugg:
+    """Минимальный DadataSuggestion — нужны только поля, которые читает отбор."""
+
+    def __init__(self, kind: str, lat: float | None, lon: float | None, kladr: str | None) -> None:
+        self.kind, self.lat, self.lon, self.kladr_id = kind, lat, lon, kladr
+        self.unrestricted_value = "г Москва, ш Варшавское, д 62 к 1"
+
+
+def test_slug_comes_from_source_url() -> None:
+    assert mod.avito_city_slug(_avito()) == "moskva"
+    himki = _avito(source_url="https://www.avito.ru/himki/kvartiry/1")
+    assert mod.avito_city_slug(himki) == "himki"
+    assert mod.avito_city_slug(_avito(source_url="")) is None
+    assert mod.avito_city_slug({}) is None
+
+
+def test_cache_key_separates_same_address_in_different_towns() -> None:
+    """1110 адресов из 21 570 живут под несколькими слагами — это разные дома."""
+    assert mod.geo_cache_key("Центральная ул.,5", "moskva") != mod.geo_cache_key(
+        "Центральная ул.,5", "himki"
+    )
+    # нормализация адреса общая с геокодером — регистр и пробелы ключ не множат
+    assert mod.geo_cache_key("  Варшавское Ш.,62к1 ", "moskva") == mod.geo_cache_key(
+        "варшавское ш.,62к1", "moskva"
+    )
+
+
+def test_moscow_slug_narrows_constraint_others_keep_both_regions() -> None:
+    """Не-московский слаг может оказаться Новой Москвой — оба региона в констрейнте."""
+    assert mod._dadata_regions_for("moskva") == [mod.DADATA_MOSCOW_REGION]
+    assert mod._dadata_regions_for("moskovskaya_oblast_troitsk") == [
+        mod.DADATA_MOSCOW_REGION,
+        mod.DADATA_OBLAST_REGION,
+    ]
+    assert mod._dadata_regions_for(None) == [
+        mod.DADATA_MOSCOW_REGION,
+        mod.DADATA_OBLAST_REGION,
+    ]
+
+
+def test_region_is_read_from_the_answer_not_from_the_slug() -> None:
+    """КЛАДР ответа 77… → регион 77, даже если слаг был подмосковным."""
+    point = mod._point_from_suggestions([_Sugg("house", 55.6, 37.6, "77000000000000000")])
+    assert point is not None and point.region_code == 77
+    oblast = mod._point_from_suggestions([_Sugg("house", 55.9, 37.4, "50000000000000000")])
+    assert oblast is not None and oblast.region_code == 50
+
+
+def test_only_house_level_answers_become_points() -> None:
+    """Улица/город дают точку в середине улицы — для radius-подбора это хуже, чем ничего."""
+    assert mod._point_from_suggestions([_Sugg("street", 55.6, 37.6, "77000000000000000")]) is None
+    assert mod._point_from_suggestions([_Sugg("house", None, None, "77000000000000000")]) is None
+    # чужой регион (констрейнт пробит) — тоже не берём
+    assert mod._point_from_suggestions([_Sugg("house", 56.8, 60.6, "66000000000000000")]) is None
+
+
+def test_budget_stops_asking_instead_of_failing() -> None:
+    budget = mod.GeocodeBudget(remaining=2)
+    assert budget.take(5) == 2 and budget.exhausted and budget.spent == 2
+    assert budget.take(1) == 0  # квота кончилась — просто больше не спрашиваем
+
+
+def _geo_rows() -> list[dict[str, Any]]:
+    return [
+        {"id": 1, "payload": _avito(source_id="1")},  # Москва
+        {
+            "id": 2,
+            "payload": _avito(
+                source_id="2",
+                source_url="https://www.avito.ru/himki/kvartiry/2",
+                address="Ивановская ул.,5",
+            ),
+        },  # область
+        {
+            "id": 3,
+            "payload": _avito(source_id="3", address="Мкр-н имени В.Н. Махалина, 33"),
+        },  # не разрешился
+    ]
+
+
+@pytest.fixture()
+def geocoded(monkeypatch: pytest.MonkeyPatch) -> list[list[dict[str, Any]]]:
+    """Подменяет резолвер целиком — в сеть и в кэш-таблицу тест не ходит."""
+    seen: list[list[dict[str, Any]]] = []
+
+    def _fake_resolve(db: Any, rows: list[Any], budget: Any, **kwargs: Any) -> dict[str, Any]:
+        seen.append(list(rows))
+        return {
+            mod.geo_cache_key("Варшавское ш.,62к1", "moskva"): mod.GeoPoint(
+                55.6412, 37.6089, 77, "г Москва, ш Варшавское, д 62 к 1"
+            ),
+            mod.geo_cache_key("Ивановская ул.,5", "himki"): mod.GeoPoint(
+                55.8970, 37.4297, 50, "Московская обл, г Химки, ул Ивановская, д 5"
+            ),
+        }
+
+    monkeypatch.setattr(mod, "_resolve_points", _fake_resolve)
+    return seen
+
+
+def test_geocode_lets_avito_through_without_allow_unfiltered(
+    captured: list[dict[str, Any]], geocoded: list[list[dict[str, Any]]]
+) -> None:
+    """Пред-геокод — штатный путь: аварийный флаг больше не нужен."""
+    counters = mod.import_msk_raw(
+        _FakeSession(_geo_rows()), source="avito", batch_size=10, geocode=True
+    )
+    assert [lot.source_id for lot in captured[0]["lots"]] == ["1"]
+    assert counters.skipped_oblast == 1
+    assert counters.skipped_ungeocoded == 1
+    assert counters.check(), "счётчики обязаны сходиться с прочитанным"
+
+
+def test_coordinates_reach_the_writer(
+    captured: list[dict[str, Any]], geocoded: list[list[dict[str, Any]]]
+) -> None:
+    """Ради geom всё и затевалось — карточка не должна лечь без координат."""
+    mod.import_msk_raw(_FakeSession(_geo_rows()), source="avito", batch_size=10, geocode=True)
+    lot = captured[0]["lots"][0]
+    assert (lot.lat, lot.lon) == (55.6412, 37.6089)
+    assert lot.geo_precision == "house"
+    assert captured[0]["region_code"] == 77
+
+
+def test_oblast_is_never_written_as_region_77(
+    captured: list[dict[str, Any]], geocoded: list[list[dict[str, Any]]]
+) -> None:
+    mod.import_msk_raw(_FakeSession(_geo_rows()), source="avito", batch_size=10, geocode=True)
+    written = [lot.source_id for call in captured for lot in call["lots"]]
+    assert "2" not in written, "химкинская карточка не должна уехать в регион 77"
+
+
+def test_avito_without_geocode_demands_opt_in(captured: list[dict[str, Any]]) -> None:
+    rows = [{"id": 1, "payload": _avito()}]
+    with pytest.raises(SystemExit):
+        mod.import_msk_raw(_FakeSession(rows), source="avito", batch_size=10)
+
+
+def test_geocode_is_refused_for_cian(captured: list[dict[str, Any]]) -> None:
+    """У Циана город виден по префиксу округа даром — квоту на него не тратим."""
+    with pytest.raises(SystemExit):
+        mod.import_msk_raw(_FakeSession(_rows()), source="cian", batch_size=10, geocode=True)