add docker-compose.vps.yml overlay for low-RAM VPS

.env.example

@@ -55,6 +55,3 @@ CELERY_BROKER_VISIBILITY_TIMEOUT=90000
 CELERY_WORKER_CONCURRENCY=1
 CELERY_BEAT_SYNC_INTERVAL_MINUTES=60
 CELERY_BEAT_SYNC_LIMIT=0
-IAAI_ALWAYS_FULL_SCAN=true
-SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_LIMIT=6
-SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_TTL_SECONDS=21600

Dockerfile

@@ -13,8 +13,8 @@ RUN pip install --no-cache-dir uv \
     && pip install --no-cache-dir -r /tmp/requirements.txt \
     && pip install --no-cache-dir playwright-stealth
 
-# Ставим Chromium и Firefox.
-# По умолчанию используем Chromium.
+# Install both Chromium and Firefox. Chromium is the default engine in Docker
+# because it works more reliably with the current IAAI listing page.
 RUN python -m playwright install chromium firefox
 
 COPY . .
@@ -27,6 +27,6 @@ STOPSIGNAL SIGINT
 
 USER app
 
-# По умолчанию запускаем API.
+# По умолчанию  API, но worker/beat переопределяют CMD в docker-compose
 ENTRYPOINT ["./entrypoint.sh"]
 CMD ["uvicorn", "iaai_scraper.api.app:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -124,10 +124,9 @@ docker compose ps
 - `worker` имеет healthcheck через `celery inspect ping`
 - `beat` имеет healthcheck по файлу `celerybeat-schedule`
 
-
 ## API
 
-**Статистика:**
+**Здоровье и статистика:**
 - `GET /health` — статус сервиса и подключения к БД
 - `GET /api/v1/stats` — сколько машин и картинок в базе, топ брендов
 
@@ -243,6 +242,11 @@ pytest -q
 
 ## `pyproject.toml` + `uv.lock`
 
+Проект переведён на современную схему зависимостей:
+
+- `pyproject.toml` — декларация зависимостей и метаданных проекта
+- `uv.lock` — зафиксированные версии для воспроизводимых установок
+
 Локально можно использовать:
 
 ```bash
@@ -250,6 +254,10 @@ uv sync
 uv run pytest -q
 ```
 
+## Runtime-фильтры
+
+Файл `runtime_config.json` управляет runtime-поведением sync и фильтрацией автомобилей.
+
 ### Секция `sync`
 
 Поддерживаются поля:

alembic/env.py

@@ -14,7 +14,6 @@ from iaai_scraper.core.config import Settings
 from iaai_scraper.storage.models import Base
 
 config = context.config
-VERSION_TABLE = "iaai_parser_alembic_version"
 
 # Logging
 if config.config_file_name is not None:
@@ -33,7 +32,6 @@ def run_migrations_offline() -> None:
     context.configure(
         url=url,
         target_metadata=target_metadata,
-        version_table=VERSION_TABLE,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
     )
@@ -49,11 +47,7 @@ def run_migrations_online() -> None:
         poolclass=pool.NullPool,
     )
     with connectable.connect() as connection:
-        context.configure(
-            connection=connection,
-            target_metadata=target_metadata,
-            version_table=VERSION_TABLE,
-        )
+        context.configure(connection=connection, target_metadata=target_metadata)
         with context.begin_transaction():
             context.run_migrations()
 

alembic/versions/001_initial.py

@@ -1,4 +1,4 @@
-# Начальная схема: iaai_cars, iaai_images, iaai_sync_runs
+# Начальная схема: cars, images, sync_runs
 from typing import Sequence, Union
 
 from alembic import op
@@ -10,29 +10,10 @@ branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
-def _schema_name() -> str | None:
-    bind = op.get_bind()
-    return "public" if bind.dialect.name == "postgresql" else None
-
-
-def _table_exists(table_name: str) -> bool:
-    inspector = sa.inspect(op.get_bind())
-    return table_name in inspector.get_table_names(schema=_schema_name())
-
-
-def _index_exists(table_name: str, index_name: str) -> bool:
-    if not _table_exists(table_name):
-        return False
-    inspector = sa.inspect(op.get_bind())
-    indexes = inspector.get_indexes(table_name, schema=_schema_name())
-    return any(index.get("name") == index_name for index in indexes)
-
-
 def upgrade() -> None:
-    # Таблица iaai_cars — аукционные машины с IAAI
-    if not _table_exists("iaai_cars"):
+    # Таблица cars — аукционные машины с IAAI
     op.create_table(
-            "iaai_cars",
+        "cars",
         sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
         sa.Column("parser_id", sa.String(50), nullable=False, unique=True),
         sa.Column("brand", sa.String(50), nullable=False),
@@ -64,27 +45,22 @@ def upgrade() -> None:
         sa.Column("slug", sa.String, nullable=False),
         sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
     )
+    op.create_index("ix_cars_origin_url", "cars", ["origin_url"])
+    op.create_index("ix_cars_origin_id", "cars", ["origin_id"], unique=True)
 
-    if not _index_exists("iaai_cars", "ix_iaai_cars_origin_url"):
-        op.create_index("ix_iaai_cars_origin_url", "iaai_cars", ["origin_url"])
-    if not _index_exists("iaai_cars", "ix_iaai_cars_origin_id"):
-        op.create_index("ix_iaai_cars_origin_id", "iaai_cars", ["origin_id"], unique=True)
-
-    # Таблица iaai_images — изображения машин
-    if not _table_exists("iaai_images"):
+    # Таблица images — изображения машин
     op.create_table(
-            "iaai_images",
+        "images",
         sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
         sa.Column("fullres_image", sa.String, nullable=False),
         sa.Column("preview_image", sa.String, nullable=False),
         sa.Column("order_index", sa.Integer, nullable=False),
-            sa.Column("car_id", sa.Integer, sa.ForeignKey("iaai_cars.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("car_id", sa.Integer, sa.ForeignKey("cars.id", ondelete="CASCADE"), nullable=False),
     )
 
-    # Таблица iaai_sync_runs — логирование синхронизаций
-    if not _table_exists("iaai_sync_runs"):
+    # Таблица sync_runs — логирование синхронизаций
     op.create_table(
-            "iaai_sync_runs",
+        "sync_runs",
         sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
         sa.Column("started_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
         sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
@@ -97,7 +73,7 @@ def upgrade() -> None:
         sa.Column("error_summary", sa.Text, nullable=True),
     )
 
-
 def downgrade() -> None:
-    # Compatibility-only migration for shared production databases.
-    pass
+    op.drop_table("sync_runs")
+    op.drop_table("images")
+    op.drop_table("cars")

alembic/versions/002_add_indexes.py

@@ -10,15 +10,20 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_brand ON iaai_cars (brand)")
-    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_brand_model ON iaai_cars (brand, model)")
-    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_year ON iaai_cars (year)")
-    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_is_sold ON iaai_cars (is_sold)")
-    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_last_seen_at ON iaai_cars (last_seen_at)")
-    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_images_car_id ON iaai_images (car_id)")
-    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_sync_runs_status ON iaai_sync_runs (status)")
+    op.create_index("ix_cars_brand", "cars", ["brand"])
+    op.create_index("ix_cars_brand_model", "cars", ["brand", "model"])
+    op.create_index("ix_cars_year", "cars", ["year"])
+    op.create_index("ix_cars_is_sold", "cars", ["is_sold"])
+    op.create_index("ix_cars_last_seen_at", "cars", ["last_seen_at"])
+    op.create_index("ix_images_car_id", "images", ["car_id"])
+    op.create_index("ix_sync_runs_status", "sync_runs", ["status"])
 
 
 def downgrade() -> None:
-    # Compatibility-only migration for shared production databases.
-    pass
+    op.drop_index("ix_sync_runs_status", table_name="sync_runs")
+    op.drop_index("ix_images_car_id", table_name="images")
+    op.drop_index("ix_cars_last_seen_at", table_name="cars")
+    op.drop_index("ix_cars_is_sold", table_name="cars")
+    op.drop_index("ix_cars_year", table_name="cars")
+    op.drop_index("ix_cars_brand_model", table_name="cars")
+    op.drop_index("ix_cars_brand", table_name="cars")

alembic/versions/003_add_composite_indexes.py

@@ -13,24 +13,26 @@ def upgrade() -> None:
     # Partial index для активных машин IAAI (is_sold = FALSE)
     # Ускоряет поиск при mark_sold операциях
     op.execute(
-        "CREATE INDEX IF NOT EXISTS ix_iaai_cars_active_iaai "
-        "ON iaai_cars (origin_url) "
+        "CREATE INDEX IF NOT EXISTS ix_cars_active_iaai "
+        "ON cars (origin_url) "
         "WHERE is_sold = FALSE AND origin_id LIKE 'iaai:%'"
     )
 
     # Composite index для проверки дубликатов изображений
-    op.execute(
-        "CREATE INDEX IF NOT EXISTS ix_iaai_images_car_id_fullres "
-        "ON iaai_images (car_id, fullres_image)"
+    op.create_index(
+        "ix_images_car_id_fullres",
+        "images",
+        ["car_id", "fullres_image"],
     )
 
     # Index для быстрого поиска existing машин по origin_url
     op.execute(
-        "CREATE INDEX IF NOT EXISTS ix_iaai_cars_origin_url_id "
-        "ON iaai_cars (origin_url, origin_id)"
+        "CREATE INDEX IF NOT EXISTS ix_cars_origin_url_id "
+        "ON cars (origin_url, origin_id)"
     )
 
 
 def downgrade() -> None:
-    # Compatibility-only migration for shared production databases.
-    pass
+    op.execute("DROP INDEX IF EXISTS ix_cars_origin_url_id")
+    op.drop_index("ix_images_car_id_fullres", table_name="images")
+    op.execute("DROP INDEX IF EXISTS ix_cars_active_iaai")

alembic/versions/004_add_ingestion_tables.py

@@ -1,17 +0,0 @@
-# Placeholder migration (ingestion tables not yet needed)
-from typing import Sequence, Union
-
-from alembic import op
-
-revision: str = "004_add_ingestion_tables"
-down_revision: Union[str, None] = "003_add_composite_indexes"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    pass
-
-
-def downgrade() -> None:
-    pass

docker-compose.vps.yml

@@ -0,0 +1,76 @@
+# Overlay для слабого VPS (4 vCPU, 4 GB RAM).
+# Применять как: docker compose -f docker-compose.yml -f docker-compose.vps.yml up -d --build
+#
+# Главное:
+#  - worker concurrency=1, max-tasks-per-child=3 (Chromium быстро течёт)
+#  - IAAI_PARALLEL_TABS=4 (важно: каждая вкладка ~80-150 MB)
+#  - IAAI_MAX_PAGES_PER_RUN=200, IAAI_MAX_VEHICLES_PER_RUN=2000 — короткие тестовые прогоны
+#  - hard memory limits на каждый сервис
+
+x-vps-env: &vps-env
+  CELERY_WORKER_CONCURRENCY: "1"
+  CELERY_WORKER_MAX_TASKS_PER_CHILD: "3"
+  CELERY_WORKER_MAX_MEMORY_PER_CHILD_KB: "350000"   # ~350 MB → перезапуск процесса
+  IAAI_PARALLEL_TABS: "4"
+  IAAI_BLOCK_RESOURCES: "true"
+  IAAI_MAX_PAGES_PER_RUN: "200"
+  IAAI_MAX_VEHICLES_PER_RUN: "2000"
+  IAAI_LISTING_SEGMENTS: "auto"
+  IAAI_HUMAN_PACE_ENABLED: "true"
+  CELERY_BATCH_SIZE: "100"
+  CELERY_TASK_SOFT_TIME_LIMIT: "1500"   # 25 мин — короче бутстрап-сегмент
+  CELERY_TASK_TIME_LIMIT: "1800"        # 30 мин
+  CELERY_BROKER_VISIBILITY_TIMEOUT: "3600"
+  IAAI_DATABASE_POOL_SIZE: "3"
+  IAAI_DATABASE_MAX_OVERFLOW: "2"
+
+services:
+  postgres:
+    mem_limit: 512m
+    mem_reservation: 256m
+    command:
+      - "postgres"
+      - "-c"
+      - "shared_buffers=128MB"
+      - "-c"
+      - "effective_cache_size=384MB"
+      - "-c"
+      - "work_mem=8MB"
+      - "-c"
+      - "maintenance_work_mem=64MB"
+      - "-c"
+      - "max_connections=50"
+
+  redis:
+    mem_limit: 192m
+    mem_reservation: 64m
+    command: >
+      redis-server
+      --appendonly yes
+      --save 60 1000
+      --maxmemory 128mb
+      --maxmemory-policy allkeys-lru
+
+  api:
+    mem_limit: 384m
+    mem_reservation: 128m
+    environment:
+      <<: *vps-env
+
+  worker:
+    mem_limit: 1800m
+    mem_reservation: 512m
+    environment:
+      <<: *vps-env
+    # Переопределяем, чтобы concurrency=1 и max-tasks-per-child=3 точно применились
+    command: >
+      celery -A iaai_scraper.worker.celery_app worker
+      --loglevel=info --concurrency=1 --pool=prefork
+      --pidfile=/tmp/celery-worker.pid
+      -Q scraping --max-tasks-per-child=3
+
+  beat:
+    mem_limit: 256m
+    mem_reservation: 64m
+    environment:
+      <<: *vps-env

docker-compose.yml

@@ -1,35 +1,27 @@
 x-app-env: &app-env
-  # Всегда используем Postgres.
+  # NB: hardcoded to Postgres — .env may contain sqlite for local CLI, don't let it leak here
   IAAI_DATABASE_URL: postgresql+psycopg2://iaai:iaai@postgres:5432/iaai_scraper
   IAAI_REDIS_URL: ${IAAI_REDIS_URL:-redis://redis:6379/0}
   CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis:6379/0}
   CELERY_RESULT_BACKEND: ${CELERY_RESULT_BACKEND:-redis://redis:6379/0}
   IAAI_DATABASE_POOL_RECYCLE_SECONDS: ${IAAI_DATABASE_POOL_RECYCLE_SECONDS:-1800}
-  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}
-  # 0 = без лимита.
+  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-3300}
+  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-3600}
+  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-7200}
+  # 0 => без лимита (в коде интерпретируется как None).
+  # После первичного полного прохода hourly-режим должен успевать за всеми новыми авто.
   CELERY_BEAT_SYNC_LIMIT: ${CELERY_BEAT_SYNC_LIMIT:-0}
   CELERY_WORKER_MAX_TASKS_PER_CHILD: ${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
   CELERY_BATCH_SIZE: ${CELERY_BATCH_SIZE:-200}
-  IAAI_INTER_BATCH_DELAY_SECONDS: ${IAAI_INTER_BATCH_DELAY_SECONDS:-0.3}
-  IAAI_FAIL_RATE_THRESHOLD: ${IAAI_FAIL_RATE_THRESHOLD:-0.9}
-  IAAI_PARALLEL_TABS: ${IAAI_PARALLEL_TABS:-8}
+  IAAI_PARALLEL_TABS: ${IAAI_PARALLEL_TABS:-40}
   IAAI_BLOCK_RESOURCES: ${IAAI_BLOCK_RESOURCES:-true}
   IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-auto}
   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:-/data/tokens.json}
   IAAI_RUNTIME_CONFIG_FILE: ${IAAI_RUNTIME_CONFIG_FILE:-/app/runtime_config.json}
   IAAI_BROWSER_ENGINE: chromium
-  # Self-heal для worker.
-  IAAI_SELF_HEAL_ENABLED: ${IAAI_SELF_HEAL_ENABLED:-true}
-  IAAI_SELF_HEAL_CHECK_INTERVAL_SECONDS: ${IAAI_SELF_HEAL_CHECK_INTERVAL_SECONDS:-30}
-  IAAI_SELF_HEAL_STALL_SECONDS: ${IAAI_SELF_HEAL_STALL_SECONDS:-900}
-  IAAI_SELF_HEAL_STARTUP_GRACE_SECONDS: ${IAAI_SELF_HEAL_STARTUP_GRACE_SECONDS:-300}
-  IAAI_SELF_HEAL_RESTART_COOLDOWN_SECONDS: ${IAAI_SELF_HEAL_RESTART_COOLDOWN_SECONDS:-300}
   TZ: ${TZ:-UTC}
 
 x-env-file: &env-file
@@ -48,9 +40,10 @@ x-worker-service: &worker-service
   volumes:
     - ./runtime_config.json:/app/runtime_config.json:ro
     - tokens_data:/data
+    - beat_data:/var/lib/celery
 
 services:
-  # PostgreSQL.
+  # PostgreSQL
   postgres:
     image: postgres:16-alpine
     container_name: iaai-postgres
@@ -74,7 +67,7 @@ services:
         max-size: "10m"
         max-file: "5"
 
-  # Redis.
+  # Redis (Celery broker)
   redis:
     image: redis:7-alpine
     container_name: iaai-redis
@@ -98,7 +91,7 @@ services:
         max-size: "10m"
         max-file: "5"
 
-  # Миграции.
+  # DB migrations
   migrate:
     <<: *app-service
     container_name: iaai-migrate
@@ -108,7 +101,7 @@ services:
         condition: service_healthy
     command: alembic upgrade head
 
-  # API.
+  # FastAPI
   api:
     <<: *app-service
     container_name: iaai-api
@@ -136,9 +129,10 @@ services:
         max-size: "10m"
         max-file: "5"
 
-  # Worker.
+  # Celery Worker
   worker:
     <<: *worker-service
+    container_name: iaai-worker
     restart: unless-stopped
     init: true
     depends_on:
@@ -149,23 +143,22 @@ services:
     stop_grace_period: 60s
     command: >
       celery -A iaai_scraper.worker.celery_app worker
-      --loglevel=info --concurrency=1 --pool=solo
+      --loglevel=info --concurrency=${CELERY_WORKER_CONCURRENCY:-4} --pool=prefork
       --pidfile=/tmp/celery-worker.pid
       -Q scraping --max-tasks-per-child=${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
     healthcheck:
-      # Проверка worker.
-      test: ["CMD-SHELL", "test -f /tmp/celery-worker.pid && kill -0 $(cat /tmp/celery-worker.pid) && pgrep -f 'iaai_scraper.worker.self_heal' >/dev/null"]
+      test: ["CMD-SHELL", "celery -A iaai_scraper.worker.celery_app inspect ping -d celery@$$HOSTNAME -t 15 >/dev/null 2>&1 || exit 1"]
       interval: 60s
-      timeout: 10s
+      timeout: 20s
       retries: 3
-      start_period: 90s
+      start_period: 120s
     logging:
       driver: json-file
       options:
         max-size: "10m"
         max-file: "5"
 
-  # Beat.
+  # Celery Beat (периодический планировщик)
   beat:
     <<: *worker-service
     container_name: iaai-beat
@@ -180,7 +173,7 @@ services:
       celery -A iaai_scraper.worker.celery_app beat
       --loglevel=info
       --pidfile=/tmp/celery-beat.pid
-      --schedule=/tmp/celerybeat-schedule
+      --schedule=/var/lib/celery/celerybeat-schedule
     healthcheck:
       test: ["CMD-SHELL", "test -f /tmp/celery-beat.pid && kill -0 $(cat /tmp/celery-beat.pid)"]
       interval: 60s
@@ -197,3 +190,4 @@ volumes:
   pgdata:
   redisdata:
   tokens_data:
+  beat_data:

entrypoint.sh

@@ -43,47 +43,39 @@ start_proxy_bridge_if_needed() {
   fi
 
   echo "[entrypoint] Using browser proxy: ${IAAI_PROXY_SERVER}"
-  python -m iaai_scraper.proxy_bridge &
-  BRIDGE_PID=$!
+
+  # Watchdog: автоматически рестартует bridge при падении.
+  # Нужно для долгоиграющих контейнеров (месяцы аптайма): без watchdog
+  # упавший bridge тихо ломает весь скрапинг и IAAI начинает банить реальный IP VPS.
+  (
+    while true; do
+      python -m iaai_scraper.proxy_bridge
+      EXIT_CODE=$?
+      echo "[entrypoint] proxy_bridge exited with code ${EXIT_CODE}; restarting in 3s..."
+      sleep 3
+    done
+  ) &
+  BRIDGE_WATCHDOG_PID=$!
   sleep 1
 
-  if ! kill -0 "${BRIDGE_PID}" 2>/dev/null; then
-    echo "[entrypoint] ERROR: iaai_scraper.proxy_bridge failed to start"
+  if ! kill -0 "${BRIDGE_WATCHDOG_PID}" 2>/dev/null; then
+    echo "[entrypoint] ERROR: proxy_bridge watchdog failed to start"
     exit 1
   fi
 
-  echo "[entrypoint] Proxy bridge started (PID ${BRIDGE_PID})"
-}
-
-start_self_heal_watchdog_if_worker() {
-  if ! is_worker_command "$@"; then
+  # Проверяем что bridge действительно слушает порт.
+  for i in 1 2 3 4 5; do
+    if python -c "import socket,sys; s=socket.socket(); s.settimeout(2); sys.exit(0 if s.connect_ex(('127.0.0.1', 8899))==0 else 1)" 2>/dev/null; then
+      echo "[entrypoint] Proxy bridge listening on :8899 (watchdog PID ${BRIDGE_WATCHDOG_PID})"
       return 0
     fi
-
-  # После reboot/restart контейнера файловая система контейнера сохраняется,
-  # поэтому stale pidfile может остаться от прошлого запуска и блокировать старт Celery.
-  # Удаляем его перед запуском worker-процесса.
-  rm -f /tmp/celery-worker.pid
-
-  if [ "${IAAI_SELF_HEAL_ENABLED:-true}" = "false" ]; then
-    echo "[entrypoint] Self-heal watchdog disabled"
-    return 0
-  fi
-
-  echo "[entrypoint] Starting self-heal watchdog for worker..."
-  python -m iaai_scraper.worker.self_heal &
-  SELF_HEAL_PID=$!
     sleep 1
+  done
 
-  if ! kill -0 "${SELF_HEAL_PID}" 2>/dev/null; then
-    echo "[entrypoint] ERROR: self-heal watchdog failed to start"
+  echo "[entrypoint] ERROR: proxy_bridge did not start listening on :8899"
   exit 1
-  fi
-
-  echo "[entrypoint] Self-heal watchdog started (PID ${SELF_HEAL_PID})"
 }
 
 start_proxy_bridge_if_needed "$@"
-start_self_heal_watchdog_if_worker "$@"
 
 exec "$@"

iaai_scraper/browser/factory.py

@@ -15,7 +15,7 @@ logger = logging.getLogger("iaai_scraper.browser")
 
 
 def _build_init_script() -> str:
-    # Маскировка браузера.
+    # Патч признаков автоматизации.
     hardware_concurrency = random.choice([4, 8, 12, 16])
     device_memory = random.choice([4, 8, 16])
     languages = ["en-US", "en"]
@@ -69,10 +69,11 @@ class BrowserFactory:
         self.settings = settings
 
     def _resolve_engine(self) -> str:
-        # Выбор движка.
+        # Выбор движка браузера.
         engine = self.settings.browser_engine.strip().lower()
         if engine == "auto":
-            # Для IAAI стабильнее Chromium.
+            # Для IAAI в headless-режиме Chromium со stealth-скриптами
+            # значительно стабильнее Firefox по anti-bot.
             return "chromium"
         if engine in ("firefox", "chromium"):
             return engine
@@ -89,11 +90,11 @@ class BrowserFactory:
             if proxy_dict:
                 launch_kwargs["proxy"] = proxy_dict
                 logger.info("Using proxy: %s", self.settings.proxy.server)
-            # Настройки Firefox.
+            # Параметры Firefox для headless-режима.
             launch_kwargs["firefox_user_prefs"] = {
                 "dom.webdriver.enabled": False,
                 "useAutomationExtension": False,
-                # Базовые оптимизации.
+                # Базовые оптимизации для VPS.
                 "media.autoplay.default": 5,
                 "media.volume_scale": "0.0",
                 "media.audio.playback.standalone": False,
@@ -110,7 +111,7 @@ class BrowserFactory:
             logger.info("Launching Firefox (headless=%s)", self.settings.headless)
             return playwright.firefox.launch(**launch_kwargs)
 
-        # Запуск Chromium.
+        # Путь запуска Chromium.
         args = [
             "--disable-blink-features=AutomationControlled",
             "--no-default-browser-check",
@@ -153,7 +154,7 @@ class BrowserFactory:
         }
 
         if is_firefox:
-            # Заголовки Firefox.
+            # Заголовки и user-agent для Firefox.
             ctx_kwargs["user_agent"] = (
                 "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) "
                 "Gecko/20100101 Firefox/128.0"
@@ -182,7 +183,7 @@ class BrowserFactory:
         context.set_default_navigation_timeout(self.settings.default_timeout_ms)
 
         if not is_firefox:
-            # Маскировка Chromium.
+            # Патчи маскировки для Chromium.
             context.add_init_script(_build_init_script())
             if stealth_sync:
                 context.on("page", lambda page: stealth_sync(page))
@@ -192,7 +193,7 @@ class BrowserFactory:
 
     @staticmethod
     def enable_resource_blocking(page) -> None:
-        # Блокируем тяжёлые ресурсы.
+        # Блокируем тяжёлые ресурсы для ускорения загрузки страниц.
         BLOCKED_TYPES = {"image", "stylesheet", "font", "media"}
         BLOCKED_URL_PATTERNS = (
             "google-analytics", "googletagmanager", "facebook.net",

iaai_scraper/browser/listing.py

@@ -133,28 +133,6 @@ class ListingCollector:
 
 		return not old_first_href
 
-	@staticmethod
-	def has_page_number(page: Page, target_page_number: int) -> bool:
-		try:
-			return bool(page.evaluate(
-				"""
-				(targetPageNumber) => {
-					const visible = (el) => !!(el && (el.offsetWidth || el.offsetHeight || el.getClientRects().length));
-					const controls = Array.from(document.querySelectorAll('a,button,[role="button"],span,div'));
-					return controls.some((el) => {
-						const text = (el.textContent || '').trim();
-						const cls = (el.getAttribute('class') || '').toLowerCase();
-						const ariaDisabled = (el.getAttribute('aria-disabled') || '').toLowerCase();
-						const disabled = el.hasAttribute('disabled') || ariaDisabled === 'true' || cls.includes('disabled');
-						return visible(el) && !disabled && /^\d+$/.test(text) && parseInt(text, 10) === targetPageNumber;
-					});
-				}
-				""",
-				target_page_number,
-			))
-		except Exception:
-			return False
-
 	def open_cars_listing(self, page: Page, *, url_override: str | None = None) -> None:
 		url = url_override or self.settings.listing.cars_url
 		logger.info("Opening cars listing page: %s", url)
@@ -647,22 +625,8 @@ class ListingCollector:
 	@staticmethod
 	def _has_next_page(page: Page) -> bool:
 		for selector in ListingCollector._NEXT_PAGE_SELECTORS:
-			locator = page.locator(selector)
-			count = locator.count()
-			if count == 0:
-				continue
-			# Проверяем, что хотя бы один элемент не disabled.
-			# Disabled "Next" на последней странице не означает наличия следующей.
-			for i in range(min(count, 3)):
-				try:
-					el = locator.nth(i)
-					disabled_attr = el.get_attribute("disabled", timeout=300)
-					aria_disabled = el.get_attribute("aria-disabled", timeout=300)
-					cls = (el.get_attribute("class", timeout=300) or "").lower()
-					if disabled_attr is None and aria_disabled != "true" and "disabled" not in cls:
+			if page.locator(selector).count() > 0:
 				return True
-				except Exception:
-					continue
 		try:
 			return bool(page.evaluate(
 				"""

iaai_scraper/core/config.py

@@ -209,16 +209,6 @@ class RedisConfig:
     health_check_interval_seconds: int = _env_int("IAAI_REDIS_HEALTH_CHECK_INTERVAL_SECONDS", 30)
 
 
-# --- Конфиг Discovery (режим обнаружения, hourly batch) ---
-
-@dataclass(slots=True)
-class DiscoveryConfig:
-    mode: str = _env_str("IAAI_DISCOVERY_MODE", "sitemap")
-    hourly_mode: str = _env_str("IAAI_HOURLY_MODE", "rolling_refresh")
-    hourly_refresh_batch_size: int = _env_int("IAAI_HOURLY_REFRESH_BATCH_SIZE", 500)
-    always_full_scan: bool = _env_bool("IAAI_ALWAYS_FULL_SCAN", True)
-
-
 # --- Конфиг Celery (лимиты задач, concurrency, beat-расписание) ---
 
 @dataclass(slots=True)
@@ -227,15 +217,15 @@ class CeleryConfig:
     result_backend: str = _env_str("CELERY_RESULT_BACKEND", "")
     task_soft_time_limit: int = _env_int("CELERY_TASK_SOFT_TIME_LIMIT", 3300)
     task_time_limit: int = _env_int("CELERY_TASK_TIME_LIMIT", 3600)
-    task_stall_timeout_seconds: int = _env_int("CELERY_TASK_STALL_TIMEOUT_SECONDS", 600)
     worker_concurrency: int = _env_int("CELERY_WORKER_CONCURRENCY", 4)
     worker_max_tasks_per_child: int = _env_int("CELERY_WORKER_MAX_TASKS_PER_CHILD", 5)
+    # KB; worker перезапустит child при превышении (защита от утечек Chromium/Playwright).
+    worker_max_memory_per_child_kb: int = _env_int("CELERY_WORKER_MAX_MEMORY_PER_CHILD_KB", 500_000)
     broker_visibility_timeout: int = _env_int("CELERY_BROKER_VISIBILITY_TIMEOUT", 7200)
     beat_sync_interval_minutes: int = _env_int("CELERY_BEAT_SYNC_INTERVAL_MINUTES", 60)
     beat_sync_limit: int | None = _env_int("CELERY_BEAT_SYNC_LIMIT", 0) or None
     batch_size: int = _env_int("CELERY_BATCH_SIZE", 50)
     parallel_tabs: int = _env_int("IAAI_PARALLEL_TABS", 8)
-    parallel_segments: bool = _env_bool("CELERY_PARALLEL_SEGMENTS", False)
     block_resources: bool = _env_bool("IAAI_BLOCK_RESOURCES", True)
 
 
@@ -294,7 +284,6 @@ class Settings:
     redis: RedisConfig = field(default_factory=RedisConfig)
     celery: CeleryConfig = field(default_factory=CeleryConfig)
     proxy: ProxyConfig = field(default_factory=ProxyConfig)
-    discovery: DiscoveryConfig = field(default_factory=DiscoveryConfig)
 
     @property
     def parallel_tabs(self) -> int:

iaai_scraper/core/exceptions.py

@@ -12,3 +12,7 @@ class SiteStructureChangedError(ScraperError):
 
 class ListingResumeError(ScraperError):
     """Вызывается, когда resume по checkpoint больше недостижим."""
+
+
+class ScraperAbortedError(ScraperError):
+    """Вызывается при внешнем прерывании (например, потеря Celery lock'а)."""

iaai_scraper/core/logs.py

@@ -1,6 +1,7 @@
 import logging
 import sys
 from contextvars import ContextVar
+from logging.handlers import RotatingFileHandler
 
 # Храним trace_id текущего потока/корутины.
 TRACE_ID: ContextVar[str] = ContextVar("trace_id", default="-")
@@ -21,7 +22,16 @@ def setup_logging(level: str = "INFO", log_file: str | None = None) -> None:
     # stderr — Docker и Celery prefork корректно его подхватывают.
     handlers: list[logging.Handler] = [logging.StreamHandler(sys.stderr)]
     if log_file:
-        handlers.append(logging.FileHandler(log_file, encoding="utf-8"))
+        # Ротация файлового лога: 50MB × 5 файлов, чтобы диск VPS не переполнился
+        # при многомесячной работе.
+        handlers.append(
+            RotatingFileHandler(
+                log_file,
+                maxBytes=50 * 1024 * 1024,
+                backupCount=5,
+                encoding="utf-8",
+            )
+        )
     trace_filter = TraceIdFilter()
     for handler in handlers:
         handler.addFilter(trace_filter)

iaai_scraper/discovery/__init__.py

@@ -1,15 +0,0 @@
-from .sitemap import (
-    SitemapDiscoveryError,
-    SitemapDiscoveryResult,
-    SitemapDiscoveryStats,
-    discover_vehicle_urls_from_sitemap,
-    discover_vehicle_urls_from_sitemap_with_stats,
-)
-
-__all__ = [
-    "SitemapDiscoveryError",
-    "SitemapDiscoveryResult",
-    "SitemapDiscoveryStats",
-    "discover_vehicle_urls_from_sitemap",
-    "discover_vehicle_urls_from_sitemap_with_stats",
-]

iaai_scraper/discovery/sitemap.py

@@ -1,210 +0,0 @@
-from __future__ import annotations
-
-import gzip
-import io
-import logging
-import re
-from dataclasses import dataclass, field
-from typing import Iterable
-from urllib.parse import urlsplit, urlunsplit
-from urllib.request import Request, urlopen, ProxyHandler, build_opener
-import xml.etree.ElementTree as ET
-
-logger = logging.getLogger("iaai_scraper.discovery.sitemap")
-
-DEFAULT_SITEMAP_INDEX_URL = "https://www.iaai.com/Xj9rDOVMEi0hc38S/sitemap_index.xml"
-_SITEMAP_TIMEOUT_SECONDS = 30
-_LOC_TAG_RE = re.compile(rb"<loc>\s*(.*?)\s*</loc>", re.IGNORECASE | re.DOTALL)
-
-
-class SitemapDiscoveryError(RuntimeError):
-    pass
-
-
-def _is_vehicle_sitemap_url(url: str) -> bool:
-    lowered = url.strip().lower()
-    # Берём только sitemap с авто.
-    if "sitemapbranches" in lowered or "sitemapauctions" in lowered:
-        return False
-    return lowered.endswith(".xml") or lowered.endswith(".xml.gz")
-
-
-def _normalize_vehicle_url(url: str) -> str:
-    parts = urlsplit(url.strip())
-    return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))
-
-
-def _download_bytes(url: str, proxy_url: str | None = None) -> bytes:
-    request = Request(
-        url,
-        headers={
-            "User-Agent": (
-                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
-                "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
-            ),
-            "Accept": "application/xml,text/xml,application/xhtml+xml,text/html;q=0.9,*/*;q=0.8",
-            "Accept-Encoding": "gzip",
-        },
-    )
-    if proxy_url:
-        handler = ProxyHandler({"http": proxy_url, "https": proxy_url})
-        opener = build_opener(handler)
-        response = opener.open(request, timeout=_SITEMAP_TIMEOUT_SECONDS)
-    else:
-        response = urlopen(request, timeout=_SITEMAP_TIMEOUT_SECONDS)
-    with response:
-        payload = response.read()
-        encoding = str(response.headers.get("Content-Encoding") or "").lower()
-        if encoding == "gzip" or url.lower().endswith(".gz"):
-            return gzip.GzipFile(fileobj=io.BytesIO(payload)).read()
-        return payload
-
-
-def _local_name(tag: str) -> str:
-    if "}" in tag:
-        return tag.rsplit("}", 1)[1]
-    return tag
-
-
-def _iter_loc_values_fallback(xml_bytes: bytes) -> Iterable[str]:
-    for match in _LOC_TAG_RE.finditer(xml_bytes):
-        try:
-            value = match.group(1).decode("utf-8", errors="ignore").strip()
-        except Exception:
-            continue
-        if value:
-            yield value
-
-
-def _iter_loc_values(xml_bytes: bytes) -> Iterable[str]:
-    try:
-        root = ET.fromstring(xml_bytes)
-    except ET.ParseError as exc:
-        fallback_values = list(_iter_loc_values_fallback(xml_bytes))
-        if fallback_values:
-            logger.warning(
-                "Falling back to regex sitemap loc extraction after XML parse error: %s",
-                exc,
-            )
-            yield from fallback_values
-            return
-        raise SitemapDiscoveryError(f"Invalid sitemap XML: {exc}") from exc
-
-    for element in root.iter():
-        if _local_name(element.tag) != "loc":
-            continue
-        if not element.text:
-            continue
-        value = element.text.strip()
-        if value:
-            yield value
-
-
-def _filter_vehicle_urls(urls: Iterable[str]) -> list[str]:
-    result: list[str] = []
-    seen: set[str] = set()
-    for url in urls:
-        normalized = _normalize_vehicle_url(url)
-        if "/VehicleDetail/" not in normalized and "/vehicledetail/" not in normalized:
-            continue
-        if normalized in seen:
-            continue
-        seen.add(normalized)
-        result.append(normalized)
-    return result
-
-
-def _looks_like_vehicle_detail_sitemap(urls: list[str]) -> bool:
-    return any("/vehicledetail/" in url.lower() for url in urls)
-
-
-def discover_vehicle_urls_from_sitemap(index_url: str = DEFAULT_SITEMAP_INDEX_URL, proxy_url: str | None = None) -> list[str]:
-    logger.info("Downloading sitemap index: %s", index_url)
-    index_xml = _download_bytes(index_url, proxy_url=proxy_url)
-    sitemap_urls = [url for url in _iter_loc_values(index_xml) if _is_vehicle_sitemap_url(url)]
-    if not sitemap_urls:
-        raise SitemapDiscoveryError("Sitemap index returned no sitemap URLs")
-
-    all_vehicle_urls: list[str] = []
-    for sitemap_url in sitemap_urls:
-        logger.info("Downloading sitemap: %s", sitemap_url)
-        try:
-            sitemap_xml = _download_bytes(sitemap_url, proxy_url=proxy_url)
-            raw_urls = list(_iter_loc_values(sitemap_xml))
-        except SitemapDiscoveryError as exc:
-            logger.warning("Skipping malformed sitemap %s: %s", sitemap_url, exc)
-            continue
-
-        vehicle_urls = _filter_vehicle_urls(raw_urls)
-        if raw_urls and not vehicle_urls and not _looks_like_vehicle_detail_sitemap(raw_urls):
-            logger.info("Skipping non-vehicle sitemap %s", sitemap_url)
-            continue
-        logger.info("Sitemap %s yielded %d vehicle URLs", sitemap_url, len(vehicle_urls))
-        all_vehicle_urls.extend(vehicle_urls)
-
-    deduped = _filter_vehicle_urls(all_vehicle_urls)
-    if not deduped:
-        raise SitemapDiscoveryError("No vehicle detail URLs discovered from vehicle sitemaps")
-    logger.info("Sitemap discovery done: %d vehicle URLs", len(deduped))
-    return deduped
-
-
-@dataclass
-class SitemapDiscoveryStats:
-    transport: str = "http"
-    fetched_sitemaps: int = 0
-    blocked_sitemaps: int = 0
-    malformed_sitemaps: int = 0
-    direct_probe_hits: int = 0
-    direct_probe_misses: int = 0
-
-
-@dataclass
-class SitemapDiscoveryResult:
-    vehicle_urls: list[str] = field(default_factory=list)
-    stats: SitemapDiscoveryStats = field(default_factory=SitemapDiscoveryStats)
-
-
-def discover_vehicle_urls_from_sitemap_with_stats(
-    settings=None,
-    index_url: str = DEFAULT_SITEMAP_INDEX_URL,
-) -> SitemapDiscoveryResult:
-    """Возвращает URL и статистику."""
-    proxy_url = None
-    if settings is not None and hasattr(settings, 'proxy') and settings.proxy.server:
-        proxy_url = settings.proxy.server
-    stats = SitemapDiscoveryStats(transport="http")
-    logger.info("Downloading sitemap index: %s", index_url)
-    index_xml = _download_bytes(index_url, proxy_url=proxy_url)
-    sitemap_urls = [url for url in _iter_loc_values(index_xml) if _is_vehicle_sitemap_url(url)]
-    if not sitemap_urls:
-        raise SitemapDiscoveryError("Sitemap index returned no sitemap URLs")
-
-    all_vehicle_urls: list[str] = []
-    for sitemap_url in sitemap_urls:
-        logger.info("Downloading sitemap: %s", sitemap_url)
-        try:
-            sitemap_xml = _download_bytes(sitemap_url, proxy_url=proxy_url)
-            raw_urls = list(_iter_loc_values(sitemap_xml))
-            stats.fetched_sitemaps += 1
-        except SitemapDiscoveryError as exc:
-            logger.warning("Skipping malformed sitemap %s: %s", sitemap_url, exc)
-            stats.malformed_sitemaps += 1
-            continue
-        except Exception as exc:
-            logger.warning("Blocked/failed sitemap %s: %s", sitemap_url, exc)
-            stats.blocked_sitemaps += 1
-            continue
-
-        vehicle_urls = _filter_vehicle_urls(raw_urls)
-        if raw_urls and not vehicle_urls and not _looks_like_vehicle_detail_sitemap(raw_urls):
-            logger.info("Skipping non-vehicle sitemap %s", sitemap_url)
-            continue
-        logger.info("Sitemap %s yielded %d vehicle URLs", sitemap_url, len(vehicle_urls))
-        all_vehicle_urls.extend(vehicle_urls)
-
-    deduped = _filter_vehicle_urls(all_vehicle_urls)
-    if not deduped:
-        raise SitemapDiscoveryError("No vehicle detail URLs discovered from vehicle sitemaps")
-    logger.info("Sitemap discovery done: %d vehicle URLs", len(deduped))
-    return SitemapDiscoveryResult(vehicle_urls=deduped, stats=stats)

iaai_scraper/parsing/mapper.py

@@ -20,7 +20,7 @@ from ..storage.schemas import CarRecord, ImageRecord
 
 
 class CarMapper:
-	# Маппер IAAI в CarRecord.
+	# Преобразование данных IAAI в CarRecord.
 
 	BODY_MAP = {
 		"sedan": "SEDAN", "coupe": "COUPE", "hatchback": "HATCHBACK", "sport utility": "SUV",
@@ -48,7 +48,7 @@ class CarMapper:
 	NO_DAMAGE_MARKERS = {"normal wear", "normal wear & tear", "normal wear and tear", "n/a", "na", "none", "no damage", "minor dents/scratches"}
 
 	def map_to_car_record(self, vehicle_url: str, vehicle_summary: dict[str, Any], payload_insights: dict[str, Any]) -> CarRecord:
-		# Собираем DB-модель.
+		# Собираем нормализованную DB-модель.
 		vehicle_summary = vehicle_summary or {}
 		payload_insights = payload_insights or {}
 		core = payload_insights.get("vehicle_core", {})
@@ -77,7 +77,7 @@ class CarMapper:
 		)
 		color = self._normalize_color(first_non_empty([core.get("color"), vehicle_summary.get("color"), "other"]))
 		drive = self._normalize_drive(first_non_empty([core.get("drive"), vehicle_summary.get("drive")]))
-		# Пробуем взять привод из двигателя.
+		# Пытаемся определить привод из строки двигателя.
 		if not drive or drive == "NA":
 			engine_text = self._as_str(first_non_empty([core.get("engine"), vehicle_summary.get("engine")]))
 			if engine_text:
@@ -144,7 +144,7 @@ class CarMapper:
 
 	@classmethod
 	def _to_money_int(cls, value: Any) -> int | None:
-		# Нормализация цены.
+		# Нормализация стоимости из разных форматов.
 		if value is None:
 			return None
 		if isinstance(value, bool):
@@ -168,14 +168,14 @@ class CarMapper:
 		for number in numbers:
 			clean = number.replace(" ", "")
 			if "," in clean and "." in clean:
-					# Поддержка двух форматов.
+					# Поддержка 1,234.56 и 1.234,56.
 				if clean.rfind(",") > clean.rfind("."):
 					clean = clean.replace(".", "").replace(",", ".")
 				else:
 					clean = clean.replace(",", "")
 			elif "," in clean:
 				parts = clean.split(",")
-				# Десятичный или тысячный разделитель.
+				# 123,45 -> 123.45, иначе разделитель тысяч.
 				if len(parts[-1]) in {1, 2} and len(parts) == 2:
 					clean = clean.replace(",", ".")
 				else:
@@ -214,7 +214,7 @@ class CarMapper:
 
 	@staticmethod
 	def _parse_odometer(value: Any) -> int:
-		# Разбор пробега.
+		# Разбор пробега из строк IAAI.
 		if value is None:
 			return 0
 		text = str(value).strip()
@@ -223,7 +223,7 @@ class CarMapper:
 		lowered = text.lower()
 		if any(skip in lowered for skip in ["not required", "exempt", "n/a", "na", "unknown", "tbd"]):
 			return 0
-		# Ищем число.
+		# Извлекаем число из строкового формата одометра.
 		numbers = re.findall(r"[\d,]+", text)
 		for num_str in numbers:
 			clean = num_str.replace(",", "")
@@ -294,7 +294,7 @@ class CarMapper:
 		empty_default: str | None,
 		fallback: str | None,
 	) -> str | None:
-		# Общая нормализация enum.
+		# Общий helper для enum-нормализации.
 		text = self._as_str(value).lower()
 		if not text:
 			return empty_default
@@ -329,7 +329,7 @@ class CarMapper:
 		return any(token in self._as_str(auction.get("sale_status")).lower() for token in ["sold", "closed", "ended"])
 
 	def _build_images(self, urls: list[Any]) -> list[ImageRecord]:
-		# Для imageKeys берём самый большой размер.
+		# Для imageKeys берем самый большой размер.
 		best_by_key: dict[str, str] = {}
 		key_order: list[str] = []
 		non_keyed: list[str] = []
@@ -375,11 +375,11 @@ class CarMapper:
 
 	@classmethod
 	def _generate_parser_id(cls, origin_id: str) -> str:
-		# Стабильный parser_id.
+		# Стабильный parser_id по origin_id.
 		digest = hashlib.sha256(origin_id.encode()).digest()
 		alphabet = cls._PARSER_ID_ALPHABET
 		base = len(alphabet)
-		num = int.from_bytes(digest[:17], "big")  # Хватает на 22 символа.
+		num = int.from_bytes(digest[:17], "big")  # 17 байт = 136 бит, хватает на 22 символа
 		chars: list[str] = []
 		for _ in range(22):
 			num, idx = divmod(num, base)
@@ -387,7 +387,7 @@ class CarMapper:
 		return "car-" + "".join(chars)
 
 	def _build_origin_id(self, vehicle_url: str, vehicle_summary: dict[str, Any], core: dict[str, Any]) -> str:
-		# Формат: iaai:{lot_number}.
+		# Формат: iaai:{lot_number} (аналог copart:94323985).
 		for value in [core.get("lot_number"), vehicle_summary.get("lot_number")]:
 			text = self._as_str(value)
 			if text:

iaai_scraper/parsing/parser.py

@@ -10,9 +10,9 @@ logger = logging.getLogger("iaai_scraper.parsers")
 
 
 class VehicleParser:
-	# Парсер страницы авто.
+	# Парсер данных страницы автомобиля.
 
-	# Регулярки парсинга и защиты.
+	# Регулярные выражения для парсинга и детекции защиты.
 	_BUY_NOW_RE = re.compile(r"Buy\s+Now[:\s]*\$\s*([\d,]+(?:\.\d{1,2})?)", re.IGNORECASE)
 	_CAPTCHA_TOKENS = frozenset(["captcha", "verify you are human", "i am human", "recaptcha", "cloudflare"])
 	_ANTIBOT_TOKENS = frozenset(["incapsula", "access denied", "request unsuccessful", "bot detection"])
@@ -101,11 +101,11 @@ class VehicleParser:
 		max_fields = len(set(self.DOM_LABEL_MAP.values()))
 
 		for i, line in enumerate(lines):
-			# Ранний выход.
+			# Ранний выход, когда уже нашли все поля.
 			if len(result) >= max_fields:
 				break
 
-			# Метка и значение в одной строке.
+			# Сценарий 1: "метка: значение" в одной строке.
 			colon_pos = line.find(":")
 			if colon_pos > 0:
 				label_part = line[:colon_pos].strip().lower()
@@ -116,7 +116,7 @@ class VehicleParser:
 						result[field_name] = value_part
 						continue
 
-			# Метка и значение в соседних строках.
+			# Сценарий 2: метка и значение на соседних строках.
 			clean = line.rstrip(":").strip().lower()
 			clean_alt = clean.rstrip("#").strip()
 			matched_label = None
@@ -164,7 +164,7 @@ class VehicleParser:
 			page_title = title_match.group(1).strip()
 		title_parsed = self._parse_title_for_year_make_model(page_title, dom_text)
 
-		# Один проход по payload.
+		# Один проход по payload для всех полей.
 		all_found = deep_find_all_keys(payloads, self.SUMMARY_KEY_MAP)
 
 		summary: dict[str, Any] = {"source_url": vehicle_url}
@@ -199,7 +199,7 @@ class VehicleParser:
 			p = item.get("payload")
 			if isinstance(p, (dict, list)):
 				payloads.append(p)
-				# Доп. проход по JSON.
+				# Дополнительный проход по встроенному JSON.
 				extra = deep_find_all_keys([p], self.SUMMARY_KEY_MAP)
 				for field, vals in extra.items():
 					if not summary.get(field):
@@ -207,7 +207,7 @@ class VehicleParser:
 						if v:
 							summary[field] = v
 
-		# Передаём готовые image_urls.
+		# Передаём image_urls без повторного извлечения.
 		image_urls = summary.get("image_urls") or []
 		return {
 			"vehicle_summary": summary,
@@ -325,7 +325,7 @@ class VehicleParser:
 		for script_text in scripts:
 			if "{" not in script_text and "[" not in script_text:
 				continue
-			# Пропускаем большие блоки.
+			# Пропускаем слишком большие минифицированные блоки.
 			if len(script_text) > 51_200:
 				continue
 			try:

iaai_scraper/storage/db.py

@@ -20,7 +20,6 @@ CAR_DB_FIELDS = {
 }
 
 _IN_CHUNK_SIZE = 5000
-CAR_TABLE_NAME = Car.__tablename__
 
 
 class PersistenceService:
@@ -36,16 +35,11 @@ class PersistenceService:
 			engine_kwargs["max_overflow"] = settings.database.max_overflow
 			engine_kwargs["pool_pre_ping"] = True
 			engine_kwargs["pool_recycle"] = settings.database.pool_recycle_seconds
-			engine_kwargs["pool_timeout"] = 30
-			# Таймауты запросов и блокировок.
-			engine_kwargs["connect_args"] = {
-				"options": "-c statement_timeout=120000 -c lock_timeout=30000"
-			}
 		self.engine = create_engine(settings.database.url, **engine_kwargs)
 		self.session_factory = sessionmaker(bind=self.engine, expire_on_commit=False, future=True)
 
 	def create_tables(self) -> None:
-		# Для SQLite можно create_all.
+		# В тестах/локально на SQLite разрешаем create_all; для non-SQLite в проде — только через миграции.
 		is_sqlite = self.settings.database.url.startswith("sqlite")
 		if not is_sqlite and not self.settings.database.auto_create_tables:
 			return
@@ -111,7 +105,7 @@ class PersistenceService:
 	def get_existing_urls_and_ids(
 		self, origin_urls: list[str], origin_ids: list[str],
 	) -> tuple[set[str], set[str]]:
-		# Загрузка URL и origin_id.
+		# Загрузка существующих URL и origin_id.
 		if not origin_urls and not origin_ids:
 			return set(), set()
 		urls: set[str] = set()
@@ -320,7 +314,7 @@ class PersistenceService:
 		return {"inserted": inserted, "updated": updated, "images_upserted": images_total}
 
 	def upsert_car(self, record: CarRecord):
-		# Вставка или обновление авто.
+		# Вставка или обновление автомобиля по origin_id/origin_url.
 		payload = self._car_payload(record)
 		images = [image.model_dump(mode="python") for image in record.images]
 		with self.session_scope() as session:
@@ -381,7 +375,7 @@ class PersistenceService:
 		- Один DELETE по car_id IN (...) вместо удаления по одному.
 		- Fallback на по-одному upsert если batch commit упал.
 		"""
-		# Дедупликация батча.
+		# ── Дедупликация записей внутри батча ──
 		seen_ids: dict[str, int] = {}
 		unique_records: list[CarRecord] = []
 		for idx, r in enumerate(records):
@@ -412,20 +406,20 @@ class PersistenceService:
 		images_total = 0
 
 		with self.session_scope() as session:
-			# Загружаем существующие записи.
+			# Получаем существующие записи chunked-запросами.
 			origin_ids = [r.origin_id for r in records if r.origin_id]
 			origin_urls = [r.origin_url for r in records if r.origin_url]
 
 			existing_by_id, existing_by_url = self._load_existing_cars(session, origin_ids, origin_urls)
 
-			# Предзагружаем изображения.
+			# Предзагружаем ВСЕ изображения для обновляемых машин одним запросом.
 			existing_car_ids = set()
 			for record in records:
 				car = existing_by_id.get(record.origin_id) or existing_by_url.get(record.origin_url)
 				if car is not None:
 					existing_car_ids.add(int(car.id))
 
-			# Готовим map car_id -> image_urls.
+			# Строим маппинг car_id → set(image_urls) для сравнения.
 			existing_images_map = self._load_existing_image_urls(session, existing_car_ids)
 
 			new_cars: list[tuple[Car, list[dict]]] = []
@@ -447,7 +441,7 @@ class PersistenceService:
 					car.last_seen_at = record.last_seen_at
 					updated += 1
 
-					# Проверяем изменения картинок.
+					# Проверяем, изменились ли изображения.
 					new_image_urls = {img.get("fullres_image", "") for img in images}
 					old_image_urls = existing_images_map.get(int(car.id), set())
 					if new_image_urls != old_image_urls:
@@ -455,15 +449,15 @@ class PersistenceService:
 					else:
 						images_total += len(old_image_urls)
 
-			# Один flush.
+			# Один flush для всех вставок.
 			session.flush()
 
-			# Добавляем картинки новым авто.
+			# Добавляем изображения для новых автомобилей.
 			for car, images in new_cars:
 				self._add_images(session, int(car.id), images)
 				images_total += len(images)
 
-			# Обновляем только изменённые картинки.
+			# Массово обновляем изображения только для машин с изменёнными картинками.
 			if update_cars_needing_images:
 				update_ids = [int(car.id) for car, _ in update_cars_needing_images]
 				for i in range(0, len(update_ids), _IN_CHUNK_SIZE):
@@ -476,7 +470,7 @@ class PersistenceService:
 		return {"inserted": inserted, "updated": updated, "images_upserted": images_total}
 
 	def _upsert_cars_individually(self, records: list[CarRecord]) -> dict[str, int]:
-		# Запасной поштучный upsert.
+		# Fallback на поштучный upsert.
 		inserted = 0
 		updated = 0
 		images_total = 0
@@ -516,107 +510,55 @@ class PersistenceService:
 
 		Для PostgreSQL использует временную таблицу + LEFT JOIN вместо NOT IN,
 		что кардинально быстрее при больших объёмах (100K+ URLs).
-		Встроенная защита: нормализация URL (тильда/дефис) + safety-check на аномальный процент.
 		"""
 		if not active_origin_urls:
 			return 0
 
-		# Нормализация URL.
-		def _norm(url: str) -> str:
-			return url.replace("~", "-")
-
-		normalized_urls = {_norm(u) for u in active_origin_urls}
-
 		is_postgres = "postgresql" in self.settings.database.url
 
 		with self.session_scope() as session:
 			if is_postgres:
-				# Считаем кандидатов на sold.
-				total_active = session.execute(
-					text(f"SELECT count(*) FROM {CAR_TABLE_NAME} WHERE is_sold = FALSE AND origin_id LIKE 'iaai:%%'")
-				).scalar() or 0
-
-				if total_active == 0:
-					return 0
-
-				# Временная таблица URL.
+				# Создаём временную таблицу с активными URL.
 				session.execute(text("CREATE TEMP TABLE IF NOT EXISTS _active_urls (url TEXT NOT NULL) ON COMMIT DROP"))
 				session.execute(text("TRUNCATE _active_urls"))
 
-				# Вставляем URL чанками.
-				url_list = list(normalized_urls)
+				# Вставляем активные URL чанками.
+				url_list = list(active_origin_urls)
 				for i in range(0, len(url_list), _IN_CHUNK_SIZE):
 					chunk = url_list[i:i + _IN_CHUNK_SIZE]
 					values = ",".join(f"(:{f'u{j}'})" for j in range(len(chunk)))
 					params = {f"u{j}": url for j, url in enumerate(chunk)}
 					session.execute(text(f"INSERT INTO _active_urls (url) VALUES {values}"), params)
 
-				# Индекс для JOIN.
+				# Создаём индекс на временной таблице для ускорения JOIN.
 				session.execute(text("CREATE INDEX IF NOT EXISTS _ix_active_urls ON _active_urls (url)"))
 
-				# Считаем будущие sold.
-				would_mark = session.execute(text("""
-					SELECT count(*)
-					FROM {car_table} c
-					LEFT JOIN _active_urls a ON replace(c.origin_url, '~', '-') = a.url
-					WHERE a.url IS NULL
-					AND c.is_sold = FALSE
-					AND c.origin_id LIKE 'iaai:%%'
-				""".format(car_table=CAR_TABLE_NAME))).scalar() or 0
-
-				# Защита от аномалии.
-				if total_active > 100 and would_mark > total_active * 0.8:
-					logger.error(
-						"mark_sold safety abort: would mark %d/%d (%.0f%%) as sold — likely URL format mismatch",
-						would_mark, total_active, would_mark / total_active * 100,
-					)
-					return 0
-
-				# Массовая пометка sold.
+				# Массовая пометка проданных в PostgreSQL.
 				result = session.execute(text("""
-					UPDATE {car_table}
+					UPDATE cars
 					SET is_sold = TRUE
 					FROM (
 						SELECT c.id
-						FROM {car_table} c
-						LEFT JOIN _active_urls a ON replace(c.origin_url, '~', '-') = a.url
+						FROM cars c
+						LEFT JOIN _active_urls a ON c.origin_url = a.url
 						WHERE a.url IS NULL
 						AND c.is_sold = FALSE
 						AND c.origin_id LIKE 'iaai:%%'
 					) sub
-					WHERE {car_table}.id = sub.id
-				""".format(car_table=CAR_TABLE_NAME)))
+					WHERE cars.id = sub.id
+				"""))
 				count = result.rowcount or 0
 			else:
 				# Упрощённый путь для SQLite.
 				stmt = (
 					update(Car)
+					.where(Car.origin_url.notin_(active_origin_urls))
 					.where(Car.is_sold == False)  # noqa: E712
 					.where(Car.origin_id.like("iaai:%"))
 					.values(is_sold=True)
 				)
-				# Загружаем active URL.
-				all_active = session.execute(
-					select(Car.id, Car.origin_url).where(
-						Car.is_sold == False, Car.origin_id.like("iaai:%")  # noqa: E712
-					)
-				).all()
-				mark_ids = [row[0] for row in all_active if _norm(row[1]) not in normalized_urls]
-
-				if not mark_ids:
-					return 0
-				total_active = len(all_active)
-				if total_active > 100 and len(mark_ids) > total_active * 0.8:
-					logger.error(
-						"mark_sold safety abort: would mark %d/%d (%.0f%%) as sold — likely URL format mismatch",
-						len(mark_ids), total_active, len(mark_ids) / total_active * 100,
-					)
-					return 0
-
-				for i in range(0, len(mark_ids), _IN_CHUNK_SIZE):
-					chunk = mark_ids[i:i + _IN_CHUNK_SIZE]
-					session.execute(update(Car).where(Car.id.in_(chunk)).values(is_sold=True))
-				count = len(mark_ids)
+				result = session.execute(stmt)
+				count = result.rowcount or 0
 
 			if count:
 				logger.info("Marked %d cars as sold by URL (no longer in listing)", count)
@@ -632,45 +574,3 @@ class PersistenceService:
 				select(Car.origin_id).where(Car.origin_id.like(f"{prefix}%")).execution_options(yield_per=10000)
 			)
 			return {str(row[0]) for row in result if row and row[0]}
-
-	def get_all_active_origin_urls_for_lane(self, prefix: str = "iaai:") -> set[str]:
-		"""Возвращает origin_url всех активных (не проданных) авто для заданного lane-префикса."""
-		with self.session_scope() as session:
-			result = session.execute(
-				select(Car.origin_url).where(
-					Car.origin_id.like(f"{prefix}%"),
-					Car.is_sold == False,  # noqa: E712
-				).execution_options(yield_per=10000)
-			)
-			return {str(row[0]) for row in result if row and row[0]}
-
-	def count_active_cars_for_lane(self, prefix: str = "iaai:") -> int:
-		"""Возвращает количество активных (не проданных) авто для заданного lane-префикса."""
-		from sqlalchemy import func as sa_func
-		with self.session_scope() as session:
-			result = session.execute(
-				select(sa_func.count()).select_from(Car).where(
-					Car.origin_id.like(f"{prefix}%"),
-					Car.is_sold == False,  # noqa: E712
-				)
-			)
-			return int(result.scalar() or 0)
-
-	def get_active_origin_urls_batch_for_refresh(
-		self,
-		prefix: str = "iaai:",
-		offset: int = 0,
-		limit: int = 500,
-	) -> list[str]:
-		"""Возвращает батч origin_url активных авто для rolling refresh.
-
-		Сортировка по last_seen_at ASC — давно не обновлённые идут первыми.
-		"""
-		with self.session_scope() as session:
-			result = session.execute(
-				select(Car.origin_url).where(
-					Car.origin_id.like(f"{prefix}%"),
-					Car.is_sold == False,  # noqa: E712
-				).order_by(Car.last_seen_at.asc()).offset(offset).limit(limit)
-			)
-			return [str(row[0]) for row in result if row and row[0]]

iaai_scraper/storage/models.py

@@ -20,10 +20,10 @@ class Base(DeclarativeBase):
 
 
 class Car(Base):
-	__tablename__ = "iaai_cars"
+	__tablename__ = "cars"
 	__table_args__ = (
-		Index("ix_iaai_cars_brand_model", "brand", "model"),
-		Index("ix_iaai_cars_origin_id_not_sold", "origin_id", "is_sold"),
+		Index("ix_cars_brand_model", "brand", "model"),
+		Index("ix_cars_origin_id_not_sold", "origin_id", "is_sold"),
 	)
 	id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
 	parser_id: Mapped[str] = mapped_column(String(50), nullable=False, unique=True)
@@ -59,17 +59,17 @@ class Car(Base):
 
 
 class Image(Base):
-	__tablename__ = "iaai_images"
+	__tablename__ = "images"
 	id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
 	fullres_image: Mapped[str] = mapped_column(String(), nullable=False)
 	preview_image: Mapped[str] = mapped_column(String(), nullable=False)
 	order_index: Mapped[int] = mapped_column(Integer, nullable=False)
-	car_id: Mapped[int] = mapped_column(Integer, ForeignKey("iaai_cars.id", ondelete="CASCADE"), nullable=False, index=True)
+	car_id: Mapped[int] = mapped_column(Integer, ForeignKey("cars.id", ondelete="CASCADE"), nullable=False, index=True)
 	car: Mapped[Car] = relationship("Car", back_populates="images")
 
 
 class SyncRun(Base):
-	__tablename__ = "iaai_sync_runs"
+	__tablename__ = "sync_runs"
 	id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
 	started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=func.now())
 	finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

iaai_scraper/worker/celery_app.py

@@ -1,7 +1,6 @@
 # Инициализация Celery-приложения и периодических задач.
 
 import logging
-import os
 
 from celery import Celery
 from celery.signals import worker_process_init, worker_ready, setup_logging as celery_setup_logging
@@ -12,12 +11,6 @@ from ..core.logs import setup_logging
 
 logger = logging.getLogger("iaai_scraper.worker.celery_app")
 STARTUP_SYNC_DISPATCH_KEY = "iaai:state:startup_sync_dispatched"
-IAAI_SYNC_QUEUE = "iaai_sync"
-
-
-def _env_bool(name: str, default: bool) -> bool:
-    raw = os.getenv(name, "true" if default else "false").strip().lower()
-    return raw in {"1", "true", "yes", "on"}
 
 
 @celery_setup_logging.connect
@@ -34,6 +27,12 @@ def _on_worker_process_init(**kwargs):
     level = settings.log_level if settings.log_level else "INFO"
     setup_logging(level, None)
 
+    # Сбрасываем cached engine/redis после fork — иначе child наследует
+    # коннекты родителя, что небезопасно (особенно с psycopg2).
+    from . import tasks as _tasks
+    _tasks._CACHED_PERSISTENCE = None
+    _tasks._CACHED_REDIS = None
+
 
 def _broker_url() -> str:
     return settings.celery.broker_url or settings.redis.url
@@ -73,32 +72,30 @@ celery_app.conf.update(
     task_acks_late=True,
     task_reject_on_worker_lost=True,
     task_track_started=True,
+    task_ignore_result=True,
     worker_concurrency=settings.celery.worker_concurrency,
     worker_max_tasks_per_child=settings.celery.worker_max_tasks_per_child,
-    worker_pool="solo",
+    worker_max_memory_per_child=settings.celery.worker_max_memory_per_child_kb,
+    worker_pool="prefork",
     worker_prefetch_multiplier=1,
     broker_connection_retry_on_startup=True,
     broker_transport_options={
         "visibility_timeout": settings.celery.broker_visibility_timeout,
     },
-    result_expires=86400,
+    result_expires=3600,
     worker_redirect_stdouts=False,
     worker_hijack_root_logger=False,
     beat_schedule={
         "periodic-sync-listing": {
-            "task": "iaai.sync_cars_feed",
+            "task": "iaai_scraper.worker.tasks.sync_listing_task",
             "schedule": settings.celery.beat_sync_interval_minutes * 60.0,
             "args": (),
-            "kwargs": {"limit": settings.celery.beat_sync_limit, "only_new": True},
-            "options": {
-                "queue": IAAI_SYNC_QUEUE,
-                "expires": settings.celery.beat_sync_interval_minutes * 60.0,
-            },
+            "kwargs": {"limit": settings.celery.beat_sync_limit, "only_new": False},
+            "options": {"queue": "scraping"},
         }
     },
     task_routes={
-        "iaai.sync_cars_feed": {"queue": IAAI_SYNC_QUEUE},
-        "iaai_scraper.worker.tasks.*": {"queue": IAAI_SYNC_QUEUE},
+        "iaai_scraper.worker.tasks.*": {"queue": "scraping"}
     },
 )
 
@@ -107,12 +104,8 @@ celery_app.autodiscover_tasks(["iaai_scraper.worker"])
 
 @worker_ready.connect
 def _on_worker_ready(**kwargs):
-    """При старте worker отправляем первый sync_listing, если очередь пуста."""
-    if not _env_bool("IAAI_STARTUP_SYNC_ENABLED", True):
-        logger.info("Worker ready: startup sync dispatch disabled by IAAI_STARTUP_SYNC_ENABLED")
-        return
-
-    redis_client = None
+    """Сразу при старте worker отправляем первую задачу sync_listing,
+    чтобы не ждать час до первого beat-цикла."""
     try:
         redis_client = Redis.from_url(
             settings.redis.url,
@@ -122,34 +115,13 @@ def _on_worker_ready(**kwargs):
             health_check_interval=settings.redis.health_check_interval_seconds,
             retry_on_timeout=True,
         )
-
-        for stale_key in ("iaai:locks:sync_listing",):
-            try:
-                ttl = redis_client.ttl(stale_key)
-                if ttl is not None and ttl != -2:
-                    redis_client.delete(stale_key)
-                    logger.warning("Cleared stale lock on startup: %s (ttl was %s)", stale_key, ttl)
-            except Exception:
-                logger.warning("Failed to clear stale lock %s on startup", stale_key, exc_info=True)
-
-        try:
-            queue_len = int(redis_client.llen(IAAI_SYNC_QUEUE) or 0)
-        except Exception:
-            queue_len = 0
-        if queue_len > 0:
-            logger.info("Worker ready: iaai_sync queue already has %d task(s); skip startup dispatch", queue_len)
-            return
-
-        should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=600))
+        # Дедуп TTL = интервал beat (по умолчанию 1ч), чтобы не плодить дубликаты
+        # при flapping-рестартах контейнера.
+        dedupe_ttl = max(600, settings.celery.beat_sync_interval_minutes * 60)
+        should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=dedupe_ttl))
     except Exception:
         logger.warning("Worker ready startup sync dedupe check failed; skipping immediate dispatch", exc_info=True)
         return
-    finally:
-        if redis_client is not None:
-            try:
-                redis_client.close()
-            except Exception:
-                pass
 
     if not should_dispatch:
         logger.info("Worker ready immediate sync already dispatched recently; skipping duplicate enqueue")
@@ -157,8 +129,7 @@ def _on_worker_ready(**kwargs):
 
     logger.info("Worker ready — dispatching initial sync_listing task")
     celery_app.send_task(
-        "iaai.sync_cars_feed",
+        "iaai_scraper.worker.tasks.sync_listing_task",
         kwargs={"limit": settings.celery.beat_sync_limit, "only_new": False},
-        queue=IAAI_SYNC_QUEUE,
-        expires=settings.celery.beat_sync_interval_minutes * 60.0,
+        queue="scraping",
     )

iaai_scraper/worker/self_heal.py

@@ -1,219 +0,0 @@
-import json
-import logging
-import os
-import random
-import signal
-import time
-
-from redis import Redis
-
-
-logger = logging.getLogger("iaai_scraper.worker.self_heal")
-
-GLOBAL_PROGRESS_TS_KEY = "iaai:state:last_progress_ts"
-SELF_HEAL_RESTART_LOCK_KEY = "iaai:state:self_heal_restart_in_progress"
-SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
-SIGKILL_FALLBACK = getattr(signal, "SIGKILL", signal.SIGTERM)
-
-
-def _env_bool(name: str, default: bool) -> bool:
-	value = os.getenv(name)
-	if value is None:
-		return default
-	return value.strip().lower() in {"1", "true", "yes", "on"}
-
-
-def _env_int(name: str, default: int) -> int:
-	value = os.getenv(name)
-	if value is None:
-		return default
-	try:
-		return int(value.strip())
-	except Exception:
-		return default
-
-
-def _get_redis() -> Redis:
-	url = os.getenv("IAAI_REDIS_URL", "redis://redis:6379/0")
-	return Redis.from_url(
-		url,
-		decode_responses=True,
-		socket_connect_timeout=5.0,
-		socket_timeout=10.0,
-		health_check_interval=30,
-		retry_on_timeout=True,
-	)
-
-
-def _safe_int(value: str | None, default: int = 0) -> int:
-	if value is None:
-		return default
-	try:
-		return int(str(value).strip())
-	except Exception:
-		return default
-
-
-def _read_last_progress_ts(redis_client: Redis) -> int | None:
-	raw = redis_client.get(GLOBAL_PROGRESS_TS_KEY)
-	if raw:
-		ts = _safe_int(raw)
-		if ts > 0:
-			return ts
-
-	# Fallback: если глобальный ключ не найден, берём max(ts) из task_progress:*.
-	# Это дороже, но выполняется только при отсутствии основного маркера.
-	max_ts = 0
-	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)
-			ts = _safe_int(data.get("ts"), 0)
-			if ts > max_ts:
-				max_ts = ts
-		except Exception:
-			continue
-	return max_ts or None
-
-
-def _has_inflight_work(redis_client: Redis, queue_name: str) -> tuple[bool, dict[str, int]]:
-	"""Есть ли признаки активной/зависшей работы, даже если очередь пуста."""
-	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
-	flags = {
-		"queue_len": queue_len,
-		"has_lock": has_lock,
-		"has_task_progress": has_task_progress,
-	}
-	return (queue_len > 0 or has_lock == 1 or has_task_progress == 1), flags
-
-
-def _kill_worker_process() -> None:
-	pid_file = "/tmp/celery-worker.pid"
-	pid: int | None = None
-	try:
-		with open(pid_file, "r", encoding="utf-8") as f:
-			pid = int(f.read().strip())
-	except Exception:
-		pid = None
-
-	if not pid:
-		logger.error("Self-heal: failed to read worker pid from %s", pid_file)
-		return
-
-	logger.error("Self-heal: terminating stuck worker process pid=%s", pid)
-	try:
-		os.kill(pid, signal.SIGTERM)
-	except Exception:
-		logger.exception("Self-heal: failed to send SIGTERM to pid=%s", pid)
-		return
-
-	time.sleep(20)
-	try:
-		# Если процесс ещё жив — принудительно убиваем.
-		os.kill(pid, 0)
-		logger.error("Self-heal: worker pid=%s did not stop after SIGTERM; sending SIGKILL", pid)
-		os.kill(pid, SIGKILL_FALLBACK)
-	except ProcessLookupError:
-		pass
-	except Exception:
-		logger.exception("Self-heal: failed to send SIGKILL to pid=%s", pid)
-
-
-def main() -> None:
-	if not _env_bool("IAAI_SELF_HEAL_ENABLED", True):
-		logger.info("Self-heal watchdog disabled via IAAI_SELF_HEAL_ENABLED")
-		return
-
-	queue_name = os.getenv("IAAI_CELERY_QUEUE", "scraping")
-	check_interval = max(5, _env_int("IAAI_SELF_HEAL_CHECK_INTERVAL_SECONDS", 30))
-	stall_seconds = max(180, _env_int("IAAI_SELF_HEAL_STALL_SECONDS", 720))
-	startup_grace = max(30, _env_int("IAAI_SELF_HEAL_STARTUP_GRACE_SECONDS", 300))
-	restart_cooldown = max(60, _env_int("IAAI_SELF_HEAL_RESTART_COOLDOWN_SECONDS", 300))
-
-	logger.warning(
-		"Self-heal watchdog enabled: queue=%s check_interval=%ss stall=%ss startup_grace=%ss cooldown=%ss",
-		queue_name,
-		check_interval,
-		stall_seconds,
-		startup_grace,
-		restart_cooldown,
-	)
-
-	started_at = time.time()
-	redis_client: Redis | None = None
-
-	while True:
-		try:
-			if redis_client is None:
-				redis_client = _get_redis()
-				redis_client.ping()
-
-			has_inflight, inflight = _has_inflight_work(redis_client, queue_name)
-			if not has_inflight:
-				time.sleep(check_interval)
-				continue
-
-			last_progress_ts = _read_last_progress_ts(redis_client)
-			now_ts = int(time.time())
-			age = None if last_progress_ts is None else max(0, now_ts - int(last_progress_ts))
-
-			if age is None:
-				if now_ts - int(started_at) < startup_grace:
-					time.sleep(check_interval)
-					continue
-				logger.warning(
-					"Self-heal: inflight=%s but no progress timestamp found after startup grace",
-					inflight,
-				)
-				age = stall_seconds + 1
-
-			if age <= stall_seconds:
-				time.sleep(check_interval)
-				continue
-
-			# Глобальный anti-storm lock: чтобы много воркеров не рестартились одновременно.
-			acquired = bool(
-				redis_client.set(
-					SELF_HEAL_RESTART_LOCK_KEY,
-					str(now_ts),
-					nx=True,
-					ex=restart_cooldown,
-				)
-			)
-			if not acquired:
-				time.sleep(check_interval)
-				continue
-
-			logger.error(
-				"Self-heal: detected global stall (inflight=%s, progress_age=%ss > %ss). Restarting worker process...",
-				inflight,
-				age,
-				stall_seconds,
-			)
-			# Небольшой джиттер, чтобы при одинаковом событии у разных контейнеров
-			# перезапуск был не строго одновременно.
-			time.sleep(random.uniform(0.3, 2.0))
-			_kill_worker_process()
-			# После kill pid1 контейнер будет перезапущен Docker restart-policy.
-			# На случай неуспеха не молотим цикл.
-			time.sleep(check_interval)
-
-		except Exception:
-			logger.exception("Self-heal watchdog iteration failed")
-			redis_client = None
-			time.sleep(check_interval)
-
-
-if __name__ == "__main__":
-	logging.basicConfig(
-		level=os.getenv("IAAI_LOG_LEVEL", "INFO"),
-		format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
-	)
-	main()

tests/test_resilience.py

@@ -1,79 +0,0 @@
-from __future__ import annotations
-
-import unittest
-from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
-
-from iaai_scraper.scraper import IAAIScraper
-from iaai_scraper.core.config import Settings
-from iaai_scraper.worker import tasks
-
-
-class TestResilience(unittest.TestCase):
-	def _make_scraper(self) -> IAAIScraper:
-		s = Settings()
-		s.log_level = "CRITICAL"
-		s.database.url = "sqlite://"
-		return IAAIScraper(s)
-
-	def test_update_task_progress_updates_global_marker(self) -> None:
-		redis_client = MagicMock()
-		pipe = MagicMock()
-		redis_client.pipeline.return_value = pipe
-
-		tasks._update_task_progress(
-			redis_client,
-			task_id="task-abc",
-			stage="batch_upserted",
-			ttl_seconds=180,
-			cars_upserted=10,
-		)
-
-		redis_client.pipeline.assert_called_once()
-		self.assertEqual(pipe.set.call_count, 2)
-		first_call = pipe.set.call_args_list[0]
-		second_call = pipe.set.call_args_list[1]
-
-		self.assertEqual(first_call.args[0], tasks._task_progress_key("task-abc"))
-		self.assertEqual(second_call.args[0], tasks.GLOBAL_PROGRESS_TS_KEY)
-		pipe.execute.assert_called_once()
-
-	def test_streaming_sync_stops_cleanly_when_page_stays_empty(self) -> None:
-		scraper = self._make_scraper()
-		scraper.settings.celery.batch_size = 50
-
-		page = MagicMock()
-		empty_page_result = SimpleNamespace(
-			page_number=1,
-			vehicle_links=[],
-			next_page_detected=True,
-		)
-
-		scraper._open_listing_for_stream = MagicMock(return_value=(
-			page,
-			{"make": None, "model": None, "year_min": None, "year_max": None},
-		))
-		scraper.listing_collector.collect_current_page = MagicMock(return_value=empty_page_result)
-		scraper._recover_empty_listing_page = MagicMock(return_value=(empty_page_result, []))
-		scraper.sync_batch = MagicMock()
-
-		result = scraper._sync_listing_streaming(
-			make=None,
-			model=None,
-			lane="iaai_cars",
-			limit=None,
-			effective_only_new=False,
-			started_at=0.0,
-			listing_url="https://www.iaai.com/Vehiclelisting/Cars?Make=TEST",
-		)
-
-		self.assertEqual(result["total"], 0)
-		self.assertEqual(result["cars_upserted"], 0)
-		self.assertEqual(result["cars_failed"], 0)
-		self.assertEqual(result["listing"]["pages_collected"], 1)
-		scraper._recover_empty_listing_page.assert_called_once()
-		scraper.sync_batch.assert_not_called()
-
-
-if __name__ == "__main__":
-	unittest.main()

tests/test_scraper.py

@@ -1,9 +1,8 @@
 from __future__ import annotations
 
 import unittest
-from concurrent.futures import TimeoutError as FuturesTimeoutError
 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 from iaai_scraper.core.config import Settings
 from iaai_scraper.core.exceptions import AntiBotDetectedError, SiteStructureChangedError
@@ -269,51 +268,6 @@ class TestScraperSync(unittest.TestCase):
         self.assertEqual(result["cars_upserted"], 1)
         self.assertEqual(result["total"], 1)
 
-    def test_sync_batch_timeout_does_not_duplicate_already_processed_urls(self) -> None:
-        scraper = self._make_scraper()
-        scraper.persistence.upsert_cars_batch = MagicMock(return_value={
-            "inserted": 1,
-            "updated": 0,
-            "images_upserted": 0,
-        })
-
-        urls = [
-            "https://www.iaai.com/VehicleDetail/111~US",
-            "https://www.iaai.com/VehicleDetail/222~US",
-            "https://www.iaai.com/VehicleDetail/333~US",
-        ]
-        first_record = CarRecord.model_validate(make_db_record("111"))
-
-        class _FakeExecutor:
-            def __init__(self, *args, **kwargs):
-                pass
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, exc_type, exc, tb):
-                return False
-
-            def map(self, fn, iterable, timeout=None):  # noqa: ARG002
-                yield 0, first_record
-                raise FuturesTimeoutError()
-
-        with patch("iaai_scraper.core.runtime_config.RuntimeFiltersConfig.is_empty", return_value=True), \
-             patch("iaai_scraper.scraper.ThreadPoolExecutor", _FakeExecutor), \
-             patch.object(scraper, "_browser_fallback_parallel", return_value={
-                 "records": [],
-                 "failures": [],
-                 "cars_failed": 0,
-                 "protection_events": 0,
-             }) as fallback_mock:
-            result = scraper.sync_batch(urls)
-
-        self.assertEqual(result["cars_upserted"], 1)
-        fallback_urls = fallback_mock.call_args.args[0]
-        self.assertEqual(len(fallback_urls), 2)
-        self.assertEqual({u for u, _ in fallback_urls}, {urls[1], urls[2]})
-        self.assertNotIn(urls[0], {u for u, _ in fallback_urls})
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_self_heal.py

@@ -1,81 +0,0 @@
-from __future__ import annotations
-
-import unittest
-from unittest.mock import MagicMock, patch
-
-from iaai_scraper.worker import self_heal
-
-
-class TestSelfHeal(unittest.TestCase):
-	def test_read_last_progress_ts_uses_global_key(self) -> None:
-		redis_client = MagicMock()
-		redis_client.get.return_value = "1776800000"
-
-		ts = self_heal._read_last_progress_ts(redis_client)
-
-		self.assertEqual(ts, 1776800000)
-		redis_client.scan_iter.assert_not_called()
-
-	def test_read_last_progress_ts_fallbacks_to_task_progress_keys(self) -> None:
-		redis_client = MagicMock()
-		redis_client.get.side_effect = lambda key: {
-			self_heal.GLOBAL_PROGRESS_TS_KEY: None,
-			"iaai:state:task_progress:a": '{"ts": 100}',
-			"iaai:state:task_progress:b": '{"ts": 250}',
-			"iaai:state:task_progress:c": '{"ts": 150}',
-		}.get(key)
-		redis_client.scan_iter.return_value = [
-			"iaai:state:task_progress:a",
-			"iaai:state:task_progress:b",
-			"iaai:state:task_progress:c",
-		]
-
-		ts = self_heal._read_last_progress_ts(redis_client)
-
-		self.assertEqual(ts, 250)
-
-	def test_read_last_progress_ts_ignores_broken_payloads(self) -> None:
-		redis_client = MagicMock()
-		redis_client.get.side_effect = lambda key: {
-			self_heal.GLOBAL_PROGRESS_TS_KEY: None,
-			"iaai:state:task_progress:a": "{bad-json}",
-			"iaai:state:task_progress:b": '{"foo": "bar"}',
-		}.get(key)
-		redis_client.scan_iter.return_value = [
-			"iaai:state:task_progress:a",
-			"iaai:state:task_progress:b",
-		]
-
-		ts = self_heal._read_last_progress_ts(redis_client)
-
-		self.assertIsNone(ts)
-
-	@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
-	@patch("iaai_scraper.worker.self_heal.os.kill")
-	@patch("builtins.open")
-	def test_kill_worker_process_sends_term_and_kill(self, open_mock, kill_mock, _sleep_mock) -> None:
-		open_mock.return_value.__enter__.return_value.read.return_value = "123"
-		# SIGTERM -> process alive check (pid,0) -> SIGKILL
-		kill_mock.side_effect = [None, None, None]
-
-		self_heal._kill_worker_process()
-
-		self.assertEqual(kill_mock.call_args_list[0].args[0], 123)
-		self.assertEqual(kill_mock.call_args_list[1].args, (123, 0))
-		self.assertEqual(kill_mock.call_args_list[2].args[0], 123)
-
-	@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
-	@patch("iaai_scraper.worker.self_heal.os.kill")
-	@patch("builtins.open")
-	def test_kill_worker_process_skips_sigkill_when_already_exited(self, open_mock, kill_mock, _sleep_mock) -> None:
-		open_mock.return_value.__enter__.return_value.read.return_value = "123"
-		kill_mock.side_effect = [None, ProcessLookupError()]
-
-		self_heal._kill_worker_process()
-
-		# Только SIGTERM и проверка существования процесса.
-		self.assertEqual(len(kill_mock.call_args_list), 2)
-
-
-if __name__ == "__main__":
-	unittest.main()

tests/test_worker_tasks.py

@@ -1,12 +1,9 @@
 from __future__ import annotations
 
-import json
-from dataclasses import replace
 import unittest
 from unittest.mock import MagicMock, patch
 
 from iaai_scraper.worker import tasks
-from iaai_scraper.core.config import settings as base_settings
 
 
 class TestWorkerTaskLockHelpers(unittest.TestCase):
@@ -70,7 +67,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
             get_redis.return_value = redis_client
             stop_event = MagicMock()
             heartbeat_thread = MagicMock()
-            start_heartbeat.return_value = (stop_event, heartbeat_thread)
+            start_heartbeat.return_value = (stop_event, MagicMock(), heartbeat_thread)
 
             tasks.sync_listing_task.push_request(id="task-123")
             try:
@@ -135,7 +132,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
             get_redis.return_value = redis_client
             stop_event = MagicMock()
             heartbeat_thread = MagicMock()
-            start_heartbeat.return_value = (stop_event, heartbeat_thread)
+            start_heartbeat.return_value = (stop_event, MagicMock(), heartbeat_thread)
 
             tasks.sync_listing_task.push_request(id="task-456")
             try:
@@ -200,7 +197,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
                 "1" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
             )
             get_redis.return_value = redis_client
-            start_heartbeat.return_value = (MagicMock(), MagicMock())
+            start_heartbeat.return_value = (MagicMock(), MagicMock(), MagicMock())
 
             sync_segmented_mock = MagicMock(return_value={
                 "run_id": 11, "status": "success", "full_scan_completed": True,
@@ -243,7 +240,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
                 "0" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
             )
             get_redis.return_value = redis_client
-            start_heartbeat.return_value = (MagicMock(), MagicMock())
+            start_heartbeat.return_value = (MagicMock(), MagicMock(), MagicMock())
 
             sync_segmented_mock = MagicMock(return_value={
                 "run_id": 14, "status": "success", "full_scan_completed": True,
@@ -284,7 +281,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
                 "99" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
             )
             get_redis.return_value = redis_client
-            start_heartbeat.return_value = (MagicMock(), MagicMock())
+            start_heartbeat.return_value = (MagicMock(), MagicMock(), MagicMock())
 
             sync_segmented_mock = MagicMock(return_value={
                 "run_id": 15, "status": "success", "full_scan_completed": True,
@@ -330,7 +327,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
             redis_client = MagicMock()
             redis_client.get.return_value = None
             get_redis.return_value = redis_client
-            start_heartbeat.return_value = (MagicMock(), MagicMock())
+            start_heartbeat.return_value = (MagicMock(), MagicMock(), MagicMock())
 
             tasks.sync_listing_task.push_request(id="task-791")
             try:
@@ -406,7 +403,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
             redis_client = MagicMock()
             redis_client.get.return_value = None
             get_redis.return_value = redis_client
-            start_heartbeat.return_value = (MagicMock(), MagicMock())
+            start_heartbeat.return_value = (MagicMock(), MagicMock(), MagicMock())
 
             with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
                 tasks.sync_listing_task.push_request(id="task-900")
@@ -420,177 +417,6 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
         clear_pending.assert_called()
         task_app.send_task.assert_not_called()
 
-    def test_sync_listing_task_soft_timeout_deduplicates_continuation(self) -> None:
-        with patch.object(tasks, "_get_persistence") as get_persistence, \
-             patch.object(tasks, "_get_redis") as get_redis, \
-             patch.object(tasks, "_acquire_lock", return_value=True), \
-             patch.object(tasks, "_is_full_scan_done", return_value=True), \
-             patch.object(tasks, "_start_lock_heartbeat", return_value=(MagicMock(), MagicMock())), \
-             patch.object(tasks, "_release_lock_if_owner"), \
-             patch.object(tasks, "_run_browser_job", side_effect=tasks.SoftTimeLimitExceeded()), \
-             patch.object(tasks, "_try_set_followup_pending", return_value=False) as set_pending, \
-             patch.object(tasks.sync_listing_task, "update_state"):
-            get_persistence.return_value = MagicMock()
-            redis_client = MagicMock()
-            redis_client.get.return_value = None
-            get_redis.return_value = redis_client
-
-            with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
-                tasks.sync_listing_task.push_request(id="task-soft-timeout")
-                try:
-                    result = tasks.sync_listing_task.run(make="Toyota")
-                finally:
-                    tasks.sync_listing_task.pop_request()
-
-        self.assertEqual(result["status"], "timed_out")
-        set_pending.assert_called_once()
-        task_app.send_task.assert_not_called()
-
-    def test_sync_listing_task_stops_immediate_bootstrap_continuation_after_limit(self) -> None:
-        with patch.object(tasks, "_get_persistence") as get_persistence, \
-             patch.object(tasks, "_get_redis") as get_redis, \
-             patch.object(tasks, "_acquire_lock", return_value=True), \
-             patch.object(tasks, "_is_full_scan_done", return_value=False), \
-             patch.object(tasks, "_start_lock_heartbeat", return_value=(MagicMock(), MagicMock())), \
-             patch.object(tasks, "_release_lock_if_owner"), \
-             patch.object(tasks, "_try_set_followup_pending", return_value=True), \
-             patch.object(tasks, "_bump_bootstrap_continuation_streak", return_value=(999, False)), \
-             patch.object(tasks, "_set_full_scan_done") as set_full_scan_done, \
-             patch.object(tasks, "_clear_sync_checkpoint") as clear_checkpoint, \
-             patch.object(tasks.sync_listing_task, "update_state"), \
-             patch.object(tasks, "_run_browser_job", return_value={
-                 "run_id": 101,
-                 "status": "partial_success",
-                 "full_scan_completed": False,
-                 "cars_upserted": 2,
-                 "cars_failed": 0,
-                 "images_upserted": 1,
-                 "skipped_existing": 0,
-                 "elapsed_seconds": 1.0,
-                 "failures": [],
-                 "listing": {"vehicles_collected": 2},
-             }):
-            get_persistence.return_value = MagicMock()
-            redis_client = MagicMock()
-            redis_client.get.return_value = None
-            get_redis.return_value = redis_client
-
-            with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
-                tasks.sync_listing_task.push_request(id="task-breaker-stop")
-                try:
-                    result = tasks.sync_listing_task.run()
-                finally:
-                    tasks.sync_listing_task.pop_request()
-
-        self.assertEqual(result["status"], "partial_success")
-        # Сначала bootstrap помечается незавершённым, затем breaker переключает на hourly.
-        self.assertTrue(any(call.args == (redis_client, False) for call in set_full_scan_done.call_args_list))
-        self.assertTrue(any(call.args == (redis_client, True) for call in set_full_scan_done.call_args_list))
-        clear_checkpoint.assert_called_once()
-        task_app.send_task.assert_not_called()
-
-    def test_sync_listing_task_always_full_scan_forces_bootstrap_even_after_done(self) -> None:
-        with patch.object(tasks, "_get_persistence") as get_persistence, \
-             patch.object(tasks, "_get_redis") as get_redis, \
-             patch.object(
-                 tasks,
-                 "Settings",
-                 return_value=replace(
-                     base_settings,
-                     discovery=replace(base_settings.discovery, always_full_scan=True),
-                 ),
-             ), \
-             patch.object(tasks, "_acquire_lock", return_value=True), \
-             patch.object(tasks, "_is_full_scan_done", return_value=True), \
-             patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
-             patch.object(tasks, "_release_lock_if_owner") as release_lock, \
-             patch.object(tasks, "_set_full_scan_done") as set_full_scan_done, \
-             patch.object(tasks.sync_listing_task, "update_state"), \
-             patch.object(tasks, "_run_browser_job") as run_job, \
-             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=[]):
-            get_persistence.return_value = MagicMock()
-            redis_client = MagicMock()
-            redis_client.get.return_value = None
-            get_redis.return_value = redis_client
-            start_heartbeat.return_value = (MagicMock(), MagicMock())
-
-            run_job.return_value = {
-                "run_id": 17,
-                "status": "success",
-                "full_scan_completed": True,
-                "cars_upserted": 1,
-                "cars_failed": 0,
-                "images_upserted": 0,
-                "skipped_existing": 0,
-                "elapsed_seconds": 1.0,
-                "failures": [],
-            }
-
-            tasks.sync_listing_task.push_request(id="task-always-full")
-            try:
-                result = tasks.sync_listing_task.run()
-            finally:
-                tasks.sync_listing_task.pop_request()
-
-        self.assertEqual(result["status"], "success")
-        set_full_scan_done.assert_called_with(redis_client, False)
-        release_lock.assert_called_once()
-
-    def test_sync_listing_task_always_full_scan_uses_segmented_resume_path(self) -> None:
-        segments = [{"make": "TOYOTA"}, {"make": "FORD"}, {"make": "HONDA"}]
-        with patch.object(tasks, "_get_persistence") as get_persistence, \
-             patch.object(tasks, "_get_redis") as get_redis, \
-             patch.object(
-                 tasks,
-                 "Settings",
-                 return_value=replace(
-                     base_settings,
-                     discovery=replace(base_settings.discovery, always_full_scan=True),
-                 ),
-             ), \
-             patch.object(tasks, "_acquire_lock", return_value=True), \
-             patch.object(tasks, "_is_full_scan_done", return_value=True), \
-             patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
-             patch.object(tasks, "_release_lock_if_owner") as release_lock, \
-             patch.object(tasks, "_run_browser_job", side_effect=lambda fn: fn()), \
-             patch.object(tasks.sync_listing_task, "update_state"), \
-             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=segments):
-            get_persistence.return_value = MagicMock()
-            redis_client = MagicMock()
-            redis_client.get.side_effect = lambda key: (
-                "0" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
-            )
-            get_redis.return_value = redis_client
-            start_heartbeat.return_value = (MagicMock(), MagicMock())
-
-            sync_segmented_mock = MagicMock(return_value={
-                "run_id": 18,
-                "status": "success",
-                "full_scan_completed": True,
-                "cars_upserted": 1,
-                "cars_failed": 0,
-                "images_upserted": 0,
-                "skipped_existing": 0,
-                "elapsed_seconds": 1.0,
-                "failures": [],
-            })
-            scraper_ctx = MagicMock()
-            scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
-            scraper_ctx.__enter__.return_value.sync_listing = MagicMock()
-            scraper_ctx.__exit__.return_value = None
-
-            with patch.object(tasks, "IAAIScraper", return_value=scraper_ctx):
-                tasks.sync_listing_task.push_request(id="task-always-full-resume")
-                try:
-                    result = tasks.sync_listing_task.run()
-                finally:
-                    tasks.sync_listing_task.pop_request()
-
-        self.assertEqual(result["status"], "success")
-        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 1)
-        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_page"], 1)
-        release_lock.assert_called_once()
-
 
 if __name__ == "__main__":
     unittest.main()