fix(scrapers): avito_full_load краш (source_id-reconcile) + cian/yandex ban-ротация #1921
5 changed files with 972 additions and 187 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
242
tradein-mvp/backend/tests/test_provider_rotation_1848.py
Normal file
242
tradein-mvp/backend/tests/test_provider_rotation_1848.py
Normal file
|
|
@ -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
|
||||
327
tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py
Normal file
327
tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue