RECORDGOD/app/dealgod.py

"""DealGod client — RecordGod's single seam to the universal brain (BaseGod). Implements the frozen
DealGod API contract: identify · value · lore · supply · metal-prices (defined + owned in the DealGod
hub repo, mirrored here). Reference consumer: the other StoreGod skins copy this call pattern.

Key from the vault (`dealgod_api_key`), sent as X-Api-Key. Base URL is env-configurable so we can mock
against a local stub / `:8001` before DG1 allowlists the routes on api.dealgod.pro:
    DEALGOD_API_BASE=http://localhost:8001   # dev mock
Endpoints: identify · value · lore · supply. The guard ships default-open, so calls work pre-entitlement.
"""
import asyncio
import os
import httpx

BASE = os.getenv("DEALGOD_API_BASE", "https://api.dealgod.pro").rstrip("/")


async def _key(db):
    # per-skin SCOPED customer key (set on the container, e.g. ToolGod's tools-only key) wins;
    # else the shared vault key (master/owner). Lets each skin run as its own scoped customer.
    k = os.getenv("DEALGOD_API_KEY")
    if k:
        return k
    from . import vault   # lazy so this module imports cleanly for the standalone selfcheck
    return await vault.get_secret(db, "dealgod_api_key")


def _norm_candidate(c: dict) -> dict:
    """Frozen contract: candidates carry ref_type:'canon'|'release' + ref_id — records resolve to a Discogs
    release_id, NOT a canon_id (SCANGOD's structural point). Echo convenience keys so callers switch on either,
    and back-fill ref_type/ref_id from a legacy/direct shape."""
    rt = c.get("ref_type")
    if rt is None:
        if c.get("release_id") is not None:
            c["ref_type"], c["ref_id"], rt = "release", c["release_id"], "release"
        elif c.get("canon_id") is not None:
            c["ref_type"], c["ref_id"], rt = "canon", c["canon_id"], "canon"
    if rt == "release":
        c.setdefault("release_id", c.get("ref_id"))
    elif rt == "canon":
        c.setdefault("canon_id", c.get("ref_id"))
    return c


async def identify(db, clues: dict, *, images=None, labels=None, hints=None, scope=None, features=None) -> dict:
    """POST /api/identify. clues = {barcode?,isbn?,catno?,text?}; images = [b64,…≤4]; labels = [ocr…];
    hints = {brand,size,weight,notes}; scope = DOMAINS (e.g. ['music']). Returns {candidates:[{ref_type,ref_id,slug,
    category,display_name,confidence,id_keys,traps,…}]} — normalised so each candidate has canon_id OR release_id set."""
    key = await _key(db)
    if not key:
        return {"candidates": [], "error": "no_key"}
    payload = dict(clues or {})
    if images:
        payload["images"] = images
    if labels:
        payload["labels"] = labels
    if hints:
        payload["hints"] = hints
    body = {"clues": payload}
    if scope:
        body["scope"] = list(scope)
    try:
        async with httpx.AsyncClient(timeout=30, headers={"X-Api-Key": key}) as c:
            r = await c.post(f"{BASE}/api/identify", json=body)
    except Exception as e:
        return {"candidates": [], "error": str(e)}
    if r.status_code != 200:
        return {"candidates": [], "status": r.status_code}
    d = r.json()
    d["candidates"] = [_norm_candidate(x) for x in d.get("candidates", [])]
    return d


async def value(db, *, ref_type=None, ref_id=None, canon_id=None, release_id=None, condition=None, region="AU") -> dict | None:
    """POST /api/value — frozen primary shape is {ref_type,ref_id} ('canon'|'release'); canon_id/release_id accepted
    as convenience aliases. Returns {low,typ,high,currency,as_of,n?,melt_floor?,comps?,supply?,source[]} or None."""
    key = await _key(db)
    if not key:
        return None
    if ref_type is None:                       # derive the universal ref from whichever alias the caller passed
        if release_id is not None:
            ref_type, ref_id = "release", release_id
        elif canon_id is not None:
            ref_type, ref_id = "canon", canon_id
    if ref_type is None or ref_id is None:
        return None
    body = {"ref_type": ref_type, "ref_id": ref_id, "region": region}
    if condition:
        body["condition"] = condition
    try:
        async with httpx.AsyncClient(timeout=25, headers={"X-Api-Key": key}) as c:
            r = await c.post(f"{BASE}/api/value", json=body)
        return r.json() if r.status_code == 200 else None
    except Exception:
        return None


async def lore(db, ref) -> dict | None:
    """GET /api/lore/{ref} — {ref} = canon_id (int) OR slug. The harvested resale wiki
    (description, tier, variants, faults, id_keys, traps, related) or None."""
    key = await _key(db)
    if not key or ref is None:
        return None
    try:
        async with httpx.AsyncClient(timeout=20, headers={"X-Api-Key": key}) as c:
            r = await c.get(f"{BASE}/api/lore/{ref}")
        return r.json() if r.status_code == 200 else None
    except Exception:
        return None


async def metal_prices(db) -> dict | None:
    """GET /api/metal-prices — live AU spot (gold/silver/platinum oz) + per-gram. Powers the jewellery melt tool."""
    key = await _key(db)
    if not key:
        return None
    try:
        async with httpx.AsyncClient(timeout=15, headers={"X-Api-Key": key}) as c:
            r = await c.get(f"{BASE}/api/metal-prices")
        return r.json() if r.status_code == 200 else None
    except Exception:
        return None


async def supply(db, ids: list) -> dict:
    """POST /api/supply (LIVE). Batched ≤200; batches run CONCURRENTLY (bounded) so a big buylist
    scan isn't a serial network crawl. Returns {<id str>: {store_count, discogs_seller_count, …}};
    degrades to {} / drops a failed batch. ponytail: Semaphore(5) caps load — the brain has its own
    rate limits + matcher caps; raise it only if buylists get huge and the brain can take it."""
    key = await _key(db)
    if not key or not ids:
        return {}
    batches = [ids[i:i + 200] for i in range(0, len(ids), 200)]
    out, sem = {}, asyncio.Semaphore(5)
    async with httpx.AsyncClient(timeout=25, headers={"X-Api-Key": key}) as c:
        async def _one(batch):
            async with sem:
                try:
                    r = await c.post(f"{BASE}/api/supply", json={"ids": batch})
                    if r.status_code == 200:
                        out.update(r.json().get("results", {}))   # single-threaded asyncio → no race
                except Exception:
                    pass
        await asyncio.gather(*(_one(b) for b in batches))
    return out


def _selfcheck():
    rel = _norm_candidate({"ref_type": "release", "ref_id": 12345, "display_name": "X"})
    assert rel["release_id"] == 12345 and rel["ref_type"] == "release"
    can = _norm_candidate({"ref_type": "canon", "ref_id": 678, "slug": "roland-jp-8000"})
    assert can["canon_id"] == 678
    legacy = _norm_candidate({"release_id": 999})        # legacy/direct shape → derives ref_type/ref_id
    assert legacy["ref_type"] == "release" and legacy["ref_id"] == 999
    legacy_c = _norm_candidate({"canon_id": 42})
    assert legacy_c["ref_type"] == "canon" and legacy_c["ref_id"] == 42
    print("ok")


if __name__ == "__main__":
    _selfcheck()