Add tools

.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,38 +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}
-  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-1200}
-  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-2400}
+  # Для больших Search-выборок (десятки тысяч лотов) run должен жить дольше одного батча.
+  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-7200}
+  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_REQUEST_JITTER_MAX_S: ${IAAI_REQUEST_JITTER_MAX_S:-0}
-  IAAI_DETAIL_RETRIES: ${IAAI_DETAIL_RETRIES:-1}
+  IAAI_LISTING_POST_ATTEMPTS: ${IAAI_LISTING_POST_ATTEMPTS:-1}
+  IAAI_REQUEST_JITTER_MAX_S: ${IAAI_REQUEST_JITTER_MAX_S:-0.03}
+  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_FILTERED_SEARCH_URL: ${IAAI_FILTERED_SEARCH_URL:-}
   IAAI_FILTERED_SEARCH_URLS: ${IAAI_FILTERED_SEARCH_URLS:-}
-  IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-runtime}
+  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_HUMAN_PACE_ENABLED: ${IAAI_HUMAN_PACE_ENABLED:-true}
-  IAAI_TOKENS_FILE: ${IAAI_TOKENS_FILE:-/home/app/tokens.json}
+  IAAI_ALWAYS_FULL_SCAN: ${IAAI_ALWAYS_FULL_SCAN:-false}
+  IAAI_HUMAN_PACE_ENABLED: ${IAAI_HUMAN_PACE_ENABLED:-false}
+  # Первый 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.
@@ -168,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

@@ -519,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

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 корректно его подхватывают.
-    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stderr)]
+    # stdout — основной поток для `docker logs`, Docker Desktop и compose logs.
+    # 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

@@ -183,6 +183,7 @@ 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 = {
@@ -232,13 +233,26 @@ 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)
-                    if self.client._settings.scraping_profile.verbose_progress_logs:
+                    verbose_progress_logs = bool(
+                        getattr(
+                            getattr(getattr(self.client, "_settings", None), "scraping_profile", None),
+                            "verbose_progress_logs",
+                            False,
+                        )
+                    )
+                    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,

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,

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:
@@ -148,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:*"):
-		has_task_progress = 1
-		break
+	progress_keys_seen = 0
+	stale_progress_deleted = 0
+	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
 

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"
@@ -867,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,
     *,
@@ -1310,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,
@@ -1317,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(
@@ -1391,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
@@ -1832,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

@@ -72,6 +72,98 @@ class TestSelfHeal(unittest.TestCase):
 
 		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")