155 lines
7.1 KiB
Python
155 lines
7.1 KiB
Python
import logging
|
||
import re
|
||
from dataclasses import asdict, dataclass, field
|
||
from urllib.parse import urljoin
|
||
|
||
from playwright.sync_api import Page
|
||
|
||
from ..browser.pace import HumanPacer
|
||
from ..core.config import Settings
|
||
from ..core.utils import first_non_empty
|
||
|
||
logger = logging.getLogger("iaai_scraper.listing")
|
||
VEHICLE_HREF_RE = re.compile(r"/VehicleDetail/\d+(?:~[A-Z]{2})?", re.IGNORECASE)
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class ListingVehicleLink:
|
||
# Ссылка на одну карточку из листинга.
|
||
href: str
|
||
title: str = ""
|
||
lot_number: str | None = None
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class ListingPageResult:
|
||
# Результат сбора ссылок с одной страницы листинга.
|
||
source_url: str
|
||
page_number: int
|
||
vehicle_links: list[ListingVehicleLink] = field(default_factory=list)
|
||
pagination_available: bool = False
|
||
next_page_detected: bool = False
|
||
|
||
|
||
class ListingCollector:
|
||
def __init__(self, settings: Settings, pacer: HumanPacer) -> None:
|
||
# Сборщик ссылок из публичного листинга IAAI.
|
||
self.settings = settings
|
||
self.pacer = pacer
|
||
|
||
def open_cars_listing(self, page: Page) -> None:
|
||
# Открываем листинг и ждём появления ссылок на карточки.
|
||
logger.info("Opening cars listing page: %s", self.settings.listing.cars_url)
|
||
page.goto(self.settings.listing.cars_url, wait_until="domcontentloaded")
|
||
try:
|
||
page.wait_for_load_state("networkidle", timeout=15000)
|
||
except Exception:
|
||
logger.debug("networkidle timeout on listing page, continuing with current state")
|
||
try:
|
||
page.wait_for_selector("a[href*='/VehicleDetail/']", timeout=20000)
|
||
except Exception:
|
||
logger.warning("Vehicle links did not appear within timeout; page may not have rendered fully")
|
||
self.pacer.after_listing_open()
|
||
|
||
def apply_filters(self, page: Page, make: str | None = None, model: str | None = None) -> dict[str, str | None]:
|
||
# Пытаемся применить make/model фильтры через UI.
|
||
applied = {"make": None, "model": None}
|
||
if make and self._try_fill_filter_input(page, ["input[placeholder*='Make']", "input[aria-label*='Make']"], make):
|
||
applied["make"] = make
|
||
self.pacer.after_filter_action()
|
||
if model and self._try_fill_filter_input(page, ["input[placeholder*='Model']", "input[aria-label*='Model']"], model):
|
||
applied["model"] = model
|
||
self.pacer.after_filter_action()
|
||
return applied
|
||
|
||
def collect_current_page(self, page: Page, page_number: int = 1) -> ListingPageResult:
|
||
# Собираем ссылки только с текущей страницы.
|
||
anchors = page.locator("a[href*='/VehicleDetail/']")
|
||
total = min(anchors.count(), self.settings.listing.page_link_limit)
|
||
links: list[ListingVehicleLink] = []
|
||
seen: set[str] = set()
|
||
for idx in range(total):
|
||
anchor = anchors.nth(idx)
|
||
href = anchor.get_attribute("href") or ""
|
||
match = VEHICLE_HREF_RE.search(href)
|
||
if not match:
|
||
continue
|
||
absolute = urljoin(self.settings.home_url, match.group(0))
|
||
if absolute in seen:
|
||
continue
|
||
seen.add(absolute)
|
||
title = first_non_empty([anchor.get_attribute("title"), anchor.text_content(), ""]) or ""
|
||
links.append(ListingVehicleLink(href=absolute, title=str(title).strip()))
|
||
if len(links) >= self.settings.listing.max_vehicles_per_run:
|
||
break
|
||
next_page_detected = self._has_next_page(page)
|
||
return ListingPageResult(source_url=page.url, page_number=page_number, vehicle_links=links, pagination_available=next_page_detected, next_page_detected=next_page_detected)
|
||
|
||
def go_to_next_page(self, page: Page) -> bool:
|
||
# Переход на следующую страницу, если кнопка доступна.
|
||
selectors = ["a[aria-label*='Next']", "button[aria-label*='Next']", "a.pagination-next", "button.pagination-next", "a:has-text('Next')", "button:has-text('Next')"]
|
||
for selector in selectors:
|
||
locator = page.locator(selector).first
|
||
if locator.count() == 0:
|
||
continue
|
||
disabled = (locator.get_attribute("disabled") or "").lower()
|
||
aria_disabled = (locator.get_attribute("aria-disabled") or "").lower()
|
||
classes = (locator.get_attribute("class") or "").lower()
|
||
if disabled or aria_disabled == "true" or "disabled" in classes:
|
||
continue
|
||
self.pacer.move_mouse_to(page, locator)
|
||
locator.click()
|
||
try:
|
||
page.wait_for_load_state("networkidle", timeout=15000)
|
||
except Exception:
|
||
pass
|
||
self.pacer.after_page_change()
|
||
return True
|
||
return False
|
||
|
||
def collect_listing_links(self, page: Page, *, make: str | None = None, model: str | None = None) -> dict[str, object]:
|
||
# Последовательно собираем ссылки с учётом лимитов и пагинации.
|
||
self.open_cars_listing(page)
|
||
applied_filters = self.apply_filters(page, make=make, model=model)
|
||
pages: list[dict[str, object]] = []
|
||
all_links: list[str] = []
|
||
for page_number in range(1, max(1, self.settings.listing.max_pages_per_run) + 1):
|
||
page_result = self.collect_current_page(page, page_number=page_number)
|
||
pages.append({"page_number": page_result.page_number, "source_url": page_result.source_url, "links_found": len(page_result.vehicle_links), "vehicle_links": [asdict(item) for item in page_result.vehicle_links], "next_page_detected": page_result.next_page_detected})
|
||
for item in page_result.vehicle_links:
|
||
if item.href not in all_links:
|
||
all_links.append(item.href)
|
||
if len(all_links) >= self.settings.listing.max_vehicles_per_run:
|
||
break
|
||
if len(all_links) >= self.settings.listing.max_vehicles_per_run or self.settings.listing.collect_current_page_only or not self.settings.listing.include_pagination or not page_result.next_page_detected:
|
||
break
|
||
if not self.go_to_next_page(page):
|
||
break
|
||
return {"listing_url": self.settings.listing.cars_url, "applied_filters": applied_filters, "pages_collected": len(pages), "vehicles_collected": len(all_links), "vehicle_urls": all_links, "pages": pages, "strategy": {"sequential": True, "collect_current_page_only": self.settings.listing.collect_current_page_only, "include_pagination": self.settings.listing.include_pagination, "max_pages_per_run": self.settings.listing.max_pages_per_run, "max_vehicles_per_run": self.settings.listing.max_vehicles_per_run}}
|
||
|
||
@staticmethod
|
||
def _try_fill_filter_input(page: Page, selectors: list[str], value: str) -> bool:
|
||
# Пробуем несколько селекторов для одного и того же фильтра.
|
||
for selector in selectors:
|
||
locator = page.locator(selector).first
|
||
if locator.count() == 0:
|
||
continue
|
||
try:
|
||
locator.click(); locator.fill(value); page.keyboard.press("Enter")
|
||
try:
|
||
page.wait_for_load_state("networkidle", timeout=15000)
|
||
except Exception:
|
||
pass
|
||
return True
|
||
except Exception:
|
||
continue
|
||
return False
|
||
|
||
@staticmethod
|
||
def _has_next_page(page: Page) -> bool:
|
||
# Грубая проверка наличия кнопки next.
|
||
for selector in ["a[aria-label*='Next']", "button[aria-label*='Next']", "a.pagination-next", "button.pagination-next", "a:has-text('Next')", "button:has-text('Next')"]:
|
||
if page.locator(selector).count() > 0:
|
||
return True
|
||
return False
|