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

refactor mobile.de parser, fix country mapping, update README

Browse Source
This commit is contained in:
qananasikq
2026-05-04 18:12:10 +03:00
parent edddfc884c
commit 800867b4e6
46 changed files with 2887 additions and 10900 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
mobilede_scraper/worker/celery_app.py
View File

@@ -127,7 +127,6 @@ celery_app.conf.update(
"mobilede.sync_runtime_segments": {"queue": MOBILEDE_SYNC_QUEUE},
"mobilede.sync_search": {"queue": MOBILEDE_SYNC_QUEUE},
"mobilede.sync_detail": {"queue": MOBILEDE_SYNC_QUEUE},
"MOBILEDE.sync_cars_feed": {"queue": MOBILEDE_SYNC_QUEUE},
"mobilede_scraper.worker.tasks.*": {"queue": MOBILEDE_SYNC_QUEUE},
},
)
@@ -137,7 +136,7 @@ celery_app.autodiscover_tasks(["mobilede_scraper.worker"])
@worker_ready.connect
def _on_worker_ready(**kwargs):
"""При старте worker отправляем первый sync_listing, если очередь пуста."""
"""При старте worker отправляем первый canonical runtime sync, если очередь пуста."""
if not _env_bool("MOBILEDE_STARTUP_SYNC_ENABLED", True):
logger.info("Worker ready: startup sync dispatch disabled by MOBILEDE_STARTUP_SYNC_ENABLED")
return
@@ -154,16 +153,6 @@ def _on_worker_ready(**kwargs):
)
has_fresh_progress = _has_fresh_active_progress(redis_client)
for stale_key in ("mobilede:locks:sync_listing",):
try:
ttl = redis_client.ttl(stale_key)
if ttl is not None and ttl != -2 and not has_fresh_progress:
redis_client.delete(stale_key)
logger.info("Cleared stale lock on startup: %s (ttl was %s)", stale_key, ttl)
elif ttl is not None and ttl != -2:
logger.info("Keeping sync lock on startup because fresh active progress exists: %s (ttl=%s)", stale_key, ttl)
except Exception:
logger.warning("Failed to inspect stale lock %s on startup", stale_key, exc_info=True)
try:
queue_len = int(redis_client.llen(MOBILEDE_SYNC_QUEUE) or 0)

137
mobilede_scraper/worker/constants.py Normal file
View File

@@ -0,0 +1,137 @@
from __future__ import annotations
import os
MOBILEDE_SYNC_QUEUE = "mobilede_sync"
MOBILEDE_SEARCH_CURSOR_KEY = "mobilede:state:search_next_page"
MOBILEDE_SEGMENT_CURSOR_KEY_FMT = "mobilede:state:search_next_page:{segment_key}"
MOBILEDE_RUNTIME_SEGMENT_INDEX_KEY = "mobilede:state:runtime_segment_index"
MOBILEDE_RUNTIME_SEGMENTS_TASK = "mobilede.sync_runtime_segments"
MOBILEDE_SYNC_TASK_NAME = "mobilede.sync_search"
MOBILEDE_SEGMENT_LOCK_KEY_FMT = "mobilede:locks:segment:{segment_key}"
MOBILEDE_SEGMENT_FOLLOWUP_PENDING_KEY_FMT = "mobilede:state:followup_pending:{segment_key}"
MOBILEDE_PROGRESS_PAGE_COUNTER_KEY_FMT = "mobilede:state:progress_pages:{segment_key}"
MOBILEDE_CONTINUOUS_SYNC_ENABLED = os.getenv("MOBILEDE_CONTINUOUS_SYNC_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS = max(0, int(float(os.getenv("MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS", "15"))))
MOBILEDE_FULL_PASS_REPEAT_DELAY_SECONDS = max(60, int(float(os.getenv("MOBILEDE_FULL_PASS_REPEAT_DELAY_SECONDS", "3600"))))
MOBILEDE_BOOTSTRAP_CONTINUATION_DELAY_SECONDS = max(0, int(float(os.getenv("MOBILEDE_BOOTSTRAP_CONTINUATION_DELAY_SECONDS", "5"))))
MOBILEDE_PROGRESS_LOG_EVERY_PAGES = max(1, int(os.getenv("MOBILEDE_PROGRESS_LOG_EVERY_PAGES", "10")))
MOBILEDE_SKIP_EMPTY_WINDOW = os.getenv("MOBILEDE_SKIP_EMPTY_WINDOW", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_ROTATE_RUNTIME_SEGMENTS = os.getenv("MOBILEDE_ROTATE_RUNTIME_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_RUNTIME_INITIAL_TASKS = max(1, int(os.getenv("MOBILEDE_RUNTIME_INITIAL_TASKS", "2")))
MOBILEDE_SEGMENT_PAGE_WINDOW = max(1, int(os.getenv("MOBILEDE_SEGMENT_PAGE_WINDOW", "10")))
MOBILEDE_RESULTS_PER_PAGE = max(1, int(os.getenv("MOBILEDE_RESULTS_PER_PAGE", "20")))
MOBILEDE_MAX_PAGE_NUMBER = max(1, int(os.getenv("MOBILEDE_MAX_PAGE_NUMBER", "50")))
MOBILEDE_SEGMENT_TARGET_RESULTS = max(
MOBILEDE_RESULTS_PER_PAGE,
int(os.getenv("MOBILEDE_SEGMENT_TARGET_RESULTS", str(MOBILEDE_RESULTS_PER_PAGE * MOBILEDE_MAX_PAGE_NUMBER))),
)
MOBILEDE_DYNAMIC_SEGMENT_PROBES = os.getenv("MOBILEDE_DYNAMIC_SEGMENT_PROBES", "false").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_PREPLAN_SEGMENT_PROBES = os.getenv("MOBILEDE_PREPLAN_SEGMENT_PROBES", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_PREPLAN_MAX_SEGMENTS = max(1, int(os.getenv("MOBILEDE_PREPLAN_MAX_SEGMENTS", "1000")))
MOBILEDE_PREPLAN_MAX_PROBES = max(0, int(os.getenv("MOBILEDE_PREPLAN_MAX_PROBES", "40")))
MOBILEDE_PREPLAN_SPLIT_THRESHOLD_RATIO = min(
5.0,
max(1.0, float(os.getenv("MOBILEDE_PREPLAN_SPLIT_THRESHOLD_RATIO", "2.5"))),
)
MOBILEDE_PREPLAN_MAX_SPLIT_DEPTH = max(0, min(3, int(os.getenv("MOBILEDE_PREPLAN_MAX_SPLIT_DEPTH", "1"))))
MOBILEDE_COMPACT_SEGMENTS = os.getenv("MOBILEDE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = os.getenv("MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE", "false").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_SKIP_EMPTY_DYNAMIC_SEGMENTS = os.getenv("MOBILEDE_SKIP_EMPTY_DYNAMIC_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_HOT_BASE_SPLIT_ENABLED = os.getenv("MOBILEDE_HOT_BASE_SPLIT_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_HOT_BASE_PRICE_MAX = max(5000, int(os.getenv("MOBILEDE_HOT_BASE_PRICE_MAX", "30000")))
MOBILEDE_HOT_RECENT_YEAR_MIN = max(2000, int(os.getenv("MOBILEDE_HOT_RECENT_YEAR_MIN", "2018")))
MOBILEDE_HOT_MILEAGE_SPLIT_ENABLED = os.getenv("MOBILEDE_HOT_MILEAGE_SPLIT_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_HOT_MILEAGE_PRICE_MIN = max(1, int(os.getenv("MOBILEDE_HOT_MILEAGE_PRICE_MIN", "15001")))
MOBILEDE_HOT_MILEAGE_PRICE_MAX = max(MOBILEDE_HOT_MILEAGE_PRICE_MIN, int(os.getenv("MOBILEDE_HOT_MILEAGE_PRICE_MAX", "30000")))
MOBILEDE_HOT_OLD_CHEAP_PRICE_MAX = max(1, int(os.getenv("MOBILEDE_HOT_OLD_CHEAP_PRICE_MAX", "5000")))
MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED = os.getenv("MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP = os.getenv("MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_INCREMENTAL_PAGE_WINDOW = max(1, int(os.getenv("MOBILEDE_INCREMENTAL_PAGE_WINDOW", "1")))
MOBILEDE_ONLY_NEW_NEWEST_FIRST = os.getenv("MOBILEDE_ONLY_NEW_NEWEST_FIRST", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_INCREMENTAL_STRICT_FIRST_PASS = os.getenv("MOBILEDE_INCREMENTAL_STRICT_FIRST_PASS", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_ONLY_NEW_ZERO_INSERT_STREAK = max(1, int(os.getenv("MOBILEDE_ONLY_NEW_ZERO_INSERT_STREAK", "1")))
MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS = max(60, int(os.getenv("MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS", "3600")))
MOBILEDE_ONLY_NEW_HOT_ONLY = os.getenv("MOBILEDE_ONLY_NEW_HOT_ONLY", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_ONLY_NEW_HOT_TTL_SECONDS = max(300, int(os.getenv("MOBILEDE_ONLY_NEW_HOT_TTL_SECONDS", "10800")))
MOBILEDE_ONLY_NEW_MIN_INSERT_RATIO = min(1.0, max(0.0, float(os.getenv("MOBILEDE_ONLY_NEW_MIN_INSERT_RATIO", "0.95"))))
MOBILEDE_BOOTSTRAP_DONE_KEY = "mobilede:state:bootstrap_full_scan_done"
MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY = "mobilede:state:bootstrap_segments_total"
MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY = "mobilede:state:bootstrap_segments_done"
MOBILEDE_BOOTSTRAP_LISTINGS_TOTAL_KEY = "mobilede:state:bootstrap_listings_total"
MOBILEDE_BOOTSTRAP_UNIQUE_TOTAL_KEY = "mobilede:state:bootstrap_unique_total"
MOBILEDE_BOOTSTRAP_INSERTED_TOTAL_KEY = "mobilede:state:bootstrap_inserted_total"
MOBILEDE_BOOTSTRAP_UPDATED_TOTAL_KEY = "mobilede:state:bootstrap_updated_total"
MOBILEDE_BOOTSTRAP_IMAGES_TOTAL_KEY = "mobilede:state:bootstrap_images_total"
MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY = "mobilede:state:bootstrap_dispatched_segments"
MOBILEDE_BOOTSTRAP_INCREMENTAL_TRANSITION_KEY = "mobilede:state:bootstrap_incremental_transition"
MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY = "mobilede:state:runtime_segments_cache"
MOBILEDE_RUNTIME_SEGMENTS_PLAN_FINALIZED_KEY = "mobilede:state:runtime_segments_plan_finalized"
MOBILEDE_RUNTIME_SEGMENTS_BUILDING_KEY = "mobilede:state:runtime_segments_building"
MOBILEDE_RUNTIME_SEGMENTS_PENDING_KEY = "mobilede:state:runtime_segments_pending"
MOBILEDE_RUNTIME_SEGMENTS_CACHE_LOCK_KEY = "mobilede:locks:runtime_segments_cache"
MOBILEDE_OVERFLOW_EXPANDED_PARENTS_KEY = "mobilede:state:overflow_expanded_parents"
MOBILEDE_REFRESH_CYCLE_ID_KEY = "mobilede:state:refresh_cycle:id"
MOBILEDE_REFRESH_CYCLE_STARTED_AT_KEY = "mobilede:state:refresh_cycle:started_at"
MOBILEDE_REFRESH_CYCLE_TOTAL_KEY = "mobilede:state:refresh_cycle:total"
MOBILEDE_REFRESH_CYCLE_DONE_KEY = "mobilede:state:refresh_cycle:done"
MOBILEDE_REFRESH_CYCLE_DONE_SEGMENTS_KEY_FMT = "mobilede:state:refresh_cycle:done_segments:{cycle_id}"
MOBILEDE_REFRESH_CYCLE_FINALIZED_KEY_FMT = "mobilede:state:refresh_cycle:finalized:{cycle_id}"
MOBILEDE_REFRESH_CYCLE_TTL_SECONDS = max(60 * 60, int(os.getenv("MOBILEDE_REFRESH_CYCLE_TTL_SECONDS", str(24 * 60 * 60))))
MOBILEDE_OVERFLOW_SPLIT_ENABLED = os.getenv("MOBILEDE_OVERFLOW_SPLIT_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_OVERFLOW_SPLIT_THRESHOLD_RATIO = min(
1.0,
max(0.5, float(os.getenv("MOBILEDE_OVERFLOW_SPLIT_THRESHOLD_RATIO", "0.98"))),
)
MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS = max(
1,
min(5, int(os.getenv("MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS", "3"))),
)
MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN = max(
250,
int(os.getenv("MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN", "1000")),
)
MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH = max(
1,
min(6, int(os.getenv("MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH", "6"))),
)
MOBILEDE_INCREMENTAL_CYCLE_KEY = "mobilede:state:incremental_cycle"
MOBILEDE_INCREMENTAL_CYCLE_SEEN_COUNT_KEY = "mobilede:state:incremental_cycle_seen_count"
MOBILEDE_INCREMENTAL_CYCLE_SEEN_SET_KEY_FMT = "mobilede:state:incremental_cycle_seen:{cycle_id}"
TASK_PROGRESS_KEY_FMT = "mobilede:state:task_progress:{task_id}"
GLOBAL_PROGRESS_TS_KEY = "mobilede:state:last_progress_ts"
GLOBAL_DB_PROGRESS_TS_KEY = "mobilede:state:last_db_progress_ts"
DB_PROGRESS_STAGES = {
"db_upsert_done",
}
STALL_WATCHDOG_NAVIGATION_STAGES = {
"page_collected",
}
STALL_WATCHDOG_LONG_RUNNING_STAGES = {
"search_collection_done",
"records_mapped",
}
STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS = max(
300,
int(os.getenv("STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS", "900")),
)
STALL_WATCHDOG_DETAIL_GRACE_SECONDS = max(
600,
int(os.getenv("STALL_WATCHDOG_DETAIL_GRACE_SECONDS", "1200")),
)
DB_IDLE_RESTART_SECONDS = max(60, int(os.getenv("MOBILEDE_DB_IDLE_RESTART_SECONDS", "3600")))
TERMINAL_PROGRESS_STAGES = {
"segment_done",
"segment_failed",
"segment_task_completed",
"segment_task_failed",
"segment_task_soft_timeout",
"sync_done",
"failed",
}

105
mobilede_scraper/worker/progress.py Normal file
View File

@@ -0,0 +1,105 @@
from __future__ import annotations
import json
import logging
import time
from redis import Redis
from .constants import (
DB_PROGRESS_STAGES,
GLOBAL_DB_PROGRESS_TS_KEY,
GLOBAL_PROGRESS_TS_KEY,
MOBILEDE_SEGMENT_FOLLOWUP_PENDING_KEY_FMT,
MOBILEDE_SEGMENT_LOCK_KEY_FMT,
STALL_WATCHDOG_DETAIL_GRACE_SECONDS,
STALL_WATCHDOG_LONG_RUNNING_STAGES,
STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS,
STALL_WATCHDOG_NAVIGATION_STAGES,
TASK_PROGRESS_KEY_FMT,
)
logger = logging.getLogger("mobilede_scraper.worker.progress")
def _safe_int(value) -> int | None:
try:
if value is None:
return None
return int(value)
except (TypeError, ValueError):
return None
def _task_progress_key(task_id: str) -> str:
return TASK_PROGRESS_KEY_FMT.format(task_id=task_id)
def _mobilede_segment_lock_key(segment_key: str) -> str:
return MOBILEDE_SEGMENT_LOCK_KEY_FMT.format(segment_key=segment_key)
def _mobilede_followup_pending_key(segment_key: str) -> str:
return MOBILEDE_SEGMENT_FOLLOWUP_PENDING_KEY_FMT.format(segment_key=segment_key)
def _update_task_progress(
redis_client: Redis,
*,
task_id: str,
stage: str,
ttl_seconds: int,
**payload,
) -> None:
try:
now_ts = int(time.time())
existing_task_started_ts: int | None = None
try:
existing_raw = redis_client.get(_task_progress_key(task_id))
if existing_raw:
existing_payload = json.loads(existing_raw)
existing_task_started_ts = _safe_int(existing_payload.get("task_started_ts"))
except Exception:
existing_task_started_ts = None
payload.setdefault("task_started_ts", existing_task_started_ts or now_ts)
if stage not in DB_PROGRESS_STAGES and "last_db_progress_ts" not in payload:
last_db_progress_ts = _safe_int(redis_client.get(GLOBAL_DB_PROGRESS_TS_KEY))
if last_db_progress_ts is not None:
payload["last_db_progress_ts"] = last_db_progress_ts
data = {
"task_id": task_id,
"stage": stage,
"ts": now_ts,
**payload,
}
ttl = max(60, int(ttl_seconds))
pipe = redis_client.pipeline()
pipe.set(
_task_progress_key(task_id),
json.dumps(data, ensure_ascii=False),
ex=ttl,
)
# Глобальный маркер активности для внешнего guard-процесса.
# Нужен, чтобы контейнер мог самовосстанавливаться при полном зависании воркера
# (когда PID жив, но прогресс по задачам не двигается).
pipe.set(GLOBAL_PROGRESS_TS_KEY, str(now_ts), ex=max(ttl, 7 * 24 * 60 * 60))
if stage in DB_PROGRESS_STAGES:
pipe.set(GLOBAL_DB_PROGRESS_TS_KEY, str(now_ts), ex=max(ttl, 7 * 24 * 60 * 60))
pipe.execute()
except Exception:
logger.warning("Failed to update task progress for %s", task_id, exc_info=True)
def _clear_task_progress(redis_client: Redis, task_id: str) -> None:
try:
redis_client.delete(_task_progress_key(task_id))
except Exception:
logger.warning("Failed to clear task progress for %s", task_id, exc_info=True)
def _stall_timeout_for_progress(stage: str | None, default_timeout: int) -> int:
if stage in STALL_WATCHDOG_NAVIGATION_STAGES:
return max(int(default_timeout), STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS)
if stage in STALL_WATCHDOG_LONG_RUNNING_STAGES:
return max(int(default_timeout), STALL_WATCHDOG_DETAIL_GRACE_SECONDS)
return int(default_timeout)

21
mobilede_scraper/worker/self_heal.py
View File

@@ -7,17 +7,14 @@ import time
from redis import Redis
from .constants import DB_PROGRESS_STAGES
logger = logging.getLogger("MOBILEDE_scraper.worker.self_heal")
MOBILEDE_SYNC_QUEUE = "MOBILEDE_sync"
MOBILEDE_SYNC_QUEUE = "mobilede_sync"
GLOBAL_PROGRESS_TS_KEY = "mobilede:state:last_progress_ts"
GLOBAL_DB_PROGRESS_TS_KEY = "mobilede:state:last_db_progress_ts"
SELF_HEAL_RESTART_LOCK_KEY = "mobilede:state:self_heal_restart_in_progress"
SYNC_LISTING_LOCK_KEY = "mobilede:locks:sync_listing"
SYNC_FULL_SCAN_DONE_KEY = "mobilede:state:sync_full_scan_done"
SYNC_LISTING_CHECKPOINT_KEY = "mobilede:state:sync_listing_checkpoint"
SYNC_LISTING_FOLLOWUP_PENDING_KEY = "mobilede:state:sync_listing_followup_pending"
SIGKILL_FALLBACK = getattr(signal, "SIGKILL", signal.SIGTERM)
@@ -97,7 +94,7 @@ def _read_last_db_progress_ts(redis_client: Redis) -> int | None:
if not payload:
continue
data = json.loads(payload)
if str(data.get("stage") or "") == "fast_db_progress":
if str(data.get("stage") or "") in DB_PROGRESS_STAGES:
ts = _safe_int(data.get("ts"), 0)
else:
ts = _safe_int(data.get("last_db_progress_ts"), 0)
@@ -110,28 +107,28 @@ def _read_last_db_progress_ts(redis_client: Redis) -> int | None:
def _reset_bootstrap_checkpoint_for_db_idle(redis_client: Redis) -> None:
pipe = redis_client.pipeline()
pipe.delete(SYNC_LISTING_CHECKPOINT_KEY)
pipe.delete(SYNC_LISTING_FOLLOWUP_PENDING_KEY)
pipe.delete(GLOBAL_PROGRESS_TS_KEY)
pipe.delete(GLOBAL_DB_PROGRESS_TS_KEY)
pipe.set(SYNC_FULL_SCAN_DONE_KEY, "0")
pipe.execute()
def _has_inflight_work(redis_client: Redis, queue_name: str) -> tuple[bool, dict[str, int]]:
"""Есть ли признаки активной/зависшей работы, даже если очередь пуста."""
queue_len = _safe_int(redis_client.llen(queue_name), 0)
has_lock = 1 if redis_client.get(SYNC_LISTING_LOCK_KEY) else 0
has_segment_lock = 0
for _ in redis_client.scan_iter(match="mobilede:locks:segment:*"):
has_segment_lock = 1
break
has_task_progress = 0
for _ in redis_client.scan_iter(match="mobilede:state:task_progress:*"):
has_task_progress = 1
break
flags = {
"queue_len": queue_len,
"has_lock": has_lock,
"has_segment_lock": has_segment_lock,
"has_task_progress": has_task_progress,
}
return (queue_len > 0 or has_lock == 1 or has_task_progress == 1), flags
return (queue_len > 0 or has_segment_lock == 1 or has_task_progress == 1), flags
def _kill_worker_process() -> None:

3256
mobilede_scraper/worker/tasks.py
View File

File diff suppressed because it is too large Load Diff