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}")