deploy from github actions

app/routers/alabanza.py

@@ -379,7 +379,9 @@ def expulsar_miembro(grupo_id: int, miembro_id: int, db: Session = Depends(get_d
     # Expulsar
     target_ug.estado = "expulsado"
     
-    # Remover lo que ese miembro compartió con este grupo
+    miembro_user = db.query(User).filter(User.id == miembro_id).first()
+    is_premium = miembro_user.is_premium if miembro_user else False
+
     kicked_lyrics = db.query(Lyric).filter(Lyric.created_by == miembro_id, Lyric.groups_shared.any(Group.id == grupo_id)).all()
     for l in kicked_lyrics:
         group_to_remove = db.query(Group).filter(Group.id == grupo_id).first()
@@ -395,6 +397,12 @@ def expulsar_miembro(grupo_id: int, miembro_id: int, db: Session = Depends(get_d
             t.groups_shared.remove(group_to_remove)
             if t.group_id == grupo_id:
                 t.group_id = t.groups_shared[0].id if t.groups_shared else None
+            # Si no quedan grupos compartidos y es usuario normal, vuelve a expirar
+            if not t.groups_shared:
+                if is_premium:
+                    t.cloud_expires_at = None
+                else:
+                    t.cloud_expires_at = datetime.utcnow() + timedelta(days=30)
     
     db.commit()
     broadcast_to_group(grupo_id, {"type": "group_updated", "grupo_id": grupo_id}, db)
@@ -426,6 +434,12 @@ def salir_del_grupo(grupo_id: int, db: Session = Depends(get_db), current_user:
             t.groups_shared.remove(group_to_remove)
             if t.group_id == grupo_id:
                 t.group_id = t.groups_shared[0].id if t.groups_shared else None
+            # Si no quedan grupos compartidos y es usuario normal, vuelve a expirar
+            if not t.groups_shared:
+                if current_user.is_premium:
+                    t.cloud_expires_at = None
+                else:
+                    t.cloud_expires_at = datetime.utcnow() + timedelta(days=30)
     
     db.commit()
     broadcast_to_group(grupo_id, {"type": "group_updated", "grupo_id": grupo_id}, db)
@@ -447,9 +461,19 @@ def eliminar_grupo(grupo_id: int, db: Session = Depends(get_db), current_user: U
             
     tasks_affected = db.query(Task).filter(Task.groups_shared.any(Group.id == grupo_id)).all()
     for t in tasks_affected:
+        group_to_remove = db.query(Group).filter(Group.id == grupo_id).first()
+        if group_to_remove in t.groups_shared:
+            t.groups_shared.remove(group_to_remove)
         if t.group_id == grupo_id:
-            other_groups = [g.id for g in t.groups_shared if g.id != grupo_id]
+            other_groups = [g.id for g in t.groups_shared]
             t.group_id = other_groups[0] if other_groups else None
+        # Si no pertenecen a ningún grupo y son de usuario normal, configurar expiración
+        if not t.groups_shared:
+            owner = db.query(User).filter(User.id == t.user_id).first()
+            if owner and not owner.is_premium:
+                t.cloud_expires_at = datetime.utcnow() + timedelta(days=30)
+            else:
+                t.cloud_expires_at = None
     
     # Eliminar invitaciones del grupo
     db.query(Invitation).filter(Invitation.group_id == grupo_id).delete()

app/routers/auth.py

@@ -30,70 +30,9 @@ router = APIRouter(prefix="/auth", tags=["auth"])
 
 def _enqueue_cloud_stems_for_new_device(db: Session, user, new_device_id: str):
     """
-    Cuando se registra un nuevo dispositivo, busca todas las tareas del usuario
-    que tienen archivos vigentes en la nube (telegram_file_ids + cloud_expires_at futuro)
-    y crea/actualiza entradas B2SyncQueue con el nuevo device_id como pending_recipient.
-    Así el PeerSyncService detecta 'pending_my_download' en el primer ciclo y descarga
-    automáticamente sin pedir confirmación al usuario.
+    No-op: La sincronización P2P por cola de dispositivos ha sido eliminada.
     """
-    try:
-        from app.models.process import Task
-        from app.models.sync import B2SyncQueue
-        from datetime import datetime
-        import json as _json
-
-        now = datetime.utcnow()
-        tasks_with_cloud = db.query(Task).filter(
-            Task.user_id == user.id,
-            Task.telegram_file_ids.isnot(None),
-        ).all()
-
-        enqueued = 0
-        for task in tasks_with_cloud:
-            # Verificar que los archivos no han expirado
-            if task.cloud_expires_at and task.cloud_expires_at < now:
-                continue
-            if not task.telegram_file_ids:
-                continue
-
-            # Obtener o crear entrada en B2SyncQueue
-            sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task.task_id).first()
-
-            if not sync_entry:
-                # Construir file_url desde telegram_file_ids
-                sync_entry = B2SyncQueue(
-                    id=task.task_id,
-                    file_key=f"stems/{task.task_id}",
-                    file_url=task.telegram_file_ids,
-                    sync_type="device",
-                    target_id=str(user.id),
-                    pending_recipients=[new_device_id],
-                    is_uploaded=True,
-                    missing_stems=None
-                )
-                db.add(sync_entry)
-            else:
-                # Añadir el nuevo device_id a pending_recipients si no está ya
-                pending = sync_entry.pending_recipients
-                if isinstance(pending, str):
-                    pending = _json.loads(pending)
-                pending = list(pending) if pending else []
-                if new_device_id not in pending:
-                    pending.append(new_device_id)
-                    sync_entry.pending_recipients = pending
-                # Asegurarse que is_uploaded sea True (hay archivos en la nube)
-                sync_entry.is_uploaded = True
-                # Actualizar file_url con las URLs actuales si estaba vacío
-                if not sync_entry.file_url or sync_entry.file_url == "{}":
-                    sync_entry.file_url = task.telegram_file_ids
-
-            enqueued += 1
-
-        if enqueued > 0:
-            db.commit()
-            logger.info(f"[auth] Encoladas {enqueued} pistas en B2SyncQueue para nuevo dispositivo {new_device_id} del usuario {user.username}")
-    except Exception as e:
-        logger.warning(f"[auth] Error encolando pistas para nuevo dispositivo {new_device_id}: {e}")
+    pass
 
 # ==========================================
 # CONFIGURACIÓN DE SEGURIDAD

app/routers/migration.py

@@ -12,10 +12,10 @@ from app.database import get_db
 from app.models.user import User
 from app.models.sync import MigrationBackup
 from app.routers.auth import get_current_user
-from app.utils.b2_storage import (
-    upload_to_b2,
-    delete_folder_from_b2,
-    get_b2_download_url
+from app.utils.hf_storage import (
+    upload_to_hf,
+    delete_folder_from_hf,
+    get_hf_download_url
 )
 
 logger = logging.getLogger(__name__)
@@ -39,7 +39,7 @@ def cleanup_expired_backups(db: Session, user_id: int):
     for eb in expired:
         try:
             logger.info(f"Limpiando respaldo expirado {eb.id} para usuario {user_id}")
-            delete_folder_from_b2(f"migrations/{user_id}/")
+            delete_folder_from_hf(f"migrations/{user_id}")
             eb.status = "expired"
         except Exception as e:
             logger.error(f"Error limpiando respaldo expirado {eb.id}: {e}")
@@ -54,7 +54,7 @@ def create_backup(
     current_user: User = Depends(get_current_user)
 ):
     """
-    Inicia un proceso de respaldo. Elimina cualquier respaldo anterior en B2 y BD.
+    Inicia un proceso de respaldo. Elimina cualquier respaldo anterior en Hugging Face y BD.
     """
     # 1. Limpieza de expirados
     cleanup_expired_backups(db, current_user.id)
@@ -65,10 +65,10 @@ def create_backup(
         MigrationBackup.status.in_(["pending", "completed"])
     ).all()
     
-    # Eliminar archivos en B2 y registros en BD de respaldos anteriores
+    # Eliminar archivos en Hugging Face y registros en BD de respaldos anteriores
     if active_backups:
         logger.info(f"Eliminando {len(active_backups)} respaldos activos previos para usuario {current_user.id}")
-        delete_folder_from_b2(f"migrations/{current_user.id}/")
+        delete_folder_from_hf(f"migrations/{current_user.id}")
         for ab in active_backups:
             db.delete(ab)
         db.commit()
@@ -102,7 +102,7 @@ def upload_backup_file(
     current_user: User = Depends(get_current_user)
 ):
     """
-    Recibe un stem individual en formato .flac o .json y lo sube a la ruta migrations/ de B2.
+    Recibe un stem individual en formato .flac o .json y lo sube a la ruta migrations/ de Hugging Face.
     """
     # Buscar y validar el respaldo
     backup = db.query(MigrationBackup).filter(
@@ -142,9 +142,9 @@ def upload_backup_file(
         with open(temp_file_path, "wb") as buffer:
             shutil.copyfileobj(file.file, buffer)
             
-        # Subir a Backblaze B2
+        # Subir a Hugging Face Dataset
         object_key = f"migrations/{current_user.id}/{task_id}/{stem_name}.{ext}"
-        public_url = upload_to_b2(temp_file_path, object_key)
+        public_url = upload_to_hf(temp_file_path, object_key)
         
         # Registrar el archivo en la BD
         files_list = list(backup.files_info) if backup.files_info else []
@@ -159,7 +159,7 @@ def upload_backup_file(
         
     except Exception as e:
         logger.error(f"Error subiendo archivo de migración {stem_name} para {task_id}: {e}")
-        raise HTTPException(status_code=500, detail=f"Error en la subida a B2: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Error en la subida a Hugging Face: {str(e)}")
     finally:
         if os.path.exists(temp_file_path):
             os.remove(temp_file_path)
@@ -225,7 +225,7 @@ def restore_backup(
     current_user: User = Depends(get_current_user)
 ):
     """
-    Retorna el mapa de AsyncStorage y las URLs firmadas de descarga de B2 para todos los stems respaldados.
+    Retorna el mapa de AsyncStorage y las URLs de descarga de Hugging Face para todos los stems respaldados.
     """
     cleanup_expired_backups(db, current_user.id)
     
@@ -237,14 +237,13 @@ def restore_backup(
     if not backup:
         raise HTTPException(status_code=404, detail="No tienes ningún respaldo activo listo para restaurar")
         
-    # Generar URLs firmadas de descarga de B2
+    # Generar URLs de descarga de Hugging Face
     restore_files = []
     files_list = backup.files_info or []
     
     for f in files_list:
         file_key = f["file_key"]
-        # Generar una URL firmada válida por 4 horas (14400s) para asegurar que le dé tiempo a descargar
-        download_url = get_b2_download_url(file_key, expires_in=14400)
+        download_url = get_hf_download_url(file_key)
         restore_files.append({
             "task_id": f["task_id"],
             "stem_name": f["stem_name"],
@@ -263,7 +262,7 @@ def complete_restore(
     current_user: User = Depends(get_current_user)
 ):
     """
-    Confirma que el nuevo dispositivo ha terminado de descargar los archivos y elimina los datos de B2.
+    Confirma que el nuevo dispositivo ha terminado de descargar los archivos y elimina los datos de Hugging Face.
     """
     backup = db.query(MigrationBackup).filter(
         MigrationBackup.user_id == current_user.id,
@@ -271,9 +270,9 @@ def complete_restore(
     ).order_by(MigrationBackup.created_at.desc()).first()
     
     if backup:
-        # Eliminar archivos físicamente de B2
-        logger.info(f"Limpiando carpeta de B2 tras restauración exitosa de {backup.id}")
-        delete_folder_from_b2(f"migrations/{current_user.id}/")
+        # Eliminar archivos físicamente de Hugging Face
+        logger.info(f"Limpiando carpeta de HF tras restauración exitosa de {backup.id}")
+        delete_folder_from_hf(f"migrations/{current_user.id}")
         backup.status = "restored"
         db.commit()
         

app/routers/songs.py

@@ -657,67 +657,11 @@ async def historial_canciones(
     else:
         total = db.query(Task).filter(Task.user_id == current_user.id).count()
 
-    from app.models.sync import B2SyncQueue
-    from datetime import datetime, timedelta
-    
-    task_ids = [t.task_id for t in tasks]
-    sync_entries = {s.id: s for s in db.query(B2SyncQueue).filter(B2SyncQueue.id.in_(task_ids)).all()}
-    
     formatted_tasks = []
-    now = datetime.utcnow()
     
     for t in tasks:
-        sync_entry = sync_entries.get(t.task_id)
         sync_status = "available"
         missing_stems_for_me = []
-        
-        if sync_entry:
-            is_recipient = False
-            recipient_key = None
-            
-            user_devices = [d.device_id for d in current_user.devices]
-            
-            pending_recipients = sync_entry.pending_recipients
-            if isinstance(pending_recipients, str):
-                pending_recipients = _json.loads(pending_recipients)
-                
-            for dev in user_devices:
-                if dev in (pending_recipients or []):
-                    is_recipient = True
-                    recipient_key = dev
-                    break
-                        
-            if is_recipient:
-                if not sync_entry.is_uploaded:
-                    if sync_entry.uploading_by:
-                        if sync_entry.uploading_claimed_at and now - sync_entry.uploading_claimed_at > timedelta(minutes=5):
-                            sync_status = "pending_upload"
-                        else:
-                            sync_status = "uploading"
-                    else:
-                        sync_status = "pending_upload"
-                else:
-                    sync_status = "pending_my_download"
-                    
-                missing_stems = sync_entry.missing_stems
-                if isinstance(missing_stems, str):
-                    missing_stems = _json.loads(missing_stems)
-                    
-                if missing_stems and recipient_key in missing_stems:
-                    missing_stems_for_me = missing_stems[recipient_key]
-                else:
-                    count = t.stems or 4
-                    missing_stems_for_me = ["vocals", "no_vocals"] if count == 2 else (["vocals", "bass", "drums", "guitar", "piano", "other"] if count == 6 else ["vocals", "bass", "drums", "other"])
-            else:
-                if not sync_entry.is_uploaded:
-                    if sync_entry.uploading_by:
-                        if sync_entry.uploading_claimed_at and now - sync_entry.uploading_claimed_at > timedelta(minutes=5):
-                            sync_status = "pending_upload"
-                        else:
-                            sync_status = "uploading"
-                    else:
-                        sync_status = "pending_upload"
-                        
         # Detectar si hay voces corales generadas en la nube (al menos 2da_voz presente)
         has_harmony = False
         if t.telegram_file_ids:
@@ -966,11 +910,6 @@ async def get_stems(
     if task_record.cloud_expires_at and task_record.cloud_expires_at < datetime.utcnow():
         task_record.telegram_file_ids = None
         task_record.cloud_expires_at = None
-        from app.models.sync import B2SyncQueue
-        sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-        if sync_entry:
-            sync_entry.is_uploaded = False
-            sync_entry.file_url = "{}"
         db.commit()
         raise HTTPException(
             status_code=404,
@@ -980,17 +919,17 @@ async def get_stems(
     if task_record.telegram_file_ids:
         try:
             stem_urls = _json.loads(task_record.telegram_file_ids)
-            from app.utils.b2_storage import get_b2_download_url
+            from app.utils.hf_storage import get_hf_download_url
             
             presigned_urls = {}
             for stem_name, file_url in stem_urls.items():
                 if "stems/" in file_url:
                     # Extraer el key del objeto de forma genérica (ej. stems/task_id/stem.m4a)
-                    # No depende del dominio (backblazeb2.com, tebi.io, etc.)
+                    # No depende del dominio
                     parts = file_url.split("stems/")
                     if len(parts) > 1:
                         object_key = "stems/" + parts[-1]
-                        presigned_urls[stem_name] = get_b2_download_url(object_key)
+                        presigned_urls[stem_name] = get_hf_download_url(object_key)
                     else:
                         presigned_urls[stem_name] = file_url
                 else:
@@ -1017,7 +956,7 @@ async def reset_cloud_stems(
     current_user: User = Depends(get_current_user)
 ):
     """
-    Limpia los metadatos de stems en la nube para forzar una resincronización desde los dispositivos helpers.
+    Limpia los metadatos de stems en la nube para forzar una resincronización.
     """
     user_group_ids = [ug.group_id for ug in current_user.groups if ug.estado == 'activo']
     task_record = db.query(Task).filter(Task.task_id == task_id).filter(
@@ -1030,15 +969,9 @@ async def reset_cloud_stems(
     # Limpiar los metadatos de la nube
     task_record.telegram_file_ids = None
     task_record.cloud_expires_at = None
-    
-    from app.models.sync import B2SyncQueue
-    sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-    if sync_entry:
-        sync_entry.is_uploaded = False
-        sync_entry.file_url = "{}"
         
     db.commit()
-    logger.info(f"[{task_id}] Se han reseteado los stems en la nube para forzar resincronización.")
+    logger.info(f"[{task_id}] Se han reseteado los stems en la nube.")
     return {"success": True, "message": "Stems en la nube reseteados correctamente."}
 
 
@@ -1046,189 +979,18 @@ async def reset_cloud_stems(
 @router.post("/confirm-download/{task_id}")
 async def confirmar_descarga(
     task_id: str,
-    device_id: str = None,
+    device_id: Optional[str] = Form(None),
     download_type: str = "stems",
     db: Session = Depends(get_db),
     current_user: User = Depends(get_current_user)
 ):
     """
     El cliente llama a este endpoint al confirmar que descargó todos los stems.
-    Si está registrado en B2SyncQueue (Premium/Grupo), se remueve al destinatario. 
-    Al quedar en 0 destinatarios pendientes, se eliminan los archivos de Backblaze B2.
-    Si no existe en B2SyncQueue (Estándar), se dispara la tarea para borrar los stems locales.
+    Registramos la descarga y disparamos la eliminación de los archivos locales en el worker de procesamiento.
     """
-    import json as _json
-    from app.models.sync import B2SyncQueue
-    from app.utils.b2_storage import delete_from_b2
-    from datetime import datetime, timedelta
-    
-    # 1. Buscar en B2SyncQueue (Premium o Grupo)
-    sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
+    from datetime import datetime
     
-    if sync_entry:
-        logger.info(f"[{task_id}] Confirmación de descarga Premium/Grupo/Estándar recibida. sync_type: {sync_entry.sync_type}")
-        
-        # Si es usuario estándar (is_premium == False), purgar de inmediato de B2 para liberar espacio
-        if sync_entry.sync_type == "device" and not current_user.is_premium:
-            # Comprobar si tiene voces pero no se han generado las armonías aún
-            has_vocals = False
-            has_harmonies = False
-            try:
-                urls_dict = _json.loads(sync_entry.file_url) if sync_entry.file_url else {}
-                has_vocals = "vocals" in urls_dict
-                has_harmonies = "2da_voz" in urls_dict
-            except Exception:
-                pass
-
-            if has_vocals and not has_harmonies:
-                logger.info(f"[{task_id}] Pospone purga de B2 para usuario estándar porque aún no se han generado las armonías corales.")
-            elif has_harmonies and download_type != "harmonies":
-                logger.info(f"[{task_id}] Pospone purga de B2 para usuario estándar porque la canción tiene armonías corales y la descarga confirmada es '{download_type}'.")
-            else:
-                logger.info(f"[{task_id}] Usuario estándar detectado (tipo: {download_type}). Eliminando de B2 de inmediato...")
-                try:
-                    urls_dict = _json.loads(sync_entry.file_url)
-                    for stem_name, stem_url in urls_dict.items():
-                        if stem_name.endswith("_pitch"):
-                            key = f"stems/{task_id}/{stem_name}.json"
-                        else:
-                            ext = "m4a" if ".m4a" in stem_url.lower() else "flac"
-                            key = f"stems/{task_id}/{stem_name}.{ext}"
-                        delete_from_b2(key)
-                    logger.info(f"[{task_id}] Stems de usuario estándar eliminados de B2.")
-                except Exception as e:
-                    logger.error(f"[{task_id}] Error eliminando stems de B2 para usuario estándar: {e}")
-                
-                db.delete(sync_entry)
-                # También limpiar metadatos de la nube en Task de inmediato
-                task_record_std = db.query(Task).filter(Task.task_id == task_id).first()
-                if task_record_std:
-                    task_record_std.telegram_file_ids = None
-                    task_record_std.cloud_expires_at = None
-                db.commit()
-                
-                # Por si acaso existieran archivos locales, gatillar la tarea de eliminación local
-                from tasks import eliminar_archivos_locales_task
-                eliminar_archivos_locales_task.delay(task_id)
-        else:
-            if sync_entry:
-                import json as _json
-                to_remove = device_id or str(current_user.id)
-                
-                # Comprobar si la canción tiene armonías
-                has_harmonies = False
-                try:
-                    urls_dict = _json.loads(sync_entry.file_url) if sync_entry.file_url else {}
-                    has_harmonies = "2da_voz" in urls_dict
-                except Exception:
-                    pass
-                
-                # Si tiene armonías y la descarga confirmada es de los stems (no armonías), posponer la remoción y limpieza
-                if has_harmonies and download_type != "harmonies":
-                    logger.info(f"[{task_id}] Pospone remoción/purga de B2 para Premium/Grupo porque la canción tiene armonías corales y la descarga confirmada es '{download_type}'.")
-                else:
-                    pending = sync_entry.pending_recipients
-                    if isinstance(pending, str):
-                        pending = _json.loads(pending)
-                    pending = list(pending) if pending else []
-                    
-                    if to_remove in pending:
-                        pending.remove(to_remove)
-                    else:
-                        # Fallback: si to_remove no está pero coincide con el user_id, intentar buscar si alguno de los dispositivos de este usuario está en pending
-                        user_devices = [d.device_id for d in current_user.devices]
-                        for dev in user_devices:
-                            if dev in pending:
-                                pending.remove(dev)
-                                logger.info(f"[{task_id}] Removido dispositivo del usuario por fallback: {dev}")
-                                break
-                    sync_entry.pending_recipients = pending
-                    
-                    # También remover de missing_stems
-                    missing = sync_entry.missing_stems
-                    if isinstance(missing, str):
-                        missing = _json.loads(missing)
-                    missing = dict(missing) if missing else {}
-                    
-                    if to_remove in missing:
-                        del missing[to_remove]
-                    else:
-                        # Fallback para missing_stems
-                        user_devices = [d.device_id for d in current_user.devices]
-                        for dev in user_devices:
-                            if dev in missing:
-                                del missing[dev]
-                                logger.info(f"[{task_id}] Removido dispositivo de missing_stems por fallback: {dev}")
-                                break
-                    sync_entry.missing_stems = missing
-                        
-                    db.commit()
-                    logger.info(f"[{task_id}] Removido {to_remove} de destinatarios. Restan: {pending}")
-                    
-                    # --- NUEVA LÓGICA DE DETECCION DE DISPOSITIVOS ACTIVOS ---
-                    from app.models.user import Device
-                    
-                    active_threshold = datetime.utcnow() - timedelta(minutes=5)
-                    
-                    if sync_entry.sync_type == "group":
-                        from app.models.lyrics import UserGroup
-                        active_members = db.query(UserGroup).filter(
-                            UserGroup.group_id == int(sync_entry.target_id),
-                            UserGroup.estado == "activo"
-                        ).all()
-                        member_ids = [m.user_id for m in active_members]
-                        other_active_recipients = [
-                            str(d.user_id) for d in db.query(Device).filter(
-                                Device.user_id.in_(member_ids),
-                                Device.last_active >= active_threshold,
-                                Device.user_id != current_user.id
-                            ).all()
-                        ]
-                        has_other_active_recipients = any(uid in pending for uid in other_active_recipients)
-                    else:
-                        other_active_devices = [
-                            d.device_id for d in current_user.devices 
-                            if d.last_active and d.last_active >= active_threshold and d.device_id != to_remove
-                        ]
-                        has_other_active_recipients = any(dev in pending for dev in other_active_devices)
-                    
-                    # Si se vacía la lista de pendientes o no queda ningún otro destinatario ACTIVO,
-                    # eliminar de Tebi para liberar espacio y marcar is_uploaded = False si quedan inactivos.
-                    if not pending or not has_other_active_recipients:
-                        try:
-                            urls_dict = _json.loads(sync_entry.file_url)
-                            for stem_name, stem_url in urls_dict.items():
-                                if stem_name.endswith("_pitch"):
-                                    key = f"stems/{task_id}/{stem_name}.json"
-                                else:
-                                    ext = "m4a" if ".m4a" in stem_url.lower() else "flac"
-                                    key = f"stems/{task_id}/{stem_name}.{ext}"
-                                delete_from_b2(key)
-                            logger.info(f"[{task_id}] No quedan destinatarios activos. Archivos eliminados de Tebi. Restantes offline: {pending}")
-                        except Exception as e:
-                            logger.error(f"[{task_id}] Error eliminando stems de B2: {e}")
-                        
-                        # También limpiar metadatos de la nube en Task de inmediato
-                        task_record_prm = db.query(Task).filter(Task.task_id == task_id).first()
-                        if task_record_prm:
-                            task_record_prm.telegram_file_ids = None
-                            task_record_prm.cloud_expires_at = None
-                        
-                        if not pending:
-                            db.delete(sync_entry)
-                            db.commit()
-                        else:
-                            sync_entry.is_uploaded = False
-                            db.commit()
-            else:
-                logger.info(f"[{task_id}] El destinatario {device_id if device_id else str(current_user.id)} no estaba en pendientes.")
-    else:
-        # Caso Estándar: Eliminar carpeta local en la laptop
-        from tasks import eliminar_archivos_locales_task
-        eliminar_archivos_locales_task.delay(task_id)
-        logger.info(f"[{task_id}] Caso Estándar: Tarea de eliminación local disparada.")
-
-    # 2. Registrar downloaded_at en la tabla Task
+    # 1. Registrar downloaded_at en la tabla Task
     user_group_ids = [ug.group_id for ug in current_user.groups if ug.estado == 'activo']
     task_record = db.query(Task).filter(Task.task_id == task_id).filter(
         (Task.user_id == current_user.id) | (Task.groups_shared.any(Group.id.in_(user_group_ids)))
@@ -1236,20 +998,18 @@ async def confirmar_descarga(
 
     if task_record:
         task_record.downloaded_at = datetime.utcnow()
-        # Eliminar pista original y carpeta de stems en Supabase (por si acaso quedara algún residuo)
-        if task_record.cloud_url:
-            orig_path = parse_supabase_path(task_record.cloud_url)
-            if orig_path:
-                delete_supabase_file(orig_path)
-        delete_supabase_folder(f"stems/{task_id}")
         db.commit()
-        logger.info(f"[{task_id}] downloaded_at actualizado y limpieza Supabase realizada.")
+        logger.info(f"[{task_id}] downloaded_at actualizado. Disparando borrado local en el worker...")
+        
+        # Eliminar carpeta local en el worker de procesamiento
+        from tasks import eliminar_archivos_locales_task
+        eliminar_archivos_locales_task.delay(task_id)
     else:
         logger.warning(f"[{task_id}] Tarea no encontrada para actualizar downloaded_at.")
 
     return {
         "message": "Descarga confirmada",
-        "pending_recipients": sync_entry.pending_recipients if sync_entry else []
+        "pending_recipients": []
     }
 
 
@@ -1309,45 +1069,15 @@ async def limpiar_tarea(
         except Exception as e:
             logger.warning(f"Error general al limpiar archivos de {task_id}: {e}. Se continuará con el borrado en BD.")
     
-    # Eliminar stems del canal de Telegram (cuando el usuario borra la pista explícitamente)
-    if task_record.telegram_message_ids:
-        try:
-            import json as _json
-            from app.utils.telegram_storage import delete_telegram_message
-            message_ids = _json.loads(task_record.telegram_message_ids)
-            for stem_name, msg_id in message_ids.items():
-                delete_telegram_message(int(msg_id))
-                logger.info(f"Stem '{stem_name}' eliminado de Telegram (message_id={msg_id})")
-        except Exception as e:
-            logger.warning(f"Error eliminando stems de Telegram para {task_id}: {e}")
-    
-    # Eliminar stems de Backblaze B2 si existen
+    # Eliminar stems de Hugging Face Dataset si existen
     if task_record.telegram_file_ids:
         try:
-            import json as _json
-            from app.utils.b2_storage import delete_from_b2
-            urls_dict = _json.loads(task_record.telegram_file_ids)
-            for stem_name, stem_url in urls_dict.items():
-                if stem_name.endswith("_pitch"):
-                    key = f"stems/{task_id}/{stem_name}.json"
-                else:
-                    ext = "m4a" if ".m4a" in stem_url.lower() else "flac"
-                    key = f"stems/{task_id}/{stem_name}.{ext}"
-                delete_from_b2(key)
-            logger.info(f"[{task_id}] Stems eliminados de B2 por eliminación explícita de la pista.")
+            from app.utils.hf_storage import delete_folder_from_hf
+            delete_folder_from_hf(f"stems/{task_id}")
+            logger.info(f"[{task_id}] Stems eliminados de Hugging Face por eliminación explícita.")
         except Exception as e:
-            logger.warning(f"Error eliminando stems de B2 al borrar pista {task_id}: {e}")
+            logger.warning(f"Error eliminando stems de Hugging Face: {e}")
 
-    # Eliminar de B2SyncQueue si existe
-    try:
-        from app.models.sync import B2SyncQueue
-        sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-        if sync_entry:
-            db.delete(sync_entry)
-            logger.info(f"[{task_id}] Entrada eliminada de B2SyncQueue por eliminación de la pista.")
-    except Exception as e:
-        logger.warning(f"Error eliminando de B2SyncQueue para {task_id}: {e}")
-    
     # Eliminar pista original y stems de Supabase storage
     if task_record.cloud_url:
         orig_path = parse_supabase_path(task_record.cloud_url)
@@ -1531,342 +1261,3 @@ async def harmonize_song(
     return {"status": "processing", "message": "Procesamiento de armonias vocales iniciado en Celery."}
 
 
-# ==========================================
-# PEER-TO-PEER B2 SYNCHRONIZATION ENDPOINTS
-# ==========================================
-from typing import List, Optional
-from pydantic import BaseModel
-
-class MissingStemItem(BaseModel):
-    task_id: str
-    missing_stems: List[str]
-
-class ReportMissingRequest(BaseModel):
-    device_id: Optional[str] = None
-    missing_items: List[MissingStemItem]
-
-@router.post("/sync-requests/report-missing")
-async def report_missing_stems(
-    request: ReportMissingRequest,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """
-    Registra que el usuario o dispositivo actual no tiene ciertos stems de una o más canciones.
-    Crea o actualiza entradas en B2SyncQueue para coordinar la subida P2P.
-    """
-    import json as _json
-    from app.models.sync import B2SyncQueue
-    
-    reported_tasks = []
-    
-    for item in request.missing_items:
-        # Verificar que la tarea existe
-        task_record = db.query(Task).filter(Task.task_id == item.task_id).first()
-        if not task_record:
-            continue
-            
-        # Verificar acceso (propietario o miembro de grupo activo)
-        user_group_ids = [ug.group_id for ug in current_user.groups if ug.estado == 'activo']
-        shared_gids_for_user = [gid for gid in task_record.group_ids if gid in user_group_ids]
-        has_access = (task_record.user_id == current_user.id) or bool(shared_gids_for_user)
-        if not has_access:
-            continue
-            
-        # Limpiar referencias de expiración/archivos en la nube, ya que se reportan como faltantes
-        task_record.telegram_file_ids = None
-        task_record.cloud_expires_at = None
-            
-        # Determinar sync_type, target_id, y to_add
-        target_group_id = shared_gids_for_user[0] if shared_gids_for_user else None
-        sync_type = "group" if target_group_id is not None else "device"
-        target_id = str(target_group_id) if sync_type == "group" else str(task_record.user_id)
-        
-        to_add = request.device_id if request.device_id else str(current_user.id)
-            
-        # Buscar entrada en B2SyncQueue
-        sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == item.task_id).first()
-        
-        if not sync_entry:
-            sync_entry = B2SyncQueue(
-                id=item.task_id,
-                file_key=f"stems/{item.task_id}",
-                file_url="{}",
-                sync_type=sync_type,
-                target_id=target_id,
-                pending_recipients=[to_add],
-                is_uploaded=False,
-                missing_stems={to_add: item.missing_stems}
-            )
-            db.add(sync_entry)
-        else:
-            to_add = request.device_id or str(current_user.id)
-            pending = sync_entry.pending_recipients
-            if isinstance(pending, str):
-                pending = _json.loads(pending)
-            pending = list(pending) if pending else []
-            
-            if to_add not in pending:
-                pending.append(to_add)
-            sync_entry.pending_recipients = pending
-            
-            missing = sync_entry.missing_stems
-            if isinstance(missing, str):
-                missing = _json.loads(missing)
-            missing = dict(missing) if missing else {}
-            
-            current_missing = set(missing.get(to_add, []))
-            current_missing.update(item.missing_stems)
-            missing[to_add] = list(current_missing)
-            sync_entry.missing_stems = missing
-            sync_entry.file_url = "{}"
-            sync_entry.is_uploaded = False
-            
-        db.commit()
-        reported_tasks.append(item.task_id)
-        
-    return {"message": "Reporte registrado", "reported_tasks": reported_tasks}
-
-
-@router.get("/sync-requests/pending")
-async def list_pending_sync_requests(
-    device_id: Optional[str] = None,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """
-    Retorna la lista de canciones en las que el usuario actual puede actuar como helper para subir stems.
-    """
-    from datetime import datetime, timedelta
-    from app.models.sync import B2SyncQueue
-    
-    user_group_ids = [ug.group_id for ug in current_user.groups if ug.estado == 'activo']
-    active_queues = db.query(B2SyncQueue).filter(B2SyncQueue.is_uploaded == False).all()
-    
-    pending_syncs = []
-    now = datetime.utcnow()
-    
-    for sq in active_queues:
-        is_locked = False
-        if sq.uploading_by:
-            if sq.uploading_claimed_at and now - sq.uploading_claimed_at <= timedelta(minutes=5):
-                is_locked = True
-                
-        if is_locked:
-            continue
-            
-        has_access = False
-        is_recipient = False
-        pending_recipients = sq.pending_recipients
-        if isinstance(pending_recipients, str):
-            import json as _json
-            pending_recipients = _json.loads(pending_recipients)
-        pending_recipients = pending_recipients or []
-            
-        if sq.sync_type == "group":
-            if int(sq.target_id) in user_group_ids:
-                has_access = True
-            if str(current_user.id) in pending_recipients or (device_id and device_id in pending_recipients):
-                is_recipient = True
-        else:
-            if str(current_user.id) == sq.target_id:
-                has_access = True
-            if device_id and device_id in pending_recipients:
-                is_recipient = True
-                
-        if has_access and not is_recipient:
-            required_stems = set()
-            missing_stems = sq.missing_stems
-            if isinstance(missing_stems, str):
-                import json as _json
-                missing_stems = _json.loads(missing_stems)
-                
-            if missing_stems:
-                for stems_list in missing_stems.values():
-                    required_stems.update(stems_list)
-                    
-            if not required_stems:
-                task_record = db.query(Task).filter(Task.task_id == sq.id).first()
-                count = task_record.stems if task_record else 4
-                required_stems = {"vocals", "no_vocals"} if count == 2 else ({"vocals", "bass", "drums", "guitar", "piano", "other"} if count == 6 else {"vocals", "bass", "drums", "other"})
-                
-            task_record = db.query(Task).filter(Task.task_id == sq.id).first()
-            song_name = task_record.song_name if task_record else "Canción Melodix"
-            
-            pending_syncs.append({
-                "task_id": sq.id,
-                "song_name": song_name,
-                "required_stems": list(required_stems),
-                "sync_type": sq.sync_type,
-                "target_id": sq.target_id
-            })
-            
-    return pending_syncs
-
-
-@router.post("/sync-requests/claim-upload/{task_id}")
-async def claim_upload_lock(
-    task_id: str,
-    device_id: str = Form(...),
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """
-    Intenta bloquear el derecho a subir los stems para evitar subidas duplicadas de múltiples miembros.
-    """
-    from datetime import datetime, timedelta
-    from app.models.sync import B2SyncQueue
-    
-    sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-    if not sync_entry:
-        raise HTTPException(status_code=404, detail="Sincronización no encontrada")
-        
-    if sync_entry.is_uploaded:
-        return {"success": False, "message": "Ya sincronizado"}
-        
-    now = datetime.utcnow()
-    if sync_entry.uploading_by and sync_entry.uploading_by != device_id:
-        if sync_entry.uploading_claimed_at and now - sync_entry.uploading_claimed_at <= timedelta(minutes=5):
-            raise HTTPException(status_code=409, detail="Sincronización bloqueada por otro miembro")
-            
-    sync_entry.uploading_by = device_id
-    sync_entry.uploading_claimed_at = now
-    db.commit()
-    return {"success": True, "message": "Lock adquirido"}
-
-
-@router.post("/sync-requests/upload/{task_id}/{stem_name}")
-async def upload_sync_stem(
-    task_id: str,
-    stem_name: str,
-    file: UploadFile = File(...),
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """
-    El dispositivo helper sube un stem individual a la API, la cual lo deposita en B2.
-    """
-    import json as _json
-    from app.models.sync import B2SyncQueue
-    from app.utils.b2_storage import upload_to_b2
-    
-    task_record = db.query(Task).filter(Task.task_id == task_id).first()
-    if not task_record:
-        raise HTTPException(status_code=404, detail="Pista no encontrada")
-        
-    user_group_ids = [ug.group_id for ug in current_user.groups if ug.estado == 'activo']
-    has_access = (task_record.user_id == current_user.id) or any(gid in user_group_ids for gid in task_record.group_ids)
-    if not has_access:
-        raise HTTPException(status_code=403, detail="No tienes acceso a esta pista")
-        
-    sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-    if not sync_entry:
-        raise HTTPException(status_code=404, detail="Sincronización no iniciada")
-        
-    temp_dir = os.path.join(os.getcwd(), "uploads", "sync_temp")
-    os.makedirs(temp_dir, exist_ok=True)
-    
-    orig_ext = "flac"
-    if file.filename:
-        parts = file.filename.split(".")
-        if len(parts) > 1:
-            orig_ext = parts[-1].lower()
-
-    ext = "json" if stem_name.endswith("_pitch") else orig_ext
-    temp_file_path = os.path.join(temp_dir, f"{task_id}_{stem_name}.{ext}")
-    
-    try:
-        with open(temp_file_path, "wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
-            
-        file_size = os.path.getsize(temp_file_path)
-        logger.info(f"[{task_id}] upload_sync_stem: stem_name={stem_name}, ext={ext}, size={file_size} bytes")
-
-        # Idempotencia: si este stem ya fue subido a B2, no volvemos a subirlo
-        # para evitar crear versiones duplicadas en B2.
-        current_urls = {}
-        if task_record.telegram_file_ids:
-            try:
-                current_urls = _json.loads(task_record.telegram_file_ids)
-            except Exception:
-                pass
-        
-        object_key = f"stems/{task_id}/{stem_name}.{ext}"
-        if stem_name in current_urls and current_urls[stem_name]:
-            logger.info(f"[{task_id}] Stem '{stem_name}' ya existe en B2 ({current_urls[stem_name]}). Omitiendo re-subida.")
-            public_url = current_urls[stem_name]
-        else:
-            public_url = upload_to_b2(temp_file_path, object_key)
-            current_urls[stem_name] = public_url
-            task_record.telegram_file_ids = _json.dumps(current_urls)
-            
-        sync_entry.file_url = _json.dumps(current_urls)
-        
-        # NO eliminamos de missing_stems — esa info la necesita el destinatario para saber qué descargar.
-        # En cambio, verificamos si ya se han subido todos los stems que necesita algún destinatario,
-        # comparando los stems presentes en file_url contra la unión de todos los stems pedidos.
-        missing = sync_entry.missing_stems
-        if isinstance(missing, str):
-            missing = _json.loads(missing)
-        missing = dict(missing) if missing else {}
-        
-        # Unión de todos los stems que algún destinatario solicitó
-        all_required = set()
-        for stems_list in missing.values():
-            if stems_list:
-                all_required.update(stems_list)
-        
-        # Stems ya subidos según file_url
-        uploaded_stems = set(current_urls.keys())
-        
-        all_uploaded = all_required.issubset(uploaded_stems)
-        if all_uploaded and all_required:
-            sync_entry.is_uploaded = True
-            sync_entry.uploading_by = None
-            sync_entry.uploading_claimed_at = None
-            logger.info(f"[{task_id}] Todos los stems requeridos se han subido ({uploaded_stems}). Marcando is_uploaded = True de forma proactiva.")
-            
-            # Reset de expiración en Tebi al completarse la sincronización P2P
-            from datetime import datetime, timedelta
-            retention_days = 7 if (bool(task_record.group_ids) or (task_record.user and task_record.user.is_premium)) else 3
-            task_record.cloud_expires_at = datetime.utcnow() + timedelta(days=retention_days)
-            logger.info(f"[{task_id}] Sincronización completa. cloud_expires_at reseteado a: {task_record.cloud_expires_at} ({retention_days}d)")
-        
-        db.commit()
-    finally:
-        if os.path.exists(temp_file_path):
-            os.remove(temp_file_path)
-            
-    return {"success": True, "url": public_url}
-
-
-@router.post("/sync-requests/complete-upload/{task_id}")
-async def complete_sync_upload(
-    task_id: str,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """
-    El helper notifica que completó la subida de todos los stems requeridos.
-    """
-    from app.models.sync import B2SyncQueue
-    
-    sync_entry = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-    if not sync_entry:
-        raise HTTPException(status_code=404, detail="Sincronización no encontrada")
-        
-    sync_entry.is_uploaded = True
-    sync_entry.uploading_by = None
-    sync_entry.uploading_claimed_at = None
-    
-    # Reset de expiración en Tebi al completarse la sincronización P2P por complete-upload
-    task_record = db.query(Task).filter(Task.task_id == task_id).first()
-    if task_record:
-        from datetime import datetime, timedelta
-        retention_days = 7 if (bool(task_record.group_ids) or (task_record.user and task_record.user.is_premium)) else 3
-        task_record.cloud_expires_at = datetime.utcnow() + timedelta(days=retention_days)
-        logger.info(f"[{task_id}] complete_sync_upload: cloud_expires_at reseteado a: {task_record.cloud_expires_at} ({retention_days}d)")
-        
-    db.commit()
-    return {"success": True}
-

app/utils/b2_storage.py

@@ -1,192 +0,0 @@
-import os
-
-# Set AWS checksum environment variables to fix Tebi boto3 issue
-os.environ["AWS_REQUEST_CHECKSUM_CALCULATION"] = "when_required"
-os.environ["AWS_RESPONSE_CHECKSUM_VALIDATION"] = "when_required"
-
-import boto3
-from botocore.config import Config
-import logging
-
-logger = logging.getLogger(__name__)
-
-B2_KEY_ID = os.getenv("B2_KEY_ID")
-B2_APPLICATION_KEY = os.getenv("B2_APPLICATION_KEY")
-B2_ENDPOINT = os.getenv("B2_ENDPOINT")  # e.g., https://s3.us-east-005.backblazeb2.com
-B2_BUCKET_NAME = os.getenv("B2_BUCKET_NAME")
-
-def get_b2_client():
-    if not all([B2_KEY_ID, B2_APPLICATION_KEY, B2_ENDPOINT]):
-        logger.warning("Faltan variables de entorno para almacenamiento S3. Usando Mock.")
-        return None
-
-    # Detectar región según proveedor
-    endpoint_lower = B2_ENDPOINT.lower()
-    if "tebi.io" in endpoint_lower:
-        # Tebi usa "auto" como región y requiere SigV4
-        region = "auto"
-    elif "backblazeb2.com" in endpoint_lower:
-        # Deducir región del endpoint B2 (ej. s3.us-east-005.backblazeb2.com → us-east-005)
-        region = "us-east-005"
-        try:
-            parts = B2_ENDPOINT.replace("https://", "").replace("http://", "").split(".")
-            if len(parts) >= 2 and "s3" in parts[0]:
-                region = parts[1]
-        except Exception as e:
-            logger.warning(f"No se pudo deducir la región de B2, usando {region}: {e}")
-    else:
-        # Fallback genérico para otros proveedores S3-compatibles
-        region = "us-east-1"
-
-    return boto3.client(
-        's3',
-        endpoint_url=B2_ENDPOINT,
-        aws_access_key_id=B2_KEY_ID,
-        aws_secret_access_key=B2_APPLICATION_KEY,
-        region_name=region,
-        config=Config(signature_version='s3v4', s3={'addressing_style': 'path'})
-    )
-
-def upload_to_b2(file_path: str, object_name: str) -> str:
-    """
-    Sube un archivo local a Backblaze B2 y retorna su URL pública.
-    Actúa como upsert: elimina versiones anteriores del mismo objeto antes de subir
-    para evitar acumulación de versiones duplicadas en B2.
-    """
-    client = get_b2_client()
-    if not client:
-        logger.warning(f"[Mock B2] Subiendo {file_path} como {object_name}")
-        return f"https://mock-b2.backblazeb2.com/{B2_BUCKET_NAME or 'mock'}/{object_name}"
-    
-    try:
-        # Eliminar versiones previas del mismo objeto antes de subir
-        try:
-            versions_resp = client.list_object_versions(Bucket=B2_BUCKET_NAME, Prefix=object_name)
-            old_versions = []
-            for version in versions_resp.get('Versions', []):
-                if version.get('Key') == object_name:
-                    old_versions.append({'Key': object_name, 'VersionId': version.get('VersionId')})
-            for marker in versions_resp.get('DeleteMarkers', []):
-                if marker.get('Key') == object_name:
-                    old_versions.append({'Key': object_name, 'VersionId': marker.get('VersionId')})
-            if old_versions:
-                client.delete_objects(
-                    Bucket=B2_BUCKET_NAME,
-                    Delete={'Objects': old_versions, 'Quiet': True}
-                )
-                logger.info(f"Eliminadas {len(old_versions)} versiones previas de B2 para: {object_name}")
-        except Exception as cleanup_err:
-            logger.warning(f"No se pudieron limpiar versiones previas de {object_name}: {cleanup_err}")
-
-        client.upload_file(file_path, B2_BUCKET_NAME, object_name)
-        endpoint_host = B2_ENDPOINT.replace("https://", "").replace("http://", "")
-        public_url = f"https://{B2_BUCKET_NAME}.{endpoint_host}/{object_name}"
-        logger.info(f"Archivo subido a B2: {object_name}. URL: {public_url}")
-        return public_url
-    except Exception as e:
-        logger.error(f"Error subiendo archivo {object_name} a B2: {e}")
-        raise e
-
-def delete_from_b2(object_name: str):
-    """
-    Elimina un archivo de Backblaze B2 de manera permanente (hard delete),
-    removiendo todas sus versiones y delete markers para evitar que queden en la papelera.
-    """
-    client = get_b2_client()
-    if not client:
-        logger.warning(f"[Mock B2] Eliminando {object_name}")
-        return
-    
-    try:
-        # Listar todas las versiones y delete markers para este objeto específico
-        versions = client.list_object_versions(Bucket=B2_BUCKET_NAME, Prefix=object_name)
-        
-        objects_to_delete = []
-        
-        # Recolectar versiones
-        for version in versions.get('Versions', []):
-            if version.get('Key') == object_name:
-                objects_to_delete.append({'Key': object_name, 'VersionId': version.get('VersionId')})
-                
-        # Recolectar delete markers
-        for marker in versions.get('DeleteMarkers', []):
-            if marker.get('Key') == object_name:
-                objects_to_delete.append({'Key': object_name, 'VersionId': marker.get('VersionId')})
-                
-        if objects_to_delete:
-            client.delete_objects(
-                Bucket=B2_BUCKET_NAME,
-                Delete={
-                    'Objects': objects_to_delete,
-                    'Quiet': True
-                }
-            )
-            logger.info(f"Archivo eliminado permanentemente de B2 (incluyendo {len(objects_to_delete)} versiones/delete markers): {object_name}")
-        else:
-            # Fallback en caso de que list_object_versions no devuelva nada o no coincida
-            client.delete_object(Bucket=B2_BUCKET_NAME, Key=object_name)
-            logger.info(f"Archivo eliminado de B2 (simple delete fallback): {object_name}")
-            
-    except Exception as e:
-        logger.error(f"Error eliminando permanentemente de B2 ({object_name}): {e}")
-
-def get_b2_download_url(object_name: str, expires_in: int = 3600) -> str:
-    """
-    Genera una URL firmada (presigned) para descargar un archivo de Backblaze B2 de manera segura.
-    Soporta buckets privados sin requerir que el usuario lo cambie a público.
-    """
-    client = get_b2_client()
-    if not client:
-        return f"https://mock-b2.backblazeb2.com/{B2_BUCKET_NAME or 'mock'}/{object_name}"
-    
-    try:
-        url = client.generate_presigned_url(
-            'get_object',
-            Params={
-                'Bucket': B2_BUCKET_NAME,
-                'Key': object_name
-            },
-            ExpiresIn=expires_in
-        )
-        logger.info(f"URL firmada generada para {object_name}: {url[:60]}...")
-        return url
-    except Exception as e:
-        logger.error(f"Error generando URL firmada para {object_name}: {e}")
-        endpoint_host = B2_ENDPOINT.replace("https://", "").replace("http://", "")
-        return f"https://{B2_BUCKET_NAME}.{endpoint_host}/{object_name}"
-
-def delete_folder_from_b2(prefix: str):
-    """
-    Elimina todos los objetos en Backblaze B2 que coincidan con el prefijo dado (ej. migrations/user_id/).
-    """
-    client = get_b2_client()
-    if not client:
-        logger.warning(f"[Mock B2] Eliminando carpeta con prefijo: {prefix}")
-        return
-    
-    try:
-        # Listar objetos con el prefijo
-        paginator = client.get_paginator('list_objects_v2')
-        pages = paginator.paginate(Bucket=B2_BUCKET_NAME, Prefix=prefix)
-        
-        delete_keys = []
-        for page in pages:
-            for obj in page.get('Contents', []):
-                delete_keys.append({'Key': obj['Key']})
-                
-        if delete_keys:
-            # Dividir en lotes de 1000 (límite de delete_objects en S3 API)
-            for i in range(0, len(delete_keys), 1000):
-                batch = delete_keys[i:i+1000]
-                client.delete_objects(
-                    Bucket=B2_BUCKET_NAME,
-                    Delete={
-                        'Objects': batch,
-                        'Quiet': True
-                    }
-                )
-            logger.info(f"Eliminados {len(delete_keys)} objetos con prefijo '{prefix}' de B2.")
-        else:
-            logger.info(f"No se encontraron objetos con el prefijo '{prefix}' para eliminar.")
-    except Exception as e:
-        logger.error(f"Error eliminando carpeta '{prefix}' de B2: {e}")

app/utils/hf_storage.py

@@ -0,0 +1,87 @@
+import os
+import logging
+from huggingface_hub import HfApi
+
+logger = logging.getLogger(__name__)
+
+HF_TOKEN = os.getenv("HF_TOKEN")
+HF_STORAGE_REPO = os.getenv("HF_STORAGE_REPO", "Isaac105/melodix-stems")
+
+def get_hf_api():
+    if not HF_TOKEN:
+        logger.warning("Falta variable de entorno HF_TOKEN. Operaciones de Hugging Face desactivadas.")
+        return None
+    return HfApi(token=HF_TOKEN)
+
+def upload_to_hf(file_path: str, object_name: str) -> str:
+    """
+    Sube un archivo local al dataset de Hugging Face y retorna su URL pública de descarga.
+    """
+    api = get_hf_api()
+    if not api:
+        logger.warning(f"[Mock HF] Subiendo {file_path} como {object_name}")
+        return f"https://mock-hf.huggingface.co/{HF_STORAGE_REPO}/{object_name}"
+
+    try:
+        api.upload_file(
+            path_or_fileobj=file_path,
+            path_in_repo=object_name,
+            repo_id=HF_STORAGE_REPO,
+            repo_type="dataset"
+        )
+        public_url = get_hf_download_url(object_name)
+        logger.info(f"Archivo subido a Hugging Face Dataset: {object_name}. URL: {public_url}")
+        return public_url
+    except Exception as e:
+        logger.error(f"Error subiendo archivo {object_name} a Hugging Face: {e}")
+        raise e
+
+def delete_from_hf(object_name: str):
+    """
+    Elimina un archivo del dataset de Hugging Face.
+    """
+    api = get_hf_api()
+    if not api:
+        logger.warning(f"[Mock HF] Eliminando {object_name}")
+        return
+
+    try:
+        api.delete_file(
+            path_in_repo=object_name,
+            repo_id=HF_STORAGE_REPO,
+            repo_type="dataset"
+        )
+        logger.info(f"Archivo eliminado de Hugging Face Dataset: {object_name}")
+    except Exception as e:
+        if "EntryNotFound" in str(e) or "404" in str(e):
+            logger.info(f"El archivo {object_name} no existía en Hugging Face. Ignorado.")
+        else:
+            logger.error(f"Error eliminando de Hugging Face ({object_name}): {e}")
+
+def delete_folder_from_hf(prefix: str):
+    """
+    Elimina una carpeta completa (ej. stems/task_id/) del dataset de Hugging Face.
+    """
+    api = get_hf_api()
+    if not api:
+        logger.warning(f"[Mock HF] Eliminando carpeta con prefijo: {prefix}")
+        return
+
+    try:
+        api.delete_folder(
+            folder_path=prefix,
+            repo_id=HF_STORAGE_REPO,
+            repo_type="dataset"
+        )
+        logger.info(f"Carpeta eliminada de Hugging Face Dataset: {prefix}")
+    except Exception as e:
+        if "EntryNotFound" in str(e) or "404" in str(e):
+            logger.info(f"La carpeta {prefix} no existía en Hugging Face. Ignorado.")
+        else:
+            logger.error(f"Error eliminando carpeta '{prefix}' de Hugging Face: {e}")
+
+def get_hf_download_url(object_name: str) -> str:
+    """
+    Devuelve la URL de descarga directa pública para un objeto en el dataset de Hugging Face.
+    """
+    return f"https://huggingface.co/datasets/{HF_STORAGE_REPO}/resolve/main/{object_name}"

cleanup.py

@@ -209,71 +209,32 @@ def show_stats():
 
 
 # ==========================================
-# EXPIRACIÓN DE COLA DE SINCRONIZACIÓN B2
+# EXPIRACIÓN DE ALMACENAMIENTO EN HUGGING FACE
 # ==========================================
 
 def clean_expired_sync_queue(max_age_days=7, dry_run=False):
     """
-    Elimina entradas de b2_sync_queue que tengan más de max_age_days días,
-    y purga sus correspondientes archivos de Backblaze B2.
+    Ejecuta la limpieza de archivos expirados en Hugging Face Dataset
+    y la purga mensual de usuarios normales si corresponde.
     """
-    logger.info(f"Limpiando cola de sincronización B2SyncQueue más antigua de {max_age_days} días...")
+    logger.info("Ejecutando limpieza de archivos expirados en Hugging Face Datasets...")
+    if dry_run:
+        logger.info("  [DRY-RUN] Ejecutaría la tarea de limpieza en Hugging Face.")
+        return 0
     
     current_dir = os.path.dirname(os.path.abspath(__file__))
     if current_dir not in sys.path:
         sys.path.insert(0, current_dir)
         
     try:
-        from app.database import SessionLocal
-        from app.models.sync import B2SyncQueue
-        from app.utils.b2_storage import delete_from_b2
-        import json
-    except ImportError as ie:
-        logger.error(f"Error de importación al limpiar cola de sync: {ie}")
-        return 0
-    
-    db = SessionLocal()
-    cutoff = datetime.utcnow() - timedelta(days=max_age_days)
-    cleaned = 0
-    
-    try:
-        expired_entries = db.query(B2SyncQueue).filter(B2SyncQueue.created_at < cutoff).all()
-        logger.info(f"Encontradas {len(expired_entries)} entradas sync vencidas.")
-        
-        for entry in expired_entries:
-            task_id = entry.id
-            if dry_run:
-                logger.info(f"  [DRY-RUN] Eliminaría entrada sync {task_id} creada el {entry.created_at}")
-            else:
-                logger.info(f"  Limpiando entrada sync {task_id} (creada {entry.created_at})...")
-                # Eliminar de B2
-                try:
-                    if entry.file_url:
-                        urls_dict = json.loads(entry.file_url)
-                        for stem_name, stem_url in urls_dict.items():
-                            if stem_name.endswith("_pitch"):
-                                key = f"stems/{task_id}/{stem_name}.json"
-                            else:
-                                ext = "m4a" if ".m4a" in stem_url.lower() else "flac"
-                                key = f"stems/{task_id}/{stem_name}.{ext}"
-                            delete_from_b2(key)
-                except Exception as b2_err:
-                    logger.error(f"  Error purgando archivos B2 para {task_id}: {b2_err}")
-                
-                # Eliminar de BD
-                db.delete(entry)
-                cleaned += 1
-                
-        if not dry_run and cleaned > 0:
-            db.commit()
-            
-        logger.info(f"Cola de sincronización B2 limpia: {cleaned} entradas eliminadas.")
+        from tasks import limpiar_stems_expirados_task
+        res = limpiar_stems_expirados_task()
+        cleaned = res.get("cleaned", 0)
+        logger.info(f"Limpieza en Hugging Face completada. Tareas limpiadas: {cleaned}")
         return cleaned
     except Exception as e:
-        logger.error(f"Error en clean_expired_sync_queue: {e}")
+        logger.error(f"Error ejecutando limpiar_stems_expirados_task desde cleanup.py: {e}")
         return 0
-    finally:
-        db.close()
 
 
 # ==========================================
@@ -306,7 +267,7 @@ def main():
     logger.info("")
     cleaned_logs = clean_logs(max_age_days=30, dry_run=args.dry_run)  # Logs: 30 días
     logger.info("")
-    cleaned_sync = clean_expired_sync_queue(max_age_days=args.days, dry_run=args.dry_run)  # B2SyncQueue: X días
+    cleaned_sync = clean_expired_sync_queue(max_age_days=args.days, dry_run=args.dry_run)  # HF/Expira
     
     logger.info("")
     logger.info("=" * 50)
@@ -315,7 +276,7 @@ def main():
     logger.info(f"Uploads limpiados:    {cleaned_uploads}")
     logger.info(f"Procesados limpiados: {cleaned_processed}")
     logger.info(f"Logs limpiados:       {cleaned_logs}")
-    logger.info(f"Cola sync limpia:     {cleaned_sync}")
+    logger.info(f"HF/Nube limpiados:    {cleaned_sync}")
     logger.info("=" * 50)
     
     if args.dry_run:

tasks.py

@@ -814,11 +814,11 @@ def procesar_cancion_completa(self, file_path, task_id, stems, trim_start="", tr
         except Exception as db_err:
             logger.error(f"[{task_id}] Error determinando premium/grupo: {db_err}")
 
-        logger.info(f"[{task_id}] Subiendo stems a Backblaze B2...")
-        self.update_state(state='PROGRESS', meta={'progress': 95, 'message': 'Subiendo stems a Backblaze B2...'})
+        logger.info(f"[{task_id}] Subiendo stems a Hugging Face Dataset...")
+        self.update_state(state='PROGRESS', meta={'progress': 95, 'message': 'Subiendo stems a Hugging Face Dataset...'})
         
-        b2_urls = {}
-        from app.utils.b2_storage import upload_to_b2
+        hf_urls = {}
+        from app.utils.hf_storage import upload_to_hf
         
         for stem_name, stem_path, ext in stems_to_upload:
             try:
@@ -826,70 +826,42 @@ def procesar_cancion_completa(self, file_path, task_id, stems, trim_start="", tr
                     raise TaskCancelledException("Tarea cancelada por el usuario.")
                 
                 object_name = f"stems/{task_id}/{stem_name}.{ext}"
-                public_url = upload_to_b2(stem_path, object_name)
-                b2_urls[stem_name] = public_url
+                public_url = upload_to_hf(stem_path, object_name)
+                hf_urls[stem_name] = public_url
                 
-                # Eliminar de disco local de inmediato tras subir a B2
+                # Eliminar de disco local de inmediato tras subir
                 if os.path.exists(stem_path):
                     os.remove(stem_path)
             except Exception as stem_err:
-                logger.warning(f"[{task_id}] Fallo al subir '{stem_name}' a B2: {stem_err}")
+                logger.warning(f"[{task_id}] Fallo al subir '{stem_name}' a Hugging Face: {stem_err}")
                 if isinstance(stem_err, TaskCancelledException):
                     raise stem_err
 
-        # Guardar URLs y crear cola de sincronización B2SyncQueue
-        if b2_urls:
+        # Guardar URLs y configurar expiración
+        if hf_urls:
             try:
                 from datetime import timedelta
                 db = SessionLocal()
                 task_record = db.query(TaskModel).filter(TaskModel.task_id == task_id).first()
                 if task_record:
-                    task_record.telegram_file_ids = json.dumps(b2_urls)
+                    task_record.telegram_file_ids = json.dumps(hf_urls)
                     if acordes:
                         task_record.chords = json.dumps(acordes)
                     task_record.status = "completed"
-                    # Calcular expiración en Tebi: 7 días premium, 3 días normal
-                    retention_days = 7 if is_premium else 3
-                    task_record.cloud_expires_at = datetime.utcnow() + timedelta(days=retention_days)
-                    logger.info(f"[{task_id}] cloud_expires_at = {task_record.cloud_expires_at} ({retention_days}d, premium={is_premium})")
-                
-                # Rellenar destinatarios pendientes
-                pending_recipients = []
-                sync_type = "device"
-                target_id = ""
-                
-                if group_id is not None:
-                    sync_type = "group"
-                    target_id = str(group_id)
-                    from app.models.lyrics import UserGroup as UserGroupModel
-                    from app.models.user import Device as DeviceModel
-                    active_members = db.query(UserGroupModel).filter(
-                        UserGroupModel.group_id == group_id,
-                        UserGroupModel.estado == "activo"
-                    ).all()
-                    member_ids = [m.user_id for m in active_members]
-                    devices = db.query(DeviceModel).filter(DeviceModel.user_id.in_(member_ids)).all()
-                    pending_recipients = [d.device_id for d in devices]
-                else:
-                    sync_type = "device"
-                    target_id = str(task_record.user_id) if task_record else ""
-                    pending_recipients = user_devices
-                
-                from app.models.sync import B2SyncQueue as B2SyncQueueModel
-                sync_entry = B2SyncQueueModel(
-                    id=task_id,
-                    file_key=f"stems/{task_id}",
-                    file_url=json.dumps(b2_urls),
-                    sync_type=sync_type,
-                    target_id=target_id,
-                    pending_recipients=pending_recipients
-                )
-                db.add(sync_entry)
+                    
+                    # Política de retención: Permanente si es Premium o pertenece a un grupo, 30 días si es normal
+                    is_group_task = bool(task_record.group_ids)
+                    if is_premium or is_group_task:
+                        task_record.cloud_expires_at = None
+                        logger.info(f"[{task_id}] cloud_expires_at = None (Permanente, premium={is_premium}, grupo={is_group_task})")
+                    else:
+                        task_record.cloud_expires_at = datetime.utcnow() + timedelta(days=30)
+                        logger.info(f"[{task_id}] cloud_expires_at = {task_record.cloud_expires_at} (30 días, normal)")
+                    
                 db.commit()
                 db.close()
-                logger.info(f"[{task_id}] Registrado en B2SyncQueue ({sync_type}) con {len(pending_recipients)} pendientes.")
             except Exception as db_err:
-                logger.error(f"[{task_id}] Error al registrar en DB B2SyncQueue: {db_err}")
+                logger.error(f"[{task_id}] Error al registrar en DB: {db_err}")
 
         # Limpiar la carpeta física local output_base de inmediato para liberar espacio en el disco del worker
         try:
@@ -1000,10 +972,9 @@ def generar_armonias_task(self, task_id: str):
     try:
         from app.database import SessionLocal
         from app.models.process import Task as TaskModel
-        from app.models.sync import B2SyncQueue as B2SyncQueueModel
         from app.utils.redis_client import get_redis_client
         from app.utils.harmony import generar_armonias
-        from app.utils.b2_storage import upload_to_b2, get_b2_download_url
+        from app.utils.hf_storage import upload_to_hf, get_hf_download_url
         import json
         import os
         import shutil
@@ -1037,9 +1008,9 @@ def generar_armonias_task(self, task_id: str):
             if os.path.exists(flac_vocal):
                 vocals_path = flac_vocal
 
-        # Si vocals no está localmente, descargarlo de B2
+        # Si vocals no está localmente, descargarlo de Hugging Face
         if not os.path.exists(vocals_path):
-            logger.info(f"[{task_id}] vocals no se encuentra localmente. Descargándolo de B2...")
+            logger.info(f"[{task_id}] vocals no se encuentra localmente. Descargándolo de Hugging Face...")
             try:
                 db = SessionLocal()
                 task_record = db.query(TaskModel).filter(TaskModel.task_id == task_id).first()
@@ -1058,9 +1029,9 @@ def generar_armonias_task(self, task_id: str):
                 ext = "m4a" if ".m4a" in vocals_url.lower() else "flac"
                 vocals_path = os.path.join(song_dir, f"vocals.{ext}")
 
-                # Generar URL firmada de B2
+                # Generar URL de descarga de Hugging Face
                 object_key = f"stems/{task_id}/vocals.{ext}"
-                download_url = get_b2_download_url(object_key)
+                download_url = get_hf_download_url(object_key)
 
                 # Descargar archivo
                 r = requests.get(download_url, stream=True)
@@ -1068,9 +1039,9 @@ def generar_armonias_task(self, task_id: str):
                 with open(vocals_path, 'wb') as f:
                     for chunk in r.iter_content(chunk_size=8192):
                         f.write(chunk)
-                logger.info(f"[{task_id}] vocals.{ext} descargado con éxito de B2.")
+                logger.info(f"[{task_id}] vocals.{ext} descargado con éxito de Hugging Face.")
             except Exception as dl_err:
-                logger.error(f"[{task_id}] Error descargando vocals de B2 para armonías: {dl_err}")
+                logger.error(f"[{task_id}] Error descargando vocals de Hugging Face para armonías: {dl_err}")
                 if redis_client:
                     redis_client.set(f"harmony_status:{task_id}", "error", ex=3600)
                 return False
@@ -1080,9 +1051,9 @@ def generar_armonias_task(self, task_id: str):
         if not res or not res.get("success"):
             raise Exception(res.get("error", "Error desconocido al armonizar."))
 
-        logger.info(f"[{task_id}] Armonías y trayectorias generadas localmente. Subiendo a B2...")
+        logger.info(f"[{task_id}] Armonías y trayectorias generadas localmente. Subiendo a Hugging Face...")
 
-        # 3. Subir a B2 los 6 stems de coro (m4a o flac) y los 7 pitch JSON
+        # 3. Subir a Hugging Face los 6 stems de coro y los 7 pitch JSON
         voces_corales = [
             "2da_voz", "2da_voz_baja", "3ra_voz", 
             "3ra_voz_baja", "oct_alta", "oct_baja"
@@ -1103,7 +1074,7 @@ def generar_armonias_task(self, task_id: str):
                 
             if os.path.exists(file_path):
                 object_key = f"stems/{task_id}/{v}.{ext}"
-                public_url = upload_to_b2(file_path, object_key)
+                public_url = upload_to_hf(file_path, object_key)
                 uploaded_files[v] = public_url
 
         # Subir jsons
@@ -1111,7 +1082,7 @@ def generar_armonias_task(self, task_id: str):
             file_path = os.path.join(song_dir, pj)
             if os.path.exists(file_path):
                 object_key = f"stems/{task_id}/{pj}"
-                public_url = upload_to_b2(file_path, object_key)
+                public_url = upload_to_hf(file_path, object_key)
                 pitch_key = pj.replace(".json", "")
                 uploaded_files[pitch_key] = public_url
 
@@ -1131,24 +1102,23 @@ def generar_armonias_task(self, task_id: str):
             current_urls.update(uploaded_files)
             task_record.telegram_file_ids = json.dumps(current_urls)
 
-            # Si cloud_expires_at no está seteado aún (tarea antigua o raro edge case),
-            # calcularlo ahora basado en el usuario
+            # Si cloud_expires_at no está seteado aún, calcularlo ahora basado en el usuario/grupo
             if task_record.cloud_expires_at is None:
                 try:
                     user = db.query(UserModel).filter(UserModel.id == task_record.user_id).first()
-                    retention_days = 7 if (user and user.is_premium) else 3
-                    task_record.cloud_expires_at = datetime.utcnow() + timedelta(days=retention_days)
-                    logger.info(f"[{task_id}] cloud_expires_at seteada por armonías: {task_record.cloud_expires_at}")
+                    is_premium = user.is_premium if user else False
+                    is_group_task = bool(task_record.group_id or task_record.groups_shared)
+                    if is_premium or is_group_task:
+                        task_record.cloud_expires_at = None
+                        logger.info(f"[{task_id}] cloud_expires_at = None (Premium/Grupo)")
+                    else:
+                        task_record.cloud_expires_at = datetime.utcnow() + timedelta(days=30)
+                        logger.info(f"[{task_id}] cloud_expires_at seteada a 30 días para usuario normal: {task_record.cloud_expires_at}")
                 except Exception as exp_err:
                     logger.warning(f"[{task_id}] No se pudo calcular cloud_expires_at en armonías: {exp_err}")
 
-            # Actualizar también B2SyncQueue
-            sync_entry = db.query(B2SyncQueueModel).filter(B2SyncQueueModel.id == task_id).first()
-            if sync_entry:
-                sync_entry.file_url = json.dumps(current_urls)
-
             db.commit()
-            logger.info(f"[{task_id}] Registro de base de datos actualizado con URLs corales.")
+            logger.info(f"[{task_id}] Registro de base de datos actualizado con URLs corales en Hugging Face.")
         db.close()
 
         # 5. Limpieza de archivos generados localmente
@@ -1192,28 +1162,26 @@ def generar_armonias_task(self, task_id: str):
 @celery_app.task(name="limpiar_stems_expirados_task")
 def limpiar_stems_expirados_task():
     """
-    Elimina de Tebi (B2) todos los stems y voces corales cuyo cloud_expires_at ya venció.
-    Limpia también el registro en DB (telegram_file_ids → NULL, B2SyncQueue eliminado).
-
-    Diseñada para ser invocada por un endpoint HTTP (POST /songs/cleanup-expired)
-    llamado por el cron-job externo cada hora. No requiere Celery Beat.
+    Elimina de Hugging Face Dataset todos los stems y voces corales cuyo cloud_expires_at ya venció.
+    También el 1 de cada mes purga los archivos de los usuarios con cuentas normales.
+    Limpia el registro en DB (telegram_file_ids -> NULL, cloud_expires_at -> NULL).
     """
     from app.database import SessionLocal
     from app.models.process import Task as TaskModel
-    from app.models.sync import B2SyncQueue as B2SyncQueueModel
-    from app.utils.b2_storage import delete_folder_from_b2
+    from app.utils.hf_storage import delete_folder_from_hf
     from datetime import datetime
 
-    logger.info("[Cleanup] Iniciando limpieza de stems expirados en Tebi...")
+    logger.info("[Cleanup] Iniciando limpieza de stems expirados en Hugging Face...")
     db = SessionLocal()
     cleaned = 0
     errors = 0
 
     try:
         now = datetime.utcnow()
+        # 1. Tareas expiradas de forma normal
         expired_tasks = db.query(TaskModel).filter(
             TaskModel.cloud_expires_at <= now,
-            TaskModel.telegram_file_ids != None  # noqa: E711
+            TaskModel.telegram_file_ids != None
         ).all()
 
         logger.info(f"[Cleanup] {len(expired_tasks)} tasks expiradas encontradas.")
@@ -1221,26 +1189,52 @@ def limpiar_stems_expirados_task():
         for task in expired_tasks:
             try:
                 task_id = task.task_id
-                logger.info(f"[Cleanup] Eliminando Tebi para task {task_id} (expiró: {task.cloud_expires_at})")
+                logger.info(f"[Cleanup] Eliminando carpeta en HF para task {task_id} (expiró: {task.cloud_expires_at})")
 
-                # 1. Eliminar carpeta completa de Tebi
-                delete_folder_from_b2(f"stems/{task_id}/")
+                # Eliminar de Hugging Face
+                delete_folder_from_hf(f"stems/{task_id}")
 
-                # 2. Limpiar campos en Task
+                # Limpiar campos en DB
                 task.telegram_file_ids = None
                 task.cloud_expires_at = None
 
-                # 3. Eliminar entrada de B2SyncQueue
-                sync_entry = db.query(B2SyncQueueModel).filter(B2SyncQueueModel.id == task_id).first()
-                if sync_entry:
-                    db.delete(sync_entry)
-
                 cleaned += 1
-                logger.info(f"[Cleanup] Task {task_id} limpiada con éxito.")
+                logger.info(f"[Cleanup] Task {task_id} limpia con éxito.")
             except Exception as e:
                 errors += 1
                 logger.error(f"[Cleanup] Error limpiando task {task.task_id}: {e}")
 
+        # 2. Si es 1 de mes, purga de usuarios con cuentas normales
+        if datetime.utcnow().day == 1:
+            logger.info("[Cleanup] Es el 1 de mes. Iniciando purga de archivos de usuarios normales...")
+            from app.models.user import User as UserModel
+            try:
+                normal_tasks = db.query(TaskModel).join(
+                    UserModel, TaskModel.user_id == UserModel.id
+                ).filter(
+                    UserModel.is_premium == False,
+                    TaskModel.group_id == None,
+                    TaskModel.telegram_file_ids != None
+                ).all()
+
+                # Filtrar que no estén en grupos compartidos
+                normal_tasks = [t for t in normal_tasks if not t.groups_shared]
+
+                logger.info(f"[Cleanup] Encontradas {len(normal_tasks)} tareas de usuarios normales para purga mensual.")
+                for task in normal_tasks:
+                    try:
+                        task_id = task.task_id
+                        logger.info(f"[Cleanup-Mensual] Eliminando stems en HF para tarea normal {task_id}")
+                        delete_folder_from_hf(f"stems/{task_id}")
+                        task.telegram_file_ids = None
+                        task.cloud_expires_at = None
+                        cleaned += 1
+                    except Exception as e:
+                        errors += 1
+                        logger.error(f"[Cleanup-Mensual] Error purgando tarea {task.task_id}: {e}")
+            except Exception as monthly_err:
+                logger.error(f"[Cleanup] Error en la purga mensual de usuarios normales: {monthly_err}")
+
         db.commit()
         logger.info(f"[Cleanup] Finalizado. Limpiadas: {cleaned}, Errores: {errors}")
         return {"cleaned": cleaned, "errors": errors}

tests/test_b2_sync_flow.py

@@ -1,310 +0,0 @@
-import sys
-import os
-import unittest
-import json
-from unittest.mock import patch, MagicMock
-
-# Force environment variables for test
-os.environ["SUPABASE_JWT_SECRET"] = "supersecretkey_change_me_in_production"
-os.environ["SECRET_KEY"] = "supersecretkey_change_me_in_production"
-os.environ["REDIS_URL"] = "rpc://"
-
-# Mock heavy modules
-sys.modules['librosa'] = MagicMock()
-sys.modules['torch'] = MagicMock()
-sys.modules['yt_dlp'] = MagicMock()
-sys.modules['pretty_midi'] = MagicMock()
-sys.modules['numpy'] = MagicMock()
-sys.modules['demucs'] = MagicMock()
-sys.modules['demucs.separate'] = MagicMock()
-
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-
-from fastapi.testclient import TestClient
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-
-from app.database import Base, get_db
-from app.models.user import User, Device
-from app.models.lyrics import Group, UserGroup
-from app.models.process import Task
-from app.models.sync import B2SyncQueue
-from main import app
-
-# Configuración de base de datos de test
-TEST_DATABASE_URL = "sqlite:///./test_b2_sync.db"
-engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
-TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-def override_get_db():
-    db = TestingSessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
-app.dependency_overrides[get_db] = override_get_db
-
-class TestB2SyncFlow(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        Base.metadata.create_all(bind=engine)
-        cls.client = TestClient(app)
-
-    @classmethod
-    def tearDownClass(cls):
-        Base.metadata.drop_all(bind=engine)
-        if os.path.exists("./test_b2_sync.db"):
-            try:
-                os.remove("./test_b2_sync.db")
-            except Exception:
-                pass
-
-    def setUp(self):
-        db = TestingSessionLocal()
-        db.query(Task).delete()
-        db.query(Device).delete()
-        db.query(UserGroup).delete()
-        db.query(Group).delete()
-        db.query(User).delete()
-        db.query(B2SyncQueue).delete()
-        db.commit()
-        db.close()
-
-    @patch("app.utils.b2_storage.get_b2_download_url")
-    @patch("urllib.request.urlopen")
-    @patch("tasks.eliminar_archivos_locales_task.delay")
-    @patch("app.utils.b2_storage.delete_from_b2")
-    def test_standard_user_local_cleanup(self, mock_delete_from_b2, mock_eliminar_locales_delay, mock_urlopen, mock_get_b2_download_url):
-        mock_get_b2_download_url.side_effect = lambda obj_key: f"https://my-b2-bucket.com/{obj_key}"
-        # 1. Registrar usuario estándar
-        reg_payload = {"username": "standard_user", "email": "std@example.com", "password": "pass123"}
-        self.client.post("/auth/register", json=reg_payload)
-        
-        token_res = self.client.post("/auth/token", data={
-            "username": "std@example.com", "password": "pass123",
-            "device_id": "std_device_1", "device_name": "Standard Device 1"
-        })
-        token = token_res.json()["access_token"]
-        headers = {"Authorization": f"Bearer {token}"}
-
-        # Crear una tarea con rutas de stems B2 (Caso Estándar B2)
-        db = TestingSessionLocal()
-        user = db.query(User).filter(User.username == "standard_user").first()
-        task_id = "test-task-std-123"
-        task = Task(
-            task_id=task_id,
-            user_id=user.id,
-            song_name="Standard Song",
-            status="completed",
-            telegram_file_ids=json.dumps({
-                "vocals": "https://my-b2-bucket.com/stems/test-task-std-123/vocals.flac",
-                "drums": "https://my-b2-bucket.com/stems/test-task-std-123/drums.flac"
-            })
-        )
-        sync_queue = B2SyncQueue(
-            id=task_id,
-            file_key=f"stems/{task_id}",
-            file_url=json.dumps({
-                "vocals": "https://my-b2-bucket.com/stems/test-task-std-123/vocals.flac",
-                "drums": "https://my-b2-bucket.com/stems/test-task-std-123/drums.flac"
-            }),
-            sync_type="device",
-            target_id=str(user.id),
-            pending_recipients=["std_device_1"]
-        )
-        db.add(task)
-        db.add(sync_queue)
-        db.commit()
-        db.close()
-
-        # 2. Consultar get_stems y verificar que retorna las URLs directas de B2 (no usa el Gist tunnel URL)
-        stems_res = self.client.get(f"/songs/stems/{task_id}", headers=headers)
-        self.assertEqual(stems_res.status_code, 200)
-        stems_data = stems_res.json()
-        self.assertEqual(stems_data["stems"]["vocals"], "https://my-b2-bucket.com/stems/test-task-std-123/vocals.flac")
-        self.assertEqual(stems_data["stems"]["drums"], "https://my-b2-bucket.com/stems/test-task-std-123/drums.flac")
-        mock_urlopen.assert_not_called()
-
-        # 3. Confirmar descarga y verificar que se eliminan los stems de B2 de inmediato y se gatilla Celery local deletion
-        confirm_res = self.client.post(f"/songs/confirm-download/{task_id}?device_id=std_device_1", headers=headers)
-        self.assertEqual(confirm_res.status_code, 200)
-        
-        self.assertEqual(mock_delete_from_b2.call_count, 2)
-        mock_delete_from_b2.assert_any_call("stems/test-task-std-123/vocals.flac")
-        mock_delete_from_b2.assert_any_call("stems/test-task-std-123/drums.flac")
-        
-        mock_eliminar_locales_delay.assert_called_once_with(task_id)
-
-        # Verificar que downloaded_at está seteado
-        db = TestingSessionLocal()
-        task_after = db.query(Task).filter(Task.task_id == task_id).first()
-        self.assertIsNotNone(task_after.downloaded_at)
-        
-        # Verificar que la cola de sincronización B2SyncQueue se eliminó
-        self.assertIsNone(db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first())
-        db.close()
-
-    @patch("app.utils.b2_storage.delete_from_b2")
-    def test_premium_user_b2_sync_flow(self, mock_delete_from_b2):
-        # 1. Registrar usuario premium
-        reg_payload = {"username": "premium_user", "email": "prem@example.com", "password": "pass123"}
-        self.client.post("/auth/register", json=reg_payload)
-        
-        # Iniciar sesión en un dispositivo para registrar dev_A
-        t1_res = self.client.post("/auth/token", data={
-            "username": "prem@example.com", "password": "pass123",
-            "device_id": "dev_A", "device_name": "Device A"
-        })
-        token_A = t1_res.json()["access_token"]
-        headers_A = {"Authorization": f"Bearer {token_A}"}
-
-        # Seteamos premium a True y obtenemos ID
-        db = TestingSessionLocal()
-        user = db.query(User).filter(User.username == "premium_user").first()
-        user.is_premium = True
-        db.commit()
-        user_id = user.id
-
-        dev_b = Device(device_id="dev_B", name="Device B", user_id=user_id)
-        db.add(dev_b)
-        db.commit()
-        db.close()
-
-        task_id = "test-task-prem-456"
-        
-        # 2. Crear cola de sincronización B2SyncQueue simulada para premium (dos dispositivos pendientes)
-        db = TestingSessionLocal()
-        sync_queue = B2SyncQueue(
-            id=task_id,
-            file_key=f"stems/{task_id}",
-            file_url=json.dumps({
-                "vocals": "https://mock-b2.com/vocals.flac",
-                "drums": "https://mock-b2.com/drums.flac"
-            }),
-            sync_type="device",
-            target_id=str(user_id),
-            pending_recipients=["dev_A", "dev_B"]
-        )
-        task = Task(
-            task_id=task_id,
-            user_id=user_id,
-            song_name="Premium Song",
-            status="completed",
-            telegram_file_ids=json.dumps({
-                "vocals": "https://mock-b2.com/vocals.flac",
-                "drums": "https://mock-b2.com/drums.flac"
-            })
-        )
-        db.add(sync_queue)
-        db.add(task)
-        db.commit()
-        db.close()
-
-        # 3. Confirmar descarga de dev_A
-        confirm_A = self.client.post(f"/songs/confirm-download/{task_id}?device_id=dev_A", headers=headers_A)
-        self.assertEqual(confirm_A.status_code, 200)
-        self.assertEqual(confirm_A.json()["pending_recipients"], ["dev_B"])
-        mock_delete_from_b2.assert_not_called()
-
-        # Verificar que el registro aún existe
-        db = TestingSessionLocal()
-        self.assertIsNotNone(db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first())
-        db.close()
-
-        # 4. Confirmar descarga de dev_B
-        confirm_B = self.client.post(f"/songs/confirm-download/{task_id}?device_id=dev_B", headers=headers_A)
-        self.assertEqual(confirm_B.status_code, 200)
-        self.assertEqual(confirm_B.json()["pending_recipients"], [])
-        
-        # Verificar que se llamó delete_from_b2 para cada stem
-        self.assertEqual(mock_delete_from_b2.call_count, 2)
-        mock_delete_from_b2.assert_any_call("stems/test-task-prem-456/vocals.flac")
-        mock_delete_from_b2.assert_any_call("stems/test-task-prem-456/drums.flac")
-
-        # Verificar que el registro en B2SyncQueue se eliminó
-        db = TestingSessionLocal()
-        self.assertIsNone(db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first())
-        db.close()
-
-    @patch("app.utils.b2_storage.delete_from_b2")
-    def test_group_b2_sync_flow(self, mock_delete_from_b2):
-        # 1. Registrar dos usuarios del grupo
-        self.client.post("/auth/register", json={"username": "m1", "email": "m1@example.com", "password": "pass"})
-        self.client.post("/auth/register", json={"username": "m2", "email": "m2@example.com", "password": "pass"})
-
-        token1 = self.client.post("/auth/token", data={"username": "m1@example.com", "password": "pass", "device_id": "dev_1", "device_name": "Dev 1"}).json()["access_token"]
-        token2 = self.client.post("/auth/token", data={"username": "m2@example.com", "password": "pass", "device_id": "dev_2", "device_name": "Dev 2"}).json()["access_token"]
-
-        headers1 = {"Authorization": f"Bearer {token1}"}
-        headers2 = {"Authorization": f"Bearer {token2}"}
-
-        # Obtener IDs de usuarios
-        db = TestingSessionLocal()
-        u1 = db.query(User).filter(User.username == "m1").first()
-        u2 = db.query(User).filter(User.username == "m2").first()
-        u1_id = u1.id
-        u2_id = u2.id
-        
-        # Crear grupo y asociar usuarios
-        group = Group(name="Coral Group")
-        db.add(group)
-        db.commit()
-        group_id = group.id
-
-        ug1 = UserGroup(user_id=u1_id, group_id=group_id, estado="activo", role="creador")
-        ug2 = UserGroup(user_id=u2_id, group_id=group_id, estado="activo", role="miembro")
-        db.add(ug1)
-        db.add(ug2)
-        db.commit()
-        db.close()
-
-        task_id = "test-task-group-789"
-        
-        # Crear cola de sincronización B2SyncQueue para grupo
-        db = TestingSessionLocal()
-        sync_queue = B2SyncQueue(
-            id=task_id,
-            file_key=f"stems/{task_id}",
-            file_url=json.dumps({
-                "vocals": "https://mock-b2.com/vocals.flac"
-            }),
-            sync_type="group",
-            target_id=str(group_id),
-            pending_recipients=[str(u1_id), str(u2_id)]
-        )
-        task = Task(
-            task_id=task_id,
-            user_id=u1_id,
-            group_id=group_id,
-            song_name="Group Song",
-            status="completed",
-            telegram_file_ids=json.dumps({
-                "vocals": "https://mock-b2.com/vocals.flac"
-            })
-        )
-        db.add(sync_queue)
-        db.add(task)
-        db.commit()
-        db.close()
-
-        # 2. Confirmar descarga de Miembro 1 (creador)
-        confirm_1 = self.client.post(f"/songs/confirm-download/{task_id}", headers=headers1)
-        self.assertEqual(confirm_1.status_code, 200)
-        self.assertEqual(confirm_1.json()["pending_recipients"], [str(u2_id)])
-        mock_delete_from_b2.assert_not_called()
-
-        # 3. Confirmar descarga de Miembro 2 (miembro)
-        confirm_2 = self.client.post(f"/songs/confirm-download/{task_id}", headers=headers2)
-        self.assertEqual(confirm_2.status_code, 200)
-        self.assertEqual(confirm_2.json()["pending_recipients"], [])
-        
-        mock_delete_from_b2.assert_called_once_with("stems/test-task-group-789/vocals.flac")
-
-        # Verificar eliminación de cola
-        db = TestingSessionLocal()
-        self.assertIsNone(db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first())
-        db.close()
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_p2p_sync_flow.py

@@ -1,222 +0,0 @@
-import sys
-import os
-import unittest
-import json
-from unittest.mock import patch, MagicMock
-
-# Force environment variables for test
-os.environ["SUPABASE_JWT_SECRET"] = "supersecretkey_change_me_in_production"
-os.environ["SECRET_KEY"] = "supersecretkey_change_me_in_production"
-os.environ["REDIS_URL"] = "rpc://"
-
-# Mock heavy modules
-sys.modules['librosa'] = MagicMock()
-sys.modules['torch'] = MagicMock()
-sys.modules['yt_dlp'] = MagicMock()
-sys.modules['pretty_midi'] = MagicMock()
-sys.modules['numpy'] = MagicMock()
-sys.modules['demucs'] = MagicMock()
-sys.modules['demucs.separate'] = MagicMock()
-
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-
-from fastapi.testclient import TestClient
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-
-from app.database import Base, get_db
-from app.models.user import User, Device
-from app.models.lyrics import Group, UserGroup
-from app.models.process import Task
-from app.models.sync import B2SyncQueue
-from main import app
-
-# Configuración de base de datos de test
-TEST_DATABASE_URL = "sqlite:///./test_p2p_sync.db"
-engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
-TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-def override_get_db():
-    db = TestingSessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
-app.dependency_overrides[get_db] = override_get_db
-
-class TestP2PSyncFlow(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        Base.metadata.create_all(bind=engine)
-        cls.client = TestClient(app)
-
-    @classmethod
-    def tearDownClass(cls):
-        Base.metadata.drop_all(bind=engine)
-        if os.path.exists("./test_p2p_sync.db"):
-            try:
-                os.remove("./test_p2p_sync.db")
-            except Exception:
-                pass
-
-    def setUp(self):
-        db = TestingSessionLocal()
-        db.query(Task).delete()
-        db.query(Device).delete()
-        db.query(UserGroup).delete()
-        db.query(Group).delete()
-        db.query(User).delete()
-        db.query(B2SyncQueue).delete()
-        db.commit()
-        db.close()
-
-    @patch("app.utils.b2_storage.upload_to_b2")
-    @patch("app.utils.b2_storage.delete_from_b2")
-    def test_p2p_group_sync_workflow(self, mock_delete_from_b2, mock_upload_to_b2):
-        mock_upload_to_b2.return_value = "https://mock-b2.com/stems/test-task-123/vocals.flac"
-        
-        # 1. Registrar dos usuarios (m1: miembro existente, m2: nuevo miembro)
-        self.client.post("/auth/register", json={"username": "m1", "email": "m1@example.com", "password": "pass"})
-        self.client.post("/auth/register", json={"username": "m2", "email": "m2@example.com", "password": "pass"})
-
-        token1 = self.client.post("/auth/token", data={"username": "m1@example.com", "password": "pass", "device_id": "dev_1", "device_name": "Dev 1"}).json()["access_token"]
-        token2 = self.client.post("/auth/token", data={"username": "m2@example.com", "password": "pass", "device_id": "dev_2", "device_name": "Dev 2"}).json()["access_token"]
-
-        headers1 = {"Authorization": f"Bearer {token1}"}
-        headers2 = {"Authorization": f"Bearer {token2}"}
-
-        db = TestingSessionLocal()
-        u1 = db.query(User).filter(User.username == "m1").first()
-        u2 = db.query(User).filter(User.username == "m2").first()
-        u1_id = u1.id
-        u2_id = u2.id
-        
-        # Crear grupo y asociar
-        group = Group(name="Rock Band")
-        db.add(group)
-        db.commit()
-        group_id = group.id
-
-        ug1 = UserGroup(user_id=u1_id, group_id=group_id, estado="activo", role="creador")
-        ug2 = UserGroup(user_id=u2_id, group_id=group_id, estado="activo", role="miembro")
-        db.add(ug1)
-        db.add(ug2)
-        db.commit()
-
-        # Crear una tarea ya completada que pertenece al grupo
-        task_id = "test-group-task-777"
-        task = Task(
-            task_id=task_id,
-            user_id=u1_id,
-            group_id=group_id,
-            song_name="Awesome Guitar Song",
-            status="completed",
-            stems=4,
-            telegram_file_ids=json.dumps({
-                "vocals": "https://mock-b2.com/vocals.flac",
-                "drums": "https://mock-b2.com/drums.flac"
-            })
-        )
-        db.add(task)
-        db.commit()
-        db.close()
-
-        # 2. El nuevo miembro (m2) reporta que le faltan stems de esta canción
-        report_res = self.client.post(
-            "/songs/sync-requests/report-missing",
-            headers=headers2,
-            json={
-                "device_id": "dev_2",
-                "missing_items": [
-                    {
-                        "task_id": task_id,
-                        "missing_stems": ["vocals", "drums"]
-                    }
-                ]
-            }
-        )
-        self.assertEqual(report_res.status_code, 200)
-        self.assertIn(task_id, report_res.json()["reported_tasks"])
-
-        # Verificar que el registro B2SyncQueue se creó con is_uploaded=False
-        db = TestingSessionLocal()
-        sq = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-        self.assertIsNotNone(sq)
-        self.assertFalse(sq.is_uploaded)
-        self.assertEqual(sq.pending_recipients, ["dev_2"])
-        db.close()
-
-        # 3. El miembro existente (m1) consulta las tareas pendientes de sincronización
-        pending_res = self.client.get("/songs/sync-requests/pending", headers=headers1)
-        self.assertEqual(pending_res.status_code, 200)
-        pending_list = pending_res.json()
-        self.assertEqual(len(pending_list), 1)
-        self.assertEqual(pending_list[0]["task_id"], task_id)
-        self.assertEqual(set(pending_list[0]["required_stems"]), {"vocals", "drums"})
-
-        # 4. El miembro 1 adquiere el bloqueo de subida (claim lock)
-        claim_res = self.client.post(
-            f"/songs/sync-requests/claim-upload/{task_id}",
-            headers=headers1,
-            data={"device_id": "dev_1"}
-        )
-        self.assertEqual(claim_res.status_code, 200)
-        self.assertTrue(claim_res.json()["success"])
-
-        # 5. El miembro 1 sube los stems requeridos
-        # Simular envío de archivo vocals
-        vocals_file = ("vocals.flac", b"dummy_audio_bytes")
-        upload_vocals_res = self.client.post(
-            f"/songs/sync-requests/upload/{task_id}/vocals",
-            headers=headers1,
-            files={"file": vocals_file}
-        )
-        self.assertEqual(upload_vocals_res.status_code, 200)
-        self.assertTrue(upload_vocals_res.json()["success"])
-
-        # Simular envío de archivo drums
-        drums_file = ("drums.flac", b"dummy_audio_bytes")
-        upload_drums_res = self.client.post(
-            f"/songs/sync-requests/upload/{task_id}/drums",
-            headers=headers1,
-            files={"file": drums_file}
-        )
-        self.assertEqual(upload_drums_res.status_code, 200)
-        self.assertTrue(upload_drums_res.json()["success"])
-
-        # Completar la subida
-        complete_res = self.client.post(
-            f"/songs/sync-requests/complete-upload/{task_id}",
-            headers=headers1
-        )
-        self.assertEqual(complete_res.status_code, 200)
-        self.assertTrue(complete_res.json()["success"])
-
-        # 6. El nuevo miembro (m2) consulta su historial de canciones y ve el estado correcto
-        history_res = self.client.get("/songs/history", headers=headers2)
-        self.assertEqual(history_res.status_code, 200)
-        tasks_list = history_res.json()["tasks"]
-        self.assertEqual(len(tasks_list), 1)
-        self.assertEqual(tasks_list[0]["sync_status"], "pending_my_download")
-        self.assertEqual(set(tasks_list[0]["missing_stems_for_me"]), {"vocals", "drums"})
-
-        # 7. El nuevo miembro descarga los archivos y confirma descarga
-        confirm_res = self.client.post(
-            f"/songs/confirm-download/{task_id}",
-            headers=headers2
-        )
-        self.assertEqual(confirm_res.status_code, 200)
-
-        # 8. Verificar que la sincronización se completó (B2SyncQueue se eliminó y delete_from_b2 se gatilló)
-        db = TestingSessionLocal()
-        sq_after = db.query(B2SyncQueue).filter(B2SyncQueue.id == task_id).first()
-        self.assertIsNone(sq_after)
-        db.close()
-
-        self.assertEqual(mock_delete_from_b2.call_count, 2)
-        mock_delete_from_b2.assert_any_call("stems/test-group-task-777/vocals.flac")
-        mock_delete_from_b2.assert_any_call("stems/test-group-task-777/drums.flac")
-
-if __name__ == "__main__":
-    unittest.main()