You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
107 lines
3.3 KiB
107 lines
3.3 KiB
from pywebpush import webpush, WebPushException
|
|
import json
|
|
from typing import List, Dict, Any
|
|
import logging
|
|
from urllib.parse import urlparse
|
|
|
|
from app.config import VAPID_PRIVATE_KEY, VAPID_CLAIMS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def get_audience(endpoint: str) -> str:
|
|
parsed = urlparse(endpoint)
|
|
return f"{parsed.scheme}://{parsed.netloc}"
|
|
|
|
class PushService:
|
|
def __init__(self):
|
|
self.vapid_private_key = VAPID_PRIVATE_KEY
|
|
self.vapid_claims = VAPID_CLAIMS
|
|
|
|
async def send_notification(self, subscription: Dict[str, Any], payload: Dict[str, Any]) -> bool:
|
|
"""
|
|
Envoie une notification push à un abonné spécifique.
|
|
|
|
Args:
|
|
subscription: Les informations de souscription push
|
|
payload: Le contenu de la notification
|
|
|
|
Returns:
|
|
bool: True si l'envoi a réussi, False sinon
|
|
"""
|
|
try:
|
|
vapid_claims = self.vapid_claims
|
|
vapid_claims["aud"] = get_audience(subscription["endpoint"])
|
|
|
|
print(f"Envoi de la notification push à {subscription} avec les claims: {vapid_claims}")
|
|
print(f"Payload: {json.dumps(payload)}")
|
|
|
|
webpush(
|
|
subscription_info=subscription,
|
|
data=json.dumps(payload),
|
|
vapid_private_key=self.vapid_private_key,
|
|
vapid_claims=vapid_claims,
|
|
verbose=True
|
|
)
|
|
return True
|
|
except WebPushException as e:
|
|
logger.error(f"Erreur lors de l'envoi de la notification push: {str(e)}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Erreur inattendue lors de l'envoi de la notification push: {str(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, icon: str = None) -> 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
|
|
},
|
|
"data": {
|
|
"url": "/map"
|
|
}
|
|
}
|
|
|
|
if icon:
|
|
payload["notification"]["icon"] = icon
|
|
|
|
return payload
|
|
|
|
# Instance singleton du service
|
|
push_service = PushService()
|