from __future__ import annotations import logging import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from collections.abc import Callable, Iterable from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit import requests from .flight import extract_detail_listing, extract_search_results from .models import MobileDeListing, MobileDeSearchPage logger = logging.getLogger("mobile_de.client") BASE_URL = "https://www.mobile.de" SEARCH_PATH = "/ru/транспортные-средства/поиск.html" DETAIL_PATH = "/ru/транспортные-средства/подробности.html" DEFAULT_HEADERS = { "user-agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "accept-language": "ru,en;q=0.9,de;q=0.8", } class MobileDeClient: """HTTP client for mobile.de search/detail pages.""" def __init__(self, session: requests.Session | None = None, *, delay_seconds: float = 0.7) -> None: self.session = session or requests.Session() self.session.headers.update(DEFAULT_HEADERS) self.delay_seconds = max(0.0, delay_seconds) @classmethod def for_worker(cls, *, delay_seconds: float = 0.0) -> "MobileDeClient": session = requests.Session() adapter = requests.adapters.HTTPAdapter(pool_connections=100, pool_maxsize=100, max_retries=0) session.mount("https://", adapter) session.mount("http://", adapter) return cls(session=session, delay_seconds=delay_seconds) @staticmethod def build_make_model_param(make_id: str | int, model_id: str | int | None = None) -> str: make = str(make_id).strip() model = str(model_id).strip() if model_id is not None else "" return f"{make};{model};;" @staticmethod def build_search_url(page_number: int = 1, **params: str | int | None) -> str: query = { "sb": "rel", "od": "up", "vc": "Car", "s": "Car", "pageNumber": page_number, } query.update({key: value for key, value in params.items() if value is not None}) return f"{BASE_URL}{SEARCH_PATH}?{urlencode(query)}" @staticmethod def build_search_url_from_existing( search_url: str, *, page_number: int | None = None, **params: str | int | None, ) -> str: parts = urlsplit(search_url) query_items = [ (key, value) for key, value in parse_qsl(parts.query, keep_blank_values=True) if key != "pageNumber" and key not in params ] if page_number is not None: query_items.append(("pageNumber", str(page_number))) query_items.extend((key, str(value)) for key, value in params.items() if value is not None) scheme = parts.scheme or "https" netloc = parts.netloc or urlsplit(BASE_URL).netloc path = parts.path or SEARCH_PATH return urlunsplit((scheme, netloc, path, urlencode(query_items), "")) @staticmethod def build_detail_url(listing_id: str | int) -> str: query = urlencode({"id": listing_id, "vc": "Car", "s": "Car"}) return f"{BASE_URL}{DETAIL_PATH}?{query}" def fetch_html(self, url: str, *, timeout: int = 30) -> str: response = self.session.get(url, timeout=timeout) response.raise_for_status() return response.text def fetch_search_page( self, page_number: int = 1, *, search_url: str | None = None, **params: str | int | None, ) -> MobileDeSearchPage: url = ( self.build_search_url_from_existing(search_url, page_number=page_number, **params) if search_url else self.build_search_url(page_number=page_number, **params) ) html = self.fetch_html(url) raw = extract_search_results(html) listings = [self._map_listing(item) for item in raw.get("listings", []) if isinstance(item, dict)] return MobileDeSearchPage( url=url, page_number=page_number, total_results=raw.get("numResultsTotal"), listings=listings, raw_search_results=raw, ) def iter_search_pages( self, *, start_page: int = 1, max_pages: int | None = None, search_url: str | None = None, stop_after_empty: bool = True, progress_callback: Callable[[MobileDeSearchPage, dict[str, int | None]], None] | None = None, **params: str | int | None, ) -> Iterable[MobileDeSearchPage]: page_number = start_page pages_seen = 0 logger.debug( "mobile.de search window started: start_page=%s max_pages=%s params=%s", start_page, max_pages, {key: value for key, value in params.items() if value is not None}, ) while max_pages is None or pages_seen < max_pages: logger.debug("mobile.de fetching search page=%s", page_number) page = self.fetch_search_page(page_number=page_number, search_url=search_url, **params) page_meta = { "page_number": page.page_number, "pages_seen": pages_seen + 1, "max_pages": max_pages, "listing_count": len(page.listings), "total_results": page.total_results, } logger.debug( "mobile.de fetched search page=%s listings=%s total_results=%s", page.page_number, len(page.listings), page.total_results, ) if progress_callback is not None: progress_callback(page, page_meta) if stop_after_empty and not page.listings: logger.debug("mobile.de stopping search window on empty page=%s", page.page_number) break yield page pages_seen += 1 page_number += 1 if self.delay_seconds: time.sleep(self.delay_seconds) logger.debug( "mobile.de search window finished: pages_seen=%s next_page=%s", pages_seen, page_number, ) def fetch_search_pages_concurrent( self, *, start_page: int = 1, max_pages: int = 1, workers: int = 8, search_url: str | None = None, stop_after_empty: bool = True, progress_callback: Callable[[MobileDeSearchPage, dict[str, int | None]], None] | None = None, **params: str | int | None, ) -> list[MobileDeSearchPage]: max_pages = max(1, int(max_pages)) workers = max(1, min(int(workers), max_pages)) page_numbers = list(range(max(1, int(start_page)), max(1, int(start_page)) + max_pages)) pages_by_number: dict[int, MobileDeSearchPage] = {} thread_local = threading.local() def _fetch_page(page_number: int) -> MobileDeSearchPage: client = getattr(thread_local, "client", None) if client is None: client = MobileDeClient.for_worker(delay_seconds=0) thread_local.client = client return client.fetch_search_page(page_number=page_number, search_url=search_url, **params) with ThreadPoolExecutor(max_workers=workers) as executor: futures = { executor.submit(_fetch_page, page_number): page_number for page_number in page_numbers } for future in as_completed(futures): page_number = futures[future] page = future.result() pages_by_number[page_number] = page if progress_callback is not None: progress_callback( page, { "page_number": page.page_number, "pages_seen": len(pages_by_number), "max_pages": max_pages, "listing_count": len(page.listings), "total_results": page.total_results, }, ) ordered_pages = [pages_by_number[page_number] for page_number in page_numbers if page_number in pages_by_number] if stop_after_empty: non_empty_pages: list[MobileDeSearchPage] = [] for page in ordered_pages: if not page.listings: break non_empty_pages.append(page) return non_empty_pages return ordered_pages def fetch_detail(self, listing_id: str | int) -> dict: html = self.fetch_html(self.build_detail_url(listing_id)) return extract_detail_listing(html) @staticmethod def _map_listing(item: dict) -> MobileDeListing: listing_id = str(item.get("id") or item.get("adId") or "") attr = item.get("attr") if isinstance(item.get("attr"), dict) else {} contact = item.get("contact") if isinstance(item.get("contact"), dict) else {} location = ", ".join( part for part in [attr.get("z"), attr.get("loc")] if isinstance(part, str) and part ) or None return MobileDeListing( id=listing_id, url=MobileDeClient.build_detail_url(listing_id) if listing_id else "", title=item.get("shortTitle"), subtitle=item.get("subTitle"), price=item.get("p"), seller_name=contact.get("name"), seller_type=contact.get("type") or item.get("st"), location=location, first_registration=attr.get("fr"), mileage=attr.get("ml"), power=attr.get("pw"), fuel=attr.get("ft"), transmission=attr.get("tr"), raw=item, )