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()