initial openlane project
This commit is contained in:
qananasikq
360
openlane_scraper/worker/tasks.py
Normal file
360
openlane_scraper/worker/tasks.py
Normal file
@@ -0,0 +1,360 @@
|
||||
# Задачи Celery для синхронизации OpenLane marketplace → DB.
|
||||
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
import uuid
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from threading import Event, Thread
|
||||
|
||||
from billiard.exceptions import SoftTimeLimitExceeded
|
||||
from celery import shared_task
|
||||
from redis import Redis
|
||||
|
||||
from ..core.config import Settings
|
||||
from ..scraper import OpenLaneScraper
|
||||
from ..storage.db import PersistenceService
|
||||
|
||||
logger = logging.getLogger("openlane_scraper.worker.tasks")
|
||||
|
||||
SYNC_LISTING_LOCK_KEY = "openlane:locks:sync_listing"
|
||||
SYNC_FULL_SCAN_DONE_KEY = "openlane:state:sync_full_scan_done"
|
||||
SYNC_LISTING_TASK_NAME = "openlane_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):
|
||||
settings = Settings()
|
||||
soft = settings.celery.task_soft_time_limit
|
||||
hard = settings.celery.task_time_limit
|
||||
wait_timeout = min(hard, soft + 120) if soft and hard else None
|
||||
|
||||
executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="openlane-browser")
|
||||
future = executor.submit(func, *args, **kwargs)
|
||||
try:
|
||||
result = future.result(timeout=wait_timeout)
|
||||
except SoftTimeLimitExceeded:
|
||||
future.cancel()
|
||||
executor.shutdown(wait=False, cancel_futures=True)
|
||||
raise
|
||||
except TimeoutError:
|
||||
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
|
||||
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,
|
||||
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
|
||||
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:
|
||||
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:
|
||||
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:
|
||||
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 _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="openlane_scraper.worker.tasks.sync_listing_task",
|
||||
bind=True,
|
||||
max_retries=3,
|
||||
default_retry_delay=120,
|
||||
acks_late=True,
|
||||
)
|
||||
def sync_listing_task(
|
||||
self,
|
||||
lane: str = "openlane_marketplace",
|
||||
limit: int | None = None,
|
||||
only_new: bool | None = None,
|
||||
max_pages: int | None = None,
|
||||
concurrency: int | None = None,
|
||||
):
|
||||
"""Полный цикл синхронизации OpenLane marketplace → DB."""
|
||||
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
|
||||
|
||||
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 = _is_full_scan_done(redis_client)
|
||||
effective_only_new = False if not full_scan_done else only_new
|
||||
|
||||
if not full_scan_done:
|
||||
logger.info("Bootstrap mode: forcing full scan (only_new=False) until first complete run")
|
||||
|
||||
# Рандомный jitter (0–120s) перед стартом.
|
||||
jitter = random.uniform(0, 120)
|
||||
logger.info("Anti-pattern jitter: waiting %.0fs before starting scrape", jitter)
|
||||
time.sleep(jitter)
|
||||
|
||||
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})
|
||||
|
||||
def _job():
|
||||
with OpenLaneScraper() as scraper:
|
||||
return scraper.sync_listing(
|
||||
lane=lane,
|
||||
limit=limit,
|
||||
only_new=effective_only_new,
|
||||
max_pages=max_pages,
|
||||
concurrency=concurrency,
|
||||
)
|
||||
|
||||
result = _run_browser_job(_job)
|
||||
|
||||
if not full_scan_done:
|
||||
if result.get("status") == "success":
|
||||
_set_full_scan_done(redis_client, True)
|
||||
logger.info("Bootstrap full scan completed; hourly schedule continues")
|
||||
else:
|
||||
_set_full_scan_done(redis_client, False)
|
||||
|
||||
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"),
|
||||
}
|
||||
logger.info(
|
||||
"sync_listing_task completed: status=%s, %d upserted, %d failed",
|
||||
summary["status"], summary["cars_upserted"], summary["cars_failed"],
|
||||
)
|
||||
return summary
|
||||
|
||||
except SoftTimeLimitExceeded:
|
||||
logger.warning("sync_listing_task soft timeout exceeded — partial progress already saved to DB")
|
||||
return {
|
||||
"status": "timed_out",
|
||||
"task_id": task_id,
|
||||
"reason": "soft_time_limit_exceeded",
|
||||
}
|
||||
|
||||
except Exception as exc:
|
||||
logger.error("sync_listing_task failed: %s", exc, exc_info=True)
|
||||
try:
|
||||
raise self.retry(exc=exc)
|
||||
except self.MaxRetriesExceededError:
|
||||
logger.error("sync_listing_task max retries exceeded, giving up")
|
||||
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)
|
||||
Reference in New Issue
Block a user