89ca9ad647
On-demand Scout service (replaces the nightly aggregator): - FastAPI agent-mesh service; card name "matchmaking-service" (HTTP /a2a/tasks + Redis-stream worker matching the sibling-service pattern) - run_search: parallel multi-board sweep (Naukri/blackfalcondata + Foundit + LinkedIn), city-filtered at the board, cheapest-per-result first - per-board adapters + normalizers -> ScoutJob with rich read-only details and offsite apply links; recall mode + MVQ guard - result cache for dev replay ($0); per-board cost knobs; retry-on-5xx - research/: engine design, board cost economics, India-actor shortlist, POC
94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
"""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
|
|
|
|
|
|
async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0) -> list[dict]:
|
|
"""POST input → return the dataset items (list). One retry on transient 5xx; raises otherwise.
|
|
|
|
Result cache (APIFY_CACHE_MODE): dev replays a saved job set for $0; production stays live.
|
|
"""
|
|
mode = 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")
|
|
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
|