Spaces:
Running
Running
File size: 3,316 Bytes
ee7d7b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | """
Push Service - Send web push notifications
Uses pywebpush library with VAPID authentication
"""
import os
import json
from pywebpush import webpush, WebPushException
import logging
logger = logging.getLogger(__name__)
# VAPID keys (generate with: webpush.generate_vapid_keys())
VAPID_PRIVATE_KEY = os.getenv("VAPID_PRIVATE_KEY")
VAPID_PUBLIC_KEY = os.getenv("VAPID_PUBLIC_KEY")
VAPID_CLAIMS = {"sub": "mailto:admin@ai-business-analyst.com"}
async def send_push_notification(
token: str,
title: str,
body: str,
data: dict = None
):
"""Send push notification to a specific token"""
if not VAPID_PRIVATE_KEY or not VAPID_PUBLIC_KEY:
logger.warning("VAPID keys not configured, skipping push send")
return
try:
# Parse subscription (token is JSON string)
subscription_info = json.loads(token)
# Create notification payload
notification_payload = json.dumps({
"title": title,
"body": body,
"icon": "/logo.png",
"badge": "/badge.png",
"data": data or {},
"actions": [
{"action": "open", "title": "View Dashboard"},
{"action": "dismiss", "title": "Dismiss"}
]
})
# Send push notification
webpush(
subscription_info=subscription_info,
data=notification_payload,
vapid_private_key=VAPID_PRIVATE_KEY,
vapid_claims=VAPID_CLAIMS
)
logger.info(f"Push notification sent: {title}")
except WebPushException as e:
logger.error(f"WebPush error: {e}")
# Check if token is invalid/expired
if e.response and e.response.status_code in [404, 410]:
raise Exception("Token invalid or expired")
raise
except Exception as e:
logger.error(f"Failed to send push notification: {e}")
raise
def generate_vapid_keys():
"""
Utility to generate VAPID key pair
Run this once and store keys in environment variables
"""
from py_vapid import Vapid
from cryptography.hazmat.primitives import serialization
vapid = Vapid()
vapid.generate_keys()
# Serialize public key
public_key = vapid.public_key.public_bytes(
encoding=serialization.Encoding.X962,
format=serialization.PublicFormat.UncompressedPoint
)
# Serialize private key
private_key = vapid.private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
import base64
public_key_b64 = base64.urlsafe_b64encode(public_key).decode('utf-8').rstrip('=')
private_key_str = private_key.decode('utf-8')
print("\n" + "="*60)
print("VAPID Keys Generated Successfully!")
print("="*60)
print("\nAdd these to your .env files:\n")
print("Backend (.env):")
print(f"VAPID_PUBLIC_KEY={public_key_b64}")
print(f"VAPID_PRIVATE_KEY={private_key_str}")
print(f"\nFrontend (.env):")
print(f"VITE_VAPID_PUBLIC_KEY={public_key_b64}")
print("\n" + "="*60 + "\n")
if __name__ == "__main__":
# Run to generate VAPID keys
generate_vapid_keys()
|