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 ThreadPoolExecutor, as_completed
from concurrent.futures import TimeoutError as FuturesTimeoutError
from datetime import datetime, timezone
from pathlib import Path
from threading import Event, Thread
from typing import Any, Callable
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, 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)

# Таймауты операций страницы.
_PAGE_COLLECT_TIMEOUT_S = 90
_PAGE_NEXT_TIMEOUT_S = 60
# Ротация контекста выключена по умолчанию.
_CONTEXT_ROTATE_EVERY_PAGES = int(os.environ.get("IAAI_CONTEXT_ROTATE_EVERY_PAGES", "0") or 0)
_SMALL_SEGMENT_SUSPICIOUS_PAGINATION_MAX_LINKS = 120
_SMALL_SEGMENT_SUSPICIOUS_PAGINATION_MAX_PAGE = 2


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 _ProgressHeartbeat:
    """Периодически пульсует progress во время долгой навигации."""

    def __init__(self, report_progress: Callable[[str, Any], None], stage: str, interval_s: float = 15.0, **meta: Any) -> None:
        self._report_progress = report_progress
        self._stage = stage
        self._interval_s = max(5.0, float(interval_s))
        self._meta = meta
        self._stop = Event()
        self._thread: Thread | None = None

    def _run(self) -> None:
        while not self._stop.wait(self._interval_s):
            self._report_progress(self._stage, **self._meta)

    def __enter__(self):
        self._thread = Thread(target=self._run, name=f"progress-heartbeat-{self._stage}", daemon=True)
        self._thread.start()
        return self

    def __exit__(self, exc_type, exc, tb):
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=1.0)
        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
        self._progress_callback: Callable[[str, dict], None] | None = None
        # HTTP pool: при parallel_tabs=24 и concurrency=4 пик ≈ 96 конкурентных сокетов;
        # даём запас до 256, чтобы не упираться в PoolError под всплесками PX/retry.
        proxy_url = self.settings.proxy.server
        if proxy_url:
            _proxy_kwargs: dict = {
                "num_pools": 4,
                "maxsize": 64,
                "block": False,
                "retries": False,
                "timeout": urllib3.Timeout(connect=5, read=20),
            }
            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,
                block=False,
                retries=False,
                timeout=urllib3.Timeout(connect=5, read=20),
            )

    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 _reload_runtime_config(self) -> None:
        """Reload runtime config from file to apply include/exclude changes without restart."""
        try:
            self.runtime_config = RuntimeConfig.from_file(self.settings.runtime_config_file)
        except Exception:
            logger.warning("Failed to reload runtime config, keeping previous values", exc_info=True)

    def set_progress_callback(self, callback: Callable[[str, dict], None]) -> None:
        self._progress_callback = callback

    def _report_progress(self, stage: str, **meta: Any) -> None:
        if self._progress_callback is not None:
            try:
                self._progress_callback(stage, meta)
            except Exception:
                pass

    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)"
            )

        self._report_progress(
            "listing_resume_started",
            target_page=target_page_number,
            max_nav_pages=max_nav_pages,
        )

        page, applied_filters = self._open_listing_page(
            make=make,
            model=model,
            listing_url=listing_url,
            year_min=year_min,
            year_max=year_max,
        )

        self._report_progress(
            "listing_resume_opened",
            target_page=target_page_number,
        )

        for expected_page in range(2, target_page_number + 1):
            self._report_progress(
                "listing_resume_progress",
                current_page=expected_page - 1,
                target_page=target_page_number,
                next_page_number=expected_page,
            )

            with _ProgressHeartbeat(
                self._report_progress,
                "listing_resume_progress",
                current_page=expected_page - 1,
                target_page=target_page_number,
                next_page_number=expected_page,
            ):
                next_ok = self.listing_collector.go_to_next_page(page, expected_page_number=expected_page)

            if not next_ok:
                self._report_progress(
                    "listing_resume_failed",
                    current_page=expected_page - 1,
                    target_page=target_page_number,
                )
                page.close()
                raise ListingResumeError(f"Failed to resume listing at page {target_page_number}")

            self._report_progress(
                "listing_resume_progress",
                current_page=expected_page,
                target_page=target_page_number,
                next_page_number=min(target_page_number, expected_page + 1),
            )

        self._report_progress(
            "listing_resume_completed",
            current_page=target_page_number,
            target_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
        protection_events = 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)
                    protection_events += int(batch_result.get("protection_events", 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)
                    protection_events += int(batch_result.get("protection_events", 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,
                "protection_events": protection_events,
                "failures": failures,
                "all_listing_origin_urls": all_listing_origin_urls,
            }

        def _process_pending_batch(batch_urls: list[str], batch_start: int) -> bool:
            nonlocal cars_upserted, cars_failed, protection_events, images_upserted, failures
            if not batch_urls:
                return True

            logger.info(
                "Processing batch %d-%d of %d",
                batch_start + 1,
                batch_start + len(batch_urls),
                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)
                protection_events += int(batch_result.get("protection_events", 0))
                images_upserted += batch_result.get("images_upserted", 0)
                failures.extend(batch_result.get("failures", []))
                self._report_progress(
                    "batch_upserted",
                    cars_upserted=cars_upserted,
                    cars_failed=cars_failed,
                    total_discovered=total,
                )
                return True
            except PlaywrightError as pw_exc:
                logger.warning(
                    "Batch %d-%d browser crashed: %s. Reinitializing browser context.",
                    batch_start + 1,
                    batch_start + len(batch_urls),
                    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")
                    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 + len(batch_urls),
                    batch_exc,
                )
                cars_failed += len(batch_urls)
                failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(batch_exc)})
                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):
                # Heartbeat для worker stall-watchdog: при малом числе ссылок на странице
                # batch_upserted может долго не вызываться, поэтому пульсуем прогресс
                # на каждом шаге пагинации.
                self._report_progress(
                    "listing_page_scan_started",
                    page_number=page_number,
                    total_discovered=total,
                    pending_urls=len(pending_urls),
                    cars_upserted=cars_upserted,
                    cars_failed=cars_failed,
                )

                # Проактивная ротация контекста (опционально, по умолчанию выключена).
                # Если включена — требует долистывания до 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=300,
                        )
                        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=300,
                        )
                        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=300,
                            )
                            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 "∞",
                )
                self._report_progress(
                    "listing_page_scan_done",
                    page_number=page_number,
                    page_links=len(page_urls),
                    page_new=len(fresh_urls),
                    page_known=page_skipped,
                    total_discovered=total,
                    pending_urls=len(pending_urls),
                    cars_upserted=cars_upserted,
                    cars_failed=cars_failed,
                )

                # Прогресс каждые 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]
                    self._report_progress(
                        "listing_batch_dispatch_started",
                        page_number=page_number,
                        batch_size=len(batch_urls),
                        pending_urls=len(pending_urls),
                        cars_upserted=cars_upserted,
                        cars_failed=cars_failed,
                    )
                    if not _process_pending_batch(batch_urls, cars_upserted + cars_failed):
                        pending_urls = []
                        break
                    pending_urls = pending_urls[batch_size:]
                    self._report_progress(
                        "listing_batch_dispatch_done",
                        page_number=page_number,
                        pending_urls=len(pending_urls),
                        cars_upserted=cars_upserted,
                        cars_failed=cars_failed,
                    )
                    # Пауза между батчами: снижает нагрузку на IAAI и риск бана.
                    if pending_urls:
                        time.sleep(random.uniform(0.5, 1.5))

                # Anti-stall: не держим хвост pending до следующей страницы/конца сегмента.
                # Если после scan остались URL меньше batch_size — отправляем их сразу.
                if pending_urls:
                    self._report_progress(
                        "listing_tail_batch_started",
                        page_number=page_number,
                        batch_size=len(pending_urls),
                        pending_urls=len(pending_urls),
                        cars_upserted=cars_upserted,
                        cars_failed=cars_failed,
                    )
                    if not _process_pending_batch(pending_urls, cars_upserted + cars_failed):
                        pending_urls = []
                        break
                    pending_urls = []
                    self._report_progress(
                        "listing_tail_batch_done",
                        page_number=page_number,
                        pending_urls=0,
                        cars_upserted=cars_upserted,
                        cars_failed=cars_failed,
                    )

                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

                if page_number == 1 and not self.listing_collector.has_page_number(page, 2):
                    logger.info(
                        "Page 2 not found on page 1 — treating segment as single-page",
                    )
                    self._report_progress(
                        "listing_single_page_no_page2",
                        page_number=page_number,
                        cars_upserted=cars_upserted,
                        cars_failed=cars_failed,
                    )
                    break

                suspicious_small_segment = (
                    total <= _SMALL_SEGMENT_SUSPICIOUS_PAGINATION_MAX_LINKS
                    and page_number >= _SMALL_SEGMENT_SUSPICIOUS_PAGINATION_MAX_PAGE
                )
                self._report_progress(
                    "listing_next_page_started",
                    page_number=page_number,
                    next_page_number=page_number + 1,
                    pending_urls=len(pending_urls),
                    cars_upserted=cars_upserted,
                    cars_failed=cars_failed,
                )
                try:
                    with _ProgressHeartbeat(
                        self._report_progress,
                        "listing_next_page_started",
                        page_number=page_number,
                        next_page_number=page_number + 1,
                        pending_urls=len(pending_urls),
                        cars_upserted=cars_upserted,
                        cars_failed=cars_failed,
                    ):
                        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
                self._report_progress(
                    "listing_next_page_done",
                    page_number=page_number,
                    next_page_number=page_number + 1,
                    next_ok=bool(next_ok),
                    pending_urls=len(pending_urls),
                    cars_upserted=cars_upserted,
                    cars_failed=cars_failed,
                )
                if not next_ok:
                    if page_number == 1:
                        logger.warning(
                            "Page 2 is not reachable from page 1 — treating segment as single-page and moving on",
                        )
                        break
                    if suspicious_small_segment:
                        logger.warning(
                            "Small segment pagination looks exhausted at page %d: total=%d, next page navigation failed — stopping segment",
                            page_number,
                            total,
                        )
                        break
                    logger.warning("Failed to navigate to page %d — reopening listing and resuming", page_number + 1)
                    self._report_progress(
                        "listing_reopen_started",
                        page_number=page_number,
                        target_page_number=page_number + 1,
                        pending_urls=len(pending_urls),
                        cars_upserted=cars_upserted,
                        cars_failed=cars_failed,
                    )
                    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=300,
                        )
                        pages_since_rotation = 0
                        self._report_progress(
                            "listing_reopen_done",
                            page_number=page_number,
                            target_page_number=page_number + 1,
                            pending_urls=len(pending_urls),
                            cars_upserted=cars_upserted,
                            cars_failed=cars_failed,
                        )
                    except ListingResumeError as resume_exc:
                        logger.warning(
                            "Cannot resume at page %d (%s) — treating pagination as exhausted",
                            page_number + 1, resume_exc,
                        )
                        self._report_progress(
                            "listing_reopen_failed",
                            page_number=page_number,
                            target_page_number=page_number + 1,
                            error=str(resume_exc),
                            pending_urls=len(pending_urls),
                            cars_upserted=cars_upserted,
                            cars_failed=cars_failed,
                        )
                        break

            if pending_urls:
                _process_pending_batch(pending_urls, cars_upserted + cars_failed)

            logger.warning(
                "sync_listing final progress: pages=%d, discovered=%d, upserted=%d, failed=%d",
                len(pages_info),
                total,
                cars_upserted,
                cars_failed,
            )
        finally:
            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,
            "protection_events": protection_events,
            "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
        вместо посимвольного сканирования скобок.
        """
        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

        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)."""
        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"):
        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

    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
            processed_local = 0
            try:
                for url, global_idx in url_slice:
                    processed_local += 1
                    # Heartbeat в Redis progress: даже если anti-bot тормозит fallback,
                    # watchdog видит живую задачу и не убивает её как stalled.
                    self._report_progress(
                        "browser_fallback_progress",
                        fallback_processed=processed_local,
                        fallback_total=len(url_slice),
                        vehicle_index=global_idx,
                        vehicle_total=total,
                    )

                    # Если anti-bot уже активен (много fallback URL), снижаем burst-нагрузку.
                    if len(url_slice) >= 20:
                        time.sleep(random.uniform(0.2, 0.7))

                    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=max(3_000, int(self.settings.fallback_navigation_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.
            # Таймаут 5 минут на весь fallback — если страницы зависли, не блокируем навсегда.
            _fallback_timeout = max(300, len(fallback_urls) * 30)
            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, timeout=_fallback_timeout):
                    try:
                        res = fut.result(timeout=60)
                    except FuturesTimeoutError:
                        logger.error("Browser fallback thread timed out")
                        cars_failed += 1
                        continue
                    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:
        # Runtime config может меняться на лету (добавили/убрали бренды, диапазоны и т.п.).
        # Подхватываем обновление перед каждым batch, чтобы фильтрация применялась сразу.
        self._reload_runtime_config()

        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
        cars_filtered = 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)
        self._report_progress(
            "sync_batch_started",
            batch_total=total,
            batch_workers=num_workers,
            http_ok=0,
            http_fallback=0,
            batch_records_ready=0,
            batch_failures=0,
        )

        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"
        )

        # ── Фаза 1: HTTP fast-path ──
        fallback_urls: list[tuple[str, int]] = []
        processed_indices: set[int] = set()
        fallback_indices: set[int] = set()

        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))
        # Таймаут на весь HTTP-пул: (connect 5 + read 20) × retries × URLs / workers + запас.
        # Если пул зависнет дольше — отпускаем зависшие потоки и уходим в fallback.
        _per_url_budget = (5 + 20) * (1 + self._HTTP_MAX_RETRIES) + 5
        _http_phase_timeout = max(120, _per_url_budget * max(1, total // max(1, num_workers)))
        http_processed = 0
        with ThreadPoolExecutor(max_workers=min(num_workers, total)) as executor:
            try:
                for idx, result in executor.map(
                    _fetch_one_with_retry, indexed_urls, timeout=_http_phase_timeout,
                ):
                    processed_indices.add(idx)
                    http_processed += 1
                    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))
                        fallback_indices.add(idx)
                    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 "?",
                        )
                    self._report_progress(
                        "sync_batch_http_progress",
                        batch_total=total,
                        http_processed=http_processed,
                        http_ok=http_successes,
                        http_fallback=http_fallbacks,
                        batch_records_ready=len(records),
                        batch_failures=cars_failed + len(failures),
                    )
            except FuturesTimeoutError:
                logger.error(
                    "HTTP phase timed out after %ds; %d/%d processed, rest go to fallback",
                    _http_phase_timeout, http_successes + http_fallbacks, total,
                )
                # Все ещё не обработанные URL → в fallback
                for i, url in enumerate(vehicle_urls):
                    if i in processed_indices or i in fallback_indices:
                        continue
                    if (url, i + 1) not in fallback_urls:
                        fallback_urls.append((url, i + 1))
                        fallback_indices.add(i)
                        http_fallbacks += 1

        logger.info(
            "HTTP phase done: %d OK, %d fallback (%.1fs)",
            http_successes, http_fallbacks, time.perf_counter() - started_at,
        )
        self._report_progress(
            "sync_batch_http_done",
            batch_total=total,
            http_processed=http_processed,
            http_ok=http_successes,
            http_fallback=http_fallbacks,
            batch_records_ready=len(records),
            batch_failures=cars_failed + len(failures),
        )

        # ── Фаза 2: браузерный fallback ──
        if fallback_urls:
            logger.info(
                "Fallback browser mode: %d/%d URLs, single page",
                len(fallback_urls), total,
            )
            self._report_progress(
                "sync_batch_fallback_started",
                batch_total=total,
                http_ok=http_successes,
                http_fallback=http_fallbacks,
                fallback_total=len(fallback_urls),
                batch_records_ready=len(records),
                batch_failures=cars_failed + len(failures),
            )
            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"])
            self._report_progress(
                "sync_batch_fallback_done",
                batch_total=total,
                http_ok=http_successes,
                http_fallback=http_fallbacks,
                fallback_total=len(fallback_urls),
                batch_records_ready=len(records),
                batch_failures=cars_failed + len(failures),
                protection_events=protection_events,
            )

        # ── Фаза 2.5: пост-фильтр по runtime_config ──
        filters_cfg = self.runtime_config.filters
        if records and not filters_cfg.is_empty():
            pre_filter_count = len(records)
            records = [
                rec for rec in records
                if filters_cfg.matches({
                    "brand": rec.brand,
                    "model": rec.model,
                    "year": rec.year,
                    "body_type": rec.body_type,
                    "color": rec.color,
                    "drive": rec.drive,
                    "gearbox": rec.gearbox,
                    "price": rec.price,
                    "mileage": rec.mileage,
                })
            ]
            cars_filtered = pre_filter_count - len(records)
            if cars_filtered:
                logger.info(
                    "Runtime filter: %d/%d records filtered out before DB upsert",
                    cars_filtered, pre_filter_count,
                )

        # ── Фаза 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:
                self._report_progress(
                    "sync_batch_db_upsert_started",
                    batch_total=total,
                    batch_records_ready=len(records),
                    http_ok=http_successes,
                    http_fallback=http_fallbacks,
                    batch_failures=cars_failed + len(failures),
                )
                _db_start = time.perf_counter()
                batch_result = self.persistence.upsert_cars_batch(records)
                _db_elapsed = time.perf_counter() - _db_start
                cars_upserted = batch_result["inserted"] + batch_result["updated"]
                images_upserted = batch_result["images_upserted"]
                if _db_elapsed > 10:
                    logger.warning("DB upsert slow: %.1fs for %d records", _db_elapsed, len(records))
                self._report_progress(
                    "sync_batch_db_upsert_done",
                    batch_total=total,
                    batch_records_ready=len(records),
                    cars_upserted=cars_upserted,
                    images_upserted=images_upserted,
                    http_ok=http_successes,
                    http_fallback=http_fallbacks,
                    batch_failures=cars_failed + len(failures),
                )
            except Exception as exc:
                logger.error("Batch upsert failed: %s", exc)
                cars_failed += len(records)
                self._report_progress(
                    "sync_batch_db_upsert_failed",
                    batch_total=total,
                    batch_records_ready=len(records),
                    http_ok=http_successes,
                    http_fallback=http_fallbacks,
                    batch_failures=cars_failed + len(failures),
                    error=str(exc),
                )

        status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
        self._report_progress(
            "sync_batch_done",
            batch_total=total,
            cars_upserted=cars_upserted,
            cars_failed=cars_failed,
            cars_filtered=cars_filtered,
            images_upserted=images_upserted,
            http_ok=http_successes,
            http_fallback=http_fallbacks,
            protection_events=protection_events,
            batch_failures=len(failures),
        )
        return {
            "trace_id": trace_id,
            "status": status,
            "cars_upserted": cars_upserted,
            "cars_failed": cars_failed,
            "cars_filtered": cars_filtered,
            "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,
        listing_url: str | None = None,
        year_min: int | None = None,
        year_max: int | None = None,
        skip_mark_sold: bool = False,
    ):
        # Подхватываем актуальный runtime_config на каждый запуск sync,
        # чтобы добавленные/удалённые бренды/модели применялись без рестарта.
        self._reload_runtime_config()

        # 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
        protection_events = 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"]
            protection_events = int(stream_result.get("protection_events", 0))
            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 skip_mark_sold:
                logger.debug("Skipping mark_sold: caller requested")
            elif 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,
        )
        discovered_total = max(0, int(total))
        fail_ratio = (cars_failed / discovered_total) if discovered_total > 0 else 0.0
        protection_ratio = (protection_events / discovered_total) if discovered_total > 0 else 0.0
        anti_bot_detected = discovered_total > 0 and (
            (protection_events >= 30 and protection_ratio >= 0.10)
            or fail_ratio >= 0.30
        )
        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") and not anti_bot_detected,
            "only_new_effective": effective_only_new,
            "listing": listing,
            "total_discovered": discovered_total,
            "cars_upserted": cars_upserted,
            "cars_failed": cars_failed,
            "cars_filtered": cars_filtered,
            "images_upserted": images_upserted,
            "protection_events": protection_events,
            "anti_bot_detected": anti_bot_detected,
            "fail_ratio": round(fail_ratio, 4),
            "protection_ratio": round(protection_ratio, 4),
            "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
        skipped_segments = 0

        # В segmented-режиме полный прогон должен проходить ВСЕ сегменты.
        # Runtime-фильтры применяются позже (на уровне конкретных карточек),
        # но не должны сужать сам обход сегментов.

        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")

            # На длительном full-scan сегмент нельзя пропускать из-за runtime include.brands,
            # иначе прогон становится частичным.
            self._reload_runtime_config()

            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,
            )

            outer_progress_callback = self._progress_callback

            def _segment_progress(stage: str, meta: dict[str, Any]) -> None:
                if outer_progress_callback is None:
                    return
                outer_progress_callback(
                    stage,
                    {
                        **meta,
                        "segment_index": seg_idx,
                        "segment_label": seg_label,
                        "segments_total": len(segments),
                    },
                )

            try:
                self.set_progress_callback(_segment_progress)
                self._report_progress(
                    "segment_started",
                    segment_index=seg_idx,
                    segment_label=seg_label,
                    segments_total=len(segments),
                )
                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))
                # Даже если сегмент завершился неполно (например, страница/пагинация сломалась),
                # в bootstrap-режиме не зацикливаемся на нём: двигаем чекпоинт дальше.
                if not segment_done:
                    skipped_segments += 1
                    logger.warning(
                        "Segment %d/%d incomplete: %s — advancing checkpoint and continuing",
                        seg_idx + 1, len(segments), seg_label,
                    )

                # Сегмент обработан до конечного состояния — фиксируем чекпоинт,
                # даже если он был пропущен/оборван с ошибками.
                if 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),
                )
                self._report_progress(
                    "segment_done",
                    segment_index=seg_idx,
                    segment_label=seg_label,
                    segments_total=len(segments),
                    segment_status=result.get("status"),
                    segment_full_scan_completed=segment_done,
                )
            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)})
                skipped_segments += 1
                segment_results.append({
                    "segment": seg,
                    "segment_index": seg_idx,
                    "status": "skipped_error",
                    "full_scan_completed": False,
                    "cars_upserted": 0,
                    "cars_failed": 0,
                    "vehicles_collected": 0,
                })
                if progress_callback is not None:
                    try:
                        progress_callback(seg_idx)
                    except Exception:
                        logger.warning(
                            "Failed to persist skipped segment checkpoint for segment=%d",
                            seg_idx, exc_info=True,
                        )
                self._report_progress(
                    "segment_failed",
                    segment_index=seg_idx,
                    segment_label=seg_label,
                    segments_total=len(segments),
                    error=str(exc),
                )
                # Продолжаем оставшиеся сегменты — одна ошибка не должна убивать весь прогон
                continue
            finally:
                self.set_progress_callback(outer_progress_callback)

        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),
            "segments_skipped": skipped_segments,
            "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)