"""Shared infrastructure for the EvalBot platform: settings, database, auth, file storage.""" import json import logging import os import re import threading import time from typing import Any, Dict, List, Optional import httpx from fastapi import Depends, Header, HTTPException logger = logging.getLogger("evalbot.platform") # -------------------------------------------------------------------------------------- # Settings (all from environment / Space secrets) # -------------------------------------------------------------------------------------- def _normalize_db_url(raw: str) -> str: """Forgive common copy-paste mistakes in the DATABASE_URL secret: stray spaces or newlines, the [ ] placeholder brackets left around the password, and unescaped symbols in the password.""" from urllib.parse import quote, unquote url = re.sub(r"\s+", "", raw or "") m = re.match(r"^(postgres(?:ql)?://)([^:/@]+):(.*)@([^@]+)$", url) if not m: return url scheme, user, password, rest = m.groups() if password.startswith("[") and password.endswith("]"): password = password[1:-1] return f"{scheme}{user}:{quote(unquote(password), safe='')}@{rest}" def _database_url() -> str: """DB_PASSWORD alone is enough: the Supabase pooler address is built from SUPABASE_URL. Otherwise use the full DATABASE_URL connection string.""" from urllib.parse import quote password = (os.getenv("DB_PASSWORD") or "").strip() ref = re.match(r"https://([a-z0-9]+)\.supabase\.co", os.getenv("SUPABASE_URL", "").strip()) if password and ref: host = os.getenv("DB_POOLER_HOST", "aws-1-eu-central-1.pooler.supabase.com") return f"postgresql://postgres.{ref.group(1)}:{quote(password, safe='')}@{host}:6543/postgres" return _normalize_db_url(os.getenv("DATABASE_URL", "")) DATABASE_URL = _database_url() SUPABASE_URL = os.getenv("SUPABASE_URL", "").rstrip("/") SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "") SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "") STORAGE_BUCKET = os.getenv("STORAGE_BUCKET", "evalbot") DEV_AUTH = os.getenv("EVALBOT_DEV_AUTH", "") == "1" # local testing only: accepts "dev::" LOCAL_STORAGE_DIR = os.getenv("EVALBOT_LOCAL_STORAGE", "./platform_storage") def enabled() -> bool: return bool(DATABASE_URL) _providers: Dict[str, Any] = {"at": 0.0, "value": {}} def _auth_providers() -> Dict[str, bool]: """Which sign-in providers are switched on in Supabase (cached for 5 minutes).""" if not SUPABASE_URL or time.time() - _providers["at"] < 300: return _providers["value"] try: r = httpx.get(f"{SUPABASE_URL}/auth/v1/settings", headers={"apikey": SUPABASE_ANON_KEY}, timeout=5) ext = r.json().get("external", {}) if r.status_code == 200 else {} _providers["value"] = {"google": bool(ext.get("google"))} except Exception: pass _providers["at"] = time.time() return _providers["value"] def public_config() -> Dict[str, Any]: return {"enabled": enabled(), "supabase_url": SUPABASE_URL, "supabase_anon_key": SUPABASE_ANON_KEY, "dev_auth": DEV_AUTH, "providers": _auth_providers()} # -------------------------------------------------------------------------------------- # Database # -------------------------------------------------------------------------------------- _pool = None _pool_lock = threading.Lock() def pool(): global _pool if _pool is None: with _pool_lock: if _pool is None: from psycopg.rows import dict_row from psycopg_pool import ConnectionPool _pool = ConnectionPool(DATABASE_URL, min_size=1, max_size=5, open=True, kwargs={"row_factory": dict_row, "autocommit": True, "prepare_threshold": None}) # pgbouncer-compatible return _pool def init_db() -> None: if not enabled(): logger.warning("DATABASE_URL not set: platform features are disabled") return sql = open(os.path.join(os.path.dirname(__file__), "schema.sql"), encoding="utf-8").read() with pool().connection() as conn: conn.execute(sql) logger.info("Database schema ready") def _adapt(params): from psycopg.types.json import Jsonb if params is None: return None if isinstance(params, dict): return {k: Jsonb(v) if isinstance(v, (dict, list)) else v for k, v in params.items()} return [Jsonb(v) if isinstance(v, (dict, list)) else v for v in params] def q(sql: str, params=None) -> List[Dict]: with pool().connection() as conn: cur = conn.execute(sql, _adapt(params)) return cur.fetchall() if cur.description else [] def q1(sql: str, params=None) -> Optional[Dict]: rows = q(sql, params) return rows[0] if rows else None # -------------------------------------------------------------------------------------- # Auth (Supabase access tokens) # -------------------------------------------------------------------------------------- _token_cache: Dict[str, tuple] = {} def _verify_token(token: str) -> Dict: hit = _token_cache.get(token) if hit and hit[1] > time.time(): return hit[0] if DEV_AUTH and token.startswith("dev:"): _, uid, email = token.split(":", 2) user = {"id": uid, "email": email} else: if not SUPABASE_URL: raise HTTPException(500, "Authentication is not configured") r = httpx.get(f"{SUPABASE_URL}/auth/v1/user", timeout=10, headers={"Authorization": f"Bearer {token}", "apikey": SUPABASE_ANON_KEY}) if r.status_code != 200: raise HTTPException(401, "Your session has expired. Please sign in again.") d = r.json() user = {"id": d["id"], "email": d.get("email"), "name": (d.get("user_metadata") or {}).get("full_name") or (d.get("user_metadata") or {}).get("name")} _token_cache[token] = (user, time.time() + 300) return user def current_user(authorization: str = Header(default="")) -> Dict: if not enabled(): raise HTTPException(503, "The platform is not configured yet") if not authorization.lower().startswith("bearer "): raise HTTPException(401, "Please sign in") user = _verify_token(authorization.split(" ", 1)[1].strip()) prof = q1("select * from profiles where id = %s", [user["id"]]) if prof is None: prof = q1("insert into profiles (id, email, full_name) values (%s, %s, %s) " "on conflict (id) do update set email = excluded.email returning *", [user["id"], user.get("email"), user.get("name")]) return prof def recruiter(user: Dict = Depends(current_user)) -> Dict: if user["role"] != "recruiter" or not user.get("company_id"): raise HTTPException(403, "This page is for company accounts") return user # -------------------------------------------------------------------------------------- # File storage (Supabase Storage, or a local folder for development) # -------------------------------------------------------------------------------------- def _storage_headers(content_type: Optional[str] = None) -> Dict[str, str]: h = {"apikey": SUPABASE_SERVICE_ROLE_KEY} if not SUPABASE_SERVICE_ROLE_KEY.startswith("sb_"): # legacy JWT keys also go in Authorization; new sb_secret keys must not h["Authorization"] = f"Bearer {SUPABASE_SERVICE_ROLE_KEY}" if content_type: h["Content-Type"] = content_type return h def _use_supabase_storage() -> bool: return bool(SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY) # Optional: keep recordings and PDFs in a private Hugging Face dataset (100 GB+ free) instead of # Supabase Storage (1 GB free). Enabled when HF_STORAGE_REPO and HF_TOKEN are both set. HF_STORAGE_REPO = os.getenv("HF_STORAGE_REPO", "") HF_TOKEN = os.getenv("HF_TOKEN", "") _hf_cache_dir = os.path.join(LOCAL_STORAGE_DIR, ".hf_cache") def _use_hf_storage() -> bool: return bool(HF_STORAGE_REPO and HF_TOKEN) def storage_backend() -> str: return "huggingface" if _use_hf_storage() else "supabase" if _use_supabase_storage() else "local" def _hf_api(): from huggingface_hub import HfApi return HfApi(token=HF_TOKEN) def ensure_bucket() -> None: if _use_hf_storage(): _hf_api().create_repo(HF_STORAGE_REPO, repo_type="dataset", private=True, exist_ok=True) os.makedirs(_hf_cache_dir, exist_ok=True) logger.info("File storage: private HF dataset %s", HF_STORAGE_REPO) return if not _use_supabase_storage(): os.makedirs(LOCAL_STORAGE_DIR, exist_ok=True) return r = httpx.get(f"{SUPABASE_URL}/storage/v1/bucket/{STORAGE_BUCKET}", headers=_storage_headers(), timeout=15) if r.status_code != 200: r = httpx.post(f"{SUPABASE_URL}/storage/v1/bucket", headers=_storage_headers("application/json"), json={"id": STORAGE_BUCKET, "name": STORAGE_BUCKET, "public": False}, timeout=15) logger.info("Created storage bucket %s: %s", STORAGE_BUCKET, r.status_code) def put_file(path: str, data: bytes, content_type: str) -> str: if _use_hf_storage(): _hf_api().upload_file(path_or_fileobj=data, path_in_repo=path, repo_id=HF_STORAGE_REPO, repo_type="dataset", commit_message=f"add {path}") full = os.path.join(_hf_cache_dir, path) # keep a local copy so reads are instant os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, "wb") as f: f.write(data) return path if not _use_supabase_storage(): full = os.path.join(LOCAL_STORAGE_DIR, path) os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, "wb") as f: f.write(data) return path r = httpx.post(f"{SUPABASE_URL}/storage/v1/object/{STORAGE_BUCKET}/{path}", content=data, timeout=120, headers={**_storage_headers(content_type), "x-upsert": "true"}) if r.status_code >= 300: raise RuntimeError(f"Storage upload failed: {r.status_code} {r.text[:200]}") return path def get_file(path: str) -> bytes: if _use_hf_storage(): full = os.path.join(_hf_cache_dir, path) if not os.path.exists(full): from huggingface_hub import hf_hub_download try: return open(hf_hub_download(HF_STORAGE_REPO, path, repo_type="dataset", token=HF_TOKEN, local_dir=_hf_cache_dir), "rb").read() except Exception as e: raise RuntimeError(f"Storage download failed: {e}") with open(full, "rb") as f: return f.read() if not _use_supabase_storage(): with open(os.path.join(LOCAL_STORAGE_DIR, path), "rb") as f: return f.read() r = httpx.get(f"{SUPABASE_URL}/storage/v1/object/{STORAGE_BUCKET}/{path}", headers=_storage_headers(), timeout=120) if r.status_code != 200: raise RuntimeError(f"Storage download failed: {r.status_code}") return r.content def dumps(obj) -> str: return json.dumps(obj, ensure_ascii=False, default=str)