web-app-python/app/services/firebase.py

import logging
from typing import List, Optional
import firebase_admin
from firebase_admin import credentials, messaging

from app.core.config import settings

logger = logging.getLogger(__name__)


class FirebaseService:
    """Servizio per gestire le notifiche push tramite Firebase Cloud Messaging"""

    _instance = None
    _initialized = False

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(FirebaseService, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        if not self._initialized:
            self._initialize_firebase()
            self.__class__._initialized = True

    def _initialize_firebase(self):
        """Inizializza Firebase Admin SDK"""
        try:
            cred = credentials.Certificate(settings.FIREBASE_CREDENTIALS_PATH)
            firebase_admin.initialize_app(cred)
            logger.info("Firebase Admin SDK inizializzato con successo")
        except Exception as e:
            logger.error(f"Errore nell'inizializzazione di Firebase: {e}")
            raise

    def send_notification(
        self,
        token: str,
        title: str,
        body: str,
        data: Optional[dict] = None,
        priority: str = "high"
    ) -> bool:
        """
        Invia una notifica push a un singolo dispositivo

        Args:
            token: FCM token del dispositivo
            title: Titolo della notifica
            body: Corpo della notifica
            data: Dati aggiuntivi da inviare
            priority: Priorità della notifica (high/normal)

        Returns:
            True se inviata con successo, False altrimenti
        """
        try:
            message = messaging.Message(
                notification=messaging.Notification(
                    title=title,
                    body=body,
                ),
                data=data or {},
                token=token,
                android=messaging.AndroidConfig(
                    priority=priority,
                    notification=messaging.AndroidNotification(
                        sound="default",
                        priority="max" if priority == "high" else "default",
                    ),
                ),
                apns=messaging.APNSConfig(
                    payload=messaging.APNSPayload(
                        aps=messaging.Aps(
                            sound="default",
                            badge=1,
                        ),
                    ),
                ),
            )

            response = messaging.send(message)
            logger.info(f"Notifica inviata con successo: {response}")
            return True

        except messaging.UnregisteredError:
            logger.warning(f"Token non registrato o scaduto: {token}")
            return False
        except Exception as e:
            logger.error(f"Errore nell'invio della notifica: {e}")
            return False

    def send_multicast(
        self,
        tokens: List[str],
        title: str,
        body: str,
        data: Optional[dict] = None,
        priority: str = "high"
    ) -> dict:
        """
        Invia una notifica push a più dispositivi

        Args:
            tokens: Lista di FCM tokens
            title: Titolo della notifica
            body: Corpo della notifica
            data: Dati aggiuntivi da inviare
            priority: Priorità della notifica

        Returns:
            Dizionario con statistiche di invio
        """
        if not tokens:
            return {"success": 0, "failure": 0}

        try:
            # Crea messaggio base
            base_message = messaging.Message(
                notification=messaging.Notification(
                    title=title,
                    body=body,
                ),
                data=data or {},
                android=messaging.AndroidConfig(
                    priority=priority,
                    notification=messaging.AndroidNotification(
                        sound="default",
                        priority="max" if priority == "high" else "default",
                    ),
                ),
                apns=messaging.APNSConfig(
                    payload=messaging.APNSPayload(
                        aps=messaging.Aps(
                            sound="default",
                            badge=1,
                        ),
                    ),
                ),
            )

            # Crea un multicast message
            multicast_message = messaging.MulticastMessage(
                notification=base_message.notification,
                data=base_message.data,
                android=base_message.android,
                apns=base_message.apns,
                tokens=tokens,
            )

            # Usa send_each_for_multicast invece di send_multicast
            response = messaging.send_each_for_multicast(multicast_message)

            success_count = response.success_count
            failure_count = response.failure_count

            logger.info(
                f"Notifiche inviate: {success_count} successi, "
                f"{failure_count} fallimenti"
            )

            return {
                "success": success_count,
                "failure": failure_count,
            }

        except Exception as e:
            logger.error(f"Errore nell'invio multicast: {e}")
            import traceback
            traceback.print_exc()
            return {"success": 0, "failure": len(tokens)}


# Singleton instance
firebase_service = FirebaseService()