2 Commits

Author SHA1 Message Date
qananasikq
31f87a9bdf update parser 2026-04-24 18:36:31 +03:00
qananasikq
1453eee83b update sync 2026-04-24 18:00:09 +03:00
7 changed files with 227 additions and 46 deletions

View File

@@ -6,6 +6,8 @@ IAAI_MAX_CAPTURED_REQUESTS=40
IAAI_MAX_CAPTURED_JSON_RESPONSES=20
IAAI_CARS_LISTING_URL=https://www.iaai.com/Vehiclelisting/Cars
IAAI_FILTERED_SEARCH_URL=
IAAI_FILTERED_SEARCH_URLS=
IAAI_LISTING_SEGMENTS=runtime
IAAI_MAX_PAGES_PER_RUN=999999
IAAI_MAX_VEHICLES_PER_RUN=999999

View File

@@ -31,6 +31,8 @@ x-app-env: &app-env
IAAI_PARALLEL_TABS: ${IAAI_PARALLEL_TABS:-8}
IAAI_FETCH_CONCURRENCY: ${IAAI_FETCH_CONCURRENCY:-32}
IAAI_BLOCK_RESOURCES: ${IAAI_BLOCK_RESOURCES:-true}
IAAI_FILTERED_SEARCH_URL: ${IAAI_FILTERED_SEARCH_URL:-}
IAAI_FILTERED_SEARCH_URLS: ${IAAI_FILTERED_SEARCH_URLS:-}
IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-runtime}
IAAI_MAX_PAGES_PER_RUN: ${IAAI_MAX_PAGES_PER_RUN:-9999}
IAAI_MAX_VEHICLES_PER_RUN: ${IAAI_MAX_VEHICLES_PER_RUN:-50000}

View File

@@ -116,13 +116,33 @@ class HybridSessionAuth:
while attempt <= retries:
try:
request_started_at = time.perf_counter()
if self._settings.scraping_profile.verbose_http_logs:
logger.debug(
"HTTP request started method=%s url=%s attempt=%s/%s timeout=%s marker=%s",
method,
url,
attempt + 1,
retries + 1,
timeout,
expected_marker,
)
response = session.request(
method=method,
url=url,
headers=headers,
data=data,
json=json_body,
timeout=timeout,
timeout=(min(10, max(1, timeout)), max(1, timeout)),
)
if self._settings.scraping_profile.verbose_http_logs or response.status_code >= 400:
logger.debug(
"HTTP request completed method=%s url=%s status=%s elapsed=%.1fs marker=%s",
method,
url,
response.status_code,
time.perf_counter() - request_started_at,
expected_marker,
)
except requests.RequestException as exc:
last_error = exc
@@ -159,6 +179,7 @@ class HybridSessionAuth:
max_refresh_attempts,
)
self._refresh_session_via_playwright(expected_marker=LISTING_MARKER, session=session)
logger.info("Retrying HTTP request after Playwright refresh url=%s", url)
if refresh_attempts > 1:
self._sleep_backoff(retry_backoff_ms, refresh_attempts - 1)
continue
@@ -291,7 +312,9 @@ class HybridSessionAuth:
logger.info("IAAI session challenge detected. Refreshing session via Playwright.")
cookies = self._fetch_cookies_via_playwright(expected_marker=expected_marker)
logger.info("Playwright refresh returned %d cookies", len(cookies))
self._apply_cookies_to_session(target_session, cookies)
logger.info("Playwright cookies applied to requests session")
with self._lock:
self._refresh_generation += 1
@@ -299,7 +322,7 @@ class HybridSessionAuth:
self._anonymous_bootstrap_attempted = True
refreshed_generation = self._refresh_generation
self._thread_local.session_generation = refreshed_generation
self._save_storage_state(target_session)
logger.info("Playwright refresh completed generation=%s", refreshed_generation)
def _fetch_cookies_via_playwright(self, *, expected_marker: str | None = None) -> list[dict[str, Any]]:
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
@@ -316,8 +339,10 @@ class HybridSessionAuth:
)
page = context.new_page()
home_target = self._settings.home_url
target = urljoin(self._settings.home_url, "Vehiclelisting/Cars")
filtered_urls = self._settings.listing.filtered_search_urls
target = filtered_urls[0] if filtered_urls else urljoin(self._settings.home_url, "Vehiclelisting/Cars")
timeout_ms = max(30_000, self._settings.default_timeout_ms)
logger.info("Playwright session refresh opening target=%s marker=%s", target, expected_marker or LISTING_MARKER)
page.goto(home_target, wait_until="domcontentloaded", timeout=timeout_ms)
self._accept_cookie_banner(page)
page.goto(target, wait_until="domcontentloaded", timeout=timeout_ms)
@@ -328,13 +353,16 @@ class HybridSessionAuth:
timeout_ms=timeout_ms,
expected_marker=expected_marker or LISTING_MARKER,
)
state = context.storage_state()
cookies = context.cookies()
logger.info("Playwright context returned %d cookies", len(cookies))
except PlaywrightTimeoutError as exc:
raise RuntimeError(f"Playwright refresh timed out: {exc}") from exc
finally:
try:
browser.close()
except Exception as exc:
logger.info("Playwright browser close failed after cookie refresh: %s", exc)
cookies = state.get("cookies") if isinstance(state, dict) else None
if not isinstance(cookies, list) or not cookies:
raise RuntimeError("Playwright refresh did not return cookies")
return [cookie for cookie in cookies if isinstance(cookie, dict)]
@@ -356,7 +384,7 @@ class HybridSessionAuth:
def _wait_until_non_challenge(*, page: Any, target: str, timeout_ms: int, expected_marker: str | None) -> None:
poll_ms = max(1000, min(5000, timeout_ms // PLAYWRIGHT_REFRESH_POLLS))
navigation_error_count = 0
for _ in range(PLAYWRIGHT_REFRESH_POLLS):
for poll_index in range(PLAYWRIGHT_REFRESH_POLLS):
try:
page.wait_for_load_state("domcontentloaded", timeout=poll_ms)
except Exception:
@@ -378,8 +406,22 @@ class HybridSessionAuth:
body_text=body,
expected_marker=expected_marker,
):
logger.info(
"Playwright session refresh passed challenge target=%s poll=%s/%s marker=%s",
target,
poll_index + 1,
PLAYWRIGHT_REFRESH_POLLS,
expected_marker,
)
return
try:
logger.info(
"Playwright session refresh still waiting target=%s poll=%s/%s marker=%s",
target,
poll_index + 1,
PLAYWRIGHT_REFRESH_POLLS,
expected_marker,
)
page.goto(target, wait_until="domcontentloaded", timeout=timeout_ms)
except Exception:
pass
@@ -457,6 +499,14 @@ class IAAIFastClient:
)
for scope_path in scope_paths:
first_page_html = self._fetch_listing_first_page(scope_path)
search_scope_path = build_search_scope_path_from_html(first_page_html)
if search_scope_path and search_scope_path != scope_path:
logger.info(
"Resolved listing scope to fast Search URL: scope=%s search_scope=%s",
scope_path,
search_scope_path,
)
scope_path = search_scope_path
first_page = parse_listing_page(first_page_html)
for vehicle in first_page.vehicles:
if vehicle.inventory_id in seen_inventory_ids:
@@ -599,10 +649,30 @@ def build_brand_scope_paths(brands: set[str]) -> list[str]:
def resolve_listing_scope_paths(*, listing_start_url: str, brands: set[str]) -> list[str]:
explicit_scope = listing_start_url.strip()
if explicit_scope:
if explicit_scope.lower().startswith(("http://", "https://", "/")):
return [explicit_scope]
return [f"/Search?url={explicit_scope}"]
return build_brand_scope_paths(brands)
def build_search_scope_path_from_html(html_text: str) -> str | None:
tiny_url = parse_attribute_value(html_text, "data-tinyurl")
if tiny_url:
return f"/Search?url={tiny_url}"
data_query_raw = parse_attribute_value(html_text, "data-query")
if data_query_raw:
try:
data_query = json.loads(data_query_raw)
except (json.JSONDecodeError, ValueError, TypeError):
data_query = None
if isinstance(data_query, dict):
url_value = parse_text(data_query.get("Url"))
if url_value:
return f"/Search?url={url_value}"
return None
def parse_listing_page(html_text: str) -> FastListingPage:
gbp_raw = parse_hidden_input_value(html_text, "GBPSearchQuery")
vehicle_raw = parse_hidden_input_value(html_text, "VehicleDetails")
@@ -677,6 +747,20 @@ def parse_hidden_input_value(html_text: str, input_id: str) -> str | None:
return None
def parse_attribute_value(html_text: str, attribute_name: str) -> str | None:
escaped_name = re.escape(attribute_name)
patterns = (
rf"\b{escaped_name}=\"([^\"]*)\"",
rf"\b{escaped_name}='([^']*)'",
)
for pattern in patterns:
match = re.search(pattern, html_text, flags=re.IGNORECASE)
if match is not None:
value = html.unescape(match.group(1)).strip()
return value or None
return None
def build_resizer_images_from_keys(image_keys: list[dict[str, Any]]) -> list[dict[str, str | int]]:
seen_fullres: set[str] = set()
images: list[dict[str, str | int]] = []

View File

@@ -105,6 +105,8 @@ class HumanPaceConfig:
@dataclass(slots=True)
class ListingConfig:
cars_url: str = _env_str("IAAI_CARS_LISTING_URL", "https://www.iaai.com/Vehiclelisting/Cars")
filtered_search_url: str | None = _env_optional_str("IAAI_FILTERED_SEARCH_URL")
filtered_search_urls_raw: str | None = _env_optional_str("IAAI_FILTERED_SEARCH_URLS")
max_pages_per_run: int = _env_int("IAAI_MAX_PAGES_PER_RUN", 9999)
max_vehicles_per_run: int = _env_int("IAAI_MAX_VEHICLES_PER_RUN", 50000)
page_link_limit: int = _env_int("IAAI_PAGE_LINK_LIMIT", 500)
@@ -120,6 +122,26 @@ class ListingConfig:
listing_segments_json: str = _env_str("IAAI_LISTING_SEGMENTS", "")
fast_segment_year_splits: bool = _env_bool("IAAI_FAST_SEGMENT_YEAR_SPLITS", True)
@property
def filtered_search_urls(self) -> list[str]:
urls: list[str] = []
if self.filtered_search_url and self.filtered_search_url.strip():
urls.append(self.filtered_search_url.strip())
if self.filtered_search_urls_raw and self.filtered_search_urls_raw.strip():
raw = self.filtered_search_urls_raw.strip()
try:
parsed = json.loads(raw)
except (json.JSONDecodeError, ValueError):
parsed = None
if isinstance(parsed, list):
candidates = [str(item or "").strip() for item in parsed]
else:
candidates = [part.strip() for part in raw.replace("\n", ",").split(",")]
for candidate in candidates:
if candidate and candidate not in urls:
urls.append(candidate)
return urls
# Список брендов IAAI для автоматической сегментации.
# Покрывает >99% автомобилей на сайте. Порядок: от крупных к мелким.
@@ -280,6 +302,8 @@ class ScrapingProfileConfig:
http_first: bool = _env_bool("IAAI_HTTP_FIRST", True)
browser_fallback_enabled: bool = _env_bool("IAAI_BROWSER_FALLBACK_ENABLED", True)
anonymous_bootstrap_enabled: bool = _env_bool("IAAI_ANONYMOUS_BOOTSTRAP_ENABLED", True)
verbose_http_logs: bool = _env_bool("IAAI_VERBOSE_HTTP_LOGS", False)
verbose_progress_logs: bool = _env_bool("IAAI_VERBOSE_PROGRESS_LOGS", False)
challenge_refresh_enabled: bool = _env_bool("IAAI_CHALLENGE_REFRESH_ENABLED", True)
challenge_refresh_attempts: int = _env_int("IAAI_CHALLENGE_REFRESH_ATTEMPTS", 3)
listing_post_attempts: int = _env_int("IAAI_LISTING_POST_ATTEMPTS", 4)

View File

@@ -65,10 +65,16 @@ class FastSyncEngine:
return any(
marker in text
for marker in (
"sslerror",
"ssleoferror",
"unexpected_eof_while_reading",
"eof occurred in violation of protocol",
"max retries exceeded",
"read timed out",
"readtimeout",
"connection reset",
"connection aborted",
"connection closed",
"temporarily unavailable",
"too many requests",
"status=429",
@@ -175,6 +181,7 @@ class FastSyncEngine:
)
prepared_rows: list[CarRecord] = []
db_processed = 0
retry_candidates: list[FastListingVehicle] = []
started_details = time.perf_counter()
with concurrent.futures.ThreadPoolExecutor(max_workers=min(self.fetch_concurrency, len(candidates))) as executor:
@@ -209,6 +216,15 @@ class FastSyncEngine:
continue
prepared_rows.append(record)
stats.ids_fetched += 1
if len(prepared_rows) >= self.batch_size:
db_processed += self._flush_db_records(
rows=prepared_rows,
stats=stats,
errors=errors,
processed=db_processed + len(prepared_rows),
total=len(candidates),
)
prepared_rows.clear()
except Exception as exc:
stats.cars_failed += 1
errors.append({"vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}", "error": str(exc)})
@@ -222,6 +238,7 @@ class FastSyncEngine:
if index % 100 == 0 or index == len(candidates):
elapsed = max(0.001, time.perf_counter() - started_details)
if self.client._settings.scraping_profile.verbose_progress_logs:
logger.info(
"Fast HTTP-first details progress: processed=%d/%d ok=%d failed=%d queued_db=%d rate=%.2f/s",
index,
@@ -283,6 +300,15 @@ class FastSyncEngine:
prepared_rows.append(record)
stats.ids_fetched += 1
stats.cars_failed = max(0, stats.cars_failed - 1)
if len(prepared_rows) >= self.batch_size:
db_processed += self._flush_db_records(
rows=prepared_rows,
stats=stats,
errors=errors,
processed=db_processed + len(prepared_rows),
total=len(candidates),
)
prepared_rows.clear()
except Exception as exc:
retry_errors.append({
"vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}",
@@ -305,32 +331,14 @@ class FastSyncEngine:
errors.extend(retry_errors)
if prepared_rows:
for batch_start in range(0, len(prepared_rows), self.batch_size):
batch = prepared_rows[batch_start:batch_start + self.batch_size]
try:
upsert = self.persistence.upsert_cars_batch(batch)
stats.cars_upserted += int(upsert.get("inserted", 0)) + int(upsert.get("updated", 0))
stats.images_upserted += int(upsert.get("images_upserted", 0))
except Exception as exc:
stats.cars_failed += len(batch)
errors.append({"vehicle_url": f"db_batch_{batch_start}", "error": str(exc)})
logger.exception("Fast DB apply failed batch_start=%s size=%s: %s", batch_start, len(batch), exc)
self._progress(
"fast_db_progress",
processed=min(batch_start + len(batch), len(prepared_rows)),
total=len(prepared_rows),
cars_upserted=stats.cars_upserted,
cars_failed=stats.cars_failed,
images_upserted=stats.images_upserted,
)
logger.info(
"Fast HTTP-first DB progress: processed=%d/%d upserted=%d failed=%d images=%d",
min(batch_start + len(batch), len(prepared_rows)),
len(prepared_rows),
stats.cars_upserted,
stats.cars_failed,
stats.images_upserted,
db_processed += self._flush_db_records(
rows=prepared_rows,
stats=stats,
errors=errors,
processed=db_processed + len(prepared_rows),
total=len(candidates),
)
prepared_rows.clear()
self.client.persist_session_state()
@@ -419,6 +427,45 @@ class FastSyncEngine:
time.sleep(random.uniform(0.0, jitter))
return self.client.fetch_vehicle_detail_payload(inventory_id)
def _flush_db_records(
self,
*,
rows: list[CarRecord],
stats: FastSyncStats,
errors: list[dict[str, str]],
processed: int,
total: int,
) -> int:
if not rows:
return 0
batch = list(rows)
try:
upsert = self.persistence.upsert_cars_batch(batch)
stats.cars_upserted += int(upsert.get("inserted", 0)) + int(upsert.get("updated", 0))
stats.images_upserted += int(upsert.get("images_upserted", 0))
except Exception as exc:
stats.cars_failed += len(batch)
errors.append({"vehicle_url": f"db_batch_{processed - len(batch)}", "error": str(exc)})
logger.exception("Fast DB apply failed processed=%s size=%s: %s", processed, len(batch), exc)
self._progress(
"fast_db_progress",
processed=processed,
total=total,
cars_upserted=stats.cars_upserted,
cars_failed=stats.cars_failed,
images_upserted=stats.images_upserted,
)
logger.info(
"Fast HTTP-first DB batch: processed=%d/%d batch=%d upserted=%d failed=%d images=%d",
processed,
total,
len(batch),
stats.cars_upserted,
stats.cars_failed,
stats.images_upserted,
)
return len(batch)
def _looks_like_protection(exc: Exception) -> bool:
message = str(exc).lower()

View File

@@ -2424,19 +2424,25 @@ class IAAIScraper:
seg_make = seg.get("make")
seg_year_min = seg.get("year_min")
seg_year_max = seg.get("year_max")
seg_listing_url = seg.get("listing_url")
# На длительном full-scan сегмент нельзя пропускать из-за runtime include.brands,
# иначе прогон становится частичным.
self._reload_runtime_config()
if seg_listing_url:
seg_url = str(seg_listing_url)
else:
seg_url = None if self.settings.scraping_profile.http_first else (self._build_segment_listing_url(base_url, seg_make) if seg_make else None)
seg_label = f"{seg_make or 'ALL'}"
if seg_listing_url:
seg_label = "FILTERED_SEARCH"
if seg_year_min is not None or seg_year_max is not None:
seg_label += f" ({seg_year_min}-{seg_year_max})"
logger.info(
"Segment %d/%d: %s",
logger.debug(
"Segment %d/%d started: %s",
seg_idx + 1, len(segments), seg_label,
)

View File

@@ -1032,6 +1032,15 @@ def _build_listing_segments(settings: Settings) -> list[dict[str, str | int | No
При IAAI_LISTING_SEGMENTS=runtime сегменты строятся из filters.brands / filters.include.brands.
Остальные значения IAAI_LISTING_SEGMENTS сохраняют прежнее поведение: auto или JSON.
"""
filtered_urls = settings.listing.filtered_search_urls
if len(filtered_urls) > 1:
return [
{"make": None, "year_min": None, "year_max": None, "listing_url": url}
for url in filtered_urls
]
if len(filtered_urls) == 1:
return []
raw_segments = settings.listing.listing_segments_json.strip()
if raw_segments.casefold() != "runtime":
return parse_listing_segments(raw_segments)
@@ -1088,7 +1097,10 @@ def sync_segment_task(
seg_make = segment.get("make")
seg_year_min = segment.get("year_min")
seg_year_max = segment.get("year_max")
seg_listing_url = segment.get("listing_url")
seg_label = f"{seg_make or 'ALL'}"
if seg_listing_url:
seg_label = "FILTERED_SEARCH"
if seg_year_min is not None or seg_year_max is not None:
seg_label += f" ({seg_year_min}-{seg_year_max})"
@@ -1135,6 +1147,7 @@ def sync_segment_task(
model=None,
lane=lane,
only_new=only_new,
listing_url=str(seg_listing_url) if seg_listing_url else None,
year_min=seg_year_min,
year_max=seg_year_max,
skip_mark_sold=True,
@@ -1422,6 +1435,8 @@ def sync_listing_task(
and not always_full_scan
)
# Определяем сегменты из конфига/env или из runtime_config при IAAI_LISTING_SEGMENTS=runtime.
filtered_listing_urls = settings.listing.filtered_search_urls
filtered_listing_url = filtered_listing_urls[0] if filtered_listing_urls else None
segments = _build_listing_segments(settings)
watchdog_stop, watchdog_thread = _start_stall_watchdog(
@@ -1716,6 +1731,7 @@ def sync_listing_task(
lane=lane,
limit=effective_limit,
only_new=effective_only_new,
listing_url=filtered_listing_url,
)
result = _run_browser_job(_job)