IAAI scraper

iaai_scraper/core/__init__.py

@@ -0,0 +1,4 @@
+from .config import *  # noqa: F401,F403
+from .logs import *  # noqa: F401,F403
+from .retry import *  # noqa: F401,F403
+from .utils import *  # noqa: F401,F403

iaai_scraper/core/config.py

@@ -0,0 +1,95 @@
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+@dataclass(slots=True)
+class FingerprintConfig:
+    user_agent: str = (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/135.0.0.0 Safari/537.36"
+    )
+    viewport_presets: tuple[dict[str, int], ...] = (
+        {"width": 1920, "height": 1080},
+        {"width": 1600, "height": 900},
+        {"width": 1536, "height": 864},
+        {"width": 1440, "height": 900},
+        {"width": 1366, "height": 768},
+    )
+    timezone_candidates: tuple[str, ...] = (
+        "America/New_York",
+        "America/Chicago",
+        "America/Los_Angeles",
+    )
+    locale: str = "en-US"
+    sec_ch_ua: str = '"Google Chrome";v="135", "Chromium";v="135", "Not.A/Brand";v="24"'
+
+
+@dataclass(slots=True)
+class GentleModeConfig:
+    enabled: bool = os.getenv("IAAI_GENTLE_MODE", "true").strip().lower() in {"1", "true", "yes", "on"}
+    capture_same_origin_only: bool = os.getenv("IAAI_CAPTURE_SAME_ORIGIN_ONLY", "true").strip().lower() in {"1", "true", "yes", "on"}
+    max_requests: int = int(os.getenv("IAAI_MAX_CAPTURED_REQUESTS", "40"))
+    max_json_responses: int = int(os.getenv("IAAI_MAX_CAPTURED_JSON_RESPONSES", "30"))
+    warm_scroll_rounds: int = int(os.getenv("IAAI_WARM_SCROLL_ROUNDS", "1"))
+    scroll_pause_ms: int = int(os.getenv("IAAI_SCROLL_PAUSE_MS", "900"))
+    post_open_idle_ms: int = int(os.getenv("IAAI_POST_OPEN_IDLE_MS", "3500"))
+
+
+@dataclass(slots=True)
+class HumanPaceConfig:
+    enabled: bool = os.getenv("IAAI_HUMAN_PACE_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"}
+    after_listing_open_min_s: float = float(os.getenv("IAAI_AFTER_LISTING_OPEN_MIN_S", "2.5"))
+    after_listing_open_max_s: float = float(os.getenv("IAAI_AFTER_LISTING_OPEN_MAX_S", "4.5"))
+    after_filter_action_min_s: float = float(os.getenv("IAAI_AFTER_FILTER_ACTION_MIN_S", "2.0"))
+    after_filter_action_max_s: float = float(os.getenv("IAAI_AFTER_FILTER_ACTION_MAX_S", "4.0"))
+    before_vehicle_open_min_s: float = float(os.getenv("IAAI_BEFORE_VEHICLE_OPEN_MIN_S", "0.5"))
+    before_vehicle_open_max_s: float = float(os.getenv("IAAI_BEFORE_VEHICLE_OPEN_MAX_S", "1.5"))
+    after_vehicle_open_min_s: float = float(os.getenv("IAAI_AFTER_VEHICLE_OPEN_MIN_S", "0.3"))
+    after_vehicle_open_max_s: float = float(os.getenv("IAAI_AFTER_VEHICLE_OPEN_MAX_S", "1.0"))
+    between_vehicles_min_s: float = float(os.getenv("IAAI_BETWEEN_VEHICLES_MIN_S", "0.5"))
+    between_vehicles_max_s: float = float(os.getenv("IAAI_BETWEEN_VEHICLES_MAX_S", "1.5"))
+    after_page_change_min_s: float = float(os.getenv("IAAI_AFTER_PAGE_CHANGE_MIN_S", "3.0"))
+    after_page_change_max_s: float = float(os.getenv("IAAI_AFTER_PAGE_CHANGE_MAX_S", "6.0"))
+
+
+@dataclass(slots=True)
+class ListingConfig:
+    cars_url: str = os.getenv("IAAI_CARS_LISTING_URL", "https://www.iaai.com/Vehiclelisting/Cars")
+    max_pages_per_run: int = int(os.getenv("IAAI_MAX_PAGES_PER_RUN", "5"))
+    max_vehicles_per_run: int = int(os.getenv("IAAI_MAX_VEHICLES_PER_RUN", "100"))
+    page_link_limit: int = int(os.getenv("IAAI_PAGE_LINK_LIMIT", "200"))
+    include_pagination: bool = os.getenv("IAAI_INCLUDE_PAGINATION", "true").strip().lower() in {"1", "true", "yes", "on"}
+    collect_current_page_only: bool = os.getenv("IAAI_COLLECT_CURRENT_PAGE_ONLY", "false").strip().lower() in {"1", "true", "yes", "on"}
+
+
+@dataclass(slots=True)
+class DatabaseConfig:
+    url: str = os.getenv("IAAI_DATABASE_URL", "sqlite:///iaai_scraper.db")
+    echo: bool = os.getenv("IAAI_DATABASE_ECHO", "false").strip().lower() in {"1", "true", "yes", "on"}
+
+
+@dataclass(slots=True)
+class Settings:
+    home_url: str = "https://www.iaai.com/"
+    default_timeout_ms: int = int(os.getenv("IAAI_TIMEOUT_MS", "45000"))
+    network_settle_ms: int = int(os.getenv("IAAI_NETWORK_SETTLE_MS", "800"))
+    max_retries: int = int(os.getenv("IAAI_MAX_RETRIES", "3"))
+    headless: bool = os.getenv("IAAI_HEADLESS", "true").strip().lower() in {"1", "true", "yes", "on"}
+    raw_output_json: str | None = os.getenv("IAAI_RAW_OUTPUT_JSON") or None
+    log_level: str = os.getenv("IAAI_LOG_LEVEL", "INFO")
+    log_file: str | None = os.getenv("IAAI_LOG_FILE") or None
+    scheduler_interval_minutes: int = int(os.getenv("IAAI_SCHEDULER_INTERVAL_MINUTES", "60"))
+    fingerprint: FingerprintConfig = field(default_factory=FingerprintConfig)
+    gentle: GentleModeConfig = field(default_factory=GentleModeConfig)
+    pace: HumanPaceConfig = field(default_factory=HumanPaceConfig)
+    listing: ListingConfig = field(default_factory=ListingConfig)
+    database: DatabaseConfig = field(default_factory=DatabaseConfig)
+
+
+settings = Settings()

iaai_scraper/core/logs.py

@@ -0,0 +1,14 @@
+import logging
+import sys
+
+
+def setup_logging(level: str = "INFO", log_file: str | None = None) -> None:
+    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)]
+    if log_file:
+        handlers.append(logging.FileHandler(log_file, encoding="utf-8"))
+    logging.basicConfig(
+        level=getattr(logging, level.upper(), logging.INFO),
+        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+        handlers=handlers,
+        force=True,
+    )

iaai_scraper/core/retry.py

@@ -0,0 +1,31 @@
+import logging
+import time
+from collections.abc import Callable
+from functools import wraps
+from typing import Any
+
+from playwright.sync_api import Error, TimeoutError as PlaywrightTimeoutError
+
+logger = logging.getLogger("iaai_scraper.retry")
+
+
+def retryable(max_attempts: int, delay_seconds: float = 2.5) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        @wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            last_error: Exception | None = None
+            for attempt in range(1, max_attempts + 1):
+                try:
+                    return func(*args, **kwargs)
+                except (PlaywrightTimeoutError, Error, ConnectionError, OSError, RuntimeError) as exc:
+                    last_error = exc
+                    logger.warning("%s failed on attempt %s/%s: %s", func.__name__, attempt, max_attempts, exc)
+                    if attempt < max_attempts:
+                        time.sleep(delay_seconds * (2 ** (attempt - 1)))
+            if last_error is not None:
+                raise last_error
+            raise RuntimeError("Retry wrapper failed without a captured exception")
+
+        return wrapper
+
+    return decorator

iaai_scraper/core/utils.py

@@ -0,0 +1,50 @@
+import json
+import random
+import re
+import time
+from pathlib import Path
+from typing import Any, Iterable
+
+
+def save_to_json(data: Any, filename: str | Path) -> None:
+    Path(filename).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def short_sleep(a: float = 0.10, b: float = 0.35) -> None:
+    time.sleep(random.uniform(a, b))
+
+
+def mask_email(email: str) -> str:
+    if "@" not in email:
+        return "***"
+    local, domain = email.split("@", 1)
+    safe_local = local[:2] + "***" if len(local) > 2 else local[:1] + "*"
+    return f"{safe_local}@{domain}"
+
+
+def first_non_empty(values: Iterable[Any]) -> Any | None:
+    for value in values:
+        if value not in (None, "", [], {}, ()):
+            return value
+    return None
+
+
+# VIN, lot, price regex
+VIN_RE = re.compile(r"\b([A-HJ-NPR-Z0-9]{17})\b", re.IGNORECASE)
+LOT_RE = re.compile(r"\b(\d{7,10})\b")
+PRICE_RE = re.compile(r"\$\s?([\d,]+(?:\.\d{1,2})?)")
+
+
+def deep_find_key(obj, target_keys: set[str], max_depth: int = 64, _depth: int = 0) -> list:
+    found = []
+    if _depth >= max_depth:
+        return found
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if key.lower() in target_keys:
+                found.append(value)
+            found.extend(deep_find_key(value, target_keys, max_depth=max_depth, _depth=_depth + 1))
+    elif isinstance(obj, list):
+        for item in obj:
+            found.extend(deep_find_key(item, target_keys, max_depth=max_depth, _depth=_depth + 1))
+    return found