api/app/push_service.py

from pywebpush import webpush, WebPushException
import json
from typing import List, Dict, Any
from urllib.parse import urlparse
from bson import ObjectId
import pymongo
import app.config as config

# Database setup
client = pymongo.MongoClient(config.MONGODB_URL, username=config.MONGODB_USERNAME, password=config.MONGODB_PASSWORD)
db = client[config.MONGODB_DATABASE]

users_collection = db["users"]

def get_audience(endpoint: str) -> str:
    parsed = urlparse(endpoint)
    return f"{parsed.scheme}://{parsed.netloc}"

class PushService:
    def __init__(self):
        self.vapid_private_key = config.VAPID_PRIVATE_KEY
        self.vapid_claims = config.VAPID_CLAIMS

    async def send_notification(self, user_id: ObjectId, payload: Dict[str, Any]) -> bool:
        """
        Envoie une notification push à un abonné spécifique.
        
        Args:
            user_id: L'ID de l'utilisateur à qui envoyer la notification
            payload: Le contenu de la notification
            
        Returns:
            bool: True si l'envoi a réussi, False sinon
        """
        try:
            # Récupérer les abonnements push de l'utilisateur
            user = users_collection.find_one({"_id": user_id}, {"push_subscriptions": 1})
            if not user:
                return False
            
            subscriptions = user.get("push_subscriptions", [])
            if not subscriptions:
                return False
            
            # Pour chaque abonnement, envoyer la notification
            for subscription in subscriptions:
                await self.send_notification_to_subscription(subscription, payload, user_id)
        except json.JSONDecodeError as e:
            return False
        except Exception as e:
            return False

    async def send_notification_to_subscription(self, subscription: str, payload: Dict[str, Any], user_id: ObjectId) -> bool:
        try:
            subscription_dict = json.loads(subscription)

            if subscription_dict and "endpoint" in subscription_dict:
                vapid_claims = self.vapid_claims
                vapid_claims["aud"] = get_audience(subscription_dict["endpoint"])

                webpush(
                    subscription_info=subscription_dict,
                    data=json.dumps(payload),
                    vapid_private_key=self.vapid_private_key,
                    vapid_claims=vapid_claims,
                    ttl=2419200 # Temps durant lequel la notification est conservée sur le serveur si l'appareil est hors ligne, ici le max (4 semaines)
                )

                return True
            else:
                await self.delete_subscription(subscription, user_id)
                return False
        except WebPushException as e:
            if int(e.response.status_code) in [404, 410]:
                # Suppression de l'abonnement de la base de données
                deleted = await self.delete_subscription(subscription, user_id)
                if not deleted:
                    return False
                
                return True
            return False
        except Exception as e:
            return False

    async def send_notification_to_all(self, subscriptions: List[Dict[str, Any]], payload: Dict[str, Any]) -> Dict[str, int]:
        """
        Envoie une notification à plusieurs abonnés.
        
        Args:
            subscriptions: Liste des informations de souscription
            payload: Le contenu de la notification
            
        Returns:
            Dict contenant le nombre de succès et d'échecs
        """
        results = {
            "success": 0,
            "failed": 0
        }
        
        for subscription in subscriptions:
            success = await self.send_notification(subscription, payload)
            if success:
                results["success"] += 1
            else:
                results["failed"] += 1
                
        return results

    def create_notification_payload(self, title: str, body: str) -> Dict[str, Any]:
        """
        Crée un payload de notification standardisé.
        
        Args:
            title: Titre de la notification
            body: Corps du message
            icon: URL de l'icône (optionnel)
            
        Returns:
            Dict contenant le payload formaté
        """
        payload = {
            "notification": {
                "title": title,
                "body": body,
                "icon": "assets/icon-128x128.png",
                "image": "assets/icon-128x128.png",
                "data": {
                    "onActionClick": {
                        "default": {"operation": "openWindow", "url": "/map"}
                    }
                }
            },
        }
            
        return payload
    
    async def delete_subscription(self, subscription: str, user_id: ObjectId) -> bool:
        """
        Supprime une souscription push de la base de données.
        
        Args:
            user_id: L'ID de l'utilisateur
            subscription: La souscription à supprimer
            
        Returns:
            bool: True si la suppression a réussi, False sinon
        """
        try:
            users_collection.update_one({"_id": user_id}, {"$pull": {"push_subscriptions": subscription}})
            return True
        except Exception as e:
            return False

# Instance singleton du service
push_service = PushService()