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)


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."""
    async with httpx.AsyncClient(timeout=60.0) as client:
        r = await client.post(f"{_BASE}/{actor_id}/runs", params={"token": token}, json=run_input)
        r.raise_for_status()
        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}", params={"token": token})
            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={"token": token, "format": "json"})
        d.raise_for_status()
        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"

    async with httpx.AsyncClient(timeout=timeout) as client:
        for attempt in range(1, _MAX_ATTEMPTS + 1):
            try:
                resp = await client.post(url, params={"token": token}, json=run_input)
                if resp.status_code >= 500:
                    raise httpx.HTTPStatusError(
                        f"{resp.status_code} from Apify", request=resp.request, response=resp)
                resp.raise_for_status()
                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
    return []  # pragma: no cover