dubizzle/dubizzle_scraper/discovery/algolia.py

from __future__ import annotations

import json
import logging
import time
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse
from urllib.request import Request, urlopen


logger = logging.getLogger("dubizzle_scraper.discovery.algolia")

ALGOLIA_HARD_PAGE_LIMIT = 100
ALGOLIA_MAX_HITS_PER_QUERY = ALGOLIA_HARD_PAGE_LIMIT * 100
ALGOLIA_ID_RANGE_START = 0
ALGOLIA_ID_RANGE_END = 20_000_000
ALGOLIA_ID_RANGE_MIN_WINDOW = 100


class AlgoliaDiscoveryError(RuntimeError):
    pass


@dataclass(slots=True)
class AlgoliaDiscoveryPageStat:
    page_number: int
    hits_count: int


@dataclass(slots=True)
class AlgoliaDiscoveryResult:
    vehicle_urls: list[str] = field(default_factory=list)
    hit_records: dict[str, dict[str, Any]] = field(default_factory=dict)
    origin_ids_by_url: dict[str, str] = field(default_factory=dict)
    pages: list[dict[str, int]] = field(default_factory=list)
    early_stopped: bool = False
    truncated_by_time_budget: bool = False
    total_hits: int = 0


@dataclass(slots=True)
class _AlgoliaShard:
    category_slug: str
    id_min: int | None = None
    id_max: int | None = None
    year_min: int | None = None
    year_max: int | None = None

    def filter_expr(self) -> str:
        parts = [f"(category_v2.slug_paths:{self.category_slug})"]
        if self.year_min is not None:
            parts.append(f"year>={int(self.year_min)}")
        if self.year_max is not None:
            parts.append(f"year<={int(self.year_max)}")
        if self.id_min is not None:
            parts.append(f"id>={int(self.id_min)}")
        if self.id_max is not None:
            parts.append(f"id<={int(self.id_max)}")
        return " AND ".join(parts)

    def label(self) -> str:
        label = self.category_slug
        if self.year_min is not None or self.year_max is not None:
            label += f"[year:{self.year_min}-{self.year_max}]"
        if self.id_min is None and self.id_max is None:
            return label
        return f"{label}[id:{self.id_min}-{self.id_max}]"


@dataclass(slots=True)
class _AlgoliaHttpClient:
    endpoint: str
    app_id: str
    api_key: str
    user_agent: str
    index_name: str
    hits_per_page: int

    def request(self, *, page: int, filters: str) -> dict[str, Any]:
        params = urlencode(
            {
                "query": "",
                "page": page,
                "hitsPerPage": self.hits_per_page,
                "filters": filters,
            }
        )
        payload = {"requests": [{"indexName": self.index_name, "params": params}]}
        request = Request(
            self.endpoint,
            data=json.dumps(payload).encode("utf-8"),
            headers={
                "content-type": "application/json",
                "x-algolia-application-id": self.app_id,
                "x-algolia-api-key": self.api_key,
                "accept": "application/json",
                "user-agent": self.user_agent,
            },
            method="POST",
        )

        last_exc: Exception | None = None
        for attempt in range(1, 4):
            try:
                with urlopen(request, timeout=15) as response:
                    body = response.read().decode("utf-8", errors="ignore")
                parsed = json.loads(body)
                results = parsed.get("results") or []
                return results[0] if results and isinstance(results[0], dict) else {}
            except Exception as exc:
                last_exc = exc
                if attempt >= 3:
                    break
                time.sleep(0.8 * attempt)

        raise last_exc or RuntimeError("Algolia request failed")


def _build_origin_id_from_hit(hit: dict[str, Any]) -> str | None:
    for key in ("id", "objectID", "uuid"):
        value = hit.get(key)
        if value is None:
            continue
        text = str(value).strip()
        if text:
            return f"dubizzle:{text}"
    permalink = str(hit.get("permalink") or "").strip()
    if permalink:
        tail = permalink.rstrip("/").split("/")[-1]
        if tail:
            return f"dubizzle:{tail}"
    return None


def _build_vehicle_url_from_hit(hit: dict[str, Any]) -> str | None:
    permalink = str(hit.get("permalink") or "").strip()
    if permalink:
        if permalink.startswith("http://") or permalink.startswith("https://"):
            return permalink
        if permalink.startswith("/"):
            return f"https://www.dubizzle.com{permalink}"
        return f"https://www.dubizzle.com/{permalink.lstrip('/')}"

    short = str(hit.get("short_url") or "").strip()
    if short:
        if short.startswith("http://") or short.startswith("https://"):
            return short
        return f"https://dubizzle.com/s/{short.strip('/')}"

    hit_id = hit.get("id") or hit.get("objectID")
    if hit_id is not None:
        return f"https://www.dubizzle.com/motors/used-cars/ad-{hit_id}/"
    return None


def _extract_make_from_listing_url(listing_url: str | None) -> str | None:
    if not listing_url:
        return None
    try:
        parsed = urlparse(listing_url)
        params = parse_qs(parsed.query)
        candidate = (params.get("Make") or params.get("make") or [None])[0]
        if candidate:
            return str(candidate).strip()
        parts = [p for p in parsed.path.split("/") if p]
        if len(parts) >= 3 and parts[0].lower() == "motors" and parts[1].lower() == "used-cars":
            slug = parts[2].strip().lower()
            if slug and slug != "s":
                return slug.replace("-", " ")
    except Exception:
        return None
    return None


def _make_slug(make: str | None) -> str | None:
    if not make:
        return None
    slug = make.strip().lower().replace(" ", "-")
    return slug or None


def _hit_matches_filters(
    hit: dict[str, Any],
    *,
    make: str | None,
    model: str | None,
    year_min: int | None,
    year_max: int | None,
) -> bool:
    if make:
        hit_make = str(hit.get("make") or "").strip().lower()
        if hit_make != make.strip().lower():
            return False
    if model:
        hit_model = str(hit.get("model") or "").strip().lower()
        if hit_model != model.strip().lower():
            return False

    hit_year_raw = hit.get("year")
    hit_year: int | None = None
    if hit_year_raw is not None:
        try:
            hit_year = int(hit_year_raw)
        except Exception:
            hit_year = None

    if year_min is not None and (hit_year is None or hit_year < year_min):
        return False
    if year_max is not None and (hit_year is None or hit_year > year_max):
        return False

    return True


def _probe_total_hits(client: _AlgoliaHttpClient, shard: _AlgoliaShard) -> int:
    first = client.request(page=0, filters=shard.filter_expr())
    return int(first.get("nbHits") or 0)


def _split_id_range(shard: _AlgoliaShard) -> tuple[_AlgoliaShard, _AlgoliaShard] | None:
    lo = shard.id_min if shard.id_min is not None else ALGOLIA_ID_RANGE_START
    hi = shard.id_max if shard.id_max is not None else ALGOLIA_ID_RANGE_END
    if hi - lo <= ALGOLIA_ID_RANGE_MIN_WINDOW:
        return None
    mid = (lo + hi) // 2
    return (
        _AlgoliaShard(
            category_slug=shard.category_slug,
            id_min=lo,
            id_max=mid,
            year_min=shard.year_min,
            year_max=shard.year_max,
        ),
        _AlgoliaShard(
            category_slug=shard.category_slug,
            id_min=mid + 1,
            id_max=hi,
            year_min=shard.year_min,
            year_max=shard.year_max,
        ),
    )


def _expand_shards(client: _AlgoliaHttpClient, initial_shard: _AlgoliaShard, max_duration_seconds: float | None, started_at: float) -> tuple[list[_AlgoliaShard], bool, int]:
    queue: list[_AlgoliaShard] = [initial_shard]
    ready: list[_AlgoliaShard] = []
    truncated = False
    total_hits = 0

    while queue:
        if max_duration_seconds is not None and (time.perf_counter() - started_at) > max_duration_seconds:
            truncated = True
            break
        shard = queue.pop(0)
        shard_total = _probe_total_hits(client, shard)
        if shard is initial_shard:
            total_hits = shard_total
        if shard_total == 0:
            continue
        if shard_total <= ALGOLIA_MAX_HITS_PER_QUERY:
            ready.append(shard)
            continue
        split = _split_id_range(shard)
        if split is None:
            logger.warning(
                "Shard %s still exceeds limit=%d but id-window is too small to split further (hits=%d)",
                shard.label(),
                ALGOLIA_MAX_HITS_PER_QUERY,
                shard_total,
            )
            ready.append(shard)
            continue
        logger.warning(
            "Splitting Algolia shard %s with hits=%d into %s and %s",
            shard.label(),
            shard_total,
            split[0].label(),
            split[1].label(),
        )
        queue.insert(0, split[1])
        queue.insert(0, split[0])

    return ready, truncated, total_hits


def _fetch_shard_hits(
    *,
    client: _AlgoliaHttpClient,
    shard: _AlgoliaShard,
    make: str | None,
    model: str | None,
    year_min: int | None,
    year_max: int | None,
    known_origin_ids: set[str] | None,
    limit: int | None,
    threshold: float,
    max_duration_seconds: float | None,
    started_at: float,
) -> tuple[list[str], dict[str, dict[str, Any]], dict[str, str], list[dict[str, int]], bool, bool]:
    urls: list[str] = []
    hit_records: dict[str, dict[str, Any]] = {}
    origin_ids_by_url: dict[str, str] = {}
    pages: list[dict[str, int]] = []
    early_stopped = False
    truncated_by_time_budget = False
    page = 0
    nb_pages = None

    while True:
        if max_duration_seconds is not None and (time.perf_counter() - started_at) > max_duration_seconds:
            truncated_by_time_budget = True
            break

        try:
            first = client.request(page=page, filters=shard.filter_expr())
        except Exception as exc:
            raise AlgoliaDiscoveryError(
                f"Algolia request failed for shard={shard.label()} page={page}: {exc}"
            ) from exc

        hits = first.get("hits") or []
        if not isinstance(hits, list):
            hits = []

        if page == 0:
            nb_pages = min(int(first.get("nbPages") or 0), ALGOLIA_HARD_PAGE_LIMIT)

        pages.append({"page_number": page + 1, "links_found": len(hits), "shard": shard.label()})
        if not hits:
            break

        page_known = 0
        page_new = 0
        for raw_hit in hits:
            if not isinstance(raw_hit, dict):
                continue
            if not _hit_matches_filters(
                raw_hit,
                make=make,
                model=model,
                year_min=year_min,
                year_max=year_max,
            ):
                continue

            url = _build_vehicle_url_from_hit(raw_hit)
            if not url:
                continue
            origin_id = _build_origin_id_from_hit(raw_hit)
            if origin_id and known_origin_ids is not None and origin_id in known_origin_ids:
                page_known += 1
            else:
                page_new += 1

            if url in hit_records:
                continue
            urls.append(url)
            hit_records[url] = raw_hit
            if origin_id:
                origin_ids_by_url[url] = origin_id
            if limit is not None and limit > 0 and len(urls) >= limit:
                early_stopped = True
                break

        if limit is not None and limit > 0 and len(urls) >= limit:
            break

        if known_origin_ids is not None and threshold > 0 and (page_known + page_new) > 0:
            ratio = page_known / (page_known + page_new)
            if ratio >= threshold and page_new == 0:
                early_stopped = True
                break

        page += 1
        if nb_pages is not None and page >= nb_pages:
            break

    return urls, hit_records, origin_ids_by_url, pages, early_stopped, truncated_by_time_budget


def discover_vehicle_hits_from_algolia(
    *,
    settings,
    make: str | None = None,
    model: str | None = None,
    limit: int | None = None,
    year_min: int | None = None,
    year_max: int | None = None,
    listing_url: str | None = None,
    known_origin_ids: set[str] | None = None,
    max_duration_seconds: float | None = None,
) -> AlgoliaDiscoveryResult:
    app_id = settings.algolia.application_id.strip()
    api_key = settings.algolia.api_key.strip()
    index_name = settings.algolia.index_name.strip()
    if not app_id or not api_key or not index_name:
        raise AlgoliaDiscoveryError("Algolia credentials/index are not configured")

    effective_make = make or _extract_make_from_listing_url(listing_url)
    hits_per_page = max(1, min(100, int(settings.algolia.hits_per_page)))
    base_url = settings.algolia.base_url.strip().rstrip("/")
    if not base_url:
        base_url = f"https://{app_id}-dsn.algolia.net"

    category_slug = settings.algolia.category_slug.strip() or "motors/used-cars"
    make_slug = _make_slug(effective_make)
    if make_slug:
        category_slug = f"{category_slug}/{make_slug}"

    client = _AlgoliaHttpClient(
        endpoint=f"{base_url}/1/indexes/*/queries",
        app_id=app_id,
        api_key=api_key,
        user_agent=settings.fingerprint.user_agent,
        index_name=index_name,
        hits_per_page=hits_per_page,
    )
    threshold = float(settings.listing.early_stop_threshold)
    started_at = time.perf_counter()

    initial_shard = _AlgoliaShard(
        category_slug=category_slug,
        id_min=ALGOLIA_ID_RANGE_START,
        id_max=ALGOLIA_ID_RANGE_END,
        year_min=year_min,
        year_max=year_max,
    )
    shards, shard_build_truncated, total_hits = _expand_shards(
        client,
        initial_shard,
        max_duration_seconds,
        started_at,
    )

    collected_urls: list[str] = []
    hits_by_url: dict[str, dict[str, Any]] = {}
    origin_ids_by_url: dict[str, str] = {}
    pages: list[dict[str, int]] = []
    early_stopped = False
    truncated_by_time_budget = shard_build_truncated

    logger.warning(
        "Algolia shard plan ready: total_hits=%d shards=%d category=%s",
        total_hits,
        len(shards),
        category_slug,
    )

    for shard in shards:
        shard_urls, shard_hits, shard_origin_ids, shard_pages, shard_early_stop, shard_truncated = _fetch_shard_hits(
            client=client,
            shard=shard,
            make=effective_make,
            model=model,
            year_min=year_min,
            year_max=year_max,
            known_origin_ids=known_origin_ids,
            limit=(limit - len(collected_urls)) if limit is not None and limit > 0 else None,
            threshold=threshold,
            max_duration_seconds=max_duration_seconds,
            started_at=started_at,
        )
        pages.extend(shard_pages)
        early_stopped = early_stopped or shard_early_stop
        truncated_by_time_budget = truncated_by_time_budget or shard_truncated
        for url in shard_urls:
            if url in hits_by_url:
                continue
            collected_urls.append(url)
            hits_by_url[url] = shard_hits[url]
            if url in shard_origin_ids:
                origin_ids_by_url[url] = shard_origin_ids[url]
        if limit is not None and limit > 0 and len(collected_urls) >= limit:
            break
        if truncated_by_time_budget:
            break

    logger.info(
        "Algolia discovery done: urls=%d total_hits=%d pages=%d shards=%d",
        len(collected_urls),
        total_hits,
        len(pages),
        len(shards),
    )

    if limit is not None and limit > 0 and len(collected_urls) > limit:
        collected_urls = collected_urls[:limit]

    return AlgoliaDiscoveryResult(
        vehicle_urls=collected_urls,
        hit_records={url: hits_by_url[url] for url in collected_urls if url in hits_by_url},
        origin_ids_by_url={url: origin_ids_by_url[url] for url in collected_urls if url in origin_ids_by_url},
        pages=pages,
        early_stopped=early_stopped,
        truncated_by_time_budget=truncated_by_time_budget,
        total_hits=total_hits,
    )