tradein-mvp/backend/data/sql/196_listings_city.sql

@@ -0,0 +1,44 @@
+-- 196_listings_city.sql
+-- Issue #2594 — критичный дефект: скрапер знает город в момент сбора (city_slug из
+-- CITY_LOCATIONS/CITY_ANCHORS, packages/scraper-kit/.../orchestration/pipeline.py), но
+-- НИКУДА его не пишет. Провайдеры (avito/cian) часто отдают адрес БЕЗ города в тексте
+-- ("ул. Победы, 30" вместо "Нижний Тагил, ул. Победы, 30") — cian даже явно вырезает
+-- location-часть перед записью (skip_types = {"location", "metro"}, providers/cian/serp.py).
+-- Без явного города такой адрес при геокодинге считается «город не назван» → попадает
+-- в EKB-only локальные реестры (ekb_geoportal_buildings/gendesign_cad_buildings) и
+-- коллизирует с одноимённой екатеринбургской улицей (Ленина/Победы/Тенистая — сотни
+-- совпадений) → объявление получает координаты Екатеринбурга и тянет медиану чужих цен.
+--
+-- Fix:
+--   Add listings.city TEXT column. Проставляется НЕПОСРЕДСТВЕННО из контекста
+--   развёртки (город известен вызывающему коду — city_slug/CITY_LOCATIONS для oblast,
+--   "Екатеринбург" для EKB-развёрток) — НЕ парсингом текста адреса. См.
+--   scraper_kit.base.save_listings(..., city=...) + scraper_kit.orchestration.pipeline
+--   .resolve_city_name(). Раздельная колонка (а не дописывание города в address) —
+--   исходный текст адреса не портится, downstream text-парсеры (geocoder._parse_street_house,
+--   geocoder._names_non_ekb_city, estimator._parse_street_house, house-matching) продолжают
+--   работать НЕИЗМЕНЁННЫМИ на исходном сыром тексте — риск регрессии на bare-form адресах
+--   без street-маркера ("Дружинина, 33") исключён.
+--
+-- Scope (#2594): только write-path для НОВЫХ листингов (go-forward). Бэкфилл city для
+-- уже накопленных строк (restore по тому, какая развёртка их когда-то принесла) —
+-- отдельная задача, НЕ эта миграция.
+--
+-- Idempotency:
+--   ALTER TABLE ... ADD COLUMN IF NOT EXISTS — safe on re-run.
+--   BEGIN/COMMIT block.
+--
+-- Dependencies:
+--   002_core_tables.sql (listings table).
+
+BEGIN;
+
+ALTER TABLE listings ADD COLUMN IF NOT EXISTS city text;
+
+COMMENT ON COLUMN listings.city IS
+    'Город объявления (#2594) — проставляется из контекста развёртки '
+    '(city_slug city-sweep / "Екатеринбург" default), НЕ парсингом address. '
+    'NULL — листинг записан до этой миграции ИЛИ путём, ещё не проставляющим город '
+    '(admin ad-hoc /admin/scrape, manual ingest-скрипты).';
+
+COMMIT;

tradein-mvp/backend/tests/test_city_sweep.py

@@ -27,6 +27,41 @@ def test_ekb_anchors_count() -> None:
         assert isinstance(name, str) and name
 
 
+# ── resolve_city_name (#2594) ────────────────────────────────────────────────
+
+
+def test_resolve_city_name_known_oblast_slugs() -> None:
+    """Каждый city_slug из CITY_LOCATIONS резолвится в человекочитаемое имя."""
+    from scraper_kit.orchestration.pipeline import CITY_LOCATIONS, resolve_city_name
+
+    expected = {
+        "nizhniy_tagil": "Нижний Тагил",
+        "kamensk_uralskiy": "Каменск-Уральский",
+        "pervouralsk": "Первоуральск",
+        "verkhnyaya_pyshma": "Верхняя Пышма",
+        "serov": "Серов",
+    }
+    # CITY_DISPLAY_NAMES обязан покрывать ровно те же slug'и, что CITY_LOCATIONS
+    # (иначе oblast-город бы тихо получил ЕКБ-дефолт вместо своего имени).
+    assert set(expected) == set(CITY_LOCATIONS)
+    for slug, name in expected.items():
+        assert resolve_city_name(slug) == name
+
+
+def test_resolve_city_name_none_defaults_to_ekaterinburg() -> None:
+    """city_slug=None — ЕКБ-развёртка той же функции, НЕ «город неизвестен» (#2594 симметрия)."""
+    from scraper_kit.orchestration.pipeline import EKATERINBURG_CITY_NAME, resolve_city_name
+
+    assert resolve_city_name(None) == EKATERINBURG_CITY_NAME == "Екатеринбург"
+
+
+def test_resolve_city_name_unknown_slug_defaults_to_ekaterinburg() -> None:
+    """Неизвестный slug — тот же ЕКБ-дефолт, что и get_city_location/get_city_anchors."""
+    from scraper_kit.orchestration.pipeline import resolve_city_name
+
+    assert resolve_city_name("nonexistent_city") == "Екатеринбург"
+
+
 # ── CitySweepCounters ───────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/test_listings_city_from_sweep.py

@@ -0,0 +1,232 @@
+"""#2594: listings.city проставляется из контекста развёртки, не парсингом адреса.
+
+Критичный дефект: скрапер ЗНАЕТ город в момент сбора (city_slug из
+scraper_kit.orchestration.pipeline.CITY_LOCATIONS/CITY_ANCHORS), но раньше нигде его
+не записывал. Провайдеры (avito/cian) часто отдают адрес БЕЗ города в тексте
+("ул. Победы, 30" вместо "Нижний Тагил, ул. Победы, 30" — cian даже явно вырезает
+location-часть, providers/cian/serp.py `_format_address` skip_types={"location",...}).
+Без города такой адрес при геокодинге считался «город не назван» и коллизировал с
+одноимённой ЕКБ-улицей (Ленина/Победы/Тенистая — сотни совпадений в ЕКБ-реестре).
+
+Fix: отдельная колонка `listings.city`, проставляется из sweep-контекста (НЕ парсингом
+address) через `scraper_kit.base.save_listings(..., city=...)` +
+`scraper_kit.orchestration.pipeline.resolve_city_name(city_slug)`. Тесты здесь проверяют
+write-path (save_listings SQL) и pure resolve_city_name; orchestration-level проверки
+(save_listings вызывается с правильным city= из каждого sweep) — в
+test_scraper_kit_pipeline_parity.py / test_scraper_kit_pipeline_parity2.py.
+
+Границы (#2594): бэкфилл уже накопленных строк — НЕ в этой задаче.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.base import ScrapedLot as KitLot
+from scraper_kit.base import save_listings as kit_save_listings
+
+
+@contextmanager
+def _nested_ctx() -> Any:
+    yield MagicMock()
+
+
+def _mock_db_insert_path(listing_id: int = 42) -> MagicMock:
+    """Session mock для fresh INSERT path (xmax = 0 → inserted)."""
+    insert_row = MagicMock()
+    insert_row.id = listing_id
+    insert_row.inserted = True
+
+    db = MagicMock()
+
+    def _execute(sql: Any, params: dict[str, Any] | None = None) -> MagicMock:
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = None
+        elif "FROM listings_snapshots" in s:
+            res.fetchone.return_value = None
+        elif "INSERT INTO listings (" in s:
+            res.fetchone.return_value = insert_row
+        else:
+            res.fetchone.return_value = None
+        return res
+
+    db.execute.side_effect = _execute
+    db.begin_nested.side_effect = _nested_ctx
+    return db
+
+
+def _find_call(db: MagicMock, needle: str) -> tuple[str, dict[str, Any]]:
+    for call in db.execute.call_args_list:
+        sql = str(call.args[0])
+        if needle in sql:
+            params = call.args[1] if len(call.args) > 1 else {}
+            return sql, params
+    raise AssertionError(f"SQL containing {needle!r} not found")
+
+
+def _kit_matcher() -> MagicMock:
+    matcher = MagicMock()
+    matcher.match_or_create_house.return_value = (101, 1.0, "new")
+    matcher.upsert_listing_source.return_value = None
+    return matcher
+
+
+def _lot(source: str = "avito", source_id: str = "1", address: str | None = None) -> KitLot:
+    return KitLot(
+        source=source,
+        source_url=f"https://www.{source}.ru/item/{source_id}",
+        source_id=source_id,
+        address=address,
+        price_rub=3_000_000,
+    )
+
+
+# ── save_listings(..., city=...) — INSERT path ────────────────────────────────
+
+
+def test_save_listings_writes_city_into_insert_sql() -> None:
+    """city="Нижний Тагил" передаётся в SQL params И колонка есть в INSERT-списке."""
+    db = _mock_db_insert_path()
+    lot = _lot(address="ул. Победы, 30")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66, city="Нижний Тагил")
+
+    sql, params = _find_call(db, "INSERT INTO listings (")
+    assert "city" in sql, "city column must be in INSERT column list"
+    assert params["city"] == "Нижний Тагил"
+    # address НЕ тронут — критичное требование #2594 (раздельная колонка, а не
+    # дописывание города в текст адреса, чтобы не сломать downstream text-парсеры).
+    assert params["address"] == "ул. Победы, 30"
+
+
+def test_save_listings_city_defaults_to_none_backward_compat() -> None:
+    """Caller без city= (старые/ad-hoc пути) — колонка остаётся NULL, backward-compatible."""
+    db = _mock_db_insert_path()
+    lot = _lot(address="ул. Малышева, 30")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    _sql, params = _find_call(db, "INSERT INTO listings (")
+    assert params["city"] is None
+
+
+def test_save_listings_ekaterinburg_city_written_unchanged_address() -> None:
+    """ЕКБ-развёртка (city="Екатеринбург") — тот же путь, address не деградирует."""
+    db = _mock_db_insert_path()
+    lot = _lot(address="ул. Малышева, 30")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66, city="Екатеринбург")
+
+    _sql, params = _find_call(db, "INSERT INTO listings (")
+    assert params["city"] == "Екатеринбург"
+    assert params["address"] == "ул. Малышева, 30"
+
+
+# ── ON CONFLICT DO UPDATE / reconcile UPDATE — COALESCE не затирает known city ──
+
+
+def test_save_listings_on_conflict_coalesces_city() -> None:
+    """ON CONFLICT DO UPDATE — city = COALESCE(EXCLUDED.city, listings.city), не blind overwrite."""
+    db = _mock_db_insert_path()
+    lot = _lot(address="ул. Победы, 30")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66, city="Нижний Тагил")
+
+    sql, _params = _find_call(db, "INSERT INTO listings (")
+    assert "city = COALESCE(EXCLUDED.city, listings.city)" in sql
+
+
+def test_save_listings_reconcile_update_coalesces_city() -> None:
+    """dedup_hash-drift reconcile UPDATE path — тоже COALESCE(:city, city), не blind overwrite."""
+    import psycopg.errors
+    from sqlalchemy.exc import IntegrityError
+
+    uv_orig = psycopg.errors.UniqueViolation()
+    integrity_err = IntegrityError("INSERT INTO listings ...", {}, uv_orig)
+    rec_row = MagicMock()
+    rec_row.id = 88
+
+    db = MagicMock()
+
+    def _execute(sql: Any, params: dict[str, Any] | None = None) -> MagicMock:
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = None
+        elif "FROM listings_snapshots" in s:
+            res.fetchone.return_value = None
+        elif "INSERT INTO listings (" in s:
+            raise integrity_err
+        elif "UPDATE listings" in s and "SET dedup_hash" in s:
+            res.fetchone.return_value = rec_row
+        else:
+            res.fetchone.return_value = None
+        return res
+
+    db.execute.side_effect = _execute
+
+    @contextmanager
+    def _nested() -> Any:
+        try:
+            yield MagicMock()
+        except IntegrityError:
+            raise
+
+    db.begin_nested.side_effect = _nested
+
+    lot = _lot(source="avito", source_id="7960764619", address="ул. Тенистая, 17")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66, city="Серов")
+
+    sql, params = _find_call(db, "SET dedup_hash")
+    assert "city = COALESCE(:city, city)" in sql
+    assert params["city"] == "Серов"
+
+
+# ── Migration 196: listings.city column ────────────────────────────────────────
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_196 = _SQL_DIR / "196_listings_city.sql"
+
+
+def test_migration_196_exists() -> None:
+    assert _MIGRATION_196.is_file(), f"missing migration: {_MIGRATION_196}"
+
+
+def test_migration_196_is_transactional() -> None:
+    sql = _MIGRATION_196.read_text("utf-8")
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_196_idempotent_add_column() -> None:
+    sql = _MIGRATION_196.read_text("utf-8")
+    assert "ADD COLUMN IF NOT EXISTS city" in sql
+
+
+def test_migration_196_no_psycopg_cast_trap() -> None:
+    """psycopg v3: никаких :param::type (не применимо тут — чистый DDL — но проверяем
+    на будущее, если файл когда-нибудь обрастёт bind-параметрами)."""
+    sql = _MIGRATION_196.read_text("utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+def test_migration_196_non_destructive() -> None:
+    sql = _MIGRATION_196.read_text("utf-8")
+    assert "DROP" not in sql.upper()
+    assert "DELETE" not in sql.upper()
+    assert "TRUNCATE" not in sql.upper()

tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py

@@ -93,6 +93,7 @@ class _Scenario:
         avito_serp_ok_not_banned: bool = True,
         avito_proxy_max_rotations: int = 0,
         lots_have_house_url: bool = False,
+        city_slug: str | None = None,
     ) -> None:
         self.anchors = anchors
         self.per_anchor = per_anchor
@@ -105,6 +106,9 @@ class _Scenario:
         self.avito_serp_ok_not_banned = avito_serp_ok_not_banned
         self.avito_proxy_max_rotations = avito_proxy_max_rotations
         self.lots_have_house_url = lots_have_house_url
+        # #2594: city_slug развёртки — прокидывается в run_avito_city_sweep(city_slug=...)
+        # для проверки, что save_listings получает правильный city=... из контекста.
+        self.city_slug = city_slug
 
     def _config(self) -> SimpleNamespace:
         return SimpleNamespace(
@@ -170,11 +174,16 @@ def _async_session_cm() -> MagicMock:
     return sess
 
 
-async def _drive(scenario: _Scenario) -> _DriveResult:
+async def _drive(scenario: _Scenario, *, capture: dict[str, Any] | None = None) -> _DriveResult:
+    """capture: опциональный dict — если передан, кладём туда save_mock (#2594) для
+    инспекции call_args (city=...) без изменения возвращаемого _DriveResult (backward-compat
+    для всех существующих вызовов _drive без capture)."""
     recorder = _RunsRecorder()
     db = _make_db(scenario)
     scraper = _make_scraper(scenario, AvitoBlockedError)
     save_mock = MagicMock(side_effect=scenario._save_side_effects())
+    if capture is not None:
+        capture["save_mock"] = save_mock
 
     imv_res = None
     if scenario.imv_result is not None:
@@ -208,6 +217,7 @@ async def _drive(scenario: _Scenario) -> _DriveResult:
             shutdown_requested=lambda: False,
             radius_m=1000,
             anchors=scenario.anchors,
+            city_slug=scenario.city_slug,
             pages_per_anchor=1,
             enrich_houses=scenario.enrich_houses,
             detail_top_n=scenario.detail_top_n,
@@ -307,3 +317,43 @@ async def test_imv_phase_counters() -> None:
     assert counters["imv_attempted"] == 3
     assert counters["imv_enriched"] == 2
     assert counters["imv_failed"] == 1
+
+
+# ── #2594: listings.city проставляется из контекста развёртки ────────────────
+#
+# Критичный дефект: развёртка ЗНАЕТ город (city_slug), но раньше НИКУДА его не
+# писала — адрес без города в тексте ("ул. Победы, 30") при геокодинге считался
+# «город не назван» и коллизировал с одноимённой ЕКБ-улицей. Тесты проверяют, что
+# save_listings() теперь получает правильный city= для обоих случаев: явный
+# oblast-город (city_slug задан) И EKB-развёртка той же функции (city_slug=None —
+# симметрия, а не «не знаем город»).
+
+
+@pytest.mark.asyncio
+async def test_city_stamped_from_city_slug() -> None:
+    """city_slug='nizhniy_tagil' → save_listings(..., city='Нижний Тагил')."""
+    scenario = _Scenario(
+        anchors=[(56.84, 60.60, "A1")],
+        per_anchor=[("lots", 3, 3, 0)],
+        city_slug="nizhniy_tagil",
+    )
+    capture: dict[str, Any] = {}
+    await _drive(scenario, capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args.kwargs["city"] == "Нижний Тагил"
+
+
+@pytest.mark.asyncio
+async def test_city_defaults_to_ekaterinburg_when_no_city_slug() -> None:
+    """city_slug=None (ЕКБ-развёртка той же run_avito_city_sweep) →
+    save_listings(..., city='Екатеринбург') — симметрия с oblast-городами (#2594),
+    а не оставленный NULL."""
+    scenario = _Scenario(
+        anchors=[(56.84, 60.60, "A1")],
+        per_anchor=[("lots", 3, 3, 0)],
+        city_slug=None,
+    )
+    capture: dict[str, Any] = {}
+    await _drive(scenario, capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args.kwargs["city"] == "Екатеринбург"

tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py

@@ -136,7 +136,10 @@ def _yandex_scraper(combos: list[tuple[str, list[Any]]]) -> MagicMock:
     return _ctx_scraper(fetch_around_multi_room=_fetch)
 
 
-async def _drive_yandex_city() -> _DriveResult:
+async def _drive_yandex_city(
+    *, city_slug: str | None = None, capture: dict[str, Any] | None = None
+) -> _DriveResult:
+    """capture: опционально — если передан, кладём save_mock (#2594, инспекция city=...)."""
     recorder = _RunsRecorder()
     db = MagicMock()
     combos = [
@@ -145,6 +148,8 @@ async def _drive_yandex_city() -> _DriveResult:
     ]
     scraper = _yandex_scraper(combos)
     save_mock = MagicMock(side_effect=[(2, 0), (1, 0)])
+    if capture is not None:
+        capture["save_mock"] = save_mock
     cfg = _config()
     enrichment = MagicMock()
     enrichment.record_yandex_price_history = MagicMock(return_value=5)
@@ -160,6 +165,7 @@ async def _drive_yandex_city() -> _DriveResult:
             enrichment=enrichment,
             run_id=1,
             anchors=None,
+            city_slug=city_slug,
             pages_per_anchor=1,
             request_delay_sec=0.0,
             enrich_address=False,
@@ -185,13 +191,18 @@ def _cian_lot(segment: str) -> MagicMock:
     return MagicMock(listing_segment=segment, house_source=None, house_ext_id=None)
 
 
-async def _drive_cian_city() -> _DriveResult:
+async def _drive_cian_city(
+    *, city_slug: str | None = None, capture: dict[str, Any] | None = None
+) -> _DriveResult:
+    """capture: опционально — если передан, кладём save_mock (#2594, инспекция city=...)."""
     recorder = _RunsRecorder()
     db = MagicMock()
     # 3 novostroyki + 2 secondary → newbuilding_only оставит 3.
     lots = [_cian_lot("novostroyki")] * 3 + [_cian_lot("vtorichnaya")] * 2
     scraper = _ctx_scraper(fetch_around_multi_room=AsyncMock(return_value=lots))
     save_mock = MagicMock(side_effect=[(3, 0)])
+    if capture is not None:
+        capture["save_mock"] = save_mock
     cfg = _config()
     with (
         patch(f"{PFX}.CianScraper", return_value=scraper),
@@ -204,6 +215,7 @@ async def _drive_cian_city() -> _DriveResult:
             matcher=MagicMock(),
             run_id=1,
             anchors=[(56.84, 60.60, "A1")],
+            city_slug=city_slug,
             radius_m=1000,
             pages_per_anchor=1,
             request_delay_sec=0.0,
@@ -228,7 +240,10 @@ async def test_cian_city_sweep() -> None:
 # ── DomClick city sweep ───────────────────────────────────────────────────────
 
 
-async def _drive_domclick(*, lots_n: int, blocked: bool) -> _DriveResult:
+async def _drive_domclick(
+    *, lots_n: int, blocked: bool, capture: dict[str, Any] | None = None
+) -> _DriveResult:
+    """capture: опционально — если передан, кладём save_mock (#2594, инспекция city=...)."""
     recorder = _RunsRecorder()
     db = MagicMock()
     lots = [MagicMock() for _ in range(lots_n)]
@@ -239,6 +254,8 @@ async def _drive_domclick(*, lots_n: int, blocked: bool) -> _DriveResult:
         fetch_errors=0,
     )
     save_mock = MagicMock(side_effect=[(lots_n, 0)] if lots_n else [])
+    if capture is not None:
+        capture["save_mock"] = save_mock
     cfg = _config()
     with (
         patch(f"{PFX}.DomClickScraper", return_value=scraper),
@@ -272,7 +289,8 @@ async def test_domclick_city_sweep_blocked_failed() -> None:
 # ── Avito newbuilding sweep ───────────────────────────────────────────────────
 
 
-async def _drive_nb_sweep() -> _DriveResult:
+async def _drive_nb_sweep(*, capture: dict[str, Any] | None = None) -> _DriveResult:
+    """capture: опционально — если передан, кладём save_mock (#2594, инспекция city=...)."""
     recorder = _RunsRecorder()
     db = MagicMock()
     lots = [MagicMock() for _ in range(6)]
@@ -281,6 +299,8 @@ async def _drive_nb_sweep() -> _DriveResult:
     scraper._browser = None
     scraper.fetch_newbuildings = AsyncMock(return_value=lots)
     save_mock = MagicMock(side_effect=[(5, 1)])
+    if capture is not None:
+        capture["save_mock"] = save_mock
     cfg = _config()
     with (
         patch(f"{PFX}.AvitoScraper", return_value=scraper),
@@ -319,7 +339,8 @@ def _full_load_scraper(buckets: list[tuple[str, list[Any]]]) -> MagicMock:
     return scraper
 
 
-async def _drive_full_load(*, source: str) -> _DriveResult:
+async def _drive_full_load(*, source: str, capture: dict[str, Any] | None = None) -> _DriveResult:
+    """capture: опционально — если передан, кладём save_mock (#2594, инспекция city=...)."""
     recorder = _RunsRecorder()
     db = MagicMock()
     buckets = [
@@ -328,6 +349,8 @@ async def _drive_full_load(*, source: str) -> _DriveResult:
     ]
     scraper = _full_load_scraper(buckets)
     save_mock = MagicMock(side_effect=[(2, 0), (1, 0)])
+    if capture is not None:
+        capture["save_mock"] = save_mock
     cfg = _config()
 
     fn_map = {
@@ -365,3 +388,77 @@ async def test_full_load_smoke(source: str) -> None:
     assert counters["saved_inserted"] == 3
     assert counters["saved_updated"] == 0
     assert calls[-1][0] == "mark_done"
+
+
+# ── #2594: listings.city проставляется из контекста развёртки ────────────────
+#
+# Критичный дефект: развёртка ЗНАЕТ город (city_slug), но раньше НИКУДА его не
+# писала. Тесты проверяют save_listings(..., city=...) для yandex/cian city-sweep
+# (oblast + EKB-симметрия), domclick (EKB-only city_id) и full_load'ов (ЕКБ вторичка).
+
+
+@pytest.mark.asyncio
+async def test_yandex_city_sweep_stamps_city_from_slug() -> None:
+    """city_slug='kamensk_uralskiy' → save_listings(..., city='Каменск-Уральский')."""
+    capture: dict[str, Any] = {}
+    await _drive_yandex_city(city_slug="kamensk_uralskiy", capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args_list[-1].kwargs["city"] == "Каменск-Уральский"
+
+
+@pytest.mark.asyncio
+async def test_yandex_city_sweep_stamps_ekaterinburg_when_no_city_slug() -> None:
+    """city_slug=None (ЕКБ-развёртка) → save_listings(..., city='Екатеринбург')."""
+    capture: dict[str, Any] = {}
+    await _drive_yandex_city(city_slug=None, capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args_list[-1].kwargs["city"] == "Екатеринбург"
+
+
+@pytest.mark.asyncio
+async def test_cian_city_sweep_stamps_city_from_slug() -> None:
+    """city_slug='pervouralsk' → save_listings(..., city='Первоуральск')."""
+    capture: dict[str, Any] = {}
+    await _drive_cian_city(city_slug="pervouralsk", capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args.kwargs["city"] == "Первоуральск"
+
+
+@pytest.mark.asyncio
+async def test_cian_city_sweep_stamps_ekaterinburg_when_no_city_slug() -> None:
+    """city_slug=None (ЕКБ-развёртка) → save_listings(..., city='Екатеринбург')."""
+    capture: dict[str, Any] = {}
+    await _drive_cian_city(city_slug=None, capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args.kwargs["city"] == "Екатеринбург"
+
+
+@pytest.mark.asyncio
+async def test_domclick_city_sweep_stamps_ekaterinburg_for_default_city_id() -> None:
+    """city_id=DOMCLICK_DEFAULT_CITY_ID (4, ЕКБ) → save_listings(..., city='Екатеринбург')."""
+    capture: dict[str, Any] = {}
+    await _drive_domclick(lots_n=4, blocked=False, capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args.kwargs["city"] == "Екатеринбург"
+
+
+@pytest.mark.asyncio
+async def test_avito_newbuilding_sweep_stamps_ekaterinburg() -> None:
+    """Citywide novostroyka-обход — только ЕКБ → save_listings(..., city='Екатеринбург')."""
+    capture: dict[str, Any] = {}
+    await _drive_nb_sweep(capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_args.kwargs["city"] == "Екатеринбург"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("source", ["avito", "cian", "yandex"])
+async def test_full_load_stamps_ekaterinburg(source: str) -> None:
+    """Exhaustive региональный сбор — только ЕКБ вторичка → city='Екатеринбург' на КАЖДОМ
+    бакете (on_bucket сохраняет инкрементально, не один batch на весь run)."""
+    capture: dict[str, Any] = {}
+    await _drive_full_load(source=source, capture=capture)
+    save_mock = capture["save_mock"]
+    assert save_mock.call_count > 0
+    for call in save_mock.call_args_list:
+        assert call.kwargs["city"] == "Екатеринбург"

tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py

@@ -303,6 +303,7 @@ def save_listings(
     region_code: int,
     run_id: int | None = None,
     skip_seen_today: bool = False,
+    city: str | None = None,
 ) -> tuple[int, int]:
     """Пишем list[ScrapedLot] в `listings` с upsert по dedup_hash.
 
@@ -324,6 +325,14 @@ def save_listings(
             Используется full_load для экономии redundant upsert + price-trigger churn
             при повторном прогоне в тот же день. Новые листинги (prior_row=None) всегда
             вставляются. False = старое поведение (всегда upsert).
+        city: человекочитаемое имя города-цели ЭТОГО batch'а (#2594), например
+            "Нижний Тагил"/"Екатеринбург". Развёртка знает город из своего контекста
+            (city_slug) — ОДИН на весь вызов save_listings (все lots одного batch'а
+            принадлежат одному city-sweep run'у), поэтому это kwarg, а НЕ поле
+            ScrapedLot. None (default) — вызывающая сторона город не знает (ad-hoc
+            admin/manual пути) — колонка остаётся NULL, backward-compatible.
+            ON CONFLICT — COALESCE (новое значение НЕ затирает уже известный город
+            NULL'ом, если какой-то caller ещё не передаёт city).
 
     Returns:
         (inserted, updated) — counters для логов.
@@ -380,6 +389,7 @@ def save_listings(
             "dedup": dedup,
             "region_code": region_code,
             "address": lot.address,
+            "city": city,
             "lat": lot.lat,
             "lon": lot.lon,
             "rooms": lot.rooms,
@@ -439,7 +449,7 @@ def save_listings(
             """
             INSERT INTO listings (
                 source, source_url, source_id, dedup_hash,
-                address, lat, lon, region_code,
+                address, city, lat, lon, region_code,
                 rooms, area_m2, floor, total_floors, year_built,
                 house_type, repair_state, has_balcony,
                 kitchen_area_m2, ceiling_height, ceiling_height_m,
@@ -460,7 +470,7 @@ def save_listings(
                 scraped_at, last_seen_at
             ) VALUES (
                 :source, :source_url, :source_id, :dedup,
-                :address, :lat, :lon, :region_code,
+                :address, :city, :lat, :lon, :region_code,
                 :rooms, :area_m2, :floor, :total_floors, :year_built,
                 :house_type, :repair_state, :has_balcony,
                 -- ceiling: один param :ceiling_height_m пишем в ОБЕ колонки —
@@ -516,6 +526,9 @@ def save_listings(
                     metro_stations = EXCLUDED.metro_stations,
                     listing_date = COALESCE(EXCLUDED.listing_date, listings.listing_date),
                     area_m2 = COALESCE(EXCLUDED.area_m2, listings.area_m2),
+                    -- #2594: город развёртки — COALESCE, чтобы caller без city (ad-hoc
+                    -- admin/manual пути, city=None) не затирал уже известный город.
+                    city = COALESCE(EXCLUDED.city, listings.city),
                     -- kitchen/ceiling (#2007): COALESCE — SERP re-scrape источника без
                     -- этих полей (avito SERP → NULL) НЕ затирает detail-enriched значение
                     -- (avito_detail пишет ceiling_height_m отдельным UPDATE).
@@ -627,6 +640,7 @@ def save_listings(
                                metro_stations = CAST(:metro_stations AS jsonb),
                                listing_date = COALESCE(:listing_date, listing_date),
                                area_m2 = COALESCE(:area_m2, area_m2),
+                               city = COALESCE(:city, city),
                                kitchen_area_m2 = COALESCE(:kitchen_area_m2, kitchen_area_m2),
                                ceiling_height = COALESCE(:ceiling_height_m, ceiling_height),
                                ceiling_height_m = COALESCE(:ceiling_height_m, ceiling_height_m),

tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py

@@ -343,6 +343,41 @@ def get_city_location(city_slug: str | None) -> CityLocation | None:
     return CITY_LOCATIONS.get(city_slug)
 
 
+# Человекочитаемые названия городов области — пишутся в `listings.city` (#2594).
+# Развёртка ЗНАЕТ город из своего контекста (city_slug), но раньше нигде его не
+# записывала — листинг терял привязку к городу и адрес без города в тексте
+# ("ул. Победы, 30" — так отдают и Avito, и Cian, см. providers/cian/serp.py
+# `_format_address` skip_types={"location",...}) при геокодинге считался «город не
+# назван» и коллизировал с одноимённой ЕКБ-улицей. Ключи СОВПАДАЮТ с CITY_LOCATIONS/
+# CITY_ANCHORS; значения — те же формы, что уже есть в geocoder.SVERDLOVSK_OBLAST_CITIES
+# (lower + word-boundary матчинг там регистронезависим, поэтому регистр здесь не
+# критичен, но человекочитаемый — для админки/логов/дальнейшего QA).
+CITY_DISPLAY_NAMES: dict[str, str] = {
+    "nizhniy_tagil": "Нижний Тагил",
+    "kamensk_uralskiy": "Каменск-Уральский",
+    "pervouralsk": "Первоуральск",
+    "verkhnyaya_pyshma": "Верхняя Пышма",
+    "serov": "Серов",
+}
+EKATERINBURG_CITY_NAME = "Екатеринбург"
+
+
+def resolve_city_name(city_slug: str | None) -> str:
+    """Человекочитаемое имя города для `save_listings(..., city=...)` (#2594).
+
+    city_slug=None → Екатеринбург. Это НЕ заглушка «не знаем» — это симметрия с
+    get_city_location/get_city_anchors (тот же None-путь = ЕКБ-дефолт): EKB-варианты
+    city-sweep функций (run_avito_city_sweep и т.д., вызванные БЕЗ city_slug) реально
+    собирают ЕКБ, поэтому их листинги тоже должны получать city="Екатеринбург" —
+    иначе была бы обратная асимметрия «у области город проставлен, у ЕКБ — нет».
+    Неизвестный slug (не в CITY_DISPLAY_NAMES) — тоже ЕКБ-дефолт, тем же путём, что и
+    get_city_location/get_city_anchors для неизвестных slug'ов.
+    """
+    if city_slug is None:
+        return EKATERINBURG_CITY_NAME
+    return CITY_DISPLAY_NAMES.get(city_slug, EKATERINBURG_CITY_NAME)
+
+
 _CHROME_HEADERS = {
     "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
     "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
@@ -896,6 +931,9 @@ async def run_avito_city_sweep(
     # kamensk-uralskiy (дефис) / verhnyaya_pyshma (kh→h) отличаются от нашего city_slug —
     # вычисляем один раз до цикла anchor'ов, не внутри closure на каждый anchor.
     _avito_slug = _loc.avito_slug if _loc else city_slug
+    # #2594: город для save_listings(..., city=...) — один на весь sweep (все anchor'ы
+    # одного run'а бьют по одному city_slug), вычисляем один раз до цикла.
+    _city_name = resolve_city_name(city_slug)
     counters = CitySweepCounters(anchors_total=len(_anchors))
     all_touched_house_ids: set[int] = set()
 
@@ -1035,7 +1073,11 @@ async def run_avito_city_sweep(
                     if anchor_lots:
                         try:
                             ins, upd = save_listings(
-                                db, anchor_lots, matcher=matcher, region_code=region_code
+                                db,
+                                anchor_lots,
+                                matcher=matcher,
+                                region_code=region_code,
+                                city=_city_name,
                             )
                             counters.lots_inserted += ins
                             counters.lots_updated += upd
@@ -1662,7 +1704,14 @@ async def run_avito_newbuilding_sweep(
             counters.lots_fetched += len(lots)
             if lots:
                 try:
-                    ins, upd = save_listings(db, lots, matcher=matcher, region_code=region_code)
+                    # #2594: citywide novostroyka-обход — только ЕКБ (см. docstring).
+                    ins, upd = save_listings(
+                        db,
+                        lots,
+                        matcher=matcher,
+                        region_code=region_code,
+                        city=EKATERINBURG_CITY_NAME,
+                    )
                     counters.lots_inserted += ins
                     counters.lots_updated += upd
                 except Exception as save_exc:
@@ -1782,6 +1831,8 @@ async def run_yandex_city_sweep(
     # city_slug (#12): rgid города-цели → YandexRealtyScraper.city_rgid скоупит SERP
     # на город вместо дефолтного ЕКБ. None/неизвестный slug → ЕКБ-дефолт в конструкторе.
     _loc = get_city_location(city_slug)
+    # #2594: город для save_listings(..., city=...) — один на весь sweep.
+    _city_name = resolve_city_name(city_slug)
 
     _rooms_list = rooms_list or list(ROOM_PATH.keys())
     _price_ranges = price_ranges or DEFAULT_PRICE_RANGES
@@ -1880,7 +1931,12 @@ async def run_yandex_city_sweep(
                     counters.lots_fetched += len(new_lots)
                     try:
                         ins, upd = save_listings(
-                            db, new_lots, matcher=matcher, region_code=region_code, run_id=run_id
+                            db,
+                            new_lots,
+                            matcher=matcher,
+                            region_code=region_code,
+                            run_id=run_id,
+                            city=_city_name,
                         )
                         counters.lots_inserted += ins
                         counters.lots_updated += upd
@@ -2288,6 +2344,8 @@ async def run_cian_city_sweep(
     # city_slug (#12): region_id города-цели → CianScraper.city_region_id скоупит SERP
     # на город вместо дефолтного ЕКБ. None/неизвестный slug → ЕКБ-дефолт в конструкторе.
     _loc = get_city_location(city_slug)
+    # #2594: город для save_listings(..., city=...) — один на весь sweep.
+    _city_name = resolve_city_name(city_slug)
     counters = CianCitySweepCounters(anchors_total=len(_anchors))
     consecutive_failures = 0
     cian_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep
@@ -2383,7 +2441,12 @@ async def run_cian_city_sweep(
                     counters.lots_dropped_secondary += _before - len(anchor_lots)
                 if anchor_lots:
                     inserted, updated = save_listings(
-                        db, anchor_lots, matcher=matcher, region_code=region_code, run_id=run_id
+                        db,
+                        anchor_lots,
+                        matcher=matcher,
+                        region_code=region_code,
+                        run_id=run_id,
+                        city=_city_name,
                     )
                     counters.lots_inserted += inserted
                     counters.lots_updated += updated
@@ -2785,6 +2848,8 @@ async def run_cian_full_load(
             region_code=region_code,
             run_id=run_id,
             skip_seen_today=config.scraper_skip_seen_today,
+            # #2594: exhaustive региональный сбор — только ЕКБ (см. docstring run_*_full_load).
+            city=EKATERINBURG_CITY_NAME,
         )
         # save_listings вызывает db.commit() внутри — данные в БД сразу
         counters.saved_inserted += inserted
@@ -3084,6 +3149,8 @@ async def run_yandex_full_load(
             region_code=region_code,
             run_id=run_id,
             skip_seen_today=config.scraper_skip_seen_today,
+            # #2594: exhaustive региональный сбор — только ЕКБ (см. docstring run_*_full_load).
+            city=EKATERINBURG_CITY_NAME,
         )
         # save_listings вызывает db.commit() внутри — данные в БД сразу
         counters.saved_inserted += inserted
@@ -3289,6 +3356,8 @@ async def run_avito_full_load(
             region_code=region_code,
             run_id=run_id,
             skip_seen_today=config.scraper_skip_seen_today,
+            # #2594: exhaustive региональный сбор — только ЕКБ (см. docstring run_*_full_load).
+            city=EKATERINBURG_CITY_NAME,
         )
         # save_listings вызывает db.commit() внутри — данные в БД сразу
         counters.saved_inserted += inserted
@@ -3499,8 +3568,18 @@ async def run_domclick_city_sweep(
                 lots = await _scraper.fetch_city(city_id=city_id, rooms=rooms, pages=pages)
             counters.lots_fetched += len(lots)
             if lots:
+                # #2594: domclick oblast-rollout (B2) ещё не wired (нет city_id→slug
+                # мэппинга, см. CITY_LOCATIONS) — известный ЕКБ city_id получает
+                # "Екатеринбург", любой другой (будущий B2) честно остаётся None, а не
+                # угадывается.
+                _dc_city = EKATERINBURG_CITY_NAME if city_id == DOMCLICK_DEFAULT_CITY_ID else None
                 inserted, updated = save_listings(
-                    db, lots, matcher=matcher, region_code=region_code, run_id=run_id
+                    db,
+                    lots,
+                    matcher=matcher,
+                    region_code=region_code,
+                    run_id=run_id,
+                    city=_dc_city,
                 )
                 counters.lots_inserted += inserted
                 counters.lots_updated += updated