104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
"""Celery task wrapper for naш.дом.рф kn-API sweeps."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import random
|
|
import time
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import redis
|
|
|
|
from app.core.config import settings
|
|
from app.services.scrapers.domrf_kn import run_region_sweep
|
|
from app.workers.celery_app import celery_app
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Lock TTL longer than the longest expected sweep (442-obj sweep with extras
|
|
# can run ~30-60 min). 2 hours is a safe upper bound — after that we assume
|
|
# the worker died and let a new run proceed.
|
|
_LOCK_TTL_SECONDS = 2 * 60 * 60
|
|
|
|
|
|
@contextmanager
|
|
def _region_lock(region_code: int, developers: list[str] | None):
|
|
"""Redis singleton lock so beat double-fire or two workers can't run the
|
|
same (region, devs) sweep concurrently. SET NX EX is atomic."""
|
|
devs_key = ",".join(developers) if developers else "*"
|
|
key = f"scrape:kn:lock:{region_code}:{devs_key}"
|
|
r = redis.Redis.from_url(settings.redis_url)
|
|
acquired = r.set(key, "1", nx=True, ex=_LOCK_TTL_SECONDS)
|
|
try:
|
|
yield bool(acquired)
|
|
finally:
|
|
if acquired:
|
|
try:
|
|
r.delete(key)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@celery_app.task(bind=True, name="tasks.scrape_kn.scrape_kn_region", max_retries=2)
|
|
def scrape_kn_region(
|
|
self: Any,
|
|
region_code: int,
|
|
developers: list[str] | None = None,
|
|
skip_jitter: bool = False,
|
|
extras: bool = True,
|
|
download_photos: bool = False,
|
|
fetch_flats: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Run a full kn-API sweep for one region. Returns summary dict.
|
|
|
|
For scheduled runs (skip_jitter=False), sleeps a random 0..SCRAPE_KN_JITTER_SECONDS
|
|
interval before starting so multiple workers / repeated weekly cycles don't hit
|
|
DOM.РФ at exactly the same minute. Manual triggers pass skip_jitter=True for
|
|
immediate execution.
|
|
"""
|
|
if not skip_jitter and settings.scrape_kn_jitter_seconds > 0:
|
|
delay = random.uniform(0, settings.scrape_kn_jitter_seconds)
|
|
logger.info("jitter delay %.0fs before scrape", delay)
|
|
time.sleep(delay)
|
|
|
|
state_path: str | None = settings.scrape_kn_state_path or None
|
|
if state_path and not Path(state_path).exists():
|
|
logger.warning("storage_state %s not found — bootstrap will run cold", state_path)
|
|
state_path = None
|
|
|
|
with _region_lock(region_code, developers) as got_lock:
|
|
if not got_lock:
|
|
logger.warning(
|
|
"scrape_kn_region SKIPPED region=%s devs=%s — another sweep is"
|
|
" already running (Redis lock held)",
|
|
region_code,
|
|
developers,
|
|
)
|
|
return {
|
|
"skipped": True,
|
|
"reason": "lock_held",
|
|
"region_code": region_code,
|
|
"developers": developers,
|
|
}
|
|
|
|
logger.info(
|
|
"scrape_kn_region start region=%s devs=%s extras=%s photos=%s state=%s",
|
|
region_code,
|
|
developers,
|
|
extras,
|
|
download_photos,
|
|
state_path,
|
|
)
|
|
return asyncio.run(
|
|
run_region_sweep(
|
|
region_code=region_code,
|
|
developers=developers,
|
|
load_state=state_path,
|
|
fetch_flats=fetch_flats,
|
|
extras=extras,
|
|
download_photos_binary=download_photos,
|
|
)
|
|
)
|