melodix-api / app /utils /supabase_storage.py
GitHub Action
deploy from github actions
440bac0
Raw
History Blame Contribute Delete
2.28 kB
import re
import logging
from supabase import create_client, Client
logger = logging.getLogger("uvicorn.error")
SUPABASE_URL = "https://egkydqberhebzkqtltic.supabase.co"
SUPABASE_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6ImVna3lkcWJlcmhlYnprcXRsdGljIiwicm9sZSI6ImFub24iLCJpYXQiOjE3NzkzNDA4NTMsImV4cCI6MjA5NDkxNjg1M30.KfLrhTHrC2W2g6IDuGuMA5Gd_wnj4YOuRdV6hhcSCqM"
BUCKET_NAME = "melodix-audio"
def get_supabase_client() -> Client:
return create_client(SUPABASE_URL, SUPABASE_KEY)
def parse_supabase_path(url: str) -> str:
"""
Extrae el path del archivo dentro del bucket a partir de la URL pública de Supabase.
Ej: https://.../public/melodix-audio/originals/xxx.mp3 -> originals/xxx.mp3
"""
if not url:
return None
pattern = rf"/storage/v1/object/public/{BUCKET_NAME}/(.+)$"
match = re.search(pattern, url)
if match:
return match.group(1)
return None
def delete_supabase_file(path_in_bucket: str):
"""
Elimina un archivo específico del bucket de Supabase.
"""
if not path_in_bucket:
return
try:
client = get_supabase_client()
res = client.storage.from_(BUCKET_NAME).remove([path_in_bucket])
logger.info(f"Archivo eliminado de Supabase: {path_in_bucket}. Resultado: {res}")
except Exception as e:
logger.warning(f"No se pudo eliminar archivo de Supabase ({path_in_bucket}): {e}")
def delete_supabase_folder(folder_path: str):
"""
Elimina todos los archivos dentro de una carpeta del bucket.
"""
if not folder_path:
return
try:
client = get_supabase_client()
# Listar archivos en la carpeta
files = client.storage.from_(BUCKET_NAME).list(folder_path)
if files:
paths_to_remove = [f"{folder_path}/{f['name']}" for f in files if f.get('name') and f['name'] != '.emptyFolderPlaceholder']
if paths_to_remove:
res = client.storage.from_(BUCKET_NAME).remove(paths_to_remove)
logger.info(f"Carpeta eliminada de Supabase: {folder_path}. Archivos eliminados: {paths_to_remove}. Resultado: {res}")
except Exception as e:
logger.warning(f"No se pudo eliminar carpeta de Supabase ({folder_path}): {e}")