211 lines
7.7 KiB
Python
211 lines
7.7 KiB
Python
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("dubizzle_scraper.discovery.sitemap")
|
||
|
||
DEFAULT_SITEMAP_INDEX_URL = "https://www.dubizzle.com/Xj9rDOVMEi0hc38S/sitemap_index.xml"
|
||
_SITEMAP_TIMEOUT_SECONDS = 30
|
||
_LOC_TAG_RE = re.compile(rb"<loc>\s*(.*?)\s*</loc>", 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)
|