matchmaking-v2/app/engine/apify_client.py

"""Thin Apify client — run an actor synchronously and get the dataset items.

One call reused by every board adapter. On-demand: we fetch live and score only what
we fetched (no corpus). `run-sync-get-dataset-items` blocks up to ~300s.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import logging
from pathlib import Path

import httpx

from app.config import get_settings

logger = logging.getLogger(__name__)

_BASE = "https://api.apify.com/v2/acts"


def _cache_path(actor_id: str, run_input: dict) -> Path:
    """Stable path per (actor, exact input) — same query replays the same saved set."""
    canon = json.dumps(run_input, sort_keys=True, ensure_ascii=False)
    digest = hashlib.sha1(f"{actor_id}|{canon}".encode()).hexdigest()[:16]
    safe = actor_id.replace("/", "~").replace("~", "_")
    return Path(get_settings().APIFY_CACHE_DIR) / f"{safe}__{digest}.json"


def _cache_read(path: Path) -> list[dict] | None:
    try:
        return json.loads(path.read_text())
    except (OSError, ValueError):
        return None


def _cache_write(path: Path, items: list[dict]) -> None:
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(items, ensure_ascii=False))
    except OSError as e:  # noqa: BLE001 — caching is best-effort
        logger.warning("apify cache write failed: %s", e)

# Apify's run-sync endpoint occasionally throws a transient 5xx (esp. 502). Retry ONCE
# on that. We do NOT retry on empty results: a completed run is billed, so re-running it
# would double the cost for no gain (an empty result is usually a real "0 matches").
_MAX_ATTEMPTS = 2
_BACKOFF = 2.0
_RUNS_BASE = "https://api.apify.com/v2"
_POLL = 3.0  # async-run poll interval (s)


def _auth_headers(token: str) -> dict:
    """Bearer auth — keeps the token OUT of request URLs (and thus out of error/log strings).

    Passing the token as a query param puts it in resp.request.url, which httpx embeds in
    HTTPStatusError.__str__ — so any logged error would leak the credential. The Apify v2 API
    accepts a Bearer header for every endpoint, so this is the single-point fix.
    """
    return {"Authorization": f"Bearer {token}"}


class _ApifyHTTPError(RuntimeError):
    """Token-free HTTP error raised in place of httpx.HTTPStatusError (whose str includes the URL)."""


async def _run_actor_async(actor_id: str, run_input: dict, *, timeout: float, token: str) -> list[dict]:
    """Start an ASYNC run, poll to completion, return its dataset items. Needed for incremental /
    stateful actors (state persisted across runs via `stateKey`) whose crawl exceeds the run-sync
    window — that window aborts the run mid-crawl. No cache: the actor itself is the dedup."""
    headers = _auth_headers(token)
    async with httpx.AsyncClient(timeout=60.0, headers=headers) as client:
        r = await client.post(f"{_BASE}/{actor_id}/runs", json=run_input)
        try:
            r.raise_for_status()
        except httpx.HTTPStatusError:
            raise _ApifyHTTPError(f"apify async run-start {actor_id}: HTTP {r.status_code}") from None
        run = r.json().get("data", {})
        run_id = run.get("id")
        ds_id = run.get("defaultDatasetId")
        if not run_id:
            return []
        elapsed = 0.0
        while elapsed < timeout:
            await asyncio.sleep(_POLL)
            elapsed += _POLL
            s = await client.get(f"{_RUNS_BASE}/actor-runs/{run_id}")
            data = s.json().get("data", {})
            status = data.get("status")
            ds_id = data.get("defaultDatasetId") or ds_id
            if status == "SUCCEEDED":
                break
            if status in ("FAILED", "ABORTED", "TIMED-OUT", "TIMING-OUT"):
                logger.warning("apify async run %s ended %s", actor_id, status)
                return []
        else:
            logger.warning("apify async run %s did not finish within %ss", actor_id, timeout)
            return []
        d = await client.get(f"{_RUNS_BASE}/datasets/{ds_id}/items", params={"format": "json"})
        try:
            d.raise_for_status()
        except httpx.HTTPStatusError:
            raise _ApifyHTTPError(f"apify async dataset {actor_id}: HTTP {d.status_code}") from None
        items = d.json()
        n = len(items) if isinstance(items, list) else 0
        logger.info("apify async run %s SUCCEEDED — %d items (incremental)", actor_id, n)
        return items if isinstance(items, list) else []


async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0,
                    cache_mode: str | None = None) -> list[dict]:
    """POST input → return the dataset items (list). One retry on transient 5xx; raises otherwise.

    Result cache: dev replays a saved job set for $0; production stays live. `cache_mode`
    overrides the configured APIFY_CACHE_MODE per call (e.g. "refresh" for a forced fresh search).
    """
    mode = cache_mode or get_settings().APIFY_CACHE_MODE
    path = _cache_path(actor_id, run_input) if mode != "off" else None

    if path and mode in ("readwrite", "read"):
        cached = _cache_read(path)
        if cached is not None:
            logger.info("apify cache HIT %s (%d items) — $0", actor_id, len(cached))
            return cached
        if mode == "read":
            raise RuntimeError(f"APIFY_CACHE_MODE=read but no cached set for {actor_id} ({path.name})")

    token = get_settings().APIFY_TOKEN
    if not token:
        raise RuntimeError("APIFY_TOKEN not configured")

    # Incremental/stateful actors need an async run (the sync window aborts the longer crawl).
    if run_input.get("incremental"):
        return await _run_actor_async(actor_id, run_input, timeout=timeout, token=token)

    url = f"{_BASE}/{actor_id}/run-sync-get-dataset-items"
    headers = _auth_headers(token)

    async with httpx.AsyncClient(timeout=timeout, headers=headers) as client:
        for attempt in range(1, _MAX_ATTEMPTS + 1):
            try:
                resp = await client.post(url, json=run_input)
                if resp.status_code >= 500:
                    raise _ApifyHTTPError(f"{resp.status_code} from Apify")
                try:
                    resp.raise_for_status()
                except httpx.HTTPStatusError:
                    raise _ApifyHTTPError(f"apify {actor_id}: HTTP {resp.status_code}") from None
                items = resp.json()
                items = items if isinstance(items, list) else []
                if path and mode in ("readwrite", "refresh"):
                    _cache_write(path, items)
                    logger.info("apify cache SAVE %s (%d items)", actor_id, len(items))
                return items
            except (httpx.HTTPStatusError, httpx.TransportError) as e:
                if attempt < _MAX_ATTEMPTS:
                    logger.warning("apify %s: %s — retrying once", actor_id, e)
                    await asyncio.sleep(_BACKOFF)
                    continue
                raise
            except _ApifyHTTPError as e:
                if attempt < _MAX_ATTEMPTS and "from Apify" in str(e):
                    logger.warning("apify %s: %s — retrying once", actor_id, e)
                    await asyncio.sleep(_BACKOFF)
                    continue
                raise
    return []  # pragma: no cover