diff --git a/mobilede_scraper/core/config.py b/mobilede_scraper/core/config.py index 9ccd241..662548a 100644 --- a/mobilede_scraper/core/config.py +++ b/mobilede_scraper/core/config.py @@ -51,7 +51,6 @@ def _env_float(name: str, default: float) -> float: @dataclass(slots=True) class ListingConfig: - cars_url: str = _env_str("MOBILEDE_CARS_LISTING_URL", "https://www.mobile.de/Vehiclelisting/Cars") filtered_search_url: str | None = _env_optional_str("MOBILEDE_FILTERED_SEARCH_URL") filtered_search_urls_raw: str | None = _env_optional_str("MOBILEDE_FILTERED_SEARCH_URLS") @@ -112,51 +111,6 @@ class CeleryConfig: worker_max_tasks_per_child: int = _env_int("CELERY_WORKER_MAX_TASKS_PER_CHILD", 5) broker_visibility_timeout: int = _env_int("CELERY_BROKER_VISIBILITY_TIMEOUT", 7200) beat_sync_interval_minutes: int = _env_int("CELERY_BEAT_SYNC_INTERVAL_MINUTES", 60) - beat_sync_limit: int | None = _env_int("CELERY_BEAT_SYNC_LIMIT", 0) or None - batch_size: int = _env_int("CELERY_BATCH_SIZE", 50) - parallel_tabs: int = _env_int("MOBILEDE_PARALLEL_TABS", 8) - fetch_concurrency: int = _env_int("MOBILEDE_FETCH_CONCURRENCY", _env_int("MOBILEDE_PARALLEL_TABS", 8)) - parallel_segments: bool = _env_bool("CELERY_PARALLEL_SEGMENTS", False) - block_resources: bool = _env_bool("MOBILEDE_BLOCK_RESOURCES", True) - - -@dataclass(slots=True) -class ScrapingProfileConfig: - name: str = _env_str("MOBILEDE_SCRAPING_PROFILE", _env_str("MOBILEDE_PROFILE", "stable")).strip().lower() - http_first: bool = _env_bool("MOBILEDE_HTTP_FIRST", True) - browser_fallback_enabled: bool = _env_bool("MOBILEDE_BROWSER_FALLBACK_ENABLED", True) - anonymous_bootstrap_enabled: bool = _env_bool("MOBILEDE_ANONYMOUS_BOOTSTRAP_ENABLED", True) - verbose_http_logs: bool = _env_bool("MOBILEDE_VERBOSE_HTTP_LOGS", False) - verbose_progress_logs: bool = _env_bool("MOBILEDE_VERBOSE_PROGRESS_LOGS", False) - challenge_refresh_enabled: bool = _env_bool("MOBILEDE_CHALLENGE_REFRESH_ENABLED", True) - challenge_refresh_attempts: int = _env_int("MOBILEDE_CHALLENGE_REFRESH_ATTEMPTS", 3) - listing_post_attempts: int = _env_int("MOBILEDE_LISTING_POST_ATTEMPTS", 4) - request_jitter_max_s: float = _env_float("MOBILEDE_REQUEST_JITTER_MAX_S", 0.15) - detail_retries: int | None = _env_int("MOBILEDE_DETAIL_RETRIES", -1) - listing_retries: int | None = _env_int("MOBILEDE_LISTING_RETRIES", -1) - - def __post_init__(self) -> None: - if self.name == "fast": - if "MOBILEDE_BROWSER_FALLBACK_ENABLED" not in os.environ: - self.browser_fallback_enabled = False - if "MOBILEDE_ANONYMOUS_BOOTSTRAP_ENABLED" not in os.environ: - self.anonymous_bootstrap_enabled = False - if "MOBILEDE_CHALLENGE_REFRESH_ATTEMPTS" not in os.environ: - self.challenge_refresh_attempts = 1 - if "MOBILEDE_LISTING_POST_ATTEMPTS" not in os.environ: - self.listing_post_attempts = 2 - if "MOBILEDE_REQUEST_JITTER_MAX_S" not in os.environ: - self.request_jitter_max_s = 0.0 - if "MOBILEDE_DETAIL_RETRIES" not in os.environ: - self.detail_retries = 1 - if "MOBILEDE_LISTING_RETRIES" not in os.environ: - self.listing_retries = 1 - else: - if self.detail_retries is not None and self.detail_retries < 0: - self.detail_retries = None - if self.listing_retries is not None and self.listing_retries < 0: - self.listing_retries = None - # Конфиг прокси. @@ -170,16 +124,6 @@ class ProxyConfig: def enabled(self) -> bool: return bool(self.server) - def to_playwright_dict(self) -> dict[str, str] | None: - if not self.server: - return None - result: dict[str, str] = {"server": self.server} - if self.username: - result["username"] = self.username - if self.password: - result["password"] = self.password - return result - def to_requests_proxy_url(self) -> str | None: if not self.server: return None @@ -211,44 +155,13 @@ class ProxyConfig: @dataclass(slots=True) class Settings: - home_url: str = "https://www.mobile.de/" - default_timeout_ms: int = _env_int("MOBILEDE_TIMEOUT_MS", 45000) - network_settle_ms: int = _env_int("MOBILEDE_NETWORK_SETTLE_MS", 400) - fast_path_timeout_ms: int = _env_int("MOBILEDE_FAST_PATH_TIMEOUT_MS", 15000) - fast_path_max_attempts: int = _env_int("MOBILEDE_FAST_PATH_MAX_ATTEMPTS", 1) - fallback_navigation_timeout_ms: int = _env_int("MOBILEDE_FALLBACK_NAV_TIMEOUT_MS", 15000) - max_retries: int = _env_int("MOBILEDE_MAX_RETRIES", 3) - retry_delay_seconds: float = _env_float("MOBILEDE_RETRY_DELAY_SECONDS", 2.5) - retry_backoff_multiplier: float = _env_float("MOBILEDE_RETRY_BACKOFF_MULTIPLIER", 2.0) - retry_jitter_seconds: float = _env_float("MOBILEDE_RETRY_JITTER_SECONDS", 0.25) headless: bool = _env_bool("MOBILEDE_HEADLESS", True) - browser_engine: str = _env_str("MOBILEDE_BROWSER_ENGINE", "auto") log_level: str = _env_str("MOBILEDE_LOG_LEVEL", "INFO") - log_file: str | None = _env_optional_str("MOBILEDE_LOG_FILE") - enable_trace_id_logs: bool = _env_bool("MOBILEDE_ENABLE_TRACE_ID_LOGS", True) - sync_only_new: bool = _env_bool("MOBILEDE_SYNC_ONLY_NEW", False) - raw_output_json: str | None = _env_optional_str("MOBILEDE_RAW_OUTPUT_JSON") - tokens_file: str | None = _env_path_str("MOBILEDE_TOKENS_FILE") runtime_config_file: str | None = _env_path_str("MOBILEDE_RUNTIME_CONFIG_FILE") - scheduler_interval_minutes: int = _env_int("MOBILEDE_SCHEDULER_INTERVAL_MINUTES", 60) listing: ListingConfig = field(default_factory=ListingConfig) database: DatabaseConfig = field(default_factory=DatabaseConfig) redis: RedisConfig = field(default_factory=RedisConfig) celery: CeleryConfig = field(default_factory=CeleryConfig) - proxy: ProxyConfig = field(default_factory=ProxyConfig) - scraping_profile: ScrapingProfileConfig = field(default_factory=ScrapingProfileConfig) - - @property - def parallel_tabs(self) -> int: - return self.celery.parallel_tabs - - @property - def fetch_concurrency(self) -> int: - return self.celery.fetch_concurrency - - @property - def block_resources(self) -> bool: - return self.celery.block_resources # Глобальные настройки. settings = Settings() diff --git a/mobilede_scraper/core/logs.py b/mobilede_scraper/core/logs.py index 4326ad9..02911a9 100644 --- a/mobilede_scraper/core/logs.py +++ b/mobilede_scraper/core/logs.py @@ -2,12 +2,12 @@ import logging import sys from contextvars import ContextVar -# Храним trace_id текущего потока/корутины. +# Trace ID текущего потока или корутины. TRACE_ID: ContextVar[str] = ContextVar("trace_id", default="-") class TraceIdFilter(logging.Filter): - # Добавляет trace_id в каждую запись лога для сквозной трассировки. + # Добавляет trace_id в запись лога. def filter(self, record: logging.LogRecord) -> bool: record.trace_id = TRACE_ID.get() return True @@ -18,7 +18,7 @@ def set_trace_id(trace_id: str) -> None: def setup_logging(level: str = "INFO", log_file: str | None = None) -> None: - # stderr — Docker и Celery prefork корректно его подхватывают. + # Пишем в stderr для Docker и Celery. handlers: list[logging.Handler] = [logging.StreamHandler(sys.stderr)] if log_file: handlers.append(logging.FileHandler(log_file, encoding="utf-8")) @@ -28,7 +28,7 @@ def setup_logging(level: str = "INFO", log_file: str | None = None) -> None: handler.setLevel(getattr(logging, level.upper(), logging.INFO)) root = logging.getLogger() root.setLevel(getattr(logging, level.upper(), logging.INFO)) - # Убираем старые хендлеры, чтобы не дублировать после fork. + # Убираем старые хендлеры после fork. for old_handler in list(root.handlers): try: old_handler.close() diff --git a/mobilede_scraper/mobile_de/client.py b/mobilede_scraper/mobile_de/client.py index 2f002d2..14c8dd0 100644 --- a/mobilede_scraper/mobile_de/client.py +++ b/mobilede_scraper/mobile_de/client.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging import os import random +import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from collections.abc import Callable, Iterable @@ -33,6 +34,7 @@ MOBILEDE_HTTP_BACKOFF_BASE_SECONDS = max(0.0, float(os.getenv("MOBILEDE_HTTP_BAC MOBILEDE_HTTP_BACKOFF_MAX_SECONDS = max(0.0, float(os.getenv("MOBILEDE_HTTP_BACKOFF_MAX_SECONDS", "20"))) MOBILEDE_HTTP_JITTER_SECONDS = max(0.0, float(os.getenv("MOBILEDE_HTTP_JITTER_SECONDS", "0.5"))) MOBILEDE_HTTP_RETRY_STATUSES = {403, 429, 500, 502, 503, 504} +MOBILEDE_DETAIL_TIMEOUT_SECONDS = max(1, int(float(os.getenv("MOBILEDE_DETAIL_TIMEOUT_SECONDS", "15")))) MOBILEDE_FLARESOLVERR_ENABLED = os.getenv("MOBILEDE_FLARESOLVERR_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_FLARESOLVERR_URL = os.getenv("MOBILEDE_FLARESOLVERR_URL", "http://flaresolverr:8191/v1").strip() MOBILEDE_FLARESOLVERR_TIMEOUT_SECONDS = max(1.0, float(os.getenv("MOBILEDE_FLARESOLVERR_TIMEOUT_SECONDS", "120"))) @@ -241,6 +243,8 @@ class MobileDeClient: page_number: int = 1, *, search_url: str | None = None, + timeout: int = 30, + max_retries: int | None = None, **params: str | int | None, ) -> MobileDeSearchPage: url = ( @@ -248,7 +252,15 @@ class MobileDeClient: if search_url else self.build_search_url(page_number=page_number, **params) ) - html = self.fetch_html(url) + if max_retries is None: + html = self.fetch_html(url, timeout=timeout) + else: + previous_retries = os.environ.get("MOBILEDE_HTTP_MAX_RETRIES") + try: + globals()["MOBILEDE_HTTP_MAX_RETRIES"] = max(0, int(max_retries)) + html = self.fetch_html(url, timeout=timeout) + finally: + globals()["MOBILEDE_HTTP_MAX_RETRIES"] = max(0, int(previous_retries or "4")) raw = extract_search_results(html) listings = [self._map_listing(item) for item in raw.get("listings", []) if isinstance(item, dict)] return MobileDeSearchPage( @@ -364,7 +376,7 @@ class MobileDeClient: return ordered_pages def fetch_detail(self, listing_id: str | int) -> dict: - html = self.fetch_html(self.build_detail_url(listing_id)) + html = self.fetch_html(self.build_detail_url(listing_id), timeout=MOBILEDE_DETAIL_TIMEOUT_SECONDS) return extract_detail_listing(html) @staticmethod diff --git a/mobilede_scraper/mobile_de/flight.py b/mobilede_scraper/mobile_de/flight.py index b85162f..1f017ef 100644 --- a/mobilede_scraper/mobile_de/flight.py +++ b/mobilede_scraper/mobile_de/flight.py @@ -15,7 +15,8 @@ def extract_next_flight_strings(html: str) -> list[str]: try: chunks.append(json.loads(f'"{raw}"')) except json.JSONDecodeError: - # Fallback keeps parser useful if one chunk has non-standard escaping. + # Резервный вариант: сохраняем работоспособность парсера, + # даже если один из фрагментов имеет нестандартное экранирование. chunks.append(raw.encode("utf-8", errors="ignore").decode("unicode_escape", errors="ignore")) return chunks diff --git a/mobilede_scraper/mobile_de/mapper.py b/mobilede_scraper/mobile_de/mapper.py index 93a0855..7f85d14 100644 --- a/mobilede_scraper/mobile_de/mapper.py +++ b/mobilede_scraper/mobile_de/mapper.py @@ -4,6 +4,7 @@ import hashlib import re from datetime import datetime, timezone from typing import Any +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from ..storage.schemas import CarRecord, ImageRecord from .client import MobileDeClient @@ -11,22 +12,47 @@ from .models import MobileDeListing _BODY_MAP = { "cabrio": "OPEN", + "cabriolet": "OPEN", + "кабриолет": "OPEN", "roadster": "OPEN", "кабрио": "OPEN", "limousine": "SEDAN", + "saloon": "SEDAN", "sedan": "SEDAN", - "сeдан": "SEDAN", + "седан": "SEDAN", "suv": "SUV", + "offroad": "SUV", + "gelandewagen": "SUV", + "geländewagen": "SUV", + "pickup": "PICKUP", + "pick-up": "PICKUP", + "пикап": "PICKUP", "внедорож": "SUV", "kombi": "STATION_WAGON", + "estatecar": "STATION_WAGON", + "touring": "STATION_WAGON", "estate": "STATION_WAGON", "универсал": "STATION_WAGON", "van": "MINIVAN", + "kleinbus": "MINIVAN", + "bus": "MINIVAN", + "active tourer": "MINIVAN", + "gran tourer": "MINIVAN", "минивэн": "MINIVAN", + "фургон": "MINIVAN", "coupe": "COUPE", + "sportscar": "COUPE", + "sports car": "COUPE", "купе": "COUPE", "hatchback": "HATCHBACK", "kleinwagen": "HATCHBACK", + "кляйнваген": "HATCHBACK", + "малолитраж": "HATCHBACK", + "хэтч": "HATCHBACK", + "smallcar": "HATCHBACK", + "small car": "HATCHBACK", + "compact": "HATCHBACK", + "compactcar": "HATCHBACK", } _GEARBOX_MAP = { @@ -40,30 +66,87 @@ _GEARBOX_MAP = { _COLOR_MAP = { "schwarz": "black", + "saphirschwarz": "black", + "carbonschwarz": "black", + "obsidianschwarz": "black", + "jet black": "black", + "jetblack": "black", "черн": "black", "black": "black", "weiss": "white", "weiß": "white", + "alpinweiss": "white", + "alpine white": "white", + "mineralweiss": "white", "бел": "white", "white": "white", - "grau": "gray", - "сер": "gray", - "gray": "gray", "silber": "silver", + "argent": "silver", "сереб": "silver", "silver": "silver", + "grau": "gray", + "grey": "gray", + "anthrazit": "gray", + "anthracite": "gray", + "graphit": "gray", + "graphite": "gray", + "spacegrau": "gray", + "brooklyn grau": "gray", + "brooklyn grey": "gray", + "sophistograu": "gray", + "skyscraper grau": "gray", + "сер": "gray", + "gray": "gray", "rot": "red", + "burgundy": "red", + "bordeaux": "red", + "maroon": "red", "красн": "red", "red": "red", "blau": "blue", + "turquoise": "blue", + "cyan": "blue", "син": "blue", "blue": "blue", "grün": "green", "gruen": "green", "зелен": "green", "green": "green", + "braun": "brown", + "корич": "brown", + "brown": "brown", + "beige": "beige", + "champagner": "beige", + "champagne": "beige", + "creme": "beige", + "cream": "beige", + "ivory": "beige", + "беж": "beige", + "gelb": "yellow", + "желт": "yellow", + "yellow": "yellow", + "orange": "orange", + "оранж": "orange", + "gold": "gold", + "золот": "gold", + "bronze": "bronze", + "бронз": "bronze", + "violett": "purple", + "lila": "purple", + "фиолет": "purple", + "purple": "purple", } +_IMAGE_FIELD_HINTS = ["image", "images", "media", "gallery", "photo", "pic", "picture", "url", "src", "uri", "ref"] +_IMAGE_URL_MARKERS = ["img.classistatic.de", "/images/", "/image/", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp"] +_MOBILEDE_IMAGE_RULE = "mo-640.jpg" +_COLOR_VALUE_KEYS = {"ecol", "color", "exteriorcolor", "manufacturercolorname", "vehiclecolor", "paint"} +_COUNTRY_VALUE_KEYS = {"country", "countrycode"} +_BODY_VALUE_KEYS = {"category", "bodytype", "vehiclecategory", "body"} +_ENGINE_VOLUME_VALUE_KEYS = {"cubiccapacity", "enginevolume", "displacement", "enginedisplacement"} +_DRIVE_VALUE_KEYS = {"wheeldrive", "drivetrain", "drive"} +_GEARBOX_VALUE_KEYS = {"transmission", "gearbox", "transmissiontype"} + class MobileDeMapper: """Map mobile.de payloads into CarRecord.""" @@ -78,35 +161,124 @@ class MobileDeMapper: model = self._text(model_payload.get("localized") or self._model_from_title(listing.title, brand) or listing.subtitle or "UNKNOWN") origin_id = self.origin_id(str(listing.id)) title = " ".join(part for part in [listing.title, listing.subtitle] if part) + contact = raw.get("contact") if isinstance(raw.get("contact"), dict) else {} + nested_country = self._find_first_value(raw, _COUNTRY_VALUE_KEYS) + nested_color = self._find_first_value(raw, _COLOR_VALUE_KEYS) + nested_body = self._find_first_value(raw, _BODY_VALUE_KEYS) + nested_engine = self._find_first_value(raw, _ENGINE_VOLUME_VALUE_KEYS) + nested_drive = self._find_first_value(raw, _DRIVE_VALUE_KEYS) + nested_gearbox = self._find_first_value(raw, _GEARBOX_VALUE_KEYS) + attr_country = self._mapping_value(attr, "cn", "countryCode", "country") + attr_color = self._mapping_value(attr, "ecol", "color", "exteriorColor", "manufacturerColorName", "paint") + attr_body = self._mapping_value(attr, "c", "category", "bodyType", "body") + attr_engine = self._mapping_value(attr, "cc", "cubicCapacity", "engineVolume", "displacement", "engineDisplacement") + attr_drive = self._mapping_value(attr, "wd", "wheelDrive", "drivetrain", "drive", "driveType", "antriebsart") + attr_gearbox = self._mapping_value(attr, "tr", "transmission", "gearbox", "transmissionType") + drive_text = " ".join( + part + for part in [ + listing.title, + listing.subtitle, + self._mapping_value(attr, "an"), + attr_drive, + raw.get("wheelDrive"), + raw.get("drivetrain"), + raw.get("drive"), + raw.get("driveType"), + raw.get("modelDescription"), + raw.get("variant"), + raw.get("trim"), + nested_drive, + ] + if isinstance(part, str) and part.strip() + ) + damage_text = self._text( + raw.get("damageCondition") + or attr.get("damageCondition") + or attr.get("dc") + ).lower() + is_damaged = ( + bool(raw.get("hasDamage")) + or ("дтп" in damage_text and "без дтп" not in damage_text) + or ("accident" in damage_text and "no accident" not in damage_text) + ) return CarRecord( parser_id=self._parser_id(origin_id), brand=brand[:50] or "UNKNOWN", model=model[:50] or "UNKNOWN", - year=self._year_from_first_registration(listing.first_registration or attr.get("fr")), - price=self._money_to_int(listing.price or raw.get("p")), + year=self._year_from_first_registration( + self._first_present( + listing.first_registration, + attr.get("fr"), + attr.get("yc"), + raw.get("firstRegistration"), + raw.get("firstRegistrationYear"), + raw.get("year"), + ) + ), + price=self._money_to_int(self._first_present(listing.price, raw.get("p"), raw.get("price"))), currency="EUR", - mileage=self._int_from_text(listing.mileage or attr.get("ml")) or 0, - country=self._normalize_country("DE"), + mileage=self._int_from_text( + self._first_present(listing.mileage, attr.get("ml"), raw.get("mileage")) + ) or 0, + country=self._normalize_country( + self._first_present( + attr_country, + contact.get("countryCode"), + contact.get("country"), + raw.get("countryCode"), + raw.get("country"), + nested_country, + "DE", + ) + ), is_sold=False, - color=self._normalize_color(attr.get("ecol")), - drive=None, - gearbox=self._normalize_gearbox(listing.transmission or attr.get("tr")), + color=self._normalize_color( + self._first_present( + attr_color, + raw.get("color"), + raw.get("manufacturerColorName"), + raw.get("exteriorColor"), + nested_color, + ) + ), + drive=self._normalize_drive(drive_text), + gearbox=self._normalize_gearbox(listing.transmission or attr_gearbox or nested_gearbox), steering_wheel="LEFT", - body_type=self._normalize_body(attr.get("c")), - engine_volume=self._int_from_text(attr.get("cc")), + body_type=self._normalize_body_from_candidates( + attr_body, + raw.get("category"), + raw.get("bodyType"), + nested_body, + listing.subtitle, + listing.title, + ), + engine_volume=self._engine_volume_from_candidates( + attr_engine, + raw.get("cubicCapacity"), + raw.get("cc"), + raw.get("engineVolume"), + raw.get("displacement"), + raw.get("engineDisplacement"), + nested_engine, + raw.get("modelDescription"), + raw.get("variant"), + listing.subtitle, + listing.title, + ), selling_type="CLASSIFIED", - one_owner=(str(attr.get("pvo") or "").strip() == "1"), - new_car=False, + one_owner=self._is_one_owner(attr.get("pvo") or raw.get("numPreviousOwners")), + new_car=bool(raw.get("isNew") or raw.get("isConditionNew")), is_hidden=False, origin="MOBILE_DE", origin_url=listing.url, origin_id=origin_id, - is_damaged=bool(raw.get("hasDamage")), - evaluation=self._text(raw.get("priceRating") or raw.get("rating")) or None, + is_damaged=is_damaged, + evaluation=self._rating_text(raw.get("priceRating") or raw.get("rating")), non_smoking=True, rental=False, - repair_history=bool(raw.get("hasDamage")), + repair_history=is_damaged, slug=self._slugify(title or f"{brand} {model}"), last_seen_at=datetime.now(timezone.utc), images=self._images_from_listing(raw), @@ -127,17 +299,45 @@ class MobileDeMapper: origin_id = self.origin_id(str(listing_id)) price_amount, price_currency = self._detail_price(detail.get("price")) - mileage = self._int_from_text(attrs.get("mileage")) or 0 - year = self._year_from_first_registration(attrs.get("firstRegistration")) - gearbox = self._normalize_gearbox(attrs.get("transmission")) - body_type = self._normalize_body(attrs.get("category") or detail.get("category")) - color = self._normalize_color(attrs.get("color") or attrs.get("manufacturerColorName")) + mileage = self._int_from_text(self._mapping_value(attrs, "mileage")) or 0 + year = self._year_from_first_registration(self._mapping_value(attrs, "firstRegistration", "year", "registrationDate")) + gearbox = self._normalize_gearbox(self._first_present(self._mapping_value(attrs, "transmission", "gearbox", "transmissionType"), detail.get("transmission"), detail.get("gearbox"))) + body_type = self._normalize_body_from_candidates( + self._mapping_value(attrs, "category", "bodyType", "body", "vehicleCategory"), + detail.get("category"), + detail.get("bodyType"), + subtitle, + short_title, + ) + color = self._normalize_color( + self._first_present( + self._mapping_value(attrs, "color", "exteriorColor", "manufacturerColorName", "paint"), + detail.get("color"), + detail.get("manufacturerColorName"), + detail.get("exteriorColor"), + self._find_first_value(detail, _COLOR_VALUE_KEYS), + ) + ) + drive_text = " ".join( + part + for part in [ + short_title, + subtitle, + self._mapping_value(attrs, "wheelDrive", "drivetrain", "drive", "driveType", "antriebsart"), + detail.get("wheelDrive"), + detail.get("drivetrain"), + detail.get("drive"), + detail.get("driveType"), + self._find_first_value(detail, _DRIVE_VALUE_KEYS), + ] + if isinstance(part, str) and part.strip() + ) damage_text = self._text(attrs.get("damageCondition")).lower() is_damaged = ("дтп" in damage_text and "без дтп" not in damage_text) or bool(detail.get("hasDamage")) owners_text = self._text(attrs.get("numPreviousOwners")) - one_owner = owners_text in {"1", "01", "1.0"} + one_owner = self._is_one_owner(owners_text) return CarRecord( parser_id=self._parser_id(origin_id), @@ -150,11 +350,21 @@ class MobileDeMapper: country=self._normalize_country(contact.get("countryCode") or contact.get("country") or "DE"), is_sold=False, color=color, - drive=self._normalize_drive(attrs.get("wheelDrive") or attrs.get("drivetrain")), + drive=self._normalize_drive(drive_text), gearbox=gearbox, steering_wheel="LEFT", body_type=body_type, - engine_volume=self._int_from_text(attrs.get("cubicCapacity") or attrs.get("cc")), + engine_volume=self._engine_volume_from_candidates( + self._mapping_value(attrs, "cubicCapacity", "cc", "engineVolume", "displacement", "engineDisplacement"), + detail.get("cubicCapacity"), + detail.get("cc"), + detail.get("engineVolume"), + detail.get("displacement"), + detail.get("engineDisplacement"), + self._find_first_value(detail, _ENGINE_VOLUME_VALUE_KEYS), + subtitle, + short_title, + ), selling_type="CLASSIFIED", one_owner=one_owner, new_car=bool(detail.get("isNew") or detail.get("isConditionNew")), @@ -163,7 +373,7 @@ class MobileDeMapper: origin_url=MobileDeClient.build_detail_url(listing_id), origin_id=origin_id, is_damaged=is_damaged, - evaluation=self._text(detail.get("priceRating") or detail.get("rating")) or None, + evaluation=self._rating_text(detail.get("priceRating") or detail.get("rating")), non_smoking=True, rental=False, repair_history=is_damaged, @@ -185,6 +395,53 @@ class MobileDeMapper: def _text(value: Any) -> str: return "" if value is None else str(value).strip() + @staticmethod + def _first_present(*values: Any) -> Any: + for value in values: + if value is None: + continue + if isinstance(value, str) and not value.strip(): + continue + return value + return None + + @staticmethod + def _normalized_key(value: Any) -> str: + return re.sub(r"[^a-z0-9]", "", str(value or "").lower()) + + @classmethod + def _find_first_value(cls, value: Any, keys: set[str], *, depth: int = 0) -> Any: + if depth > 6: + return None + if isinstance(value, dict): + for key, item in value.items(): + if cls._normalized_key(key) in keys and cls._first_present(item) is not None: + if not isinstance(item, (dict, list)): + return item + for item in value.values(): + found = cls._find_first_value(item, keys, depth=depth + 1) + if cls._first_present(found) is not None: + return found + elif isinstance(value, list): + for item in value: + found = cls._find_first_value(item, keys, depth=depth + 1) + if cls._first_present(found) is not None: + return found + return None + + @classmethod + def _mapping_value(cls, mapping: Any, *keys: str) -> Any: + if not isinstance(mapping, dict): + return None + for key in keys: + if key in mapping and cls._first_present(mapping.get(key)) is not None: + return mapping.get(key) + normalized_keys = {cls._normalized_key(key) for key in keys if key} + for existing_key, value in mapping.items(): + if cls._normalized_key(existing_key) in normalized_keys and cls._first_present(value) is not None: + return value + return None + @classmethod def _money_to_int(cls, value: Any) -> int | None: if isinstance(value, dict): @@ -231,11 +488,11 @@ class MobileDeMapper: @staticmethod def _normalize_drive(value: Any) -> str | None: text = "" if value is None else str(value).lower() - if any(marker in text for marker in ("front", "fwd", "перед")): + if any(marker in text for marker in ("front", "fwd", "перед", "frontantrieb", "vorderrad", "antrieb vorne", "front-wheel", "front wheel")): return "FWD" - if any(marker in text for marker in ("rear", "rwd", "зад")): + if any(marker in text for marker in ("rear", "rwd", "зад", "heckantrieb", "hinterrad", "antrieb hinten", "rear-wheel", "rear wheel")): return "RWD" - if any(marker in text for marker in ("all", "awd", "4x4", "quattro", "полный")): + if any(marker in text for marker in ("awd", "4wd", "4x4", "quattro", "полный", "xdrive", "4matic", "4motion", "allrad", "all-wheel", "all wheel", "four-wheel", "four wheel")): return "4WD" return None @@ -254,24 +511,116 @@ class MobileDeMapper: return "JP" if text in {"KR", "KOREA", "SOUTH KOREA"}: return "KR" + if re.fullmatch(r"[A-Z]{2}", text): + return text return "NA" @staticmethod def _normalize_body(value: Any) -> str: text = "" if value is None else str(value).lower() + text = text.replace("ё", "е") + if re.search(r"\bbmw\s+x(?:[1-7]|m)\b", text): + return "SUV" for marker, mapped in _BODY_MAP.items(): if marker in text: return mapped return "OTHER" + @classmethod + def _normalize_body_from_candidates(cls, *values: Any) -> str: + for value in values: + normalized = cls._normalize_body(value) + if normalized != "OTHER": + return normalized + return "OTHER" + + @classmethod + def _engine_volume_from_candidates(cls, *values: Any) -> int | None: + fallback_texts: list[str] = [] + free_text_start = max(0, len(values) - 2) + for index, value in enumerate(values): + if value is None: + continue + if isinstance(value, str): + text = value.strip() + if not text: + continue + fallback_texts.append(text) + if index < free_text_start and re.fullmatch(r"[\d\s.,]+", text): + parsed = cls._int_from_text(text) + if parsed and 500 <= parsed <= 9000: + return parsed + if re.search(r"(ccm|cm3|cm³|\bcc\b|куб|cubic|displacement)", text, re.IGNORECASE): + parsed = cls._engine_volume_from_cc_text(text) + if parsed: + return parsed + if re.search(r"\b(?:liter|litre|l)\b", text, re.IGNORECASE): + parsed = cls._engine_volume_from_liter_text(text) + if parsed: + return parsed + continue + parsed = cls._int_from_text(value) + if parsed and 500 <= parsed <= 9000: + return parsed + + for text in fallback_texts: + match = re.search(r"(? int | None: + text = str(value or "") + match = re.search(r"(\d{1,2}(?:[\s.,]\d{3})|\d{3,5})\s*(?:ccm|cm3|cm³|cc|куб)", text, re.IGNORECASE) + if not match: + return None + parsed = cls._int_from_text(match.group(1)) + if parsed and 500 <= parsed <= 9000: + return parsed + return None + + @classmethod + def _engine_volume_from_liter_text(cls, value: str) -> int | None: + text = str(value or "") + match = re.search(r"(? str: text = "" if value is None else str(value).lower().strip() + text = text.replace("ё", "е") + text = text.replace("ä", "a").replace("ö", "o").replace("ü", "u").replace("ß", "ss") + text = re.sub(r"[_\-/]+", " ", text) for marker, mapped in _COLOR_MAP.items(): if marker in text: return mapped return text[:50] if text else "other" + @staticmethod + def _is_one_owner(value: Any) -> bool: + text = "" if value is None else str(value).strip().lower() + return text in {"1", "01", "1.0", "one", "one owner", "1 owner", "1 previous owner"} + + @staticmethod + def _rating_text(value: Any) -> str | None: + if isinstance(value, dict): + for key in ("rating", "ratingLabel", "label", "value"): + text = MobileDeMapper._text(value.get(key)) + if text: + return text + return None + text = MobileDeMapper._text(value) + return text or None + @staticmethod def _slugify(value: str) -> str: slug = re.sub(r"[^a-zA-Z0-9а-яА-ЯёЁ]+", "-", value.lower()).strip("-") @@ -306,38 +655,69 @@ class MobileDeMapper: @staticmethod def _images_from_listing(raw: dict[str, Any]) -> list[ImageRecord]: - urls: list[str] = [] - image = raw.get("image") - if isinstance(image, str): - urls.append(MobileDeMapper._normalize_image_url(image)) - images = raw.get("images") - if isinstance(images, list): - for item in images: - if isinstance(item, str): - urls.append(MobileDeMapper._normalize_image_url(item)) - elif isinstance(item, dict): - src = item.get("src") or item.get("url") or item.get("uri") - if src: - urls.append(MobileDeMapper._normalize_image_url(str(src))) - media_gallery = raw.get("mediaGallery") - if isinstance(media_gallery, list): - for item in media_gallery: - if isinstance(item, dict): - src = item.get("uri") or item.get("url") - if src: - urls.append(MobileDeMapper._normalize_image_url(str(src))) + urls = MobileDeMapper._extract_image_urls(raw) return [ ImageRecord(fullres_image=url, preview_image=url, order_index=index) for index, url in enumerate(dict.fromkeys(url for url in urls if url)) ] + @staticmethod + def _extract_image_urls(value: Any, *, parent_key: str = "", depth: int = 0) -> list[str]: + if depth > 8: + return [] + urls: list[str] = [] + parent_hint = MobileDeMapper._has_image_field_hint(parent_key) + if isinstance(value, str): + if parent_hint or MobileDeMapper._looks_like_image_url(value): + normalized = MobileDeMapper._normalize_image_url(value) + if MobileDeMapper._looks_like_image_url(normalized): + urls.append(normalized) + return urls + if isinstance(value, list): + for item in value: + urls.extend(MobileDeMapper._extract_image_urls(item, parent_key=parent_key, depth=depth + 1)) + return urls + if isinstance(value, dict): + for key, item in value.items(): + key_text = str(key or "") + urls.extend(MobileDeMapper._extract_image_urls(item, parent_key=key_text, depth=depth + 1)) + return urls + return urls + + @staticmethod + def _has_image_field_hint(value: str) -> bool: + normalized = re.sub(r"[^a-z0-9]", "", str(value or "").lower()) + return normalized in _IMAGE_FIELD_HINTS or any(hint in normalized for hint in _IMAGE_FIELD_HINTS) + + @staticmethod + def _looks_like_image_url(value: str) -> bool: + text = str(value or "").strip().lower() + if not text: + return False + if not (text.startswith("http://") or text.startswith("https://") or text.startswith("//") or text.startswith("/")): + return False + return any(marker in text for marker in _IMAGE_URL_MARKERS) + @staticmethod def _normalize_image_url(value: str) -> str: url = str(value).strip() if not url: return "" if url.startswith("//"): - return f"https:{url}" - if url.startswith("http://") or url.startswith("https://"): + url = f"https:{url}" + elif not (url.startswith("http://") or url.startswith("https://")): + url = f"https://{url.lstrip('/')}" + return MobileDeMapper._normalize_mobilede_image_rule(url) + + @staticmethod + def _normalize_mobilede_image_rule(url: str) -> str: + parsed = urlsplit(url) + if "img.classistatic.de" not in parsed.netloc.lower(): return url - return f"https://{url.lstrip('/')}" + if "/api/v1/mo-prod/images/" not in parsed.path: + return url + query_pairs = parse_qsl(parsed.query, keep_blank_values=True) + if any(key.lower() == "rule" for key, _value in query_pairs): + return url + query_pairs.append(("rule", _MOBILEDE_IMAGE_RULE)) + return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, urlencode(query_pairs), parsed.fragment)) diff --git a/mobilede_scraper/mobile_de/scraper.py b/mobilede_scraper/mobile_de/scraper.py index 733d607..2b257fa 100644 --- a/mobilede_scraper/mobile_de/scraper.py +++ b/mobilede_scraper/mobile_de/scraper.py @@ -21,6 +21,10 @@ logger = logging.getLogger("mobile_de.scraper") MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK = max(0, int(os.getenv("MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK", "2"))) MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS = max(0, int(os.getenv("MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS", "1"))) MOBILEDE_LIGHT_REFRESH_EXISTING = os.getenv("MOBILEDE_LIGHT_REFRESH_EXISTING", "false").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_SELECTIVE_DETAIL_ENRICH_ENABLED = os.getenv("MOBILEDE_SELECTIVE_DETAIL_ENRICH_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_DETAIL_ENRICH_IMAGES_ENABLED = os.getenv("MOBILEDE_DETAIL_ENRICH_IMAGES_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_SELECTIVE_DETAIL_ENRICH_MAX_PER_RUN = max(0, int(os.getenv("MOBILEDE_SELECTIVE_DETAIL_ENRICH_MAX_PER_RUN", "20"))) +MOBILEDE_SELECTIVE_DETAIL_ENRICH_MAX_PER_PAGE = max(0, int(os.getenv("MOBILEDE_SELECTIVE_DETAIL_ENRICH_MAX_PER_PAGE", "2"))) MOBILEDE_SEARCH_STRATEGY_NOTE = ( "mobile.de search pages return about 20 listings per page and are limited to about 50 pages; " "for full coverage split into narrower segments and deduplicate by id." @@ -94,6 +98,32 @@ class MobileDeScraper: deduped_page_records.append(record) return deduped_page_records + @staticmethod + def _record_key(record: CarRecord) -> str: + return record.origin_id or record.origin_url + + @staticmethod + def _record_needs_detail_enrich(record: CarRecord) -> bool: + return any( + ( + record.year is None, + record.mileage == 0, + record.engine_volume is None, + record.body_type == "OTHER", + record.color == "other", + ) + ) + + @staticmethod + def _detail_enrich_priority(record: CarRecord) -> tuple[int, int, int, int, int]: + return ( + int(record.year is None), + int(record.engine_volume is None), + int(record.mileage == 0), + int(record.body_type == "OTHER"), + int(record.color == "other"), + ) + def _apply_only_new_page_policy( self, *, @@ -310,6 +340,8 @@ class MobileDeScraper: inserted_total = 0 updated_total = 0 images_upserted = 0 + detail_enriched = 0 + detail_enrich_failed = 0 existing_streak = 0 new_records_kept = 0 head_cut_triggered = False @@ -348,19 +380,36 @@ class MobileDeScraper: pages_collected = 0 early_stopped = False - for page in self.client.iter_search_pages( - start_page=start_page, - max_pages=max_pages, - search_url=search_url, - progress_callback=_on_page, - **params, - ): + concurrent_pages = max(1, int(os.getenv("MOBILEDE_CONCURRENT_PAGES", "1"))) + if concurrent_pages > 1 and max_pages and max_pages > 1 and only_new is not True: + page_iterator = self.client.fetch_search_pages_concurrent( + start_page=start_page, + max_pages=max_pages, + workers=concurrent_pages, + search_url=search_url, + progress_callback=_on_page, + **params, + ) + else: + page_iterator = self.client.iter_search_pages( + start_page=start_page, + max_pages=max_pages, + search_url=search_url, + progress_callback=_on_page, + **params, + ) + for page in page_iterator: pages_collected += 1 pages_payload.append(asdict(page)) listing_count += len(page.listings) unique_ids.update(str(listing.id) for listing in page.listings if listing.id) page_records = [self.mapper.listing_to_car_record(listing) for listing in page.listings] + listing_by_record_key = { + self._record_key(record): listing + for listing, record in zip(page.listings, page_records, strict=False) + if self._record_key(record) + } page_records = self._dedupe_page_records(page_records, seen_record_keys) for record in page_records: record.is_sold = False @@ -369,7 +418,7 @@ class MobileDeScraper: record.sold_at = None record.skip_image_sync = False existing_origin_ids: set[str] = set() - if page_records and (only_new or MOBILEDE_LIGHT_REFRESH_EXISTING): + if page_records and (only_new or MOBILEDE_LIGHT_REFRESH_EXISTING or MOBILEDE_SELECTIVE_DETAIL_ENRICH_ENABLED): existing_origin_ids = self.persistence.get_existing_origin_ids( [record.origin_id for record in page_records if record.origin_id] ) @@ -389,6 +438,66 @@ class MobileDeScraper: for record in page_records: if record.origin_id and record.origin_id in existing_origin_ids: record.skip_image_sync = True + if ( + MOBILEDE_SELECTIVE_DETAIL_ENRICH_ENABLED + and detail_enriched < MOBILEDE_SELECTIVE_DETAIL_ENRICH_MAX_PER_RUN + and page_records + ): + remaining_budget = MOBILEDE_SELECTIVE_DETAIL_ENRICH_MAX_PER_RUN - detail_enriched + page_budget = min(MOBILEDE_SELECTIVE_DETAIL_ENRICH_MAX_PER_PAGE, remaining_budget) + candidate_records = [ + record + for record in page_records + if record.origin_id + and record.origin_id not in existing_origin_ids + and self._record_needs_detail_enrich(record) + ] + candidate_records.sort(key=self._detail_enrich_priority, reverse=True) + selected_keys = { + self._record_key(record) + for record in candidate_records[:page_budget] + } + if selected_keys: + enriched_records: list[CarRecord] = [] + for record in page_records: + record_key = self._record_key(record) + listing = listing_by_record_key.get(record_key) + if record_key not in selected_keys or listing is None: + enriched_records.append(record) + continue + try: + if progress_callback is not None: + progress_callback( + "detail_enriching", + { + "run_id": run_id, + "pages_collected": pages_collected, + "page_number": page.page_number, + "listing_id": str(listing.id), + "detail_enriched": detail_enriched, + "detail_enrich_failed": detail_enrich_failed, + }, + ) + enriched = self.mapper.detail_to_car_record(str(listing.id), self.client.fetch_detail(listing.id)) + enriched.is_sold = False + enriched.first_seen_at = run_seen_at + enriched.last_seen_at = run_seen_at + enriched.sold_at = None + enriched.skip_image_sync = False + if not MOBILEDE_DETAIL_ENRICH_IMAGES_ENABLED: + enriched.images = record.images + enriched_records.append(enriched) + detail_enriched += 1 + except Exception: + logger.warning( + "mobile.de selective detail enrich failed: listing_id=%s title=%s", + getattr(listing, "id", None), + getattr(listing, "title", None), + exc_info=True, + ) + detail_enrich_failed += 1 + enriched_records.append(record) + page_records = enriched_records if progress_callback is not None: progress_callback( @@ -396,6 +505,8 @@ class MobileDeScraper: { "record_count": len(page_records), "skipped_existing": skipped_existing, + "detail_enriched": detail_enriched, + "detail_enrich_failed": detail_enrich_failed, "only_new": bool(only_new), "run_id": run_id, "pages_collected": pages_collected, @@ -486,6 +597,8 @@ class MobileDeScraper: "updated": updated_total, "images_upserted": images_upserted, }, + "detail_enriched": detail_enriched, + "detail_enrich_failed": detail_enrich_failed, "skipped_existing": skipped_existing, **data, } diff --git a/mobilede_scraper/worker/celery_app.py b/mobilede_scraper/worker/celery_app.py index 3a2b0f3..882e871 100644 --- a/mobilede_scraper/worker/celery_app.py +++ b/mobilede_scraper/worker/celery_app.py @@ -1,4 +1,4 @@ -# Инициализация Celery-приложения и периодических задач. +# Celery-приложение и периодические задачи. import json import logging @@ -11,12 +11,22 @@ from redis import Redis from ..core.config import settings from ..core.logs import setup_logging -from .constants import GLOBAL_DB_PROGRESS_TS_KEY, GLOBAL_PROGRESS_TS_KEY +from .constants import ( + GLOBAL_DB_PROGRESS_TS_KEY, + GLOBAL_PROGRESS_TS_KEY, + MOBILEDE_RUNTIME_SEGMENTS_TASK, + MOBILEDE_SYNC_QUEUE, + MOBILEDE_SYNC_TASK_NAME, +) logger = logging.getLogger("mobilede_scraper.worker.celery_app") STARTUP_SYNC_DISPATCH_KEY = "mobilede:state:startup_sync_dispatched" -MOBILEDE_SYNC_QUEUE = "mobilede_sync" PROGRESS_KEY_PREFIX = "mobilede:state:task_progress:" +MOBILEDE_SYNC_DETAIL_TASK = "mobilede.sync_detail" +MOBILEDE_ENRICH_IMAGES_TASK = "mobilede.enrich_images_batch" +STARTUP_SYNC_DISPATCH_TTL_SECONDS = 10 * 60 +STARTUP_PROGRESS_MAX_AGE_SECONDS = 180 +STARTUP_GLOBAL_PROGRESS_MAX_AGE_SECONDS = 300 def _env_bool(name: str, default: bool) -> bool: @@ -27,7 +37,23 @@ def _env_bool(name: str, default: bool) -> bool: MOBILEDE_BEAT_SYNC_ENABLED = _env_bool("MOBILEDE_BEAT_SYNC_ENABLED", True) -def _has_fresh_active_progress(redis_client: Redis, *, max_age_seconds: int = 180) -> bool: +def _runtime_sync_kwargs() -> dict[str, bool | float]: + return { + "delay_seconds": float(os.getenv("MOBILEDE_REQUEST_DELAY_SECONDS", "0.7")), + "use_cursor": _env_bool("MOBILEDE_CURSOR_ENABLED", True), + "continuous": _env_bool("MOBILEDE_CONTINUOUS_SYNC_ENABLED", True), + } + + +def _runtime_sync_expires_seconds() -> float: + return settings.celery.beat_sync_interval_minutes * 60.0 + + +def _has_fresh_active_progress( + redis_client: Redis, + *, + max_age_seconds: int = STARTUP_PROGRESS_MAX_AGE_SECONDS, +) -> bool: now = int(time.time()) try: for raw_key in redis_client.scan_iter(f"{PROGRESS_KEY_PREFIX}*"): @@ -47,7 +73,11 @@ def _has_fresh_active_progress(redis_client: Redis, *, max_age_seconds: int = 18 return False -def _has_recent_global_progress(redis_client: Redis, *, max_age_seconds: int = 300) -> bool: +def _has_recent_global_progress( + redis_client: Redis, + *, + max_age_seconds: int = STARTUP_GLOBAL_PROGRESS_MAX_AGE_SECONDS, +) -> bool: now = int(time.time()) try: progress_ts = int(redis_client.get(GLOBAL_PROGRESS_TS_KEY) or 0) @@ -59,17 +89,40 @@ def _has_recent_global_progress(redis_client: Redis, *, max_age_seconds: int = 3 return freshest_ts > 0 and now - freshest_ts <= max_age_seconds +def _has_live_startup_progress(redis_client: Redis, *, queue_len: int) -> bool: + if _has_recent_global_progress(redis_client): + return True + if queue_len <= 0: + return False + return _has_fresh_active_progress(redis_client) + + +def _claim_startup_dispatch(redis_client: Redis, *, reset_stale: bool) -> bool: + if redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=STARTUP_SYNC_DISPATCH_TTL_SECONDS): + return True + if not reset_stale: + return False + redis_client.delete(STARTUP_SYNC_DISPATCH_KEY) + return bool( + redis_client.set( + STARTUP_SYNC_DISPATCH_KEY, + "1", + nx=True, + ex=STARTUP_SYNC_DISPATCH_TTL_SECONDS, + ) + ) + + @celery_setup_logging.connect def _configure_logging(loglevel=None, **kwargs): - # Перехватываем логирование Celery и пишем только в stderr (Docker logs). + # Пишем логи Celery в stderr. level = settings.log_level if settings.log_level else "INFO" setup_logging(level, None) @worker_process_init.connect def _on_worker_process_init(**kwargs): - # Повторно настраиваем логирование в каждом дочернем prefork-процессе, - # чтобы StreamHandler(stderr) корректно работал после fork. + # Повторно настраиваем логирование после fork. level = settings.log_level if settings.log_level else "INFO" setup_logging(level, None) @@ -88,8 +141,7 @@ celery_app = Celery( backend=_result_backend(), ) -# Auto-clamp: если hard limit слишком далёк от soft (> soft + 120), -# ограничиваем, чтобы зависший worker не жил вечно. +# Если hard limit слишком большой, сжимаем его до soft + 120. _soft = settings.celery.task_soft_time_limit _hard = settings.celery.task_time_limit _max_hard = _soft + 120 if _soft else _hard @@ -105,17 +157,13 @@ beat_schedule = {} if MOBILEDE_BEAT_SYNC_ENABLED: beat_schedule = { "periodic-mobilede-sync-search": { - "task": "mobilede.sync_runtime_segments", - "schedule": settings.celery.beat_sync_interval_minutes * 60.0, + "task": MOBILEDE_RUNTIME_SEGMENTS_TASK, + "schedule": _runtime_sync_expires_seconds(), "args": (), - "kwargs": { - "delay_seconds": float(os.getenv("MOBILEDE_REQUEST_DELAY_SECONDS", "0.7")), - "use_cursor": _env_bool("MOBILEDE_CURSOR_ENABLED", True), - "continuous": _env_bool("MOBILEDE_CONTINUOUS_SYNC_ENABLED", True), - }, + "kwargs": _runtime_sync_kwargs(), "options": { "queue": MOBILEDE_SYNC_QUEUE, - "expires": settings.celery.beat_sync_interval_minutes * 60.0, + "expires": _runtime_sync_expires_seconds(), }, } } @@ -144,9 +192,10 @@ celery_app.conf.update( worker_hijack_root_logger=False, beat_schedule=beat_schedule, task_routes={ - "mobilede.sync_runtime_segments": {"queue": MOBILEDE_SYNC_QUEUE}, - "mobilede.sync_search": {"queue": MOBILEDE_SYNC_QUEUE}, - "mobilede.sync_detail": {"queue": MOBILEDE_SYNC_QUEUE}, + MOBILEDE_RUNTIME_SEGMENTS_TASK: {"queue": MOBILEDE_SYNC_QUEUE}, + MOBILEDE_SYNC_TASK_NAME: {"queue": MOBILEDE_SYNC_QUEUE}, + MOBILEDE_SYNC_DETAIL_TASK: {"queue": MOBILEDE_SYNC_QUEUE}, + MOBILEDE_ENRICH_IMAGES_TASK: {"queue": MOBILEDE_SYNC_QUEUE}, "mobilede_scraper.worker.tasks.*": {"queue": MOBILEDE_SYNC_QUEUE}, }, ) @@ -161,6 +210,7 @@ def _on_worker_ready(**kwargs): logger.info("Worker ready: startup sync dispatch disabled by MOBILEDE_STARTUP_SYNC_ENABLED") return + should_dispatch = False redis_client = None try: redis_client = Redis.from_url( @@ -172,29 +222,27 @@ def _on_worker_ready(**kwargs): retry_on_timeout=True, ) - has_fresh_progress = _has_fresh_active_progress(redis_client) - has_recent_global_progress = _has_recent_global_progress(redis_client) - has_live_progress = bool(has_fresh_progress or has_recent_global_progress) + queue_len = int(redis_client.llen(MOBILEDE_SYNC_QUEUE) or 0) + has_live_progress = _has_live_startup_progress(redis_client, queue_len=queue_len) - try: - queue_len = int(redis_client.llen(MOBILEDE_SYNC_QUEUE) or 0) - except Exception: - queue_len = 0 if queue_len > 0 and has_live_progress: logger.info("Worker ready: MOBILEDE_sync queue already has %d task(s); skip startup dispatch", queue_len) return + has_fresh_progress = _has_fresh_active_progress(redis_client) if queue_len > 0 and not has_live_progress: logger.warning( "Worker ready: MOBILEDE_sync queue has %d task(s), but no fresh progress is visible; forcing runtime sync dispatch", queue_len, ) + elif queue_len <= 0 and has_fresh_progress: + logger.info("Worker ready: queue is empty but active progress is still visible; relying on startup dedupe") - should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=600)) - if not should_dispatch and not has_live_progress: - redis_client.delete(STARTUP_SYNC_DISPATCH_KEY) - should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=600)) - if should_dispatch: - logger.info("Worker ready: stale startup dedupe key ignored because queue is empty and no fresh active progress exists") + should_dispatch = _claim_startup_dispatch( + redis_client, + reset_stale=not has_live_progress, + ) + if should_dispatch and not has_live_progress: + logger.info("Worker ready: claimed startup dispatch after stale or missing dedupe state") except Exception: logger.warning("Worker ready startup sync dedupe check failed; skipping immediate dispatch", exc_info=True) return @@ -211,12 +259,8 @@ def _on_worker_ready(**kwargs): logger.info("Worker ready - dispatching initial mobile.de sync_runtime_segments task") celery_app.send_task( - "mobilede.sync_runtime_segments", - kwargs={ - "delay_seconds": float(os.getenv("MOBILEDE_REQUEST_DELAY_SECONDS", "0.7")), - "use_cursor": _env_bool("MOBILEDE_CURSOR_ENABLED", True), - "continuous": _env_bool("MOBILEDE_CONTINUOUS_SYNC_ENABLED", True), - }, + MOBILEDE_RUNTIME_SEGMENTS_TASK, + kwargs=_runtime_sync_kwargs(), queue=MOBILEDE_SYNC_QUEUE, - expires=settings.celery.beat_sync_interval_minutes * 60.0, + expires=_runtime_sync_expires_seconds(), ) diff --git a/mobilede_scraper/worker/constants.py b/mobilede_scraper/worker/constants.py index 0e3acda..9208ef9 100644 --- a/mobilede_scraper/worker/constants.py +++ b/mobilede_scraper/worker/constants.py @@ -16,6 +16,7 @@ MOBILEDE_CONTINUOUS_SYNC_ENABLED = os.getenv("MOBILEDE_CONTINUOUS_SYNC_ENABLED", 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_ANTIBOT_BACKOFF_SECONDS = max(300, int(float(os.getenv("MOBILEDE_ANTIBOT_BACKOFF_SECONDS", "1800")))) 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"} @@ -31,6 +32,7 @@ MOBILEDE_DYNAMIC_SEGMENT_PROBES = os.getenv("MOBILEDE_DYNAMIC_SEGMENT_PROBES", " 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_ADAPTIVE_URL_MAX_SECONDS = max(0, int(float(os.getenv("MOBILEDE_ADAPTIVE_URL_MAX_SECONDS", "300")))) MOBILEDE_PREPLAN_SPLIT_THRESHOLD_RATIO = min( 5.0, max(1.0, float(os.getenv("MOBILEDE_PREPLAN_SPLIT_THRESHOLD_RATIO", "2.5"))), @@ -127,6 +129,30 @@ MOBILEDE_SEGMENT_TINY_RATIO = min( 0.8, max(0.1, float(os.getenv("MOBILEDE_SEGMENT_TINY_RATIO", "0.45"))), ) +MOBILEDE_OVERFLOW_MIN_YEAR_SPLIT_SPAN = max( + 1, + min(8, int(os.getenv("MOBILEDE_OVERFLOW_MIN_YEAR_SPLIT_SPAN", "4"))), +) +MOBILEDE_OVERFLOW_YEAR_DEEP_SPLIT_DEPTH = max( + 0, + min(6, int(os.getenv("MOBILEDE_OVERFLOW_YEAR_DEEP_SPLIT_DEPTH", "1"))), +) +MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY = max( + 0, + int(os.getenv("MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY", "220")), +) +MOBILEDE_OVERFLOW_SCORE_YEAR_PENALTY = max( + 0, + int(os.getenv("MOBILEDE_OVERFLOW_SCORE_YEAR_PENALTY", "320")), +) +MOBILEDE_OVERFLOW_SCORE_MILEAGE_PENALTY = max( + 0, + int(os.getenv("MOBILEDE_OVERFLOW_SCORE_MILEAGE_PENALTY", "80")), +) +MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY = max( + 0, + int(os.getenv("MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY", "420")), +) MOBILEDE_OVERFLOW_MIN_USEFUL_CHILD_RATIO = min( 1.0, max(0.2, float(os.getenv("MOBILEDE_OVERFLOW_MIN_USEFUL_CHILD_RATIO", "0.55"))), @@ -153,6 +179,7 @@ STALL_WATCHDOG_NAVIGATION_STAGES = { STALL_WATCHDOG_LONG_RUNNING_STAGES = { "search_collection_done", "records_mapped", + "detail_enriching", } STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS = max( 300, diff --git a/mobilede_scraper/worker/planner.py b/mobilede_scraper/worker/planner.py index e36cfb0..d509a85 100644 --- a/mobilede_scraper/worker/planner.py +++ b/mobilede_scraper/worker/planner.py @@ -1,9 +1,24 @@ from __future__ import annotations -# Segment planning and overflow splitting live here to keep tasks.py focused -# on Celery orchestration. The module intentionally reuses task-layer -# helpers and globals that are synchronized from tasks.py before calls. +# Планирование сегментов вынесено сюда, чтобы `tasks.py` оставался короче. +# Модуль использует хелперы из `tasks.py`. from .tasks import * # noqa: F401,F403 +from .tasks import ( + _mobilede_make_segment_url, + _mobilede_segment_fingerprint, + _mobilede_segment_key, + _mobilede_segment_label, + _mobilede_short_segment_label, + _mobilede_short_segment_ref, + _mobilede_should_skip_dynamic_segment, + _mobilede_should_skip_planned_segment, + _mobilede_site_make_options_cache, + _mobilede_try_mark_bootstrap_segment_dispatched, + _mobilede_url_query_values, +) + + +_MOBILEDE_RUNTIME_PLAN_VERSION = 2 def _mobilede_price_ranges() -> list[tuple[int, int | None]]: @@ -40,10 +55,10 @@ def _mobilede_price_ranges() -> list[tuple[int, int | None]]: def _mobilede_price_ranges_for_segment(segment: dict[str, object] | None = None) -> list[tuple[int, int | None]]: - make_id = str((segment or {}).get("make_id") or "").strip() - if make_id == "3500" and MOBILEDE_COMPACT_SEGMENTS: - # Используем границы, близкие к самим фильтрам mobile.de, чтобы BMW - # сразу попадали в более плотные корзины и реже дорезались в хвосте. + del segment + if MOBILEDE_COMPACT_SEGMENTS and os.getenv("MOBILEDE_DENSE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}: + # Плотные корзины включаются для всех марок из пользовательской ссылки. + # Так покрытие не зависит от конкретных make_id и не теряет хвосты за лимитом 50 страниц. return [ (1, 5000), (5001, 10000), @@ -63,6 +78,14 @@ def _mobilede_price_ranges_for_segment(segment: dict[str, object] | None = None) return _mobilede_price_ranges() +def _mobilede_filtered_url_uses_adaptive_plan() -> bool: + return os.getenv("MOBILEDE_FILTERED_URL_ADAPTIVE_PLAN", "false").strip().lower() in {"1", "true", "yes", "on"} + + +def _mobilede_skip_late_overflow_children_during_bootstrap() -> bool: + return os.getenv("MOBILEDE_SKIP_LATE_OVERFLOW_CHILDREN_DURING_BOOTSTRAP", "true").strip().lower() in {"1", "true", "yes", "on"} + + def _mobilede_year_ranges() -> list[tuple[int | None, int | None]]: if MOBILEDE_COMPACT_SEGMENTS: return [(None, 2009), (2010, 2017), (2018, 2022), (2023, None)] @@ -93,8 +116,8 @@ def _mobilede_year_ranges_for_segment_price( price_min: int, price_max: int | None, ) -> list[tuple[int | None, int | None]]: - make_id = str((segment or {}).get("make_id") or "").strip() - if make_id != "3500": + del segment + if not (MOBILEDE_COMPACT_SEGMENTS and os.getenv("MOBILEDE_DENSE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}): return _mobilede_year_ranges_for_price(price_min, price_max) upper_bound = int(price_max) if price_max is not None else int(price_min) @@ -107,6 +130,35 @@ def _mobilede_year_ranges_for_segment_price( return _mobilede_year_ranges_for_price(price_min, price_max) +def _mobilede_interleave_segments_by_make(segments: list[dict[str, object]]) -> list[dict[str, object]]: + """Mix make blocks so an early bootstrap pass covers the whole source URL.""" + groups: dict[str, list[dict[str, object]]] = {} + order: list[str] = [] + passthrough: list[dict[str, object]] = [] + for segment in segments: + make_id = str(segment.get("make_id") or "").strip() + if not make_id: + passthrough.append(segment) + continue + if make_id not in groups: + groups[make_id] = [] + order.append(make_id) + groups[make_id].append(segment) + + if len(order) <= 1: + return segments + + mixed: list[dict[str, object]] = [] + max_len = max(len(items) for items in groups.values()) + for index in range(max_len): + for make_id in order: + items = groups[make_id] + if index < len(items): + mixed.append(items[index]) + mixed.extend(passthrough) + return mixed + + def _mobilede_mileage_ranges() -> list[tuple[int | None, int | None]]: if MOBILEDE_COMPACT_SEGMENTS: return [(None, 100000), (100001, 200000), (200001, None)] @@ -125,8 +177,7 @@ def _mobilede_should_pre_split_mileage( if MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE: return True - # Restore the older balanced plan: split mileage only in buckets that are - # usually dense enough to hit mobile.de's 50-page cap. + # Делим по пробегу только плотные корзины. if price_max is not None and price_max <= MOBILEDE_HOT_OLD_CHEAP_PRICE_MAX: return year_max is not None and year_max <= 2009 @@ -310,7 +361,7 @@ def _mobilede_load_learned_runtime_segments(source_segments: list[dict[str, obje payload = json.load(fh) if not isinstance(payload, dict): return None - if payload.get("plan_version") != 1: + if payload.get("plan_version") != _MOBILEDE_RUNTIME_PLAN_VERSION: return None expected = _mobilede_segments_source_fingerprint(source_segments) if str(payload.get("source_fingerprint") or "") != expected: @@ -346,7 +397,7 @@ def _mobilede_save_learned_runtime_segments( os.makedirs(os.path.dirname(path) or ".", exist_ok=True) pruned_segments = _mobilede_prune_overflow_parent_segments(runtime_segments) payload = { - "plan_version": 1, + "plan_version": _MOBILEDE_RUNTIME_PLAN_VERSION, "source_fingerprint": _mobilede_segments_source_fingerprint(source_segments), "updated_at": datetime.now(timezone.utc).isoformat(), "segments": pruned_segments, @@ -602,16 +653,36 @@ def _mobilede_split_price_ranges_for_overflow( left = int(price_min or 1) right = price_max if right is None: - step = max(5000, min(50000, left)) - pivot = left + step - return [(left, pivot), (pivot + 1, None)] + child_budget = max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)) + step = max(MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN, 5000, min(50000, left)) + ranges: list[tuple[int, int | None]] = [] + current = left + for _index in range(child_budget - 1): + upper = current + step + ranges.append((current, upper)) + current = upper + 1 + step = min(step * 2, 100000) + ranges.append((current, None)) + return ranges span = int(right) - int(left) if span < MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN: return [] - pivot = int(left) + span // 2 - if pivot <= int(left) or pivot >= int(right): - return [] - return [(int(left), pivot), (pivot + 1, int(right))] + child_count = min( + max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)), + max(2, span // MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN + 1), + ) + step = max(1, span // child_count) + ranges = [] + current = int(left) + for index in range(child_count): + upper = int(right) if index == child_count - 1 else min(int(right), current + step) + if upper < current: + break + ranges.append((current, upper)) + current = upper + 1 + if current > int(right): + break + return ranges def _mobilede_split_mileage_ranges_for_overflow( @@ -632,9 +703,16 @@ def _mobilede_split_mileage_ranges_for_overflow( return [(None, pivot), (pivot + 1, right)] if mileage_min is not None and mileage_max is None: left = int(mileage_min) + child_budget = max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)) step = max(25000, min(100000, left)) - pivot = left + step - return [(left, pivot), (pivot + 1, None)] + ranges: list[tuple[int | None, int | None]] = [] + current = left + for _index in range(child_budget - 1): + upper = current + step + ranges.append((current, upper)) + current = upper + 1 + ranges.append((current, None)) + return ranges assert mileage_min is not None and mileage_max is not None left = int(mileage_min) @@ -642,10 +720,19 @@ def _mobilede_split_mileage_ranges_for_overflow( span = right - left if span < 10000: return _mobilede_split_fine_mileage_ranges(left, right) - pivot = left + span // 2 - if pivot <= left or pivot >= right: - return [] - return [(left, pivot), (pivot + 1, right)] + child_count = min(max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)), max(2, span // 10000 + 1)) + step = max(1, span // child_count) + ranges = [] + current = left + for index in range(child_count): + upper = right if index == child_count - 1 else min(right, current + step) + if upper < current: + break + ranges.append((current, upper)) + current = upper + 1 + if current > right: + break + return ranges def _mobilede_root_mileage_ranges_for_overflow(max_children: int) -> list[tuple[int | None, int | None]]: @@ -659,6 +746,15 @@ def _mobilede_root_mileage_ranges_for_overflow(max_children: int) -> list[tuple[ pivot = 150000 if MOBILEDE_COMPACT_SEGMENTS else 100000 low_cap = 75000 if MOBILEDE_COMPACT_SEGMENTS else 50000 return [(None, low_cap), (low_cap + 1, pivot), (pivot + 1, None)] + if MOBILEDE_COMPACT_SEGMENTS: + ranges: list[tuple[int | None, int | None]] = [ + (None, 50000), + (50001, 100000), + (100001, 150000), + (150001, 200000), + (200001, None), + ] + return ranges[:child_budget] return _mobilede_mileage_ranges() @@ -694,7 +790,7 @@ def _mobilede_range_span(value_min: int | None, value_max: int | None) -> int | return max(0, int(value_max) - int(value_min)) -def _mobilede_should_avoid_bmw_mileage_split( +def _mobilede_should_avoid_tiny_mileage_split( *, depth: int, price_min: int | None, @@ -809,11 +905,10 @@ def _mobilede_build_overflow_candidate_groups( year_max = _mobilede_parse_optional_int(segment.get("year_max")) price_min = _mobilede_parse_optional_int(segment.get("price_min")) price_max = _mobilede_parse_optional_int(segment.get("price_max")) - make_id = str(segment.get("make_id") or "").strip() previous_split_kind = str(segment.get("overflow_split") or "").strip() mileage_splits = _mobilede_split_mileage_ranges_for_overflow(mileage_min, mileage_max) - if make_id == "3500" and mileage_splits: - if _mobilede_should_avoid_bmw_mileage_split( + if mileage_splits: + if _mobilede_should_avoid_tiny_mileage_split( depth=depth, price_min=price_min, price_max=price_max, @@ -844,7 +939,7 @@ def _mobilede_build_overflow_candidate_groups( price_splits = _mobilede_split_price_ranges_for_overflow(price_min, price_max) if price_splits: child_segments = [] - for child_price_min, child_price_max in price_splits[:2]: + for child_price_min, child_price_max in price_splits[:MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS]: child_segments.append( _mobilede_make_overflow_child_segment( segment, @@ -869,8 +964,8 @@ def _mobilede_build_overflow_candidate_groups( previous_split_kind == "year" and depth >= MOBILEDE_OVERFLOW_YEAR_DEEP_SPLIT_DEPTH ) - if allow_year_split and make_id == "3500": - # Для BMW не уходим в слишком узкие year-ветки. + if allow_year_split: + # Не уходим в слишком узкие year-ветки, если уже есть более полезная разбивка. if year_span is not None and year_span < max(4, MOBILEDE_OVERFLOW_MIN_YEAR_SPLIT_SPAN): allow_year_split = False elif price_splits or mileage_splits: @@ -879,7 +974,7 @@ def _mobilede_build_overflow_candidate_groups( year_splits = _mobilede_split_year_ranges_for_overflow(year_min, year_max) if allow_year_split else [] if year_splits: child_segments = [] - for child_year_min, child_year_max in year_splits[:2]: + for child_year_min, child_year_max in year_splits[:MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS]: child_segments.append( _mobilede_make_overflow_child_segment( segment, @@ -897,7 +992,7 @@ def _mobilede_build_overflow_candidate_groups( if mileage_splits: child_segments = [] - for child_mileage_min, child_mileage_max in mileage_splits[:2]: + for child_mileage_min, child_mileage_max in mileage_splits[:MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS]: child_segments.append( _mobilede_make_overflow_child_segment( segment, @@ -913,10 +1008,7 @@ def _mobilede_build_overflow_candidate_groups( ) _mobilede_append_overflow_candidate_group(candidate_groups, "mileage", child_segments) - if make_id == "11000": - candidate_groups.sort(key=lambda item: {"price": 0, "recent_price": 0, "year": 1, "root_mileage": 2, "mileage": 3}.get(item[0], 9)) - elif make_id == "3500": - candidate_groups.sort(key=lambda item: {"recent_price": 0, "price": 1, "year": 2, "mileage": 3, "root_mileage": 4}.get(item[0], 9)) + candidate_groups.sort(key=lambda item: {"recent_price": 0, "price": 1, "year": 2, "mileage": 3, "root_mileage": 4}.get(item[0], 9)) return candidate_groups @@ -1003,17 +1095,16 @@ def _mobilede_score_overflow_candidate_group( split_bias += depth * max(1, MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY // 2) if kind == "mileage" and previous_split_kind == "mileage": split_bias += MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY + MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY - if make_id == "3500": - if price_span is not None and price_span <= 10000: - split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY - if price_span is not None and price_span <= 5000: - split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY - if mileage_span is not None and mileage_span <= 100000: - split_bias += max(1, MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY // 2) - if mileage_span is not None and mileage_span <= 50000: - split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY - if mileage_span is not None and mileage_span <= 25000: - split_bias += MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY + MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY + if price_span is not None and price_span <= 10000: + split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY + if price_span is not None and price_span <= 5000: + split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY + if mileage_span is not None and mileage_span <= 100000: + split_bias += max(1, MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY // 2) + if mileage_span is not None and mileage_span <= 50000: + split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY + if mileage_span is not None and mileage_span <= 25000: + split_bias += MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY + MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY distance_to_target += micro_children * max(1, MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY // 2) @@ -1050,10 +1141,12 @@ def _mobilede_overflow_candidate_group_is_useful( return True useful_children = sum(total >= useful_floor for total in child_totals) + non_empty_children = sum(total > 0 for total in child_totals) micro_children = sum(total < lower_target for total in child_totals) tiny_children = sum(total < tiny_threshold for total in child_totals) max_child_total = max(child_totals) - make_id = str(segment.get("make_id") or "").strip() + sum_child_total = sum(child_totals) + parent_total = _mobilede_segment_total_results(segment) depth = max(0, int(_mobilede_parse_optional_int(segment.get("overflow_depth")) or 0)) previous_split_kind = str(segment.get("overflow_split") or "").strip() price_span = _mobilede_range_span( @@ -1065,23 +1158,28 @@ def _mobilede_overflow_candidate_group_is_useful( _mobilede_parse_optional_int(segment.get("mileage_max")), ) + if parent_total is not None and int(parent_total) > target: + if non_empty_children > 0 and sum_child_total >= int(parent_total * 0.5): + return True + if max_child_total >= lower_target: + return True + if tiny_children >= len(child_totals): return False if micro_children > MOBILEDE_OVERFLOW_MAX_MICRO_CHILDREN and useful_children == 0: return False - if make_id == "3500": - if kind in {"mileage", "root_mileage"}: - if previous_split_kind == "mileage" and useful_children == 0: - return False - if depth >= 2 and useful_children == 0: - return False - if price_span is not None and price_span <= 10000 and useful_children == 0: - return False - if mileage_span is not None and mileage_span <= 100000 and micro_children > 0 and useful_children == 0: - return False - if kind == "year" and useful_children == 0 and max_child_total < lower_target: + if kind in {"mileage", "root_mileage"}: + if previous_split_kind == "mileage" and useful_children == 0: return False + if depth >= 2 and useful_children == 0: + return False + if price_span is not None and price_span <= 10000 and useful_children == 0: + return False + if mileage_span is not None and mileage_span <= 100000 and micro_children > 0 and useful_children == 0: + return False + if kind == "year" and useful_children == 0 and max_child_total < lower_target: + return False return useful_children > 0 or max_child_total >= lower_target @@ -1164,8 +1262,8 @@ def _mobilede_finalize_preplanned_segment(segment: dict[str, object], total_resu item["total_results"] = total_results if total_results is not None: label = str(item.get("label") or _mobilede_segment_key(item)) - if "total=" not in label: - item["label"] = f"{label} | total={total_results}" + label = re.sub(r"\s*\|\s*total=\d+", "", label) + item["label"] = f"{label} | total={total_results}" return item @@ -1464,6 +1562,12 @@ def _queue_mobilede_overflow_child_segments( ) -> int: if not parent_segment: return 0 + if bootstrap_run and only_new is not True and _mobilede_skip_late_overflow_children_during_bootstrap(): + logger.info( + "mobile.de late overflow children not queued during bootstrap: parent=%s reason=preplan_first_pass", + _mobilede_short_segment_label(parent_segment), + ) + return 0 parent_fingerprint = _mobilede_segment_fingerprint(parent_segment) children = _mobilede_get_overflow_child_segments( redis_client, @@ -1529,7 +1633,14 @@ def _queue_mobilede_overflow_child_segments( def _mobilede_probe_total(search_url: str, **params: str | int | None) -> int | None: try: client = MobileDeClient.for_worker(delay_seconds=0) - page = client.fetch_search_page(page_number=1, search_url=search_url, **params) + probe_timeout = max(5, int(os.getenv("MOBILEDE_PLAN_PROBE_TIMEOUT_SECONDS", "12"))) + page = client.fetch_search_page( + page_number=1, + search_url=search_url, + timeout=probe_timeout, + max_retries=0, + **params, + ) return int(page.total_results or 0) except Exception as exc: logger.warning("mobile.de segment probe failed: params=%s error=%s", params, exc) @@ -1662,7 +1773,7 @@ def _mobilede_split_search_url_segment_by_make(segment: dict[str, object]) -> li def _expand_mobilede_search_url_segment_by_probe(segment: dict[str, object]) -> list[dict[str, object]] | None: - if not MOBILEDE_PREPLAN_SEGMENT_PROBES: + if not MOBILEDE_PREPLAN_SEGMENT_PROBES or not _mobilede_filtered_url_uses_adaptive_plan(): return None search_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip() @@ -1674,11 +1785,15 @@ def _expand_mobilede_search_url_segment_by_probe(segment: dict[str, object]) -> planned: list[dict[str, object]] = [] probes_used = 0 probe_limit = max(1, int(MOBILEDE_PREPLAN_MAX_PROBES)) + started_at = time.monotonic() + max_seconds = int(MOBILEDE_ADAPTIVE_URL_MAX_SECONDS) def _probe(params: dict[str, str | int | None]) -> int | None: nonlocal probes_used if probes_used >= probe_limit: return None + if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: + return None probes_used += 1 if probes_used == 1 or probes_used % 25 == 0: _mobilede_touch_planning_progress("adaptive_url_planning") @@ -1692,7 +1807,11 @@ def _expand_mobilede_search_url_segment_by_probe(segment: dict[str, object]) -> ) return _mobilede_probe_total(search_url, **params) + time_budget_reached = False for price_min, price_max in _mobilede_price_ranges_for_segment(segment): + if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: + time_budget_reached = True + break price_params: dict[str, str | int | None] = {"p": _mobilede_range_value(price_min, price_max)} price_label = _mobilede_price_label(price_min, price_max) price_total = _probe(price_params) @@ -1714,8 +1833,14 @@ def _expand_mobilede_search_url_segment_by_probe(segment: dict[str, object]) -> continue for year_min, year_max in _mobilede_year_ranges_for_segment_price(segment, price_min, price_max): + if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: + time_budget_reached = True + break year_label = _mobilede_year_label(year_min, year_max) for refined_price_min, refined_price_max in _mobilede_price_subranges_for_hot_year(price_min, price_max, year_min, year_max): + if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: + time_budget_reached = True + break refined_price_label = _mobilede_price_label(refined_price_min, refined_price_max) year_params = { "p": _mobilede_range_value(refined_price_min, refined_price_max), @@ -1741,6 +1866,9 @@ def _expand_mobilede_search_url_segment_by_probe(segment: dict[str, object]) -> continue for mileage_min, mileage_max in _mobilede_mileage_ranges(): + if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: + time_budget_reached = True + break mileage_params = dict(year_params) mileage_params["ml"] = _mobilede_range_value(mileage_min, mileage_max) mileage_label = _mobilede_mileage_label(mileage_min, mileage_max) @@ -1761,6 +1889,20 @@ def _expand_mobilede_search_url_segment_by_probe(segment: dict[str, object]) -> mileage_range=(mileage_min, mileage_max), ) ) + if time_budget_reached: + break + if time_budget_reached: + break + + if time_budget_reached: + logger.warning( + "mobile.de adaptive URL planning time budget reached: base=%s seconds=%s segments=%s probes=%s/%s", + _mobilede_short_segment_label(segment), + max_seconds, + len(planned), + probes_used, + probe_limit, + ) logger.info( "mobile.de adaptive URL segments planned: base=%s segments=%s probes=%s limit=%s", @@ -1874,6 +2016,33 @@ def _mobilede_refine_dense_planned_segments(segments: list[dict[str, object]]) - return refined +def _mobilede_refine_dense_planned_segments_until_stable(segments: list[dict[str, object]]) -> list[dict[str, object]]: + max_passes = max(1, int(os.getenv("MOBILEDE_REFINE_ADAPTIVE_MAX_PASSES", "2"))) + refined = [dict(item) for item in segments] + for pass_index in range(1, max_passes + 1): + before = len(refined) + refined = _mobilede_refine_dense_planned_segments(refined) + dense_count = sum( + 1 + for item in refined + if (_mobilede_segment_total_results(item) or 0) > min( + MOBILEDE_SEGMENT_TARGET_RESULTS, + _mobilede_overflow_threshold(MOBILEDE_MAX_PAGE_NUMBER), + ) + ) + logger.info( + "mobile.de dense refine pass complete: pass=%s/%s before=%s after=%s dense_left=%s", + pass_index, + max_passes, + before, + len(refined), + dense_count, + ) + if dense_count <= 0 or len(refined) >= MOBILEDE_PREPLAN_MAX_SEGMENTS: + break + return refined + + def _mobilede_should_refine_adaptive_segments() -> bool: return os.getenv("MOBILEDE_REFINE_ADAPTIVE_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"} @@ -2079,17 +2248,15 @@ def _expand_mobilede_search_url_segment( def _build_mobilede_runtime_segments(settings: Settings) -> list[dict[str, object]]: env_search_urls = settings.listing.filtered_search_urls - # Ready-made filtered URLs default to the old deterministic price/year split: - # it starts immediately and is easier to keep stable. Probe-based adaptive - # planning can still be enabled explicitly when we need denser pre-plans. + # Готовые фильтрованные URL по умолчанию используют старое + # детерминированное деление по цене/году: оно стартует сразу и его проще + # держать стабильным. Адаптивное планирование на основе probe можно + # включить явно, когда нужен более плотный pre-plan. fast_start_filtered_urls = ( bool(env_search_urls) and os.getenv("MOBILEDE_FILTERED_URL_FAST_START", "false").strip().lower() in {"1", "true", "yes", "on"} ) - adaptive_filtered_urls = ( - bool(env_search_urls) - and os.getenv("MOBILEDE_FILTERED_URL_ADAPTIVE_PLAN", "false").strip().lower() in {"1", "true", "yes", "on"} - ) + adaptive_filtered_urls = bool(env_search_urls) and _mobilede_filtered_url_uses_adaptive_plan() if env_search_urls: segments = _mobilede_source_segments_from_settings(settings) if any(str(item.get("runtime_brand") or "").strip() for item in segments): @@ -2104,7 +2271,7 @@ def _build_mobilede_runtime_segments(settings: Settings) -> list[dict[str, objec segments = _mobilede_source_segments_from_settings(settings) learned_segments = _mobilede_load_learned_runtime_segments(segments) if learned_segments is not None: - return learned_segments + return _mobilede_interleave_segments_by_make(learned_segments) expanded: list[dict[str, object]] = [] expanded_from_adaptive_url = False for segment in segments: @@ -2121,6 +2288,7 @@ def _build_mobilede_runtime_segments(settings: Settings) -> list[dict[str, objec expanded_from_adaptive_url = True if len(expanded) != len(segments): logger.info("mobile.de segments planned: input=%s total=%s", len(segments), len(expanded)) + expanded = _mobilede_interleave_segments_by_make(expanded) if fast_start_filtered_urls: logger.info( "mobile.de fast-start runtime segments ready: input_urls=%s final=%s reason=filtered_search_urls", @@ -2130,7 +2298,7 @@ def _build_mobilede_runtime_segments(settings: Settings) -> list[dict[str, objec return [_mobilede_finalize_preplanned_segment(item, item.get("total_results")) for item in expanded] if expanded_from_adaptive_url: if _mobilede_should_refine_adaptive_segments(): - refined = _mobilede_refine_dense_planned_segments(expanded) + refined = _mobilede_refine_dense_planned_segments_until_stable(expanded) logger.info( "mobile.de adaptive URL plan refined: before=%s after=%s reason=dense_segments", len(expanded), diff --git a/mobilede_scraper/worker/progress.py b/mobilede_scraper/worker/progress.py index 68b5b37..6294929 100644 --- a/mobilede_scraper/worker/progress.py +++ b/mobilede_scraper/worker/progress.py @@ -79,9 +79,7 @@ def _update_task_progress( json.dumps(data, ensure_ascii=False), ex=ttl, ) - # Глобальный маркер активности для внешнего guard-процесса. - # Нужен, чтобы контейнер мог самовосстанавливаться при полном зависании воркера - # (когда PID жив, но прогресс по задачам не двигается). + # Глобальный маркер активности для self-heal. 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)) diff --git a/mobilede_scraper/worker/search_sync.py b/mobilede_scraper/worker/search_sync.py index 880e5a5..44d683b 100644 --- a/mobilede_scraper/worker/search_sync.py +++ b/mobilede_scraper/worker/search_sync.py @@ -1,9 +1,16 @@ from __future__ import annotations -# This module intentionally reuses the task module globals and helpers so -# the giant search task can live outside tasks.py without changing runtime -# behavior. -from .tasks import * # noqa: F401,F403 +# Модуль переиспользует хелперы из `tasks.py`, чтобы не менять runtime-логику. +from . import tasks as _tasks + + +globals().update( + { + name: getattr(_tasks, name) + for name in dir(_tasks) + if not name.startswith("__") + } +) def run_mobilede_sync_search_task( @@ -76,21 +83,28 @@ def run_mobilede_sync_search_task( lock_owner=lock_owner, ) _clear_mobilede_followup_pending(redis_client, segment_key=segment_runtime_key) + if continuous is None: + continuous = MOBILEDE_CONTINUOUS_SYNC_ENABLED if only_new is None and runtime_config.sync.only_new is not None: only_new = runtime_config.sync.only_new - guarded_only_new = _mobilede_force_full_scan_only_new(only_new, redis_client=redis_client) + guarded_only_new = _mobilede_force_full_scan_only_new( + only_new, + redis_client=redis_client, + continuous=continuous, + ) if only_new is True and guarded_only_new is False: logger.info("mobile.de full-pass mode: forcing only_new=False in sync task") only_new = guarded_only_new - # Full-pass tasks must keep contributing to bootstrap progress until all - # segments are completed. Some already queued tasks may carry - # bootstrap_run=False from a post-bootstrap refresh attempt; do not let - # that stale flag turn an incomplete full pass into endless refresh mode. + # Пока full-pass не завершён, сохраняем вклад в bootstrap-прогресс. if only_new is not True: bootstrap_run_active = _mobilede_bootstrap_active(redis_client) else: bootstrap_run_active = _mobilede_bootstrap_active(redis_client) if bootstrap_run is None else bool(bootstrap_run and _mobilede_bootstrap_active(redis_client)) - post_bootstrap_refresh = _mobilede_post_bootstrap_full_refresh(redis_client, only_new) + post_bootstrap_refresh = _mobilede_post_bootstrap_full_refresh( + redis_client, + only_new, + continuous=continuous, + ) runtime_segments_enabled = False if segment is None and not make_id and not model_id: reserved_segment = _reserve_mobilede_runtime_segment(redis_client, settings, only_new=only_new) @@ -289,8 +303,6 @@ def run_mobilede_sync_search_task( segment_index=segment_index, segment_label=(segment or {}).get("label") if segment else None, ) - if continuous is None: - continuous = MOBILEDE_CONTINUOUS_SYNC_ENABLED segment_label = _mobilede_segment_label(segment) make_name = _mobilede_segment_make(segment, make_id) model_name = _mobilede_segment_model(segment, model_id) @@ -608,7 +620,7 @@ def run_mobilede_sync_search_task( if allow_followup: progress_done_now, progress_total_now, progress_left_now, progress_dispatched_now = _mobilede_bootstrap_progress_snapshot(redis_client) incremental_mode = bool(only_new and segment and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP) - full_pass_continuous_mode = bool(continuous and runtime_config.sync.only_new is False and not post_bootstrap_refresh) + full_pass_continuous_mode = bool(continuous and only_new is not True and not post_bootstrap_refresh) full_pass_cycle_complete = bool( full_pass_continuous_mode and progress_total_now > 0 @@ -627,6 +639,7 @@ def run_mobilede_sync_search_task( should_start_incremental = bool( runtime_config.sync.only_new is True and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP + and not full_pass_continuous_mode ) if should_start_incremental: if _try_queue_mobilede_incremental_transition( @@ -783,6 +796,17 @@ def run_mobilede_sync_search_task( ) refresh_followup_mode = bool(post_bootstrap_refresh and refresh_cycle_id) if late_overflow_pending: + bootstrap_recovery_mode = bool(bootstrap_run_active and not bootstrap_done_now) + if bootstrap_recovery_mode: + _queue_mobilede_bootstrap_recovery( + redis_client, + lane=lane, + delay_seconds=delay_seconds, + use_cursor=use_cursor, + continuous=True, + segment_label=segment_label, + reason="late_overflow", + ) logger.info( "mobile.de late overflow keeps current full pass open: runtime=%s added=%s queued=%s", segment_label, @@ -935,6 +959,24 @@ def run_mobilede_sync_search_task( followup_segment_max_pages, followup_phase, ) + if ( + bootstrap_run_active + and not late_overflow_pending + and not runtime_rotation + and progress_total_now > 0 + and progress_done_now < progress_total_now + and int(redis_client.llen(MOBILEDE_SYNC_QUEUE) or 0) <= 0 + ): + _queue_mobilede_bootstrap_recovery( + redis_client, + lane=lane, + delay_seconds=delay_seconds, + use_cursor=use_cursor, + continuous=True, + segment_label=segment_label, + reason="window_exhausted_without_rotation", + countdown=max(5, int(MOBILEDE_BOOTSTRAP_CONTINUATION_DELAY_SECONDS)), + ) return _mobilede_task_result_summary( result=result, segment=segment, @@ -1013,6 +1055,8 @@ def run_mobilede_sync_search_task( ) if segment is not None: _release_mobilede_bootstrap_dispatched_marker(redis_client, segment) + status_code = getattr(getattr(exc, "response", None), "status_code", None) + is_antibot_block = int(status_code or 0) in {401, 403, 429} is_transient_request_error = _is_mobilede_transient_request_error(exc) max_retries = int(getattr(self, "max_retries", 0) or 0) current_retries = int(getattr(self.request, "retries", 0) or 0) @@ -1057,7 +1101,11 @@ def run_mobilede_sync_search_task( if segment is not None: followup_kwargs["segment"] = segment followup_kwargs["segment_index"] = segment_index - delayed_retry = max(300, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS * 4) + delayed_retry = ( + MOBILEDE_ANTIBOT_BACKOFF_SECONDS + if is_antibot_block + else max(300, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS * 4) + ) if _try_set_mobilede_followup_pending( redis_client, segment_key=segment_runtime_key, @@ -1069,12 +1117,13 @@ def run_mobilede_sync_search_task( countdown=delayed_retry, ) logger.warning( - "mobile.de delayed retry queued after network issue: runtime=%s filter=%s pages=%s-%s delay=%ss", + "mobile.de delayed retry queued after network issue: runtime=%s filter=%s pages=%s-%s delay=%ss status=%s", _mobilede_segment_label(segment), _mobilede_filter_source(segment, search_url), actual_start_page, actual_end_page, delayed_retry, + status_code, ) else: logger.info("mobile.de delayed retry already pending for segment=%s", segment_runtime_key) diff --git a/mobilede_scraper/worker/self_heal.py b/mobilede_scraper/worker/self_heal.py index 9eec882..139b1a8 100644 --- a/mobilede_scraper/worker/self_heal.py +++ b/mobilede_scraper/worker/self_heal.py @@ -63,8 +63,7 @@ def _read_last_progress_ts(redis_client: Redis) -> int | None: if ts > 0: return ts - # Fallback: если глобальный ключ не найден, берём max(ts) из task_progress:*. - # Это дороже, но выполняется только при отсутствии основного маркера. + # Резервно берём max(ts) из task_progress:*. max_ts = 0 for key in redis_client.scan_iter(match="mobilede:state:task_progress:*"): try: @@ -153,7 +152,7 @@ def _kill_worker_process() -> None: time.sleep(20) try: - # Если процесс ещё жив — принудительно убиваем. + # Если процесс жив, добиваем SIGKILL. os.kill(pid, 0) logger.error("Self-heal: worker pid=%s did not stop after SIGTERM; sending SIGKILL", pid) os.kill(pid, SIGKILL_FALLBACK) @@ -230,7 +229,7 @@ def main() -> None: db_idle_restart = True restart_reason = f"db_idle_age={db_age}s > {db_idle_seconds}s" - # Глобальный anti-storm lock: чтобы много воркеров не рестартились одновременно. + # Не даём нескольким воркерам рестартовать одновременно. acquired = bool( redis_client.set( SELF_HEAL_RESTART_LOCK_KEY, @@ -251,12 +250,10 @@ def main() -> None: if db_idle_restart: logger.error("Self-heal: no DB writes for too long; clearing checkpoint to restart from segment 1") _reset_bootstrap_checkpoint_for_db_idle(redis_client) - # Небольшой джиттер, чтобы при одинаковом событии у разных контейнеров - # перезапуск был не строго одновременно. + # Добавляем небольшой джиттер перед рестартом. time.sleep(random.uniform(0.3, 2.0)) _kill_worker_process() - # После kill pid1 контейнер будет перезапущен Docker restart-policy. - # На случай неуспеха не молотим цикл. + # Даём Docker время на рестарт. time.sleep(check_interval) except Exception: diff --git a/mobilede_scraper/worker/tasks.py b/mobilede_scraper/worker/tasks.py index 7f90224..d49f2b7 100644 --- a/mobilede_scraper/worker/tasks.py +++ b/mobilede_scraper/worker/tasks.py @@ -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,