iaai-parser/iaai_scraper/worker/tasks.py

# Задачи Celery для синхронизации автомобилей и листинга IAAI.

from concurrent.futures import ThreadPoolExecutor
import json
import logging
import os
import signal
from threading import Event, Thread
import time
import uuid

from billiard.exceptions import SoftTimeLimitExceeded
from celery import shared_task
from redis import Redis

from ..core.config import Settings, parse_listing_segments
from ..scraper import IAAIScraper
from ..storage.db import PersistenceService
from ..discovery import SitemapDiscoveryError, discover_vehicle_urls_from_sitemap_with_stats

logger = logging.getLogger("iaai_scraper.worker.tasks")

SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
SYNC_LISTING_CHECKPOINT_KEY = "iaai:state:sync_listing_checkpoint"
SYNC_LISTING_CHECKPOINT_TTL_SECONDS = 7 * 24 * 60 * 60
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY = "iaai:state:sync_listing_bootstrap_failure_streak"
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT = 3
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS = 24 * 60 * 60
SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
SYNC_LISTING_TASK_NAME = "iaai_scraper.worker.tasks.sync_listing_task"
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"
SYNC_SEGMENTS_PROGRESS_TTL_SECONDS = 24 * 60 * 60
TASK_PROGRESS_KEY_FMT = "iaai:state:task_progress:{task_id}"
SITEMAP_HOURLY_LAST_COUNT_KEY = "iaai:state:sitemap_hourly_last_count"
SITEMAP_HOURLY_REFRESH_OFFSET_KEY = "iaai:state:sitemap_hourly_refresh_offset"


def _retry_with_backoff(func, *, attempts: int = 5, base_delay_s: float = 1.0):
    last_exc: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            return func()
        except Exception as exc:
            last_exc = exc
            if attempt >= attempts:
                break
            delay = base_delay_s * (2 ** (attempt - 1))
            logger.warning(
                "Operation failed (attempt %d/%d): %s. Retrying in %.1fs",
                attempt,
                attempts,
                exc,
                delay,
            )
            time.sleep(delay)
    if last_exc is not None:
        raise last_exc


def _run_browser_job(func, *args, **kwargs):
    # Браузерный код запускаем в отдельном потоке без активного loop.
    # НЕ используем `with` — иначе shutdown(wait=True) заблокирует main thread
    # если SoftTimeLimitExceeded прервёт future.result(), а browser thread ещё работает.
    settings = Settings()
    soft = settings.celery.task_soft_time_limit
    hard = settings.celery.task_time_limit
    # Таймаут для future.result(): берём hard limit (или soft + 120), чтобы не висеть вечно.
    wait_timeout = min(hard, soft + 120) if soft and hard else None

    executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="iaai-browser")
    future = executor.submit(func, *args, **kwargs)
    try:
        result = future.result(timeout=wait_timeout)
    except SoftTimeLimitExceeded:
        # Отпускаем executor без ожидания — thread умрёт когда Celery убьёт процесс (hard limit).
        future.cancel()
        executor.shutdown(wait=False, cancel_futures=True)
        raise
    except TimeoutError:
        # future.result(timeout=...) вышел по таймауту — browser thread завис.
        future.cancel()
        executor.shutdown(wait=False, cancel_futures=True)
        raise SoftTimeLimitExceeded("Browser thread did not finish within time limit")
    except Exception:
        executor.shutdown(wait=False, cancel_futures=True)
        raise
    else:
        executor.shutdown(wait=True)
    return result


def _sync_listing_lock_ttl_seconds() -> int:
    settings = Settings()
    soft = settings.celery.task_soft_time_limit
    hard = settings.celery.task_time_limit
    # Используем clamped hard limit (soft + 120), а не сырой task_time_limit,
    # чтобы lock не висел 11 дней при CELERY_TASK_TIME_LIMIT=999999.
    effective_hard = min(hard, soft + 120) if soft else hard
    return max(effective_hard + 120, 300)


def _sync_segment_lock_ttl_seconds() -> int:
    return 300


def _task_progress_key(task_id: str) -> str:
    return TASK_PROGRESS_KEY_FMT.format(task_id=task_id)


def _update_task_progress(
    redis_client: Redis,
    *,
    task_id: str,
    stage: str,
    ttl_seconds: int,
    **payload,
) -> None:
    try:
        data = {
            "task_id": task_id,
            "stage": stage,
            "ts": int(time.time()),
            **payload,
        }
        redis_client.set(
            _task_progress_key(task_id),
            json.dumps(data, ensure_ascii=False),
            ex=max(60, int(ttl_seconds)),
        )
    except Exception:
        logger.warning("Failed to update task progress for %s", task_id, exc_info=True)


def _clear_task_progress(redis_client: Redis, task_id: str) -> None:
    try:
        redis_client.delete(_task_progress_key(task_id))
    except Exception:
        logger.warning("Failed to clear task progress for %s", task_id, exc_info=True)


def _hourly_sitemap_diff_sync(*, lane: str, limit: int | None, only_new: bool | None) -> dict[str, object]:
    del limit, only_new
    settings = Settings()
    persistence = _get_persistence()
    persistence.create_tables()

    discovery_result = discover_vehicle_urls_from_sitemap_with_stats(settings=settings)
    discovered_urls = discovery_result.vehicle_urls
    active_urls = set(discovered_urls)
    existing_urls = persistence.get_all_active_origin_urls_for_lane("iaai:")

    new_urls = [url for url in discovered_urls if url not in existing_urls]
    sold_count = 0
    if active_urls:
        sold_count = persistence.mark_sold_not_in_listing_by_urls(active_urls, lane="iaai")

    cars_upserted = 0
    cars_failed = 0
    images_upserted = 0
    failures: list[dict[str, str]] = []

    if new_urls:
        with IAAIScraper(settings) as scraper:
            batch_size = settings.celery.batch_size
            for batch_start in range(0, len(new_urls), batch_size):
                batch_urls = new_urls[batch_start:batch_start + batch_size]
                batch_result = scraper.sync_batch(batch_urls, lane=lane)
                cars_upserted += int(batch_result.get("cars_upserted", 0))
                cars_failed += int(batch_result.get("cars_failed", 0))
                images_upserted += int(batch_result.get("images_upserted", 0))
                failures.extend(batch_result.get("failures", []))

    status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
    return {
        "status": status,
        "run_id": None,
        "cars_upserted": cars_upserted,
        "cars_failed": cars_failed,
        "images_upserted": images_upserted,
        "skipped_existing": len(discovered_urls) - len(new_urls),
        "elapsed_seconds": None,
        "failures": failures,
        "full_scan_completed": True,
        "hourly_mode": "sitemap_diff",
        "discovered_urls": len(discovered_urls),
        "new_urls": len(new_urls),
        "sold_marked": sold_count,
        "transport": discovery_result.stats.transport,
    }


def _hourly_sitemap_full_refresh_sync(*, lane: str, limit: int | None, only_new: bool | None) -> dict[str, object]:
    del limit, only_new
    settings = Settings()
    persistence = _get_persistence()
    persistence.create_tables()

    discovery_result = discover_vehicle_urls_from_sitemap_with_stats(settings=settings)
    discovered_urls = discovery_result.vehicle_urls
    active_urls = set(discovered_urls)
    sold_count = 0
    if active_urls:
        sold_count = persistence.mark_sold_not_in_listing_by_urls(active_urls, lane="iaai")

    cars_upserted = 0
    cars_failed = 0
    images_upserted = 0
    failures: list[dict[str, str]] = []

    with IAAIScraper(settings) as scraper:
        batch_size = settings.celery.batch_size
        for batch_start in range(0, len(discovered_urls), batch_size):
            batch_urls = discovered_urls[batch_start:batch_start + batch_size]
            batch_result = scraper.sync_batch(batch_urls, lane=lane)
            cars_upserted += int(batch_result.get("cars_upserted", 0))
            cars_failed += int(batch_result.get("cars_failed", 0))
            images_upserted += int(batch_result.get("images_upserted", 0))
            failures.extend(batch_result.get("failures", []))

    status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
    return {
        "status": status,
        "run_id": None,
        "cars_upserted": cars_upserted,
        "cars_failed": cars_failed,
        "images_upserted": images_upserted,
        "skipped_existing": 0,
        "elapsed_seconds": None,
        "failures": failures,
        "full_scan_completed": True,
        "hourly_mode": "sitemap_full_refresh",
        "discovered_urls": len(discovered_urls),
        "new_urls": None,
        "sold_marked": sold_count,
        "transport": discovery_result.stats.transport,
    }


def _hourly_sitemap_rolling_refresh_sync(
    *,
    redis_client: Redis,
    lane: str,
    limit: int | None,
    only_new: bool | None,
) -> dict[str, object]:
    del limit, only_new
    settings = Settings()
    persistence = _get_persistence()
    persistence.create_tables()

    discovery_result = discover_vehicle_urls_from_sitemap_with_stats(settings=settings)
    discovered_urls = discovery_result.vehicle_urls
    active_urls = set(discovered_urls)
    sold_count = 0
    if active_urls:
        sold_count = persistence.mark_sold_not_in_listing_by_urls(active_urls, lane="iaai")

    existing_urls = persistence.get_all_active_origin_urls_for_lane("iaai:")
    new_urls = [url for url in discovered_urls if url not in existing_urls]

    total_active = persistence.count_active_cars_for_lane("iaai:")
    batch_size = max(1, int(settings.discovery.hourly_refresh_batch_size))
    try:
        offset = int(redis_client.get(SITEMAP_HOURLY_REFRESH_OFFSET_KEY) or 0)
    except Exception:
        offset = 0

    refresh_urls = persistence.get_active_origin_urls_batch_for_refresh(
        prefix="iaai:",
        offset=offset,
        limit=batch_size,
    )
    if not refresh_urls and total_active > 0:
        offset = 0
        refresh_urls = persistence.get_active_origin_urls_batch_for_refresh(
            prefix="iaai:",
            offset=0,
            limit=batch_size,
        )

    next_offset = 0
    if total_active > 0:
        next_offset = offset + len(refresh_urls)
        if next_offset >= total_active:
            next_offset = 0

    try:
        redis_client.set(SITEMAP_HOURLY_REFRESH_OFFSET_KEY, str(next_offset))
    except Exception:
        logger.warning("Failed to persist hourly rolling refresh offset", exc_info=True)

    seen: set[str] = set()
    target_urls: list[str] = []
    for url in new_urls + refresh_urls:
        if url in seen:
            continue
        seen.add(url)
        target_urls.append(url)

    cars_upserted = 0
    cars_failed = 0
    images_upserted = 0
    failures: list[dict[str, str]] = []

    if target_urls:
        with IAAIScraper(settings) as scraper:
            worker_batch_size = settings.celery.batch_size
            for batch_start in range(0, len(target_urls), worker_batch_size):
                batch_urls = target_urls[batch_start:batch_start + worker_batch_size]
                batch_result = scraper.sync_batch(batch_urls, lane=lane)
                cars_upserted += int(batch_result.get("cars_upserted", 0))
                cars_failed += int(batch_result.get("cars_failed", 0))
                images_upserted += int(batch_result.get("images_upserted", 0))
                failures.extend(batch_result.get("failures", []))

    status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
    return {
        "status": status,
        "run_id": None,
        "cars_upserted": cars_upserted,
        "cars_failed": cars_failed,
        "images_upserted": images_upserted,
        "skipped_existing": max(0, len(discovered_urls) - len(new_urls)),
        "elapsed_seconds": None,
        "failures": failures,
        "full_scan_completed": True,
        "hourly_mode": "sitemap_rolling_refresh",
        "discovered_urls": len(discovered_urls),
        "new_urls": len(new_urls),
        "refresh_urls": len(refresh_urls),
        "sold_marked": sold_count,
        "transport": discovery_result.stats.transport,
        "refresh_offset": offset,
        "refresh_next_offset": next_offset,
        "active_total": total_active,
    }


def _start_stall_watchdog(
    redis_client: Redis,
    *,
    task_id: str,
    stall_timeout_seconds: int,
) -> tuple[Event, Thread]:
    stop_event = Event()
    interval_seconds = max(5.0, min(30.0, stall_timeout_seconds / 3))

    def _watchdog() -> None:
        key = _task_progress_key(task_id)
        while not stop_event.wait(interval_seconds):
            try:
                raw = redis_client.get(key)
                if not raw:
                    continue
                data = json.loads(raw)
                last_ts = int(data.get("ts") or 0)
                if not last_ts:
                    continue
                age = int(time.time()) - last_ts
                if age < stall_timeout_seconds:
                    continue
                logger.error(
                    "Task %s stalled for %ss at stage=%s payload=%s; killing worker process for redelivery",
                    task_id,
                    age,
                    data.get("stage"),
                    data,
                )
            except Exception:
                logger.warning("Failed to inspect task progress for stall watchdog", exc_info=True)
                continue
            os.kill(os.getpid(), signal.SIGKILL)

    thread = Thread(target=_watchdog, name=f"task-stall-watchdog-{task_id[:8]}", daemon=True)
    thread.start()
    return stop_event, thread


def _get_persistence() -> PersistenceService:
    settings = Settings()
    persistence = PersistenceService(settings)

    def _ping_db() -> None:
        with persistence.engine.connect() as conn:
            conn.exec_driver_sql("SELECT 1")

    _retry_with_backoff(_ping_db, attempts=5, base_delay_s=1.0)
    return persistence


def _get_redis() -> Redis:
    settings = Settings()
    redis_client = Redis.from_url(
        settings.redis.url,
        decode_responses=True,
        socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
        socket_timeout=settings.redis.socket_timeout_seconds,
        health_check_interval=settings.redis.health_check_interval_seconds,
        retry_on_timeout=True,
    )

    def _ping_redis() -> None:
        redis_client.ping()

    _retry_with_backoff(_ping_redis, attempts=5, base_delay_s=1.0)
    return redis_client


def _acquire_lock(redis_client: Redis, key: str, owner_token: str, ttl_seconds: int) -> bool:
    try:
        acquired = bool(redis_client.set(key, owner_token, nx=True, ex=ttl_seconds))
        if acquired:
            return True

        # Автовосстановление: если lock завис без TTL, считаем stale и пересоздаём.
        ttl = redis_client.ttl(key)
        if ttl is not None and ttl < 0:
            logger.warning("Detected stale lock without TTL, removing: %s", key)
            redis_client.delete(key)
            return bool(redis_client.set(key, owner_token, nx=True, ex=ttl_seconds))

        return False
    except Exception as exc:
        logger.warning("Failed to acquire lock %s", key, exc_info=True)
        return False


def _refresh_lock_if_owner(redis_client: Redis, key: str, owner_token: str, ttl_seconds: int) -> bool | None:
    try:
        refreshed = redis_client.eval(
            """
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
            end
            return 0
            """,
            1,
            key,
            owner_token,
            int(ttl_seconds),
        )
        return bool(refreshed)
    except Exception as exc:
        logger.warning("Failed to refresh lock %s", key, exc_info=True)
        return None


def _release_lock_if_owner(redis_client: Redis, key: str, owner_token: str) -> None:
    try:
        redis_client.eval(
            """
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('DEL', KEYS[1])
            end
            return 0
            """,
            1,
            key,
            owner_token,
        )
    except Exception as exc:
        logger.warning("Failed to release lock %s", key, exc_info=True)


def _has_running_sync_listing_tasks(celery_app, *, exclude_task_id: str | None = None) -> bool:
    try:
        inspector = celery_app.control.inspect(timeout=1.0)
        snapshots = [
            inspector.active() or {},
            inspector.reserved() or {},
            inspector.scheduled() or {},
        ]
    except Exception:
        logger.warning("Failed to inspect Celery workers for running sync tasks", exc_info=True)
        return True

    for snapshot in snapshots:
        for entries in snapshot.values():
            for entry in entries or []:
                task_name = str(entry.get("name") or entry.get("request", {}).get("name") or "")
                if task_name != SYNC_LISTING_TASK_NAME:
                    continue
                # Исключаем текущую задачу — она не считается "другой запущенной"
                entry_id = str(entry.get("id") or entry.get("request", {}).get("id") or "")
                if exclude_task_id and entry_id == exclude_task_id:
                    continue
                return True
    return False


def _clear_orphan_sync_listing_lock(redis_client: Redis, celery_app, *, current_task_id: str | None = None) -> bool:
    try:
        owner_token = redis_client.get(SYNC_LISTING_LOCK_KEY)
        if not owner_token:
            return False
    except Exception:
        logger.warning("Failed to read sync listing lock before cleanup", exc_info=True)
        return False

    if _has_running_sync_listing_tasks(celery_app, exclude_task_id=current_task_id):
        logger.info("sync_listing lock preserved: active task still detected")
        return False

    try:
        ttl = redis_client.ttl(SYNC_LISTING_LOCK_KEY)
        redis_client.delete(SYNC_LISTING_LOCK_KEY)
        logger.warning(
            "Removed orphan sync_listing lock owner=%s ttl=%s after worker restart",
            owner_token,
            ttl,
        )
        return True
    except Exception:
        logger.warning("Failed to clear orphan sync listing lock", exc_info=True)
        return False


def _is_full_scan_done(redis_client: Redis) -> bool:
    try:
        value = redis_client.get(SYNC_FULL_SCAN_DONE_KEY)
    except Exception:
        logger.warning("Failed to read full scan state", exc_info=True)
        return False
    return str(value or "").strip() == "1"


def _set_full_scan_done(redis_client: Redis, done: bool) -> None:
    try:
        redis_client.set(SYNC_FULL_SCAN_DONE_KEY, "1" if done else "0")
    except Exception:
        logger.warning("Failed to persist full scan state", exc_info=True)


def _load_last_completed_segment(redis_client: Redis) -> int | None:
    # Segment-only checkpoint: хранит индекс последнего ПОЛНОСТЬЮ пройденного сегмента.
    try:
        raw = redis_client.get(SYNC_LISTING_CHECKPOINT_KEY)
    except Exception:
        logger.warning("Failed to read sync listing checkpoint", exc_info=True)
        return None
    if raw is None:
        return None
    try:
        value = int(str(raw).strip())
    except (TypeError, ValueError):
        logger.warning("Invalid sync listing checkpoint value %r; clearing", raw)
        _clear_sync_checkpoint(redis_client)
        return None
    if value < 0:
        _clear_sync_checkpoint(redis_client)
        return None
    return value


def _save_last_completed_segment(redis_client: Redis, segment_index: int) -> None:
    try:
        redis_client.set(
            SYNC_LISTING_CHECKPOINT_KEY,
            str(int(segment_index)),
            ex=SYNC_LISTING_CHECKPOINT_TTL_SECONDS,
        )
    except Exception:
        logger.warning("Failed to save sync listing checkpoint", exc_info=True)


def _clear_sync_checkpoint(redis_client: Redis) -> None:
    try:
        redis_client.delete(SYNC_LISTING_CHECKPOINT_KEY)
    except Exception:
        logger.warning("Failed to clear sync listing checkpoint", exc_info=True)


def _try_set_followup_pending(redis_client: Redis, *, ttl_seconds: int) -> bool:
    try:
        return bool(redis_client.set(SYNC_LISTING_FOLLOWUP_PENDING_KEY, "1", nx=True, ex=max(60, int(ttl_seconds))))
    except Exception:
        logger.warning("Failed to set follow-up pending flag", exc_info=True)
        return True


def _clear_followup_pending(redis_client: Redis) -> None:
    try:
        redis_client.delete(SYNC_LISTING_FOLLOWUP_PENDING_KEY)
    except Exception:
        logger.warning("Failed to clear follow-up pending flag", exc_info=True)


def _bump_bootstrap_failure_streak(
    redis_client: Redis,
    *,
    reason: str,
) -> tuple[int, bool]:
    try:
        streak = int(redis_client.incr(SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY))
        redis_client.expire(
            SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY,
            SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS,
        )
    except Exception:
        logger.warning("Failed to update bootstrap failure streak", exc_info=True)
        return 0, True

    should_enqueue = streak < SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT
    if should_enqueue:
        logger.warning(
            "Bootstrap failure streak %d/%d recorded (reason=%s)",
            streak,
            SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT,
            reason,
        )
    else:
        logger.error(
            "Bootstrap follow-up circuit breaker opened after %d consecutive failures (reason=%s)",
            streak,
            reason,
        )
    return streak, should_enqueue


def _clear_bootstrap_failure_streak(redis_client: Redis) -> None:
    try:
        redis_client.delete(SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY)
    except Exception:
        logger.warning("Failed to clear bootstrap failure streak", exc_info=True)


def _start_lock_heartbeat(
    redis_client: Redis,
    key: str,
    owner_token: str,
    ttl_seconds: int,
) -> tuple[Event, Thread]:
    stop_event = Event()
    interval_seconds = max(5.0, min(30.0, ttl_seconds / 3))

    def _heartbeat() -> None:
        while not stop_event.wait(interval_seconds):
            refreshed = _refresh_lock_if_owner(redis_client, key, owner_token, ttl_seconds)
            if refreshed is False:
                logger.warning("Lost sync_listing lock ownership for %s", owner_token)
                return

    thread = Thread(target=_heartbeat, name="sync-listing-lock-heartbeat", daemon=True)
    thread.start()
    return stop_event, thread


def _reset_segments_progress(redis_client: Redis, total: int) -> None:
    try:
        pipe = redis_client.pipeline()
        pipe.delete(SYNC_SEGMENTS_PROGRESS_KEY)
        pipe.set(SYNC_SEGMENTS_TOTAL_KEY, str(int(total)), ex=SYNC_SEGMENTS_PROGRESS_TTL_SECONDS)
        pipe.execute()
    except Exception:
        logger.warning("Failed to reset segments progress", exc_info=True)


def _mark_segment_completed(redis_client: Redis, segment_index: int) -> tuple[int, int]:
    """Помечает сегмент завершённым. Возвращает (completed_count, total)."""
    try:
        pipe = redis_client.pipeline()
        pipe.sadd(SYNC_SEGMENTS_PROGRESS_KEY, str(int(segment_index)))
        pipe.expire(SYNC_SEGMENTS_PROGRESS_KEY, SYNC_SEGMENTS_PROGRESS_TTL_SECONDS)
        pipe.scard(SYNC_SEGMENTS_PROGRESS_KEY)
        pipe.get(SYNC_SEGMENTS_TOTAL_KEY)
        results = pipe.execute()
        completed = int(results[2] or 0)
        total = int(results[3] or 0) if results[3] else 0
        return completed, total
    except Exception:
        logger.warning("Failed to mark segment %d completed", segment_index, exc_info=True)
        return 0, 0


@shared_task(
    name="iaai_scraper.worker.tasks.sync_segment_task",
    bind=True,
    max_retries=2,
    default_retry_delay=60,
    acks_late=True,
)
def sync_segment_task(
    self,
    segment_index: int,
    segment: dict,
    lane: str = "iaai_cars",
    only_new: bool | None = None,
    is_bootstrap: bool = False,
):
    """Обработка одного сегмента листинга. Запускается параллельно несколькими воркерами."""
    persistence = _get_persistence()
    persistence.create_tables()
    task_id = self.request.id or "unknown"
    redis_client = _get_redis()
    settings = Settings()

    # Per-segment lock — защита от случайного дубля
    seg_lock_key = SYNC_SEGMENT_LOCK_KEY_FMT.format(idx=int(segment_index))
    owner_token = f"{task_id}:{uuid.uuid4().hex}"
    lock_ttl = _sync_segment_lock_ttl_seconds()
    lock_acquired = _acquire_lock(redis_client, seg_lock_key, owner_token, lock_ttl)
    if not lock_acquired:
        logger.info("sync_segment_task[%d] skipped: already running", segment_index)
        return {"status": "skipped", "segment_index": segment_index, "reason": "duplicate"}

    seg_make = segment.get("make")
    seg_year_min = segment.get("year_min")
    seg_year_max = segment.get("year_max")
    seg_label = f"{seg_make or 'ALL'}"
    if seg_year_min is not None or seg_year_max is not None:
        seg_label += f" ({seg_year_min}-{seg_year_max})"

    heartbeat_stop: Event | None = None
    heartbeat_thread: Thread | None = None
    watchdog_stop: Event | None = None
    watchdog_thread: Thread | None = None
    stall_timeout = max(120, int(settings.celery.task_stall_timeout_seconds))
    progress_ttl = max(lock_ttl + 120, stall_timeout + 120)

    _update_task_progress(
        redis_client,
        task_id=task_id,
        stage="segment_task_started",
        ttl_seconds=progress_ttl,
        segment_index=segment_index,
        segment_label=seg_label,
    )
    heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(redis_client, seg_lock_key, owner_token, lock_ttl)
    watchdog_stop, watchdog_thread = _start_stall_watchdog(
        redis_client,
        task_id=task_id,
        stall_timeout_seconds=stall_timeout,
    )

    try:
        def _job():
            with IAAIScraper() as scraper:
                scraper.set_progress_callback(
                    lambda stage, meta: _update_task_progress(
                        redis_client,
                        task_id=task_id,
                        stage=stage,
                        ttl_seconds=progress_ttl,
                        segment_index=segment_index,
                        segment_label=seg_label,
                        **meta,
                    )
                )
                base_url = scraper.settings.listing.cars_url
                seg_url = scraper._build_segment_listing_url(base_url, seg_make) if seg_make else None
                return scraper.sync_listing(
                    make=None if seg_url else seg_make,
                    model=None,
                    lane=lane,
                    only_new=only_new,
                    listing_url=seg_url,
                    year_min=seg_year_min,
                    year_max=seg_year_max,
                    skip_mark_sold=True,
                )

        result = _run_browser_job(_job)
        segment_done = bool(result.get("full_scan_completed", False))
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="segment_task_completed",
            ttl_seconds=progress_ttl,
            segment_index=segment_index,
            segment_label=seg_label,
            status=result.get("status", "success"),
            cars_upserted=result.get("cars_upserted", 0),
            cars_failed=result.get("cars_failed", 0),
        )

        if is_bootstrap and segment_done:
            completed, total = _mark_segment_completed(redis_client, segment_index)
            logger.warning(
                "Segment %d (%s) bootstrap done: %d/%d completed",
                segment_index, seg_label, completed, total,
            )
            if total > 0 and completed >= total:
                _set_full_scan_done(redis_client, True)
                _clear_sync_checkpoint(redis_client)
                _clear_bootstrap_failure_streak(redis_client)
                logger.warning("All %d segments completed; bootstrap full scan done", total)

        return {
            "status": result.get("status", "success"),
            "segment_index": segment_index,
            "segment_label": seg_label,
            "cars_upserted": result.get("cars_upserted", 0),
            "cars_failed": result.get("cars_failed", 0),
            "vehicles_collected": result.get("listing", {}).get("vehicles_collected", 0),
            "full_scan_completed": segment_done,
        }

    except SoftTimeLimitExceeded:
        logger.warning("sync_segment_task[%d] soft timeout — partial progress saved", segment_index)
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="segment_task_soft_timeout",
            ttl_seconds=progress_ttl,
            segment_index=segment_index,
            segment_label=seg_label,
        )
        return {
            "status": "timed_out",
            "segment_index": segment_index,
            "segment_label": seg_label,
        }
    except Exception as exc:
        logger.error("sync_segment_task[%d] failed: %s — %s", segment_index, seg_label, exc, exc_info=True)
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="segment_task_failed",
            ttl_seconds=progress_ttl,
            segment_index=segment_index,
            segment_label=seg_label,
            error=str(exc),
        )
        try:
            raise self.retry(exc=exc)
        except self.MaxRetriesExceededError:
            return {
                "status": "failed",
                "segment_index": segment_index,
                "segment_label": seg_label,
                "error": str(exc),
            }
    finally:
        if watchdog_stop is not None:
            watchdog_stop.set()
        if watchdog_thread is not None:
            watchdog_thread.join(timeout=1)
        if heartbeat_stop is not None:
            heartbeat_stop.set()
        if heartbeat_thread is not None:
            heartbeat_thread.join(timeout=1)
        _clear_task_progress(redis_client, task_id)
        _release_lock_if_owner(redis_client, seg_lock_key, owner_token)


@shared_task(
    name="iaai_scraper.worker.tasks.sync_vehicle_task",
    bind=True,
    max_retries=2,
    default_retry_delay=30,
    acks_late=True,
)
def sync_vehicle_task(self, vehicle_url: str, lane: str = "iaai"):
    # Скрапинг и upsert одного автомобиля.
    persistence = _get_persistence()
    persistence.create_tables()

    try:
        def _job():
            with IAAIScraper() as scraper:
                return scraper.sync_vehicle(vehicle_url, lane=lane)

        result = _run_browser_job(_job)
        logger.info("sync_vehicle_task completed: %s", vehicle_url)
        return {
            "status": "success",
            "vehicle_url": vehicle_url,
            "db_action": result.get("db_action"),
            "images_upserted": result.get("images_upserted", 0),
        }

    except Exception as exc:
        logger.error("sync_vehicle_task failed: %s — %s", vehicle_url, exc, exc_info=True)
        raise self.retry(exc=exc)


@shared_task(
    name="iaai_scraper.worker.tasks.sync_listing_task",
    bind=True,
    max_retries=3,
    default_retry_delay=120,
    acks_late=True,
)
def sync_listing_task(
    self,
    make: str | None = None,
    model: str | None = None,
    lane: str = "iaai_cars",
    limit: int | None = None,
    only_new: bool | None = None,
):
    # Полный цикл: листинг + sync всех найденных машин.
    persistence = _get_persistence()
    persistence.create_tables()
    task_id = self.request.id or "unknown"
    owner_token = f"{task_id}:{uuid.uuid4().hex}"
    redis_client = _get_redis()

    lock_acquired = False
    lock_ttl = _sync_listing_lock_ttl_seconds()
    heartbeat_stop: Event | None = None
    heartbeat_thread: Thread | None = None
    force_bootstrap_full_scan = False

    def _enqueue_bootstrap_followup(
        reason: str,
        delay_seconds: int = 5,
        *,
        count_as_failure: bool = False,
    ) -> None:
        flag_ttl = max(lock_ttl, delay_seconds + 300)
        if not _try_set_followup_pending(redis_client, ttl_seconds=flag_ttl):
            logger.info(
                "Bootstrap follow-up already pending; skip enqueue (reason=%s)",
                reason,
            )
            return
        if count_as_failure:
            _, should_enqueue = _bump_bootstrap_failure_streak(redis_client, reason=reason)
            if not should_enqueue:
                _clear_followup_pending(redis_client)
                return
        else:
            _clear_bootstrap_failure_streak(redis_client)
        try:
            self.app.send_task(
                "iaai_scraper.worker.tasks.sync_listing_task",
                kwargs={
                    "make": make,
                    "model": model,
                    "lane": lane,
                    "limit": limit,
                    "only_new": only_new,
                },
                queue="scraping",
                countdown=max(0, int(delay_seconds)),
            )
            logger.info(
                "Bootstrap follow-up sync queued in %ss (reason=%s)",
                delay_seconds,
                reason,
            )
        except Exception:
            _clear_followup_pending(redis_client)
            logger.warning("Failed to enqueue bootstrap follow-up sync", exc_info=True)

    lock_acquired = _acquire_lock(redis_client, SYNC_LISTING_LOCK_KEY, owner_token, lock_ttl)

    if not lock_acquired:
        orphan_cleared = _clear_orphan_sync_listing_lock(redis_client, self.app, current_task_id=task_id)
        if orphan_cleared:
            lock_acquired = _acquire_lock(redis_client, SYNC_LISTING_LOCK_KEY, owner_token, lock_ttl)

    if not lock_acquired:
        logger.info("sync_listing_task skipped: another sync is already running")
        return {
            "status": "skipped",
            "reason": "sync_already_running",
            "task_id": task_id,
        }

    try:
        settings = Settings()
        full_scan_done_before_run = _is_full_scan_done(redis_client)
        force_bootstrap_full_scan = not full_scan_done_before_run
        effective_limit = None if force_bootstrap_full_scan else limit
        effective_only_new = False if force_bootstrap_full_scan else only_new
        hourly_mode = settings.discovery.hourly_mode.strip().lower()
        discovery_mode = settings.discovery.mode.strip().lower()
        prefer_sitemap_mainline = (
            make is None
            and model is None
            and effective_limit is None
            and not effective_only_new
        )
        use_hourly_sitemap_sync = full_scan_done_before_run and prefer_sitemap_mainline

        # Segment-level checkpoint: хранит индекс последнего ПОЛНОСТЬЮ пройденного сегмента.
        # Используется только во время bootstrap для пропуска уже обработанных сегментов.
        # Никаких page-level resume — внутри сегмента всегда стартуем с page 1.
        last_completed_segment: int | None = None
        if force_bootstrap_full_scan:
            last_completed_segment = _load_last_completed_segment(redis_client)
        else:
            # После завершения bootstrap чекпоинт не нужен никогда.
            _clear_sync_checkpoint(redis_client)

        if force_bootstrap_full_scan:
            logger.info(
                "Bootstrap mode: forcing full scan (only_new=False, limit=None) until first complete run",
            )

        logger.info(
            "sync_listing options: only_new=%s, limit=%s",
            effective_only_new,
            effective_limit,
        )

        if use_hourly_sitemap_sync:
            if hourly_mode == "diff":
                logger.info("Hourly mode: running sitemap diff sync instead of full listing traversal")
                result = _hourly_sitemap_diff_sync(
                    lane=lane,
                    limit=effective_limit,
                    only_new=effective_only_new,
                )
            elif hourly_mode == "full_refresh":
                logger.info("Hourly mode: running sitemap full refresh of all active vehicles")
                result = _hourly_sitemap_full_refresh_sync(
                    lane=lane,
                    limit=effective_limit,
                    only_new=effective_only_new,
                )
            else:
                logger.info("Hourly mode: running sitemap rolling refresh of active vehicles")
                result = _hourly_sitemap_rolling_refresh_sync(
                    redis_client=redis_client,
                    lane=lane,
                    limit=effective_limit,
                    only_new=effective_only_new,
                )
            try:
                redis_client.set(SITEMAP_HOURLY_LAST_COUNT_KEY, str(result.get("discovered_urls", 0)))
            except Exception:
                logger.warning("Failed to persist hourly sitemap count", exc_info=True)

            summary = {
                "task_id": task_id,
                "run_id": result.get("run_id"),
                "status": result.get("status", "success"),
                "cars_upserted": result.get("cars_upserted", 0),
                "cars_failed": result.get("cars_failed", 0),
                "images_upserted": result.get("images_upserted", 0),
                "skipped_existing": result.get("skipped_existing", 0),
                "elapsed_seconds": result.get("elapsed_seconds"),
                "failures_count": len(result.get("failures") or []),
                "hourly_mode": result.get("hourly_mode"),
                "discovered_urls": result.get("discovered_urls", 0),
                "new_urls": result.get("new_urls", 0),
                "sold_marked": result.get("sold_marked", 0),
            }
            logger.info(
                "sync_listing_task hourly diff completed: status=%s, new=%d, sold=%d, skipped=%d",
                summary["status"],
                summary["new_urls"],
                summary["sold_marked"],
                summary["skipped_existing"],
            )
            return summary

        _clear_followup_pending(redis_client)

        heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(
            redis_client,
            SYNC_LISTING_LOCK_KEY,
            owner_token,
            lock_ttl,
        )
        self.update_state(state="STARTED", meta={"stage": "sync_listing_started", "task_id": task_id})

        # Определяем сегменты из конфига.
        segments = parse_listing_segments(settings.listing.listing_segments_json)
        use_segmented = (
            bool(segments)
            and make is None
            and model is None
            and not prefer_sitemap_mainline
            and discovery_mode != "sitemap"
        )

        if prefer_sitemap_mainline:
            if discovery_mode != "sitemap":
                logger.warning(
                    "Unfiltered full scan forcing sitemap discovery despite IAAI_DISCOVERY_MODE=%s",
                    discovery_mode or "unset",
                )
            if segments:
                logger.info("Ignoring configured listing segments for unfiltered sitemap full scan")

        # --- Параллельный диспатч сегментов: dispatch & exit ---
        if use_segmented and settings.celery.parallel_segments:
            # Сегменты уже завершённые (для bootstrap resume) пропускаем по Redis SET.
            already_completed: set[int] = set()
            if force_bootstrap_full_scan:
                try:
                    raw = redis_client.smembers(SYNC_SEGMENTS_PROGRESS_KEY) or set()
                    already_completed = {int(x) for x in raw if str(x).strip().lstrip("-").isdigit()}
                except Exception:
                    already_completed = set()

            pending = [
                (idx, seg) for idx, seg in enumerate(segments)
                if idx not in already_completed
            ]

            if not pending:
                # Всё уже сделано — фиксируем bootstrap done.
                if force_bootstrap_full_scan:
                    _set_full_scan_done(redis_client, True)
                    _clear_sync_checkpoint(redis_client)
                    _clear_bootstrap_failure_streak(redis_client)
                logger.warning("Parallel segments: nothing to dispatch (all completed)")
                return {
                    "status": "success",
                    "task_id": task_id,
                    "mode": "parallel_segments",
                    "segments_total": len(segments),
                    "segments_dispatched": 0,
                    "segments_already_completed": len(already_completed),
                }

            # При первом запуске bootstrap фиксируем total, чтобы знать когда остановиться.
            if force_bootstrap_full_scan and not already_completed:
                _reset_segments_progress(redis_client, len(segments))

            dispatched = 0
            for idx, seg in pending:
                try:
                    self.app.send_task(
                        "iaai_scraper.worker.tasks.sync_segment_task",
                        kwargs={
                            "segment_index": idx,
                            "segment": seg,
                            "lane": lane,
                            "only_new": effective_only_new,
                            "is_bootstrap": force_bootstrap_full_scan,
                        },
                        queue="scraping",
                    )
                    dispatched += 1
                except Exception:
                    logger.warning("Failed to dispatch segment %d", idx, exc_info=True)

            logger.warning(
                "Parallel segments dispatched: %d/%d (already_completed=%d, bootstrap=%s)",
                dispatched, len(segments), len(already_completed), force_bootstrap_full_scan,
            )
            return {
                "status": "success",
                "task_id": task_id,
                "mode": "parallel_segments",
                "segments_total": len(segments),
                "segments_dispatched": dispatched,
                "segments_already_completed": len(already_completed),
            }
        # --- конец параллельной ветки ---

        resume_from_segment = 0
        if force_bootstrap_full_scan and use_segmented and last_completed_segment is not None:
            resume_from_segment = max(0, last_completed_segment + 1)
            if resume_from_segment >= len(segments):
                # Все сегменты уже пройдены — чекпоинт устарел, начинаем заново.
                logger.info(
                    "Stored checkpoint segment=%d is beyond configured segments (%d); restarting bootstrap from segment 0",
                    last_completed_segment, len(segments),
                )
                resume_from_segment = 0
                _clear_sync_checkpoint(redis_client)
            elif resume_from_segment > 0:
                logger.warning(
                    "Resuming segmented bootstrap from segment=%d (last completed=%d)",
                    resume_from_segment, last_completed_segment,
                )

        def _job():
            with IAAIScraper() as scraper:
                if use_segmented:
                    return scraper.sync_listing_segmented(
                        segments=segments,
                        lane=lane,
                        only_new=effective_only_new,
                        start_segment=resume_from_segment,
                        start_page=1,
                        progress_callback=(
                            (lambda seg_idx: _save_last_completed_segment(redis_client, seg_idx))
                            if force_bootstrap_full_scan else None
                        ),
                    )
                return scraper.sync_listing(
                    make=make,
                    model=model,
                    lane=lane,
                    limit=effective_limit,
                    only_new=effective_only_new,
                )

        result = _run_browser_job(_job)

        if force_bootstrap_full_scan:
            bootstrap_completed = bool(result.get("full_scan_completed"))
            if bootstrap_completed:
                _set_full_scan_done(redis_client, True)
                _clear_sync_checkpoint(redis_client)
                _clear_bootstrap_failure_streak(redis_client)
                logger.info("Bootstrap full scan completed; hourly schedule continues")
            else:
                _set_full_scan_done(redis_client, False)
                listing_payload = result.get("listing") if isinstance(result.get("listing"), dict) else {}
                had_progress = any(
                    int(result.get(key) or 0) > 0
                    for key in ("cars_upserted", "images_upserted", "skipped_existing", "total_discovered")
                ) or int(listing_payload.get("vehicles_collected") or 0) > 0
                count_as_failure = str(result.get("status") or "") == "failed" and not had_progress
                logger.info("Bootstrap full scan not complete yet; queuing immediate continuation")
                _enqueue_bootstrap_followup(
                    "bootstrap_not_completed",
                    count_as_failure=count_as_failure,
                )
        else:
            _clear_sync_checkpoint(redis_client)

        summary = {
            "task_id": task_id,
            "run_id": result.get("run_id"),
            "status": result.get("status", "success"),
            "cars_upserted": result.get("cars_upserted", 0),
            "cars_failed": result.get("cars_failed", 0),
            "images_upserted": result.get("images_upserted", 0),
            "skipped_existing": result.get("skipped_existing", 0),
            "elapsed_seconds": result.get("elapsed_seconds"),
            "failures_count": len(result.get("failures") or []),
        }
        logger.info(
            "sync_listing_task completed: status=%s, %d upserted, %d failed, failures=%d",
            summary["status"],
            summary["cars_upserted"],
            summary["cars_failed"],
            summary["failures_count"],
        )
        return summary

    except SoftTimeLimitExceeded:
        logger.warning(
            "sync_listing_task soft timeout exceeded — partial progress already saved to DB"
        )
        if force_bootstrap_full_scan:
            _set_full_scan_done(redis_client, False)
            _enqueue_bootstrap_followup("soft_time_limit_exceeded", count_as_failure=False)
        # Partial progress уже записан в БД через finish_sync_run.
        # Не retry — следующий запуск продолжит обработку по расписанию.
        return {
            "status": "timed_out",
            "task_id": task_id,
            "reason": "soft_time_limit_exceeded",
            "note": "partial progress saved to DB; bootstrap continuation queued",
        }

    except Exception as exc:
        logger.error("sync_listing_task failed: %s", exc, exc_info=True)
        # Retry только на не-таймаутные ошибки (сеть, БД, браузер).
        try:
            if force_bootstrap_full_scan:
                _set_full_scan_done(redis_client, False)
                raise self.retry(exc=exc, countdown=5)
            raise self.retry(exc=exc)
        except self.MaxRetriesExceededError:
            logger.error("sync_listing_task max retries exceeded, giving up")
            if force_bootstrap_full_scan:
                _set_full_scan_done(redis_client, False)
                _enqueue_bootstrap_followup("max_retries_exceeded", count_as_failure=True)
            return {
                "status": "failed",
                "task_id": task_id,
                "error": str(exc),
            }
    finally:
        if heartbeat_stop is not None:
            heartbeat_stop.set()
        if heartbeat_thread is not None:
            heartbeat_thread.join(timeout=max(1.0, min(5.0, lock_ttl / 10)))
        if lock_acquired:
            _release_lock_if_owner(redis_client, SYNC_LISTING_LOCK_KEY, owner_token)


# Новые задачи для ingestion pipeline

@shared_task(
    name="iaai_scraper.worker.tasks.discover_vehicles_task",
    bind=True,
    max_retries=2,
    default_retry_delay=60,
    acks_late=True,
)
def discover_vehicles_task(self, max_urls: int | None = None):
    """Задача для обнаружения новых URL автомобилей."""
    from ..discovery_service import DiscoveryService

    try:
        discovery = DiscoveryService()
        count = discovery.discover_new_vehicles(max_urls=max_urls)
        logger.info("discover_vehicles_task completed: %d candidates added", count)
        return {"status": "success", "candidates_added": count}
    except Exception as exc:
        logger.error("discover_vehicles_task failed: %s", exc, exc_info=True)
        raise self.retry(exc=exc)


@shared_task(
    name="iaai_scraper.worker.tasks.fetch_pending_candidates_task",
    bind=True,
    max_retries=2,
    default_retry_delay=30,
    acks_late=True,
)
def fetch_pending_candidates_task(self, limit: int = 10):
    """Задача для захвата данных ожидающих кандидатов."""
    from ..fetch_service import FetchService

    try:
        fetch = FetchService()
        count = fetch.process_pending_candidates(limit=limit)
        logger.info("fetch_pending_candidates_task completed: %d candidates processed", count)
        return {"status": "success", "candidates_processed": count}
    except Exception as exc:
        logger.error("fetch_pending_candidates_task failed: %s", exc, exc_info=True)
        raise self.retry(exc=exc)


@shared_task(
    name="iaai_scraper.worker.tasks.enrich_snapshots_task",
    bind=True,
    max_retries=2,
    default_retry_delay=30,
    acks_late=True,
)
def enrich_snapshots_task(self, limit: int = 10):
    """Задача для парсинга и обогащения snapshots."""
    from ..enrichment_service import EnrichmentService

    try:
        enrichment = EnrichmentService()
        count = enrichment.process_unparsed_snapshots(limit=limit)
        logger.info("enrich_snapshots_task completed: %d snapshots enriched", count)
        return {"status": "success", "snapshots_enriched": count}
    except Exception as exc:
        logger.error("enrich_snapshots_task failed: %s", exc, exc_info=True)
        raise self.retry(exc=exc)


@shared_task(
    name="iaai_scraper.worker.tasks.run_ingestion_pipeline_task",
    bind=True,
    max_retries=2,
    default_retry_delay=120,
    acks_late=True,
)
def run_ingestion_pipeline_task(self):
    """Задача для запуска полного ingestion pipeline."""
    from ..scheduler_service import SchedulerService

    try:
        scheduler = SchedulerService()
        scheduler.run_full_pipeline()
        logger.info("run_ingestion_pipeline_task completed")
        return {"status": "success"}
    except Exception as exc:
        logger.error("run_ingestion_pipeline_task failed: %s", exc, exc_info=True)
        raise self.retry(exc=exc)