b1e9cdd182
- A1: fire the hard filter (Apify is precise); embedding vibe-ranker (embed.py) blends with the floor-free white-box (rank.py) to sift ~90 -> top 18 - A2: Opus curator (curate.py) reads the 18 -> honest <=count + Gemini-voiced report cards; graceful fallback to the white-box + templated cards - LLM via opencode.ai/zen gateway (llm.py): Opus chat + direct-OpenAI embeddings - Run LLM work off the event loop (asyncio.to_thread) so the Redis response publishes - C1: dedicated Postgres (app/db/) persists the per-user feed; get_scout_feed replays it so matches survive navigation/refresh - match contract + report-card fields (schema.py); skills.py; tests/
96 lines
3.8 KiB
Python
96 lines
3.8 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,
|
|
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")
|
|
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
|