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