Spaces:
Running
Running
| """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:<uuid>:<email>" | |
| 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) | |