# Задачи Celery для синхронизации автомобилей и листинга IAAI. from concurrent.futures import ThreadPoolExecutor import json import logging 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 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_TASK_NAME = "iaai_scraper.worker.tasks.sync_listing_task" 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 _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) 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_sync_checkpoint(redis_client: Redis) -> dict[str, object] | None: 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 not raw: return None if isinstance(raw, str) and not raw.strip(): return None try: data = json.loads(str(raw)) except Exception: logger.warning("Failed to decode sync listing checkpoint", exc_info=True) return None return data if isinstance(data, dict) else None def _save_sync_checkpoint( redis_client: Redis, *, task_id: str, page_number: int, make: str | None, model: str | None, lane: str, segment_index: int | None = None, ) -> None: payload = { "status": "in_progress", "task_id": task_id, "last_successful_page": int(page_number), "make": make, "model": model, "lane": lane, "segment_index": segment_index, "updated_at": int(time.time()), } try: redis_client.set(SYNC_LISTING_CHECKPOINT_KEY, json.dumps(payload)) 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 _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 @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) -> None: 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: 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: 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 checkpoint = _load_sync_checkpoint(redis_client) resume_from_page = 1 if checkpoint and str(checkpoint.get("status") or "") == "in_progress": checkpoint_page = int(checkpoint.get("last_successful_page") or 0) checkpoint_make = checkpoint.get("make") checkpoint_model = checkpoint.get("model") checkpoint_lane = checkpoint.get("lane") checkpoint_segment = checkpoint.get("segment_index") same_scope = ( checkpoint_make == make and checkpoint_model == model and checkpoint_lane == lane ) # Для сегментированного режима: совпадение по lane + наличие segment_index is_segmented_checkpoint = checkpoint_segment is not None and checkpoint_lane == lane if checkpoint_page > 0 and (same_scope or is_segmented_checkpoint): resume_from_page = checkpoint_page + 1 logger.warning( "Resuming sync_listing from page %d (segment=%s) using checkpoint", resume_from_page, checkpoint_segment, ) elif checkpoint_page > 0: logger.info("Ignoring stale checkpoint due to different sync parameters") _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, ) 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}) # Определяем сегменты из конфига. settings = Settings() segments = parse_listing_segments(settings.listing.listing_segments_json) use_segmented = bool(segments) and make is None and model is None resume_from_segment = 0 if use_segmented and checkpoint and str(checkpoint.get("status") or "") == "in_progress": cp_segment = checkpoint.get("segment_index") if cp_segment is not None and int(cp_segment) >= 0: resume_from_segment = int(cp_segment) # resume_from_page уже вычислен выше 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=resume_from_page, progress_callback=lambda seg_idx, page_number: _save_sync_checkpoint( redis_client, task_id=task_id, page_number=page_number, make=None, model=None, lane=lane, segment_index=seg_idx, ), ) return scraper.sync_listing( make=make, model=model, lane=lane, limit=effective_limit, only_new=effective_only_new, start_page=resume_from_page, progress_callback=lambda page_number: _save_sync_checkpoint( redis_client, task_id=task_id, page_number=page_number, make=make, model=model, lane=lane, ), ) 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) logger.info("Bootstrap full scan completed; hourly schedule continues") else: _set_full_scan_done(redis_client, False) logger.info("Bootstrap full scan not complete yet; queuing immediate continuation") _enqueue_bootstrap_followup("bootstrap_not_completed") elif result.get("status") == "success": _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") # 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") 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)