import json import logging import os import re import signal import time import uuid import html as html_module from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlsplit, urlunsplit from urllib.request import Request, urlopen import urllib3 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 from .core.exceptions import AntiBotDetectedError, 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 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)[^>]*>.*?", re.IGNORECASE | re.DOTALL) 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-клиент для fast-path. # При наличии прокси используем ProxyManager. proxy_url = self.settings.proxy.server if proxy_url: _proxy_kwargs: dict = { "num_pools": 4, "maxsize": 64, "retries": False, "timeout": urllib3.Timeout(connect=5, read=10), } 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=4, maxsize=64, retries=False, timeout=urllib3.Timeout(connect=5, read=10), ) 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 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) # Ждём domcontentloaded вместо networkidle. 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 # Короткая пауза для установки cookies. 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]: # Нормализация и дедупликация URL. 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. 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 _collect_listing_iterative( self, *, make: str | None = None, model: str | None = None, limit: int, max_duration_seconds: float | None = None, ) -> tuple[list[str], list[str], dict, int]: # Поэтапный сбор листинга до нужного лимита новых URL. new_urls: list[str] = [] all_raw_urls: list[str] = [] seen: set[str] = set() skipped_existing = 0 pages_info: list[dict] = [] max_pages = self.settings.listing.max_pages_per_run started_at = time.perf_counter() truncated_by_time_budget = False page = self._get_page_with_warmup() try: self.listing_collector.open_cars_listing(page) self.listing_collector.apply_filters(page, make=make, model=model) for page_number in range(1, max_pages + 1): if max_duration_seconds is not None and max_duration_seconds > 0: elapsed = time.perf_counter() - started_at if elapsed >= max_duration_seconds: truncated_by_time_budget = True logger.warning( "Iterative listing stopped by time budget: page=%d elapsed=%.1fs budget=%.1fs", page_number, elapsed, max_duration_seconds, ) break page_result = self.listing_collector.collect_current_page(page, page_number=page_number) pages_info.append({ "page_number": page_result.page_number, "links_found": len(page_result.vehicle_links), }) # Собираем URL со страницы. 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 not in seen: seen.add(normalized) page_urls.append(normalized) if not page_urls: # Страница 1 пустая — скорее всего transient network issue. # Пробуем перезагрузить листинг ещё раз. if page_number == 1: logger.warning("Page 1 returned 0 links — retrying listing open...") time.sleep(3) self.listing_collector.open_cars_listing(page) page_result = self.listing_collector.collect_current_page(page, page_number=page_number) for item in page_result.vehicle_links: normalized = self._normalize_vehicle_url(item.href) all_raw_urls.append(item.href) if normalized not in seen: seen.add(normalized) page_urls.append(normalized) if not page_urls: logger.info("Page %d: 0 new links, stopping pagination", page_number) break # Фильтруем known. fresh, page_skipped = self._filter_known_urls(page_urls) skipped_existing += page_skipped new_urls.extend(fresh) logger.info( "Page %d: %d links, %d new, %d known (total new: %d/%d)", page_number, len(page_urls), len(fresh), page_skipped, len(new_urls), limit, ) # Стопаем только если набрали нужное количество И на этой странице уже нет новых. # Если последняя страница дала новые — проверяем следующую (там могут быть ещё). if len(new_urls) >= limit and len(fresh) == 0: break if len(new_urls) >= limit: # Набрали достаточно, дальше не листаем break if not page_result.next_page_detected: logger.info("No next page detected, stopping") break if not self.listing_collector.go_to_next_page(page): logger.info("Failed to navigate to next page, stopping") break finally: page.close() # Обрезаем до limit. new_urls = new_urls[:limit] listing = { "status": "ok", "listing_url": self.settings.listing.cars_url, "pages_collected": len(pages_info), "vehicles_collected": len(all_raw_urls), "vehicle_urls": all_raw_urls, "early_stopped": False, "truncated_by_time_budget": truncated_by_time_budget, "pages": pages_info, } return new_urls, all_raw_urls, listing, skipped_existing 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 _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): # Открыть страницу, перехватить JSON, вернуть данные. page = self._get_page() try: return self._scrape_on_page(page, vehicle_url) finally: page.close() # Быстрый извлекатель данных из inline JSON. _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: # Сбор vehicle_summary из JS-данных. 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: # Сбор payload_insights из vehicle_summary. 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) # На VPS достаточно domcontentloaded. try: page.wait_for_load_state("domcontentloaded", timeout=5_000) except PlaywrightTimeoutError: pass # Быстрый путь: читаем данные из inline JSON. 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 # Резервный путь: полный парсинг страницы. 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 вместо посимвольного сканирования скобок. """ search = "inventoryView" pos = html.find(search) if pos < 0: return None # Ищем начало внешнего JSON-объекта. script_start = html.rfind("", script_start) if tag_end < 0: return None content_start = html.find("{", tag_end) if content_start < 0: return None # Быстрый поиск конца JSON через raw_decode. 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: # Быстрый запрос через context.request. 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() # Пытаемся извлечь JSON напрямую из HTML. 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).""" 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", "Cache-Control": "no-cache", "Pragma": "no-cache", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", } if cookie_header: headers["Cookie"] = cookie_header response = self._http_pool.request("GET", vehicle_url, headers=headers) 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"): # Скрапинг и upsert одного авто. 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, ) # ── Tunables for sync_batch ── _HTTP_MICRO_BATCH = 25 # URLs per micro-batch (avoid mass rate-limit) _HTTP_MAX_RETRIES = 2 # Retries per URL before giving up to browser _HTTP_RETRY_DELAYS = (0.4, 1.0) # Backoff between retries _FALLBACK_PARALLEL_PAGES = 4 # Concurrent browser tabs for fallback _FALLBACK_NAV_TIMEOUT_MS = 8000 # Reduced from 15 000 def _browser_fallback_parallel( self, fallback_urls: list[tuple[str, int]], total: int, n_pages: int, ) -> dict: # Обработка fallback URL через браузер. records: list[CarRecord] = [] failures: list[dict[str, str]] = [] cars_failed = 0 protection_events = 0 # Деление URL по страницам. 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, } # Один page обрабатываем в главном потоке. 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, } def sync_batch( self, vehicle_urls: list[str], lane: str = "iaai_cars", parallel_tabs: int | None = None, ) -> dict: # Пакетный скрапинг списка URL. 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] = [] failures: list[dict[str, str]] = [] 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) # Подготовка для HTTP fast-path: cookies + user-agent из Playwright сессии. cookie_header = self._cookie_header_for_context() user_agent = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) " "Gecko/20100101 Firefox/128.0" ) # ── Phase 1: HTTP fast-path — все URLs одним ThreadPoolExecutor ── fallback_urls: list[tuple[str, int]] = [] def _fetch_one_with_retry(idx_url: tuple[int, str]) -> tuple[int, CarRecord | Exception]: idx, url = idx_url 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] # Запускаем все URLs сразу — один пул потоков для минимального времени ожидания. 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, ) # ── Phase 2: browser fallback (single-threaded — Playwright sync API is not thread-safe) ── if fallback_urls: logger.info( "Fallback browser mode: %d/%d URLs, single page", len(fallback_urls), total, ) fb_results = self._browser_fallback_parallel(fallback_urls, total, 1) records.extend(fb_results["records"]) cars_failed += fb_results["cars_failed"] protection_events += fb_results["protection_events"] failures.extend(fb_results["failures"]) # ── Phase 3: single DB flush ── 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, } 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, ): # Листинг + sync всех найденных машин. # Применяем 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 failures: list[dict[str, str]] = [] listing: dict = {} try: effective_only_new = self.settings.sync_only_new if only_new is None else only_new soft_time_limit = self.settings.celery.task_soft_time_limit listing_time_budget_s: float | None = None if soft_time_limit and soft_time_limit > 0: # Делим время: ~40% на листинг, ~60% на парсинг + запись. # При soft=120 → listing=48s, processing=72s (достаточно для ~7 батчей). # При soft=600 → listing=240s, processing=360s (достаточно для ~30+ батчей). # Минимум 30s на листинг, оставляем 30s запас до hard limit. soft_f = float(soft_time_limit) listing_time_budget_s = max(30.0, soft_f * 0.4) # ── Сбор листинга ── # При only_new + limit используем итеративный подход: # листаем страницы одну за другой, фильтруем known на лету, # останавливаемся когда набрали limit новых. original_listing_cap = self.settings.listing.max_vehicles_per_run original_include_pagination = self.settings.listing.include_pagination original_collect_current_page_only = self.settings.listing.collect_current_page_only original_max_pages_per_run = self.settings.listing.max_pages_per_run if effective_only_new and limit is not None and limit > 0: # Ограничиваем кол-во URL тем, что реально успеем обработать. # При soft=600: processing=360s, ~3s на батч из 10 → ~120 батчей → ~1200 машин. effective_limit = limit if soft_time_limit and soft_time_limit > 0 and listing_time_budget_s is not None: processing_budget_s = max(60.0, float(soft_time_limit) - listing_time_budget_s) seconds_per_batch = 4.0 # ~3-4s на batch из 10 через HTTP fast-path batches_fit = max(1, int(processing_budget_s / seconds_per_batch)) max_processable = batches_fit * self.settings.celery.batch_size if max_processable < effective_limit: logger.info( "Capping iterative limit %d → %d (processing_budget=%.0fs, batch_size=%d)", effective_limit, max_processable, processing_budget_s, self.settings.celery.batch_size, ) effective_limit = max_processable # Итеративный сбор: страница→фильтр→проверка→следующая страница. vehicle_urls, raw_urls, listing, skipped_existing = self._collect_listing_iterative( make=make, model=model, limit=effective_limit, max_duration_seconds=listing_time_budget_s, ) else: # Обычный сбор (без only_new или без limit). prefetch_cap: int | None = None if limit is not None and limit > 0: prefetch_cap = limit elif soft_time_limit and soft_time_limit > 0 and listing_time_budget_s is not None: # При ограниченном soft-time-limit не раздуваем prefetch, # чтобы быстрее перейти к парсингу/записи. remaining_processing_s = max(30.0, float(soft_time_limit) - listing_time_budget_s) batches_fit = max(2, int(remaining_processing_s // 60)) prefetch_cap = max(self.settings.celery.batch_size, self.settings.celery.batch_size * batches_fit) if prefetch_cap is not None and prefetch_cap < original_listing_cap: self.settings.listing.max_vehicles_per_run = prefetch_cap # Если включён режим только новых — заранее загружаем все известные origin_id, # чтобы listing_collector мог остановиться при встрече старых страниц. known_origin_ids: set[str] | None = None if effective_only_new and self.settings.listing.early_stop_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) try: listing = self.collect_listing( make=make, model=model, known_origin_ids=known_origin_ids, max_duration_seconds=listing_time_budget_s, ) finally: self.settings.listing.max_vehicles_per_run = original_listing_cap self.settings.listing.include_pagination = original_include_pagination self.settings.listing.collect_current_page_only = original_collect_current_page_only self.settings.listing.max_pages_per_run = original_max_pages_per_run raw_urls = list(listing.get("vehicle_urls", [])) vehicle_urls = self._dedupe_urls(raw_urls) if effective_only_new: vehicle_urls, skipped_existing = self._filter_known_urls(vehicle_urls) if limit is not None: vehicle_urls = vehicle_urls[:max(0, limit)] total = len(vehicle_urls) if listing.get("truncated_by_time_budget"): logger.warning( "Listing was truncated by time budget; processing collected subset (%d URLs)", total, ) logger.info("Starting sync: %d vehicles to process (batch mode)", total) # Собираем нормализованные origin_url для последующей пометки проданных. # Используем URL (а не origin_id из URL), т.к. в БД origin_id берётся из parsed lot_number, # который может отличаться от числа в URL листинга. all_listing_origin_urls = set() for raw_url in raw_urls: normalized_url = self._normalize_vehicle_url(raw_url) if normalized_url: all_listing_origin_urls.add(normalized_url) # Обрабатываем пакетами. # Каждый batch сохраняется в БД сразу — partial progress не теряется при crash. batch_size = self.settings.celery.batch_size for batch_start in range(0, total, batch_size): # Проверяем оставшееся время перед каждым batch. if soft_time_limit and soft_time_limit > 0: elapsed = time.perf_counter() - started_at remaining = float(soft_time_limit) - elapsed if remaining < 30: logger.warning( "Time budget exhausted before batch %d-%d (elapsed=%.1fs, soft=%ds). " "Saving partial progress: %d upserted so far.", batch_start + 1, min(batch_start + batch_size, total), elapsed, soft_time_limit, cars_upserted, ) break batch_urls = vehicle_urls[batch_start:batch_start + batch_size] logger.info( "Processing batch %d-%d of %d", batch_start + 1, min(batch_start + batch_size, total), total, ) 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 PlaywrightError as pw_exc: # Browser crashed — пробуем пересоздать контекст и продолжить. logger.warning( "Batch %d-%d browser crashed: %s. Reinitializing browser context.", batch_start + 1, min(batch_start + batch_size, total), pw_exc, ) cars_failed += len(batch_urls) failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(pw_exc)}) try: self._new_context() logger.info("Browser context reinitialized successfully after crash") except Exception as reinit_exc: logger.error("Failed to reinitialize browser: %s. Aborting remaining batches.", reinit_exc) break except Exception as batch_exc: # Batch failed — логируем и не теряем уже накопленный прогресс. logger.error( "Batch %d-%d failed: %s. Continuing with next batch.", batch_start + 1, min(batch_start + batch_size, total), batch_exc, ) cars_failed += len(batch_urls) failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(batch_exc)}) # Помечаем авто как проданные, если они исчезли из листинга. # Только если сканирование было полным (не ограниченным limit/only_new/early_stop). is_partial_scan = ( effective_only_new or (limit is not None and limit > 0) or listing.get("early_stopped", 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) # Не пробрасываем — partial progress уже записан в БД через finish_sync_run ниже. 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, "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 run_scheduled(self) -> None: # Резерв для standalone-режима. # В текущей архитектуре планирование выполняется через Celery beat + worker/tasks.py. 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)