"""Sarvam API credit health checks for Master Panel (read-only, cached).""" from __future__ import annotations import glob import hashlib import logging import os import re import time from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional, Tuple import requests logger = logging.getLogger(__name__) _DEFAULT_CHAT_URL = "https://api.sarvam.ai/v1/chat/completions" _DEFAULT_MODEL = "sarvam-30b" _DEFAULT_TTL = 600 _HEALTH_PRIORITY = { "exhausted": 0, "invalid_key": 1, "unknown": 2, "not_configured": 3, "available": 4, } _LOG_PATTERNS = ( re.compile(r"\b402\b", re.IGNORECASE), re.compile(r"no\s+credits", re.IGNORECASE), re.compile(r"insufficient\s+credits", re.IGNORECASE), re.compile(r"credit[s]?\s+exhausted", re.IGNORECASE), ) _TS_RE = re.compile( r"^(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:\.\d+)?)" ) _probe_cache: Dict[str, Dict[str, Any]] = {} _summary_cache: Dict[str, Any] = {"expires_at": 0.0, "fingerprint": "", "payload": None} def _utc_now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def mask_key(key: str) -> str: key = str(key or "") if len(key) <= 8: return "****" if key else "" return f"{key[:4]}...{key[-4:]}" def _key_fingerprint(key: str) -> str: return hashlib.sha256(str(key or "").encode("utf-8")).hexdigest()[:16] def _probe_ttl_seconds() -> int: try: return max(60, int(os.getenv("SARVAM_CREDITS_PROBE_TTL_SECONDS", str(_DEFAULT_TTL)))) except (TypeError, ValueError): return _DEFAULT_TTL def _chat_url() -> str: return os.getenv("RAG_SARVAM_CHAT_URL", "").strip() or _DEFAULT_CHAT_URL def _chat_model() -> str: return os.getenv("RAG_SARVAM_MODEL", "").strip() or _DEFAULT_MODEL def _probe_timeout() -> int: try: return max(5, min(60, int(os.getenv("SARVAM_CREDITS_PROBE_TIMEOUT_SECONDS", "15")))) except (TypeError, ValueError): return 15 def _parse_log_timestamp(line: str) -> Optional[str]: match = _TS_RE.match(line.strip()) if not match: return None raw = match.group(1).replace(" ", "T") if not raw.endswith("Z") and "+" not in raw: raw = f"{raw}Z" return raw def _log_paths() -> List[str]: paths: List[str] = [] paths.extend(sorted(glob.glob("/tmp/stt_worker_*.log"))) for fixed in ("/tmp/pca_stt-worker.log", "/tmp/pca_stt_worker.log"): if os.path.isfile(fixed): paths.append(fixed) return paths def scan_stt_logs_for_credit_exhaustion(max_lines_per_file: int = 4000) -> Dict[str, Any]: """Read recent STT log lines for credit-exhaustion signals (read-only).""" last_exhausted_at: Optional[str] = None last_message: Optional[str] = None recent_count = 0 cutoff = datetime.now(timezone.utc) - timedelta(hours=24) for path in _log_paths(): try: with open(path, "r", encoding="utf-8", errors="ignore") as fh: lines = fh.readlines() except OSError: continue tail = lines[-max_lines_per_file:] if len(lines) > max_lines_per_file else lines for line in tail: if not any(p.search(line) for p in _LOG_PATTERNS): continue ts = _parse_log_timestamp(line) recent_count += 1 if ts: try: parsed = datetime.fromisoformat(ts.replace("Z", "+00:00")) if parsed >= cutoff: if last_exhausted_at is None or ts > last_exhausted_at: last_exhausted_at = ts last_message = line.strip()[:240] except ValueError: if last_exhausted_at is None: last_exhausted_at = ts last_message = line.strip()[:240] elif last_exhausted_at is None: last_message = line.strip()[:240] return { "last_exhausted_at": last_exhausted_at, "last_exhaustion_message": last_message, "recent_exhaustion_signals": recent_count, "log_paths_scanned": len(_log_paths()), } def probe_sarvam_credits(api_key: str) -> Dict[str, Any]: """Minimal Sarvam chat probe — detects exhausted vs valid key (not exact balance).""" key = str(api_key or "").strip() if not key: return { "credits_probe": "skipped", "credits_health": "not_configured", "credits_health_label": "Not configured", "probe_message": "No Sarvam subscription key configured", "http_status": None, "last_checked_at": _utc_now_iso(), } url = _chat_url() model = _chat_model() headers = { "api-subscription-key": key, "Content-Type": "application/json", } payload = { "model": model, "messages": [{"role": "user", "content": "ping"}], "max_tokens": 1, } try: response = requests.post(url, headers=headers, json=payload, timeout=_probe_timeout()) status = response.status_code body_snippet = (response.text or "")[:300] if status == 200: return { "credits_probe": "ok", "credits_health": "available", "credits_health_label": "Credits available", "probe_message": "Sarvam API accepted a minimal request", "http_status": status, "last_checked_at": _utc_now_iso(), } if status == 402: return { "credits_probe": "exhausted", "credits_health": "exhausted", "credits_health_label": "Credits exhausted", "probe_message": body_snippet or "Sarvam returned HTTP 402 (no credits)", "http_status": status, "last_checked_at": _utc_now_iso(), } if status == 403: return { "credits_probe": "invalid_key", "credits_health": "invalid_key", "credits_health_label": "Invalid API key", "probe_message": body_snippet or "Sarvam returned HTTP 403 (invalid or missing key)", "http_status": status, "last_checked_at": _utc_now_iso(), } if status == 429: return { "credits_probe": "rate_limited", "credits_health": "available", "credits_health_label": "Credits likely available (rate limited)", "probe_message": "Sarvam returned HTTP 429 — key accepted but rate limited", "http_status": status, "last_checked_at": _utc_now_iso(), } return { "credits_probe": "error", "credits_health": "unknown", "credits_health_label": "Could not verify credits", "probe_message": f"Sarvam returned HTTP {status}: {body_snippet}", "http_status": status, "last_checked_at": _utc_now_iso(), } except requests.Timeout: return { "credits_probe": "error", "credits_health": "unknown", "credits_health_label": "Could not verify credits", "probe_message": "Sarvam probe timed out", "http_status": None, "last_checked_at": _utc_now_iso(), } except Exception as exc: logger.warning("Sarvam credits probe failed: %s", exc) return { "credits_probe": "error", "credits_health": "unknown", "credits_health_label": "Could not verify credits", "probe_message": str(exc), "http_status": None, "last_checked_at": _utc_now_iso(), } def _probe_for_key(key: str, *, force_probe: bool) -> Tuple[Dict[str, Any], bool]: """Return probe payload and whether cache was used.""" ttl = _probe_ttl_seconds() now = time.time() fp = _key_fingerprint(key) cached = _probe_cache.get(fp) if not force_probe and cached and now < float(cached.get("expires_at") or 0): result = dict(cached.get("result") or {}) result["probe_skipped"] = True return result, True result = probe_sarvam_credits(key) result["probe_skipped"] = False _probe_cache[fp] = {"expires_at": now + ttl, "result": dict(result)} return result, False def _status_from_health(key_set: bool, health: str) -> Tuple[str, str]: if not key_set: return "not_configured", "Not configured" mapping = { "available": ("active", "Active"), "exhausted": ("exhausted", "Credits exhausted"), "invalid_key": ("invalid_key", "Invalid API key"), "unknown": ("active", "Active"), "not_configured": ("not_configured", "Not configured"), } return mapping.get(health, ("active", "Active")) def _worst_health(healths: List[str]) -> str: configured = [h for h in healths if h and h != "not_configured"] if not configured: return "not_configured" return min(configured, key=lambda h: _HEALTH_PRIORITY.get(h, 99)) def _slots_fingerprint(slots: List[Dict[str, Any]]) -> str: parts = [] for slot in slots: parts.append(f"{slot.get('source_id')}:{_key_fingerprint(str(slot.get('key') or ''))}") return "|".join(parts) def get_sarvam_keys_health( slots: List[Dict[str, str]], *, force_probe: bool = False, ) -> Dict[str, Any]: """ Probe each unique Sarvam key used across backend, STT, and RAG. slots items: source_id, source_label, env_hint, key """ global _summary_cache ttl = _probe_ttl_seconds() now = time.time() fp = _slots_fingerprint(slots) if not force_probe and _summary_cache.get("payload") and _summary_cache.get("fingerprint") == fp: if now < float(_summary_cache.get("expires_at") or 0): cached = dict(_summary_cache["payload"]) cached["probe_skipped"] = True cached["cache_ttl_seconds"] = ttl return cached logs = scan_stt_logs_for_credit_exhaustion() probe_by_fp: Dict[str, Dict[str, Any]] = {} first_source_by_fp: Dict[str, str] = {} any_probe_ran = False public_slots: List[Dict[str, Any]] = [] for slot in slots: source_id = str(slot.get("source_id") or "") source_label = str(slot.get("source_label") or source_id) env_hint = str(slot.get("env_hint") or "") key = str(slot.get("key") or "").strip() key_fp = _key_fingerprint(key) if key and key_fp not in probe_by_fp: probe, from_cache = _probe_for_key(key, force_probe=force_probe) probe_by_fp[key_fp] = probe first_source_by_fp[key_fp] = source_id any_probe_ran = any_probe_ran or not from_cache elif not key: probe_by_fp.setdefault( key_fp, { "credits_probe": "skipped", "credits_health": "not_configured", "credits_health_label": "Not configured", "probe_message": "Key not set for this slot", "http_status": None, "last_checked_at": _utc_now_iso(), "probe_skipped": True, }, ) first_source_by_fp.setdefault(key_fp, source_id) probe = probe_by_fp.get(key_fp, {}) same_as = None if key and first_source_by_fp.get(key_fp) != source_id: same_as = first_source_by_fp[key_fp] slot_status, slot_status_label = _status_from_health(bool(key), probe.get("credits_health", "not_configured")) public_slots.append( { "source_id": source_id, "source_label": source_label, "env_hint": env_hint, "key_set": bool(key), "key_masked": mask_key(key), "same_key_as": same_as, "status": slot_status, "status_label": slot_status_label, "credits_health": probe.get("credits_health"), "credits_health_label": probe.get("credits_health_label"), "credits_probe": probe.get("credits_probe"), "probe_message": probe.get("probe_message"), "last_checked_at": probe.get("last_checked_at"), "probe_skipped": probe.get("probe_skipped", False), } ) unique_keys = {str(s.get("key") or "").strip() for s in slots if str(s.get("key") or "").strip()} keys_in_sync = len(unique_keys) <= 1 keys_mismatch_warning = None if len(unique_keys) > 1: keys_mismatch_warning = ( "Different Sarvam keys are configured for Dashboard, STT, and/or MCube AI — " "each row below is checked separately." ) slot_healths = [s.get("credits_health") or "not_configured" for s in public_slots] aggregate_health = _worst_health(slot_healths) status, status_label = _status_from_health(any(s.get("key_set") for s in public_slots), aggregate_health) aggregate_probe = next( (probe_by_fp[_key_fingerprint(str(s.get("key") or ""))] for s in slots if str(s.get("key") or "").strip()), probe_by_fp.get(_key_fingerprint(""), {}), ) payload: Dict[str, Any] = { "key_set": any(str(s.get("key") or "").strip() for s in slots), "keys_in_sync": keys_in_sync, "keys_mismatch_warning": keys_mismatch_warning, "unique_key_count": len(unique_keys), "key_slots": public_slots, "credits_balance": None, "credits_balance_note": "Exact balance is only available on the Sarvam Dashboard → Usage", "status": status, "status_label": status_label, "credits_health": aggregate_health, "credits_health_label": aggregate_probe.get("credits_health_label"), "credits_probe": aggregate_probe.get("credits_probe"), "probe_message": aggregate_probe.get("probe_message"), "last_checked_at": aggregate_probe.get("last_checked_at") or _utc_now_iso(), "probe_skipped": not any_probe_ran and not force_probe, "dashboard_url": "https://dashboard.sarvam.ai/usage", "cache_ttl_seconds": ttl, **logs, } if aggregate_health == "available" and logs.get("last_exhausted_at"): stt_health = next((s for s in public_slots if s.get("source_id") == "stt"), None) if stt_health and stt_health.get("credits_health") == "exhausted": payload["credits_health_warning"] = ( "STT worker key reports exhausted credits (matches recent STT log errors)." ) elif keys_in_sync: payload["credits_health_warning"] = ( "STT logs reported credit errors earlier — likely resolved if you topped up or updated the key." ) else: payload["credits_health_warning"] = ( "STT logs recently reported credit errors — Dashboard, STT, and RAG use different keys; " "check each row below." ) _summary_cache = {"expires_at": now + ttl, "fingerprint": fp, "payload": dict(payload)} return dict(payload) def invalidate_credit_cache() -> None: """Clear cached probe results (e.g. after Sarvam key change).""" global _summary_cache, _probe_cache _summary_cache = {"expires_at": 0.0, "fingerprint": "", "payload": None} _probe_cache = {} def get_sarvam_credit_status(api_key: str, *, force_probe: bool = False) -> Dict[str, Any]: """Backward-compatible single-key wrapper.""" key = str(api_key or "").strip() return get_sarvam_keys_health( [ { "source_id": "backend", "source_label": "Dashboard / backend", "env_hint": "dashboard-backend/.env → SARVAM_SUBSCRIPTION_KEY", "key": key, } ], force_probe=force_probe, )