from __future__ import annotations import gzip import io import logging import re from dataclasses import dataclass, field from typing import Iterable from urllib.parse import urlsplit, urlunsplit from urllib.request import Request, urlopen, ProxyHandler, build_opener import xml.etree.ElementTree as ET logger = logging.getLogger("MOBILEDE_scraper.discovery.sitemap") DEFAULT_SITEMAP_INDEX_URL = "https://www.MOBILEDE.com/Xj9rDOVMEi0hc38S/sitemap_index.xml" _SITEMAP_TIMEOUT_SECONDS = 30 _LOC_TAG_RE = re.compile(rb"\s*(.*?)\s*", re.IGNORECASE | re.DOTALL) class SitemapDiscoveryError(RuntimeError): pass def _is_vehicle_sitemap_url(url: str) -> bool: lowered = url.strip().lower() # Берём только sitemap с авто. if "sitemapbranches" in lowered or "sitemapauctions" in lowered: return False return lowered.endswith(".xml") or lowered.endswith(".xml.gz") def _normalize_vehicle_url(url: str) -> str: parts = urlsplit(url.strip()) return urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")) def _download_bytes(url: str, proxy_url: str | None = None) -> bytes: request = Request( url, headers={ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36" ), "Accept": "application/xml,text/xml,application/xhtml+xml,text/html;q=0.9,*/*;q=0.8", "Accept-Encoding": "gzip", }, ) if proxy_url: handler = ProxyHandler({"http": proxy_url, "https": proxy_url}) opener = build_opener(handler) response = opener.open(request, timeout=_SITEMAP_TIMEOUT_SECONDS) else: response = urlopen(request, timeout=_SITEMAP_TIMEOUT_SECONDS) with response: payload = response.read() encoding = str(response.headers.get("Content-Encoding") or "").lower() if encoding == "gzip" or url.lower().endswith(".gz"): return gzip.GzipFile(fileobj=io.BytesIO(payload)).read() return payload def _local_name(tag: str) -> str: if "}" in tag: return tag.rsplit("}", 1)[1] return tag def _iter_loc_values_fallback(xml_bytes: bytes) -> Iterable[str]: for match in _LOC_TAG_RE.finditer(xml_bytes): try: value = match.group(1).decode("utf-8", errors="ignore").strip() except Exception: continue if value: yield value def _iter_loc_values(xml_bytes: bytes) -> Iterable[str]: try: root = ET.fromstring(xml_bytes) except ET.ParseError as exc: fallback_values = list(_iter_loc_values_fallback(xml_bytes)) if fallback_values: logger.warning( "Falling back to regex sitemap loc extraction after XML parse error: %s", exc, ) yield from fallback_values return raise SitemapDiscoveryError(f"Invalid sitemap XML: {exc}") from exc for element in root.iter(): if _local_name(element.tag) != "loc": continue if not element.text: continue value = element.text.strip() if value: yield value def _filter_vehicle_urls(urls: Iterable[str]) -> list[str]: result: list[str] = [] seen: set[str] = set() for url in urls: normalized = _normalize_vehicle_url(url) if "/VehicleDetail/" not in normalized and "/vehicledetail/" not in normalized: continue if normalized in seen: continue seen.add(normalized) result.append(normalized) return result def _looks_like_vehicle_detail_sitemap(urls: list[str]) -> bool: return any("/vehicledetail/" in url.lower() for url in urls) def discover_vehicle_urls_from_sitemap(index_url: str = DEFAULT_SITEMAP_INDEX_URL, proxy_url: str | None = None) -> list[str]: logger.info("Downloading sitemap index: %s", index_url) index_xml = _download_bytes(index_url, proxy_url=proxy_url) sitemap_urls = [url for url in _iter_loc_values(index_xml) if _is_vehicle_sitemap_url(url)] if not sitemap_urls: raise SitemapDiscoveryError("Sitemap index returned no sitemap URLs") all_vehicle_urls: list[str] = [] for sitemap_url in sitemap_urls: logger.info("Downloading sitemap: %s", sitemap_url) try: sitemap_xml = _download_bytes(sitemap_url, proxy_url=proxy_url) raw_urls = list(_iter_loc_values(sitemap_xml)) except SitemapDiscoveryError as exc: logger.warning("Skipping malformed sitemap %s: %s", sitemap_url, exc) continue vehicle_urls = _filter_vehicle_urls(raw_urls) if raw_urls and not vehicle_urls and not _looks_like_vehicle_detail_sitemap(raw_urls): logger.info("Skipping non-vehicle sitemap %s", sitemap_url) continue logger.info("Sitemap %s yielded %d vehicle URLs", sitemap_url, len(vehicle_urls)) all_vehicle_urls.extend(vehicle_urls) deduped = _filter_vehicle_urls(all_vehicle_urls) if not deduped: raise SitemapDiscoveryError("No vehicle detail URLs discovered from vehicle sitemaps") logger.info("Sitemap discovery done: %d vehicle URLs", len(deduped)) return deduped @dataclass class SitemapDiscoveryStats: transport: str = "http" fetched_sitemaps: int = 0 blocked_sitemaps: int = 0 malformed_sitemaps: int = 0 direct_probe_hits: int = 0 direct_probe_misses: int = 0 @dataclass class SitemapDiscoveryResult: vehicle_urls: list[str] = field(default_factory=list) stats: SitemapDiscoveryStats = field(default_factory=SitemapDiscoveryStats) def discover_vehicle_urls_from_sitemap_with_stats( settings=None, index_url: str = DEFAULT_SITEMAP_INDEX_URL, ) -> SitemapDiscoveryResult: """Возвращает URL и статистику.""" proxy_url = None if settings is not None and hasattr(settings, 'proxy') and settings.proxy.server: proxy_url = settings.proxy.server stats = SitemapDiscoveryStats(transport="http") logger.info("Downloading sitemap index: %s", index_url) index_xml = _download_bytes(index_url, proxy_url=proxy_url) sitemap_urls = [url for url in _iter_loc_values(index_xml) if _is_vehicle_sitemap_url(url)] if not sitemap_urls: raise SitemapDiscoveryError("Sitemap index returned no sitemap URLs") all_vehicle_urls: list[str] = [] for sitemap_url in sitemap_urls: logger.info("Downloading sitemap: %s", sitemap_url) try: sitemap_xml = _download_bytes(sitemap_url, proxy_url=proxy_url) raw_urls = list(_iter_loc_values(sitemap_xml)) stats.fetched_sitemaps += 1 except SitemapDiscoveryError as exc: logger.warning("Skipping malformed sitemap %s: %s", sitemap_url, exc) stats.malformed_sitemaps += 1 continue except Exception as exc: logger.warning("Blocked/failed sitemap %s: %s", sitemap_url, exc) stats.blocked_sitemaps += 1 continue vehicle_urls = _filter_vehicle_urls(raw_urls) if raw_urls and not vehicle_urls and not _looks_like_vehicle_detail_sitemap(raw_urls): logger.info("Skipping non-vehicle sitemap %s", sitemap_url) continue logger.info("Sitemap %s yielded %d vehicle URLs", sitemap_url, len(vehicle_urls)) all_vehicle_urls.extend(vehicle_urls) deduped = _filter_vehicle_urls(all_vehicle_urls) if not deduped: raise SitemapDiscoveryError("No vehicle detail URLs discovered from vehicle sitemaps") logger.info("Sitemap discovery done: %d vehicle URLs", len(deduped)) return SitemapDiscoveryResult(vehicle_urls=deduped, stats=stats)