diff --git a/iaai_scraper/worker/tasks.py b/iaai_scraper/worker/tasks.py index 3dda455..b11bf5f 100644 --- a/iaai_scraper/worker/tasks.py +++ b/iaai_scraper/worker/tasks.py @@ -1,9 +1,11 @@ # 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 @@ -11,11 +13,64 @@ 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, @@ -29,8 +84,11 @@ def sync_vehicle_task(self, vehicle_url: str, lane: str = "iaai"): persistence.create_tables() try: - with IAAIScraper() as scraper: - result = scraper.sync_vehicle(vehicle_url, lane=lane) + 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", @@ -62,18 +120,71 @@ def sync_listing_task( # Полный цикл: листинг + 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: - with IAAIScraper() as scraper: - result = scraper.sync_listing( - make=make, - model=model, - lane=lane, - limit=limit, - only_new=only_new, - ) + 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), @@ -89,3 +200,6 @@ def sync_listing_task( 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)