qananasikq/mobile.de
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve runtime sync flow

Browse Source
This commit is contained in:
qananasikq
2026-05-11 12:45:51 +03:00
parent 124dfc514f
commit d9d073641b
13 changed files with 1224 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files

284
mobilede_scraper/worker/tasks.py
View File

@@ -1,4 +1,4 @@
# Задачи Celery для синхронизации автомобилей и листинга MOBILEDE.
# Celery-задачи для синхронизации MOBILEDE.
import json
import logging
@@ -62,10 +62,9 @@ _MOBILEDE_REFDATA_MAKE_KEY_ALIASES = {
}
_mobilede_site_make_options_cache: dict[str, tuple[float, dict[str, str]]] = {}
_mobilede_refdata_make_keys_cache: tuple[float, dict[str, str]] | None = None
MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY = "mobilede:state:bootstrap_recovery_pending"
# Compatibility guard for partially updated deployments where tasks.py may
# temporarily get ahead of constants.py. Falling back keeps the worker alive
# instead of wedging the tail of a refresh cycle on NameError.
# Резервные значения для частично обновлённых деплоев.
_MOBILEDE_COMPAT_DEFAULTS: dict[str, object] = {
"MOBILEDE_OVERFLOW_SMART_SPLIT_ENABLED": True,
"MOBILEDE_OVERFLOW_SPLIT_PROBE_CANDIDATES": 3,
@@ -73,6 +72,12 @@ _MOBILEDE_COMPAT_DEFAULTS: dict[str, object] = {
"MOBILEDE_SEGMENT_TARGET_MIN_RATIO": 0.65,
"MOBILEDE_SEGMENT_TARGET_MAX_RATIO": 0.98,
"MOBILEDE_SEGMENT_TINY_RATIO": 0.45,
"MOBILEDE_OVERFLOW_MIN_YEAR_SPLIT_SPAN": 4,
"MOBILEDE_OVERFLOW_YEAR_DEEP_SPLIT_DEPTH": 1,
"MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY": 220,
"MOBILEDE_OVERFLOW_SCORE_YEAR_PENALTY": 320,
"MOBILEDE_OVERFLOW_SCORE_MILEAGE_PENALTY": 80,
"MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY": 420,
"MOBILEDE_OVERFLOW_MIN_USEFUL_CHILD_RATIO": 0.55,
"MOBILEDE_OVERFLOW_MAX_MICRO_CHILDREN": 1,
}
@@ -112,8 +117,7 @@ def _mobilede_segment_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.
# Не даём lock жить слишком долго.
effective_hard = min(hard, soft + 120) if soft else hard
return max(effective_hard + 120, 300)
@@ -133,7 +137,7 @@ def _start_stall_watchdog(
def _watchdog() -> None:
key = _task_progress_key(task_id)
no_data_count = 0
# Абсолютный дедлайн: если watchdog работает дольше 3× stall_timeout без прогресса — убиваем.
# Жёсткий дедлайн без прогресса.
watchdog_born = time.monotonic()
absolute_deadline = stall_timeout_seconds * 3
while not stop_event.wait(interval_seconds):
@@ -148,7 +152,7 @@ def _start_stall_watchdog(
"Stall watchdog: no progress data for task %s after %d checks (%.0fs)",
task_id, no_data_count, elapsed_since_born,
)
# Если прогресс-данных нет дольше stall_timeout — считаем задачу мёртвой.
# Без прогресса считаем задачу зависшей.
if elapsed_since_born > stall_timeout_seconds:
logger.error(
"Task %s has no progress data for %.0fs (> %ds); treating as stalled",
@@ -191,7 +195,7 @@ def _start_stall_watchdog(
)
except Exception:
logger.warning("Failed to inspect task progress for stall watchdog", exc_info=True)
# Если Redis тоже не отвечает дольше дедлайна — убиваем.
# Если Redis молчит слишком долго, завершаем процесс.
if time.monotonic() - watchdog_born > absolute_deadline:
logger.error("Stall watchdog: Redis unreachable for %.0fs; forcing kill", time.monotonic() - watchdog_born)
else:
@@ -203,28 +207,26 @@ def _start_stall_watchdog(
reason=f"no DB writes for >{db_idle_restart_seconds}s",
)
# ── Pre-SIGTERM cleanup: release lock so next task can run ──
# Перед остановкой освобождаем lock.
if lock_key and lock_owner:
try:
_release_lock_if_owner(redis_client, lock_key, lock_owner)
logger.info("Stall watchdog: released lock %s before SIGTERM", lock_key)
except Exception:
# Force-delete if owner check fails (process is dying anyway)
# Если проверка владельца не прошла, удаляем lock принудительно.
try:
redis_client.delete(lock_key)
logger.info("Stall watchdog: force-deleted lock %s", lock_key)
except Exception:
logger.warning("Stall watchdog: failed to release lock %s", lock_key, exc_info=True)
# Runtime follow-up is handled by the canonical mobile.de task chain.
# SIGTERM даёт процессу время на cleanup (закрыть DB, browser).
# Celery перехватит SIGTERM и поднимет Terminated / warm shutdown.
# Runtime продолжит каноническая цепочка задач.
# SIGTERM даёт время закрыть ресурсы.
try:
os.kill(os.getpid(), signal.SIGTERM)
except OSError:
pass
# Даём 30 секунд на graceful shutdown, потом SIGKILL как последний resort.
# Ждём graceful shutdown, затем даём SIGKILL.
stop_event.wait(30)
if not stop_event.is_set():
logger.error("Task %s did not stop after SIGTERM; forcing SIGKILL", task_id)
@@ -669,7 +671,7 @@ def _mobilede_filter_source(segment: dict[str, object] | None, search_url: str |
def _is_mobilede_transient_request_error(exc: Exception) -> bool:
if isinstance(exc, requests.exceptions.HTTPError):
status_code = getattr(getattr(exc, "response", None), "status_code", None)
if status_code in {408, 409, 425, 429, 500, 502, 503, 504}:
if status_code in {401, 403, 408, 409, 425, 429, 500, 502, 503, 504}:
return True
if isinstance(
exc,
@@ -691,6 +693,9 @@ def _is_mobilede_transient_request_error(exc: Exception) -> bool:
"connection refused",
"read timed out",
"connect timeout",
"403 client error",
"forbidden",
"too many requests",
)
)
@@ -838,15 +843,23 @@ def _mobilede_try_recover_stalled_bootstrap_queue(
*,
queue_name: str = MOBILEDE_SYNC_QUEUE,
) -> bool:
"""Сбрасывает залипшие bootstrap-dispatched маркеры, если очередь пуста и нет активного прогресса."""
"""Сбрасывает залипшие bootstrap-dispatched маркеры, если очередь пуста и нет свежего прогресса.
Важно проверять именно свежесть прогресса, а не просто наличие глобального
ключа. Иначе после падения воркера старый `last_progress_ts` может жить ещё
несколько дней и бесконечно блокировать recovery/finalize полного прохода.
"""
if not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED or _mobilede_bootstrap_done(redis_client):
return False
try:
queue_len = int(redis_client.llen(queue_name) or 0)
if queue_len > 0:
return False
has_active_progress = bool(redis_client.exists(GLOBAL_PROGRESS_TS_KEY))
if has_active_progress:
has_recent_progress = _has_recent_global_progress(
redis_client,
max_age_seconds=max(180, int(MOBILEDE_BOOTSTRAP_CONTINUATION_DELAY_SECONDS) + 120),
)
if has_recent_progress:
return False
done = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY) or 0)
total = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY) or 0)
@@ -857,7 +870,7 @@ def _mobilede_try_recover_stalled_bootstrap_queue(
return False
redis_client.delete(MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY)
logger.warning(
"mobile.de bootstrap queue stall recovered: queue=0 active=0 progress=%s/%s dispatched=%s -> cleared",
"mobile.de bootstrap queue stall recovered: queue=0 recent_progress=0 progress=%s/%s dispatched=%s -> cleared",
done,
total,
dispatched,
@@ -868,6 +881,53 @@ def _mobilede_try_recover_stalled_bootstrap_queue(
return False
def _queue_mobilede_bootstrap_recovery(
redis_client: Redis,
*,
lane: str,
delay_seconds: float,
use_cursor: bool,
continuous: bool,
segment_label: str,
reason: str,
countdown: int | None = None,
force: bool = False,
) -> bool:
recovery_delay = max(1, int(countdown or MOBILEDE_BOOTSTRAP_CONTINUATION_DELAY_SECONDS))
recovery_ttl = max(30, recovery_delay + 30)
if force:
redis_client.set(MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY, "1", ex=recovery_ttl)
should_queue = True
else:
should_queue = bool(
redis_client.set(MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY, "1", nx=True, ex=recovery_ttl)
)
if should_queue:
mobilede_sync_runtime_segments_task.apply_async(
kwargs={
"lane": lane,
"delay_seconds": delay_seconds,
"use_cursor": use_cursor,
"continuous": continuous,
},
queue=MOBILEDE_SYNC_QUEUE,
countdown=recovery_delay,
)
logger.info(
"mobile.de bootstrap recovery queued: reason=%s runtime=%s delay=%ss",
reason,
segment_label,
recovery_delay,
)
return True
logger.info(
"mobile.de bootstrap recovery already pending: reason=%s runtime=%s",
reason,
segment_label,
)
return False
def _mobilede_current_cycle_id(redis_client: Redis) -> str:
cycle_id = str(redis_client.get(MOBILEDE_INCREMENTAL_CYCLE_KEY) or "").strip()
if not cycle_id:
@@ -981,10 +1041,29 @@ def _planner_module():
name for name in planner.__dict__
if callable(planner.__dict__.get(name)) and (name.startswith("_mobilede_") or name.startswith("_is_mobilede_"))
}
planner_value_names = {
"_MOBILEDE_SITE_MAKE_OPTION_RE",
"_MOBILEDE_RUNTIME_BRAND_ALIASES",
"_MOBILEDE_REFDATA_MAKE_KEY_ALIASES",
"_mobilede_site_make_options_cache",
"_mobilede_refdata_make_keys_cache",
}
planner_helper_names = {
"_get_redis",
"_get_cached_mobilede_runtime_segments",
"_acquire_lock",
"_release_lock_if_owner",
}
for name, value in globals().items():
if name.startswith("MOBILEDE_"):
setattr(planner, name, value)
continue
if name in planner_value_names:
setattr(planner, name, value)
continue
if name in planner_helper_names:
setattr(planner, name, value)
continue
if (name.startswith("_mobilede_") or name.startswith("_is_mobilede_")) and name not in planner_func_names:
setattr(planner, name, value)
planner.logger = logger
@@ -993,10 +1072,6 @@ def _planner_module():
return planner
def _mobilede_price_ranges() -> list[tuple[int, int | None]]:
return _planner_module()._mobilede_price_ranges()
def _mobilede_price_ranges_for_segment(segment: dict[str, object] | None = None) -> list[tuple[int, int | None]]:
return _planner_module()._mobilede_price_ranges_for_segment(segment)
@@ -1033,10 +1108,6 @@ def _mobilede_price_subranges_for_hot_year(price_min: int, price_max: int | None
return _planner_module()._mobilede_price_subranges_for_hot_year(price_min, price_max, year_min, year_max)
def _mobilede_target_results_band() -> tuple[int, int, int]:
return _planner_module()._mobilede_target_results_band()
def _mobilede_range_value(min_value: int | None, max_value: int | None) -> str:
return _planner_module()._mobilede_range_value(min_value, max_value)
@@ -1069,10 +1140,6 @@ def _mobilede_prune_overflow_parent_segments(segments: list[dict[str, object]])
return _planner_module()._mobilede_prune_overflow_parent_segments(segments)
def _mobilede_learned_segments_file() -> str:
return _planner_module()._mobilede_learned_segments_file()
def _mobilede_segments_source_fingerprint(segments: list[dict[str, object]]) -> str:
return _planner_module()._mobilede_segments_source_fingerprint(segments)
@@ -1089,10 +1156,6 @@ def _mobilede_normalize_make_name(value: str) -> str:
return _planner_module()._mobilede_normalize_make_name(value)
def _mobilede_runtime_filter_brands(settings: Settings) -> tuple[str, ...]:
return _planner_module()._mobilede_runtime_filter_brands(settings)
def _mobilede_make_alias_candidates(value: str) -> tuple[str, ...]:
return _planner_module()._mobilede_make_alias_candidates(value)
@@ -1109,22 +1172,10 @@ def _mobilede_parse_site_make_options(html: str) -> dict[str, str]:
return _planner_module()._mobilede_parse_site_make_options(html)
def _mobilede_fetch_site_make_options(search_url: str) -> dict[str, str]:
return _planner_module()._mobilede_fetch_site_make_options(search_url)
def _mobilede_match_site_make_option(brand_name: str, make_options: dict[str, str]) -> tuple[str, str] | None:
return _planner_module()._mobilede_match_site_make_option(brand_name, make_options)
def _mobilede_source_segments_from_runtime_brands(settings: Settings, env_search_urls: list[str]) -> list[dict[str, object]]:
return _planner_module()._mobilede_source_segments_from_runtime_brands(settings, env_search_urls)
def _mobilede_source_segments_from_settings(settings: Settings) -> list[dict[str, object]]:
return _planner_module()._mobilede_source_segments_from_settings(settings)
def _mobilede_split_year_ranges_for_overflow(year_min: int | None, year_max: int | None) -> list[tuple[int | None, int | None]]:
return _planner_module()._mobilede_split_year_ranges_for_overflow(year_min, year_max)
@@ -1645,23 +1696,40 @@ def _mobilede_try_finalize_bootstrap(redis_client: Redis) -> bool:
return True
def _mobilede_post_bootstrap_full_refresh(redis_client: Redis, only_new: bool | None) -> bool:
return bool(MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED and only_new is not True and _mobilede_bootstrap_done(redis_client))
def _mobilede_force_full_scan_only_new(
only_new: bool | None,
*,
redis_client: Redis | None = None,
continuous: bool | None = None,
) -> bool | None:
"""Принудительный full-pass включён только до завершения bootstrap."""
"""Выбирает режим full-pass для bootstrap и почасового continuous-цикла.
Если включён continuous-режим, то даже при `only_new=true` после bootstrap
продолжаем запускать полный проход каждый час: он и обновляет старые авто,
и добирает новые объявления по всем сегментам.
"""
effective_continuous = bool(continuous if continuous is not None else MOBILEDE_CONTINUOUS_SYNC_ENABLED)
if only_new is True and redis_client is not None and _mobilede_bootstrap_done(redis_client):
if effective_continuous and MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED:
return False
return True
if only_new is True:
return False
return only_new
def _mobilede_post_bootstrap_full_refresh(
redis_client: Redis,
only_new: bool | None,
*,
continuous: bool | None = None,
) -> bool:
effective_continuous = bool(continuous if continuous is not None else MOBILEDE_CONTINUOUS_SYNC_ENABLED)
if effective_continuous:
return False
return bool(MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED and only_new is not True and _mobilede_bootstrap_done(redis_client))
def _mobilede_segment_scan_complete(redis_client: Redis, segment: dict[str, object] | None) -> bool:
if not segment:
return False
@@ -1753,11 +1821,23 @@ def _enqueue_mobilede_runtime_segments(
settings = Settings()
runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
redis_client = _get_redis()
only_new = _mobilede_force_full_scan_only_new(runtime_config.sync.only_new, redis_client=redis_client)
effective_continuous_requested = bool(continuous if continuous is not None else MOBILEDE_CONTINUOUS_SYNC_ENABLED)
only_new = _mobilede_force_full_scan_only_new(
runtime_config.sync.only_new,
redis_client=redis_client,
continuous=effective_continuous_requested,
)
full_pass_mode = only_new is not True
bootstrap_active = _mobilede_bootstrap_active(redis_client)
post_bootstrap_refresh = bool(full_pass_mode and _mobilede_post_bootstrap_full_refresh(redis_client, only_new))
effective_continuous = bool(continuous and not post_bootstrap_refresh)
post_bootstrap_refresh = bool(
full_pass_mode
and _mobilede_post_bootstrap_full_refresh(
redis_client,
only_new,
continuous=effective_continuous_requested,
)
)
effective_continuous = bool(effective_continuous_requested and not post_bootstrap_refresh)
effective_use_cursor = use_cursor if only_new is True else False
if runtime_config.sync.only_new is True and only_new is False:
logger.info("mobile.de full-pass mode: forcing only_new=False")
@@ -2143,13 +2223,19 @@ def mobilede_sync_runtime_segments_task(
settings = Settings()
cached_segments = _get_cached_mobilede_runtime_segments(redis_client)
runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
full_pass_mode = _mobilede_force_full_scan_only_new(runtime_config.sync.only_new, redis_client=redis_client) is not True
effective_continuous_requested = bool(continuous if continuous is not None else MOBILEDE_CONTINUOUS_SYNC_ENABLED)
full_pass_mode = _mobilede_force_full_scan_only_new(
runtime_config.sync.only_new,
redis_client=redis_client,
continuous=effective_continuous_requested,
) is not True
repeat_pending = bool(redis_client.get("mobilede:state:full_pass_repeat_pending"))
if full_pass_mode and not full_pass_repeat and repeat_pending:
logger.info(
"mobile.de runtime sync skipped: hourly full-pass repeat is already pending",
)
return {"status": "waiting_repeat"}
bootstrap_recovery_pending = bool(redis_client.get(MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY))
queue_len = int(redis_client.llen(MOBILEDE_SYNC_QUEUE) or 0)
if not cached_segments:
_update_task_progress(
@@ -2194,7 +2280,14 @@ def mobilede_sync_runtime_segments_task(
reset_info["dropped_followup_markers"],
reset_info["dropped_cycle_cursors"],
)
post_bootstrap_refresh = bool(full_pass_mode and _mobilede_post_bootstrap_full_refresh(redis_client, runtime_config.sync.only_new))
post_bootstrap_refresh = bool(
full_pass_mode
and _mobilede_post_bootstrap_full_refresh(
redis_client,
runtime_config.sync.only_new,
continuous=effective_continuous_requested,
)
)
if post_bootstrap_refresh and not full_pass_repeat:
logger.info(
"mobile.de post-bootstrap refresh starting immediately: queue_len=%s",
@@ -2228,6 +2321,18 @@ def mobilede_sync_runtime_segments_task(
if not full_pass_repeat and total_now > 0 and done_now < total_now and (
dispatched_now > 0 or (queue_len > 0 and has_recent_progress)
):
if bootstrap_recovery_pending:
_queue_mobilede_bootstrap_recovery(
redis_client,
lane=lane,
delay_seconds=delay_seconds,
use_cursor=use_cursor,
continuous=bool(continuous if continuous is not None else MOBILEDE_CONTINUOUS_SYNC_ENABLED),
segment_label="runtime_segments",
reason="waiting_for_active_bootstrap_tasks",
countdown=max(5, int(MOBILEDE_BOOTSTRAP_CONTINUATION_DELAY_SECONDS)),
force=True,
)
logger.info(
"mobile.de bootstrap dispatch skipped: progress=%s/%s left=%s dispatched=%s queue_len=%s",
done_now,
@@ -2259,6 +2364,8 @@ def mobilede_sync_runtime_segments_task(
redis_client,
total_segments=len(cached_segments or []),
)
if bootstrap_recovery_pending:
redis_client.delete(MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY)
segments = _enqueue_mobilede_runtime_segments(
lane=lane,
delay_seconds=delay_seconds,
@@ -2322,6 +2429,65 @@ def mobilede_sync_detail_task(self, listing_id: str, lane: str = "mobile_de_cars
raise self.retry(exc=exc)
@shared_task(
name="mobilede.enrich_images_batch",
queue=MOBILEDE_SYNC_QUEUE,
bind=True,
max_retries=1,
default_retry_delay=120,
acks_late=True,
)
def mobilede_enrich_images_batch_task(
self,
limit: int = 50,
lane: str = "mobile_de_cars",
max_existing_images: int = 1,
delay_seconds: float = 1.0,
):
persistence = _get_persistence()
scraper = MobileDeScraper(persistence=persistence)
candidates = persistence.get_active_cars_batch_for_image_enrich(
limit=limit,
max_existing_images=max_existing_images,
)
enriched = 0
failed = 0
skipped = 0
for _car_id, origin_id, origin_url, image_count in candidates:
listing_id = _mobilede_extract_listing_id(origin_url) or str(origin_id).rsplit(":", 1)[-1]
if not listing_id:
skipped += 1
continue
try:
result = scraper.sync_detail(str(listing_id), lane=lane)
enriched += 1
logger.info(
"mobile.de image enrich completed: listing_id=%s origin_id=%s old_images=%s result=%s",
listing_id,
origin_id,
image_count,
result.get("upsert", {}),
)
except Exception as exc:
failed += 1
logger.warning(
"mobile.de image enrich failed: listing_id=%s origin_id=%s error=%s",
listing_id,
origin_id,
exc,
exc_info=True,
)
if delay_seconds:
time.sleep(max(0.0, float(delay_seconds)))
return {
"status": "success",
"candidates": len(candidates),
"enriched": enriched,
"failed": failed,
"skipped": skipped,
}
@shared_task(
name="mobilede.sync_search",
queue=MOBILEDE_SYNC_QUEUE,