125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""Routes de notifications Web Push."""
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_membre
|
|
from app.models.push_subscription import PushSubscription
|
|
from app.services.notifications import (
|
|
NOTIFICATION_TYPES,
|
|
normalized_preferences,
|
|
notifications_configured,
|
|
send_to_subscriptions,
|
|
)
|
|
|
|
router = APIRouter(prefix="/notifications", tags=["Notifications"])
|
|
|
|
|
|
class PushSubscriptionIn(BaseModel):
|
|
endpoint: str
|
|
expirationTime: int | None = None
|
|
keys: dict
|
|
|
|
|
|
class NotificationPreferencesIn(BaseModel):
|
|
preferences: dict[str, bool]
|
|
|
|
|
|
@router.get("/configuration")
|
|
async def configuration_notifications():
|
|
return {
|
|
"enabled": notifications_configured(),
|
|
"public_key": settings.VAPID_PUBLIC_KEY,
|
|
}
|
|
|
|
|
|
@router.get("/preferences")
|
|
async def lire_preferences_notifications(membre=Depends(get_current_membre)):
|
|
return {
|
|
"types": NOTIFICATION_TYPES,
|
|
"preferences": normalized_preferences(membre.notification_preferences),
|
|
}
|
|
|
|
|
|
@router.patch("/preferences")
|
|
async def modifier_preferences_notifications(
|
|
req: NotificationPreferencesIn,
|
|
membre=Depends(get_current_membre),
|
|
):
|
|
current = normalized_preferences(membre.notification_preferences)
|
|
for item in NOTIFICATION_TYPES:
|
|
code = item["code"]
|
|
if code in req.preferences:
|
|
current[code] = bool(req.preferences[code])
|
|
membre.notification_preferences = current
|
|
return {
|
|
"types": NOTIFICATION_TYPES,
|
|
"preferences": current,
|
|
}
|
|
|
|
|
|
@router.post("/abonnement")
|
|
async def enregistrer_abonnement(
|
|
abonnement: PushSubscriptionIn,
|
|
request: Request,
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not notifications_configured():
|
|
raise HTTPException(status_code=400, detail="Les notifications ne sont pas configurées sur ce serveur")
|
|
result = await db.execute(select(PushSubscription).where(PushSubscription.endpoint == abonnement.endpoint))
|
|
sub = result.scalar_one_or_none()
|
|
if sub:
|
|
sub.groupe_id = membre.groupe_id
|
|
sub.membre_id = membre.id
|
|
sub.abonnement = abonnement.model_dump()
|
|
sub.user_agent = (request.headers.get("user-agent") or "")[:255] or None
|
|
else:
|
|
db.add(PushSubscription(
|
|
groupe_id=membre.groupe_id,
|
|
membre_id=membre.id,
|
|
endpoint=abonnement.endpoint,
|
|
abonnement=abonnement.model_dump(),
|
|
user_agent=(request.headers.get("user-agent") or "")[:255] or None,
|
|
))
|
|
membre.push_sub = abonnement.model_dump()
|
|
await db.flush()
|
|
return {"message": "Notifications activées"}
|
|
|
|
|
|
@router.delete("/abonnement")
|
|
async def retirer_abonnement(
|
|
endpoint: str,
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(PushSubscription).where(
|
|
PushSubscription.endpoint == endpoint,
|
|
PushSubscription.membre_id == membre.id,
|
|
)
|
|
)
|
|
sub = result.scalar_one_or_none()
|
|
if sub:
|
|
await db.delete(sub)
|
|
membre.push_sub = None
|
|
await db.flush()
|
|
return {"message": "Notifications désactivées"}
|
|
|
|
|
|
@router.post("/test")
|
|
async def envoyer_test(
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(PushSubscription).where(PushSubscription.membre_id == membre.id))
|
|
sent = await send_to_subscriptions(db, result.scalars().all(), {
|
|
"title": "Notifications activées",
|
|
"body": "Cet appareil recevra les avis du groupe.",
|
|
"url": "/",
|
|
})
|
|
await db.flush()
|
|
return {"message": "Notification de test envoyée", "sent": sent}
|