dubizzle/iaai_scraper/worker/tasks.py

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

from concurrent.futures import ThreadPoolExecutor
import json
import logging

from celery import shared_task
from redis import Redis

from ..core.config import Settings
from ..scraper import IAAIScraper
from ..storage.db import PersistenceService

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

SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"


def _run_browser_job(func, *args, **kwargs):
    # Playwright Sync API нельзя запускать в потоке с активным asyncio loop.
    # Celery/зависимости могут поднимать loop в worker-процессе, поэтому
    # браузерный код выполняем в отдельном thread без loop.
    with ThreadPoolExecutor(max_workers=1, thread_name_prefix="iaai-browser") as executor:
        future = executor.submit(func, *args, **kwargs)
        return future.result()


def _sync_listing_lock_ttl_seconds() -> int:
    settings = Settings()
    # Небольшой запас к hard time limit задачи, чтобы lock самоснимался после сбоев.
    return max(settings.celery.task_time_limit + 120, 300)


def _get_persistence() -> PersistenceService:
    return PersistenceService(Settings())


def _get_redis() -> Redis:
    settings = Settings()
    return Redis.from_url(settings.redis.url, decode_responses=True)


def _release_lock_if_owner(redis_client: Redis, key: str, owner: str) -> None:
    try:
        current_owner = redis_client.get(key)
        if current_owner == owner:
            redis_client.delete(key)
    except Exception as exc:
        logger.warning("Failed to release lock %s: %s", key, exc)


def _has_other_active_sync_listing_task(task) -> bool:
    try:
        inspector = task.app.control.inspect(timeout=1.0)
        active_map = inspector.active() or {}
    except Exception as exc:
        logger.warning("Failed to inspect active tasks: %s", exc)
        return False

    current_task_id = task.request.id
    for worker_tasks in active_map.values():
        for item in worker_tasks or []:
            name = str(item.get("name") or "")
            task_id = str(item.get("id") or "")
            if (
                name == "iaai_scraper.worker.tasks.sync_listing_task"
                and task_id
                and task_id != current_task_id
            ):
                return True
    return False


@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)
        raise self.retry(exc=exc)


@shared_task(
    name="iaai_scraper.worker.tasks.sync_listing_task",
    bind=True,
    max_retries=1,
    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"
    redis_client = _get_redis()

    lock_acquired = False
    lock_ttl = _sync_listing_lock_ttl_seconds()
    try:
        lock_acquired = bool(
            redis_client.set(
                SYNC_LISTING_LOCK_KEY,
                task_id,
                nx=True,
                ex=lock_ttl,
            )
        )
    except Exception as exc:
        logger.warning("Failed to acquire sync lock in Redis: %s", exc)

    if not lock_acquired:
        # Возможен stale lock после рестарта worker. Если активного sync_listing нет —
        # снимаем lock и пытаемся взять его заново.
        if not _has_other_active_sync_listing_task(self):
            try:
                stale_owner = redis_client.get(SYNC_LISTING_LOCK_KEY)
                if stale_owner:
                    logger.warning(
                        "Removing stale sync lock held by task %s",
                        stale_owner,
                    )
                redis_client.delete(SYNC_LISTING_LOCK_KEY)
                lock_acquired = bool(
                    redis_client.set(
                        SYNC_LISTING_LOCK_KEY,
                        task_id,
                        nx=True,
                        ex=lock_ttl,
                    )
                )
            except Exception as exc:
                logger.warning("Failed to recover stale sync lock: %s", exc)

    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:
        self.update_state(state="STARTED", meta={"stage": "sync_listing_started", "task_id": task_id})
        def _job():
            with IAAIScraper() as scraper:
                return scraper.sync_listing(
                    make=make,
                    model=model,
                    lane=lane,
                    limit=limit,
                    only_new=only_new,
                )

        result = _run_browser_job(_job)

        summary = {
            "task_id": task_id,
            "run_id": result.get("run_id"),
            "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"),
        }
        logger.info(
            "sync_listing_task completed: %d upserted, %d failed",
            summary["cars_upserted"], summary["cars_failed"],
        )
        return {"status": "success", **summary}

    except Exception as exc:
        logger.error("sync_listing_task failed: %s", exc)
        raise self.retry(exc=exc)
    finally:
        if lock_acquired:
            _release_lock_if_owner(redis_client, SYNC_LISTING_LOCK_KEY, task_id)