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

@@ -1207,9 +1207,8 @@ async def run_cian_city_sweep(
                 from sqlalchemy import text as _text2
 
                 # Собираем cian_internal_house_ids из SERP-объявлений этого anchor'а.
-                # lot.raw_payload["newbuilding_id"] → SERP предоставляет house_ext_id.
-                # Используем houses.cian_zhk_url для тех домов, у которых она заполнена.
-                # Fallback: дома у которых cian_internal_house_id совпадает с house_ext_id.
+                # lot.house_ext_id → SERP предоставляет house_ext_id (newbuilding_id).
+                # Резолвим house_id через house_sources (ext_source, ext_id) → houses.id.
                 cian_nb_ext_ids = {
                     lot.house_ext_id
                     for lot in anchor_lots
@@ -1217,20 +1216,37 @@ async def run_cian_city_sweep(
                 }
                 if cian_nb_ext_ids:
                     nb_id_list = list(cian_nb_ext_ids)
-                    house_rows = (
-                        db.execute(
-                            _text2("""
-                                SELECT h.id, h.cian_zhk_url
-                                FROM houses h
-                                WHERE h.ext_source = 'cian_newbuilding'
-                                  AND h.ext_id = ANY(CAST(:ids AS text[]))
-                                  AND h.cian_zhk_url IS NOT NULL
-                            """),
-                            {"ids": nb_id_list},
+                    try:
+                        house_rows = (
+                            db.execute(
+                                _text2("""
+                                    SELECT h.id, h.cian_zhk_url
+                                    FROM houses h
+                                    JOIN house_sources hs ON hs.house_id = h.id
+                                    WHERE hs.ext_source = 'cian_newbuilding'
+                                      AND hs.ext_id = ANY(CAST(:ids AS text[]))
+                                      AND h.cian_zhk_url IS NOT NULL
+                                """),
+                                {"ids": nb_id_list},
+                            )
+                            .mappings()
+                            .all()
                         )
-                        .mappings()
-                        .all()
-                    )
+                    except Exception as exc:
+                        logger.exception(
+                            "cian-sweep run_id=%d anchor %s: houses DB query failed — "
+                            "skipping houses phase (%d ids): %s",
+                            run_id,
+                            name,
+                            len(nb_id_list),
+                            exc,
+                        )
+                        counters.houses_failed += len(nb_id_list)
+                        try:
+                            db.rollback()
+                        except Exception:
+                            pass
+                        house_rows = []
                     for hidx, hrow in enumerate(house_rows):
                         house_id_val: int = hrow["id"]
                         zhk_url: str = hrow["cian_zhk_url"]

tradein-mvp/backend/tests/test_cian_city_sweep.py

@@ -462,6 +462,122 @@ async def test_mark_done_always_called(
     assert counters.anchors_done == len(TEST_ANCHORS)
 
 
+async def test_houses_phase_uses_house_sources_join(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Houses-фаза формирует JOIN на house_sources, не h.ext_source напрямую."""
+
+    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        # Один lot с house_source="cian_newbuilding" → попадёт в houses-фазу
+        return [_fake_lot("nb-1", nb=True)]
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        AsyncMock(return_value=None),
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
+        AsyncMock(return_value=None),
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_newbuilding.save_newbuilding_enrichment",
+        MagicMock(),
+    )
+
+    mock_db = MagicMock()
+    # DB execute → возвращает одну house-строку
+    mock_db.execute.return_value.mappings.return_value.all.return_value = [
+        {"id": 501, "cian_zhk_url": "https://ekb.cian.ru/kupit-kvartiru/zhk-test/"},
+    ]
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=mock_db,
+        run_id=10,
+        anchors=TEST_ANCHORS[:1],
+        detail_top_n=0,
+        enrich_houses=True,
+        request_delay_sec=0.0,
+    )
+
+    # Проверяем SQL — должен содержать JOIN house_sources, а не h.ext_source
+    assert mock_db.execute.called
+    executed_sql_args = [str(c.args[0]) for c in mock_db.execute.call_args_list]
+    houses_sql_calls = [s for s in executed_sql_args if "house" in s.lower()]
+    assert houses_sql_calls, "должен быть хотя бы один SQL-вызов связанный с houses"
+    houses_query = houses_sql_calls[-1]
+    assert (
+        "house_sources" in houses_query
+    ), f"запрос должен JOIN house_sources, а не обращаться к h.ext_source: {houses_query!r}"
+    assert "ext_source" not in houses_query.split("house_sources")[0] or True  # tolerate join expr
+    # Главное: нет обращения к h.ext_source (старый баг)
+    assert (
+        "h.ext_source" not in houses_query
+    ), f"колонка h.ext_source не существует в houses: {houses_query!r}"
+    assert (
+        "h.ext_id" not in houses_query
+    ), f"колонка h.ext_id не существует в houses: {houses_query!r}"
+
+    assert fake.done is not None, "mark_done должен быть вызван"
+    assert fake.failed is None, "mark_failed НЕ должен быть вызван"
+    assert counters.houses_attempted == 1
+
+
+async def test_houses_db_query_error_graceful_mark_done(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Ошибка SQL в houses-фазе (напр. UndefinedColumn) → graceful: mark_done,
+    НЕ mark_failed, houses_failed увеличивается."""
+
+    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        # Два nb-лота → houses-фаза попытается запросить 2 id
+        return [_fake_lot("nb-10", nb=True), _fake_lot("nb-11", nb=True)]
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        AsyncMock(return_value=None),
+    )
+
+    mock_db = MagicMock()
+
+    def _db_execute_raises(sql, params=None):
+        sql_text = str(sql)
+        if "house" in sql_text.lower() and "cian_newbuilding" in sql_text:
+            raise RuntimeError("column h.ext_source does not exist")
+        result = MagicMock()
+        result.mappings.return_value.all.return_value = []
+        return result
+
+    mock_db.execute.side_effect = _db_execute_raises
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=mock_db,
+        run_id=11,
+        anchors=TEST_ANCHORS[:1],
+        detail_top_n=0,
+        enrich_houses=True,
+        request_delay_sec=0.0,
+    )
+
+    assert fake.done is not None, "mark_done должен быть вызван даже при ошибке houses-фазы"
+    assert fake.failed is None, "mark_failed НЕ должен быть вызван из-за ошибки houses-фазы"
+    assert counters.houses_failed > 0, "houses_failed должен вырасти при ошибке"
+    assert counters.houses_attempted == 0, "houses_attempted не растёт — до него не дошли"
+
+
 # ── Admin endpoint tests (offline) ──────────────────────────────────────────