bot-backend
dc7a84cf2e
run_{avito,cian,yandex}_city_sweep had no per-anchor watchdog: a stalled HTTP
request (proxy/connection hang) blocked the asyncio task forever — heartbeat
froze mid-anchor, run stayed 'running' until the 6h zombie-detect. Live-repro:
cian sweep #24 hung on anchor 3, #29 on anchor 1.
Wrap each anchor's processing in asyncio.wait_for(coro, ANCHOR_TIMEOUT_SEC=240).
TimeoutError -> logger.warning + errors_count++ + continue (sweep finishes,
reaches mark_done). Timeout branch sits before the existing except Exception.
yandex/cian also bump consecutive_failures -> existing MAX_CONSECUTIVE_FAILURES
abort fires on 3 consecutive hung anchors. CianScraper SERP session already
has timeout=30. 5 unit tests (all 3 sweeps), network-free.
Refs #880, #581, #559
1507 lines
66 KiB
Python
1507 lines
66 KiB
Python
"""scrape_pipeline.py — Avito full orchestrator (Stage 2e + city sweep).
|
||
|
||
Координирует 5 шагов одного pipeline run:
|
||
1. SEARCH — AvitoScraper().fetch_around(lat, lon, radius_m, pages=N)
|
||
2. SAVE — save_listings(db, lots) → (inserted, updated)
|
||
3. GROUP — dedupe house_url из lots → unique set
|
||
4. ENRICH_HOUSES — fetch_house_catalog + save_house_catalog_enrichment per unique house
|
||
5. ENRICH_DETAIL — top-N listings (detail_enriched_at IS NULL) → fetch_detail + save
|
||
|
||
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 (кроме blocked exceptions — они abort весь sweep).
|
||
|
||
IMV-фаза (финальная, после всех anchor'ов): если enrich_imv=True —
|
||
для домов, тронутых в этом sweep'е (touched house_id из houses-enrichment),
|
||
вызывается house_imv_backfill.process_houses_imv_batch. Ошибки per-house
|
||
логируются и считаются, но не валят sweep.
|
||
"""
|
||
|
||
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.core.config import settings
|
||
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
|
||
from app.services.scrapers.n1 import N1Scraper
|
||
from app.services.scrapers.yandex_realty import YandexRealtyScraper
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает
|
||
# дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for
|
||
# прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом.
|
||
# Диапазон 180-300 с; 240 — разумный середина (4 мин > любой нормальный anchor).
|
||
ANCHOR_TIMEOUT_SEC: int = 240
|
||
|
||
# Default anchors ЕКБ — 5 точек покрытия города
|
||
EKB_ANCHORS: list[tuple[float, float, str]] = [
|
||
(56.8400, 60.6050, "Центр"),
|
||
(56.7950, 60.5300, "ЮЗ"),
|
||
(56.8970, 60.6100, "Уралмаш"),
|
||
(56.7700, 60.5500, "Академический"),
|
||
(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",
|
||
}
|
||
|
||
|
||
def _avito_proxies() -> dict[str, str] | None:
|
||
"""#623/#806: mobile-proxy egress для shared-сессии sweep'а.
|
||
|
||
КРИТИЧНО: sweep создаёт собственную shared AsyncSession и присваивает её
|
||
`scraper._cffi`, перезатирая проксированную сессию из `AvitoScraper.__aenter__`.
|
||
Через эту же сессию идёт и SERP, и detail-энричмент (fetch_detail). Без прокси
|
||
detail-страницы летят с datacenter-IP → HTTP 429 → весь run mark_banned.
|
||
Пусто (env не задан) → прямое подключение (dev/no-op).
|
||
|
||
Читает settings.scraper_proxy_url (SCRAPER_PROXY_URL > AVITO_PROXY_URL fallback).
|
||
"""
|
||
url = settings.scraper_proxy_url
|
||
return {"http": url, "https": url} if url else None
|
||
|
||
|
||
# ── Result dataclasses ──────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class PipelineCounters:
|
||
"""Counters одного pipeline run для логов/админки."""
|
||
|
||
lots_fetched: int = 0
|
||
lots_inserted: int = 0
|
||
lots_updated: int = 0
|
||
unique_houses: int = 0
|
||
houses_enriched: int = 0
|
||
houses_failed: int = 0
|
||
detail_attempted: int = 0
|
||
detail_enriched: int = 0
|
||
detail_failed: int = 0
|
||
errors: list[str] = field(default_factory=list)
|
||
|
||
|
||
@dataclass
|
||
class PipelineResult:
|
||
"""Полный результат full Avito pipeline."""
|
||
|
||
anchor_lat: float
|
||
anchor_lon: float
|
||
radius_m: int
|
||
counters: PipelineCounters
|
||
enrich_houses: bool
|
||
enrich_detail_top_n: int
|
||
touched_house_ids: set[int] = field(default_factory=set)
|
||
|
||
|
||
# ── Main orchestrator ───────────────────────────────────────────
|
||
|
||
|
||
async def run_avito_pipeline(
|
||
db: Session,
|
||
lat: float,
|
||
lon: float,
|
||
radius_m: int = 1500,
|
||
*,
|
||
enrich_houses: bool = True,
|
||
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.
|
||
|
||
Steps:
|
||
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: fetch_house_catalog + save per house (with sleep between)
|
||
5. ENRICH_DETAIL: top-N listings → fetch_detail + save (abort on 3 blocks)
|
||
|
||
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
|
||
|
||
own_session = shared_session is None
|
||
session = shared_session or AsyncSession(
|
||
impersonate="chrome120",
|
||
timeout=25,
|
||
headers=_CHROME_HEADERS,
|
||
proxies=_avito_proxies(),
|
||
)
|
||
|
||
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,
|
||
)
|
||
counters.lots_fetched = len(lots)
|
||
logger.info(
|
||
"pipeline:search anchor=(%.4f,%.4f) r=%dm fetched=%d",
|
||
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)
|
||
|
||
# ── 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 ──────────────────────────────
|
||
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 ──────
|
||
touched_house_ids: set[int] = set()
|
||
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)
|
||
house_counters = save_house_catalog_enrichment(db, enrichment)
|
||
counters.houses_enriched += 1
|
||
hid = house_counters.get("house_id")
|
||
if hid:
|
||
touched_house_ids.add(int(hid))
|
||
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
|
||
detail_house_paths: set[str] = set()
|
||
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
|
||
)
|
||
enrichment_detail = await fetch_detail(item_url, cffi_session=session)
|
||
if save_detail_enrichment(db, enrichment_detail):
|
||
counters.detail_enriched += 1
|
||
# #871: house-link есть в detail SSR (в SERP-карточке — JS-only).
|
||
# Копим catalog-пути для house-enrich ниже (Step 5b).
|
||
if enrichment_detail.house_catalog_url:
|
||
hp = urlparse(enrichment_detail.house_catalog_url).path
|
||
if hp:
|
||
detail_house_paths.add(hp)
|
||
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,
|
||
)
|
||
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
|
||
|
||
if idx < len(priority_rows) - 1:
|
||
jitter = random.uniform(0.8, 1.2)
|
||
await asyncio.sleep(detail_delay * jitter)
|
||
|
||
# ── Step 5b: enrich houses, найденные через detail-страницы (#871) ──
|
||
# SERP-карточки больше не несут house-link (JS-render) → дома берём из
|
||
# detail SSR. Дедуп против обработанных в Step 4 (unique_house_paths).
|
||
new_house_paths = detail_house_paths - unique_house_paths
|
||
if enrich_houses and new_house_paths:
|
||
counters.unique_houses += len(new_house_paths)
|
||
nh_list = list(new_house_paths)
|
||
for h_idx, house_path in enumerate(nh_list):
|
||
try:
|
||
enrichment = await fetch_house_catalog(house_path, cffi_session=session)
|
||
house_counters = save_house_catalog_enrichment(db, enrichment)
|
||
counters.houses_enriched += 1
|
||
hid = house_counters.get("house_id")
|
||
if hid:
|
||
touched_house_ids.add(int(hid))
|
||
except (AvitoBlockedError, AvitoRateLimitedError):
|
||
logger.error(
|
||
"pipeline:house(detail) BLOCKED at %s — propagating", house_path
|
||
)
|
||
raise
|
||
except Exception as e:
|
||
logger.warning(
|
||
"pipeline:house(detail) enrich failed for %s: %s", house_path, e
|
||
)
|
||
counters.houses_failed += 1
|
||
counters.errors.append(f"house(detail) {house_path}: {e}")
|
||
if h_idx < len(nh_list) - 1:
|
||
await asyncio.sleep(detail_delay)
|
||
|
||
logger.info(
|
||
"pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) "
|
||
"houses=%d/%d detail=%d/%d touched_house_ids=%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(touched_house_ids),
|
||
len(counters.errors),
|
||
)
|
||
return PipelineResult(
|
||
lat,
|
||
lon,
|
||
radius_m,
|
||
counters,
|
||
enrich_houses,
|
||
enrich_detail_top_n,
|
||
touched_house_ids=touched_house_ids,
|
||
)
|
||
|
||
finally:
|
||
if own_session:
|
||
await session.close()
|
||
|
||
|
||
@dataclass
|
||
class CitySweepCounters:
|
||
"""Aggregate counters для full city sweep run."""
|
||
|
||
anchors_total: int = 0
|
||
anchors_done: int = 0
|
||
lots_fetched: int = 0
|
||
lots_inserted: int = 0
|
||
lots_updated: int = 0
|
||
unique_houses: int = 0
|
||
houses_enriched: int = 0
|
||
houses_failed: int = 0
|
||
detail_attempted: int = 0
|
||
detail_enriched: int = 0
|
||
detail_failed: int = 0
|
||
imv_attempted: int = 0
|
||
imv_enriched: int = 0
|
||
imv_failed: int = 0
|
||
errors_count: int = 0
|
||
|
||
def to_dict(self) -> dict[str, int]:
|
||
return {f.name: getattr(self, f.name) for f in fields(self)}
|
||
|
||
|
||
async def run_avito_city_sweep(
|
||
db: Session,
|
||
*,
|
||
run_id: int,
|
||
anchors: list[tuple[float, float, str]] | None = None,
|
||
radius_m: int = 1500,
|
||
pages_per_anchor: int = 3,
|
||
enrich_houses: bool = True,
|
||
detail_top_n: int = 10,
|
||
request_delay_sec: float = 7.0,
|
||
enrich_imv: bool = True,
|
||
) -> CitySweepCounters:
|
||
"""Full city sweep: iterate anchors × pages → save → enrich houses + detail → IMV.
|
||
|
||
- Single shared AsyncSession на весь sweep (один TLS fingerprint)
|
||
- AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned')
|
||
- Прочие errors per-anchor логируются, не валят весь sweep
|
||
- После всех anchor'ов: если enrich_imv=True — IMV-оценка тронутых домов
|
||
(cooperative cancel + per-house graceful error handling)
|
||
"""
|
||
from app.services.house_imv_backfill import process_houses_imv_batch
|
||
|
||
_anchors = anchors if anchors is not None else EKB_ANCHORS
|
||
counters = CitySweepCounters(anchors_total=len(_anchors))
|
||
all_touched_house_ids: set[int] = set()
|
||
|
||
async with AsyncSession(
|
||
impersonate="chrome120",
|
||
timeout=25,
|
||
headers=_CHROME_HEADERS,
|
||
proxies=_avito_proxies(),
|
||
) 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 anchor #%d/%d (%s, %.4f, %.4f)",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
lat,
|
||
lon,
|
||
)
|
||
try:
|
||
result = await asyncio.wait_for(
|
||
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,
|
||
),
|
||
timeout=ANCHOR_TIMEOUT_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)
|
||
all_touched_house_ids.update(result.touched_house_ids)
|
||
except TimeoutError:
|
||
logger.warning(
|
||
"city-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
|
||
"timed out after %ds — skipping",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
lat,
|
||
lon,
|
||
ANCHOR_TIMEOUT_SEC,
|
||
)
|
||
counters.errors_count += 1
|
||
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())
|
||
|
||
# ── IMV-фаза: финальный обход тронутых домов ──────────
|
||
if enrich_imv and all_touched_house_ids:
|
||
if scrape_runs.is_cancelled(db, run_id):
|
||
logger.info(
|
||
"city-sweep run_id=%d: cancelled before IMV phase (%d houses skipped)",
|
||
run_id,
|
||
len(all_touched_house_ids),
|
||
)
|
||
scrape_runs.mark_done(db, run_id, counters.to_dict())
|
||
return counters
|
||
|
||
logger.info(
|
||
"city-sweep run_id=%d: IMV phase — %d touched houses",
|
||
run_id,
|
||
len(all_touched_house_ids),
|
||
)
|
||
try:
|
||
imv_result = await process_houses_imv_batch(
|
||
db,
|
||
all_touched_house_ids,
|
||
request_delay_sec=request_delay_sec,
|
||
)
|
||
counters.imv_attempted += imv_result.checked
|
||
counters.imv_enriched += imv_result.saved
|
||
counters.imv_failed += imv_result.errors
|
||
counters.errors_count += imv_result.errors
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
logger.info(
|
||
"city-sweep run_id=%d: IMV phase done — attempted=%d enriched=%d failed=%d",
|
||
run_id,
|
||
imv_result.checked,
|
||
imv_result.saved,
|
||
imv_result.errors,
|
||
)
|
||
except Exception as exc:
|
||
logger.error(
|
||
"city-sweep run_id=%d: IMV phase crashed — %r (sweep still done)",
|
||
run_id,
|
||
exc,
|
||
)
|
||
counters.errors_count += 1
|
||
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 done: anchors=%d/%d lots=%d (ins=%d/upd=%d) "
|
||
"houses=%d/%d detail=%d/%d imv=%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.imv_enriched,
|
||
counters.imv_attempted,
|
||
counters.errors_count,
|
||
)
|
||
return counters
|
||
|
||
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
|
||
|
||
|
||
# ── Yandex city sweep ───────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class YandexCitySweepCounters:
|
||
"""Aggregate counters для Yandex.Недвижимость city sweep run.
|
||
|
||
Фазы: SERP (fetch_around_multi_room) → save_listings → address-enrich.
|
||
address_* — результат T10-обогащения (detail <title> → полный адрес с номером дома).
|
||
"""
|
||
|
||
anchors_total: int = 0
|
||
anchors_done: int = 0
|
||
lots_fetched: int = 0
|
||
lots_inserted: int = 0
|
||
lots_updated: int = 0
|
||
address_attempted: int = 0
|
||
address_enriched: int = 0
|
||
address_failed: int = 0
|
||
errors_count: int = 0
|
||
|
||
def to_dict(self) -> dict[str, int]:
|
||
return {f.name: getattr(self, f.name) for f in fields(self)}
|
||
|
||
|
||
# Defensive abort threshold: у YandexRealtyScraper нет block/rate-exception классов
|
||
# (fetch_around глотает ошибки и возвращает []). Чтобы не молотить Yandex впустую при
|
||
# системном блоке/бане IP, прерываем sweep после N подряд anchor'ов, упавших с
|
||
# исключением. NB: пустой результат (0 lots) — это НЕ ошибка, счётчик не растёт.
|
||
YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES = 3
|
||
|
||
|
||
async def run_yandex_city_sweep(
|
||
db: Session,
|
||
*,
|
||
run_id: int,
|
||
anchors: list[tuple[float, float, str]] | None = None,
|
||
radius_m: int = 1500,
|
||
pages_per_anchor: int = 2,
|
||
request_delay_sec: float | None = None,
|
||
enrich_address: bool = True,
|
||
) -> YandexCitySweepCounters:
|
||
"""Yandex.Недвижимость city sweep: SERP → save → address-enrich.
|
||
|
||
Фазы per anchor:
|
||
1. SERP: YandexRealtyScraper.fetch_around_multi_room(lat, lon, radius_m,
|
||
max_pages=pages_per_anchor) — rooms × price-range combos, dedup внутри.
|
||
2. SAVE: save_listings(db, lots, run_id=run_id).
|
||
3. ADDRESS-ENRICH (если enrich_address=True): для yandex-листингов этого anchor'а
|
||
без номера дома в address (address IS NOT NULL AND NOT (address ~ ',\\s*\\d+'))
|
||
запрашиваем detail-страницу и извлекаем полный адрес из <title>.
|
||
Переиспользует приватные функции yandex_address_backfill:
|
||
_eligible_listings_for_ids → _fetch_and_update_one.
|
||
|
||
Anti-bot / cancel семантика:
|
||
- Cooperative cancel: is_cancelled(db, run_id) проверяется перед каждым anchor.
|
||
- YandexRealtyScraper не propagate'ит block/rate exceptions (глотает → []).
|
||
Поэтому per-anchor ловим broad Exception; при N подряд — mark_banned + abort.
|
||
- request_delay_sec: доп. asyncio.sleep МЕЖДУ anchor'ами (поверх per-page sleep
|
||
внутри scraper'а). Также используется как inter-request delay для address-enrich.
|
||
|
||
Возвращает YandexCitySweepCounters (агрегат по всем anchor'ам).
|
||
mark_done вызывается ВСЕГДА (кроме cancel / ban / fatal).
|
||
"""
|
||
from sqlalchemy import text as _text
|
||
|
||
from app.services.yandex_address_backfill import (
|
||
_RE_HAS_HOUSE_NUMBER,
|
||
_extract_address_from_title,
|
||
)
|
||
|
||
_anchors = anchors if anchors is not None else EKB_ANCHORS
|
||
counters = YandexCitySweepCounters(anchors_total=len(_anchors))
|
||
inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0
|
||
enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0
|
||
consecutive_failures = 0
|
||
|
||
# Прокси для address-enrich curl_cffi сессии (зеркало yandex_address_backfill).
|
||
_proxy_url = settings.scraper_proxy_url
|
||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||
|
||
try:
|
||
for idx, (lat, lon, name) in enumerate(_anchors, start=1):
|
||
# ── Cooperative cancel ───────────────────────────────────────────
|
||
if scrape_runs.is_cancelled(db, run_id):
|
||
logger.info(
|
||
"yandex-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(
|
||
"yandex-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
lat,
|
||
lon,
|
||
)
|
||
|
||
# ── Per-anchor phases под watchdog-таймаутом ────────────────────
|
||
# Весь SERP + address-enrich одного anchor'а оборачиваем в wait_for.
|
||
# TimeoutError → log + errors_count++ + continue (sweep не прерывается).
|
||
|
||
anchor_lots: list[ScrapedLot] = []
|
||
_anchor_timed_out = False
|
||
# Capture loop variables in default args to satisfy B023 (flake8/ruff):
|
||
# без этого inner async def захватывает изменяемые переменные цикла
|
||
# по ссылке → неправильные значения при asyncio scheduling.
|
||
_lat, _lon, _name = lat, lon, name
|
||
|
||
async def _yandex_anchor_phases(
|
||
_a_lat: float = _lat,
|
||
_a_lon: float = _lon,
|
||
_a_name: str = _name,
|
||
) -> None:
|
||
"""Все фазы одного anchor'а (SERP + address-enrich)."""
|
||
nonlocal anchor_lots, consecutive_failures
|
||
|
||
# ── Phase 1+2: SERP + save ─────────────────────────────────
|
||
async with YandexRealtyScraper() as scraper:
|
||
anchor_lots = await scraper.fetch_around_multi_room(
|
||
_a_lat,
|
||
_a_lon,
|
||
radius_m,
|
||
max_pages=pages_per_anchor,
|
||
)
|
||
counters.lots_fetched += len(anchor_lots)
|
||
if anchor_lots:
|
||
inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
|
||
counters.lots_inserted += inserted
|
||
counters.lots_updated += updated
|
||
consecutive_failures = 0
|
||
logger.info(
|
||
"yandex-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d",
|
||
run_id,
|
||
_a_name,
|
||
len(anchor_lots),
|
||
counters.lots_inserted,
|
||
counters.lots_updated,
|
||
)
|
||
|
||
# ── Phase 3: address-enrich ────────────────────────────────
|
||
if not (enrich_address and anchor_lots):
|
||
return
|
||
source_ids = [lot.source_id for lot in anchor_lots if lot.source_id]
|
||
if not source_ids:
|
||
return
|
||
id_rows = (
|
||
db.execute(
|
||
_text("""
|
||
SELECT id, source_id, address, source_url
|
||
FROM listings
|
||
WHERE source = 'yandex'
|
||
AND source_id = ANY(CAST(:ids AS text[]))
|
||
AND address IS NOT NULL
|
||
AND NOT (address ~ ',\\s*\\d+')
|
||
AND source_url IS NOT NULL
|
||
"""),
|
||
{"ids": source_ids},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
items = [dict(r) for r in id_rows]
|
||
counters.address_attempted += len(items)
|
||
|
||
if items:
|
||
from curl_cffi.requests import AsyncSession as _AsyncSession
|
||
|
||
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):
|
||
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:
|
||
logger.warning(
|
||
"yandex-sweep run_id=%d address-enrich:"
|
||
" HTTP %d listing_id=%d",
|
||
run_id,
|
||
resp.status_code,
|
||
lid,
|
||
)
|
||
counters.address_failed += 1
|
||
else:
|
||
new_addr = _extract_address_from_title(resp.text)
|
||
if (
|
||
new_addr is None
|
||
or new_addr == old_addr
|
||
or not _RE_HAS_HOUSE_NUMBER.search(new_addr)
|
||
):
|
||
pass
|
||
else:
|
||
try:
|
||
db.execute(
|
||
_text("""
|
||
UPDATE listings
|
||
SET address = :addr,
|
||
geocode_tried_at = NULL
|
||
WHERE id = CAST(:id AS bigint)
|
||
"""),
|
||
{"addr": new_addr, "id": lid},
|
||
)
|
||
db.commit()
|
||
counters.address_enriched += 1
|
||
logger.info(
|
||
"yandex-sweep run_id=%d: "
|
||
"address updated listing_id=%d "
|
||
"%r -> %r",
|
||
run_id,
|
||
lid,
|
||
old_addr,
|
||
new_addr,
|
||
)
|
||
except Exception as upd_exc:
|
||
counters.address_failed += 1
|
||
counters.errors_count += 1
|
||
logger.warning(
|
||
"yandex-sweep run_id=%d: "
|
||
"address DB update failed "
|
||
"listing_id=%d: %s",
|
||
run_id,
|
||
lid,
|
||
upd_exc,
|
||
)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
except Exception as fetch_exc:
|
||
counters.address_failed += 1
|
||
logger.warning(
|
||
"yandex-sweep run_id=%d: address-enrich "
|
||
"fetch failed listing_id=%d: %s",
|
||
run_id,
|
||
lid,
|
||
fetch_exc,
|
||
)
|
||
if eidx < len(items) - 1:
|
||
await asyncio.sleep(enrich_delay)
|
||
|
||
logger.info(
|
||
"yandex-sweep run_id=%d anchor %s: " "address enrich=%d/%d failed=%d",
|
||
run_id,
|
||
_a_name,
|
||
counters.address_enriched,
|
||
counters.address_attempted,
|
||
counters.address_failed,
|
||
)
|
||
|
||
try:
|
||
await asyncio.wait_for(_yandex_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC)
|
||
except TimeoutError:
|
||
logger.warning(
|
||
"yandex-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
|
||
"timed out after %ds — skipping",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
lat,
|
||
lon,
|
||
ANCHOR_TIMEOUT_SEC,
|
||
)
|
||
counters.errors_count += 1
|
||
consecutive_failures += 1
|
||
_anchor_timed_out = True
|
||
if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES:
|
||
logger.error(
|
||
"yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
|
||
"%d consecutive failures (last: timeout): "
|
||
"IP likely blocked or proxy hung",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
consecutive_failures,
|
||
)
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
scrape_runs.mark_banned(
|
||
db,
|
||
run_id,
|
||
f"yandex sweep aborted: {consecutive_failures} consecutive "
|
||
f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)",
|
||
counters.to_dict(),
|
||
)
|
||
return counters
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
if idx < len(_anchors):
|
||
await asyncio.sleep(inter_anchor_delay)
|
||
continue
|
||
except Exception as e:
|
||
# YandexRealtyScraper не propagate'ит block/rate exceptions — ловим broad.
|
||
logger.exception("yandex-sweep run_id=%d: anchor %s SERP failed", run_id, name)
|
||
counters.errors_count += 1
|
||
consecutive_failures += 1
|
||
if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES:
|
||
logger.error(
|
||
"yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
|
||
"%d consecutive failures (IP likely blocked): %s",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
consecutive_failures,
|
||
e,
|
||
)
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
scrape_runs.mark_banned(
|
||
db,
|
||
run_id,
|
||
f"yandex sweep aborted: {consecutive_failures} consecutive "
|
||
f"anchor failures (last: {e})",
|
||
counters.to_dict(),
|
||
)
|
||
return counters
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
# Пауза перед следующим anchor'ом даже при ошибке.
|
||
if idx < len(_anchors):
|
||
await asyncio.sleep(inter_anchor_delay)
|
||
continue
|
||
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
|
||
# Доп. пауза между anchor'ами (поверх per-page sleep внутри scraper'а).
|
||
if idx < len(_anchors):
|
||
await asyncio.sleep(inter_anchor_delay)
|
||
|
||
scrape_runs.mark_done(db, run_id, counters.to_dict())
|
||
logger.info(
|
||
"yandex-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) "
|
||
"address=%d/%d failed=%d errors=%d",
|
||
run_id,
|
||
counters.anchors_done,
|
||
counters.anchors_total,
|
||
counters.lots_fetched,
|
||
counters.lots_inserted,
|
||
counters.lots_updated,
|
||
counters.address_enriched,
|
||
counters.address_attempted,
|
||
counters.address_failed,
|
||
counters.errors_count,
|
||
)
|
||
return counters
|
||
|
||
except Exception as exc:
|
||
logger.exception("yandex-sweep run_id=%d: fatal error", run_id)
|
||
scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
|
||
raise
|
||
|
||
|
||
# ── N1.ru city sweep (#575) ─────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class N1CitySweepCounters:
|
||
"""Aggregate counters для N1.ru city sweep run.
|
||
|
||
Как Yandex: N1 SERP — простой search → save, без house-catalog / detail-enrich
|
||
шага, поэтому houses/detail-полей нет.
|
||
"""
|
||
|
||
anchors_total: int = 0
|
||
anchors_done: int = 0
|
||
pages_fetched: int = 0
|
||
lots_fetched: int = 0
|
||
lots_inserted: int = 0
|
||
lots_updated: int = 0
|
||
errors_count: int = 0
|
||
|
||
def to_dict(self) -> dict[str, int]:
|
||
return {f.name: getattr(self, f.name) for f in fields(self)}
|
||
|
||
|
||
# Defensive abort: N1Scraper.fetch_around глотает ошибки и возвращает [] (нет
|
||
# block/rate exception-классов). Прерываем sweep после N подряд anchor'ов,
|
||
# упавших с исключением — сигнал системного блока IP. Пустой результат (0 lots)
|
||
# — НЕ ошибка, счётчик не растёт (зеркало yandex).
|
||
N1_SWEEP_MAX_CONSECUTIVE_FAILURES = 3
|
||
|
||
|
||
async def run_n1_city_sweep(
|
||
db: Session,
|
||
*,
|
||
run_id: int,
|
||
anchors: list[tuple[float, float, str]] | None = None,
|
||
radius_m: int = 1500,
|
||
pages_per_anchor: int = 2,
|
||
request_delay_sec: float | None = None,
|
||
) -> N1CitySweepCounters:
|
||
"""N1.ru city sweep: iterate anchors × pages → save (#575).
|
||
|
||
Зеркало run_yandex_city_sweep — N1 это региональный SERP-портал ЕКБ без
|
||
house-catalog / detail-enrich шага:
|
||
per anchor: N1Scraper().fetch_around(lat, lon, radius_m, page=p)
|
||
→ save_listings(db, lots) → accumulate counters.
|
||
|
||
Anti-bot / cancel семантика (зеркало yandex/avito sweep):
|
||
- Cooperative cancel: is_cancelled(db, run_id) перед каждым anchor.
|
||
- N1Scraper не определяет block/rate exception-классов (fetch_around ловит
|
||
ошибки внутри → []). Per-anchor ловим broad Exception, логируем, continue;
|
||
защитно прерываем (early abort + mark_banned) при N подряд исключениях.
|
||
- request_delay_sec применяется как доп. asyncio.sleep МЕЖДУ anchor'ами
|
||
(N1Scraper сам спит self.request_delay_sec между страницами).
|
||
"""
|
||
_anchors = anchors if anchors is not None else EKB_ANCHORS
|
||
counters = N1CitySweepCounters(anchors_total=len(_anchors))
|
||
inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0
|
||
consecutive_failures = 0
|
||
|
||
try:
|
||
for idx, (lat, lon, name) in enumerate(_anchors, start=1):
|
||
if scrape_runs.is_cancelled(db, run_id):
|
||
logger.info(
|
||
"n1-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(
|
||
"n1-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
lat,
|
||
lon,
|
||
)
|
||
try:
|
||
anchor_lots: list[ScrapedLot] = []
|
||
async with N1Scraper() as scraper:
|
||
for page in range(1, pages_per_anchor + 1):
|
||
page_lots = await scraper.fetch_around(lat, lon, radius_m, page=page)
|
||
counters.pages_fetched += 1
|
||
if not page_lots:
|
||
break
|
||
anchor_lots.extend(page_lots)
|
||
|
||
counters.lots_fetched += len(anchor_lots)
|
||
if anchor_lots:
|
||
inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
|
||
counters.lots_inserted += inserted
|
||
counters.lots_updated += updated
|
||
consecutive_failures = 0
|
||
except Exception as e:
|
||
logger.exception("n1-sweep run_id=%d: anchor %s failed", run_id, name)
|
||
counters.errors_count += 1
|
||
consecutive_failures += 1
|
||
if consecutive_failures >= N1_SWEEP_MAX_CONSECUTIVE_FAILURES:
|
||
logger.error(
|
||
"n1-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
|
||
"%d consecutive failures (IP likely blocked): %s",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
consecutive_failures,
|
||
e,
|
||
)
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
scrape_runs.mark_banned(
|
||
db,
|
||
run_id,
|
||
f"n1 sweep aborted: {consecutive_failures} consecutive "
|
||
f"anchor failures (last: {e})",
|
||
counters.to_dict(),
|
||
)
|
||
return counters
|
||
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
|
||
if idx < len(_anchors):
|
||
await asyncio.sleep(inter_anchor_delay)
|
||
|
||
scrape_runs.mark_done(db, run_id, counters.to_dict())
|
||
logger.info(
|
||
"n1-sweep run_id=%d done: anchors=%d/%d pages=%d lots=%d (ins=%d/upd=%d) errors=%d",
|
||
run_id,
|
||
counters.anchors_done,
|
||
counters.anchors_total,
|
||
counters.pages_fetched,
|
||
counters.lots_fetched,
|
||
counters.lots_inserted,
|
||
counters.lots_updated,
|
||
counters.errors_count,
|
||
)
|
||
return counters
|
||
|
||
except Exception as exc:
|
||
logger.exception("n1-sweep run_id=%d: fatal error", run_id)
|
||
scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
|
||
raise
|
||
|
||
|
||
# ── Cian city sweep (#860) ──────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class CianCitySweepCounters:
|
||
"""Aggregate counters для Cian city sweep run.
|
||
|
||
Поля симметричны CitySweepCounters, но без imv_* (у Cian нет IMV-фазы).
|
||
detail_* — обогащение через cian_detail (вкл. price-history).
|
||
houses_* — обогащение newbuilding/ЖК через cian_newbuilding.
|
||
"""
|
||
|
||
anchors_total: int = 0
|
||
anchors_done: int = 0
|
||
lots_fetched: int = 0
|
||
lots_inserted: int = 0
|
||
lots_updated: int = 0
|
||
detail_attempted: int = 0
|
||
detail_enriched: int = 0
|
||
detail_failed: int = 0
|
||
houses_attempted: int = 0
|
||
houses_enriched: int = 0
|
||
houses_failed: int = 0
|
||
errors_count: int = 0
|
||
|
||
def to_dict(self) -> dict[str, int]:
|
||
return {f.name: getattr(self, f.name) for f in fields(self)}
|
||
|
||
|
||
# Defensive abort threshold: если N подряд anchor'ов упали с исключением
|
||
# (не просто вернули пустой результат) — прерываем sweep.
|
||
CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES = 3
|
||
|
||
|
||
async def run_cian_city_sweep(
|
||
db: Session,
|
||
*,
|
||
run_id: int,
|
||
anchors: list[tuple[float, float, str]] | None = None,
|
||
radius_m: int = 1500,
|
||
pages_per_anchor: int = 3,
|
||
request_delay_sec: float = 5.0,
|
||
detail_top_n: int = 10,
|
||
enrich_houses: bool = True,
|
||
) -> CianCitySweepCounters:
|
||
"""Cian full city sweep: SERP → detail(+price-history) → newbuilding/houses.
|
||
|
||
Симметрично run_avito_city_sweep. Использует EKB_ANCHORS (общий список).
|
||
|
||
Фазы per anchor:
|
||
1. SERP: CianScraper.fetch_around_multi_room(lat, lon, radius_m, pages=pages_per_anchor)
|
||
2. SAVE: save_listings(db, lots, run_id=run_id)
|
||
3. DETAIL: top-N свежих cian-листингов без detail_enriched_at →
|
||
fetch_detail(source_url) + save_detail_enrichment(db, listing_id, enrichment)
|
||
(save_detail_enrichment уже пишет price_changes → offer_price_history)
|
||
4. HOUSES (если enrich_houses=True): для домов из этого anchor'а с cian_zhk_url →
|
||
fetch_newbuilding(zhk_url) + save_newbuilding_enrichment(db, house_id, enr)
|
||
|
||
Anti-bot:
|
||
- CianScraper управляет своей curl_cffi-сессией (прокси уже внутри).
|
||
- asyncio.sleep(request_delay_sec) между detail-запросами (jitter ±20%).
|
||
- Defensive abort при CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES SERP-ошибок подряд.
|
||
|
||
Cooperative cancel: scrape_runs.is_cancelled(db, run_id) перед каждым anchor.
|
||
mark_done вызывается ВСЕГДА (finally outer).
|
||
"""
|
||
from app.services.scrapers.cian import CianScraper
|
||
from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment
|
||
from app.services.scrapers.cian_newbuilding import (
|
||
fetch_newbuilding,
|
||
save_newbuilding_enrichment,
|
||
)
|
||
|
||
_anchors = anchors if anchors is not None else EKB_ANCHORS
|
||
counters = CianCitySweepCounters(anchors_total=len(_anchors))
|
||
consecutive_failures = 0
|
||
|
||
try:
|
||
for idx, (lat, lon, name) in enumerate(_anchors, start=1):
|
||
# Cooperative cancel перед каждым anchor
|
||
if scrape_runs.is_cancelled(db, run_id):
|
||
logger.info(
|
||
"cian-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(
|
||
"cian-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
lat,
|
||
lon,
|
||
)
|
||
|
||
# ── Per-anchor phases под watchdog-таймаутом ────────────────────
|
||
# Весь SERP + detail + houses одного anchor'а оборачиваем в wait_for.
|
||
# TimeoutError → log + errors_count++ + continue (sweep не прерывается).
|
||
|
||
anchor_lots: list[ScrapedLot] = []
|
||
# Capture loop variables in default args (B023): prevents stale binding
|
||
# if the coroutine is scheduled after the loop variable changes.
|
||
_c_lat, _c_lon, _c_name = lat, lon, name
|
||
|
||
async def _cian_anchor_phases(
|
||
_a_lat: float = _c_lat,
|
||
_a_lon: float = _c_lon,
|
||
_a_name: str = _c_name,
|
||
) -> None:
|
||
"""Все фазы одного cian anchor'а (SERP + detail + houses)."""
|
||
nonlocal anchor_lots, consecutive_failures
|
||
from sqlalchemy import text as _text
|
||
|
||
# ── Phase 1+2: SERP + save ─────────────────────────────────
|
||
async with CianScraper() as scraper:
|
||
anchor_lots = await scraper.fetch_around_multi_room(
|
||
_a_lat,
|
||
_a_lon,
|
||
radius_m,
|
||
pages=pages_per_anchor,
|
||
)
|
||
counters.lots_fetched += len(anchor_lots)
|
||
if anchor_lots:
|
||
inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
|
||
counters.lots_inserted += inserted
|
||
counters.lots_updated += updated
|
||
consecutive_failures = 0
|
||
logger.info(
|
||
"cian-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d",
|
||
run_id,
|
||
_a_name,
|
||
len(anchor_lots),
|
||
counters.lots_inserted,
|
||
counters.lots_updated,
|
||
)
|
||
|
||
# ── Phase 3: detail enrichment (top-N) ────────────────────
|
||
if detail_top_n > 0:
|
||
priority_rows = (
|
||
db.execute(
|
||
_text("""
|
||
SELECT id, source_url
|
||
FROM listings
|
||
WHERE source = 'cian'
|
||
AND source_url IS NOT NULL
|
||
AND detail_enriched_at IS NULL
|
||
AND scraped_at > NOW() - INTERVAL '2 hours'
|
||
ORDER BY scraped_at DESC NULLS LAST
|
||
LIMIT :lim
|
||
"""),
|
||
{"lim": detail_top_n},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
for didx, row in enumerate(priority_rows):
|
||
listing_id: int = row["id"]
|
||
source_url: str = row["source_url"]
|
||
counters.detail_attempted += 1
|
||
try:
|
||
enrichment = await fetch_detail(source_url)
|
||
if enrichment is not None:
|
||
save_detail_enrichment(db, listing_id, enrichment)
|
||
counters.detail_enriched += 1
|
||
else:
|
||
counters.detail_failed += 1
|
||
logger.warning(
|
||
"cian-sweep run_id=%d: detail returned None for "
|
||
"listing_id=%d url=%s",
|
||
run_id,
|
||
listing_id,
|
||
source_url,
|
||
)
|
||
except Exception as exc:
|
||
counters.detail_failed += 1
|
||
counters.errors_count += 1
|
||
logger.warning(
|
||
"cian-sweep run_id=%d: detail failed listing_id=%d: %s",
|
||
run_id,
|
||
listing_id,
|
||
exc,
|
||
)
|
||
if didx < len(priority_rows) - 1:
|
||
jitter = random.uniform(0.8, 1.2)
|
||
await asyncio.sleep(request_delay_sec * jitter)
|
||
logger.info(
|
||
"cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d",
|
||
run_id,
|
||
_a_name,
|
||
counters.detail_enriched,
|
||
counters.detail_attempted,
|
||
counters.detail_failed,
|
||
)
|
||
|
||
# ── Phase 4: newbuilding/houses enrichment ─────────────────
|
||
if not (enrich_houses and anchor_lots):
|
||
return
|
||
cian_nb_ext_ids = {
|
||
lot.house_ext_id
|
||
for lot in anchor_lots
|
||
if lot.house_source == "cian_newbuilding" and lot.house_ext_id
|
||
}
|
||
if not cian_nb_ext_ids:
|
||
return
|
||
nb_id_list = list(cian_nb_ext_ids)
|
||
try:
|
||
house_rows = (
|
||
db.execute(
|
||
_text("""
|
||
SELECT h.id, h.cian_zhk_url
|
||
FROM houses h
|
||
JOIN house_sources hs ON hs.house_id = h.id
|
||
WHERE hs.ext_source = 'cian_newbuilding'
|
||
AND hs.ext_id = ANY(CAST(:ids AS text[]))
|
||
AND h.cian_zhk_url IS NOT NULL
|
||
"""),
|
||
{"ids": nb_id_list},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"cian-sweep run_id=%d anchor %s: houses DB query failed — "
|
||
"skipping houses phase (%d ids): %s",
|
||
run_id,
|
||
_a_name,
|
||
len(nb_id_list),
|
||
exc,
|
||
)
|
||
counters.houses_failed += len(nb_id_list)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
return
|
||
for hidx, hrow in enumerate(house_rows):
|
||
house_id_val: int = hrow["id"]
|
||
zhk_url: str = hrow["cian_zhk_url"]
|
||
counters.houses_attempted += 1
|
||
try:
|
||
nb_enrichment = await fetch_newbuilding(zhk_url)
|
||
if nb_enrichment is not None:
|
||
save_newbuilding_enrichment(db, house_id_val, nb_enrichment)
|
||
counters.houses_enriched += 1
|
||
else:
|
||
counters.houses_failed += 1
|
||
logger.warning(
|
||
"cian-sweep run_id=%d: newbuilding returned None "
|
||
"house_id=%d url=%s",
|
||
run_id,
|
||
house_id_val,
|
||
zhk_url,
|
||
)
|
||
except Exception as exc:
|
||
counters.houses_failed += 1
|
||
counters.errors_count += 1
|
||
logger.warning(
|
||
"cian-sweep run_id=%d: houses failed house_id=%d: %s",
|
||
run_id,
|
||
house_id_val,
|
||
exc,
|
||
)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
if hidx < len(house_rows) - 1:
|
||
await asyncio.sleep(request_delay_sec)
|
||
logger.info(
|
||
"cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d",
|
||
run_id,
|
||
_a_name,
|
||
counters.houses_enriched,
|
||
counters.houses_attempted,
|
||
counters.houses_failed,
|
||
)
|
||
|
||
try:
|
||
await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC)
|
||
except TimeoutError:
|
||
logger.warning(
|
||
"cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
|
||
"timed out after %ds — skipping",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
lat,
|
||
lon,
|
||
ANCHOR_TIMEOUT_SEC,
|
||
)
|
||
counters.errors_count += 1
|
||
consecutive_failures += 1
|
||
if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES:
|
||
logger.error(
|
||
"cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
|
||
"%d consecutive failures (last: timeout): "
|
||
"IP likely blocked or proxy hung",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
consecutive_failures,
|
||
)
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
scrape_runs.mark_banned(
|
||
db,
|
||
run_id,
|
||
f"cian sweep aborted: {consecutive_failures} consecutive "
|
||
f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)",
|
||
counters.to_dict(),
|
||
)
|
||
return counters
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
continue
|
||
except Exception as e:
|
||
logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name)
|
||
counters.errors_count += 1
|
||
consecutive_failures += 1
|
||
if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES:
|
||
logger.error(
|
||
"cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
|
||
"%d consecutive SERP failures (IP likely blocked): %s",
|
||
run_id,
|
||
idx,
|
||
len(_anchors),
|
||
name,
|
||
consecutive_failures,
|
||
e,
|
||
)
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
scrape_runs.mark_banned(
|
||
db,
|
||
run_id,
|
||
f"cian sweep aborted: {consecutive_failures} consecutive "
|
||
f"anchor SERP failures (last: {e})",
|
||
counters.to_dict(),
|
||
)
|
||
return counters
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
continue
|
||
|
||
counters.anchors_done = idx
|
||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||
|
||
# Пауза между anchor'ами (поверх per-request sleep внутри scraper'а)
|
||
if idx < len(_anchors):
|
||
await asyncio.sleep(request_delay_sec)
|
||
|
||
scrape_runs.mark_done(db, run_id, counters.to_dict())
|
||
logger.info(
|
||
"cian-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) "
|
||
"detail=%d/%d houses=%d/%d errors=%d",
|
||
run_id,
|
||
counters.anchors_done,
|
||
counters.anchors_total,
|
||
counters.lots_fetched,
|
||
counters.lots_inserted,
|
||
counters.lots_updated,
|
||
counters.detail_enriched,
|
||
counters.detail_attempted,
|
||
counters.houses_enriched,
|
||
counters.houses_attempted,
|
||
counters.errors_count,
|
||
)
|
||
return counters
|
||
|
||
except Exception as exc:
|
||
logger.exception("cian-sweep run_id=%d: fatal error", run_id)
|
||
scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
|
||
raise
|
||
|
||
|
||
# ── Public re-exports ───────────────────────────────────────────
|
||
__all__ = [
|
||
"ANCHOR_TIMEOUT_SEC",
|
||
"CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES",
|
||
"EKB_ANCHORS",
|
||
"CianCitySweepCounters",
|
||
"CitySweepCounters",
|
||
"PipelineCounters",
|
||
"PipelineResult",
|
||
"YandexCitySweepCounters",
|
||
"run_avito_city_sweep",
|
||
"run_avito_pipeline",
|
||
"run_cian_city_sweep",
|
||
"run_yandex_city_sweep",
|
||
]
|