diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index fc6b61cc..210bcbc7 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -556,7 +556,7 @@ async def scrape_avito_imv( class CitySweepStartRequest(BaseModel): pages_per_anchor: int = Field(default=3, ge=1, le=10) - detail_top_n: int = Field(default=20, ge=0, le=30) + detail_top_n: int = Field(default=10, ge=0, le=30) request_delay_sec: float = Field(default=7.0, ge=3.0, le=15.0) enrich_houses: bool = True radius_m: int = Field(default=1500, ge=500, le=5000) diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index b674ade0..a6198a60 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -10,23 +10,34 @@ City sweep: run_avito_city_sweep — итерирует EKB_ANCHORS × pages_per_anchor, с cooperative cancel через scrape_runs.status и heartbeat update каждый anchor. +Anti-bot hardening (2023-05-23): + - Shared AsyncSession на весь sweep (один TLS handshake) + - asyncio.sleep с random +-20% jitter между detail requests + - AvitoBlockedError/AvitoRateLimitedError propagation → early abort + - После 3 consecutive blocks — abort sweep + mark_banned (status='banned') + Graceful degradation на каждом step: exception в одной house/listing -не валит весь pipeline. +не валит весь pipeline (кроме blocked exceptions — они abort весь sweep). IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration), не cron. """ from __future__ import annotations +import asyncio import logging +import random from dataclasses import dataclass, field, fields from urllib.parse import urlparse +from curl_cffi.requests import AsyncSession from sqlalchemy import text from sqlalchemy.orm import Session +from app.services import scrape_runs from app.services.scrapers.avito import AvitoScraper from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment +from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment from app.services.scrapers.base import ScrapedLot, save_listings @@ -41,8 +52,19 @@ EKB_ANCHORS: list[tuple[float, float, str]] = [ (56.8650, 60.6200, "Пионерский"), ] +_CHROME_HEADERS = { + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", + "Cache-Control": "max-age=0", + "Sec-Fetch-Dest": "document", + "Sec-Fetch-Mode": "navigate", + "Sec-Fetch-Site": "none", + "Sec-Fetch-User": "?1", + "Upgrade-Insecure-Requests": "1", +} -# ── Result dataclasses ────────────────────────────────────────────────────── + +# ── Result dataclasses ────────────────────────────────────────── @dataclass @@ -73,7 +95,7 @@ class PipelineResult: enrich_detail_top_n: int -# ── Main orchestrator ─────────────────────────────────────────────────────── +# ── Main orchestrator ─────────────────────────────────────────── async def run_avito_pipeline( @@ -86,6 +108,7 @@ async def run_avito_pipeline( enrich_detail_top_n: int = 10, pages: int = 1, request_delay_sec: float | None = None, + shared_session: AsyncSession | None = None, ) -> PipelineResult: """Full Avito search → houses → detail enrichment pipeline. @@ -93,151 +116,187 @@ async def run_avito_pipeline( 1. SEARCH: AvitoScraper().fetch_around(lat, lon, radius_m, pages=pages) 2. SAVE: save_listings(db, lots) → (inserted, updated) 3. GROUP: dedupe house_url из новых lots → unique house path set - 4. ENRICH_HOUSES (если enrich_houses=True): для каждого unique house path - → fetch_house_catalog + save_house_catalog_enrichment (try/except per house) - 5. ENRICH_DETAIL (если enrich_detail_top_n > 0): select top-N priority listings - (приоритет — detail_enriched_at IS NULL + price > 0) - → fetch_detail + save_detail_enrichment (try/except per listing) - 6. Return PipelineResult с counters + 4. ENRICH_HOUSES: fetch_house_catalog + save per house (with sleep between) + 5. ENRICH_DETAIL: top-N listings → fetch_detail + save (abort on 3 blocks) - pages=1 (default) — backward compat, одна страница. - pages>1 — pagination (city sweep). request_delay_sec — override задержки между страницами. - - Graceful degradation на каждом step — exception на одной house/listing - не должна провалить весь pipeline. + shared_session — если передан, используется без закрытия (lifecycle у вызывающего). + AvitoBlockedError / AvitoRateLimitedError propagate наружу. """ counters = PipelineCounters() lots: list[ScrapedLot] = [] + detail_delay = request_delay_sec if request_delay_sec is not None else 7.0 - # ── Step 1: search ────────────────────────────────────────────────────── - async with AvitoScraper() as scraper: + own_session = shared_session is None + session = shared_session or AsyncSession( + impersonate="chrome120", + timeout=25, + headers=_CHROME_HEADERS, + ) + + try: + # ── Step 1: search ────────────────────────────────────── + scraper = AvitoScraper() + scraper._cffi = session try: lots = await scraper.fetch_around( - lat, lon, radius_m, pages=pages, delay_override_sec=request_delay_sec + lat, lon, radius_m, + pages=pages, + delay_override_sec=request_delay_sec, ) counters.lots_fetched = len(lots) logger.info( "pipeline:search anchor=(%.4f,%.4f) r=%dm fetched=%d", - lat, - lon, - radius_m, - len(lots), + lat, lon, radius_m, len(lots), ) + except (AvitoBlockedError, AvitoRateLimitedError): + logger.error( + "pipeline:search BLOCKED by Avito at anchor=(%.4f,%.4f)", lat, lon + ) + raise except Exception as e: logger.exception("pipeline:search failed") counters.errors.append(f"search: {e}") - return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n) + return PipelineResult( + lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n + ) - # ── Step 2: save listings ─────────────────────────────────────────────── - if lots: - try: - counters.lots_inserted, counters.lots_updated = save_listings(db, lots) - except Exception as e: - logger.exception("pipeline:save_listings failed") - counters.errors.append(f"save_listings: {e}") - - # ── Step 3: group by house ────────────────────────────────────────────── - # Dedupe house path (без host) для consistency; fetch_house_catalog принимает path - unique_house_paths: set[str] = set() - for lot in lots: - if lot.house_url: + # ── Step 2: save listings ─────────────────────────────── + if lots: try: - parsed = urlparse(lot.house_url) - path = parsed.path if parsed.path else lot.house_url - unique_house_paths.add(path) - except Exception: - continue - - counters.unique_houses = len(unique_house_paths) - logger.info("pipeline:group_by_house unique=%d", counters.unique_houses) - - # ── Step 4: enrich houses ─────────────────────────────────────────────── - if enrich_houses and unique_house_paths: - for house_path in unique_house_paths: - try: - enrichment = await fetch_house_catalog(house_path) - save_house_catalog_enrichment(db, enrichment) - counters.houses_enriched += 1 + counters.lots_inserted, counters.lots_updated = save_listings(db, lots) except Exception as e: - logger.warning( - "pipeline:house_enrich failed for %s: %s", house_path, e - ) - counters.houses_failed += 1 - counters.errors.append(f"house {house_path}: {e}") + logger.exception("pipeline:save_listings failed") + counters.errors.append(f"save_listings: {e}") - # ── Step 5: enrich detail для top-N priority listings ────────────────── - if enrich_detail_top_n > 0: - # Priority: листинги ещё не enriched (detail_enriched_at IS NULL) - # в радиусе 2× от anchor ИЛИ только что scraped (последние 2 часа). - # OR обёрнут в скобки — приоритет AND выше OR. - priority_rows = db.execute( - text(""" - SELECT source_url - FROM listings - WHERE source = 'avito' - AND source_url IS NOT NULL - AND ( - ( - detail_enriched_at IS NULL - AND price_rub > 0 - AND ST_DWithin( - geom::geography, - ST_MakePoint(:lon, :lat)::geography, - :radius + # ── Step 3: group by house ────────────────────────────── + unique_house_paths: set[str] = set() + for lot in lots: + if lot.house_url: + try: + parsed = urlparse(lot.house_url) + path = parsed.path if parsed.path else lot.house_url + unique_house_paths.add(path) + except Exception: + continue + + counters.unique_houses = len(unique_house_paths) + logger.info("pipeline:group_by_house unique=%d", counters.unique_houses) + + # ── Step 4: enrich houses с shared session + sleep ────── + if enrich_houses and unique_house_paths: + house_paths_list = list(unique_house_paths) + for idx, house_path in enumerate(house_paths_list): + try: + enrichment = await fetch_house_catalog(house_path, cffi_session=session) + save_house_catalog_enrichment(db, enrichment) + counters.houses_enriched += 1 + except (AvitoBlockedError, AvitoRateLimitedError): + logger.error( + "pipeline:house BLOCKED at %s — propagating", house_path + ) + raise + except Exception as e: + logger.warning( + "pipeline:house_enrich failed for %s: %s", house_path, e + ) + counters.houses_failed += 1 + counters.errors.append(f"house {house_path}: {e}") + + if idx < len(house_paths_list) - 1: + await asyncio.sleep(detail_delay) + + # ── Step 5: enrich detail для top-N priority listings ─── + if enrich_detail_top_n > 0: + priority_rows = db.execute( + text(""" + SELECT source_url + FROM listings + WHERE source = 'avito' + AND source_url IS NOT NULL + AND ( + ( + detail_enriched_at IS NULL + AND price_rub > 0 + AND ST_DWithin( + geom::geography, + ST_MakePoint(:lon, :lat)::geography, + :radius + ) + ) + OR ( + detail_enriched_at IS NULL + AND scraped_at > NOW() - INTERVAL '2 hours' + ) ) + ORDER BY scraped_at DESC NULLS LAST + LIMIT :limit + """), + { + "lat": lat, + "lon": lon, + "radius": radius_m * 2, + "limit": enrich_detail_top_n, + }, + ).mappings().all() + + consecutive_blocks = 0 + for idx, row in enumerate(priority_rows): + source_url: str = row["source_url"] + counters.detail_attempted += 1 + try: + item_url = ( + urlparse(source_url).path + if source_url.startswith("http") + else source_url ) - OR ( - detail_enriched_at IS NULL - AND scraped_at > NOW() - INTERVAL '2 hours' + enrichment_detail = await fetch_detail(item_url, cffi_session=session) + if save_detail_enrichment(db, enrichment_detail): + counters.detail_enriched += 1 + consecutive_blocks = 0 + except (AvitoBlockedError, AvitoRateLimitedError) as e: + consecutive_blocks += 1 + counters.detail_failed += 1 + counters.errors.append(f"BLOCKED_DETAIL #{idx}: {e}") + logger.warning( + "pipeline:detail BLOCKED #%d/%d (consecutive=%d): %s", + idx + 1, len(priority_rows), consecutive_blocks, e, ) - ) - ORDER BY scraped_at DESC NULLS LAST - LIMIT :limit - """), - { - "lat": lat, - "lon": lon, - "radius": radius_m * 2, - "limit": enrich_detail_top_n, - }, - ).mappings().all() + if consecutive_blocks >= 3: + logger.error( + "pipeline:detail ABORT — %d consecutive blocks, IP rate-limited", + consecutive_blocks, + ) + counters.errors.append( + f"AVITO_RATE_LIMITED — sweep aborted after " + f"{idx + 1}/{len(priority_rows)} details" + ) + raise + except Exception as e: + counters.detail_failed += 1 + counters.errors.append(f"detail {source_url}: {e}") + logger.warning( + "pipeline:detail_enrich failed for %s: %s", source_url, e + ) + consecutive_blocks = 0 - for row in priority_rows: - source_url: str = row["source_url"] - counters.detail_attempted += 1 - try: - # fetch_detail принимает absolute URL или relative path - item_url = ( - urlparse(source_url).path - if source_url.startswith("http") - else source_url - ) - enrichment_detail = await fetch_detail(item_url) - if save_detail_enrichment(db, enrichment_detail): - counters.detail_enriched += 1 - except Exception as e: - logger.warning( - "pipeline:detail_enrich failed for %s: %s", source_url, e - ) - counters.detail_failed += 1 - counters.errors.append(f"detail {source_url}: {e}") + if idx < len(priority_rows) - 1: + jitter = random.uniform(0.8, 1.2) + await asyncio.sleep(detail_delay * jitter) - logger.info( - "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) " - "houses=%d/%d detail=%d/%d errors=%d", - lat, - lon, - counters.lots_fetched, - counters.lots_inserted, - counters.lots_updated, - counters.houses_enriched, - counters.unique_houses, - counters.detail_enriched, - counters.detail_attempted, - len(counters.errors), - ) + logger.info( + "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) " + "houses=%d/%d detail=%d/%d errors=%d", + lat, lon, + counters.lots_fetched, counters.lots_inserted, counters.lots_updated, + counters.houses_enriched, counters.unique_houses, + counters.detail_enriched, counters.detail_attempted, + len(counters.errors), + ) + return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n) - return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n) + finally: + if own_session: + await session.close() @dataclass @@ -269,91 +328,103 @@ async def run_avito_city_sweep( radius_m: int = 1500, pages_per_anchor: int = 3, enrich_houses: bool = True, - detail_top_n: int = 20, + detail_top_n: int = 10, request_delay_sec: float = 7.0, ) -> CitySweepCounters: """Full city sweep: iterate anchors × pages → save → enrich houses + detail. - Алгоритм: - - Для каждого anchor: run_avito_pipeline(pages=pages_per_anchor) - - Перед каждым anchor проверяет scrape_runs.status — если 'cancelled' → stop - - Heartbeat update каждый anchor с текущими counters - - Errors per-anchor логируются, не валят весь sweep - - По завершению вызывает mark_done; при исключении — mark_failed + - Single shared AsyncSession на весь sweep (один TLS fingerprint) + - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned') + - Прочие errors per-anchor логируются, не валят весь sweep """ - from app.services import scrape_runs as runs - _anchors = anchors if anchors is not None else EKB_ANCHORS counters = CitySweepCounters(anchors_total=len(_anchors)) - try: - for idx, (lat, lon, name) in enumerate(_anchors, start=1): - # Cooperative cancel: проверяем статус перед каждым anchor - if runs.is_cancelled(db, run_id): + async with AsyncSession( + impersonate="chrome120", + timeout=25, + headers=_CHROME_HEADERS, + ) as session: + try: + for idx, (lat, lon, name) in enumerate(_anchors, start=1): + if scrape_runs.is_cancelled(db, run_id): + logger.info( + "city-sweep run_id=%d: cancelled at anchor #%d/%d (%s)", + run_id, idx, len(_anchors), name, + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + return counters + logger.info( - "city-sweep run_id=%d: cancelled by signal at anchor #%d/%d (%s)", - run_id, idx, len(_anchors), name, + "city-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", + run_id, idx, len(_anchors), name, lat, lon, ) - runs.update_heartbeat(db, run_id, counters.to_dict()) - return counters + try: + result = await run_avito_pipeline( + db, + lat=lat, + lon=lon, + radius_m=radius_m, + enrich_houses=enrich_houses, + enrich_detail_top_n=detail_top_n, + pages=pages_per_anchor, + request_delay_sec=request_delay_sec, + shared_session=session, + ) + counters.lots_fetched += result.counters.lots_fetched + counters.lots_inserted += result.counters.lots_inserted + counters.lots_updated += result.counters.lots_updated + counters.unique_houses += result.counters.unique_houses + counters.houses_enriched += result.counters.houses_enriched + counters.houses_failed += result.counters.houses_failed + counters.detail_attempted += result.counters.detail_attempted + counters.detail_enriched += result.counters.detail_enriched + counters.detail_failed += result.counters.detail_failed + counters.errors_count += len(result.counters.errors) + except (AvitoBlockedError, AvitoRateLimitedError) as e: + logger.error( + "city-sweep run_id=%d ABORT at anchor #%d/%d (%s) — blocked: %s", + run_id, idx, len(_anchors), name, e, + ) + counters.errors_count += 1 + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict()) + return counters + except Exception: + logger.exception( + "city-sweep run_id=%d: anchor %s failed", run_id, name + ) + counters.errors_count += 1 + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + + scrape_runs.mark_done(db, run_id, counters.to_dict()) logger.info( - "city-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", - run_id, idx, len(_anchors), name, lat, lon, + "city-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " + "houses=%d/%d detail=%d/%d errors=%d", + run_id, + counters.anchors_done, + counters.anchors_total, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.houses_enriched, + counters.unique_houses, + counters.detail_enriched, + counters.detail_attempted, + counters.errors_count, ) - try: - result = await run_avito_pipeline( - db, - lat=lat, - lon=lon, - radius_m=radius_m, - enrich_houses=enrich_houses, - enrich_detail_top_n=detail_top_n, - pages=pages_per_anchor, - request_delay_sec=request_delay_sec, - ) - counters.lots_fetched += result.counters.lots_fetched - counters.lots_inserted += result.counters.lots_inserted - counters.lots_updated += result.counters.lots_updated - counters.unique_houses += result.counters.unique_houses - counters.houses_enriched += result.counters.houses_enriched - counters.houses_failed += result.counters.houses_failed - counters.detail_attempted += result.counters.detail_attempted - counters.detail_enriched += result.counters.detail_enriched - counters.detail_failed += result.counters.detail_failed - counters.errors_count += len(result.counters.errors) - except Exception: - logger.exception("city-sweep run_id=%d: anchor %s failed", run_id, name) - counters.errors_count += 1 + return counters - counters.anchors_done = idx - runs.update_heartbeat(db, run_id, counters.to_dict()) - - runs.mark_done(db, run_id, counters.to_dict()) - logger.info( - "city-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " - "houses=%d/%d detail=%d/%d errors=%d", - run_id, - counters.anchors_done, - counters.anchors_total, - counters.lots_fetched, - counters.lots_inserted, - counters.lots_updated, - counters.houses_enriched, - counters.unique_houses, - counters.detail_enriched, - counters.detail_attempted, - counters.errors_count, - ) - return counters - - except Exception as exc: - logger.exception("city-sweep run_id=%d: fatal error", run_id) - runs.mark_failed(db, run_id, str(exc), counters.to_dict()) - raise + except Exception as exc: + logger.exception("city-sweep run_id=%d: fatal error", run_id) + scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise -# ── Public re-exports ─────────────────────────────────────────────────────── +# ── Public re-exports ─────────────────────────────────────────── __all__ = [ "EKB_ANCHORS", "CitySweepCounters", diff --git a/tradein-mvp/backend/app/services/scrape_runs.py b/tradein-mvp/backend/app/services/scrape_runs.py index ffbf9a51..697a6daf 100644 --- a/tradein-mvp/backend/app/services/scrape_runs.py +++ b/tradein-mvp/backend/app/services/scrape_runs.py @@ -63,33 +63,77 @@ def is_cancelled(db: Session, run_id: int) -> bool: def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None: """Финализация run: status='done', finished_at=NOW(), counters.""" - db.execute( + row = db.execute( text( """ UPDATE scrape_runs SET status = 'done', finished_at = NOW(), heartbeat_at = NOW(), counters = CAST(:counters AS jsonb) WHERE id = :run_id AND status = 'running' + RETURNING id """ ), {"run_id": run_id, "counters": json.dumps(counters)}, - ) + ).first() + if row is None: + logger.warning("mark_done no-op: run_id=%d not in 'running' state", run_id) db.commit() def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None: - """Финализация run: status='failed', error (первые 1000 символов).""" - db.execute( + """Финализация run: status='failed', error (первые 1000 символов). + + Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE, + он мог оставить сессию в error state — rollback сбрасывает состояние. + """ + try: + db.rollback() # defensive: cascade-safe if prior UPDATE left txn in error state + except Exception: + pass + row = db.execute( text( """ UPDATE scrape_runs SET status = 'failed', finished_at = NOW(), heartbeat_at = NOW(), error = :error, counters = CAST(:counters AS jsonb) WHERE id = :run_id AND status = 'running' + RETURNING id """ ), {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)}, - ) + ).first() + if row is None: + logger.warning("mark_failed no-op: run_id=%d not in 'running' state", run_id) + db.commit() + + +def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None: + """Финализация run: status='banned' (IP заблокирован Avito — 403/captcha). + + Per migration 015 — 'banned' задокументирован как 'Avito вернул 403/captcha'. + Отличается от 'failed': это external constraint, не наш bug. Cooldown 2-4 часа. + + Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE, + он мог оставить сессию в error state — rollback сбрасывает состояние. + """ + try: + db.rollback() # defensive: cascade-safe if prior UPDATE left txn in error state + except Exception: + pass + row = db.execute( + text( + """ + UPDATE scrape_runs + SET status = 'banned', finished_at = NOW(), heartbeat_at = NOW(), + error = :error, counters = CAST(:counters AS jsonb) + WHERE id = :run_id AND status = 'running' + RETURNING id + """ + ), + {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)}, + ).first() + if row is None: + logger.warning("mark_banned no-op: run_id=%d not in 'running' state", run_id) db.commit() diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index 6530ab22..c6e19f0e 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/tradein-mvp/backend/app/services/scrapers/avito.py @@ -24,6 +24,7 @@ from curl_cffi.requests import AsyncSession from selectolax.parser import HTMLParser from app.services.scraper_settings import get_scraper_delay +from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError from app.services.scrapers.base import BaseScraper, ScrapedLot logger = logging.getLogger(__name__) @@ -101,11 +102,27 @@ class AvitoScraper(BaseScraper): try: assert self._cffi is not None response = await self._cffi.get(url) + if response.status_code == 403: + logger.error( + "avito SERP HTTP 403 (IP blocked) page=%d url=%s", page, url + ) + raise AvitoBlockedError( + f"Avito SERP returned 403 — IP blocked at page={page}" + ) + if response.status_code == 429: + logger.error( + "avito SERP HTTP 429 (rate limited) page=%d url=%s", page, url + ) + raise AvitoRateLimitedError( + f"Avito SERP returned 429 — rate limited at page={page}" + ) if response.status_code != 200: logger.warning( "avito HTML page=%d returned %d for %s", page, response.status_code, url ) break + except (AvitoBlockedError, AvitoRateLimitedError): + raise except Exception: logger.exception("avito HTML page=%d fetch failed for %s", page, url) break diff --git a/tradein-mvp/backend/app/services/scrapers/avito_detail.py b/tradein-mvp/backend/app/services/scrapers/avito_detail.py index 274485b4..3ac657bd 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito_detail.py +++ b/tradein-mvp/backend/app/services/scrapers/avito_detail.py @@ -32,6 +32,8 @@ from selectolax.parser import HTMLParser, Node from sqlalchemy import text from sqlalchemy.orm import Session +from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError + logger = logging.getLogger(__name__) AVITO_BASE = "https://www.avito.ru" @@ -252,6 +254,10 @@ async def fetch_detail( full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url) try: response = await session.get(full_url) + if response.status_code == 403: + raise AvitoBlockedError(f"Avito detail HTTP 403 for {full_url}") + if response.status_code == 429: + raise AvitoRateLimitedError(f"Avito detail HTTP 429 for {full_url}") if response.status_code != 200: raise ValueError( f"avito detail HTTP {response.status_code} for {full_url}" diff --git a/tradein-mvp/backend/app/services/scrapers/avito_exceptions.py b/tradein-mvp/backend/app/services/scrapers/avito_exceptions.py new file mode 100644 index 00000000..0ea56350 --- /dev/null +++ b/tradein-mvp/backend/app/services/scrapers/avito_exceptions.py @@ -0,0 +1,17 @@ +"""Avito-specific exceptions для anti-bot detection.""" + + +class AvitoError(Exception): + """Base для всех Avito scraper exceptions.""" + + +class AvitoBlockedError(AvitoError): + """HTTP 403 от Avito — IP-level block detected.""" + + +class AvitoRateLimitedError(AvitoError): + """HTTP 429 от Avito — rate limit triggered.""" + + +class AvitoParseError(AvitoError): + """Cannot parse Avito HTML structure (selector changes).""" diff --git a/tradein-mvp/backend/app/services/scrapers/avito_houses.py b/tradein-mvp/backend/app/services/scrapers/avito_houses.py index 6bed1445..743e8619 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito_houses.py +++ b/tradein-mvp/backend/app/services/scrapers/avito_houses.py @@ -32,6 +32,8 @@ from typing import Any from sqlalchemy import text from sqlalchemy.orm import Session +from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError + logger = logging.getLogger(__name__) AVITO_BASE = "https://www.avito.ru" @@ -621,6 +623,10 @@ async def fetch_house_catalog( logger.info("Houses Catalog fetch: %s", url) resp = await cffi_session.get(url) + if resp.status_code == 403: + raise AvitoBlockedError(f"Avito houses HTTP 403 for {url}") + if resp.status_code == 429: + raise AvitoRateLimitedError(f"Avito houses HTTP 429 for {url}") resp.raise_for_status() html: str = resp.text diff --git a/tradein-mvp/backend/tests/scrapers/__init__.py b/tradein-mvp/backend/tests/scrapers/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py b/tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py new file mode 100644 index 00000000..0e64073d --- /dev/null +++ b/tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py @@ -0,0 +1,229 @@ +"""Test anti-bot hardening: blocked exceptions + early abort.""" + +from unittest.mock import AsyncMock, MagicMock, call, patch + +import pytest + +from app.services.scrapers.avito_exceptions import ( + AvitoBlockedError, + AvitoRateLimitedError, +) + +# ── fetch_detail tests ────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_fetch_detail_raises_blocked_on_403() -> None: + """fetch_detail должен raise AvitoBlockedError на HTTP 403.""" + from app.services.scrapers.avito_detail import fetch_detail + + mock_session = AsyncMock() + mock_response = MagicMock() + mock_response.status_code = 403 + mock_session.get = AsyncMock(return_value=mock_response) + + with pytest.raises(AvitoBlockedError, match="403"): + await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session) + + +@pytest.mark.asyncio +async def test_fetch_detail_raises_rate_limited_on_429() -> None: + """fetch_detail должен raise AvitoRateLimitedError на HTTP 429.""" + from app.services.scrapers.avito_detail import fetch_detail + + mock_session = AsyncMock() + mock_response = MagicMock() + mock_response.status_code = 429 + mock_session.get = AsyncMock(return_value=mock_response) + + with pytest.raises(AvitoRateLimitedError, match="429"): + await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session) + + +@pytest.mark.asyncio +async def test_fetch_detail_raises_value_error_on_other_non_200() -> None: + """fetch_detail должен raise ValueError на прочие non-200 (не 403/429).""" + from app.services.scrapers.avito_detail import fetch_detail + + mock_session = AsyncMock() + mock_response = MagicMock() + mock_response.status_code = 503 + mock_session.get = AsyncMock(return_value=mock_response) + + with pytest.raises(ValueError, match="503"): + await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session) + + +# ── SERP (avito.py) tests ─────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_fetch_around_raises_blocked_on_403() -> None: + """AvitoScraper.fetch_around должен raise AvitoBlockedError на HTTP 403.""" + from app.services.scrapers.avito import AvitoScraper + + scraper = AvitoScraper() + mock_session = AsyncMock() + mock_response = MagicMock() + mock_response.status_code = 403 + mock_session.get = AsyncMock(return_value=mock_response) + scraper._cffi = mock_session + + with pytest.raises(AvitoBlockedError): + await scraper.fetch_around(56.84, 60.60, 1000, pages=1) + + +@pytest.mark.asyncio +async def test_fetch_around_raises_rate_limited_on_429() -> None: + """AvitoScraper.fetch_around должен raise AvitoRateLimitedError на HTTP 429.""" + from app.services.scrapers.avito import AvitoScraper + + scraper = AvitoScraper() + mock_session = AsyncMock() + mock_response = MagicMock() + mock_response.status_code = 429 + mock_session.get = AsyncMock(return_value=mock_response) + scraper._cffi = mock_session + + with pytest.raises(AvitoRateLimitedError): + await scraper.fetch_around(56.84, 60.60, 1000, pages=1) + + +# ── Pipeline abort tests ──────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_pipeline_aborts_after_3_consecutive_blocks() -> None: + """Pipeline должен прервать sweep после 3 consecutive 403 (re-raise).""" + from app.services.scrape_pipeline import run_avito_pipeline + + mock_db = MagicMock() + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper.fetch_around = AsyncMock(return_value=[]) + mock_scraper._cffi = None + + mock_row = MagicMock() + mock_row.__getitem__ = lambda self, key: ( + "https://www.avito.ru/ekaterinburg/kvartiry/test-1" if key == "source_url" else None + ) + mock_mappings = MagicMock() + mock_mappings.all.return_value = [mock_row, mock_row, mock_row, mock_row] + mock_execute = MagicMock() + mock_execute.mappings.return_value = mock_mappings + mock_db.execute.return_value = mock_execute + + block_error = AvitoBlockedError("HTTP 403") + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper), + patch( + "app.services.scrape_pipeline.fetch_detail", + AsyncMock(side_effect=block_error), + ), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch( + "curl_cffi.requests.AsyncSession", + return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()), + ), + ): + with pytest.raises(AvitoBlockedError): + await run_avito_pipeline( + mock_db, 56.8, 60.6, 1000, + enrich_houses=False, + enrich_detail_top_n=4, + ) + + +@pytest.mark.asyncio +async def test_pipeline_sleep_between_detail_requests() -> None: + """Pipeline должен вызывать asyncio.sleep между каждым detail request.""" + from app.services.scrape_pipeline import run_avito_pipeline + + mock_db = MagicMock() + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper.fetch_around = AsyncMock(return_value=[]) + mock_scraper._cffi = None + + mock_row = MagicMock() + mock_row.__getitem__ = lambda self, key: ( + "https://www.avito.ru/ekaterinburg/kvartiry/test-1" if key == "source_url" else None + ) + mock_mappings = MagicMock() + mock_mappings.all.return_value = [mock_row, mock_row, mock_row] + mock_execute = MagicMock() + mock_execute.mappings.return_value = mock_mappings + mock_db.execute.return_value = mock_execute + + mock_enrichment = MagicMock() + sleep_mock = AsyncMock() + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper), + patch( + "app.services.scrape_pipeline.fetch_detail", + AsyncMock(return_value=mock_enrichment), + ), + patch("app.services.scrape_pipeline.save_detail_enrichment", return_value=True), + patch("app.services.scrape_pipeline.asyncio.sleep", sleep_mock), + patch( + "curl_cffi.requests.AsyncSession", + return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()), + ), + patch("app.services.scrape_pipeline.random.uniform", return_value=1.0), + ): + await run_avito_pipeline( + mock_db, 56.8, 60.6, 1000, + enrich_houses=False, + enrich_detail_top_n=3, + request_delay_sec=7.0, + ) + + assert sleep_mock.call_count == 2 + for c in sleep_mock.call_args_list: + assert c == call(7.0) + + +@pytest.mark.asyncio +async def test_city_sweep_marks_banned_on_block() -> None: + """City sweep mark'ит run как banned при AvitoBlockedError. + + Migration 015 задокументировал 'banned' как 'Avito вернул 403/captcha' — именно этот сценарий. + """ + from app.services.scrape_pipeline import run_avito_city_sweep + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.update_heartbeat = MagicMock() + mock_runs.mark_banned = MagicMock() + mock_runs.mark_done = MagicMock() + + block_error = AvitoBlockedError("IP blocked at page=1") + + with ( + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch( + "app.services.scrape_pipeline.run_avito_pipeline", + AsyncMock(side_effect=block_error), + ), + patch( + "curl_cffi.requests.AsyncSession", + return_value=AsyncMock( + __aenter__=AsyncMock(return_value=AsyncMock()), + __aexit__=AsyncMock(return_value=None), + ), + ), + ): + result = await run_avito_city_sweep( + mock_db, + run_id=42, + anchors=[(56.84, 60.60, "Центр"), (56.79, 60.53, "ЮЗ")], + ) + + assert mock_runs.mark_banned.called + assert not mock_runs.mark_done.called + assert result.anchors_done == 1