Spaces:
Running on Zero
Running on Zero
File size: 4,430 Bytes
f1ef7e2 1817714 f1ef7e2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """
Supabase Storage access over plain REST (no SDK dependency).
Artifacts (call JSON, mp3, sentence_segments, sentiment, raw transcript, and
the original upload for restart recovery) live in one PUBLIC bucket so reads
need no auth and the Supabase CDN serves audio Range requests directly. Writes
use the service-role key.
Layout in the bucket (keys):
calls/{id}.json audio/{id}.mp3 sentence_segments/{id}.json
sentiment/{id}.json transcripts/{id}.json uploads/{id}/<name>
"""
import os
import json
import mimetypes
from pathlib import Path
import requests
BUCKET = os.environ.get("SUPABASE_BUCKET", "call-artifacts")
_TIMEOUT = 30
class StorageError(RuntimeError):
pass
def _base():
url = os.environ.get("SUPABASE_URL", "").rstrip("/")
if not url:
raise StorageError("SUPABASE_URL is not set")
return url
def _service_key():
# secret/service-role key preferred (storage writes bypass RLS); falls back
# to SUPABASE_KEY, which only authorizes writes if the bucket grants anon
# insert. Reads from a public bucket work with either.
key = os.environ.get("SUPABASE_SERVICE_KEY") or os.environ.get("SUPABASE_KEY", "")
if not key:
raise StorageError("neither SUPABASE_SERVICE_KEY nor SUPABASE_KEY is set")
return key
def is_configured():
return bool(os.environ.get("SUPABASE_URL")
and (os.environ.get("SUPABASE_SERVICE_KEY")
or os.environ.get("SUPABASE_KEY")))
def object_url(key):
"""Authenticated object endpoint (used for writes)."""
return f"{_base()}/storage/v1/object/{BUCKET}/{key}"
def public_url(key):
"""Public read URL served by the Supabase CDN (Range-capable)."""
return f"{_base()}/storage/v1/object/public/{BUCKET}/{key}"
# ── writes ──────────────────────────────────────────────────────────────────
def upload_bytes(key, data, content_type="application/octet-stream"):
# `apikey` header (not just Authorization: Bearer) is required for the new
# sb_secret_ key format -- the object endpoint parses Bearer tokens as JWTs
# and rejects the non-JWT key with "Invalid Compact JWS". Sending both is
# compatible with legacy JWT service_role keys too.
k = _service_key()
headers = {
"apikey": k,
"Authorization": f"Bearer {k}",
"Content-Type": content_type,
"x-upsert": "true", # overwrite -> idempotent re-runs
}
r = requests.post(object_url(key), headers=headers, data=data, timeout=_TIMEOUT)
if r.status_code not in (200, 201):
raise StorageError(f"upload {key} failed: {r.status_code} {r.text[:200]}")
return key
def upload_file(key, path, content_type=None):
if content_type is None:
content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
with open(path, "rb") as f:
return upload_bytes(key, f.read(), content_type)
def upload_json(key, obj):
return upload_bytes(key, json.dumps(obj).encode("utf-8"), "application/json")
# ── reads (public bucket; small local cache since artifacts are immutable) ──
_CACHE_DIR = Path(os.environ.get("CAPSTONE_DATA_ROOT", "/tmp")) / "_storage_cache"
def _cache_path(key):
return _CACHE_DIR / key
def exists(key):
try:
r = requests.head(public_url(key), timeout=_TIMEOUT)
return r.status_code == 200
except requests.RequestException:
return False
def download_bytes(key, use_cache=True):
cp = _cache_path(key)
if use_cache and cp.exists():
return cp.read_bytes()
r = requests.get(public_url(key), timeout=_TIMEOUT)
missing = r.status_code == 404
if r.status_code == 400:
try:
payload = r.json()
missing = (str(payload.get("statusCode")) == "404"
or payload.get("error") == "not_found")
except ValueError:
pass
if missing:
raise FileNotFoundError(key)
if r.status_code != 200:
raise StorageError(f"download {key} failed: {r.status_code}")
if use_cache:
cp.parent.mkdir(parents=True, exist_ok=True)
cp.write_bytes(r.content)
return r.content
def stream_json(key, use_cache=True):
return json.loads(download_bytes(key, use_cache=use_cache).decode("utf-8"))
|