2102 lines
92 KiB
Python
2102 lines
92 KiB
Python
import json
|
||
import logging
|
||
import os
|
||
import random
|
||
import re
|
||
import signal
|
||
import time
|
||
import uuid
|
||
import html as html_module
|
||
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Callable
|
||
from urllib.parse import urlsplit, urlunsplit
|
||
from urllib.request import Request, urlopen
|
||
|
||
import urllib3
|
||
|
||
try:
|
||
import orjson as _orjson # ~3-5× быстрее stdlib json на парсинге
|
||
_HAS_ORJSON = True
|
||
except ImportError:
|
||
_orjson = None
|
||
_HAS_ORJSON = False
|
||
|
||
from playwright.sync_api import Error as PlaywrightError
|
||
from playwright.sync_api import BrowserContext, Page, sync_playwright
|
||
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
|
||
|
||
from .browser import BrowserFactory, HumanPacer, NetworkCapture
|
||
from .core.config import Settings, settings, parse_listing_segments
|
||
from .core.exceptions import AntiBotDetectedError, ListingResumeError, SiteStructureChangedError
|
||
from .core.logs import set_trace_id, setup_logging
|
||
from .core.retry import retryable
|
||
from .core.runtime_config import RuntimeConfig
|
||
from .core.utils import save_to_json
|
||
from .parsing.mapper import CarMapper
|
||
from .parsing.parser import VehicleParser
|
||
from .storage.db import PersistenceService
|
||
from .browser.listing import ListingCollector, ListingPageResult
|
||
from .storage.schemas import CarRecord
|
||
|
||
logger = logging.getLogger("iaai_scraper.scraper")
|
||
VEHICLE_ID_RE = re.compile(r"/VehicleDetail/(\d+)(?:~[A-Z]{2})?", re.IGNORECASE)
|
||
HTML_TAG_RE = re.compile(r"<[^>]+>")
|
||
SCRIPT_STYLE_RE = re.compile(r"<(script|style)[^>]*>.*?</\1>", re.IGNORECASE | re.DOTALL)
|
||
|
||
# ── Anti-stall watchdog ──
|
||
# Жёсткие верхние границы на per-page операции браузера. Если Playwright "тихо" завис
|
||
# (частая реакция IAAI на затяжной скрапинг — page hangs без raise), SIGALRM вернёт
|
||
# управление и существующая recover-логика (_reopen_listing_and_resume) стартует
|
||
# новый context с нуля.
|
||
_PAGE_COLLECT_TIMEOUT_S = 90
|
||
_PAGE_NEXT_TIMEOUT_S = 60
|
||
# Проактивная ротация контекста отключена: reactive-восстановление
|
||
# (retry при пустой странице + _reopen_listing_and_resume) уже закрывает
|
||
# anti-bot соскоки, а проактивная ротация требовала долистывать кликами
|
||
# после reopen и конфликтовала с лимитом max_nav_pages в recovery-пути.
|
||
# Включить можно переменной окружения IAAI_CONTEXT_ROTATE_EVERY_PAGES.
|
||
_CONTEXT_ROTATE_EVERY_PAGES = int(os.environ.get("IAAI_CONTEXT_ROTATE_EVERY_PAGES", "0") or 0)
|
||
|
||
|
||
class PageOperationTimeoutError(Exception):
|
||
"""Browser page operation exceeded per-op watchdog timeout."""
|
||
|
||
|
||
class _PageOpWatchdog:
|
||
"""SIGALRM-based watchdog.
|
||
|
||
Работает только в главном потоке процесса (Celery prefork child — это ок).
|
||
В других контекстах — no-op.
|
||
"""
|
||
|
||
def __init__(self, seconds: int, label: str) -> None:
|
||
self.seconds = max(1, int(seconds))
|
||
self.label = label
|
||
self._old_handler = None
|
||
self._active = False
|
||
|
||
def _handler(self, signum, frame): # noqa: ARG002
|
||
raise PageOperationTimeoutError(
|
||
f"Page operation '{self.label}' exceeded {self.seconds}s watchdog"
|
||
)
|
||
|
||
def __enter__(self):
|
||
import threading as _threading
|
||
if _threading.current_thread() is not _threading.main_thread():
|
||
return self
|
||
if not hasattr(signal, "SIGALRM"):
|
||
return self
|
||
try:
|
||
self._old_handler = signal.signal(signal.SIGALRM, self._handler)
|
||
signal.alarm(self.seconds)
|
||
self._active = True
|
||
except (ValueError, OSError):
|
||
# signal можно выставлять только из main thread; в остальном пропускаем.
|
||
self._active = False
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc, tb):
|
||
if not self._active:
|
||
return False
|
||
try:
|
||
signal.alarm(0)
|
||
if self._old_handler is not None:
|
||
signal.signal(signal.SIGALRM, self._old_handler)
|
||
except (ValueError, OSError):
|
||
pass
|
||
return False
|
||
|
||
|
||
class IAAIScraper:
|
||
|
||
@staticmethod
|
||
def _is_protection_or_network_error(exc: Exception) -> bool:
|
||
message = str(exc).lower()
|
||
signals = (
|
||
"captcha",
|
||
"antibot",
|
||
"blocked",
|
||
"challenge",
|
||
"ns_error_net_interrupt",
|
||
"navigation",
|
||
"timeout",
|
||
"403",
|
||
"429",
|
||
)
|
||
return any(signal in message for signal in signals)
|
||
|
||
@staticmethod
|
||
def _html_to_text(html: str) -> str:
|
||
if not html:
|
||
return ""
|
||
cleaned = SCRIPT_STYLE_RE.sub(" ", html)
|
||
text = HTML_TAG_RE.sub(" ", cleaned)
|
||
text = html_module.unescape(text)
|
||
return re.sub(r"\s+", " ", text).strip()
|
||
|
||
@staticmethod
|
||
def _raise_if_blocked_or_incomplete(parsed: dict, vehicle_url: str) -> None:
|
||
dom_hints = parsed.get("dom_hints", {}) or {}
|
||
access_notes = parsed.get("access_notes", {}) or {}
|
||
summary = parsed.get("vehicle_summary", {}) or {}
|
||
|
||
has_identity = bool(summary.get("lot_number") or summary.get("make") or summary.get("model"))
|
||
|
||
if (dom_hints.get("has_captcha_text") or dom_hints.get("has_antibot_text")) and not has_identity:
|
||
raise AntiBotDetectedError(f"IAAI anti-bot detected for {vehicle_url}")
|
||
|
||
if (access_notes.get("possible_captcha") or access_notes.get("possible_antibot")) and not has_identity:
|
||
raise AntiBotDetectedError(f"IAAI blocked or challenged request for {vehicle_url}")
|
||
|
||
if not has_identity:
|
||
raise SiteStructureChangedError(f"Vehicle page returned no recognizable vehicle data: {vehicle_url}")
|
||
|
||
@staticmethod
|
||
def _extract_origin_id_from_url(vehicle_url: str) -> str | None:
|
||
match = VEHICLE_ID_RE.search(vehicle_url)
|
||
if not match:
|
||
return None
|
||
return match.group(1)
|
||
|
||
@staticmethod
|
||
def _extract_db_origin_id_from_url(vehicle_url: str) -> str | None:
|
||
raw_id = IAAIScraper._extract_origin_id_from_url(vehicle_url)
|
||
if not raw_id:
|
||
return None
|
||
return f"iaai:{raw_id}"
|
||
|
||
@staticmethod
|
||
def _normalize_vehicle_url(vehicle_url: str) -> str:
|
||
try:
|
||
parts = urlsplit(vehicle_url)
|
||
return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))
|
||
except Exception:
|
||
return vehicle_url
|
||
|
||
def __init__(self, runtime_settings: Settings | None = None) -> None:
|
||
self.settings = runtime_settings or settings
|
||
setup_logging(self.settings.log_level, self.settings.log_file)
|
||
self.runtime_config = RuntimeConfig.from_file(self.settings.runtime_config_file)
|
||
self.trace_id = uuid.uuid4().hex[:12]
|
||
set_trace_id(self.trace_id)
|
||
self.playwright = None
|
||
self.browser = None
|
||
self.context: BrowserContext | None = None
|
||
self.browser_factory = BrowserFactory(self.settings)
|
||
self.pacer = HumanPacer(self.settings)
|
||
self.listing_collector = ListingCollector(self.settings, self.pacer)
|
||
self.vehicle_parser = VehicleParser()
|
||
self.car_mapper = CarMapper()
|
||
self.persistence = PersistenceService(self.settings)
|
||
self._shutdown_requested = False
|
||
# HTTP pool: при parallel_tabs=24 и concurrency=4 пик ≈ 96 конкурентных сокетов;
|
||
# даём запас до 256, чтобы не упираться в PoolError под всплесками PX/retry.
|
||
# TCP_NODELAY выключает Nagle: для коротких HTTPS-запросов с keep-alive
|
||
# это снижает tail-latency на десятки ms.
|
||
import socket as _socket
|
||
_socket_options = [
|
||
(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1),
|
||
(_socket.SOL_SOCKET, _socket.SO_KEEPALIVE, 1),
|
||
]
|
||
proxy_url = self.settings.proxy.server
|
||
if proxy_url:
|
||
_proxy_kwargs: dict = {
|
||
"num_pools": 8,
|
||
"maxsize": 256,
|
||
"block": False,
|
||
"retries": False,
|
||
"timeout": urllib3.Timeout(connect=5, read=20),
|
||
"socket_options": _socket_options,
|
||
}
|
||
if self.settings.proxy.username:
|
||
_proxy_kwargs["proxy_headers"] = urllib3.make_headers(
|
||
proxy_basic_auth=f"{self.settings.proxy.username}:{self.settings.proxy.password or ''}"
|
||
)
|
||
self._http_pool: urllib3.PoolManager = urllib3.ProxyManager(proxy_url, **_proxy_kwargs)
|
||
logger.info("HTTP fast-path using proxy: %s", proxy_url)
|
||
else:
|
||
self._http_pool = urllib3.PoolManager(
|
||
num_pools=8,
|
||
maxsize=256,
|
||
block=False,
|
||
retries=False,
|
||
timeout=urllib3.Timeout(connect=5, read=20),
|
||
socket_options=_socket_options,
|
||
)
|
||
|
||
def _new_trace_id(self, prefix: str) -> str:
|
||
trace_id = f"{prefix}-{uuid.uuid4().hex[:8]}"
|
||
self.trace_id = trace_id
|
||
set_trace_id(trace_id)
|
||
return trace_id
|
||
|
||
def __enter__(self) -> "IAAIScraper":
|
||
if self.playwright is None:
|
||
self.playwright = sync_playwright().start()
|
||
if self.browser is None:
|
||
self.browser = self.browser_factory.create_browser(self.playwright)
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc, tb) -> None:
|
||
self.close()
|
||
|
||
def close(self) -> None:
|
||
if self.context is not None:
|
||
try:
|
||
self.context.close()
|
||
except PlaywrightError:
|
||
pass
|
||
finally:
|
||
self.context = None
|
||
if self.browser is not None:
|
||
try:
|
||
self.browser.close()
|
||
except PlaywrightError:
|
||
pass
|
||
finally:
|
||
self.browser = None
|
||
if self.playwright is not None:
|
||
try:
|
||
self.playwright.stop()
|
||
except PlaywrightError:
|
||
pass
|
||
finally:
|
||
self.playwright = None
|
||
if getattr(self, "_http_pool", None) is not None:
|
||
try:
|
||
self._http_pool.clear()
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
self._http_pool = None
|
||
|
||
def _new_context(self) -> BrowserContext:
|
||
if self.browser is None:
|
||
self.__enter__()
|
||
if self.context:
|
||
try:
|
||
self.context.close()
|
||
except PlaywrightError:
|
||
pass
|
||
self.context = self.browser_factory.create_context(self.browser)
|
||
return self.context
|
||
|
||
def init_db(self):
|
||
self.persistence.create_tables()
|
||
return {"status": "ok", "database_url": self.settings.database.url}
|
||
|
||
def _warmup_visit(self, page: Page) -> None:
|
||
try:
|
||
logger.info("Warmup: visiting homepage to pass anti-bot challenge...")
|
||
page.goto(self.settings.home_url, wait_until="commit", timeout=30_000)
|
||
try:
|
||
page.wait_for_load_state("domcontentloaded", timeout=6_000)
|
||
except PlaywrightTimeoutError:
|
||
pass
|
||
try:
|
||
page.wait_for_function("() => document.title && document.title.length > 3", timeout=4_000)
|
||
except PlaywrightTimeoutError:
|
||
pass
|
||
time.sleep(0.3)
|
||
logger.info("Warmup done: %s (title=%s)", page.url, page.title()[:50])
|
||
except Exception as e:
|
||
logger.warning("Warmup visit failed: %s — continuing anyway", e)
|
||
time.sleep(1.5)
|
||
|
||
def _get_page_with_warmup(self) -> Page:
|
||
context = self._new_context()
|
||
page = context.new_page()
|
||
self._warmup_visit(page)
|
||
return page
|
||
|
||
def _dedupe_urls(self, raw_urls: list[str]) -> list[str]:
|
||
vehicle_urls: list[str] = []
|
||
seen: set[str] = set()
|
||
for raw in raw_urls:
|
||
normalized = self._normalize_vehicle_url(raw)
|
||
if normalized not in seen:
|
||
seen.add(normalized)
|
||
vehicle_urls.append(normalized)
|
||
return vehicle_urls
|
||
|
||
def _filter_known_urls(self, vehicle_urls: list[str]) -> tuple[list[str], int]:
|
||
url_to_origin_id = {
|
||
url: self._extract_db_origin_id_from_url(url)
|
||
for url in vehicle_urls
|
||
}
|
||
candidate_origin_ids = [oid for oid in url_to_origin_id.values() if oid]
|
||
existing_urls, existing_ids = self.persistence.get_existing_urls_and_ids(
|
||
vehicle_urls, candidate_origin_ids,
|
||
)
|
||
known_urls = {
|
||
url for url in vehicle_urls
|
||
if (url in existing_urls) or (url_to_origin_id.get(url) in existing_ids)
|
||
}
|
||
skipped = len(known_urls)
|
||
if skipped:
|
||
logger.info("Filtering already known vehicles: skipped %d", skipped)
|
||
new_urls = [url for url in vehicle_urls if url not in known_urls]
|
||
return new_urls, skipped
|
||
|
||
def _extract_page_urls(
|
||
self,
|
||
page_result: ListingPageResult,
|
||
all_raw_urls: list[str],
|
||
seen_urls: set[str],
|
||
all_listing_origin_urls: set[str] | None = None,
|
||
) -> list[str]:
|
||
page_urls: list[str] = []
|
||
for item in page_result.vehicle_links:
|
||
normalized = self._normalize_vehicle_url(item.href)
|
||
all_raw_urls.append(item.href)
|
||
if normalized and all_listing_origin_urls is not None:
|
||
all_listing_origin_urls.add(normalized)
|
||
if normalized and normalized not in seen_urls:
|
||
seen_urls.add(normalized)
|
||
page_urls.append(normalized)
|
||
return page_urls
|
||
|
||
@staticmethod
|
||
def _build_segment_listing_url(base_url: str, make: str | None) -> str:
|
||
"""Построить URL листинга для сегмента. Make передаётся через query-параметр."""
|
||
if not make:
|
||
return base_url
|
||
sep = "&" if "?" in base_url else "?"
|
||
return f"{base_url}{sep}Make={make.replace(' ', '%20')}"
|
||
|
||
def _recover_empty_listing_page(
|
||
self,
|
||
page: Page,
|
||
*,
|
||
page_number: int,
|
||
all_raw_urls: list[str],
|
||
seen_urls: set[str],
|
||
all_listing_origin_urls: set[str] | None = None,
|
||
listing_url: str | None = None,
|
||
) -> tuple[ListingPageResult, list[str]]:
|
||
if page_number == 1:
|
||
logger.warning("Page 1 returned 0 links — retrying listing open...")
|
||
time.sleep(3)
|
||
self.listing_collector.open_cars_listing(page, url_override=listing_url)
|
||
page_result = self.listing_collector.collect_current_page(page, page_number=page_number)
|
||
return page_result, self._extract_page_urls(page_result, all_raw_urls, seen_urls, all_listing_origin_urls)
|
||
|
||
logger.warning(
|
||
"Page %d returned 0 links — retrying current page before stopping pagination",
|
||
page_number,
|
||
)
|
||
try:
|
||
page.reload(wait_until="domcontentloaded", timeout=30_000)
|
||
except Exception as exc:
|
||
logger.debug("Page %d reload failed during empty-page recovery: %s", page_number, exc)
|
||
page_result = self.listing_collector.collect_current_page(page, page_number=page_number)
|
||
return page_result, self._extract_page_urls(page_result, all_raw_urls, seen_urls, all_listing_origin_urls)
|
||
|
||
def _open_listing_page(
|
||
self,
|
||
*,
|
||
make: str | None,
|
||
model: str | None,
|
||
listing_url: str | None = None,
|
||
year_min: int | None = None,
|
||
year_max: int | None = None,
|
||
) -> tuple[Page, dict[str, str | int | None]]:
|
||
page = self._get_page_with_warmup()
|
||
try:
|
||
self.listing_collector.open_cars_listing(page, url_override=listing_url)
|
||
applied_filters = self.listing_collector.apply_filters(
|
||
page,
|
||
make=make,
|
||
model=model,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
)
|
||
except Exception:
|
||
page.close()
|
||
raise
|
||
return page, applied_filters
|
||
|
||
def _reopen_listing_and_resume(
|
||
self,
|
||
*,
|
||
target_page_number: int,
|
||
make: str | None,
|
||
model: str | None,
|
||
listing_url: str | None = None,
|
||
year_min: int | None = None,
|
||
year_max: int | None = None,
|
||
max_nav_pages: int = 10,
|
||
) -> tuple[Page, dict[str, str | int | None]]:
|
||
# Ограничиваем глубину навигации: если до цели > max_nav_pages кликов — не пытаемся.
|
||
if target_page_number > max_nav_pages + 1:
|
||
raise ListingResumeError(
|
||
f"Cannot resume at page {target_page_number}: "
|
||
f"exceeds max navigation depth ({max_nav_pages} pages)"
|
||
)
|
||
page, applied_filters = self._open_listing_page(
|
||
make=make,
|
||
model=model,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
)
|
||
|
||
for expected_page in range(2, target_page_number + 1):
|
||
if not self.listing_collector.go_to_next_page(page, expected_page_number=expected_page):
|
||
page.close()
|
||
raise ListingResumeError(f"Failed to resume listing at page {target_page_number}")
|
||
|
||
logger.warning("Listing resumed at page %d after recovery", target_page_number)
|
||
return page, applied_filters
|
||
|
||
def _open_listing_for_stream(
|
||
self,
|
||
*,
|
||
make: str | None,
|
||
model: str | None,
|
||
listing_url: str | None = None,
|
||
year_min: int | None = None,
|
||
year_max: int | None = None,
|
||
) -> tuple[Page, dict[str, str | int | None]]:
|
||
# Всегда открываем с page 1. Page-level resume убран: эфемерные URL и короткие
|
||
# сегменты делали pagination-resume хрупким. Bootstrap прогресс сохраняется
|
||
# на уровне сегментов в worker/tasks.py (_save_last_completed_segment).
|
||
return self._open_listing_page(
|
||
make=make,
|
||
model=model,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
)
|
||
|
||
def collect_listing(
|
||
self,
|
||
make: str | None = None,
|
||
model: str | None = None,
|
||
known_origin_ids: set[str] | None = None,
|
||
max_duration_seconds: float | None = None,
|
||
):
|
||
page = self._get_page_with_warmup()
|
||
|
||
try:
|
||
listing = self.listing_collector.collect_listing_links(
|
||
page,
|
||
make=make,
|
||
model=model,
|
||
known_origin_ids=known_origin_ids,
|
||
max_duration_seconds=max_duration_seconds,
|
||
)
|
||
finally:
|
||
page.close()
|
||
|
||
return {
|
||
"status": "ok",
|
||
**listing,
|
||
}
|
||
|
||
def _sync_listing_streaming(
|
||
self,
|
||
*,
|
||
make: str | None,
|
||
model: str | None,
|
||
lane: str,
|
||
limit: int | None,
|
||
effective_only_new: bool,
|
||
started_at: float,
|
||
listing_url: str | None = None,
|
||
year_min: int | None = None,
|
||
year_max: int | None = None,
|
||
) -> dict[str, Any]:
|
||
batch_size = self.settings.celery.batch_size
|
||
pending_urls: list[str] = []
|
||
seen_urls: set[str] = set()
|
||
all_raw_urls: list[str] = []
|
||
all_listing_origin_urls: set[str] = set()
|
||
pages_info: list[dict[str, Any]] = []
|
||
skipped_existing = 0
|
||
total = 0
|
||
cars_upserted = 0
|
||
cars_failed = 0
|
||
images_upserted = 0
|
||
failures: list[dict[str, str]] = []
|
||
early_stopped = False
|
||
truncated_by_time_budget = False
|
||
|
||
known_origin_ids: set[str] | None = None
|
||
threshold = self.settings.listing.early_stop_threshold
|
||
if effective_only_new and threshold > 0.0:
|
||
try:
|
||
known_origin_ids = self.persistence.get_all_origin_ids_for_lane("iaai:")
|
||
logger.info(
|
||
"Loaded %d known origin_ids for early-stop listing",
|
||
len(known_origin_ids),
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("Could not load known origin_ids for early-stop: %s", exc)
|
||
|
||
# Совместимость: если only_new без limit и без сегментного URL — старая схема.
|
||
# При сегментированном скрапинге (listing_url/year фильтры) всегда streaming.
|
||
if effective_only_new and (limit is None or limit <= 0) and not listing_url and year_min is None and year_max is None:
|
||
listing_payload = self.collect_listing(
|
||
make=make,
|
||
model=model,
|
||
known_origin_ids=known_origin_ids,
|
||
max_duration_seconds=None,
|
||
)
|
||
raw_urls = list(listing_payload.get("vehicle_urls", []))
|
||
deduped_urls = self._dedupe_urls(raw_urls)
|
||
fresh_urls, page_skipped = self._filter_known_urls(deduped_urls)
|
||
skipped_existing += page_skipped
|
||
pending_urls.extend(fresh_urls)
|
||
total = len(fresh_urls)
|
||
|
||
for raw in raw_urls:
|
||
normalized = self._normalize_vehicle_url(raw)
|
||
if normalized:
|
||
all_listing_origin_urls.add(normalized)
|
||
|
||
logger.info(
|
||
"Starting sync: %d vehicles to process (batch mode, only_new legacy path)",
|
||
total,
|
||
)
|
||
|
||
while len(pending_urls) >= batch_size:
|
||
batch_urls = pending_urls[:batch_size]
|
||
try:
|
||
batch_result = self.sync_batch(batch_urls, lane=lane)
|
||
cars_upserted += batch_result.get("cars_upserted", 0)
|
||
cars_failed += batch_result.get("cars_failed", 0)
|
||
images_upserted += batch_result.get("images_upserted", 0)
|
||
failures.extend(batch_result.get("failures", []))
|
||
except Exception as batch_exc:
|
||
logger.error("Legacy only_new batch failed: %s", batch_exc)
|
||
cars_failed += len(batch_urls)
|
||
failures.append({"vehicle_url": "legacy_only_new_batch", "error": str(batch_exc)})
|
||
pending_urls = pending_urls[batch_size:]
|
||
|
||
if pending_urls:
|
||
try:
|
||
batch_result = self.sync_batch(pending_urls, lane=lane)
|
||
cars_upserted += batch_result.get("cars_upserted", 0)
|
||
cars_failed += batch_result.get("cars_failed", 0)
|
||
images_upserted += batch_result.get("images_upserted", 0)
|
||
failures.extend(batch_result.get("failures", []))
|
||
except Exception as batch_exc:
|
||
logger.error("Legacy only_new final batch failed: %s", batch_exc)
|
||
cars_failed += len(pending_urls)
|
||
failures.append({"vehicle_url": "legacy_only_new_final_batch", "error": str(batch_exc)})
|
||
|
||
return {
|
||
"listing": listing_payload,
|
||
"total": total,
|
||
"skipped_existing": skipped_existing,
|
||
"cars_upserted": cars_upserted,
|
||
"cars_failed": cars_failed,
|
||
"images_upserted": images_upserted,
|
||
"failures": failures,
|
||
"all_listing_origin_urls": all_listing_origin_urls,
|
||
}
|
||
|
||
# ── Pipeline-режим: батчи (HTTP + DB upsert) выполняются в фоне,
|
||
# пока главный поток продолжает собирать listing-страницы через
|
||
# Playwright. Это убирает простой Playwright во время HTTP-фазы
|
||
# (~50-65 сек на 400 машин) и даёт ~25-30% к throughput.
|
||
# max_workers=1: батчи исполняются последовательно, но ВНЕ
|
||
# главного потока, чтобы не блокировать сбор listing-а.
|
||
# Browser fallback откладывается (skip_browser_fallback=True),
|
||
# потому что Playwright не thread-safe — обработаем после loop'а.
|
||
batch_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="batch-pipeline")
|
||
pending_future: Future | None = None
|
||
pending_future_meta: tuple[int, int] | None = None # (batch_start, batch_size)
|
||
deferred_fallbacks: list[tuple[str, int]] = []
|
||
|
||
def _drain_pending_future() -> bool:
|
||
"""Waits for in-flight batch and accumulates its result.
|
||
|
||
Returns False if a non-recoverable error occurred and remaining
|
||
batches should be aborted (mirrors the legacy contract).
|
||
"""
|
||
nonlocal pending_future, pending_future_meta
|
||
nonlocal cars_upserted, cars_failed, images_upserted, failures
|
||
if pending_future is None:
|
||
return True
|
||
future = pending_future
|
||
meta = pending_future_meta or (0, 0)
|
||
pending_future = None
|
||
pending_future_meta = None
|
||
batch_start, batch_size_actual = meta
|
||
try:
|
||
batch_result = future.result()
|
||
except PlaywrightError as pw_exc:
|
||
logger.warning(
|
||
"Batch %d-%d browser crashed: %s. Reinitializing browser context.",
|
||
batch_start + 1,
|
||
batch_start + batch_size_actual,
|
||
pw_exc,
|
||
)
|
||
cars_failed += batch_size_actual
|
||
failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(pw_exc)})
|
||
try:
|
||
self._new_context()
|
||
logger.info("Browser context reinitialized successfully after crash")
|
||
return True
|
||
except Exception as reinit_exc:
|
||
logger.error("Failed to reinitialize browser: %s. Aborting remaining batches.", reinit_exc)
|
||
return False
|
||
except Exception as batch_exc:
|
||
logger.error(
|
||
"Batch %d-%d failed: %s. Continuing with next batch.",
|
||
batch_start + 1,
|
||
batch_start + batch_size_actual,
|
||
batch_exc,
|
||
)
|
||
cars_failed += batch_size_actual
|
||
failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(batch_exc)})
|
||
return True
|
||
cars_upserted += batch_result.get("cars_upserted", 0)
|
||
cars_failed += batch_result.get("cars_failed", 0)
|
||
images_upserted += batch_result.get("images_upserted", 0)
|
||
failures.extend(batch_result.get("failures", []))
|
||
deferred_fallbacks.extend(batch_result.get("deferred_fallback_urls", []))
|
||
return True
|
||
|
||
def _process_pending_batch(batch_urls: list[str], batch_start: int) -> bool:
|
||
nonlocal pending_future, pending_future_meta
|
||
# Drain previous batch first (if still running).
|
||
if not _drain_pending_future():
|
||
return False
|
||
if not batch_urls:
|
||
return True
|
||
logger.info(
|
||
"Processing batch %d-%d of %d (pipelined)",
|
||
batch_start + 1,
|
||
batch_start + len(batch_urls),
|
||
total,
|
||
)
|
||
# Cookie header обязательно вычисляем в главном потоке — Playwright
|
||
# context.cookies() использует greenlet, который нельзя дёргать из background thread'а.
|
||
cookie_header = self._cookie_header_for_context()
|
||
pending_future = batch_executor.submit(
|
||
self.sync_batch,
|
||
batch_urls,
|
||
lane=lane,
|
||
skip_browser_fallback=True,
|
||
cookie_header_override=cookie_headerkground thread'а.
|
||
cookie_header = self._cookie_header_for_context()
|
||
pending_future = batch_executor.submit(
|
||
self.sync_batch,
|
||
batch_urls,
|
||
lane=lane,
|
||
skip_browser_fallback=True,
|
||
cookie_header_override=cookie_header,
|
||
)
|
||
pending_future_meta = (batch_start, len(batch_urls))
|
||
return True
|
||
|
||
page = None
|
||
try:
|
||
page, applied_filters = self._open_listing_for_stream(
|
||
make=make,
|
||
model=model,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
)
|
||
|
||
pages_since_rotation = 0
|
||
for page_number in range(1, self.settings.listing.max_pages_per_run + 1):
|
||
# Проактивная ротация контекста (опционально, по умолчанию выключена).
|
||
# Если включена — требует долистывания до page_number после reopen,
|
||
# поэтому max_nav_pages поднят под реальную глубину.
|
||
if _CONTEXT_ROTATE_EVERY_PAGES > 0 and pages_since_rotation >= _CONTEXT_ROTATE_EVERY_PAGES:
|
||
logger.warning(
|
||
"Rotating browser context at page %d (every %d pages) to reset anti-bot state",
|
||
page_number, _CONTEXT_ROTATE_EVERY_PAGES,
|
||
)
|
||
try:
|
||
page.close()
|
||
except Exception:
|
||
pass
|
||
page = None
|
||
try:
|
||
page, _ = self._reopen_listing_and_resume(
|
||
target_page_number=page_number,
|
||
make=make,
|
||
model=model,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
max_nav_pages=max(page_number, 500),
|
||
)
|
||
pages_since_rotation = 0
|
||
except ListingResumeError as resume_exc:
|
||
logger.warning(
|
||
"Context rotation could not resume at page %d (%s) — stopping segment",
|
||
page_number, resume_exc,
|
||
)
|
||
break
|
||
|
||
try:
|
||
with _PageOpWatchdog(_PAGE_COLLECT_TIMEOUT_S, f"collect page {page_number}"):
|
||
page_result = self.listing_collector.collect_current_page(page, page_number=page_number)
|
||
except (PageOperationTimeoutError, PlaywrightError) as op_exc:
|
||
logger.warning(
|
||
"Page %d collect stalled/failed (%s) — reopening listing and resuming",
|
||
page_number, op_exc,
|
||
)
|
||
try:
|
||
page.close()
|
||
except Exception:
|
||
pass
|
||
page = None
|
||
try:
|
||
page, _ = self._reopen_listing_and_resume(
|
||
target_page_number=page_number,
|
||
make=make,
|
||
model=model,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
max_nav_pages=max(page_number, 500),
|
||
)
|
||
pages_since_rotation = 0
|
||
with _PageOpWatchdog(_PAGE_COLLECT_TIMEOUT_S, f"collect page {page_number} (after reopen)"):
|
||
page_result = self.listing_collector.collect_current_page(page, page_number=page_number)
|
||
except (ListingResumeError, PageOperationTimeoutError, PlaywrightError) as resume_exc:
|
||
logger.warning(
|
||
"Cannot recover at page %d (%s) — stopping pagination for this segment",
|
||
page_number, resume_exc,
|
||
)
|
||
break
|
||
|
||
pages_info.append({
|
||
"page_number": page_result.page_number,
|
||
"links_found": len(page_result.vehicle_links),
|
||
})
|
||
pages_since_rotation += 1
|
||
|
||
page_urls = self._extract_page_urls(
|
||
page_result,
|
||
all_raw_urls,
|
||
seen_urls,
|
||
all_listing_origin_urls,
|
||
)
|
||
|
||
if not page_urls:
|
||
page_result, page_urls = self._recover_empty_listing_page(
|
||
page,
|
||
page_number=page_number,
|
||
all_raw_urls=all_raw_urls,
|
||
seen_urls=seen_urls,
|
||
all_listing_origin_urls=all_listing_origin_urls,
|
||
listing_url=listing_url,
|
||
)
|
||
if not page_urls and page_number > 1:
|
||
logger.warning("Page %d still empty after retry — reopening listing and resuming", page_number)
|
||
page.close()
|
||
try:
|
||
page, _ = self._reopen_listing_and_resume(
|
||
target_page_number=page_number,
|
||
make=make,
|
||
model=model,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
max_nav_pages=max(page_number, 500),
|
||
)
|
||
page_result = self.listing_collector.collect_current_page(page, page_number=page_number)
|
||
page_urls = self._extract_page_urls(
|
||
page_result,
|
||
all_raw_urls,
|
||
seen_urls,
|
||
all_listing_origin_urls,
|
||
)
|
||
except ListingResumeError as resume_exc:
|
||
logger.warning(
|
||
"Cannot resume at page %d (%s) — stopping pagination for this segment",
|
||
page_number, resume_exc,
|
||
)
|
||
page = None
|
||
page_urls = []
|
||
if not page_urls:
|
||
logger.info("Page %d: 0 new links, stopping pagination", page_number)
|
||
break
|
||
|
||
if known_origin_ids is not None and threshold > 0.0 and page_result.vehicle_links:
|
||
page_known = sum(
|
||
1
|
||
for item in page_result.vehicle_links
|
||
if item.lot_number and f"iaai:{item.lot_number}" in known_origin_ids
|
||
)
|
||
page_new = len(page_result.vehicle_links) - page_known
|
||
ratio = page_known / len(page_result.vehicle_links)
|
||
if ratio >= threshold and page_new == 0:
|
||
logger.info(
|
||
"Early stop on page %d: %.0f%% known (%d/%d), 0 new >= threshold %.0f%%",
|
||
page_number,
|
||
ratio * 100,
|
||
page_known,
|
||
len(page_result.vehicle_links),
|
||
threshold * 100,
|
||
)
|
||
early_stopped = True
|
||
break
|
||
|
||
fresh_urls = page_urls
|
||
page_skipped = 0
|
||
if effective_only_new:
|
||
fresh_urls, page_skipped = self._filter_known_urls(page_urls)
|
||
skipped_existing += page_skipped
|
||
|
||
if limit is not None and limit > 0:
|
||
remaining_limit = max(0, limit - total - len(pending_urls))
|
||
if remaining_limit <= 0:
|
||
break
|
||
fresh_urls = fresh_urls[:remaining_limit]
|
||
|
||
pending_urls.extend(fresh_urls)
|
||
total += len(fresh_urls)
|
||
|
||
logger.info(
|
||
"Page %d: %d links, %d new, %d known (total new: %d/%s)",
|
||
page_number,
|
||
len(page_urls),
|
||
len(fresh_urls),
|
||
page_skipped,
|
||
total,
|
||
limit if limit is not None else "∞",
|
||
)
|
||
|
||
# Прогресс каждые 10 страниц на уровне WARNING.
|
||
if page_number % 10 == 0:
|
||
logger.warning(
|
||
"sync_listing progress: pages=%d, discovered=%d, upserted=%d, failed=%d, pending=%d",
|
||
page_number,
|
||
total,
|
||
cars_upserted,
|
||
cars_failed,
|
||
len(pending_urls),
|
||
)
|
||
|
||
while len(pending_urls) >= batch_size:
|
||
batch_urls = pending_urls[:batch_size]
|
||
if not _process_pending_batch(batch_urls, cars_upserted + cars_failed):
|
||
pending_urls = []
|
||
break
|
||
pending_urls = pending_urls[batch_size:]
|
||
|
||
if limit is not None and limit > 0 and total >= limit:
|
||
break
|
||
if (
|
||
self.settings.listing.collect_current_page_only
|
||
or not self.settings.listing.include_pagination
|
||
or not page_result.next_page_detected
|
||
):
|
||
break
|
||
try:
|
||
with _PageOpWatchdog(_PAGE_NEXT_TIMEOUT_S, f"next page {page_number + 1}"):
|
||
next_ok = self.listing_collector.go_to_next_page(page, expected_page_number=page_number + 1)
|
||
except (PageOperationTimeoutError, PlaywrightError) as nav_exc:
|
||
logger.warning(
|
||
"go_to_next_page stalled/failed at page %d (%s) — will reopen",
|
||
page_number + 1, nav_exc,
|
||
)
|
||
next_ok = False
|
||
if not next_ok:
|
||
logger.warning("Failed to navigate to page %d — reopening listing and resuming", page_number + 1)
|
||
try:
|
||
page.close()
|
||
except Exception:
|
||
pass
|
||
page = None
|
||
try:
|
||
page, _ = self._reopen_listing_and_resume(
|
||
target_page_number=page_number + 1,
|
||
make=make,
|
||
model=model,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
max_nav_pages=max(page_number + 1, 500),
|
||
)
|
||
pages_since_rotation = 0
|
||
except ListingResumeError as resume_exc:
|
||
logger.warning(
|
||
"Cannot resume at page %d (%s) — treating pagination as exhausted",
|
||
page_number + 1, resume_exc,
|
||
)
|
||
break
|
||
|
||
if pending_urls:
|
||
_process_pending_batch(pending_urls, cars_upserted + cars_failed)
|
||
|
||
# Финальный drain последнего in-flight батча.
|
||
_drain_pending_future()
|
||
|
||
# Обрабатываем отложенные browser fallback'и (Playwright теперь свободен).
|
||
if deferred_fallbacks:
|
||
fb_pages = min(len(deferred_fallbacks), self._FALLBACK_PARALLEL_PAGES)
|
||
logger.info(
|
||
"Processing %d deferred browser fallbacks (%d parallel pages)",
|
||
len(deferred_fallbacks), fb_pages,
|
||
)
|
||
try:
|
||
fb_results = self._browser_fallback_parallel(
|
||
deferred_fallbacks, len(deferred_fallbacks), fb_pages,
|
||
)
|
||
fb_records = fb_results.get("records", [])
|
||
if fb_records:
|
||
try:
|
||
upsert_result = self.persistence.upsert_cars_batch(fb_records)
|
||
cars_upserted += upsert_result["inserted"] + upsert_result["updated"]
|
||
images_upserted += upsert_result["images_upserted"]
|
||
except Exception as exc:
|
||
logger.error("Deferred fallback upsert failed: %s", exc)
|
||
cars_failed += len(fb_records)
|
||
cars_failed += fb_results.get("cars_failed", 0)
|
||
failures.extend(fb_results.get("failures", []))
|
||
except Exception as fb_exc:
|
||
logger.error("Deferred fallback batch failed: %s", fb_exc)
|
||
cars_failed += len(deferred_fallbacks)
|
||
|
||
logger.warning(
|
||
"sync_listing final progress: pages=%d, discovered=%d, upserted=%d, failed=%d",
|
||
len(pages_info),
|
||
total,
|
||
cars_upserted,
|
||
cars_failed,
|
||
)
|
||
finally:
|
||
# Ensure no future is left dangling on exception path.
|
||
try:
|
||
_drain_pending_future()
|
||
except Exception:
|
||
pass
|
||
batch_executor.shutdown(wait=True)
|
||
if page is not None:
|
||
page.close()
|
||
|
||
return {
|
||
"listing": {
|
||
"status": "ok",
|
||
"listing_url": self.settings.listing.cars_url,
|
||
"applied_filters": applied_filters,
|
||
"pages_collected": len(pages_info),
|
||
"vehicles_collected": len(all_raw_urls),
|
||
"vehicle_urls": all_raw_urls,
|
||
"early_stopped": early_stopped,
|
||
"truncated_by_time_budget": truncated_by_time_budget,
|
||
"pages": pages_info,
|
||
},
|
||
"total": total,
|
||
"skipped_existing": skipped_existing,
|
||
"cars_upserted": cars_upserted,
|
||
"cars_failed": cars_failed,
|
||
"images_upserted": images_upserted,
|
||
"failures": failures,
|
||
"all_listing_origin_urls": all_listing_origin_urls,
|
||
}
|
||
|
||
def _get_page(self) -> Page:
|
||
if not self.context:
|
||
self._new_context()
|
||
return self.context.new_page()
|
||
|
||
@retryable(max_attempts=3, jitter_seconds=0.25)
|
||
def scrape_vehicle_detail(self, vehicle_url: str):
|
||
page = self._get_page()
|
||
try:
|
||
return self._scrape_on_page(page, vehicle_url)
|
||
finally:
|
||
page.close()
|
||
|
||
_JS_EXTRACT = """
|
||
() => {
|
||
try {
|
||
const scripts = document.querySelectorAll('script:not([src])');
|
||
for (const s of scripts) {
|
||
const t = s.textContent || '';
|
||
if (!t.includes('inventoryView') || !t.includes('attributes')) continue;
|
||
const start = t.indexOf('{');
|
||
if (start < 0) continue;
|
||
let depth = 0, end = -1;
|
||
for (let i = start; i < t.length; i++) {
|
||
if (t[i] === '{') depth++;
|
||
else if (t[i] === '}') { depth--; if (depth === 0) { end = i; break; } }
|
||
}
|
||
if (end < 0) continue;
|
||
try {
|
||
const obj = JSON.parse(t.slice(start, end + 1));
|
||
const iv = obj.inventoryView;
|
||
if (!iv || !iv.attributes) continue;
|
||
const attr = iv.attributes;
|
||
const imgs = (iv.imageDimensions && iv.imageDimensions.keys && iv.imageDimensions.keys.$values)
|
||
? iv.imageDimensions.keys.$values : [];
|
||
const bid = (obj.auctionInformation && obj.auctionInformation.biddingInformation)
|
||
? obj.auctionInformation.biddingInformation : {};
|
||
const prebid = (obj.auctionInformation && obj.auctionInformation.prebidInformation)
|
||
? obj.auctionInformation.prebidInformation : {};
|
||
return {
|
||
ok: true,
|
||
SalvageId: attr.SalvageId || '',
|
||
StockNumber: attr.StockNumber || '',
|
||
Year: attr.Year || '',
|
||
Make: attr.Make || '',
|
||
Model: attr.Model || '',
|
||
Series: attr.Series || '',
|
||
BodyStyleName: attr.BodyStyleName || '',
|
||
Cylinders: attr.Cylinders || '',
|
||
DriveLineTypeDesc: attr.DriveLineTypeDesc || '',
|
||
EngineSize: (attr.EngineInformation || attr.EngineSize || '').trim(),
|
||
FuelTypeCode: attr.FuelTypeCode || '',
|
||
Transmission: attr.Transmission || '',
|
||
ExteriorColor: attr.ExteriorColor || '',
|
||
PrimaryDamageDesc: attr.PrimaryDamageDesc || '',
|
||
SecondaryDamageDesc: attr.SecondaryDamageDesc || '',
|
||
ODOValue: attr.ODOValue || '',
|
||
ODOBrand: attr.ODOBrand || '',
|
||
RunAndDrive: attr.RunAndDrive || '',
|
||
Keys: attr.Keys || '',
|
||
BranchName: attr.BranchName || '',
|
||
AuctionDateTime: attr.AuctionDateTime || '',
|
||
Title: attr.Title || '',
|
||
TitleBrand: attr.TitleBrand || '',
|
||
TitleCode: attr.TitleCode || '',
|
||
EstRepairCost: attr.EstRepairCost || '',
|
||
VehicleGrade: attr.VehicleGrade || '',
|
||
highBidAmount: prebid.highBidAmount || bid.highBidAmount || '',
|
||
buyNowPrice: prebid.buyNowPrice || bid.buyNowPrice || '',
|
||
acv: attr.ProviderACV || '',
|
||
BranchNumber: attr.BranchNumber || '',
|
||
imageKeys: imgs.map(i => i.k || '').filter(Boolean),
|
||
};
|
||
} catch (_) { continue; }
|
||
}
|
||
} catch (_) {}
|
||
return { ok: false };
|
||
}
|
||
"""
|
||
|
||
@staticmethod
|
||
def _build_car_from_js(js_data: dict, vehicle_url: str) -> dict:
|
||
if not js_data or not js_data.get("ok"):
|
||
return {}
|
||
|
||
brnch = str(js_data.get("BranchNumber", "") or "").strip()
|
||
img_keys = js_data.get("imageKeys") or []
|
||
image_urls = [
|
||
f"https://vis.iaai.com/resizer?imageKeys={k}&width=845&height=633"
|
||
for k in img_keys
|
||
]
|
||
|
||
return {
|
||
"source_url": vehicle_url,
|
||
"lot_number": js_data.get("StockNumber") or js_data.get("SalvageId"),
|
||
"year": js_data.get("Year"),
|
||
"make": js_data.get("Make"),
|
||
"model": js_data.get("Model"),
|
||
"trim": js_data.get("Series"),
|
||
"body_type": js_data.get("BodyStyleName"),
|
||
"cylinders": js_data.get("Cylinders"),
|
||
"drive": js_data.get("DriveLineTypeDesc"),
|
||
"engine": js_data.get("EngineSize"),
|
||
"fuel_type": js_data.get("FuelTypeCode"),
|
||
"gearbox": js_data.get("Transmission"),
|
||
"color": js_data.get("ExteriorColor"),
|
||
"primary_damage": js_data.get("PrimaryDamageDesc"),
|
||
"secondary_damage": js_data.get("SecondaryDamageDesc"),
|
||
"odometer": js_data.get("ODOValue"),
|
||
"run_and_drive": js_data.get("RunAndDrive"),
|
||
"keys": js_data.get("Keys"),
|
||
"location": js_data.get("BranchName"),
|
||
"auction_date": js_data.get("AuctionDateTime"),
|
||
"title": js_data.get("Title"),
|
||
"current_bid": js_data.get("highBidAmount"),
|
||
"buy_now": js_data.get("buyNowPrice"),
|
||
"actual_cash_value": js_data.get("acv"),
|
||
"estimated_repair_cost": js_data.get("EstRepairCost"),
|
||
"image_urls": image_urls,
|
||
}
|
||
|
||
@staticmethod
|
||
def _build_payload_insights(vehicle_summary: dict) -> dict:
|
||
return {
|
||
"vehicle_core": vehicle_summary,
|
||
"pricing": {
|
||
"buy_now": vehicle_summary.get("buy_now"),
|
||
"current_bid": vehicle_summary.get("current_bid"),
|
||
"actual_cash_value": vehicle_summary.get("actual_cash_value"),
|
||
"estimated_repair_cost": vehicle_summary.get("estimated_repair_cost"),
|
||
"currency": "USD",
|
||
},
|
||
"bids": {"amount": vehicle_summary.get("current_bid"), "currency": "USD"},
|
||
"damage": {"primary": vehicle_summary.get("primary_damage"), "secondary": vehicle_summary.get("secondary_damage")},
|
||
"auction": {"auction_date": vehicle_summary.get("auction_date"), "branch": vehicle_summary.get("location")},
|
||
"images": {"count": len(vehicle_summary.get("image_urls") or []), "urls": vehicle_summary.get("image_urls") or []},
|
||
}
|
||
|
||
def _scrape_on_page(self, page: Page, vehicle_url: str):
|
||
trace_id = self._new_trace_id("scrape")
|
||
started_at = time.perf_counter()
|
||
|
||
if self.settings.block_resources:
|
||
BrowserFactory.enable_resource_blocking(page)
|
||
|
||
capture = NetworkCapture(self.settings)
|
||
capture.attach(page, origin_url=vehicle_url)
|
||
|
||
page.goto(vehicle_url, wait_until="commit", timeout=60_000)
|
||
|
||
try:
|
||
page.wait_for_load_state("domcontentloaded", timeout=5_000)
|
||
except PlaywrightTimeoutError:
|
||
pass
|
||
|
||
js_data: dict = {}
|
||
try:
|
||
js_data = page.evaluate(self._JS_EXTRACT) or {}
|
||
except Exception:
|
||
pass
|
||
|
||
if js_data.get("ok"):
|
||
vehicle_summary = self._build_car_from_js(js_data, vehicle_url)
|
||
has_identity = bool(vehicle_summary.get("make") or vehicle_summary.get("lot_number"))
|
||
if not has_identity:
|
||
raise SiteStructureChangedError(f"JS extraction returned no vehicle identity for {vehicle_url}")
|
||
|
||
db_record = self.car_mapper.map_to_car_record(
|
||
vehicle_url=vehicle_url,
|
||
vehicle_summary=vehicle_summary,
|
||
payload_insights=self._build_payload_insights(vehicle_summary),
|
||
)
|
||
network_dump = capture.export()
|
||
result = {
|
||
"trace_id": trace_id,
|
||
"source_url": vehicle_url,
|
||
"fetched_at_epoch": int(time.time()),
|
||
"elapsed_seconds": round(time.perf_counter() - started_at, 3),
|
||
"network": network_dump,
|
||
"vehicle_summary": vehicle_summary,
|
||
"payload_insights": {},
|
||
"embedded_json": [],
|
||
"dom_hints": {"has_captcha_text": False, "has_antibot_text": False},
|
||
"access_notes": {},
|
||
"db_record": db_record.model_dump(mode="json"),
|
||
}
|
||
if self.settings.raw_output_json:
|
||
save_to_json(network_dump, self.settings.raw_output_json)
|
||
return result
|
||
|
||
# Резервный путь: полный HTML-парсинг.
|
||
try:
|
||
page.wait_for_selector("#VehicleDetailViewModel, .veh-details, .vehicle-details, [data-uname='vehicleDetailPage']", timeout=400)
|
||
except PlaywrightTimeoutError:
|
||
pass
|
||
|
||
html = page.content()
|
||
try:
|
||
dom_text = page.evaluate("() => document.body?.textContent || ''")
|
||
except Exception:
|
||
dom_text = ""
|
||
network_dump = capture.export()
|
||
parsed = self.vehicle_parser.normalize(vehicle_url, html, dom_text, network_dump)
|
||
self._raise_if_blocked_or_incomplete(parsed, vehicle_url)
|
||
|
||
db_record = self.car_mapper.map_to_car_record(
|
||
vehicle_url=vehicle_url,
|
||
vehicle_summary=parsed.get("vehicle_summary", {}),
|
||
payload_insights=parsed.get("payload_insights", {}),
|
||
)
|
||
|
||
result = {
|
||
"trace_id": trace_id,
|
||
"source_url": vehicle_url,
|
||
"fetched_at_epoch": int(time.time()),
|
||
"elapsed_seconds": round(time.perf_counter() - started_at, 3),
|
||
"network": network_dump,
|
||
**parsed,
|
||
"db_record": db_record.model_dump(mode="json"),
|
||
}
|
||
|
||
if self.settings.raw_output_json:
|
||
save_to_json(network_dump, self.settings.raw_output_json)
|
||
return result
|
||
|
||
@staticmethod
|
||
def _extract_iaai_json_from_html(html: str, vehicle_url: str) -> dict | None:
|
||
"""Извлекает IAAI inventoryView.attributes из HTML без браузера.
|
||
|
||
Использует json.JSONDecoder.raw_decode для быстрого поиска JSON
|
||
вместо посимвольного сканирования скобок. Если получившийся срез
|
||
целиком отдаётся декодеру — пробуем orjson (3-5× быстрее stdlib).
|
||
"""
|
||
search = "inventoryView"
|
||
pos = html.find(search)
|
||
if pos < 0:
|
||
return None
|
||
|
||
script_start = html.rfind("<script", 0, pos)
|
||
if script_start < 0:
|
||
return None
|
||
tag_end = html.find(">", script_start)
|
||
if tag_end < 0:
|
||
return None
|
||
content_start = html.find("{", tag_end)
|
||
if content_start < 0:
|
||
return None
|
||
|
||
# Быстрый путь: вырезаем всё до </script>, пытаемся orjson; иначе stdlib.
|
||
obj: dict | None = None
|
||
if _HAS_ORJSON:
|
||
script_end = html.find("</script>", content_start)
|
||
if script_end > content_start:
|
||
# Подрезаем хвост до последней закрывающей скобки JSON.
|
||
snippet = html[content_start:script_end].rstrip().rstrip(";")
|
||
try:
|
||
obj = _orjson.loads(snippet)
|
||
except Exception:
|
||
obj = None
|
||
|
||
if obj is None:
|
||
decoder = json.JSONDecoder()
|
||
try:
|
||
obj, _ = decoder.raw_decode(html, content_start)
|
||
except (json.JSONDecodeError, ValueError):
|
||
return None
|
||
|
||
iv = obj.get("inventoryView")
|
||
if not iv or not iv.get("attributes"):
|
||
return None
|
||
|
||
attr = iv["attributes"]
|
||
imgs = []
|
||
try:
|
||
imgs = iv.get("imageDimensions", {}).get("keys", {}).get("$values", []) or []
|
||
except Exception:
|
||
pass
|
||
|
||
bid = {}
|
||
prebid = {}
|
||
try:
|
||
ai = obj.get("auctionInformation", {})
|
||
bid = ai.get("biddingInformation", {}) or {}
|
||
prebid = ai.get("prebidInformation", {}) or {}
|
||
except Exception:
|
||
pass
|
||
|
||
img_keys = [i.get("k", "") for i in imgs if i.get("k")]
|
||
return {
|
||
"ok": True,
|
||
"SalvageId": attr.get("SalvageId", ""),
|
||
"StockNumber": attr.get("StockNumber", ""),
|
||
"Year": attr.get("Year", ""),
|
||
"Make": attr.get("Make", ""),
|
||
"Model": attr.get("Model", ""),
|
||
"Series": attr.get("Series", ""),
|
||
"BodyStyleName": attr.get("BodyStyleName", ""),
|
||
"Cylinders": attr.get("Cylinders", ""),
|
||
"DriveLineTypeDesc": attr.get("DriveLineTypeDesc", ""),
|
||
"EngineSize": (attr.get("EngineInformation") or attr.get("EngineSize") or "").strip(),
|
||
"FuelTypeCode": attr.get("FuelTypeCode", ""),
|
||
"Transmission": attr.get("Transmission", ""),
|
||
"ExteriorColor": attr.get("ExteriorColor", ""),
|
||
"PrimaryDamageDesc": attr.get("PrimaryDamageDesc", ""),
|
||
"SecondaryDamageDesc": attr.get("SecondaryDamageDesc", ""),
|
||
"ODOValue": attr.get("ODOValue", ""),
|
||
"ODOBrand": attr.get("ODOBrand", ""),
|
||
"RunAndDrive": attr.get("RunAndDrive", ""),
|
||
"Keys": attr.get("Keys", ""),
|
||
"BranchName": attr.get("BranchName", ""),
|
||
"AuctionDateTime": attr.get("AuctionDateTime", ""),
|
||
"Title": attr.get("Title", ""),
|
||
"TitleBrand": attr.get("TitleBrand", ""),
|
||
"TitleCode": attr.get("TitleCode", ""),
|
||
"EstRepairCost": attr.get("EstRepairCost", ""),
|
||
"VehicleGrade": attr.get("VehicleGrade", ""),
|
||
"highBidAmount": prebid.get("highBidAmount") or bid.get("highBidAmount", ""),
|
||
"buyNowPrice": prebid.get("buyNowPrice") or bid.get("buyNowPrice", ""),
|
||
"acv": attr.get("ProviderACV", ""),
|
||
"BranchNumber": attr.get("BranchNumber", ""),
|
||
"imageKeys": img_keys,
|
||
}
|
||
|
||
def _scrape_via_context_request(self, vehicle_url: str) -> CarRecord:
|
||
if not self.context:
|
||
self._new_context()
|
||
assert self.context is not None
|
||
last_error: Exception | None = None
|
||
max_attempts = max(1, self.settings.fast_path_max_attempts)
|
||
timeout_ms = max(1000, self.settings.fast_path_timeout_ms)
|
||
for attempt in range(1, max_attempts + 1):
|
||
try:
|
||
response = self.context.request.get(vehicle_url, timeout=timeout_ms)
|
||
if not response.ok:
|
||
raise RuntimeError(f"HTTP fetch failed for {vehicle_url}: {response.status}")
|
||
|
||
html = response.text()
|
||
|
||
js_data = self._extract_iaai_json_from_html(html, vehicle_url)
|
||
if not js_data:
|
||
raise RuntimeError(f"HTTP fast-path missing embedded JSON for {vehicle_url}")
|
||
|
||
vehicle_summary = self._build_car_from_js(js_data, vehicle_url)
|
||
has_identity = bool(vehicle_summary.get("make") or vehicle_summary.get("lot_number"))
|
||
if not has_identity:
|
||
raise RuntimeError(f"HTTP fast-path returned incomplete identity for {vehicle_url}")
|
||
|
||
db_record = self.car_mapper.map_to_car_record(
|
||
vehicle_url=vehicle_url,
|
||
vehicle_summary=vehicle_summary,
|
||
payload_insights=self._build_payload_insights(vehicle_summary),
|
||
)
|
||
return CarRecord.model_validate(db_record.model_dump(mode="json"))
|
||
except Exception as exc:
|
||
last_error = exc
|
||
if attempt < max_attempts:
|
||
time.sleep(0.2)
|
||
|
||
if last_error is not None:
|
||
raise last_error
|
||
raise RuntimeError(f"HTTP fast-path failed for {vehicle_url}")
|
||
|
||
def _cookie_header_for_context(self) -> str:
|
||
if not self.context:
|
||
return ""
|
||
try:
|
||
cookies = self.context.cookies()
|
||
except Exception:
|
||
return ""
|
||
pairs = [
|
||
f"{item.get('name')}={item.get('value')}"
|
||
for item in cookies
|
||
if item.get("name") and item.get("value") is not None
|
||
]
|
||
return "; ".join(pairs)
|
||
|
||
def _scrape_via_raw_http(
|
||
self,
|
||
vehicle_url: str,
|
||
*,
|
||
cookie_header: str,
|
||
user_agent: str,
|
||
) -> CarRecord:
|
||
"""Быстрый HTTP-запрос через urllib3 (connection pooling / keep-alive).
|
||
|
||
Заголовок Accept-Encoding включает gzip/deflate/br — IAAI отдаёт HTML
|
||
со сжатием 5-10×, что снижает байты по сети и время декода.
|
||
urllib3 сам прозрачно распаковывает gzip и deflate; для br нужен
|
||
пакет `brotli` (он в зависимостях). Если brotli не установлен — он
|
||
просто не появится в Accept-Encoding и сервер ответит gzip.
|
||
"""
|
||
headers = {
|
||
"User-Agent": user_agent,
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
"Accept-Encoding": self._ACCEPT_ENCODING,
|
||
"Cache-Control": "no-cache",
|
||
"Pragma": "no-cache",
|
||
"Connection": "keep-alive",
|
||
"Upgrade-Insecure-Requests": "1",
|
||
}
|
||
if cookie_header:
|
||
headers["Cookie"] = cookie_header
|
||
|
||
# decode_content=True — urllib3 сам распакует gzip/deflate/br.
|
||
response = self._http_pool.request(
|
||
"GET", vehicle_url, headers=headers, decode_content=True,
|
||
)
|
||
if response.status >= 400:
|
||
raise RuntimeError(f"HTTP fetch failed for {vehicle_url}: {response.status}")
|
||
html = response.data.decode("utf-8", errors="ignore")
|
||
|
||
js_data = self._extract_iaai_json_from_html(html, vehicle_url)
|
||
if not js_data:
|
||
raise RuntimeError(f"HTTP fast-path missing embedded JSON for {vehicle_url}")
|
||
|
||
vehicle_summary = self._build_car_from_js(js_data, vehicle_url)
|
||
has_identity = bool(vehicle_summary.get("make") or vehicle_summary.get("lot_number"))
|
||
if not has_identity:
|
||
raise RuntimeError(f"HTTP fast-path returned incomplete identity for {vehicle_url}")
|
||
|
||
db_record = self.car_mapper.map_to_car_record(
|
||
vehicle_url=vehicle_url,
|
||
vehicle_summary=vehicle_summary,
|
||
payload_insights=self._build_payload_insights(vehicle_summary),
|
||
)
|
||
return CarRecord.model_validate(db_record.model_dump(mode="json"))
|
||
|
||
def sync_vehicle(self, vehicle_url: str, lane: str = "iaai"):
|
||
trace_id = self._new_trace_id("sync-vehicle")
|
||
started_at = time.perf_counter()
|
||
self.persistence.create_tables()
|
||
run_id = self.persistence.start_sync_run(lane=lane)
|
||
ids_fetched = 1
|
||
cars_upserted = 0
|
||
cars_failed = 0
|
||
images_upserted = 0
|
||
status = "failed"
|
||
error_summary = None
|
||
try:
|
||
scrape_result = self.scrape_vehicle_detail(vehicle_url)
|
||
db_record = scrape_result.get("db_record")
|
||
if not db_record:
|
||
raise RuntimeError("Scrape result does not contain db_record")
|
||
record = CarRecord.model_validate(db_record)
|
||
upsert = self.persistence.upsert_car(record)
|
||
cars_upserted = 1
|
||
images_upserted = int(upsert.get("images_upserted", 0))
|
||
status = "success"
|
||
return {
|
||
"trace_id": trace_id,
|
||
"status": status,
|
||
"run_id": run_id,
|
||
"vehicle_url": vehicle_url,
|
||
"db_action": upsert.get("action"),
|
||
"images_upserted": images_upserted,
|
||
"elapsed_seconds": round(time.perf_counter() - started_at, 3),
|
||
"db_record": db_record,
|
||
}
|
||
except Exception as exc:
|
||
cars_failed = 1
|
||
status = "failed"
|
||
error_summary = str(exc)
|
||
raise
|
||
finally:
|
||
self.persistence.finish_sync_run(
|
||
run_id,
|
||
status=status,
|
||
ids_fetched=ids_fetched,
|
||
cars_upserted=cars_upserted,
|
||
cars_failed=cars_failed,
|
||
images_upserted=images_upserted,
|
||
error_summary=error_summary,
|
||
)
|
||
|
||
# ── Настройки sync_batch ──
|
||
_HTTP_MICRO_BATCH = 25 # URL за микро-батч
|
||
_HTTP_MAX_RETRIES = 2 # Повторов на URL до перехода в браузер
|
||
_HTTP_RETRY_DELAYS = (0.4, 1.0) # Задержки между повторами
|
||
_FALLBACK_PARALLEL_PAGES = 4 # Параллельных вкладок для fallback
|
||
_FALLBACK_NAV_TIMEOUT_MS = 8000 # Таймаут навигации fallback
|
||
# gzip+deflate: нативно в urllib3 (zlib, ~0 CPU). Brotli отключён намеренно —
|
||
# python-brotli декодирует на CPU и под 32 параллельными потоками становится
|
||
# узким местом (хуже, чем гонять чуть больше байт по сети).
|
||
_ACCEPT_ENCODING = "gzip, deflate"
|
||
|
||
def _browser_fallback_parallel(
|
||
self,
|
||
fallback_urls: list[tuple[str, int]],
|
||
total: int,
|
||
n_pages: int,
|
||
) -> dict:
|
||
records: list[CarRecord] = []
|
||
failures: list[dict[str, str]] = []
|
||
cars_failed = 0
|
||
protection_events = 0
|
||
|
||
slices: list[list[tuple[str, int]]] = [[] for _ in range(n_pages)]
|
||
for i, item in enumerate(fallback_urls):
|
||
slices[i % n_pages].append(item)
|
||
|
||
def _process_slice(url_slice: list[tuple[str, int]]) -> dict:
|
||
local_records: list[CarRecord] = []
|
||
local_failures: list[dict[str, str]] = []
|
||
local_failed = 0
|
||
local_protection = 0
|
||
page: Page | None = None
|
||
try:
|
||
for url, global_idx in url_slice:
|
||
if page is None:
|
||
page = self._get_page()
|
||
if self.settings.block_resources:
|
||
BrowserFactory.enable_resource_blocking(page)
|
||
|
||
try:
|
||
page.goto(
|
||
url,
|
||
wait_until="commit",
|
||
timeout=self._FALLBACK_NAV_TIMEOUT_MS,
|
||
)
|
||
except Exception as exc:
|
||
if self._is_protection_or_network_error(exc):
|
||
local_protection += 1
|
||
local_failed += 1
|
||
local_failures.append({"vehicle_url": url, "error": str(exc)})
|
||
logger.error("[%d/%d] Failed to open %s: %s", global_idx, total, url, exc)
|
||
try:
|
||
page.close()
|
||
except Exception:
|
||
pass
|
||
page = None
|
||
continue
|
||
|
||
try:
|
||
js_data: dict = {}
|
||
try:
|
||
js_data = page.evaluate(self._JS_EXTRACT) or {}
|
||
except Exception:
|
||
pass
|
||
|
||
if not js_data.get("ok"):
|
||
try:
|
||
page.wait_for_load_state("domcontentloaded", timeout=3_000)
|
||
except PlaywrightTimeoutError:
|
||
pass
|
||
try:
|
||
js_data = page.evaluate(self._JS_EXTRACT) or {}
|
||
except Exception:
|
||
pass
|
||
|
||
if js_data.get("ok"):
|
||
vehicle_summary = self._build_car_from_js(js_data, url)
|
||
has_identity = bool(vehicle_summary.get("make") or vehicle_summary.get("lot_number"))
|
||
if not has_identity:
|
||
raise SiteStructureChangedError(
|
||
f"JS extraction returned no vehicle identity for {url}"
|
||
)
|
||
db_record = self.car_mapper.map_to_car_record(
|
||
vehicle_url=url,
|
||
vehicle_summary=vehicle_summary,
|
||
payload_insights=self._build_payload_insights(vehicle_summary),
|
||
)
|
||
else:
|
||
try:
|
||
page.wait_for_selector(
|
||
"#VehicleDetailViewModel, .veh-details, .vehicle-details, "
|
||
"[data-uname='vehicleDetailPage']",
|
||
timeout=400,
|
||
)
|
||
except PlaywrightTimeoutError:
|
||
pass
|
||
page_html = page.content()
|
||
try:
|
||
dom_text = page.evaluate("() => document.body?.textContent || ''")
|
||
except Exception:
|
||
dom_text = ""
|
||
network_dump = {
|
||
"requests": [],
|
||
"json_responses": [],
|
||
"capture_limits": {},
|
||
}
|
||
parsed = self.vehicle_parser.normalize(url, page_html, dom_text, network_dump)
|
||
self._raise_if_blocked_or_incomplete(parsed, url)
|
||
db_record = self.car_mapper.map_to_car_record(
|
||
vehicle_url=url,
|
||
vehicle_summary=parsed.get("vehicle_summary", {}),
|
||
payload_insights=parsed.get("payload_insights", {}),
|
||
)
|
||
|
||
record = CarRecord.model_validate(db_record.model_dump(mode="json"))
|
||
local_records.append(record)
|
||
logger.debug(
|
||
"[%d/%d] Parsed %s %s %s (fallback)",
|
||
global_idx, total, record.brand, record.model, record.year or "?",
|
||
)
|
||
except Exception as exc:
|
||
if self._is_protection_or_network_error(exc):
|
||
local_protection += 1
|
||
local_failed += 1
|
||
local_failures.append({"vehicle_url": url, "error": str(exc)})
|
||
logger.error("[%d/%d] Failed %s: %s", global_idx, total, url, exc)
|
||
finally:
|
||
if page is not None:
|
||
try:
|
||
page.close()
|
||
except Exception:
|
||
pass
|
||
return {
|
||
"records": local_records,
|
||
"failures": local_failures,
|
||
"cars_failed": local_failed,
|
||
"protection_events": local_protection,
|
||
}
|
||
|
||
# Одна страница — в главном потоке.
|
||
if n_pages == 1:
|
||
res = _process_slice(slices[0] if slices else [])
|
||
records.extend(res["records"])
|
||
failures.extend(res["failures"])
|
||
cars_failed += res["cars_failed"]
|
||
protection_events += res["protection_events"]
|
||
else:
|
||
# Несколько страниц — параллельно, каждый поток со своей Page.
|
||
with ThreadPoolExecutor(max_workers=n_pages) as executor:
|
||
futures = [executor.submit(_process_slice, s) for s in slices if s]
|
||
for fut in as_completed(futures):
|
||
res = fut.result()
|
||
records.extend(res["records"])
|
||
failures.extend(res["failures"])
|
||
cars_failed += res["cars_failed"]
|
||
protection_events += res["protection_events"]
|
||
|
||
return {
|
||
"records": records,
|
||
"failures": failures,
|
||
"cars_failed": cars_failed,
|
||
"protection_events": protection_events,
|
||
}
|
||
|
||
cookie_header_override: str | None = None,
|
||
def sync_batch(
|
||
self,
|
||
vehicle_urls: list[str],
|
||
lane: str = "iaai_cars",
|
||
parallel_tabs: int | None = None,
|
||
*,
|
||
skip_browser_fallback: bool = False,
|
||
cookie_header_override: str | None = None,
|
||
) -> dict:
|
||
trace_id = self._new_trace_id("sync-batch")
|
||
started_at = time.perf_counter()
|
||
num_workers = parallel_tabs or self.settings.parallel_tabs
|
||
num_workers = min(num_workers, len(vehicle_urls), 48)
|
||
|
||
cars_upserted = 0
|
||
cars_failed = 0
|
||
images_upserted = 0
|
||
records: list[CarRecord] = []
|
||
# cookie_header_override передаётся из главного потока при pipeline-вызове —
|
||
# вычисление cookie_header из background thread'а ломает Playwright sync greenlet.
|
||
cookie_header = (
|
||
cookie_header_override
|
||
if cookie_header_override is not None
|
||
else self._cookie_header_for_context()
|
||
|
||
http_successes = 0
|
||
http_fallbacks = 0
|
||
protection_events = 0
|
||
|
||
total = len(vehicle_urls)
|
||
logger.info("sync_batch: %d vehicles, %d parallel workers", total, num_workers)
|
||
|
||
# cookie_header_override передаётся из главного потока при pipeline-вызове —
|
||
# вычисление cookie_header из background thread'а ломает Playwright sync greenlet.
|
||
cookie_header = (
|
||
cookie_header_override
|
||
if cookie_header_override is not None
|
||
else self._cookie_header_for_context()
|
||
)
|
||
# UA должен совпадать с UA Playwright-контекста, иначе Imperva видит
|
||
# «cookies от Chrome, ходим под Firefox» — типичный сигнал бота.
|
||
user_agent = self.settings.fingerprint.user_agent
|
||
|
||
# ── Фаза 1: HTTP fast-path ──
|
||
fallback_urls: list[tuple[str, int]] = []
|
||
|
||
def _fetch_one_with_retry(idx_url: tuple[int, str]) -> tuple[int, CarRecord | Exception]:
|
||
idx, url = idx_url
|
||
# Небольшой анти-бёрст джиттер: разносим старт запросов в пуле,
|
||
# чтобы N воркеров не били в одну TLS-/PX-волну.
|
||
time.sleep(random.uniform(0.0, 0.15))
|
||
last_exc: Exception | None = None
|
||
for attempt in range(1 + self._HTTP_MAX_RETRIES):
|
||
try:
|
||
return idx, self._scrape_via_raw_http(
|
||
url,
|
||
cookie_header=cookie_header,
|
||
user_agent=user_agent,
|
||
)
|
||
except Exception as exc:
|
||
last_exc = exc
|
||
if attempt < self._HTTP_MAX_RETRIES:
|
||
time.sleep(self._HTTP_RETRY_DELAYS[min(attempt, len(self._HTTP_RETRY_DELAYS) - 1)])
|
||
return idx, last_exc # type: ignore[return-value]
|
||
|
||
indexed_urls = list(enumerate(vehicle_urls))
|
||
with ThreadPoolExecutor(max_workers=min(num_workers, total)) as executor:
|
||
for idx, result in executor.map(_fetch_one_with_retry, indexed_urls):
|
||
if isinstance(result, Exception):
|
||
http_fallbacks += 1
|
||
logger.debug(
|
||
"[%d/%d] HTTP fast-path failed for %s: %s",
|
||
idx + 1, total, vehicle_urls[idx], result,
|
||
)
|
||
fallback_urls.append((vehicle_urls[idx], idx + 1))
|
||
else:
|
||
records.append(result)
|
||
http_successes += 1
|
||
logger.debug(
|
||
"[%d/%d] Parsed %s %s %s (raw HTTP)",
|
||
idx + 1, total, result.brand, result.model, result.year or "?",
|
||
)
|
||
|
||
logger.info(
|
||
"HTTP phase done: %d OK, %d fallback (%.1fs)",
|
||
http_successes, http_fallbacks, time.perf_counter() - started_at,
|
||
)
|
||
|
||
# ── Фаза 2: браузерный fallback ──
|
||
# При skip_browser_fallback=True (pipeline-режим) возвращаем fallback_urls
|
||
# вызывающему — Playwright занят сбором листинга в главном потоке,
|
||
# его нельзя трогать из фонового executor'а.
|
||
deferred_fallback_urls: list[tuple[str, int]] = []
|
||
if fallback_urls:
|
||
if skip_browser_fallback:
|
||
deferred_fallback_urls = fallback_urls
|
||
logger.info(
|
||
"Deferring browser fallback: %d/%d URLs (pipeline mode)",
|
||
len(fallback_urls), total,
|
||
)
|
||
else:
|
||
fb_pages = min(len(fallback_urls), self._FALLBACK_PARALLEL_PAGES)
|
||
logger.info(
|
||
"Fallback browser mode: %d/%d URLs, %d parallel pages",
|
||
len(fallback_urls), total, fb_pages,
|
||
)
|
||
fb_results = self._browser_fallback_parallel(fallback_urls, total, fb_pages)
|
||
records.extend(fb_results["records"])
|
||
cars_failed += fb_results["cars_failed"]
|
||
protection_events += fb_results["protection_events"]
|
||
failures.extend(fb_results["failures"])
|
||
|
||
# ── Фаза 3: запись в БД ──
|
||
if records:
|
||
seen_origins: set[str] = set()
|
||
unique_records: list[CarRecord] = []
|
||
for rec in records:
|
||
key = rec.origin_id or rec.origin_url
|
||
if key not in seen_origins:
|
||
seen_origins.add(key)
|
||
unique_records.append(rec)
|
||
if len(unique_records) < len(records):
|
||
logger.info("Dedup before DB: %d → %d", len(records), len(unique_records))
|
||
records = unique_records
|
||
|
||
try:
|
||
batch_result = self.persistence.upsert_cars_batch(records)
|
||
cars_upserted = batch_result["inserted"] + batch_result["updated"]
|
||
images_upserted = batch_result["images_upserted"]
|
||
except Exception as exc:
|
||
logger.error("Batch upsert failed: %s", exc)
|
||
cars_failed += len(records)
|
||
|
||
status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
|
||
return {
|
||
"trace_id": trace_id,
|
||
"status": status,
|
||
"cars_upserted": cars_upserted,
|
||
"cars_failed": cars_failed,
|
||
"images_upserted": images_upserted,
|
||
"http_successes": http_successes,
|
||
"http_fallbacks": http_fallbacks,
|
||
"protection_events": protection_events,
|
||
"elapsed_seconds": round(time.perf_counter() - started_at, 3),
|
||
"failures": failures,
|
||
"deferred_fallback_urls": deferred_fallback_urls,
|
||
}
|
||
|
||
def sync_listing(
|
||
self,
|
||
make: str | None = None,
|
||
model: str | None = None,
|
||
lane: str = "iaai_cars",
|
||
limit: int | None = None,
|
||
only_new: bool | None = None,
|
||
listing_url: str | None = None,
|
||
year_min: int | None = None,
|
||
year_max: int | None = None,
|
||
):
|
||
# runtime_config — дефолты; CLI/API аргументы приоритетнее.
|
||
rc = self.runtime_config.sync
|
||
if limit is None and rc.limit is not None:
|
||
limit = rc.limit
|
||
if only_new is None and rc.only_new is not None:
|
||
only_new = rc.only_new
|
||
if lane == "iaai_cars" and rc.lane is not None:
|
||
lane = rc.lane
|
||
|
||
trace_id = self._new_trace_id("sync-listing")
|
||
started_at = time.perf_counter()
|
||
self.persistence.create_tables()
|
||
run_id = self.persistence.start_sync_run(lane=lane)
|
||
cars_upserted = 0
|
||
cars_failed = 0
|
||
cars_filtered = 0
|
||
images_upserted = 0
|
||
total = 0
|
||
skipped_existing = 0
|
||
is_partial_scan = True
|
||
failures: list[dict[str, str]] = []
|
||
listing: dict = {}
|
||
|
||
try:
|
||
effective_only_new = self.settings.sync_only_new if only_new is None else only_new
|
||
stream_result = self._sync_listing_streaming(
|
||
make=make,
|
||
model=model,
|
||
lane=lane,
|
||
limit=limit,
|
||
effective_only_new=effective_only_new,
|
||
started_at=started_at,
|
||
listing_url=listing_url,
|
||
year_min=year_min,
|
||
year_max=year_max,
|
||
)
|
||
listing = stream_result["listing"]
|
||
total = stream_result["total"]
|
||
skipped_existing = stream_result["skipped_existing"]
|
||
cars_upserted = stream_result["cars_upserted"]
|
||
cars_failed = stream_result["cars_failed"]
|
||
images_upserted = stream_result["images_upserted"]
|
||
failures.extend(stream_result["failures"])
|
||
all_listing_origin_urls = stream_result["all_listing_origin_urls"]
|
||
|
||
logger.info("Streaming sync processed %d vehicles", total)
|
||
|
||
# Помечаем проданные авто, исчезнувшие из листинга (только при полном скане).
|
||
is_partial_scan = (
|
||
effective_only_new
|
||
or (limit is not None and limit > 0)
|
||
or listing.get("early_stopped", False)
|
||
or listing.get("truncated_by_time_budget", False)
|
||
)
|
||
if all_listing_origin_urls and not is_partial_scan:
|
||
try:
|
||
sold_count = self.persistence.mark_sold_not_in_listing_by_urls(all_listing_origin_urls)
|
||
if sold_count:
|
||
logger.info("Marked %d cars as sold", sold_count)
|
||
except Exception as exc:
|
||
logger.warning("Failed to mark sold cars: %s", exc)
|
||
elif is_partial_scan:
|
||
logger.debug("Skipping mark_sold: partial/incremental scan (only_new=%s, limit=%s, early_stopped=%s)",
|
||
effective_only_new, limit, listing.get("early_stopped", False))
|
||
except Exception as exc:
|
||
if not failures:
|
||
failures.append({"vehicle_url": "collect_listing", "error": str(exc)})
|
||
logger.error("sync_listing failed: %s (partial progress: %d upserted)", exc, cars_upserted)
|
||
finally:
|
||
status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
|
||
error_summary = "; ".join(item["error"] for item in failures[:10]) if failures else None
|
||
self.persistence.finish_sync_run(
|
||
run_id,
|
||
status=status,
|
||
ids_fetched=total,
|
||
cars_upserted=cars_upserted,
|
||
cars_failed=cars_failed,
|
||
images_upserted=images_upserted,
|
||
error_summary=error_summary,
|
||
)
|
||
|
||
logger.info(
|
||
"Sync run #%d finished: %d/%d upserted, %d failed, %d filtered, %d images",
|
||
run_id, cars_upserted, total, cars_failed, cars_filtered, images_upserted,
|
||
)
|
||
return {
|
||
"trace_id": trace_id,
|
||
"status": status,
|
||
"run_id": run_id,
|
||
# partial_success допустим — отдельные машины могли не спарситься,
|
||
# это не повод повторять весь bootstrap.
|
||
"full_scan_completed": (not is_partial_scan) and status in ("success", "partial_success"),
|
||
"only_new_effective": effective_only_new,
|
||
"listing": listing,
|
||
"cars_upserted": cars_upserted,
|
||
"cars_failed": cars_failed,
|
||
"cars_filtered": cars_filtered,
|
||
"images_upserted": images_upserted,
|
||
"skipped_existing": skipped_existing,
|
||
"elapsed_seconds": round(time.perf_counter() - started_at, 3),
|
||
"failures": failures,
|
||
}
|
||
|
||
def sync_listing_segmented(
|
||
self,
|
||
segments: list[dict[str, Any]],
|
||
lane: str = "iaai_cars",
|
||
only_new: bool | None = None,
|
||
start_segment: int = 0,
|
||
start_page: int = 1,
|
||
progress_callback: Callable[[int], None] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Итеративный sync_listing по списку сегментов (бренд / бренд+годы).
|
||
|
||
Args:
|
||
segments: список dict с ключами make, year_min, year_max.
|
||
start_segment: индекс сегмента для resume (0-based).
|
||
start_page: не используется (остаётся для обратной совместимости API).
|
||
progress_callback: вызывается (segment_index) после каждого ПОЛНОСТЬЮ пройденного сегмента.
|
||
"""
|
||
trace_id = self._new_trace_id("sync-segmented")
|
||
started_at = time.perf_counter()
|
||
base_url = self.settings.listing.cars_url
|
||
_ = start_page # обратная совместимость — page-level resume удалён
|
||
|
||
total_cars_upserted = 0
|
||
total_cars_failed = 0
|
||
total_images_upserted = 0
|
||
total_skipped = 0
|
||
total_discovered = 0
|
||
all_failures: list[dict[str, str]] = []
|
||
segment_results: list[dict[str, Any]] = []
|
||
completed_all = True
|
||
|
||
logger.warning(
|
||
"Starting segmented sync: %d segments, resume from segment=%d",
|
||
len(segments), start_segment,
|
||
)
|
||
|
||
for seg_idx in range(start_segment, len(segments)):
|
||
seg = segments[seg_idx]
|
||
seg_make = seg.get("make")
|
||
seg_year_min = seg.get("year_min")
|
||
seg_year_max = seg.get("year_max")
|
||
seg_url = self._build_segment_listing_url(base_url, seg_make) if seg_make else None
|
||
|
||
seg_label = f"{seg_make or 'ALL'}"
|
||
if seg_year_min is not None or seg_year_max is not None:
|
||
seg_label += f" ({seg_year_min}-{seg_year_max})"
|
||
|
||
logger.warning(
|
||
"Segment %d/%d: %s",
|
||
seg_idx + 1, len(segments), seg_label,
|
||
)
|
||
|
||
try:
|
||
result = self.sync_listing(
|
||
make=None if seg_url else seg_make,
|
||
model=None,
|
||
lane=lane,
|
||
only_new=only_new,
|
||
listing_url=seg_url,
|
||
year_min=seg_year_min,
|
||
year_max=seg_year_max,
|
||
)
|
||
total_cars_upserted += result.get("cars_upserted", 0)
|
||
total_cars_failed += result.get("cars_failed", 0)
|
||
total_images_upserted += result.get("images_upserted", 0)
|
||
total_skipped += result.get("skipped_existing", 0)
|
||
total_discovered += result.get("listing", {}).get("vehicles_collected", 0)
|
||
all_failures.extend(result.get("failures", []))
|
||
segment_results.append({
|
||
"segment": seg,
|
||
"segment_index": seg_idx,
|
||
"status": result.get("status"),
|
||
"full_scan_completed": bool(result.get("full_scan_completed", False)),
|
||
"cars_upserted": result.get("cars_upserted", 0),
|
||
"cars_failed": result.get("cars_failed", 0),
|
||
"vehicles_collected": result.get("listing", {}).get("vehicles_collected", 0),
|
||
})
|
||
|
||
segment_done = bool(result.get("full_scan_completed", False))
|
||
if not segment_done:
|
||
completed_all = False
|
||
|
||
# Сегмент полностью пройден — фиксируем чекпоинт, даже если часть машин failed.
|
||
if segment_done and progress_callback is not None:
|
||
try:
|
||
progress_callback(seg_idx)
|
||
except Exception:
|
||
logger.warning(
|
||
"Failed to persist segment checkpoint for segment=%d",
|
||
seg_idx, exc_info=True,
|
||
)
|
||
|
||
logger.warning(
|
||
"Segment %d/%d done: %s → upserted=%d, failed=%d, collected=%d",
|
||
seg_idx + 1, len(segments), seg_label,
|
||
result.get("cars_upserted", 0),
|
||
result.get("cars_failed", 0),
|
||
result.get("listing", {}).get("vehicles_collected", 0),
|
||
)
|
||
except Exception as exc:
|
||
logger.error("Segment %d/%d failed: %s — %s", seg_idx + 1, len(segments), seg_label, exc)
|
||
all_failures.append({"vehicle_url": f"segment_{seg_idx}_{seg_label}", "error": str(exc)})
|
||
completed_all = False
|
||
# Продолжаем оставшиеся сегменты — одна ошибка не должна убивать весь прогон
|
||
continue
|
||
|
||
elapsed = round(time.perf_counter() - started_at, 3)
|
||
status = "success" if not all_failures else "partial_success" if total_cars_upserted else "failed"
|
||
|
||
logger.warning(
|
||
"Segmented sync done: %d/%d segments, upserted=%d, failed=%d, discovered=%d, elapsed=%.1fs",
|
||
len(segment_results), len(segments),
|
||
total_cars_upserted, total_cars_failed, total_discovered, elapsed,
|
||
)
|
||
|
||
return {
|
||
"trace_id": trace_id,
|
||
"status": status,
|
||
# Bootstrap считается завершённым если все сегменты пройдены,
|
||
# даже если часть машин failed (они будут обновлены в следующих циклах).
|
||
"full_scan_completed": completed_all,
|
||
"segments_total": len(segments),
|
||
"segments_completed": len(segment_results),
|
||
"cars_upserted": total_cars_upserted,
|
||
"cars_failed": total_cars_failed,
|
||
"images_upserted": total_images_upserted,
|
||
"skipped_existing": total_skipped,
|
||
"total_discovered": total_discovered,
|
||
"elapsed_seconds": elapsed,
|
||
"failures": all_failures,
|
||
"segment_results": segment_results,
|
||
}
|
||
|
||
def run_scheduled(self) -> None:
|
||
"""Standalone-планировщик (в продакшене используется Celery beat)."""
|
||
def _handle_shutdown(signum, frame):
|
||
logger.info("Received signal %s, shutting down gracefully...", signum)
|
||
self._shutdown_requested = True
|
||
|
||
signal.signal(signal.SIGINT, _handle_shutdown)
|
||
signal.signal(signal.SIGTERM, _handle_shutdown)
|
||
|
||
interval = self.settings.scheduler_interval_minutes * 60
|
||
logger.info(
|
||
"Scheduler started: syncing every %d minutes",
|
||
self.settings.scheduler_interval_minutes,
|
||
)
|
||
cycle = 0
|
||
while not self._shutdown_requested:
|
||
cycle += 1
|
||
logger.info("Scheduler cycle #%d starting", cycle)
|
||
start = time.time()
|
||
try:
|
||
if self.context:
|
||
try:
|
||
self.context.close()
|
||
except PlaywrightError:
|
||
pass
|
||
self.context = None
|
||
|
||
if self.browser is None or self.playwright is None:
|
||
logger.info("Browser/Playwright not available, re-initializing...")
|
||
self.close()
|
||
self.__enter__()
|
||
|
||
result = self.sync_listing()
|
||
elapsed = time.time() - start
|
||
logger.info(
|
||
"Cycle #%d done in %.1fs: %d upserted, %d failed",
|
||
cycle, elapsed,
|
||
result.get("cars_upserted", 0),
|
||
result.get("cars_failed", 0),
|
||
)
|
||
except Exception as exc:
|
||
elapsed = time.time() - start
|
||
logger.error("Cycle #%d failed after %.1fs: %s", cycle, elapsed, exc)
|
||
try:
|
||
self.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.__enter__()
|
||
except Exception as reinit_exc:
|
||
logger.error("Failed to re-initialize browser: %s", reinit_exc)
|
||
|
||
if self._shutdown_requested:
|
||
break
|
||
|
||
sleep_time = max(0, interval - (time.time() - start))
|
||
if sleep_time > 0:
|
||
logger.info("Sleeping %.0f seconds until next cycle...", sleep_time)
|
||
slept = 0.0
|
||
while slept < sleep_time and not self._shutdown_requested:
|
||
chunk = min(5.0, sleep_time - slept)
|
||
time.sleep(chunk)
|
||
slept += chunk
|
||
|
||
logger.info("Scheduler stopped gracefully after %d cycles.", cycle)
|