"""Thin async client for the cleaned Planner mirror (`planner.maxify.it`). This is the "Hub DB API" entry point of the pipeline. It is a fork of `meta-ads-mcp/agent/context/scc.py` (kept standalone so this service has no dependency on that repo). Read-only by design: never sends credentials, authenticates only via the `Authorization: Bearer $SCC_API_KEY` header. The upstream rule from `metabase/am_metabase_map/README.md` applies: "Runtime consumers must read cleaned Planner APIs, not scrape Hub or Meta directly." We honor that — we never touch `hub.maxify.it`. """ from __future__ import annotations import logging from datetime import date, datetime from typing import Any import httpx from app.config import settings logger = logging.getLogger(__name__) SCC_TIMEOUT = httpx.Timeout(20.0, connect=8.0) class SccApiError(RuntimeError): def __init__(self, path: str, status: int, body: str) -> None: self.path = path self.status = status self.body = body[:500] super().__init__(f"SCC {path} -> HTTP {status}") class SccClient: """Read-only access to `planner.maxify.it` clean Hub mirror.""" def __init__(self, base_url: str | None = None, api_key: str | None = None) -> None: self.base_url = (base_url or settings.SCC_BASE_URL).rstrip("/") api_key = api_key if api_key is not None else settings.SCC_API_KEY headers = {} if api_key: headers["Authorization"] = f"Bearer {api_key}" self._client = httpx.AsyncClient( base_url=self.base_url, timeout=SCC_TIMEOUT, follow_redirects=True, headers=headers, ) async def close(self) -> None: await self._client.aclose() async def get(self, path: str, params: dict[str, Any] | None = None) -> Any: try: response = await self._client.get(path, params=params) except httpx.HTTPError as exc: logger.warning("SCC GET %s failed: %s", path, exc) raise SccApiError(path, 0, str(exc)) from exc if response.status_code >= 400: raise SccApiError(path, response.status_code, response.text) try: return response.json() except ValueError as exc: raise SccApiError(path, response.status_code, "invalid json") from exc async def get_shows(self) -> list[dict[str, Any]]: """Live show directory (source of truth for the show list).""" data = await self.get("/api/show-directory") return data.get("shows", data) if isinstance(data, dict) else data async def get_tour_shows(self) -> list[dict[str, Any]]: data = await self.get("/api/tour-shows") return data.get("shows", data) if isinstance(data, dict) else data async def get_ticket_sales(self) -> dict[str, Any]: return await self.get("/api/ticket-sales") async def ping(self) -> bool: """Live reachability probe. Never raises.""" try: await self.get("/api/health") return True except SccApiError as exc: if exc.status == 404: return True logger.warning("SCC ping failed: %s", exc) return False except Exception as exc: logger.warning("SCC ping failed: %s", exc) return False # ── Pipeline-specific convenience queries ────────────────────────── async def find_shows_on_date(self, target_date: date) -> list[dict[str, Any]]: """Return every tour show whose `show_date` matches `target_date`.""" tour = await self.get_tour_shows() matches = [] for entry in tour or []: entry_date = _parse_show_date(entry.get("show_date")) if entry_date == target_date: matches.append(entry) logger.info("found %d shows for %s", len(matches), target_date.isoformat()) return matches async def find_previous_show_in_city(self, city: str, before_date: date) -> dict[str, Any] | None: """Most recent past tour show in the same city (ignore venue).""" tour = await self.get_tour_shows() candidates = [] for entry in tour or []: if _normalize_city(entry.get("city")) != _normalize_city(city): continue entry_date = _parse_show_date(entry.get("show_date")) if entry_date is None or entry_date >= before_date: continue candidates.append((entry_date, entry)) if not candidates: return None candidates.sort(key=lambda x: x[0], reverse=True) return candidates[0][1] async def historical_sales( self, show_key: str, *, show_date: date | None = None, city: str | None = None, ) -> dict[str, Any]: """Best-effort lookup of historical ticket sales. Planner currently returns `shows` as a mapping of `show_key` to sales records, while some older responses returned a list. Normalize both shapes and use the catalog's date/city mapping when the tour-show ID is not the same as the ticket-sales key. """ sales = await self.get_ticket_sales() records: dict[str, dict[str, Any]] = {} catalog: list[dict[str, Any]] = [] if isinstance(sales, dict): raw_shows = sales.get("shows", sales.get("data", [])) raw_catalog = sales.get("catalog", []) if isinstance(raw_shows, dict): records = { str(key): value for key, value in raw_shows.items() if isinstance(value, dict) } elif isinstance(raw_shows, list): records = { str(record.get("show_key") or record.get("show_id")): record for record in raw_shows if isinstance(record, dict) and (record.get("show_key") or record.get("show_id")) } if isinstance(raw_catalog, list): catalog = [row for row in raw_catalog if isinstance(row, dict)] elif isinstance(sales, list): records = { str(record.get("show_key") or record.get("show_id")): record for record in sales if isinstance(record, dict) and (record.get("show_key") or record.get("show_id")) } candidate_keys = [show_key] if show_date and city: target_city = _normalize_city(city) for entry in catalog: entry_date = _parse_show_date(entry.get("show_date")) entry_city = _normalize_city( entry.get("clean_city") or entry.get("city") or entry.get("display_name") ) catalog_key = entry.get("show_key") if entry_date == show_date and entry_city == target_city and catalog_key: candidate_keys.append(str(catalog_key)) for key in candidate_keys: record = records.get(key) if record: return {"show_key": key, **record} return {} def _normalize_city(value: str | None) -> str: return (value or "").strip().lower().replace(" ", "-") def _parse_show_date(value: Any) -> date | None: if value is None: return None if isinstance(value, date): return value if isinstance(value, datetime): return value.date() if isinstance(value, str): try: return date.fromisoformat(value[:10]) except ValueError: return None return None