From 197068c515689c788866dfa75b050aaabd01903e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 26 Jun 2026 19:10:12 +0300 Subject: [PATCH 1/2] =?UTF-8?q?fix(scrapers):=20save=5Flistings=20?= =?UTF-8?q?=E2=80=94=20UPDATE=20by=20(source,source=5Fid)=20=D0=BF=D1=80?= =?UTF-8?q?=D0=B8=20dedup=5Fhash-=D0=B4=D1=80=D0=B5=D0=B9=D1=84=D0=B5=20(a?= =?UTF-8?q?vito=5Ffull=5Fload=20=D0=BA=D1=80=D0=B0=D1=88)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backend/app/services/scrapers/base.py | 443 +++++++++++------- .../test_save_listings_upsert_reconcile.py | 327 +++++++++++++ 2 files changed, 601 insertions(+), 169 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py diff --git a/tradein-mvp/backend/app/services/scrapers/base.py b/tradein-mvp/backend/app/services/scrapers/base.py index cef1c2a6..4cfd5893 100644 --- a/tradein-mvp/backend/app/services/scrapers/base.py +++ b/tradein-mvp/backend/app/services/scrapers/base.py @@ -25,8 +25,10 @@ from typing import Any from zoneinfo import ZoneInfo import httpx +import psycopg.errors from pydantic import BaseModel, Field from sqlalchemy import text +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from tenacity import retry, stop_after_attempt, wait_exponential @@ -282,6 +284,7 @@ def save_listings( inserted = 0 updated = 0 skipped = 0 + reconciled = 0 # UPDATE by (source,source_id) при dedup_hash-дрейфе matched = 0 match_failures = 0 @@ -312,176 +315,277 @@ def save_listings( skipped += 1 continue - result = db.execute( - text( - """ - INSERT INTO listings ( - source, source_url, source_id, dedup_hash, - address, lat, lon, region_code, - rooms, area_m2, floor, total_floors, year_built, - house_type, repair_state, has_balcony, - house_source, house_ext_id, house_url, listing_segment, - newbuilding_id, newbuilding_url, - price_rub, price_per_m2, - listing_date, publish_date, days_on_market, description, - photo_urls, raw_payload, - living_area_m2, bedrooms_count, balconies_count, loggias_count, - description_minhash, cadastral_number, building_cadastral_number, - phones, is_homeowner, is_pro_seller, - bargain_allowed, sale_type, metro_stations, agency_name, - yandex_offer_id, predicted_price_rub, - predicted_price_min, predicted_price_max, - price_trend, price_previous_rub, - geo_precision, card_hash, - scraped_at, last_seen_at - ) VALUES ( - :source, :source_url, :source_id, :dedup, - :address, :lat, :lon, 66, - :rooms, :area_m2, :floor, :total_floors, :year_built, - :house_type, :repair_state, :has_balcony, - :house_source, :house_ext_id, :house_url, :listing_segment, - :newbuilding_id, :newbuilding_url, - :price_rub, :ppm2, - :listing_date, :publish_date, :days_on_market, :description, - CAST(:photos AS jsonb), - CAST(:raw AS jsonb), - :living_area_m2, :bedrooms_count, :balconies_count, :loggias_count, - :description_minhash, :cadastral_number, :building_cadastral_number, - CAST(:phones AS jsonb), :is_homeowner, :is_pro_seller, - :bargain_allowed, :sale_type, CAST(:metro_stations AS jsonb), :agency_name, - :yandex_offer_id, :predicted_price_rub, - :predicted_price_min, :predicted_price_max, - :price_trend, :price_previous_rub, - :geo_precision, :card_hash, - NOW(), NOW() - ) - -- Конфликт-арбитр — dedup_hash (sha256(source|source_id)). - -- Для одного (source, source_id) формула даёт ОДИН dedup_hash, - -- поэтому этот апсерт уже не вставит второй раз → согласован с - -- частичным UNIQUE (source, source_id) WHERE source_id IS NOT NULL - -- (133_listings_uq_source_source_id.sql). Менять арбитр на - -- (source, source_id) НЕЛЬЗЯ: yandex/url-only дают source_id=NULL, - -- а NULL не годится как conflict-target → дубли вернулись бы (#1773). - ON CONFLICT (dedup_hash) DO UPDATE - SET last_seen_at = NOW(), - is_active = true, - -- если цена изменилась — обновляем - price_rub = EXCLUDED.price_rub, - price_per_m2 = EXCLUDED.price_per_m2, - -- Cian-specific: обновляем при каждом re-scrape - living_area_m2 = EXCLUDED.living_area_m2, - bedrooms_count = EXCLUDED.bedrooms_count, - balconies_count = EXCLUDED.balconies_count, - loggias_count = EXCLUDED.loggias_count, - description_minhash = EXCLUDED.description_minhash, - cadastral_number = EXCLUDED.cadastral_number, - building_cadastral_number = EXCLUDED.building_cadastral_number, - phones = EXCLUDED.phones, - is_homeowner = EXCLUDED.is_homeowner, - is_pro_seller = EXCLUDED.is_pro_seller, - bargain_allowed = EXCLUDED.bargain_allowed, - sale_type = EXCLUDED.sale_type, - metro_stations = EXCLUDED.metro_stations, - listing_date = COALESCE(EXCLUDED.listing_date, listings.listing_date), - area_m2 = COALESCE(EXCLUDED.area_m2, listings.area_m2), - -- Yandex rich fields (+ shared description/agency/publish_date): - -- COALESCE so a source that does not provide them never wipes - -- a value previously written by another source / scrape. - publish_date = COALESCE(EXCLUDED.publish_date, listings.publish_date), - days_on_market = COALESCE( - EXCLUDED.days_on_market, listings.days_on_market - ), - description = COALESCE(EXCLUDED.description, listings.description), - agency_name = COALESCE(EXCLUDED.agency_name, listings.agency_name), - yandex_offer_id = COALESCE( - EXCLUDED.yandex_offer_id, listings.yandex_offer_id - ), - predicted_price_rub = COALESCE( - EXCLUDED.predicted_price_rub, listings.predicted_price_rub - ), - predicted_price_min = COALESCE( - EXCLUDED.predicted_price_min, listings.predicted_price_min - ), - predicted_price_max = COALESCE( - EXCLUDED.predicted_price_max, listings.predicted_price_max - ), - price_trend = COALESCE(EXCLUDED.price_trend, listings.price_trend), - price_previous_rub = COALESCE( - EXCLUDED.price_previous_rub, listings.price_previous_rub - ), - newbuilding_id = COALESCE( - EXCLUDED.newbuilding_id, listings.newbuilding_id - ), - newbuilding_url = COALESCE( - EXCLUDED.newbuilding_url, listings.newbuilding_url - ), - card_hash = EXCLUDED.card_hash - RETURNING id, (xmax = 0) AS inserted - """ - ), - { - "source": lot.source, - "source_url": lot.source_url, - "source_id": lot.source_id, - "dedup": dedup, - "address": lot.address, - "lat": lot.lat, - "lon": lot.lon, - "rooms": lot.rooms, - "area_m2": lot.area_m2, - "floor": lot.floor, - "total_floors": lot.total_floors, - "year_built": lot.year_built, - "house_type": lot.house_type, - "repair_state": lot.repair_state, - "has_balcony": lot.has_balcony, - "house_source": lot.house_source, - "house_ext_id": lot.house_ext_id, - "house_url": lot.house_url, - "listing_segment": lot.listing_segment, - "newbuilding_id": lot.newbuilding_id, - "newbuilding_url": lot.newbuilding_url, - "price_rub": lot.price_rub, - "ppm2": ppm2, - "listing_date": lot.listing_date, - "publish_date": lot.publish_date, - "days_on_market": lot.days_on_market, - "description": lot.description, - "photos": _to_json(lot.photo_urls), - "raw": _to_json(lot.raw_payload) if lot.raw_payload else None, - # Cian-specific columns — None for non-Cian scrapers (→ SQL NULL) - "living_area_m2": lot.living_area_m2, - "bedrooms_count": lot.bedrooms_count, - "balconies_count": lot.balconies_count, - "loggias_count": lot.loggias_count, - "description_minhash": lot.description_minhash, - "cadastral_number": lot.cadastral_number, - "building_cadastral_number": lot.building_cadastral_number, - "phones": _to_json(lot.phones) if lot.phones else None, - "is_homeowner": lot.is_homeowner, - "is_pro_seller": lot.is_pro_seller, - "bargain_allowed": lot.bargain_allowed, - "sale_type": lot.sale_type, - "metro_stations": _to_json(lot.metro_stations) if lot.metro_stations else None, - "agency_name": lot.agency_name, - "yandex_offer_id": lot.yandex_offer_id, - "predicted_price_rub": lot.predicted_price_rub, - "predicted_price_min": lot.predicted_price_min, - "predicted_price_max": lot.predicted_price_max, - "price_trend": lot.price_trend, - "price_previous_rub": lot.price_previous_rub, - "geo_precision": lot.geo_precision, - "card_hash": card_hash, - }, - ).fetchone() + # ── Per-lot SAVEPOINT: изолируем listings-upsert ──────────────────── + # Когда контент существующего ad дрейфит → новый dedup_hash → INSERT + # находит prior_row=None (SELECT по старому dedup_hash пуст) → INSERT + # летит в UniqueViolation по (source, source_id) при source_id IS NOT NULL + # (constraint 133_listings_uq_source_source_id). Без SAVEPOINT это роняет + # весь прогон. С SAVEPOINT ловим IntegrityError, rollback SAVEPOINT'а, + # и делаем прямой UPDATE по (source, source_id) — reconcile drifted row. + # source_id IS NULL (yandex) → constraint не фаерит, обычный путь. + _upsert_params = { + "source": lot.source, + "source_url": lot.source_url, + "source_id": lot.source_id, + "dedup": dedup, + "address": lot.address, + "lat": lot.lat, + "lon": lot.lon, + "rooms": lot.rooms, + "area_m2": lot.area_m2, + "floor": lot.floor, + "total_floors": lot.total_floors, + "year_built": lot.year_built, + "house_type": lot.house_type, + "repair_state": lot.repair_state, + "has_balcony": lot.has_balcony, + "house_source": lot.house_source, + "house_ext_id": lot.house_ext_id, + "house_url": lot.house_url, + "listing_segment": lot.listing_segment, + "newbuilding_id": lot.newbuilding_id, + "newbuilding_url": lot.newbuilding_url, + "price_rub": lot.price_rub, + "ppm2": ppm2, + "listing_date": lot.listing_date, + "publish_date": lot.publish_date, + "days_on_market": lot.days_on_market, + "description": lot.description, + "photos": _to_json(lot.photo_urls), + "raw": _to_json(lot.raw_payload) if lot.raw_payload else None, + # Cian-specific columns — None for non-Cian scrapers (→ SQL NULL) + "living_area_m2": lot.living_area_m2, + "bedrooms_count": lot.bedrooms_count, + "balconies_count": lot.balconies_count, + "loggias_count": lot.loggias_count, + "description_minhash": lot.description_minhash, + "cadastral_number": lot.cadastral_number, + "building_cadastral_number": lot.building_cadastral_number, + "phones": _to_json(lot.phones) if lot.phones else None, + "is_homeowner": lot.is_homeowner, + "is_pro_seller": lot.is_pro_seller, + "bargain_allowed": lot.bargain_allowed, + "sale_type": lot.sale_type, + "metro_stations": _to_json(lot.metro_stations) if lot.metro_stations else None, + "agency_name": lot.agency_name, + "yandex_offer_id": lot.yandex_offer_id, + "predicted_price_rub": lot.predicted_price_rub, + "predicted_price_min": lot.predicted_price_min, + "predicted_price_max": lot.predicted_price_max, + "price_trend": lot.price_trend, + "price_previous_rub": lot.price_previous_rub, + "geo_precision": lot.geo_precision, + "card_hash": card_hash, + } + _upsert_sql = text( + """ + INSERT INTO listings ( + source, source_url, source_id, dedup_hash, + address, lat, lon, region_code, + rooms, area_m2, floor, total_floors, year_built, + house_type, repair_state, has_balcony, + house_source, house_ext_id, house_url, listing_segment, + newbuilding_id, newbuilding_url, + price_rub, price_per_m2, + listing_date, publish_date, days_on_market, description, + photo_urls, raw_payload, + living_area_m2, bedrooms_count, balconies_count, loggias_count, + description_minhash, cadastral_number, building_cadastral_number, + phones, is_homeowner, is_pro_seller, + bargain_allowed, sale_type, metro_stations, agency_name, + yandex_offer_id, predicted_price_rub, + predicted_price_min, predicted_price_max, + price_trend, price_previous_rub, + geo_precision, card_hash, + scraped_at, last_seen_at + ) VALUES ( + :source, :source_url, :source_id, :dedup, + :address, :lat, :lon, 66, + :rooms, :area_m2, :floor, :total_floors, :year_built, + :house_type, :repair_state, :has_balcony, + :house_source, :house_ext_id, :house_url, :listing_segment, + :newbuilding_id, :newbuilding_url, + :price_rub, :ppm2, + :listing_date, :publish_date, :days_on_market, :description, + CAST(:photos AS jsonb), + CAST(:raw AS jsonb), + :living_area_m2, :bedrooms_count, :balconies_count, :loggias_count, + :description_minhash, :cadastral_number, :building_cadastral_number, + CAST(:phones AS jsonb), :is_homeowner, :is_pro_seller, + :bargain_allowed, :sale_type, CAST(:metro_stations AS jsonb), :agency_name, + :yandex_offer_id, :predicted_price_rub, + :predicted_price_min, :predicted_price_max, + :price_trend, :price_previous_rub, + :geo_precision, :card_hash, + NOW(), NOW() + ) + -- Конфликт-арбитр — dedup_hash (sha256(source|source_id)). + -- Для одного (source, source_id) формула даёт ОДИН dedup_hash, + -- поэтому этот апсерт уже не вставит второй раз → согласован с + -- частичным UNIQUE (source, source_id) WHERE source_id IS NOT NULL + -- (133_listings_uq_source_source_id.sql). Менять арбитр на + -- (source, source_id) НЕЛЬЗЯ: yandex/url-only дают source_id=NULL, + -- а NULL не годится как conflict-target → дубли вернулись бы (#1773). + ON CONFLICT (dedup_hash) DO UPDATE + SET last_seen_at = NOW(), + is_active = true, + -- если цена изменилась — обновляем + price_rub = EXCLUDED.price_rub, + price_per_m2 = EXCLUDED.price_per_m2, + -- Cian-specific: обновляем при каждом re-scrape + living_area_m2 = EXCLUDED.living_area_m2, + bedrooms_count = EXCLUDED.bedrooms_count, + balconies_count = EXCLUDED.balconies_count, + loggias_count = EXCLUDED.loggias_count, + description_minhash = EXCLUDED.description_minhash, + cadastral_number = EXCLUDED.cadastral_number, + building_cadastral_number = EXCLUDED.building_cadastral_number, + phones = EXCLUDED.phones, + is_homeowner = EXCLUDED.is_homeowner, + is_pro_seller = EXCLUDED.is_pro_seller, + bargain_allowed = EXCLUDED.bargain_allowed, + sale_type = EXCLUDED.sale_type, + metro_stations = EXCLUDED.metro_stations, + listing_date = COALESCE(EXCLUDED.listing_date, listings.listing_date), + area_m2 = COALESCE(EXCLUDED.area_m2, listings.area_m2), + -- Yandex rich fields (+ shared description/agency/publish_date): + -- COALESCE so a source that does not provide them never wipes + -- a value previously written by another source / scrape. + publish_date = COALESCE(EXCLUDED.publish_date, listings.publish_date), + days_on_market = COALESCE( + EXCLUDED.days_on_market, listings.days_on_market + ), + description = COALESCE(EXCLUDED.description, listings.description), + agency_name = COALESCE(EXCLUDED.agency_name, listings.agency_name), + yandex_offer_id = COALESCE( + EXCLUDED.yandex_offer_id, listings.yandex_offer_id + ), + predicted_price_rub = COALESCE( + EXCLUDED.predicted_price_rub, listings.predicted_price_rub + ), + predicted_price_min = COALESCE( + EXCLUDED.predicted_price_min, listings.predicted_price_min + ), + predicted_price_max = COALESCE( + EXCLUDED.predicted_price_max, listings.predicted_price_max + ), + price_trend = COALESCE(EXCLUDED.price_trend, listings.price_trend), + price_previous_rub = COALESCE( + EXCLUDED.price_previous_rub, listings.price_previous_rub + ), + newbuilding_id = COALESCE( + EXCLUDED.newbuilding_id, listings.newbuilding_id + ), + newbuilding_url = COALESCE( + EXCLUDED.newbuilding_url, listings.newbuilding_url + ), + card_hash = EXCLUDED.card_hash + RETURNING id, (xmax = 0) AS inserted + """ + ) + + # ── Per-lot SAVEPOINT вокруг listings-upsert ──────────────────────── + # Когда контент существующего ad дрейфит → новый dedup_hash → INSERT + # не находит existing row по dedup_hash → пробует fresh INSERT → летит + # UniqueViolation по (source, source_id) WHERE source_id IS NOT NULL + # (constraint listings_source_source_id_uq). Без SAVEPOINT это переводит + # весь Session в InFailedSqlTransaction → весь прогон падает. + # С SAVEPOINT ловим IntegrityError/UniqueViolation, делаем rollback, + # и применяем UPDATE by (source, source_id) — reconcile drifted row. + # source_id IS NULL (yandex) → constraint не фаерит, обычный upsert путь. + result = None listing_id: int | None = None - if result is not None: - listing_id = int(result.id) - if result.inserted: - inserted += 1 + try: + with db.begin_nested(): + result = db.execute(_upsert_sql, _upsert_params).fetchone() + except IntegrityError as ie: + # Проверяем что это UniqueViolation именно по (source, source_id). + if lot.source_id is not None and isinstance(ie.orig, psycopg.errors.UniqueViolation): + logger.warning( + "save_listings:dedup_hash_drift source=%s source_id=%s " + "— reconcile UPDATE by (source, source_id)", + lot.source, + lot.source_id, + ) + # SAVEPOINT уже откачен; session чистая — применяем прямой UPDATE. + rec_row = db.execute( + text( + """ + UPDATE listings + SET dedup_hash = :dedup, + last_seen_at = NOW(), + is_active = true, + price_rub = :price_rub, + price_per_m2 = :ppm2, + living_area_m2 = :living_area_m2, + bedrooms_count = :bedrooms_count, + balconies_count = :balconies_count, + loggias_count = :loggias_count, + description_minhash = :description_minhash, + cadastral_number = :cadastral_number, + building_cadastral_number = :building_cadastral_number, + phones = CAST(:phones AS jsonb), + is_homeowner = :is_homeowner, + is_pro_seller = :is_pro_seller, + bargain_allowed = :bargain_allowed, + sale_type = :sale_type, + metro_stations = CAST(:metro_stations AS jsonb), + listing_date = COALESCE(:listing_date, listing_date), + area_m2 = COALESCE(:area_m2, area_m2), + publish_date = COALESCE(:publish_date, publish_date), + days_on_market = COALESCE(:days_on_market, days_on_market), + description = COALESCE(:description, description), + agency_name = COALESCE(:agency_name, agency_name), + yandex_offer_id = COALESCE(:yandex_offer_id, yandex_offer_id), + predicted_price_rub = COALESCE( + :predicted_price_rub, predicted_price_rub + ), + predicted_price_min = COALESCE( + :predicted_price_min, predicted_price_min + ), + predicted_price_max = COALESCE( + :predicted_price_max, predicted_price_max + ), + price_trend = COALESCE(:price_trend, price_trend), + price_previous_rub = COALESCE( + :price_previous_rub, price_previous_rub + ), + newbuilding_id = COALESCE(:newbuilding_id, newbuilding_id), + newbuilding_url = COALESCE(:newbuilding_url, newbuilding_url), + card_hash = :card_hash + WHERE source = :source + AND source_id = :source_id + RETURNING id + """ + ), + _upsert_params, + ).fetchone() + if rec_row is not None: + listing_id = int(rec_row.id) + reconciled += 1 + else: + logger.error( + "save_listings:reconcile_failed source=%s source_id=%s " + "— UPDATE returned no row (unexpected)", + lot.source, + lot.source_id, + ) else: - updated += 1 + logger.error( + "save_listings:integrity_error source=%s source_id=%s: %s", + lot.source, + lot.source_id, + ie, + ) + raise + else: + if result is not None: + listing_id = int(result.id) + if result.inserted: + inserted += 1 + else: + updated += 1 # ── Snapshot: point-in-time observation in listings_snapshots ─── # Fault-tolerant: failure here MUST NOT abort the listings batch. @@ -554,11 +658,12 @@ def save_listings( db.commit() logger.info( - "save_listings: source=%s inserted=%d updated=%d skipped_seen_today=%d " - "matched=%d match_failures=%d (total %d)", + "save_listings: source=%s inserted=%d updated=%d reconciled=%d " + "skipped_seen_today=%d matched=%d match_failures=%d (total %d)", lots[0].source if lots else "?", inserted, updated, + reconciled, skipped, matched, match_failures, diff --git a/tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py b/tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py new file mode 100644 index 00000000..ddcb4385 --- /dev/null +++ b/tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py @@ -0,0 +1,327 @@ +"""Тесты Fix 1: save_listings — SAVEPOINT вокруг INSERT + reconcile по (source,source_id). + +Корень бага: при dedup_hash-дрейфе (контент ad изменился, новый source_id не меняется) +INSERT не находит existing row по dedup_hash → пробует свежий INSERT → UniqueViolation +по listings_source_source_id_uq (source, source_id) → весь прогон падает. + +Фикс: per-lot SAVEPOINT + IntegrityError catch + UPDATE by (source, source_id). +""" + +from __future__ import annotations + +import os +from contextlib import contextmanager +from unittest.mock import MagicMock, patch + +import psycopg.errors +import pytest +from sqlalchemy.exc import IntegrityError + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from app.services.scrapers.base import ScrapedLot, save_listings + +# ── Fixtures ───────────────────────────────────────────────────────────────── + + +@pytest.fixture(autouse=True) +def _patch_matching(): + """Stub matching + snapshot so tests focus on the upsert/reconcile logic.""" + with ( + patch("app.services.scrapers.base.match_or_create_house") as m_house, + patch("app.services.scrapers.base.upsert_listing_source") as m_link, + patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap, + ): + m_house.return_value = (101, 1.0, "new") + m_link.return_value = None + m_snap.return_value = None + yield {"house": m_house, "link": m_link, "snap": m_snap} + + +def _base_lot(**overrides) -> ScrapedLot: + defaults = { + "source": "avito", + "source_url": "https://www.avito.ru/ekaterinburg/test/1", + "source_id": "7960764619", + "price_rub": 5_000_000, + } + defaults.update(overrides) + return ScrapedLot(**defaults) + + +@contextmanager +def _nested_ctx(): + yield MagicMock() + + +def _mock_db_normal(inserted: bool = True, listing_id: int = 42) -> MagicMock: + """DB mock для нормального пути (нет UniqueViolation).""" + insert_row = MagicMock() + insert_row.id = listing_id + insert_row.inserted = inserted + + db = MagicMock() + + def _execute(sql, params=None): + s = str(sql) + res = MagicMock() + if "SELECT card_hash" in s and "WHERE dedup_hash" in s: + res.fetchone.return_value = None # нет prior row → fresh insert + 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 _mock_db_with_unique_violation(reconcile_id: int = 77) -> MagicMock: + """DB mock: первый INSERT вызывает UniqueViolation (dedup_hash drift). + + Симулирует кейс: INSERT по новому dedup_hash → коллизия по (source, source_id). + Reconcile UPDATE by (source, source_id) → возвращает id=reconcile_id. + """ + uv_orig = psycopg.errors.UniqueViolation() + integrity_err = IntegrityError( + statement="INSERT INTO listings ...", + params={}, + orig=uv_orig, + ) + + rec_row = MagicMock() + rec_row.id = reconcile_id + + db = MagicMock() + + def _execute(sql, params=None): + s = str(sql) + res = MagicMock() + if "SELECT card_hash" in s and "WHERE dedup_hash" in s: + # prior_row=None → INSERT будет сделан (no prior dedup_hash match) + res.fetchone.return_value = None + elif "FROM listings_snapshots" in s: + res.fetchone.return_value = None + elif "INSERT INTO listings (" in s: + # Имитируем исключение внутри begin_nested() context + raise integrity_err + elif "UPDATE listings" in s and "SET dedup_hash" in s: + # reconcile UPDATE by (source, source_id) + res.fetchone.return_value = rec_row + elif "UPDATE listings" in s and "house_id_fk" in s: + res.fetchone.return_value = None + else: + res.fetchone.return_value = None + return res + + db.execute.side_effect = _execute + + # begin_nested() нужен для per-lot SAVEPOINT + snapshot + matching hook + # Используем реальный contextmanager который НЕ перехватывает IntegrityError + # (savepoint rollback симулируется тем, что db.execute после rollback'а + # уже вызывает reconcile UPDATE, а не снова INSERT). + savepoint_entered = [False] + + @contextmanager + def _nested(): + savepoint_entered[0] = True + try: + yield MagicMock() + except IntegrityError: + # SAVEPOINT rollback — пробрасываем наверх (SQLAlchemy делает rollback) + raise + + db.begin_nested.side_effect = _nested + return db + + +# ── Tests ───────────────────────────────────────────────────────────────────── + + +def test_normal_insert_path_unchanged(): + """Нормальный путь (нет UniqueViolation) → inserted=1, updated=0.""" + db = _mock_db_normal(inserted=True, listing_id=10) + lot = _base_lot() + + inserted, updated = save_listings(db, [lot]) + + assert inserted == 1 + assert updated == 0 + + +def test_normal_update_path_unchanged(): + """Нормальный update path (ON CONFLICT DO UPDATE) → inserted=0, updated=1.""" + insert_row = MagicMock() + insert_row.id = 42 + insert_row.inserted = False # xmax != 0 → update + + db = MagicMock() + + def _execute(sql, params=None): + 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 + + lot = _base_lot(source_id="upd_test") + inserted, updated = save_listings(db, [lot]) + + assert inserted == 0 + assert updated == 1 + + +def test_null_source_id_no_savepoint_needed(): + """source_id=None (yandex) → обычный INSERT, UniqueViolation не может возникнуть. + + Проверяем, что нормальный путь не изменился при source_id=None. + """ + db = _mock_db_normal(inserted=True, listing_id=99) + lot = _base_lot( + source="yandex", + source_url="https://realty.yandex.ru/offer/123/", + source_id=None, + ) + + inserted, updated = save_listings(db, [lot]) + + assert inserted == 1 + assert updated == 0 + + +def test_unique_violation_triggers_reconcile_update(caplog, _patch_matching): + """UniqueViolation по (source, source_id) → SAVEPOINT rollback → UPDATE by (s,sid). + + После reconcile UPDATE: listing_id определён, matching hook вызывается. + inserted=0, updated=0 (reconciled не входит в возврат, но лог есть). + """ + import logging + + db = _mock_db_with_unique_violation(reconcile_id=77) + lot = _base_lot(source="avito", source_id="7960764619") + + with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"): + inserted, updated = save_listings(db, [lot]) + + # Основные counters: reconcile не добавляет к inserted/updated + assert inserted == 0 + assert updated == 0 + + # Лог предупреждения о reconcile + warns = [r for r in caplog.records if "dedup_hash_drift" in r.getMessage()] + assert len(warns) == 1 + assert "7960764619" in warns[0].getMessage() + + # Matching hook должен вызваться (listing_id=77 из reconcile UPDATE) + assert _patch_matching["link"].call_count == 1 + + +def test_unique_violation_reconcile_uses_update_by_source_source_id(): + """Reconcile UPDATE содержит WHERE source = :source AND source_id = :source_id.""" + db = _mock_db_with_unique_violation(reconcile_id=55) + lot = _base_lot(source="cian", source_id="330200428") + + save_listings(db, [lot]) + + # Ищем вызов reconcile UPDATE (содержит SET dedup_hash) + reconcile_calls = [ + c + for c in db.execute.call_args_list + if "UPDATE listings" in str(c.args[0]) and "SET dedup_hash" in str(c.args[0]) + ] + assert len(reconcile_calls) == 1 + _sql, params = str(reconcile_calls[0].args[0]), reconcile_calls[0].args[1] + assert "WHERE source = :source" in _sql + assert "AND source_id = :source_id" in _sql + assert params["source"] == "cian" + assert params["source_id"] == "330200428" + + +def test_reconcile_includes_new_dedup_hash(): + """Reconcile UPDATE обновляет dedup_hash на новый (дрейфнувший) хэш.""" + db = _mock_db_with_unique_violation(reconcile_id=66) + lot = _base_lot(source="avito", source_id="7960764619", price_rub=5_500_000) + + expected_dedup = lot.compute_dedup_hash() + save_listings(db, [lot]) + + reconcile_calls = [ + c + for c in db.execute.call_args_list + if "UPDATE listings" in str(c.args[0]) and "SET dedup_hash" in str(c.args[0]) + ] + assert len(reconcile_calls) == 1 + params = reconcile_calls[0].args[1] + assert params["dedup"] == expected_dedup + + +def test_reconcile_run_does_not_crash_full_batch(_patch_matching): + """Один лот с UniqueViolation в batch из 3 → остальные 2 успешно обрабатываются. + + Ключевая проверка что один битый lot НЕ роняет прогон. + """ + rec_row = MagicMock() + rec_row.id = 77 + + ok_insert_row = MagicMock() + ok_insert_row.id = 88 + ok_insert_row.inserted = True + + uv_orig = psycopg.errors.UniqueViolation() + integrity_err = IntegrityError("INSERT", {}, uv_orig) + + def _execute(sql, params=None): + 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: + if params and params.get("source_id") == "CONFLICT_ID": + raise integrity_err + res.fetchone.return_value = ok_insert_row + elif "UPDATE listings" in s and "SET dedup_hash" in s: + res.fetchone.return_value = rec_row + elif "UPDATE listings" in s and "house_id_fk" in s: + res.fetchone.return_value = None + else: + res.fetchone.return_value = None + return res + + db = MagicMock() + db.execute.side_effect = _execute + + @contextmanager + def _nested(): + try: + yield MagicMock() + except IntegrityError: + raise + + db.begin_nested.side_effect = _nested + + lot_conflict = _base_lot(source_id="CONFLICT_ID") + lot_ok1 = _base_lot(source_id="OK_1", source_url="https://www.avito.ru/ok1") + lot_ok2 = _base_lot(source_id="OK_2", source_url="https://www.avito.ru/ok2") + + inserted, updated = save_listings(db, [lot_conflict, lot_ok1, lot_ok2]) + + # 2 нормальных INSERT + 1 reconcile (не считается в inserted/updated) + assert inserted == 2 + assert updated == 0 + # Все 3 лота дошли до matching hook + assert _patch_matching["link"].call_count == 3 -- 2.45.3 From c1cdf3969c51d320b688e13937a1698ea09631cf Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 26 Jun 2026 19:10:26 +0300 Subject: [PATCH 2/2] =?UTF-8?q?fix(scrapers):=20ban/stuck-=D1=80=D0=BE?= =?UTF-8?q?=D1=82=D0=B0=D1=86=D0=B8=D1=8F=20=D0=B4=D0=BB=D1=8F=20cian+yand?= =?UTF-8?q?ex=20(provider-aware=20=5Frotate=5Fproxy=5Fip)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/core/config.py | 6 + .../backend/app/services/scrape_pipeline.py | 141 ++++++++-- .../tests/test_provider_rotation_1848.py | 242 ++++++++++++++++++ 3 files changed, 371 insertions(+), 18 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_provider_rotation_1848.py diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 9e2c7b94..50c9a881 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -359,6 +359,9 @@ class Settings(BaseSettings): # changeip-ссылка для Cian-прокси (ротация IP при бане/таймауте). Если не задан — # fallback на avito_proxy_rotate_url. ENV: CIAN_PROXY_ROTATE_URL. cian_proxy_rotate_url: str | None = None + # Максимум IP-ротаций для Cian на один sweep-прогон. Аналог avito_proxy_max_rotations. + # ENV: CIAN_PROXY_MAX_ROTATIONS. + cian_proxy_max_rotations: int = 4 @property def cian_proxy_url(self) -> str | None: @@ -373,6 +376,9 @@ class Settings(BaseSettings): # changeip-ссылка для Yandex-прокси (ротация IP при капче/таймауте). Если не задан — # fallback на avito_proxy_rotate_url. ENV: YANDEX_PROXY_ROTATE_URL. yandex_proxy_rotate_url: str | None = None + # Максимум IP-ротаций для Yandex на один sweep-прогон. Аналог avito_proxy_max_rotations. + # ENV: YANDEX_PROXY_MAX_ROTATIONS. + yandex_proxy_max_rotations: int = 4 @property def yandex_proxy_url(self) -> str | None: diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 5f75bf97..d5632d4f 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -53,17 +53,31 @@ from app.services.yandex_price_history import record_yandex_price_history logger = logging.getLogger(__name__) -async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool: - """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790). +async def _rotate_proxy_ip(*, reason: str, rotations_done: int, source: str = "avito") -> bool: + """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790/#1848). + + Provider-aware: выбирает rotate_url и max_rotations по параметру `source` + (avito / cian / yandex). Fallback-цепочка для rotate_url: + avito: avito_proxy_rotate_url + cian: cian_proxy_rotate_url → avito_proxy_rotate_url + yandex: yandex_proxy_rotate_url → avito_proxy_rotate_url Аналог AvitoScraper._rotate_ip, но на уровне pipeline — используется в - enrichment-фазах (houses + detail) где нет доступа к экземпляру scraper'а. + enrichment-фазах где нет доступа к экземпляру scraper'а. Returns True при успешной смене IP, False если rotate_url не задан или ошибка. - Логирует каждую ротацию (причина, порядковый номер, остаток лимита). + Логирует каждую ротацию (причина, порядковый номер, source, остаток лимита). """ - rotate_url = settings.avito_proxy_rotate_url - max_rot = settings.avito_proxy_max_rotations + if source == "cian": + rotate_url = settings.cian_proxy_rotate_url or settings.avito_proxy_rotate_url + max_rot = settings.cian_proxy_max_rotations + elif source == "yandex": + rotate_url = settings.yandex_proxy_rotate_url or settings.avito_proxy_rotate_url + max_rot = settings.yandex_proxy_max_rotations + else: + # avito (default) — backward compat + rotate_url = settings.avito_proxy_rotate_url + max_rot = settings.avito_proxy_max_rotations if not rotate_url: return False sep = "&" if "?" in rotate_url else "?" @@ -78,7 +92,9 @@ async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool: pass await asyncio.sleep(settings.avito_proxy_rotate_settle_s) logger.info( - "pipeline: IP rotated via changeip — reason=%s rotation=#%d remaining=%d new_ip=%s", + "pipeline: IP rotated via changeip — source=%s reason=%s " + "rotation=#%d remaining=%d new_ip=%s", + source, reason, rotations_done + 1, max_rot - rotations_done - 1, @@ -87,7 +103,8 @@ async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool: return True except Exception: logger.warning( - "pipeline: IP rotation failed (reason=%s, rotation=#%d)", + "pipeline: IP rotation failed (source=%s reason=%s rotation=#%d)", + source, reason, rotations_done + 1, exc_info=True, @@ -759,9 +776,7 @@ async def run_avito_city_sweep( scraper._browser = shared_bf # Shared-browser режим: _cffi=None → curl_cffi-fallback на # firewall пропускается (by design). Логируем для observability. - logger.debug( - "avito pipeline-mode: shared browser only, no cffi fallback" - ) + logger.debug("avito pipeline-mode: shared browser only, no cffi fallback") else: scraper._cffi = session @@ -1482,6 +1497,7 @@ async def run_yandex_city_sweep( enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0 _resolved_delay = request_delay_sec if request_delay_sec is not None else 9.0 consecutive_failures = 0 + yandex_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None). # В combos-режиме один "anchor" выполняет num_segments × num_combos × max_pages @@ -1565,7 +1581,7 @@ async def run_yandex_city_sweep( каждого (room×price) combo, сохраняет порцию сразу в БД + обновляет heartbeat. anchor_lots накапливается (через _al) для address-enrich и price-history. """ - nonlocal consecutive_failures + nonlocal consecutive_failures, yandex_rotations_done # ── Phase 1+2: SERP + инкрементальный save per-combo ────── def _on_combo( @@ -1675,19 +1691,25 @@ async def run_yandex_city_sweep( if items: from curl_cffi.requests import AsyncSession as _AsyncSession + # #1848: rotation tracking для Yandex address-enrich (stuck-proxy) + _yandex_enrich_consec_fail = 0 + _yandex_enrich_abort = 3 # abort address-enrich после N подряд + async with _AsyncSession( impersonate="chrome120", timeout=30.0, proxies=_proxies, headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"}, ) as enrich_session: - for eidx, item in enumerate(items): + eidx = 0 + while eidx < len(items): + item = items[eidx] lid: int = item["id"] old_addr: str = item["address"] url: str = item["source_url"] try: resp = await enrich_session.get(url, allow_redirects=True) - if resp.status_code != 200: + if resp.status_code not in {200}: logger.warning( "yandex-sweep run_id=%d address-enrich:" " HTTP %d listing_id=%d", @@ -1696,6 +1718,11 @@ async def run_yandex_city_sweep( lid, ) counters.address_failed += 1 + # 503/429 = прокси заблокирован → считаем как failure + if resp.status_code in {429, 503}: + _yandex_enrich_consec_fail += 1 + else: + _yandex_enrich_consec_fail = 0 else: new_addr = _extract_address_from_title(resp.text) if ( @@ -1741,17 +1768,53 @@ async def run_yandex_city_sweep( db.rollback() except Exception: pass + _yandex_enrich_consec_fail = 0 except Exception as fetch_exc: counters.address_failed += 1 + _yandex_enrich_consec_fail += 1 logger.warning( "yandex-sweep run_id=%d: address-enrich " - "fetch failed listing_id=%d: %s", + "fetch failed listing_id=%d " + "(consecutive=%d): %s", run_id, lid, + _yandex_enrich_consec_fail, fetch_exc, ) + # Ротация Yandex-прокси при stuck/ban (#1848) + if _yandex_enrich_consec_fail >= _yandex_enrich_abort: + if yandex_rotations_done < settings.yandex_proxy_max_rotations: + rotated = await _rotate_proxy_ip( + reason=( + f"yandex enrich consecutive=" + f"{_yandex_enrich_consec_fail}" + ), + rotations_done=yandex_rotations_done, + source="yandex", + ) + yandex_rotations_done += 1 + if rotated: + _yandex_enrich_consec_fail = 0 + logger.info( + "yandex-sweep run_id=%d: enrich ROTATE — " + "IP changed, reset consecutive " + "(rotation #%d/%d)", + run_id, + yandex_rotations_done, + settings.yandex_proxy_max_rotations, + ) + # retry текущего item после ротации + continue + logger.error( + "yandex-sweep run_id=%d: address-enrich ABORT — " + "%d consecutive failures (proxy likely stuck/banned)", + run_id, + _yandex_enrich_consec_fail, + ) + break if eidx < len(items) - 1: await asyncio.sleep(enrich_delay) + eidx += 1 logger.info( "yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d", @@ -1954,6 +2017,7 @@ async def run_cian_city_sweep( _anchors = anchors if anchors is not None else EKB_ANCHORS counters = CianCitySweepCounters(anchors_total=len(_anchors)) consecutive_failures = 0 + cian_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): @@ -1994,7 +2058,7 @@ async def run_cian_city_sweep( _a_name: str = _c_name, ) -> None: """Все фазы одного cian anchor'а (SERP + detail + houses).""" - nonlocal anchor_lots, consecutive_failures + nonlocal anchor_lots, consecutive_failures, cian_rotations_done from sqlalchemy import text as _text # ── Phase 1+2: SERP + save ───────────────────────────────── @@ -2051,7 +2115,12 @@ async def run_cian_city_sweep( .mappings() .all() ) - for didx, row in enumerate(priority_rows): + # #1848: Cian bans/stuck-proxy tracking + rotation (зеркало avito) + _cian_detail_consec_failures = 0 + _cian_detail_abort = 3 # abort detail-фазы после N подряд + didx = 0 + while didx < len(priority_rows): + row = priority_rows[didx] listing_id: int = row["id"] source_url: str = row["source_url"] counters.detail_attempted += 1 @@ -2069,18 +2138,54 @@ async def run_cian_city_sweep( listing_id, source_url, ) + _cian_detail_consec_failures = 0 except Exception as exc: counters.detail_failed += 1 counters.errors_count += 1 + _cian_detail_consec_failures += 1 logger.warning( - "cian-sweep run_id=%d: detail failed listing_id=%d: %s", + "cian-sweep run_id=%d: detail failed listing_id=%d " + "(consecutive=%d): %s", run_id, listing_id, + _cian_detail_consec_failures, exc, ) + if _cian_detail_consec_failures >= _cian_detail_abort: + # Попытка ротации Cian-прокси перед abort'ом (#1848) + if cian_rotations_done < settings.cian_proxy_max_rotations: + rotated = await _rotate_proxy_ip( + reason=( + f"cian detail consecutive=" + f"{_cian_detail_consec_failures}" + ), + rotations_done=cian_rotations_done, + source="cian", + ) + cian_rotations_done += 1 + if rotated: + _cian_detail_consec_failures = 0 + logger.info( + "cian-sweep run_id=%d: detail ROTATE — " + "IP changed, reset consecutive " + "(rotation #%d/%d)", + run_id, + cian_rotations_done, + settings.cian_proxy_max_rotations, + ) + # retry текущего detail (не инкрементируем didx) + continue + logger.error( + "cian-sweep run_id=%d: detail ABORT — " + "%d consecutive failures (proxy likely banned/stuck)", + run_id, + _cian_detail_consec_failures, + ) + break if didx < len(priority_rows) - 1: jitter = random.uniform(0.8, 1.2) await asyncio.sleep(request_delay_sec * jitter) + didx += 1 logger.info( "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d", run_id, diff --git a/tradein-mvp/backend/tests/test_provider_rotation_1848.py b/tradein-mvp/backend/tests/test_provider_rotation_1848.py new file mode 100644 index 00000000..c9021119 --- /dev/null +++ b/tradein-mvp/backend/tests/test_provider_rotation_1848.py @@ -0,0 +1,242 @@ +"""Тесты Fix 2: provider-aware _rotate_proxy_ip (#1848). + +Тест 1: source='cian' → cian_proxy_rotate_url используется. +Тест 2: source='yandex' → yandex_proxy_rotate_url используется. +Тест 3: source='avito' (default) → avito_proxy_rotate_url (backward compat). +Тест 4: cian без cian url → fallback на avito_proxy_rotate_url. +Тест 5: rotate_url=None → returns False (no HTTP call). +Тест 6: network error → returns False (no crash). +Тест 7: yandex fallback на avito. +""" + +from __future__ import annotations + +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from app.services.scrape_pipeline import _rotate_proxy_ip + +# ── Helpers ────────────────────────────────────────────────────────────────── + + +def _async_session_ctx(resp_json: dict | None = None): + """Возвращает (async_cm_class, session_mock) для curl_cffi AsyncSession.""" + session = AsyncMock() + if resp_json is not None: + mock_resp = MagicMock() + mock_resp.json.return_value = resp_json + session.get = AsyncMock(return_value=mock_resp) + else: + session.get = AsyncMock(side_effect=RuntimeError("network error")) + + # curl_cffi AsyncSession используется как `async with AsyncSession(...) as rot:` + # Нам нужен callable класс-замена, вызов которого возвращает async context manager. + class _MockAsyncSession: + def __init__(self, timeout=30): + pass + + async def __aenter__(self): + return session + + async def __aexit__(self, *_): + pass + + return _MockAsyncSession, session + + +# ── Тест 1: cian_proxy_rotate_url ──────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_rotate_proxy_ip_cian_uses_cian_url() -> None: + """source='cian' → cian_proxy_rotate_url (не avito_proxy_rotate_url).""" + mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "1.2.3.4"}) + + with ( + patch("app.services.scrape_pipeline.settings") as s, + patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + ): + s.cian_proxy_rotate_url = "http://cian-rotate.test/" + s.avito_proxy_rotate_url = "http://avito-rotate.test/" + s.yandex_proxy_rotate_url = None + s.cian_proxy_max_rotations = 4 + s.avito_proxy_max_rotations = 4 + s.yandex_proxy_max_rotations = 4 + s.avito_proxy_rotate_settle_s = 0.0 + + result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian") + + assert result is True + called_url: str = session.get.call_args[0][0] + assert "cian-rotate.test" in called_url + assert "avito-rotate.test" not in called_url + + +# ── Тест 2: yandex_proxy_rotate_url ───────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_rotate_proxy_ip_yandex_uses_yandex_url() -> None: + """source='yandex' → yandex_proxy_rotate_url.""" + mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "5.6.7.8"}) + + with ( + patch("app.services.scrape_pipeline.settings") as s, + patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + ): + s.yandex_proxy_rotate_url = "http://yandex-rotate.test/" + s.avito_proxy_rotate_url = "http://avito-rotate.test/" + s.cian_proxy_rotate_url = None + s.yandex_proxy_max_rotations = 4 + s.avito_proxy_max_rotations = 4 + s.cian_proxy_max_rotations = 4 + s.avito_proxy_rotate_settle_s = 0.0 + + result = await _rotate_proxy_ip(reason="timeout", rotations_done=0, source="yandex") + + assert result is True + called_url: str = session.get.call_args[0][0] + assert "yandex-rotate.test" in called_url + assert "avito-rotate.test" not in called_url + + +# ── Тест 3: avito backward compat ──────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_rotate_proxy_ip_avito_backward_compat() -> None: + """source не передан (дефолт 'avito') → avito_proxy_rotate_url.""" + mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "9.9.9.9"}) + + with ( + patch("app.services.scrape_pipeline.settings") as s, + patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + ): + s.avito_proxy_rotate_url = "http://avito-only.test/" + s.cian_proxy_rotate_url = None + s.yandex_proxy_rotate_url = None + s.avito_proxy_max_rotations = 4 + s.cian_proxy_max_rotations = 4 + s.yandex_proxy_max_rotations = 4 + s.avito_proxy_rotate_settle_s = 0.0 + + # Вызов без source= — должен работать как раньше + result = await _rotate_proxy_ip(reason="block", rotations_done=0) + + assert result is True + called_url: str = session.get.call_args[0][0] + assert "avito-only.test" in called_url + + +# ── Тест 4: cian fallback на avito когда нет cian url ───────────────────── + + +@pytest.mark.asyncio +async def test_rotate_proxy_ip_cian_fallback_to_avito() -> None: + """cian_proxy_rotate_url=None → fallback на avito_proxy_rotate_url.""" + mock_session_cls, session = _async_session_ctx(resp_json={}) + + with ( + patch("app.services.scrape_pipeline.settings") as s, + patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + ): + s.cian_proxy_rotate_url = None + s.avito_proxy_rotate_url = "http://avito-fallback.test/" + s.yandex_proxy_rotate_url = None + s.cian_proxy_max_rotations = 4 + s.avito_proxy_max_rotations = 4 + s.yandex_proxy_max_rotations = 4 + s.avito_proxy_rotate_settle_s = 0.0 + + result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian") + + assert result is True + called_url: str = session.get.call_args[0][0] + assert "avito-fallback.test" in called_url + + +# ── Тест 5: rotate_url=None → returns False ────────────────────────────────── + + +@pytest.mark.asyncio +async def test_rotate_proxy_ip_returns_false_when_no_url() -> None: + """Никакой URL не задан → False, HTTP-запрос не делается.""" + mock_session_cls, session = _async_session_ctx(resp_json={}) + + with ( + patch("app.services.scrape_pipeline.settings") as s, + patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls), + ): + s.cian_proxy_rotate_url = None + s.avito_proxy_rotate_url = None + s.yandex_proxy_rotate_url = None + s.cian_proxy_max_rotations = 4 + s.avito_proxy_max_rotations = 4 + s.yandex_proxy_max_rotations = 4 + + result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian") + + assert result is False + # HTTP-запрос не делался + session.get.assert_not_called() + + +# ── Тест 6: network error → returns False ──────────────────────────────────── + + +@pytest.mark.asyncio +async def test_rotate_proxy_ip_returns_false_on_network_error() -> None: + """Ошибка сети → returns False (не крашит caller).""" + mock_session_cls, _session = _async_session_ctx(resp_json=None) # raises RuntimeError + + with ( + patch("app.services.scrape_pipeline.settings") as s, + patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls), + ): + s.cian_proxy_rotate_url = "http://cian-rotate.test/" + s.avito_proxy_rotate_url = None + s.yandex_proxy_rotate_url = None + s.cian_proxy_max_rotations = 4 + s.avito_proxy_max_rotations = 4 + s.yandex_proxy_max_rotations = 4 + s.avito_proxy_rotate_settle_s = 0.0 + + result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian") + + assert result is False + + +# ── Тест 7: yandex fallback на avito ───────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_rotate_proxy_ip_yandex_fallback_to_avito() -> None: + """yandex_proxy_rotate_url=None → fallback на avito_proxy_rotate_url.""" + mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "11.22.33.44"}) + + with ( + patch("app.services.scrape_pipeline.settings") as s, + patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + ): + s.yandex_proxy_rotate_url = None + s.avito_proxy_rotate_url = "http://shared-proxy.test/" + s.cian_proxy_rotate_url = None + s.yandex_proxy_max_rotations = 4 + s.avito_proxy_max_rotations = 4 + s.cian_proxy_max_rotations = 4 + s.avito_proxy_rotate_settle_s = 0.0 + + result = await _rotate_proxy_ip(reason="503", rotations_done=0, source="yandex") + + assert result is True + called_url: str = session.get.call_args[0][0] + assert "shared-proxy.test" in called_url -- 2.45.3