"""Импорт московского сырья (`msk_raw.*_latest`) в `listings`.

Сырьё собрано отдельным коллектором и лежит в прод-схеме `msk_raw`: каждая строка
несёт `payload` — сериализованный `ScrapedLot` один в один (те же 54 ключа, что и
поля модели, см. `scraper_kit/base.py`). Свой писатель поэтому не нужен: собираем
`ScrapedLot(**payload)` и отдаём в штатный `save_listings(..., region_code=77)`.

Отбор Москвы (source=cian). Адрес карточки Циана города НЕ содержит, зато
начинается с округа: «ЦАО, ...», «СВАО, ...». По этому префиксу Москва и
опознаётся. Замер по проду (60 464 карточки): с округом — 35 551, ВСЕ внутри
bbox региона 77; без округа внутри bbox — 17 576 (это Московская область, регион
50, которого в реестре ещё нет, в этот импорт не берём); без округа вне bbox —
7 337. Отдельно 212 карточек с адресом вида «Екатеринбург (Cian)» — артефакт
парсера, считаются своим счётчиком, чтобы не растворяться в «не Москва».

Отбор Москвы (source=avito) по адресу НЕВОЗМОЖЕН: у Авито адрес — голая улица с
домом («Варшавское ш.,62к1»), ни города, ни округа, и координат нет НИ У ОДНОЙ
карточки (замер: lat/lon/cadastral_number/geo_precision пусты у всех 50 335).
Поэтому для Авито работает ПРЕД-ГЕОКОД (`--geocode`), а не префиксный фильтр.

Два сигнала, и оба нужны — ни один по отдельности не годится.

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`,
то есть по ссылке на ЖК, а не по продаже застройщиком. Замер по всем 60 464:
`raw_payload.is_from_developer` = true у НУЛЯ карточек, false у 29 000,
отсутствует у 31 464 — застройщик в этом корпусе не продаёт ничего, это вся
вторичка. В estimator'е стоит гвард (`estimator.py:5992-5995`): в аналоги идут
только строки с `listing_segment IS NULL` или 'vtorichka'. Скопируй мы метку
кита — 29 000 карточек выпали бы из подбора. Поэтому метка считается заново:
is_from_developer is True → 'novostroyki', иначе → 'vtorichka'.

Идемпотентность — на стороне `save_listings`: он делает upsert
`ON CONFLICT (dedup_hash) DO UPDATE` плюс reconcile-UPDATE по
`(source, source_id)` на случай дрейфа хеша. `dedup_hash` = sha256(source +
source_id) считает сам кит (`ScrapedLot.compute_dedup_hash`), цена в ключ не
входит. Повторный прогон поэтому обновляет те же строки, а не плодит дубли;
курсор идёт по `id` вью, так что порядок и полнота обхода от прогона к прогону
одинаковы.

Отбор Москвы (source=yandex) стоит ноль вызовов: адрес приходит полным и
нормализованным («Россия, Москва, Коробейников переулок, 1»), регион читается
вторым компонентом. Замер по 21 393 карточкам первого прохода — ровно два
значения, «Москва» 10 610 и «Московская область» 10 783. Координаты у Яндекса
заполнены у 100% карточек, поэтому ни геокод, ни `geocode_missing` ему не нужны.

Отбор Москвы (source=domclick) тоже стоит ноль вызовов, но по другой причине:
сырьё собрано запросом с GUID-ом Москвы и отфильтровано по bbox ещё на стороне
сборщика, поэтому областных карточек в нём нет по построению. Регион читается
ПЕРВЫМ компонентом адреса («Москва, улица Нижние Мнёвники, 16 к2»), и фильтр
оставлен второй линией — на случай, если в ту же таблицу когда-нибудь приедет
второй город. Замер по 5 024 карточкам первого прохода: единственное значение
первого компонента — «Москва». Новая Москва приходит как «Москва, пос. Птичное,
…», то есть посёлок стоит вторым компонентом и регион 77 не ломает. Координаты
заполнены у 100% карточек.

Запуск:
    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 yandex
    python -m app.tasks.msk_raw_import --source domclick
    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
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 = (
    "ЦАО",
    "САО",
    "СВАО",
    "ВАО",
    "ЮВАО",
    "ЮАО",
    "ЮЗАО",
    "ЗАО",
    "СЗАО",
    "ЗелАО",
    "НАО",
    "ТАО",
)
# Lookahead вместо \b: следом за округом идёт запятая/пробел, но НЕ буква — иначе
# «ЗАО» матчило бы начало гипотетического «ЗАОзёрная».
_MOSCOW_OKRUG_RE = re.compile(
    r"^(?:" + "|".join(MOSCOW_OKRUGS) + r")(?![А-Яа-яЁёA-Za-z])",
)
# Артефакт парсера: адрес вида «Екатеринбург (Cian)» в московском корпусе.
_ARTIFACT_RE = re.compile(r"Екатеринбург", re.IGNORECASE)

# Вью-источники. Только whitelist: имя подставляется в SQL текстом, параметром
# идентификатор не передать.
SOURCE_VIEWS = {
    "cian": "msk_raw.cian_latest",
    "avito": "msk_raw.avito_latest",
    "yandex": "msk_raw.yandex_latest",
    "domclick": "msk_raw.domclick_latest",
}

_PAGE_SQL = """
SELECT id, payload
FROM {view}
WHERE id > :after
ORDER BY id
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_EXISTS = "SELECT to_regclass('msk_raw.avito_geocode')"
_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] = {}
    # Кэш создаётся только боевым прогоном (`geocode and not dry_run`), поэтому на
    # первом `--dry-run` таблицы ещё нет и SELECT по ней роняет весь замер — то
    # есть ломается ровно та репетиция, ради которой сухой прогон и существует.
    # Проверяем наличие отношения, а не ловим исключение: в Postgres упавший
    # оператор кладёт транзакцию целиком, и except потребовал бы rollback.
    cache_exists = db.execute(text(_GEO_CACHE_EXISTS)).scalar() is not None
    cached = (
        db.execute(text(_GEO_CACHE_SELECT), {"keys": list(wanted)}).mappings().all()
        if cache_exists
        else ()
    )
    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()`."""

    read: int = 0
    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:
        return self.inserted + self.updated

    @property
    def writer_skipped(self) -> int:
        """Отобрано, но писатель строку не тронул.

        `save_listings` возвращает только (inserted, updated); неизменные строки,
        уже виденные сегодня, он пропускает своим гейтом (#2992). Остаток честно
        показываем отдельно, а не растворяем в «записано».
        """
        return self.selected - self.written

    def check(self) -> bool:
        return (
            self.read
            == self.selected
            + self.skipped_artifact
            + self.skipped_not_moscow
            + self.skipped_oblast
            + self.skipped_ungeocoded
            + self.skipped_invalid
        )


def is_artifact_address(address: str | None) -> bool:
    """Адрес чужого города в московском корпусе (артефакт парсера)."""
    return bool(address) and _ARTIFACT_RE.search(address) is not None


def is_moscow_address(address: str | None) -> bool:
    """Москва опознаётся префиксом административного округа."""
    if not address:
        return False
    return _MOSCOW_OKRUG_RE.match(address.strip()) is not None


def is_moscow_yandex_address(address: str | None) -> bool:
    """У Яндекса регион — второй компонент полного адреса.

    Адрес приходит нормализованным и с городом: «Россия, Москва, Коробейников
    переулок, 1». Замер по 21 393 карточкам первого прохода: во втором
    компоненте ровно ДВА значения — «Москва» 10 610 и «Московская область»
    10 783, третьего не встречается. Поэтому ни префикса округа (как у Циана),
    ни внешнего геокода (как у Авито) источнику не нужно: разделение 77 и 50
    читается из самой карточки и стоит ноль вызовов.

    Новая Москва отдельным значением НЕ приходит — Троицк и Зеленоград Яндекс
    кладёт под «Москва», что совпадает с кодом региона 77.
    """
    parts = [part.strip() for part in (address or "").split(",")]
    return len(parts) > 1 and parts[1] == "Москва"


def is_moscow_domclick_address(address: str | None) -> bool:
    """У ДомКлика регион — ПЕРВЫЙ компонент адреса: «Москва, улица …».

    Источник собран запросом с GUID-ом Москвы в параметре address и дополнительно
    отфильтрован по bbox на стороне сборщика, так что областных карточек в сырье
    и не должно быть. Фильтр здесь — вторая линия: сменится GUID в сборщике или
    появится второй город в той же таблице — импорт не потащит его в Москву
    молча. Замер по 5 024 карточкам первого прохода: первый компонент имеет ровно
    одно значение, «Москва», областных нет ни одной.

    Новая Москва отдельным значением НЕ приходит: «Москва, x. Ильичевка, …»,
    «Москва, пос. Птичное, …» — посёлок стоит ВТОРЫМ компонентом, первый всегда
    город, что совпадает с кодом региона 77.
    """
    parts = [part.strip() for part in (address or "").split(",")]
    return bool(parts) and parts[0] == "Москва"


# Источники, у которых город виден в самой карточке. Ключ отсутствует —
# источник про город молчит, и без пред-геокода писать его нельзя (avito).
CITY_FILTERS = {
    "cian": is_moscow_address,
    "yandex": is_moscow_yandex_address,
    "domclick": is_moscow_domclick_address,
}


def recompute_listing_segment(payload: dict) -> str:
    """Заново считаем сегмент: 'novostroyki' только при продаже застройщиком.

    Обоснование — в докстринге модуля: метка кита означает лишь ссылку на ЖК.
    """
    raw = payload.get("raw_payload") or {}
    if not isinstance(raw, dict):
        return "vtorichka"
    return "novostroyki" if raw.get("is_from_developer") is True else "vtorichka"


def build_lot(payload: dict) -> ScrapedLot:
    """`payload` → `ScrapedLot` с пересчитанным сегментом.

    Ключи, которых в модели нет, отбрасываем явно (по `model_fields`), а не
    полагаемся на настройку extra у pydantic-модели.
    """
    known = {k: v for k, v in payload.items() if k in ScrapedLot.model_fields}
    known["listing_segment"] = recompute_listing_segment(payload)
    return ScrapedLot(**known)


def _iter_pages(db: Session, view: str, *, batch_size: int, limit: int | None):
    """Keyset-пагинация по `id` — весь корпус в память не тянем."""
    after = 0
    taken = 0
    sql = text(_PAGE_SQL.format(view=view))
    while True:
        page_size = batch_size
        if limit is not None:
            page_size = min(batch_size, limit - taken)
            if page_size <= 0:
                return
        rows = db.execute(sql, {"after": after, "limit": page_size}).mappings().all()
        if not rows:
            return
        after = rows[-1]["id"]
        taken += len(rows)
        yield rows


def import_msk_raw(
    db: Session,
    *,
    source: str = "cian",
    batch_size: int = DEFAULT_BATCH_SIZE,
    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))

    # Источник, который сам говорит про город: у Циана это префикс округа, у
    # Яндекса — второй компонент полного адреса. Авито не говорит ничего, ему
    # нужен пред-геокод, поэтому его в словаре нет.
    city_filter = CITY_FILTERS.get(source)
    if geocode and city_filter is not None:
        # Город опознаётся даром и без ошибок — тратить на него внешнюю квоту
        # незачем.
        raise SystemExit(f"source={source}: --geocode нужен только для avito")
    if city_filter is None and not geocode:
        # Без пред-геокода у Авито по-прежнему нечем отделить Москву от области:
        # ни города в адресе, ни координат. Пишем только по явному разрешению.
        if not (dry_run or allow_unfiltered):
            raise SystemExit(
                f"source={source}: адрес не содержит признака города, Москву от "
                "области не отличить. Нужен --geocode (штатный путь), "
                "--allow-unfiltered (аварийный) или --dry-run."
            )
        logger.warning(
            "source=%s: пред-геокод ВЫКЛЮЧЕН — фильтра по городу нет вовсе; "
            "строки лягут без 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
            payload = row["payload"] or {}
            address = payload.get("address")
            if is_artifact_address(address):
                counters.skipped_artifact += 1
                continue
            if city_filter is not None and not city_filter(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:
                counters.skipped_invalid += 1
                logger.warning("msk_raw id=%s не собрался в ScrapedLot: %s", row["id"], exc)

        counters.selected += len(lots)
        if dry_run or not lots:
            continue

        inserted, updated = save_listings(
            db,
            lots,
            matcher=matcher,
            region_code=MOSCOW_REGION_CODE,
            city=MOSCOW_CITY,
        )
        counters.inserted += inserted
        counters.updated += updated
        db.commit()  # батч зафиксирован — обрыв не отматывает всю работу
        logger.info(
            "msk_raw %s: прочитано=%d отобрано=%d записано=%d (new=%d upd=%d)",
            source,
            counters.read,
            counters.selected,
            counters.written,
            counters.inserted,
            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 "
        "область(50)=%d не разрешён=%d артефакт=%d невалидный payload=%d | "
        "геокод-вызовов=%d | сходится=%s",
        source,
        " (dry-run)" if dry_run else "",
        counters.read,
        counters.selected,
        counters.written,
        counters.inserted,
        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


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    parser = argparse.ArgumentParser(description="Импорт сырья msk_raw в listings (регион 77)")
    parser.add_argument("--source", choices=sorted(SOURCE_VIEWS), default="cian")
    parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE)
    parser.add_argument("--limit", type=int, default=None, help="обработать не больше N карточек")
    parser.add_argument("--dry-run", action="store_true", help="ничего не пишет, только счётчики")
    parser.add_argument(
        "--allow-unfiltered",
        action="store_true",
        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()

    db = SessionLocal()
    try:
        import_msk_raw(
            db,
            source=args.source,
            batch_size=args.batch_size,
            limit=args.limit,
            dry_run=args.dry_run,
            allow_unfiltered=args.allow_unfiltered,
            geocode=args.geocode,
            geocode_limit=args.geocode_limit,
        )
    finally:
        db.close()


if __name__ == "__main__":
    main()