Add tools

.env.example

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

.gitignore

@@ -19,4 +19,6 @@ build/
 artifacts/
 celerybeat-schedule*
 tokens_data/
-tokens.json
+tokens.json
+iaai_deploy.tar.gz
+.tmp_iaai_fast_ref/

check_vps.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import os
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+
+cmds = [
+    "cd /root/iaai-parser && docker compose ps",
+    "docker logs --since 3m iaai-parser-worker-1 2>&1 | tail -n 120",
+    "docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c \"SELECT now() AS ts, count(*) AS cars FROM iaai_cars; SELECT count(*) AS runs FROM iaai_sync_runs;\"",
+]
+
+client = paramiko.SSHClient()
+client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+client.connect(
+    HOST,
+    username=USER,
+    password=PASSWORD,
+    look_for_keys=False,
+    allow_agent=False,
+    timeout=30,
+    banner_timeout=30,
+    auth_timeout=30,
+)
+try:
+    for cmd in cmds:
+        print(f"REMOTE_RUN {cmd}", flush=True)
+        stdin, stdout, stderr = client.exec_command(cmd, timeout=120)
+        code = stdout.channel.recv_exit_status()
+        print(stdout.read().decode("utf-8", "replace"), end="")
+        print(stderr.read().decode("utf-8", "replace"), end="")
+        print(f"EXIT {code}", flush=True)
+finally:
+    client.close()

clean_vps_db.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import os
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+
+commands = [
+    "cd /root/iaai-parser && docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c 'TRUNCATE TABLE iaai_images, iaai_cars, iaai_sync_runs RESTART IDENTITY CASCADE;'",
+    "docker exec -i iaai-redis redis-cli FLUSHALL",
+    "docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c 'SELECT count(*) AS cars FROM iaai_cars; SELECT count(*) AS runs FROM iaai_sync_runs;'",
+    "cd /root/iaai-parser && docker compose ps",
+    "docker logs --since 2m iaai-parser-worker-1 2>&1 | tail -n 100",
+]
+
+client = paramiko.SSHClient()
+client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+client.connect(
+    HOST,
+    username=USER,
+    password=PASSWORD,
+    look_for_keys=False,
+    allow_agent=False,
+    timeout=30,
+    banner_timeout=30,
+    auth_timeout=30,
+)
+try:
+    for command in commands:
+        print(f"REMOTE_RUN {command}", flush=True)
+        stdin, stdout, stderr = client.exec_command(command, timeout=180)
+        exit_code = stdout.channel.recv_exit_status()
+        print(stdout.read().decode("utf-8", "replace"), end="")
+        print(stderr.read().decode("utf-8", "replace"), end="")
+        print(f"EXIT {exit_code}", flush=True)
+        if exit_code != 0:
+            raise SystemExit(exit_code)
+finally:
+    client.close()

deploy_vps.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import os
+import posixpath
+import stat
+import tarfile
+import time
+import socket
+from pathlib import Path
+
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+LOCAL_ROOT = Path(__file__).resolve().parent
+ARCHIVE = LOCAL_ROOT / "iaai_deploy.tar.gz"
+REMOTE_DIR = "/root/iaai-parser"
+REMOTE_ARCHIVE = "/root/iaai_deploy.tar.gz"
+
+EXCLUDE_DIRS = {
+    ".git",
+    ".venv",
+    "__pycache__",
+    ".pytest_cache",
+    ".mypy_cache",
+    ".ruff_cache",
+    "iaai_scraper.egg-info",
+}
+EXCLUDE_FILES = {"iaai_deploy.tar.gz", "deploy_vps.py"}
+
+
+def _include(path: Path) -> bool:
+    rel = path.relative_to(LOCAL_ROOT)
+    parts = set(rel.parts)
+    if parts & EXCLUDE_DIRS:
+        return False
+    if path.name in EXCLUDE_FILES:
+        return False
+    if path.suffix in {".pyc", ".pyo"}:
+        return False
+    return True
+
+
+def make_archive() -> None:
+    if ARCHIVE.exists():
+        ARCHIVE.unlink()
+    with tarfile.open(ARCHIVE, "w:gz") as tf:
+        for path in LOCAL_ROOT.rglob("*"):
+            if not _include(path):
+                continue
+            tf.add(path, arcname=str(path.relative_to(LOCAL_ROOT)))
+    print(f"ARCHIVE_READY {ARCHIVE} {ARCHIVE.stat().st_size} bytes", flush=True)
+
+
+def connect() -> paramiko.SSHClient:
+    last_exc: BaseException | None = None
+    for attempt in range(1, 6):
+        client = paramiko.SSHClient()
+        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        try:
+            print(f"SSH_CONNECT attempt={attempt}", flush=True)
+            client.connect(
+                HOST,
+                username=USER,
+                password=PASSWORD,
+                look_for_keys=False,
+                allow_agent=False,
+                timeout=45,
+                banner_timeout=60,
+                auth_timeout=45,
+            )
+            return client
+        except (paramiko.SSHException, socket.timeout, OSError) as exc:
+            last_exc = exc
+            client.close()
+            print(f"SSH_CONNECT_RETRY attempt={attempt} error={type(exc).__name__}: {exc}", flush=True)
+            time.sleep(3 * attempt)
+    raise SystemExit(f"SSH_CONNECT_FAILED: {last_exc}")
+
+
+def run(client: paramiko.SSHClient, cmd: str, timeout: int | None = None) -> None:
+    print(f"REMOTE_RUN {cmd}", flush=True)
+    stdin, stdout, stderr = client.exec_command(cmd, timeout=timeout)
+    exit_code = stdout.channel.recv_exit_status()
+    out = stdout.read().decode("utf-8", "replace")
+    err = stderr.read().decode("utf-8", "replace")
+    if out:
+        print(out, end="", flush=True)
+    if err:
+        print(err, end="", flush=True)
+    if exit_code != 0:
+        raise SystemExit(f"REMOTE_FAILED code={exit_code}: {cmd}")
+
+
+def upload_file(sftp: paramiko.SFTPClient, local: Path, remote: str) -> None:
+    total = local.stat().st_size
+    last = 0.0
+
+    def cb(done: int, _total: int) -> None:
+        nonlocal last
+        now = time.time()
+        if now - last >= 2 or done == total:
+            print(f"UPLOAD {done}/{total}", flush=True)
+            last = now
+
+    sftp.put(str(local), remote, callback=cb)
+
+
+def main() -> None:
+    make_archive()
+    client = connect()
+    try:
+        run(client, "echo VPS_OK && hostname && docker --version && docker compose version")
+        sftp = client.open_sftp()
+        try:
+            upload_file(sftp, ARCHIVE, REMOTE_ARCHIVE)
+        finally:
+            sftp.close()
+        run(
+            client,
+            f"mkdir -p {REMOTE_DIR} && cd {REMOTE_DIR} && docker compose down || true && "
+            f"find {REMOTE_DIR} -mindepth 1 -maxdepth 1 ! -name '.env' -exec rm -rf {{}} + && "
+            f"tar -xzf {REMOTE_ARCHIVE} -C {REMOTE_DIR} && rm -f {REMOTE_ARCHIVE}",
+            timeout=300,
+        )
+        run(client, f"cd {REMOTE_DIR} && docker compose up -d --build", timeout=1800)
+        run(client, f"cd {REMOTE_DIR} && docker compose ps", timeout=120)
+        run(client, "docker logs --since 2m iaai-parser-worker-1 2>&1 | tail -n 120 || docker compose -f /root/iaai-parser/docker-compose.yml logs --since 2m worker | tail -n 120", timeout=120)
+    finally:
+        client.close()
+
+
+if __name__ == "__main__":
+    main()

docker-compose.yml

@@ -7,36 +7,53 @@ x-app-env: &app-env
   IAAI_DATABASE_POOL_RECYCLE_SECONDS: ${IAAI_DATABASE_POOL_RECYCLE_SECONDS:-1800}
   IAAI_DATABASE_POOL_SIZE: ${IAAI_DATABASE_POOL_SIZE:-20}
   IAAI_DATABASE_MAX_OVERFLOW: ${IAAI_DATABASE_MAX_OVERFLOW:-40}
-  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-900}
+  # Для больших Search-выборок (десятки тысяч лотов) run должен жить дольше одного батча.
-  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-1200}
+  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-7200}
-  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-2400}
+  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-7500}
+  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-10800}
   IAAI_PROFILE: ${IAAI_PROFILE:-fast}
   IAAI_SCRAPING_PROFILE: ${IAAI_SCRAPING_PROFILE:-${IAAI_PROFILE:-fast}}
   IAAI_HTTP_FIRST: ${IAAI_HTTP_FIRST:-true}
   IAAI_BROWSER_FALLBACK_ENABLED: ${IAAI_BROWSER_FALLBACK_ENABLED:-false}
   IAAI_ANONYMOUS_BOOTSTRAP_ENABLED: ${IAAI_ANONYMOUS_BOOTSTRAP_ENABLED:-false}
   IAAI_CHALLENGE_REFRESH_ATTEMPTS: ${IAAI_CHALLENGE_REFRESH_ATTEMPTS:-1}
-  IAAI_LISTING_POST_ATTEMPTS: ${IAAI_LISTING_POST_ATTEMPTS:-2}
+  IAAI_LISTING_POST_ATTEMPTS: ${IAAI_LISTING_POST_ATTEMPTS:-1}
-  IAAI_REQUEST_JITTER_MAX_S: ${IAAI_REQUEST_JITTER_MAX_S:-0}
+  IAAI_REQUEST_JITTER_MAX_S: ${IAAI_REQUEST_JITTER_MAX_S:-0.03}
-  IAAI_DETAIL_RETRIES: ${IAAI_DETAIL_RETRIES:-1}
+  IAAI_DETAIL_RETRIES: ${IAAI_DETAIL_RETRIES:-2}
   IAAI_LISTING_RETRIES: ${IAAI_LISTING_RETRIES:-1}
   # 0 = без лимита.
   CELERY_BEAT_SYNC_LIMIT: ${CELERY_BEAT_SYNC_LIMIT:-0}
+  # Следующий плановый прогон — не раньше чем через час после предыдущего tick beat.
+  CELERY_BEAT_SYNC_INTERVAL_MINUTES: ${CELERY_BEAT_SYNC_INTERVAL_MINUTES:-60}
+  # Любой внутренний follow-up после неполного bootstrap тоже не стартует сразу.
+  IAAI_SYNC_FOLLOWUP_MIN_DELAY_SECONDS: ${IAAI_SYNC_FOLLOWUP_MIN_DELAY_SECONDS:-3600}
   CELERY_WORKER_CONCURRENCY: ${CELERY_WORKER_CONCURRENCY:-4}
   CELERY_WORKER_POOL: ${CELERY_WORKER_POOL:-prefork}
   CELERY_WORKER_MAX_TASKS_PER_CHILD: ${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
-  CELERY_BATCH_SIZE: ${CELERY_BATCH_SIZE:-1000}
+  CELERY_BATCH_SIZE: ${CELERY_BATCH_SIZE:-1500}
   IAAI_INTER_BATCH_DELAY_SECONDS: ${IAAI_INTER_BATCH_DELAY_SECONDS:-0}
+  # Стабильный fast-профиль: не душим IAAI слишком большим числом HTTPS detail-соединений.
+  IAAI_FETCH_CONCURRENCY: ${IAAI_FETCH_CONCURRENCY:-96}
+  IAAI_MAX_RETRIES: ${IAAI_MAX_RETRIES:-2}
+  IAAI_RETRY_DELAY_SECONDS: ${IAAI_RETRY_DELAY_SECONDS:-0.35}
+  CELERY_PARALLEL_SEGMENTS: ${CELERY_PARALLEL_SEGMENTS:-false}
   IAAI_FAIL_RATE_THRESHOLD: ${IAAI_FAIL_RATE_THRESHOLD:-0.9}
   IAAI_PARALLEL_TABS: ${IAAI_PARALLEL_TABS:-8}
-  IAAI_FETCH_CONCURRENCY: ${IAAI_FETCH_CONCURRENCY:-32}
   IAAI_BLOCK_RESOURCES: ${IAAI_BLOCK_RESOURCES:-true}
-  IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-runtime}
+  IAAI_FILTERED_SEARCH_URL: ${IAAI_FILTERED_SEARCH_URL:-}
+  IAAI_FILTERED_SEARCH_URLS: ${IAAI_FILTERED_SEARCH_URLS:-}
+  IAAI_FAST_PATH_TIMEOUT_MS: ${IAAI_FAST_PATH_TIMEOUT_MS:-10000}
+  IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-[]}
+  IAAI_DISCOVERY_MODE: ${IAAI_DISCOVERY_MODE:-listing}
+  IAAI_HOURLY_MODE: ${IAAI_HOURLY_MODE:-rolling_refresh}
+  IAAI_HOURLY_REFRESH_BATCH_SIZE: ${IAAI_HOURLY_REFRESH_BATCH_SIZE:-500}
   IAAI_MAX_PAGES_PER_RUN: ${IAAI_MAX_PAGES_PER_RUN:-9999}
   IAAI_MAX_VEHICLES_PER_RUN: ${IAAI_MAX_VEHICLES_PER_RUN:-50000}
-  IAAI_ALWAYS_FULL_SCAN: ${IAAI_ALWAYS_FULL_SCAN:-true}
+  IAAI_ALWAYS_FULL_SCAN: ${IAAI_ALWAYS_FULL_SCAN:-false}
-  IAAI_HUMAN_PACE_ENABLED: ${IAAI_HUMAN_PACE_ENABLED:-true}
+  IAAI_HUMAN_PACE_ENABLED: ${IAAI_HUMAN_PACE_ENABLED:-false}
-  IAAI_TOKENS_FILE: ${IAAI_TOKENS_FILE:-/home/app/tokens.json}
+  # Первый sync после clean restart стартует сразу; следующий запуск — только через hourly guard/beat.
+  IAAI_STARTUP_SYNC_ENABLED: ${IAAI_STARTUP_SYNC_ENABLED:-true}
+  IAAI_TOKENS_FILE: ${IAAI_TOKENS_FILE:-/data/tokens.json}
   IAAI_RUNTIME_CONFIG_FILE: ${IAAI_RUNTIME_CONFIG_FILE:-/app/runtime_config.json}
   IAAI_BROWSER_ENGINE: chromium
   # Self-heal для worker.
@@ -166,7 +183,9 @@ services:
     stop_grace_period: 60s
     command: >
       celery -A iaai_scraper.worker.celery_app worker
-      --loglevel=info --concurrency=${CELERY_WORKER_CONCURRENCY:-4} --pool=${CELERY_WORKER_POOL:-prefork}
+      --loglevel=info
+      --concurrency=${CELERY_WORKER_CONCURRENCY:-4}
+      --pool=${CELERY_WORKER_POOL:-prefork}
       --pidfile=/tmp/celery-worker.pid
       -Q iaai_sync --max-tasks-per-child=${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
     healthcheck:

iaai_scraper/browser/fast_client.py

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

iaai_scraper/core/config.py

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

iaai_scraper/core/logs.py

@@ -18,8 +18,9 @@ def set_trace_id(trace_id: str) -> None:
 
 
 def setup_logging(level: str = "INFO", log_file: str | None = None) -> None:
-    # stderr — Docker и Celery prefork корректно его подхватывают.
+    # stdout — основной поток для `docker logs`, Docker Desktop и compose logs.
-    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stderr)]
+    # stderr в PowerShell часто выглядит как NativeCommandError, хотя это обычные логи.
+    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)]
     if log_file:
         handlers.append(logging.FileHandler(log_file, encoding="utf-8"))
     trace_filter = TraceIdFilter()

iaai_scraper/fast_sync.py

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

iaai_scraper/scraper.py

@@ -2234,6 +2234,8 @@ class IAAIScraper:
             only_new = rc.only_new
         if lane == "iaai_cars" and rc.lane is not None:
             lane = rc.lane
+        if listing_url is None and rc.name:
+            listing_url = rc.name
 
         trace_id = self._new_trace_id("sync-listing")
         started_at = time.perf_counter()
@@ -2260,6 +2262,8 @@ class IAAIScraper:
                         year_min,
                         year_max,
                     )
+                if listing_url:
+                    logger.info("sync_listing uses explicit listing_url from runtime/args: %s", listing_url)
                 fast_engine = FastSyncEngine(
                     client=self.fast_client,
                     mapper=self.fast_mapper,
@@ -2424,19 +2428,25 @@ class IAAIScraper:
             seg_make = seg.get("make")
             seg_year_min = seg.get("year_min")
             seg_year_max = seg.get("year_max")
+            seg_listing_url = seg.get("listing_url")
 
             # На длительном full-scan сегмент нельзя пропускать из-за runtime include.brands,
             # иначе прогон становится частичным.
             self._reload_runtime_config()
 
-            seg_url = None if self.settings.scraping_profile.http_first else (self._build_segment_listing_url(base_url, seg_make) if seg_make else None)
+            if seg_listing_url:
+                seg_url = str(seg_listing_url)
+            else:
+                seg_url = None if self.settings.scraping_profile.http_first else (self._build_segment_listing_url(base_url, seg_make) if seg_make else None)
 
             seg_label = f"{seg_make or 'ALL'}"
+            if seg_listing_url:
+                seg_label = "FILTERED_SEARCH"
             if seg_year_min is not None or seg_year_max is not None:
                 seg_label += f" ({seg_year_min}-{seg_year_max})"
 
-            logger.info(
+            logger.debug(
-                "Segment %d/%d: %s",
+                "Segment %d/%d started: %s",
                 seg_idx + 1, len(segments), seg_label,
             )
 

iaai_scraper/worker/celery_app.py

@@ -16,6 +16,9 @@ logger = logging.getLogger("iaai_scraper.worker.celery_app")
 STARTUP_SYNC_DISPATCH_KEY = "iaai:state:startup_sync_dispatched"
 IAAI_SYNC_QUEUE = "iaai_sync"
 PROGRESS_KEY_PREFIX = "iaai:state:task_progress:"
+SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
+SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
+SYNC_LAST_COMPLETED_AT_KEY = "iaai:state:sync_listing_last_completed_at"
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -43,6 +46,26 @@ def _has_fresh_active_progress(redis_client: Redis, *, max_age_seconds: int = 18
     return False
 
 
+def _seconds_until_next_allowed_sync(redis_client: Redis) -> int:
+    """Запрещает новый автозапуск раньше чем через интервал beat после завершения полного run."""
+    min_interval = max(0, int(settings.celery.beat_sync_interval_minutes * 60))
+    if min_interval <= 0:
+        return 0
+    try:
+        full_done = str(redis_client.get(SYNC_FULL_SCAN_DONE_KEY) or "").strip().lower() in {"1", "true", "yes", "on"}
+        if not full_done:
+            return 0
+        completed_raw = redis_client.get(SYNC_LAST_COMPLETED_AT_KEY)
+        if not completed_raw:
+            return 0
+        completed_at = int(float(completed_raw))
+    except Exception:
+        logger.warning("Failed to inspect last sync completion timestamp", exc_info=True)
+        return 0
+    elapsed = int(time.time()) - completed_at
+    return max(0, min_interval - elapsed)
+
+
 @celery_setup_logging.connect
 def _configure_logging(loglevel=None, **kwargs):
     # Перехватываем логирование Celery и пишем только в stderr (Docker logs).
@@ -116,6 +139,7 @@ celery_app.conf.update(
             "options": {
                 "queue": IAAI_SYNC_QUEUE,
                 "expires": settings.celery.beat_sync_interval_minutes * 60.0,
+                "headers": {"iaai_beat_task": True},
             },
         }
     },
@@ -147,7 +171,15 @@ def _on_worker_ready(**kwargs):
         )
 
         has_fresh_progress = _has_fresh_active_progress(redis_client)
-        for stale_key in ("iaai:locks:sync_listing",):
+        next_allowed_delay = _seconds_until_next_allowed_sync(redis_client)
+        if next_allowed_delay > 0:
+            logger.info(
+                "Worker ready: last full sync finished recently; next auto sync allowed in %ss, skip startup dispatch",
+                next_allowed_delay,
+            )
+            return
+
+        for stale_key in (SYNC_LISTING_LOCK_KEY,):
             try:
                 ttl = redis_client.ttl(stale_key)
                 if ttl is not None and ttl != -2 and not has_fresh_progress:

iaai_scraper/worker/fast_sync.py

@@ -0,0 +1,479 @@
+from __future__ import annotations
+
+import concurrent.futures
+import logging
+import random
+import time
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from .browser.fast_client import FastListingVehicle, IAAIFastClient
+from .core.runtime_config import RuntimeConfig
+from .parsing.fast_mapper import INACTIVE_STATUS_VALUES, FastCarMapper
+from .storage.db import PersistenceService
+from .storage.schemas import CarRecord
+
+logger = logging.getLogger("iaai_scraper.fast_sync")
+
+
+@dataclass(slots=True)
+class FastSyncStats:
+    ids_fetched: int = 0
+    cars_upserted: int = 0
+    cars_failed: int = 0
+    cars_filtered: int = 0
+    images_upserted: int = 0
+    skipped_existing: int = 0
+    protection_events: int = 0
+
+
+def passes_condition_check(row: FastListingVehicle) -> bool:
+    if row.timed_auction_closed:
+        return False
+    status = (row.inventory_status or "").strip().upper()
+    return status not in INACTIVE_STATUS_VALUES
+
+
+class FastSyncEngine:
+    """Full iaai-fast style sync pipeline adapted to this project's storage.
+
+    Flow: hidden listing payloads -> concurrent ProductDetailsVM HTTP fetch ->
+    CarRecord preparation in memory -> single batch DB upsert. Browser is used
+    only inside IAAIFastClient to refresh cookies when IAAI challenge appears.
+    """
+
+    def __init__(
+        self,
+        *,
+        client: IAAIFastClient,
+        mapper: FastCarMapper,
+        persistence: PersistenceService,
+        batch_size: int,
+        fetch_concurrency: int,
+        report_progress: Callable[[str, Any], None] | None = None,
+    ) -> None:
+        self.client = client
+        self.mapper = mapper
+        self.persistence = persistence
+        self.batch_size = max(1, int(batch_size))
+        self.fetch_concurrency = max(1, int(fetch_concurrency))
+        self.report_progress = report_progress
+
+    @staticmethod
+    def _is_transient_detail_error(exc: Exception) -> bool:
+        text = str(exc).lower()
+        return any(
+            marker in text
+            for marker in (
+                "sslerror",
+                "ssleoferror",
+                "unexpected_eof_while_reading",
+                "eof occurred in violation of protocol",
+                "max retries exceeded",
+                "read timed out",
+                "readtimeout",
+                "connection reset",
+                "connection aborted",
+                "connection closed",
+                "temporarily unavailable",
+                "too many requests",
+                "status=429",
+                "status=500",
+                "status=502",
+                "status=503",
+                "status=504",
+            )
+        )
+
+    def sync_listing(
+        self,
+        *,
+        runtime_config: RuntimeConfig,
+        make: str | None = None,
+        model: str | None = None,
+        lane: str = "iaai_cars",
+        limit: int | None = None,
+        only_new: bool = False,
+        listing_url: str | None = None,
+        max_pages: int | None = None,
+        skip_mark_sold: bool = False,
+    ) -> dict[str, Any]:
+        del lane
+        started_at = time.perf_counter()
+        stats = FastSyncStats()
+        errors: list[dict[str, str]] = []
+        selected: dict[str, FastListingVehicle] = {}
+        rows_seen = 0
+        rows_skipped_condition = 0
+
+        filters = runtime_config.filters
+        logger.info(
+            "Fast HTTP-first listing started: make=%s listing_url=%s max_pages=%s concurrency=%s batch_size=%s",
+            make or "ALL",
+            listing_url or "default",
+            max_pages,
+            self.fetch_concurrency,
+            self.batch_size,
+        )
+        for vehicle in self.client.iter_listing_vehicles(
+            listing_start_url=listing_url,
+            make=make,
+            max_pages=max_pages,
+        ):
+            rows_seen += 1
+            if runtime_config.sync.condition_check_enabled and not passes_condition_check(vehicle):
+                rows_skipped_condition += 1
+                continue
+            if vehicle.inventory_id in selected:
+                continue
+            selected[vehicle.inventory_id] = vehicle
+            if limit is not None and limit > 0 and len(selected) >= limit:
+                break
+
+        candidates = list(selected.values())
+        all_listing_origin_urls = {f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates}
+        if only_new and candidates:
+            origin_urls = [f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates]
+            origin_ids = [f"iaai:{v.inventory_id}" for v in candidates]
+            existing_urls, existing_ids = self.persistence.get_existing_urls_and_ids(origin_urls, origin_ids)
+            fresh: list[FastListingVehicle] = []
+            for vehicle in candidates:
+                if f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}" in existing_urls or f"iaai:{vehicle.inventory_id}" in existing_ids:
+                    stats.skipped_existing += 1
+                    continue
+                fresh.append(vehicle)
+            candidates = fresh
+
+        self._progress(
+            "fast_listing_collected",
+            rows_seen=rows_seen,
+            rows_filtered_condition=rows_skipped_condition,
+            rows_selected=len(candidates),
+            skipped_existing=stats.skipped_existing,
+        )
+        logger.info(
+            "Fast HTTP-first listing collected: rows_seen=%d selected=%d skipped_existing=%d filtered_condition=%d only_new=%s",
+            rows_seen,
+            len(candidates),
+            stats.skipped_existing,
+            rows_skipped_condition,
+            only_new,
+        )
+
+        scan_completed = not (limit is not None and limit > 0) and not errors
+
+        if not candidates:
+            return self._result(
+                started_at=started_at,
+                stats=stats,
+                failures=errors,
+                listing={
+                    "mode": "fast_hidden_payload",
+                    "vehicles_collected": 0,
+                    "vehicle_urls": [],
+                    "early_stopped": False,
+                    "truncated_by_time_budget": False,
+                    "rows_seen": rows_seen,
+                    "rows_filtered_condition": rows_skipped_condition,
+                },
+                all_listing_origin_urls=all_listing_origin_urls,
+                full_scan_completed=scan_completed,
+            )
+
+        prepared_rows: list[CarRecord] = []
+        db_processed = 0
+        retry_candidates: list[FastListingVehicle] = []
+        transient_failure_log_count = 0
+        started_details = time.perf_counter()
+        with concurrent.futures.ThreadPoolExecutor(max_workers=min(self.fetch_concurrency, len(candidates))) as executor:
+            future_to_vehicle = {
+                executor.submit(self._fetch_detail_payload, vehicle.inventory_id): vehicle
+                for vehicle in candidates
+            }
+            for index, future in enumerate(concurrent.futures.as_completed(future_to_vehicle), start=1):
+                vehicle = future_to_vehicle[future]
+                try:
+                    payload = future.result()
+                    record = self.mapper.map_payload_to_record(
+                        detail_payload=payload,
+                        vehicle_url=f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}",
+                        listing_vehicle=vehicle,
+                    )
+                    if model and model.casefold() not in record.model.casefold():
+                        stats.cars_filtered += 1
+                        continue
+                    if not filters.matches({
+                        "brand": record.brand,
+                        "model": record.model,
+                        "year": record.year,
+                        "body_type": record.body_type,
+                        "color": record.color,
+                        "drive": record.drive,
+                        "gearbox": record.gearbox,
+                        "price": record.price,
+                        "mileage": record.mileage,
+                    }):
+                        stats.cars_filtered += 1
+                        continue
+                    prepared_rows.append(record)
+                    stats.ids_fetched += 1
+                    if len(prepared_rows) >= self.batch_size:
+                        db_processed += self._flush_db_records(
+                            rows=prepared_rows,
+                            stats=stats,
+                            errors=errors,
+                            processed=db_processed + len(prepared_rows),
+                            total=len(candidates),
+                        )
+                        prepared_rows.clear()
+                except Exception as exc:
+                    stats.cars_failed += 1
+                    errors.append({"vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}", "error": str(exc)})
+                    if _looks_like_protection(exc):
+                        stats.protection_events += 1
+                    if self._is_transient_detail_error(exc):
+                        retry_candidates.append(vehicle)
+                        transient_failure_log_count += 1
+                        if transient_failure_log_count % 100 == 0:
+                            logger.warning(
+                                "Fast detail transient failures queued for retry: count=%d latest_inventory_id=%s",
+                                transient_failure_log_count,
+                                vehicle.inventory_id,
+                            )
+                    else:
+                        logger.exception("Fast detail parse failed inventory_id=%s: %s", vehicle.inventory_id, exc)
+
+                if index % 100 == 0 or index == len(candidates):
+                    elapsed = max(0.001, time.perf_counter() - started_details)
+                    if self.client._settings.scraping_profile.verbose_progress_logs:
+                        logger.info(
+                            "Fast HTTP-first details progress: processed=%d/%d ok=%d failed=%d queued_db=%d rate=%.2f/s",
+                            index,
+                            len(candidates),
+                            stats.ids_fetched,
+                            stats.cars_failed,
+                            len(prepared_rows),
+                            index / elapsed,
+                        )
+                    self._progress(
+                        "fast_detail_progress",
+                        processed=index,
+                        total=len(candidates),
+                        ids_fetched=stats.ids_fetched,
+                        cars_failed=stats.cars_failed,
+                        queued_for_db=len(prepared_rows),
+                        throughput=round(index / elapsed, 2),
+                    )
+
+        if retry_candidates:
+            retry_started = time.perf_counter()
+            retry_workers = max(1, min(8, self.fetch_concurrency // 2, len(retry_candidates)))
+            retry_errors: list[dict[str, str]] = []
+            logger.info(
+                "Fast HTTP-first retrying transient detail failures: total=%d workers=%d",
+                len(retry_candidates),
+                retry_workers,
+            )
+            with concurrent.futures.ThreadPoolExecutor(max_workers=retry_workers) as executor:
+                future_to_vehicle = {
+                    executor.submit(self._fetch_detail_payload, vehicle.inventory_id): vehicle
+                    for vehicle in retry_candidates
+                }
+                for retry_index, future in enumerate(concurrent.futures.as_completed(future_to_vehicle), start=1):
+                    vehicle = future_to_vehicle[future]
+                    try:
+                        payload = future.result()
+                        record = self.mapper.map_payload_to_record(
+                            detail_payload=payload,
+                            vehicle_url=f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}",
+                            listing_vehicle=vehicle,
+                        )
+                        if model and model.casefold() not in record.model.casefold():
+                            stats.cars_filtered += 1
+                            continue
+                        if not filters.matches({
+                            "brand": record.brand,
+                            "model": record.model,
+                            "year": record.year,
+                            "body_type": record.body_type,
+                            "color": record.color,
+                            "drive": record.drive,
+                            "gearbox": record.gearbox,
+                            "price": record.price,
+                            "mileage": record.mileage,
+                        }):
+                            stats.cars_filtered += 1
+                            continue
+                        prepared_rows.append(record)
+                        stats.ids_fetched += 1
+                        stats.cars_failed = max(0, stats.cars_failed - 1)
+                        if len(prepared_rows) >= self.batch_size:
+                            db_processed += self._flush_db_records(
+                                rows=prepared_rows,
+                                stats=stats,
+                                errors=errors,
+                                processed=db_processed + len(prepared_rows),
+                                total=len(candidates),
+                            )
+                            prepared_rows.clear()
+                    except Exception as exc:
+                        retry_errors.append({
+                            "vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}",
+                            "error": str(exc),
+                        })
+                        if _looks_like_protection(exc):
+                            stats.protection_events += 1
+                    if retry_index % 100 == 0 or retry_index == len(retry_candidates):
+                        elapsed = max(0.001, time.perf_counter() - retry_started)
+                        logger.info(
+                            "Fast HTTP-first retry progress: processed=%d/%d recovered=%d remaining_failed=%d rate=%.2f/s",
+                            retry_index,
+                            len(retry_candidates),
+                            len(retry_candidates) - len(retry_errors),
+                            len(retry_errors),
+                            retry_index / elapsed,
+                        )
+            transient_urls = {f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in retry_candidates}
+            errors = [error for error in errors if error.get("vehicle_url") not in transient_urls]
+            errors.extend(retry_errors)
+
+        if prepared_rows:
+            db_processed += self._flush_db_records(
+                rows=prepared_rows,
+                stats=stats,
+                errors=errors,
+                processed=db_processed + len(prepared_rows),
+                total=len(candidates),
+            )
+            prepared_rows.clear()
+
+        self.client.persist_session_state()
+
+        scan_completed = not (limit is not None and limit > 0) and not errors
+        mark_sold_scope_partial = bool(
+            (limit is not None and limit > 0)
+            or make
+            or model
+            or listing_url
+            or only_new
+        )
+        if all_listing_origin_urls and not mark_sold_scope_partial and not skip_mark_sold and scan_completed:
+            try:
+                sold_count = self.persistence.mark_sold_not_in_listing_by_urls(all_listing_origin_urls, lane="iaai")
+            except Exception as exc:
+                sold_count = 0
+                logger.warning("Fast sold reconcile failed: %s", exc)
+        else:
+            sold_count = 0
+
+        listing = {
+            "mode": "fast_hidden_payload",
+            "vehicles_collected": len(candidates),
+            "vehicle_urls": [f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates],
+            "early_stopped": False,
+            "truncated_by_time_budget": False,
+            "rows_seen": rows_seen,
+            "rows_filtered_condition": rows_skipped_condition,
+            "sold_marked": sold_count,
+        }
+        return self._result(
+            started_at=started_at,
+            stats=stats,
+            failures=errors,
+            listing=listing,
+            all_listing_origin_urls=all_listing_origin_urls,
+            full_scan_completed=scan_completed,
+        )
+
+    def _result(
+        self,
+        *,
+        started_at: float,
+        stats: FastSyncStats,
+        failures: list[dict[str, str]],
+        listing: dict[str, Any],
+        all_listing_origin_urls: set[str],
+        full_scan_completed: bool,
+    ) -> dict[str, Any]:
+        status = "success" if not failures else ("partial_success" if stats.cars_upserted else "failed")
+        total = int(listing.get("vehicles_collected") or 0)
+        fail_ratio = (stats.cars_failed / total) if total > 0 else 0.0
+        protection_ratio = (stats.protection_events / total) if total > 0 else 0.0
+        anti_bot_detected = total > 0 and ((stats.protection_events >= 30 and protection_ratio >= 0.10) or fail_ratio >= 0.30)
+        return {
+            "status": status,
+            "listing": listing,
+            "total": total,
+            "total_discovered": total,
+            "skipped_existing": stats.skipped_existing,
+            "cars_upserted": stats.cars_upserted,
+            "cars_failed": stats.cars_failed,
+            "cars_filtered": stats.cars_filtered,
+            "images_upserted": stats.images_upserted,
+            "protection_events": stats.protection_events,
+            "failures": failures,
+            "all_listing_origin_urls": all_listing_origin_urls,
+            "full_scan_completed": full_scan_completed and not anti_bot_detected,
+            "anti_bot_detected": anti_bot_detected,
+            "fail_ratio": round(fail_ratio, 4),
+            "protection_ratio": round(protection_ratio, 4),
+            "elapsed_seconds": round(time.perf_counter() - started_at, 3),
+        }
+
+    def _progress(self, stage: str, **meta: Any) -> None:
+        if self.report_progress is None:
+            return
+        try:
+            self.report_progress(stage, **meta)
+        except Exception:
+            pass
+
+    def _fetch_detail_payload(self, inventory_id: str) -> dict[str, Any]:
+        jitter = float(self.client._settings.scraping_profile.request_jitter_max_s)
+        if jitter > 0:
+            time.sleep(random.uniform(0.0, jitter))
+        return self.client.fetch_vehicle_detail_payload(inventory_id)
+
+    def _flush_db_records(
+        self,
+        *,
+        rows: list[CarRecord],
+        stats: FastSyncStats,
+        errors: list[dict[str, str]],
+        processed: int,
+        total: int,
+    ) -> int:
+        if not rows:
+            return 0
+        batch = list(rows)
+        try:
+            upsert = self.persistence.upsert_cars_batch(batch)
+            stats.cars_upserted += int(upsert.get("inserted", 0)) + int(upsert.get("updated", 0))
+            stats.images_upserted += int(upsert.get("images_upserted", 0))
+        except Exception as exc:
+            stats.cars_failed += len(batch)
+            errors.append({"vehicle_url": f"db_batch_{processed - len(batch)}", "error": str(exc)})
+            logger.exception("Fast DB apply failed processed=%s size=%s: %s", processed, len(batch), exc)
+        self._progress(
+            "fast_db_progress",
+            processed=processed,
+            total=total,
+            cars_upserted=stats.cars_upserted,
+            cars_failed=stats.cars_failed,
+            images_upserted=stats.images_upserted,
+        )
+        logger.info(
+            "Fast HTTP-first DB batch: processed=%d/%d batch=%d upserted=%d failed=%d images=%d",
+            processed,
+            total,
+            len(batch),
+            stats.cars_upserted,
+            stats.cars_failed,
+            stats.images_upserted,
+        )
+        return len(batch)
+
+
+def _looks_like_protection(exc: Exception) -> bool:
+    message = str(exc).lower()
+    return any(token in message for token in ("captcha", "antibot", "challenge", "blocked", "403", "429", "incapsula"))

iaai_scraper/worker/self_heal.py

@@ -19,6 +19,15 @@ SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
 SYNC_LISTING_CHECKPOINT_KEY = "iaai:state:sync_listing_checkpoint"
 SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
 SIGKILL_FALLBACK = getattr(signal, "SIGKILL", signal.SIGTERM)
+TERMINAL_PROGRESS_STAGES = {
+	"segment_done",
+	"segment_failed",
+	"segment_task_completed",
+	"segment_task_failed",
+	"segment_task_soft_timeout",
+	"sync_done",
+	"failed",
+}
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -108,6 +117,31 @@ def _read_last_db_progress_ts(redis_client: Redis) -> int | None:
 	return max_ts or None
 
 
+def _has_active_recent_progress(redis_client: Redis, now_ts: int, stall_seconds: int) -> bool:
+	"""Return True when a task is still reporting non-DB progress.
+
+	Hourly only_new runs can legitimately skip every listed vehicle as existing.
+	Those runs may not emit fast_db_progress for a long time, but they still emit
+	regular listing/segment progress. Treating old DB timestamps as fatal caused
+	the watchdog to kill healthy production workers during such scans.
+	"""
+	for key in redis_client.scan_iter(match="iaai:state:task_progress:*"):
+		try:
+			payload = redis_client.get(key)
+			if not payload:
+				continue
+			data = json.loads(payload)
+			stage = str(data.get("stage") or "")
+			if stage in {"failed", "sync_done", "segment_task_failed", "segment_task_soft_timeout"}:
+				continue
+			ts = _safe_int(data.get("ts"), 0)
+			if ts > 0 and now_ts - ts <= max(60, int(stall_seconds)):
+				return True
+		except Exception:
+			continue
+	return False
+
+
 def _reset_bootstrap_checkpoint_for_db_idle(redis_client: Redis) -> None:
 	pipe = redis_client.pipeline()
 	pipe.delete(SYNC_LISTING_CHECKPOINT_KEY)
@@ -123,13 +157,43 @@ def _has_inflight_work(redis_client: Redis, queue_name: str) -> tuple[bool, dict
 	queue_len = _safe_int(redis_client.llen(queue_name), 0)
 	has_lock = 1 if redis_client.get(SYNC_LISTING_LOCK_KEY) else 0
 	has_task_progress = 0
-	for _ in redis_client.scan_iter(match="iaai:state:task_progress:*"):
+	progress_keys_seen = 0
-		has_task_progress = 1
+	stale_progress_deleted = 0
-		break
+	stale_seconds = max(180, min(_env_int("IAAI_SELF_HEAL_STALL_SECONDS", 720), 1800))
+	now_ts = int(time.time())
+	for key in redis_client.scan_iter(match="iaai:state:task_progress:*"):
+		progress_keys_seen += 1
+		if queue_len > 0 or has_lock == 1:
+			has_task_progress = 1
+			break
+		try:
+			payload = redis_client.get(key)
+			if not payload:
+				continue
+			data = json.loads(payload)
+			stage = str(data.get("stage") or "")
+			ts = _safe_int(data.get("ts"), 0)
+			is_terminal = stage in TERMINAL_PROGRESS_STAGES
+			is_stale = ts <= 0 or now_ts - ts > stale_seconds
+			if is_terminal or is_stale:
+				redis_client.delete(key)
+				stale_progress_deleted += 1
+				continue
+			has_task_progress = 1
+			break
+		except Exception:
+			# Битые progress payload не должны держать worker в вечном false-stall цикле.
+			try:
+				redis_client.delete(key)
+				stale_progress_deleted += 1
+			except Exception:
+				pass
 	flags = {
 		"queue_len": queue_len,
 		"has_lock": has_lock,
 		"has_task_progress": has_task_progress,
+		"progress_keys_seen": progress_keys_seen,
+		"stale_progress_deleted": stale_progress_deleted,
 	}
 	return (queue_len > 0 or has_lock == 1 or has_task_progress == 1), flags
 
@@ -219,6 +283,12 @@ def main() -> None:
 			restart_reason = f"progress_age={age}s > {stall_seconds}s"
 			db_idle_restart = False
 			if age <= stall_seconds:
+				# If the task is actively reporting progress, do not require DB writes.
+				# Hourly only_new listing scans often skip already-known cars and can
+				# legitimately have no DB writes while still moving through segments.
+				if _has_active_recent_progress(redis_client, now_ts, stall_seconds):
+					time.sleep(check_interval)
+					continue
 				last_db_ts = _read_last_db_progress_ts(redis_client)
 				db_age = None if last_db_ts is None else max(0, now_ts - int(last_db_ts))
 				if db_age is None:

iaai_scraper/worker/tasks.py

@@ -48,6 +48,7 @@ HOURLY_FAILURE_STREAK_LIMIT = 3
 HOURLY_FAILURE_STREAK_TTL_SECONDS = 6 * 60 * 60  # сброс через 6 часов
 SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
 SYNC_LISTING_TASK_NAME = "iaai.sync_cars_feed"
+SYNC_LAST_COMPLETED_AT_KEY = "iaai:state:sync_listing_last_completed_at"
 SYNC_SEGMENT_LOCK_KEY_FMT = "iaai:locks:sync_segment:{idx}"
 SYNC_SEGMENTS_PROGRESS_KEY = "iaai:state:sync_segments_progress"
 SYNC_SEGMENTS_TOTAL_KEY = "iaai:state:sync_segments_total"
@@ -578,9 +579,14 @@ def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) ->
     stage = str(progress.get("stage") or "")
     if stage in TERMINAL_PROGRESS_STAGES:
         return False
+    # Listing/segment progress means the task is alive even if it writes no new
+    # cars. This is normal for hourly only_new runs when all vehicles are already
+    # present in DB. Do not kill healthy scans just because DB progress is idle.
+    progress_ts = _safe_int(progress.get("ts")) or 0
+    if progress_ts > 0 and int(time.time()) - progress_ts < int(db_idle_restart_seconds):
+        return False
     if stage in STALL_WATCHDOG_LONG_RUNNING_STAGES:
         timeout = _stall_timeout_for_progress(stage, db_idle_restart_seconds)
-        progress_ts = _safe_int(progress.get("ts")) or 0
         return progress_ts > 0 and int(time.time()) - progress_ts >= timeout
 
     segments_total = _safe_int(progress.get("segments_total"))
@@ -591,7 +597,6 @@ def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) ->
         return False
 
     now_ts = int(time.time())
-    progress_ts = _safe_int(progress.get("ts")) or 0
     if progress_ts <= 0:
         return False
 
@@ -863,6 +868,33 @@ def _clear_followup_pending(redis_client: Redis) -> None:
         logger.warning("Failed to clear follow-up pending flag", exc_info=True)
 
 
+def _mark_sync_completed(redis_client: Redis) -> None:
+    try:
+        redis_client.set(SYNC_LAST_COMPLETED_AT_KEY, str(int(time.time())), ex=7 * 24 * 60 * 60)
+    except Exception:
+        logger.warning("Failed to mark sync completion timestamp", exc_info=True)
+
+
+def _seconds_until_next_allowed_sync(redis_client: Redis, settings: Settings, *, is_beat_task: bool = False) -> int:
+    if is_beat_task:
+        return 0
+    min_interval = max(0, int(settings.celery.beat_sync_interval_minutes * 60))
+    if min_interval <= 0:
+        return 0
+    try:
+        if not _is_full_scan_done(redis_client):
+            return 0
+        completed_raw = redis_client.get(SYNC_LAST_COMPLETED_AT_KEY)
+        if not completed_raw:
+            return 0
+        completed_at = int(float(completed_raw))
+    except Exception:
+        logger.warning("Failed to inspect next allowed sync time", exc_info=True)
+        return 0
+    elapsed = int(time.time()) - completed_at
+    return max(0, min_interval - elapsed)
+
+
 def _bump_bootstrap_failure_streak(
     redis_client: Redis,
     *,
@@ -1032,6 +1064,15 @@ def _build_listing_segments(settings: Settings) -> list[dict[str, str | int | No
     При IAAI_LISTING_SEGMENTS=runtime сегменты строятся из filters.brands / filters.include.brands.
     Остальные значения IAAI_LISTING_SEGMENTS сохраняют прежнее поведение: auto или JSON.
     """
+    filtered_urls = settings.listing.filtered_search_urls
+    if len(filtered_urls) > 1:
+        return [
+            {"make": None, "year_min": None, "year_max": None, "listing_url": url}
+            for url in filtered_urls
+        ]
+    if len(filtered_urls) == 1:
+        return []
+
     raw_segments = settings.listing.listing_segments_json.strip()
     if raw_segments.casefold() != "runtime":
         return parse_listing_segments(raw_segments)
@@ -1042,10 +1083,17 @@ def _build_listing_segments(settings: Settings) -> list[dict[str, str | int | No
 
     runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
     brands = runtime_config.filters.include.brands
-    if settings.scraping_profile.http_first and settings.listing.fast_segment_year_splits:
+
+    # Fast HTTP-first режим не использует UI year-фильтры IAAI.
+    # Значит runtime-сегменты должны быть "один бренд = один сегмент"
+    # без доп. year split, иначе получаем много лишних долгих сегментов,
+    # которые fast-профиль всё равно игнорирует.
+    if settings.scraping_profile.http_first:
         segments = build_fast_listing_segments_for_makes(list(brands))
-    else:
+    elif settings.listing.fast_segment_year_splits:
         segments = build_listing_segments_for_makes(list(brands))
+    else:
+        segments = build_fast_listing_segments_for_makes(list(brands))
     if not segments:
         logger.warning(
             "IAAI_LISTING_SEGMENTS=runtime, but runtime_config filters.brands is empty; segmented listing disabled"
@@ -1088,7 +1136,10 @@ def sync_segment_task(
     seg_make = segment.get("make")
     seg_year_min = segment.get("year_min")
     seg_year_max = segment.get("year_max")
+    seg_listing_url = segment.get("listing_url")
     seg_label = f"{seg_make or 'ALL'}"
+    if seg_listing_url:
+        seg_label = "FILTERED_SEARCH"
     if seg_year_min is not None or seg_year_max is not None:
         seg_label += f" ({seg_year_min}-{seg_year_max})"
 
@@ -1135,6 +1186,7 @@ def sync_segment_task(
                     model=None,
                     lane=lane,
                     only_new=only_new,
+                    listing_url=str(seg_listing_url) if seg_listing_url else None,
                     year_min=seg_year_min,
                     year_max=seg_year_max,
                     skip_mark_sold=True,
@@ -1286,6 +1338,12 @@ def sync_listing_task(
     watchdog_stop: Event | None = None
     watchdog_thread: Thread | None = None
     force_bootstrap_full_scan = False
+    settings = Settings()
+    followup_min_delay_seconds = max(
+        0,
+        int(os.getenv("IAAI_SYNC_FOLLOWUP_MIN_DELAY_SECONDS", str(settings.celery.beat_sync_interval_minutes * 60))),
+    )
+    is_beat_task = bool(self.request.headers and self.request.headers.get("iaai_beat_task"))
 
     def _enqueue_bootstrap_followup(
         reason: str,
@@ -1293,6 +1351,7 @@ def sync_listing_task(
         *,
         count_as_failure: bool = False,
     ) -> None:
+        delay_seconds = max(int(delay_seconds), followup_min_delay_seconds)
         flag_ttl = max(lock_ttl, delay_seconds + 300)
         if not _try_set_followup_pending(redis_client, ttl_seconds=flag_ttl):
             logger.info(
@@ -1367,7 +1426,18 @@ def sync_listing_task(
         }
 
     try:
-        settings = Settings()
+        next_allowed_delay = _seconds_until_next_allowed_sync(redis_client, settings, is_beat_task=is_beat_task)
+        if next_allowed_delay > 0:
+            logger.info(
+                "sync_listing_task skipped: previous full run finished recently; next run allowed in %ss",
+                next_allowed_delay,
+            )
+            return {
+                "status": "skipped",
+                "reason": "next_sync_not_due_yet",
+                "retry_after_seconds": next_allowed_delay,
+                "task_id": task_id,
+            }
 
         # Любой реально стартовавший sync_listing снимает pending-флаг followup,
         # чтобы watchdog/continuation могли корректно планировать следующий run
@@ -1422,6 +1492,8 @@ def sync_listing_task(
             and not always_full_scan
         )
         # Определяем сегменты из конфига/env или из runtime_config при IAAI_LISTING_SEGMENTS=runtime.
+        filtered_listing_urls = settings.listing.filtered_search_urls
+        filtered_listing_url = filtered_listing_urls[0] if filtered_listing_urls else None
         segments = _build_listing_segments(settings)
 
         watchdog_stop, watchdog_thread = _start_stall_watchdog(
@@ -1716,6 +1788,7 @@ def sync_listing_task(
                     lane=lane,
                     limit=effective_limit,
                     only_new=effective_only_new,
+                    listing_url=filtered_listing_url,
                 )
 
         result = _run_browser_job(_job)
@@ -1805,6 +1878,7 @@ def sync_listing_task(
             summary["cars_failed"],
             summary["failures_count"],
         )
+        _mark_sync_completed(redis_client)
         return summary
 
     except SoftTimeLimitExceeded:

runtime_config.json

@@ -1,6 +1,6 @@
 {
   "sync": {
-    "name": null,
+    "name": "https://www.iaai.com/Search?url=B%2bU066dM8%2flZtRvnzTwIEi8Pib9%2fM2fLpCTQDIgQRm4%3d",
     "ids_initial_size": null,
     "ids_next_size": null,
     "ids_max_pages": null,

status_vps.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import os
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+
+commands = [
+    "cd /root/iaai-parser && docker compose ps",
+    "cd /root/iaai-parser && docker compose exec -T worker celery -A iaai_scraper.worker.celery_app inspect active reserved scheduled",
+    "docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c \"SELECT now() AS ts, (SELECT count(*) FROM iaai_cars) AS cars, (SELECT count(*) FROM iaai_sync_runs) AS runs, (SELECT status FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_status, (SELECT cars_upserted FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_upserted, (SELECT cars_failed FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_failed, (SELECT started_at FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_started;\"",
+    "docker logs --since 10m iaai-parser-worker-1 2>&1 | tail -n 160",
+]
+
+client = paramiko.SSHClient()
+client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+client.connect(
+    HOST,
+    username=USER,
+    password=PASSWORD,
+    look_for_keys=False,
+    allow_agent=False,
+    timeout=30,
+    banner_timeout=30,
+    auth_timeout=30,
+)
+try:
+    for command in commands:
+        print(f"\n=== REMOTE_RUN {command} ===", flush=True)
+        stdin, stdout, stderr = client.exec_command(command, timeout=180)
+        exit_code = stdout.channel.recv_exit_status()
+        print(stdout.read().decode("utf-8", "replace"), end="")
+        print(stderr.read().decode("utf-8", "replace"), end="")
+        print(f"\n=== EXIT {exit_code} ===", flush=True)
+finally:
+    client.close()

tests/test_self_heal.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -50,6 +51,119 @@ class TestSelfHeal(unittest.TestCase):
 
 		self.assertIsNone(ts)
 
+	def test_active_recent_progress_prevents_db_idle_restart(self) -> None:
+		redis_client = MagicMock()
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:hourly"]
+		redis_client.get.return_value = json.dumps(
+			{
+				"task_id": "hourly",
+				"stage": "fast_listing_collected",
+				"ts": 1000,
+				"last_db_progress_ts": 1,
+				"skipped_existing": 2417,
+			}
+		)
+
+		active = self_heal._has_active_recent_progress(
+			redis_client,
+			now_ts=1010,
+			stall_seconds=900,
+		)
+
+		self.assertTrue(active)
+
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=5000)
+	def test_has_inflight_work_deletes_stale_terminal_progress_without_work(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: None,
+			"iaai:state:task_progress:old": json.dumps(
+				{
+					"task_id": "old",
+					"stage": "segment_done",
+					"ts": 1000,
+					"segment_full_scan_completed": True,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:old"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertFalse(has_inflight)
+		self.assertEqual(flags["has_task_progress"], 0)
+		redis_client.delete.assert_called_once_with("iaai:state:task_progress:old")
+
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=1010)
+	def test_has_inflight_work_keeps_recent_non_terminal_progress(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: None,
+			"iaai:state:task_progress:active": json.dumps(
+				{
+					"task_id": "active",
+					"stage": "fast_listing_collected",
+					"ts": 1000,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:active"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertTrue(has_inflight)
+		self.assertEqual(flags["has_task_progress"], 1)
+		redis_client.delete.assert_not_called()
+
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=5000)
+	def test_has_inflight_work_deletes_stale_non_terminal_progress_without_work(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: None,
+			"iaai:state:task_progress:stale": json.dumps(
+				{
+					"task_id": "stale",
+					"stage": "segment_started",
+					"ts": 1000,
+					"segment_index": 11,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:stale"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertFalse(has_inflight)
+		self.assertEqual(flags["has_task_progress"], 0)
+		self.assertEqual(flags["stale_progress_deleted"], 1)
+		redis_client.delete.assert_called_once_with("iaai:state:task_progress:stale")
+
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=5000)
+	def test_has_inflight_work_keeps_stale_progress_when_lock_exists(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: "owner",
+			"iaai:state:task_progress:stale": json.dumps(
+				{
+					"task_id": "stale",
+					"stage": "segment_started",
+					"ts": 1000,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:stale"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertTrue(has_inflight)
+		self.assertEqual(flags["has_lock"], 1)
+		self.assertEqual(flags["has_task_progress"], 1)
+		redis_client.delete.assert_not_called()
+
 	@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
 	@patch("iaai_scraper.worker.self_heal.os.kill")
 	@patch("builtins.open")

tests/test_worker_tasks.py

@@ -19,6 +19,8 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
 
         settings = Settings(runtime_config_file=runtime_config_file)
         settings.listing.listing_segments_json = "runtime"
+        settings.scraping_profile.http_first = False
+        settings.listing.fast_segment_year_splits = True
 
         segs = tasks._build_listing_segments(settings)
 

vps_check.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd /root/iaai-parser
+
+echo '--- services ---'
+docker compose ps
+
+echo '--- env speed ---'
+grep -nE 'IAAI_PROFILE|IAAI_SCRAPING_PROFILE|IAAI_HTTP_FIRST|IAAI_LISTING_SEGMENTS|IAAI_FAST_SEGMENT_YEAR_SPLITS|CELERY_WORKER_CONCURRENCY|IAAI_FETCH_CONCURRENCY|CELERY_BATCH_SIZE' .env
+
+echo '--- db count ---'
+docker compose exec -T postgres psql -U iaai -d iaai_scraper -t -c "select count(*) as cars_count, max(last_seen_at) as last_seen from iaai_cars;"
+
+echo '--- redis progress ---'
+docker compose exec -T redis redis-cli MGET iaai:state:sync_listing_checkpoint iaai:state:sync_segments_done iaai:state:sync_segments_total iaai:state:sync_full_scan_done
+
+echo '--- recent worker progress ---'
+docker compose logs --tail=80 worker | grep -E 'Segment [0-9]+/[0-9]+|Fast HTTP-first listing started|Fast HTTP-first listing collected|Fast HTTP-first DB batch|ERROR|WARNING' | tail -40