5 changed files with 664 additions and 16 deletions
|
|
@ -277,6 +277,9 @@ class Settings(BaseSettings):
|
||||||
# #1731: 2→4 — больше шансов восстановиться mid-sweep после проактивной
|
# #1731: 2→4 — больше шансов восстановиться mid-sweep после проактивной
|
||||||
# ротации на старте (Datadome ban recovery).
|
# ротации на старте (Datadome ban recovery).
|
||||||
avito_proxy_max_rotations: int = 4
|
avito_proxy_max_rotations: int = 4
|
||||||
|
# Settle-sleep после changeip-вызова: мобильный модем поднимает новый IP.
|
||||||
|
# ~9с по умолчанию (эмпирика mobileproxy.space). ENV: AVITO_PROXY_ROTATE_SETTLE_S.
|
||||||
|
avito_proxy_rotate_settle_s: float = 9.0
|
||||||
|
|
||||||
# ── Cian dedicated mobile proxy (separate egress from Avito) ──────────────
|
# ── Cian dedicated mobile proxy (separate egress from Avito) ──────────────
|
||||||
# Cian и Avito делят один мобильный IP при общем scraper_proxy_url → конкуренция
|
# Cian и Avito делят один мобильный IP при общем scraper_proxy_url → конкуренция
|
||||||
|
|
|
||||||
|
|
@ -52,6 +52,49 @@ from app.services.yandex_price_history import record_yandex_price_history
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool:
|
||||||
|
"""Вызвать changeip-ссылку mobileproxy и подождать settle (#1790).
|
||||||
|
|
||||||
|
Аналог AvitoScraper._rotate_ip, но на уровне pipeline — используется в
|
||||||
|
enrichment-фазах (houses + detail) где нет доступа к экземпляру scraper'а.
|
||||||
|
|
||||||
|
Returns True при успешной смене IP, False если rotate_url не задан или ошибка.
|
||||||
|
Логирует каждую ротацию (причина, порядковый номер, остаток лимита).
|
||||||
|
"""
|
||||||
|
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 "?"
|
||||||
|
try:
|
||||||
|
async with AsyncSession(timeout=30) as rot:
|
||||||
|
resp = await rot.get(f"{rotate_url}{sep}format=json")
|
||||||
|
new_ip: str = ""
|
||||||
|
try:
|
||||||
|
data = resp.json()
|
||||||
|
new_ip = str(data.get("new_ip", ""))
|
||||||
|
except Exception:
|
||||||
|
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",
|
||||||
|
reason,
|
||||||
|
rotations_done + 1,
|
||||||
|
max_rot - rotations_done - 1,
|
||||||
|
new_ip or "unknown",
|
||||||
|
)
|
||||||
|
return True
|
||||||
|
except Exception:
|
||||||
|
logger.warning(
|
||||||
|
"pipeline: IP rotation failed (reason=%s, rotation=#%d)",
|
||||||
|
reason,
|
||||||
|
rotations_done + 1,
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
# Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает
|
# Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает
|
||||||
# дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for
|
# дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for
|
||||||
# прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом.
|
# прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом.
|
||||||
|
|
@ -200,8 +243,10 @@ async def run_avito_pipeline(
|
||||||
lots: list[ScrapedLot] = []
|
lots: list[ScrapedLot] = []
|
||||||
detail_delay = request_delay_sec if request_delay_sec is not None else 7.0
|
detail_delay = request_delay_sec if request_delay_sec is not None else 7.0
|
||||||
# #1820: общий счётчик ПОСЛЕДОВАТЕЛЬНЫХ блоков на все enrichment-фазы (4/5/5b).
|
# #1820: общий счётчик ПОСЛЕДОВАТЕЛЬНЫХ блоков на все enrichment-фазы (4/5/5b).
|
||||||
# Сбрасывается на любом успешном fetch; при достижении порога — enrichment_aborted.
|
# Сбрасывается на любом успешном fetch; при достижении порога — ротация IP (#1790)
|
||||||
|
# или abort если бюджет ротаций исчерпан.
|
||||||
enrich_consecutive_blocks = 0
|
enrich_consecutive_blocks = 0
|
||||||
|
enrich_rotations_done = 0 # #1790: сколько раз уже сменили IP в этом run
|
||||||
enrichment_aborted = False
|
enrichment_aborted = False
|
||||||
|
|
||||||
browser_mode = settings.scraper_fetch_mode == "browser"
|
browser_mode = settings.scraper_fetch_mode == "browser"
|
||||||
|
|
@ -302,6 +347,24 @@ async def run_avito_pipeline(
|
||||||
enrich_consecutive_blocks,
|
enrich_consecutive_blocks,
|
||||||
)
|
)
|
||||||
if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
|
if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
|
||||||
|
# #1790: Перед abort'ом пробуем сменить IP (changeip).
|
||||||
|
# После ротации сбрасываем счётчик и продолжаем — НЕ abort.
|
||||||
|
# Если ротации исчерпаны — abort enrichment (listings уже сохранены).
|
||||||
|
if enrich_rotations_done < settings.avito_proxy_max_rotations:
|
||||||
|
rotated = await _rotate_proxy_ip(
|
||||||
|
reason=f"house consecutive={enrich_consecutive_blocks}",
|
||||||
|
rotations_done=enrich_rotations_done,
|
||||||
|
)
|
||||||
|
enrich_rotations_done += 1
|
||||||
|
if rotated:
|
||||||
|
enrich_consecutive_blocks = 0
|
||||||
|
logger.info(
|
||||||
|
"pipeline:enrich ROTATE — IP changed, "
|
||||||
|
"reset consecutive blocks (rotation #%d/%d)",
|
||||||
|
enrich_rotations_done,
|
||||||
|
settings.avito_proxy_max_rotations,
|
||||||
|
)
|
||||||
|
continue
|
||||||
logger.error(
|
logger.error(
|
||||||
"pipeline:enrich ABORT — %d consecutive blocks (IP rate-limited), "
|
"pipeline:enrich ABORT — %d consecutive blocks (IP rate-limited), "
|
||||||
"listings preserved",
|
"listings preserved",
|
||||||
|
|
@ -400,6 +463,24 @@ async def run_avito_pipeline(
|
||||||
e,
|
e,
|
||||||
)
|
)
|
||||||
if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
|
if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
|
||||||
|
# #1790: Перед abort'ом пробуем сменить IP (changeip).
|
||||||
|
# После ротации сбрасываем счётчик — продолжаем detail-обход.
|
||||||
|
# Если ротации исчерпаны — abort detail-фазы (listings уже сохранены).
|
||||||
|
if enrich_rotations_done < settings.avito_proxy_max_rotations:
|
||||||
|
rotated = await _rotate_proxy_ip(
|
||||||
|
reason=f"detail consecutive={enrich_consecutive_blocks}",
|
||||||
|
rotations_done=enrich_rotations_done,
|
||||||
|
)
|
||||||
|
enrich_rotations_done += 1
|
||||||
|
if rotated:
|
||||||
|
enrich_consecutive_blocks = 0
|
||||||
|
logger.info(
|
||||||
|
"pipeline:enrich ROTATE — IP changed, "
|
||||||
|
"reset consecutive blocks (rotation #%d/%d)",
|
||||||
|
enrich_rotations_done,
|
||||||
|
settings.avito_proxy_max_rotations,
|
||||||
|
)
|
||||||
|
continue
|
||||||
logger.error(
|
logger.error(
|
||||||
"pipeline:detail ABORT — %d consecutive blocks (IP rate-limited), "
|
"pipeline:detail ABORT — %d consecutive blocks (IP rate-limited), "
|
||||||
"listings preserved",
|
"listings preserved",
|
||||||
|
|
@ -464,6 +545,22 @@ async def run_avito_pipeline(
|
||||||
enrich_consecutive_blocks,
|
enrich_consecutive_blocks,
|
||||||
)
|
)
|
||||||
if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
|
if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
|
||||||
|
# #1790: попытка ротации перед abort'ом (общий бюджет run).
|
||||||
|
if enrich_rotations_done < settings.avito_proxy_max_rotations:
|
||||||
|
rotated = await _rotate_proxy_ip(
|
||||||
|
reason=f"house-detail consecutive={enrich_consecutive_blocks}",
|
||||||
|
rotations_done=enrich_rotations_done,
|
||||||
|
)
|
||||||
|
enrich_rotations_done += 1
|
||||||
|
if rotated:
|
||||||
|
enrich_consecutive_blocks = 0
|
||||||
|
logger.info(
|
||||||
|
"pipeline:enrich ROTATE — IP changed, "
|
||||||
|
"reset consecutive blocks (rotation #%d/%d)",
|
||||||
|
enrich_rotations_done,
|
||||||
|
settings.avito_proxy_max_rotations,
|
||||||
|
)
|
||||||
|
continue
|
||||||
logger.error(
|
logger.error(
|
||||||
"pipeline:enrich ABORT — %d consecutive blocks "
|
"pipeline:enrich ABORT — %d consecutive blocks "
|
||||||
"(IP rate-limited), listings preserved",
|
"(IP rate-limited), listings preserved",
|
||||||
|
|
@ -599,6 +696,9 @@ async def run_avito_city_sweep(
|
||||||
)
|
)
|
||||||
|
|
||||||
browser_mode = settings.scraper_fetch_mode == "browser"
|
browser_mode = settings.scraper_fetch_mode == "browser"
|
||||||
|
# #1790: общий бюджет IP-ротаций на весь sweep (разделяется по всем anchor'ам).
|
||||||
|
sweep_rotations_done = 0
|
||||||
|
|
||||||
async with AsyncExitStack() as stack:
|
async with AsyncExitStack() as stack:
|
||||||
session: AsyncSession | None = None
|
session: AsyncSession | None = None
|
||||||
shared_bf: BrowserFetcher | None = None
|
shared_bf: BrowserFetcher | None = None
|
||||||
|
|
@ -651,7 +751,7 @@ async def run_avito_city_sweep(
|
||||||
НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что
|
НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что
|
||||||
TimeoutError из wait_for теряет только detail-счётчики, но не SERP.
|
TimeoutError из wait_for теряет только detail-счётчики, но не SERP.
|
||||||
"""
|
"""
|
||||||
nonlocal all_touched_house_ids
|
nonlocal all_touched_house_ids, sweep_rotations_done
|
||||||
|
|
||||||
# ── Phase 1+2: SERP + save ─────────────────────────────────
|
# ── Phase 1+2: SERP + save ─────────────────────────────────
|
||||||
scraper = AvitoScraper()
|
scraper = AvitoScraper()
|
||||||
|
|
@ -719,9 +819,12 @@ async def run_avito_city_sweep(
|
||||||
|
|
||||||
# ── Phase 4: enrich houses ─────────────────────────────────
|
# ── Phase 4: enrich houses ─────────────────────────────────
|
||||||
touched_house_ids: set[int] = set()
|
touched_house_ids: set[int] = set()
|
||||||
|
sweep_consecutive_house_blocks = 0 # #1790: per-anchor счётчик
|
||||||
if enrich_houses and unique_house_paths:
|
if enrich_houses and unique_house_paths:
|
||||||
house_paths_list = list(unique_house_paths)
|
house_paths_list = list(unique_house_paths)
|
||||||
for h_idx, house_path in enumerate(house_paths_list):
|
h_idx = 0
|
||||||
|
while h_idx < len(house_paths_list):
|
||||||
|
house_path = house_paths_list[h_idx]
|
||||||
try:
|
try:
|
||||||
enrichment = await fetch_house_catalog(
|
enrichment = await fetch_house_catalog(
|
||||||
house_path,
|
house_path,
|
||||||
|
|
@ -730,16 +833,60 @@ async def run_avito_city_sweep(
|
||||||
)
|
)
|
||||||
hc = save_house_catalog_enrichment(db, enrichment)
|
hc = save_house_catalog_enrichment(db, enrichment)
|
||||||
counters.houses_enriched += 1
|
counters.houses_enriched += 1
|
||||||
|
sweep_consecutive_house_blocks = 0
|
||||||
hid = hc.get("house_id")
|
hid = hc.get("house_id")
|
||||||
if hid:
|
if hid:
|
||||||
touched_house_ids.add(int(hid))
|
touched_house_ids.add(int(hid))
|
||||||
except (AvitoBlockedError, AvitoRateLimitedError):
|
if h_idx < len(house_paths_list) - 1:
|
||||||
logger.error(
|
await asyncio.sleep(request_delay_sec)
|
||||||
"city-sweep run_id=%d: house BLOCKED at %s — propagating",
|
h_idx += 1
|
||||||
|
except (AvitoBlockedError, AvitoRateLimitedError) as be:
|
||||||
|
# #1790: пробуем сменить IP перед re-raise.
|
||||||
|
sweep_consecutive_house_blocks += 1
|
||||||
|
counters.houses_failed += 1
|
||||||
|
counters.errors_count += 1
|
||||||
|
logger.warning(
|
||||||
|
"city-sweep run_id=%d: house BLOCKED at %s "
|
||||||
|
"(consecutive=%d) — %s",
|
||||||
run_id,
|
run_id,
|
||||||
house_path,
|
house_path,
|
||||||
|
sweep_consecutive_house_blocks,
|
||||||
|
be,
|
||||||
)
|
)
|
||||||
raise
|
if (
|
||||||
|
sweep_consecutive_house_blocks
|
||||||
|
>= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
|
||||||
|
and sweep_rotations_done < settings.avito_proxy_max_rotations
|
||||||
|
):
|
||||||
|
rotated = await _rotate_proxy_ip(
|
||||||
|
reason=(
|
||||||
|
f"city-sweep house "
|
||||||
|
f"consecutive={sweep_consecutive_house_blocks}"
|
||||||
|
),
|
||||||
|
rotations_done=sweep_rotations_done,
|
||||||
|
)
|
||||||
|
sweep_rotations_done += 1
|
||||||
|
if rotated:
|
||||||
|
sweep_consecutive_house_blocks = 0
|
||||||
|
logger.info(
|
||||||
|
"city-sweep run_id=%d: house ROTATE — "
|
||||||
|
"IP changed, reset consecutive "
|
||||||
|
"(rotation #%d/%d)",
|
||||||
|
run_id,
|
||||||
|
sweep_rotations_done,
|
||||||
|
settings.avito_proxy_max_rotations,
|
||||||
|
)
|
||||||
|
# Не инкрементируем h_idx — retry текущего дома
|
||||||
|
continue
|
||||||
|
elif sweep_consecutive_house_blocks >= 3:
|
||||||
|
logger.error(
|
||||||
|
"city-sweep run_id=%d: house ABORT — "
|
||||||
|
"%d consecutive blocks, propagating",
|
||||||
|
run_id,
|
||||||
|
sweep_consecutive_house_blocks,
|
||||||
|
)
|
||||||
|
raise
|
||||||
|
h_idx += 1
|
||||||
except Exception as he:
|
except Exception as he:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"city-sweep run_id=%d: house_enrich failed %s: %s",
|
"city-sweep run_id=%d: house_enrich failed %s: %s",
|
||||||
|
|
@ -753,8 +900,7 @@ async def run_avito_city_sweep(
|
||||||
db.rollback()
|
db.rollback()
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
if h_idx < len(house_paths_list) - 1:
|
h_idx += 1
|
||||||
await asyncio.sleep(request_delay_sec)
|
|
||||||
|
|
||||||
# ── Phase 5: enrich detail (top-N) ────────────────────────
|
# ── Phase 5: enrich detail (top-N) ────────────────────────
|
||||||
if detail_top_n > 0 and anchor_lots:
|
if detail_top_n > 0 and anchor_lots:
|
||||||
|
|
@ -833,6 +979,27 @@ async def run_avito_city_sweep(
|
||||||
be,
|
be,
|
||||||
)
|
)
|
||||||
if consecutive_blocks >= 3:
|
if consecutive_blocks >= 3:
|
||||||
|
# #1790: попытка ротации перед abort'ом.
|
||||||
|
if sweep_rotations_done < settings.avito_proxy_max_rotations:
|
||||||
|
rotated = await _rotate_proxy_ip(
|
||||||
|
reason=(
|
||||||
|
f"city-sweep detail "
|
||||||
|
f"consecutive={consecutive_blocks}"
|
||||||
|
),
|
||||||
|
rotations_done=sweep_rotations_done,
|
||||||
|
)
|
||||||
|
sweep_rotations_done += 1
|
||||||
|
if rotated:
|
||||||
|
consecutive_blocks = 0
|
||||||
|
logger.info(
|
||||||
|
"city-sweep run_id=%d: detail ROTATE — "
|
||||||
|
"IP changed, reset consecutive "
|
||||||
|
"(rotation #%d/%d)",
|
||||||
|
run_id,
|
||||||
|
sweep_rotations_done,
|
||||||
|
settings.avito_proxy_max_rotations,
|
||||||
|
)
|
||||||
|
continue
|
||||||
logger.error(
|
logger.error(
|
||||||
"city-sweep run_id=%d: detail ABORT — "
|
"city-sweep run_id=%d: detail ABORT — "
|
||||||
"%d consecutive blocks (IP rate-limited)",
|
"%d consecutive blocks (IP rate-limited)",
|
||||||
|
|
@ -877,7 +1044,10 @@ async def run_avito_city_sweep(
|
||||||
if enrich_houses and new_house_paths:
|
if enrich_houses and new_house_paths:
|
||||||
counters.unique_houses += len(new_house_paths)
|
counters.unique_houses += len(new_house_paths)
|
||||||
nh_list = list(new_house_paths)
|
nh_list = list(new_house_paths)
|
||||||
for nh_idx, house_path in enumerate(nh_list):
|
nh_idx = 0
|
||||||
|
sweep_consecutive_detail_house_blocks = 0 # #1790
|
||||||
|
while nh_idx < len(nh_list):
|
||||||
|
house_path = nh_list[nh_idx]
|
||||||
try:
|
try:
|
||||||
enrichment = await fetch_house_catalog(
|
enrichment = await fetch_house_catalog(
|
||||||
house_path,
|
house_path,
|
||||||
|
|
@ -886,17 +1056,59 @@ async def run_avito_city_sweep(
|
||||||
)
|
)
|
||||||
hc = save_house_catalog_enrichment(db, enrichment)
|
hc = save_house_catalog_enrichment(db, enrichment)
|
||||||
counters.houses_enriched += 1
|
counters.houses_enriched += 1
|
||||||
|
sweep_consecutive_detail_house_blocks = 0
|
||||||
hid = hc.get("house_id")
|
hid = hc.get("house_id")
|
||||||
if hid:
|
if hid:
|
||||||
touched_house_ids.add(int(hid))
|
touched_house_ids.add(int(hid))
|
||||||
except (AvitoBlockedError, AvitoRateLimitedError):
|
if nh_idx < len(nh_list) - 1:
|
||||||
logger.error(
|
await asyncio.sleep(request_delay_sec)
|
||||||
|
nh_idx += 1
|
||||||
|
except (AvitoBlockedError, AvitoRateLimitedError) as beh:
|
||||||
|
# #1790: ротация перед re-raise.
|
||||||
|
sweep_consecutive_detail_house_blocks += 1
|
||||||
|
counters.houses_failed += 1
|
||||||
|
counters.errors_count += 1
|
||||||
|
logger.warning(
|
||||||
"city-sweep run_id=%d: house(detail) BLOCKED %s"
|
"city-sweep run_id=%d: house(detail) BLOCKED %s"
|
||||||
" — propagating",
|
" (consecutive=%d) — %s",
|
||||||
run_id,
|
run_id,
|
||||||
house_path,
|
house_path,
|
||||||
|
sweep_consecutive_detail_house_blocks,
|
||||||
|
beh,
|
||||||
)
|
)
|
||||||
raise
|
if (
|
||||||
|
sweep_consecutive_detail_house_blocks >= 3
|
||||||
|
and sweep_rotations_done
|
||||||
|
< settings.avito_proxy_max_rotations
|
||||||
|
):
|
||||||
|
rotated = await _rotate_proxy_ip(
|
||||||
|
reason=(
|
||||||
|
"city-sweep house(detail) "
|
||||||
|
f"consecutive="
|
||||||
|
f"{sweep_consecutive_detail_house_blocks}"
|
||||||
|
),
|
||||||
|
rotations_done=sweep_rotations_done,
|
||||||
|
)
|
||||||
|
sweep_rotations_done += 1
|
||||||
|
if rotated:
|
||||||
|
sweep_consecutive_detail_house_blocks = 0
|
||||||
|
logger.info(
|
||||||
|
"city-sweep run_id=%d: house(detail) "
|
||||||
|
"ROTATE — IP changed (rotation #%d/%d)",
|
||||||
|
run_id,
|
||||||
|
sweep_rotations_done,
|
||||||
|
settings.avito_proxy_max_rotations,
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
elif sweep_consecutive_detail_house_blocks >= 3:
|
||||||
|
logger.error(
|
||||||
|
"city-sweep run_id=%d: house(detail) ABORT"
|
||||||
|
" — %d consecutive blocks, propagating",
|
||||||
|
run_id,
|
||||||
|
sweep_consecutive_detail_house_blocks,
|
||||||
|
)
|
||||||
|
raise
|
||||||
|
nh_idx += 1
|
||||||
except Exception as he:
|
except Exception as he:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"city-sweep run_id=%d: house(detail) failed %s: %s",
|
"city-sweep run_id=%d: house(detail) failed %s: %s",
|
||||||
|
|
@ -910,8 +1122,7 @@ async def run_avito_city_sweep(
|
||||||
db.rollback()
|
db.rollback()
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
if nh_idx < len(nh_list) - 1:
|
nh_idx += 1
|
||||||
await asyncio.sleep(request_delay_sec)
|
|
||||||
|
|
||||||
all_touched_house_ids.update(touched_house_ids)
|
all_touched_house_ids.update(touched_house_ids)
|
||||||
logger.info(
|
logger.info(
|
||||||
|
|
|
||||||
|
|
@ -793,6 +793,35 @@ def parse_houses_state(state: dict[str, Any], house_url: str) -> HouseCatalogEnr
|
||||||
reviews, next_page_url = _parse_reviews(reviews_widget)
|
reviews, next_page_url = _parse_reviews(reviews_widget)
|
||||||
if not reviews:
|
if not reviews:
|
||||||
logger.warning("Виджет 'reviews' найден, но 0 отзывов распарсено — проверь шейп props")
|
logger.warning("Виджет 'reviews' найден, но 0 отзывов распарсено — проверь шейп props")
|
||||||
|
# #1789: рейтинг переехал из housePage.developmentData.ratingPreview в
|
||||||
|
# reviews.filterEntries[type=="score"]. Применяем как fallback — только
|
||||||
|
# если housePage не дал рейтинг (чтобы не затирать уже правильный источник).
|
||||||
|
if house_info.rating_score is None:
|
||||||
|
reviews_payload = _widget_payload(reviews_widget, "reviews")
|
||||||
|
filter_entries: list[dict[str, Any]] = reviews_payload.get("filterEntries", [])
|
||||||
|
for fe in filter_entries:
|
||||||
|
if fe.get("type") == "score":
|
||||||
|
score_value = fe.get("value", {})
|
||||||
|
if isinstance(score_value, dict):
|
||||||
|
score_float = score_value.get("scoreFloat")
|
||||||
|
if score_float is not None:
|
||||||
|
house_info.rating_score = float(score_float)
|
||||||
|
house_info.rating_string = score_value.get("scoreString")
|
||||||
|
rc = score_value.get("reviewCount")
|
||||||
|
if rc is not None:
|
||||||
|
house_info.reviews_count = int(rc)
|
||||||
|
rating_stat = score_value.get("ratingStat")
|
||||||
|
if isinstance(rating_stat, list):
|
||||||
|
house_info.rating_distribution = rating_stat
|
||||||
|
logger.info(
|
||||||
|
"parse_houses_state: rating из reviews.filterEntries "
|
||||||
|
"url=%s score=%.3f string=%s reviews=%s",
|
||||||
|
house_url,
|
||||||
|
house_info.rating_score,
|
||||||
|
house_info.rating_string,
|
||||||
|
house_info.reviews_count,
|
||||||
|
)
|
||||||
|
break
|
||||||
else:
|
else:
|
||||||
logger.debug("Виджет 'reviews' не найден — пропускаем")
|
logger.debug("Виджет 'reviews' не найден — пропускаем")
|
||||||
|
|
||||||
|
|
|
||||||
216
tradein-mvp/backend/tests/test_ban_rotation_1790.py
Normal file
216
tradein-mvp/backend/tests/test_ban_rotation_1790.py
Normal file
|
|
@ -0,0 +1,216 @@
|
||||||
|
"""#1790 — ban-triggered IP rotation в enrichment-фазах scrape_pipeline.
|
||||||
|
|
||||||
|
Тест 1: AvitoBlockedError при доступных ротациях → вызывается _rotate_proxy_ip
|
||||||
|
(changeip) + consecutive счётчик сбрасывается → enrichment продолжается.
|
||||||
|
|
||||||
|
Тест 2: Ротаций больше нет (бюджет исчерпан) → abort enrichment (bounded N ротаций).
|
||||||
|
|
||||||
|
Тест 3: Graceful per-item (#1820) сохранён — один блок без порога ротации НЕ вызывает.
|
||||||
|
|
||||||
|
Без сети, без БД.
|
||||||
|
"""
|
||||||
|
|
||||||
|
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 (
|
||||||
|
_AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT,
|
||||||
|
PipelineResult,
|
||||||
|
run_avito_pipeline,
|
||||||
|
)
|
||||||
|
from app.services.scrapers.avito_exceptions import AvitoBlockedError
|
||||||
|
|
||||||
|
|
||||||
|
def _make_lot(
|
||||||
|
item_id: str,
|
||||||
|
house_url: str | None = None,
|
||||||
|
) -> object:
|
||||||
|
from app.services.scrapers.base import ScrapedLot
|
||||||
|
|
||||||
|
return ScrapedLot(
|
||||||
|
source="avito",
|
||||||
|
source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/test_{item_id}",
|
||||||
|
source_id=item_id,
|
||||||
|
price_rub=5_000_000,
|
||||||
|
address="Test address",
|
||||||
|
house_source="avito" if house_url else None,
|
||||||
|
house_ext_id=house_url.rstrip("/").split("/")[-1] if house_url else None,
|
||||||
|
house_url=house_url,
|
||||||
|
listing_segment="vtorichka",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _block_scraper(lots: list) -> MagicMock: # type: ignore[type-arg]
|
||||||
|
scraper = MagicMock()
|
||||||
|
scraper.__aenter__ = AsyncMock(return_value=scraper)
|
||||||
|
scraper.__aexit__ = AsyncMock(return_value=None)
|
||||||
|
scraper.fetch_around = AsyncMock(return_value=lots)
|
||||||
|
scraper._cffi = None
|
||||||
|
return scraper
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Тест 1: ротация срабатывает при consecutive blocks == порог
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_rotation_called_on_consecutive_blocks() -> None:
|
||||||
|
"""При N подряд-блоках (порог) вызывается _rotate_proxy_ip — IP меняется.
|
||||||
|
|
||||||
|
После успешной ротации consecutive счётчик сбрасывается и enrichment продолжается:
|
||||||
|
следующий (N+1)-й дом фетчится успешно.
|
||||||
|
"""
|
||||||
|
n = _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
|
||||||
|
lots = [
|
||||||
|
_make_lot(str(i), f"https://www.avito.ru/catalog/houses/ekb/h{i}/{100 + i}")
|
||||||
|
for i in range(n + 2) # n блокируются, потом ротация, потом успех
|
||||||
|
]
|
||||||
|
scraper = _block_scraper(lots)
|
||||||
|
|
||||||
|
# Первые n обращений — блок, (n+1)-й — успех
|
||||||
|
fetch_house = AsyncMock(side_effect=[AvitoBlockedError("banned")] * n + [MagicMock()])
|
||||||
|
rotate_mock = AsyncMock(return_value=True) # ротация успешна
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper),
|
||||||
|
patch("app.services.scrape_pipeline.save_listings", return_value=(len(lots), 0)),
|
||||||
|
patch("app.services.scrape_pipeline.fetch_house_catalog", fetch_house),
|
||||||
|
patch(
|
||||||
|
"app.services.scrape_pipeline.save_house_catalog_enrichment",
|
||||||
|
return_value={"house_id": 99},
|
||||||
|
),
|
||||||
|
patch("app.services.scrape_pipeline._rotate_proxy_ip", rotate_mock),
|
||||||
|
patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
|
||||||
|
patch(
|
||||||
|
"curl_cffi.requests.AsyncSession",
|
||||||
|
return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
|
||||||
|
),
|
||||||
|
):
|
||||||
|
result = await run_avito_pipeline(
|
||||||
|
MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0
|
||||||
|
)
|
||||||
|
|
||||||
|
assert isinstance(result, PipelineResult)
|
||||||
|
# _rotate_proxy_ip вызвана хотя бы один раз
|
||||||
|
assert rotate_mock.await_count >= 1
|
||||||
|
# Enrichment продолжился после ротации — (n+1)-й дом обогащён
|
||||||
|
assert result.counters.houses_enriched >= 1
|
||||||
|
# Listings сохранены
|
||||||
|
assert result.counters.lots_inserted == len(lots)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Тест 2: бюджет ротаций исчерпан → abort
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_rotation_bounded_abort_when_budget_exhausted() -> None:
|
||||||
|
"""После max_rotations попыток ротации enrichment abort'ится (bounded).
|
||||||
|
|
||||||
|
Используем avito_proxy_max_rotations=1 → только одна попытка ротации.
|
||||||
|
После неё блоки продолжаются → abort.
|
||||||
|
"""
|
||||||
|
n = _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
|
||||||
|
# Достаточно домов чтобы вызвать и ротацию и повторные блоки
|
||||||
|
lots = [
|
||||||
|
_make_lot(str(i), f"https://www.avito.ru/catalog/houses/ekb/h{i}/{200 + i}")
|
||||||
|
for i in range(n * 3)
|
||||||
|
]
|
||||||
|
scraper = _block_scraper(lots)
|
||||||
|
|
||||||
|
# Все дома блокируются (не будет успешного фетча)
|
||||||
|
fetch_house = AsyncMock(side_effect=AvitoBlockedError("banned"))
|
||||||
|
# Ротация успешна, но дом после неё тоже блокируется
|
||||||
|
rotate_mock = AsyncMock(return_value=True)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper),
|
||||||
|
patch("app.services.scrape_pipeline.save_listings", return_value=(len(lots), 0)),
|
||||||
|
patch("app.services.scrape_pipeline.fetch_house_catalog", fetch_house),
|
||||||
|
patch("app.services.scrape_pipeline.save_house_catalog_enrichment", return_value={}),
|
||||||
|
patch("app.services.scrape_pipeline._rotate_proxy_ip", rotate_mock),
|
||||||
|
patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
|
||||||
|
patch("app.services.scrape_pipeline.settings") as mock_settings,
|
||||||
|
patch(
|
||||||
|
"curl_cffi.requests.AsyncSession",
|
||||||
|
return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
|
||||||
|
),
|
||||||
|
):
|
||||||
|
# Лимит 1 ротация: после неё бюджет исчерпан → abort на следующей серии блоков
|
||||||
|
mock_settings.avito_proxy_max_rotations = 1
|
||||||
|
mock_settings.avito_proxy_rotate_url = "http://rotate.test/"
|
||||||
|
mock_settings.avito_proxy_rotate_settle_s = 0.0
|
||||||
|
mock_settings.scraper_fetch_mode = "curl_cffi"
|
||||||
|
mock_settings.scraper_proxy_url = None
|
||||||
|
mock_settings.scraper_skip_seen_today = False
|
||||||
|
|
||||||
|
result = await run_avito_pipeline(
|
||||||
|
MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0
|
||||||
|
)
|
||||||
|
|
||||||
|
assert isinstance(result, PipelineResult)
|
||||||
|
# Ровно 1 ротация (бюджет = 1)
|
||||||
|
assert rotate_mock.await_count == 1
|
||||||
|
# Enrichment abort'нут (есть AVITO_BLOCKED в ошибках)
|
||||||
|
assert any("AVITO_BLOCKED" in e for e in result.counters.errors)
|
||||||
|
# Listings всё равно сохранены
|
||||||
|
assert result.counters.lots_inserted == len(lots)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Тест 3: один блок (ниже порога) — ротация НЕ вызывается (#1820 graceful)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_single_block_below_threshold_no_rotation() -> None:
|
||||||
|
"""Один блок (< порога) — _rotate_proxy_ip НЕ вызывается.
|
||||||
|
|
||||||
|
Graceful per-item #1820: единственный блок пишется в errors, enrichment
|
||||||
|
продолжается, ротация не тратится.
|
||||||
|
"""
|
||||||
|
n = _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
|
||||||
|
# Только 2 дома: первый блокируется, второй успешен
|
||||||
|
lots = [
|
||||||
|
_make_lot("1", "https://www.avito.ru/catalog/houses/ekb/a/100"),
|
||||||
|
_make_lot("2", "https://www.avito.ru/catalog/houses/ekb/b/200"),
|
||||||
|
]
|
||||||
|
assert n >= 2, "Порог должен быть >= 2 чтобы тест был корректен"
|
||||||
|
scraper = _block_scraper(lots)
|
||||||
|
|
||||||
|
fetch_house = AsyncMock(side_effect=[AvitoBlockedError("HTTP 403"), MagicMock()])
|
||||||
|
rotate_mock = AsyncMock(return_value=True)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper),
|
||||||
|
patch("app.services.scrape_pipeline.save_listings", return_value=(2, 0)),
|
||||||
|
patch("app.services.scrape_pipeline.fetch_house_catalog", fetch_house),
|
||||||
|
patch(
|
||||||
|
"app.services.scrape_pipeline.save_house_catalog_enrichment",
|
||||||
|
return_value={"house_id": 55},
|
||||||
|
),
|
||||||
|
patch("app.services.scrape_pipeline._rotate_proxy_ip", rotate_mock),
|
||||||
|
patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
|
||||||
|
patch(
|
||||||
|
"curl_cffi.requests.AsyncSession",
|
||||||
|
return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
|
||||||
|
),
|
||||||
|
):
|
||||||
|
result = await run_avito_pipeline(
|
||||||
|
MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0
|
||||||
|
)
|
||||||
|
|
||||||
|
# Ротация НЕ вызвана — один блок ниже порога
|
||||||
|
assert rotate_mock.await_count == 0
|
||||||
|
# Первый дом упал, второй обогащён
|
||||||
|
assert result.counters.houses_failed == 1
|
||||||
|
assert result.counters.houses_enriched == 1
|
||||||
|
assert any("BLOCKED_HOUSE" in e for e in result.counters.errors)
|
||||||
|
|
@ -0,0 +1,189 @@
|
||||||
|
"""#1789 — rating_score из reviews.filterEntries[type==score] (fallback).
|
||||||
|
|
||||||
|
Avito перенёс сводный рейтинг из housePage.developmentData.ratingPreview в виджет
|
||||||
|
reviews.filterEntries[{type: 'score', value: {scoreFloat, scoreString, reviewCount, ratingStat}}].
|
||||||
|
|
||||||
|
Тест 1: когда housePage НЕ даёт рейтинг (ratingBadge пуст), rating берётся из
|
||||||
|
reviews.filterEntries — проверяем rating_score/rating_string/reviews_count/
|
||||||
|
rating_distribution.
|
||||||
|
|
||||||
|
Тест 2: housePage даёт рейтинг (старый путь ratingBadge) → filterEntries НЕ
|
||||||
|
перезаписывает (приоритет housePage).
|
||||||
|
|
||||||
|
Тест 3: новостройка (developmentPage) без reviews-виджета → graceful, нет краша.
|
||||||
|
|
||||||
|
Без сети, без БД.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||||||
|
|
||||||
|
from app.services.scrapers.avito_houses import parse_houses_state
|
||||||
|
|
||||||
|
|
||||||
|
def _make_state(
|
||||||
|
*,
|
||||||
|
house_page_rating: dict | None = None,
|
||||||
|
reviews_filter_entries: list | None = None,
|
||||||
|
) -> dict:
|
||||||
|
"""Минимальный __preloadedState__ с housePage и опциональными reviews.
|
||||||
|
|
||||||
|
house_page_rating — если задан, проставляем в developmentData (старый путь).
|
||||||
|
reviews_filter_entries — если задан, проставляем в reviews.props.reviews.filterEntries
|
||||||
|
"""
|
||||||
|
dev_data: dict = {
|
||||||
|
"avitoId": 555666,
|
||||||
|
"id": "aGFzaA==",
|
||||||
|
"title": "Test House",
|
||||||
|
"address": "ул. Тестовая, 1",
|
||||||
|
"coords": {"lat": 56.8, "lng": 60.6},
|
||||||
|
"aboutDevelopment": {"expandParams": {"items": []}},
|
||||||
|
"developer": {},
|
||||||
|
"mapPreview": {},
|
||||||
|
}
|
||||||
|
if house_page_rating is not None:
|
||||||
|
dev_data.update(house_page_rating)
|
||||||
|
|
||||||
|
placeholders: list = [
|
||||||
|
{
|
||||||
|
"type": "housePage",
|
||||||
|
"props": {"developmentData": dev_data},
|
||||||
|
},
|
||||||
|
]
|
||||||
|
|
||||||
|
if reviews_filter_entries is not None:
|
||||||
|
placeholders.append(
|
||||||
|
{
|
||||||
|
"type": "reviews",
|
||||||
|
"props": {
|
||||||
|
# MFE-форма: данные под ключом "reviews" внутри props
|
||||||
|
"reviews": {
|
||||||
|
"entries": [],
|
||||||
|
"filterEntries": reviews_filter_entries,
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return {"data": {"data": {"page": {"placeholders": placeholders}}}}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Тест 1: рейтинг из reviews.filterEntries (housePage пуст)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_rating_from_filter_entries_when_housepage_empty() -> None:
|
||||||
|
"""Когда housePage не даёт рейтинг — берём из reviews.filterEntries."""
|
||||||
|
state = _make_state(
|
||||||
|
reviews_filter_entries=[
|
||||||
|
{
|
||||||
|
"type": "score",
|
||||||
|
"value": {
|
||||||
|
"scoreFloat": 4.666,
|
||||||
|
"scoreString": "4,7",
|
||||||
|
"reviewCount": 3,
|
||||||
|
"ratingStat": [
|
||||||
|
{"score": 5, "count": 2},
|
||||||
|
{"score": 4, "count": 1},
|
||||||
|
],
|
||||||
|
},
|
||||||
|
}
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
|
||||||
|
h = e.house
|
||||||
|
|
||||||
|
assert h.rating_score == pytest.approx(4.666, rel=1e-3)
|
||||||
|
assert h.rating_string == "4,7"
|
||||||
|
assert h.reviews_count == 3
|
||||||
|
assert len(h.rating_distribution) == 2
|
||||||
|
assert h.rating_distribution[0]["score"] == 5
|
||||||
|
|
||||||
|
|
||||||
|
def test_rating_filter_entries_float_type() -> None:
|
||||||
|
"""scoreFloat из filterEntries конвертируется в float (не остаётся int/str)."""
|
||||||
|
state = _make_state(
|
||||||
|
reviews_filter_entries=[
|
||||||
|
{
|
||||||
|
"type": "score",
|
||||||
|
"value": {"scoreFloat": 5, "scoreString": "5,0", "reviewCount": 1},
|
||||||
|
}
|
||||||
|
]
|
||||||
|
)
|
||||||
|
e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
|
||||||
|
assert isinstance(e.house.rating_score, float)
|
||||||
|
assert e.house.rating_score == 5.0
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Тест 2: housePage даёт рейтинг → filterEntries НЕ перезаписывает
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_housepage_rating_takes_priority_over_filter_entries() -> None:
|
||||||
|
"""Если housePage.ratingBadge уже заполнен — filterEntries не перезаписывает."""
|
||||||
|
state = _make_state(
|
||||||
|
house_page_rating={
|
||||||
|
"ratingBadge": {"info": {"score": 3.9, "scoreString": "3,9"}},
|
||||||
|
"ratingSummaryStat": {"reviewCount": 10, "ratingStat": []},
|
||||||
|
},
|
||||||
|
reviews_filter_entries=[
|
||||||
|
{
|
||||||
|
"type": "score",
|
||||||
|
"value": {
|
||||||
|
"scoreFloat": 4.9,
|
||||||
|
"scoreString": "4,9",
|
||||||
|
"reviewCount": 99,
|
||||||
|
"ratingStat": [],
|
||||||
|
},
|
||||||
|
}
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
|
||||||
|
h = e.house
|
||||||
|
|
||||||
|
# Значение из housePage должно сохраниться
|
||||||
|
assert h.rating_score == pytest.approx(3.9, rel=1e-3)
|
||||||
|
assert h.rating_string == "3,9"
|
||||||
|
assert h.reviews_count == 10
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Тест 3: нет reviews-виджета вообще → None, не падаем
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_no_reviews_widget_rating_is_none() -> None:
|
||||||
|
"""Без виджета reviews и без ratingBadge → rating_score=None, graceful."""
|
||||||
|
state = _make_state() # нет reviews_filter_entries
|
||||||
|
|
||||||
|
e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
|
||||||
|
h = e.house
|
||||||
|
|
||||||
|
assert h.rating_score is None
|
||||||
|
assert h.rating_string is None
|
||||||
|
assert h.reviews_count is None
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Тест 4: filterEntries без entry type==score → rating остаётся None
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_entries_without_score_type() -> None:
|
||||||
|
"""filterEntries без type==score → рейтинг не заполняется."""
|
||||||
|
state = _make_state(
|
||||||
|
reviews_filter_entries=[
|
||||||
|
{"type": "something_else", "value": {"scoreFloat": 9.9}},
|
||||||
|
]
|
||||||
|
)
|
||||||
|
e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
|
||||||
|
assert e.house.rating_score is None
|
||||||
Loading…
Add table
Reference in a new issue