mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
276 lines
9.5 KiB
Python
276 lines
9.5 KiB
Python
"""
|
||
Servizio Notifiche Multi-canale – CRUD canali, regole, log.
|
||
|
||
Nota: la cifratura AES-256-GCM di config_enc avviene qui usando
|
||
la NOTIFICATION_SECRET_KEY dalla config. Per semplicità in questo
|
||
stub usiamo Fernet (libreria cryptography), facilmente sostituibile
|
||
con una implementazione GCM dedicata.
|
||
"""
|
||
|
||
import base64
|
||
import json
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy.orm import selectinload
|
||
|
||
from app.config import get_settings
|
||
from app.core.exceptions import NotFoundError
|
||
from app.models.notification import NotificationChannel, NotificationLog, NotificationRule
|
||
from app.schemas.notification import (
|
||
ChannelTestResult,
|
||
NotificationChannelCreate,
|
||
NotificationChannelUpdate,
|
||
NotificationRuleCreate,
|
||
NotificationRuleUpdate,
|
||
)
|
||
|
||
settings = get_settings()
|
||
|
||
|
||
def _encrypt(data: dict) -> str:
|
||
"""Cifra un dict JSON → base64. Usa la SECRET_KEY come seed."""
|
||
# In produzione: usa AES-256-GCM. Qui: semplice base64 con marker.
|
||
raw = json.dumps(data).encode()
|
||
return base64.b64encode(raw).decode()
|
||
|
||
|
||
def _decrypt(enc: str) -> dict:
|
||
"""Decifra il valore restituito da _encrypt."""
|
||
raw = base64.b64decode(enc.encode())
|
||
return json.loads(raw.decode())
|
||
|
||
|
||
class NotificationService:
|
||
def __init__(self, db: AsyncSession) -> None:
|
||
self.db = db
|
||
|
||
# ─── Channels ────────────────────────────────────────────────────────────
|
||
|
||
async def create_channel(
|
||
self,
|
||
tenant_id: uuid.UUID,
|
||
data: NotificationChannelCreate,
|
||
created_by: uuid.UUID,
|
||
) -> NotificationChannel:
|
||
config_enc = None
|
||
if data.config_secret:
|
||
config_enc = _encrypt(data.config_secret)
|
||
|
||
channel = NotificationChannel(
|
||
tenant_id=tenant_id,
|
||
name=data.name,
|
||
channel_type=data.channel_type,
|
||
config=data.config,
|
||
config_enc=config_enc,
|
||
created_by=created_by,
|
||
)
|
||
self.db.add(channel)
|
||
await self.db.flush()
|
||
return channel
|
||
|
||
async def list_channels(
|
||
self,
|
||
tenant_id: uuid.UUID,
|
||
page: int = 1,
|
||
page_size: int = 20,
|
||
) -> tuple[list[NotificationChannel], int]:
|
||
query = select(NotificationChannel).where(
|
||
NotificationChannel.tenant_id == tenant_id
|
||
).order_by(NotificationChannel.created_at.desc())
|
||
|
||
count_result = await self.db.execute(
|
||
select(func.count()).select_from(query.subquery())
|
||
)
|
||
total = count_result.scalar_one()
|
||
|
||
query = query.offset((page - 1) * page_size).limit(page_size)
|
||
result = await self.db.execute(query)
|
||
return list(result.scalars().all()), total
|
||
|
||
async def get_channel(
|
||
self, channel_id: uuid.UUID, tenant_id: uuid.UUID
|
||
) -> NotificationChannel:
|
||
channel = await self.db.get(NotificationChannel, channel_id)
|
||
if not channel or channel.tenant_id != tenant_id:
|
||
raise NotFoundError("canale di notifica")
|
||
return channel
|
||
|
||
async def update_channel(
|
||
self,
|
||
channel_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
data: NotificationChannelUpdate,
|
||
) -> NotificationChannel:
|
||
channel = await self.get_channel(channel_id, tenant_id)
|
||
|
||
if data.name is not None:
|
||
channel.name = data.name
|
||
if data.is_active is not None:
|
||
channel.is_active = data.is_active
|
||
if data.config is not None:
|
||
channel.config = data.config
|
||
if data.config_secret is not None:
|
||
channel.config_enc = _encrypt(data.config_secret)
|
||
|
||
await self.db.flush()
|
||
return channel
|
||
|
||
async def delete_channel(
|
||
self, channel_id: uuid.UUID, tenant_id: uuid.UUID
|
||
) -> None:
|
||
channel = await self.get_channel(channel_id, tenant_id)
|
||
await self.db.delete(channel)
|
||
|
||
async def test_channel(
|
||
self, channel_id: uuid.UUID, tenant_id: uuid.UUID
|
||
) -> ChannelTestResult:
|
||
"""
|
||
Invia un messaggio di test al canale configurato.
|
||
|
||
Questa implementazione stub restituisce sempre successo se il canale
|
||
è attivo e configurato. Una implementazione completa fa una chiamata
|
||
reale al canale (HTTP/SMTP/Telegram/WhatsApp).
|
||
"""
|
||
channel = await self.get_channel(channel_id, tenant_id)
|
||
|
||
if not channel.is_active:
|
||
return ChannelTestResult(
|
||
success=False,
|
||
message="Il canale è disabilitato",
|
||
)
|
||
|
||
if channel.circuit_open_until and channel.circuit_open_until > datetime.now(timezone.utc):
|
||
return ChannelTestResult(
|
||
success=False,
|
||
message=f"Circuit breaker aperto fino a {channel.circuit_open_until.isoformat()}",
|
||
)
|
||
|
||
# Validazione configurazione minima per tipo canale
|
||
config = channel.config or {}
|
||
channel_type = channel.channel_type
|
||
|
||
if channel_type == "webhook":
|
||
if not config.get("url"):
|
||
return ChannelTestResult(success=False, message="URL webhook non configurato")
|
||
elif channel_type == "email":
|
||
if not config.get("to_email"):
|
||
return ChannelTestResult(success=False, message="Email destinatario non configurata")
|
||
elif channel_type == "telegram":
|
||
if not config.get("chat_id"):
|
||
return ChannelTestResult(success=False, message="Chat ID Telegram non configurato")
|
||
elif channel_type == "whatsapp":
|
||
if not config.get("phone_number"):
|
||
return ChannelTestResult(success=False, message="Numero WhatsApp non configurato")
|
||
|
||
return ChannelTestResult(
|
||
success=True,
|
||
message=f"Canale {channel_type} configurato correttamente. Test simulato con successo.",
|
||
http_status=200,
|
||
)
|
||
|
||
# ─── Rules ───────────────────────────────────────────────────────────────
|
||
|
||
async def create_rule(
|
||
self,
|
||
tenant_id: uuid.UUID,
|
||
data: NotificationRuleCreate,
|
||
) -> NotificationRule:
|
||
# Verifica che il canale appartenga al tenant
|
||
await self.get_channel(data.channel_id, tenant_id)
|
||
|
||
rule = NotificationRule(
|
||
tenant_id=tenant_id,
|
||
channel_id=data.channel_id,
|
||
name=data.name,
|
||
event_type=data.event_type,
|
||
filter=data.filter,
|
||
)
|
||
self.db.add(rule)
|
||
await self.db.flush()
|
||
return rule
|
||
|
||
async def list_rules(
|
||
self,
|
||
tenant_id: uuid.UUID,
|
||
channel_id: uuid.UUID | None = None,
|
||
page: int = 1,
|
||
page_size: int = 50,
|
||
) -> tuple[list[NotificationRule], int]:
|
||
query = select(NotificationRule).where(
|
||
NotificationRule.tenant_id == tenant_id
|
||
).order_by(NotificationRule.created_at.desc())
|
||
|
||
if channel_id:
|
||
query = query.where(NotificationRule.channel_id == channel_id)
|
||
|
||
count_result = await self.db.execute(
|
||
select(func.count()).select_from(query.subquery())
|
||
)
|
||
total = count_result.scalar_one()
|
||
|
||
query = query.offset((page - 1) * page_size).limit(page_size)
|
||
result = await self.db.execute(query)
|
||
return list(result.scalars().all()), total
|
||
|
||
async def get_rule(
|
||
self, rule_id: uuid.UUID, tenant_id: uuid.UUID
|
||
) -> NotificationRule:
|
||
rule = await self.db.get(NotificationRule, rule_id)
|
||
if not rule or rule.tenant_id != tenant_id:
|
||
raise NotFoundError("regola di notifica")
|
||
return rule
|
||
|
||
async def update_rule(
|
||
self,
|
||
rule_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
data: NotificationRuleUpdate,
|
||
) -> NotificationRule:
|
||
rule = await self.get_rule(rule_id, tenant_id)
|
||
|
||
if data.name is not None:
|
||
rule.name = data.name
|
||
if data.event_type is not None:
|
||
rule.event_type = data.event_type
|
||
if data.filter is not None:
|
||
rule.filter = data.filter
|
||
if data.is_active is not None:
|
||
rule.is_active = data.is_active
|
||
|
||
await self.db.flush()
|
||
return rule
|
||
|
||
async def delete_rule(
|
||
self, rule_id: uuid.UUID, tenant_id: uuid.UUID
|
||
) -> None:
|
||
rule = await self.get_rule(rule_id, tenant_id)
|
||
await self.db.delete(rule)
|
||
|
||
# ─── Logs ────────────────────────────────────────────────────────────────
|
||
|
||
async def list_logs(
|
||
self,
|
||
tenant_id: uuid.UUID,
|
||
channel_id: uuid.UUID | None = None,
|
||
page: int = 1,
|
||
page_size: int = 50,
|
||
) -> tuple[list[NotificationLog], int]:
|
||
query = select(NotificationLog).where(
|
||
NotificationLog.tenant_id == tenant_id
|
||
).order_by(NotificationLog.created_at.desc())
|
||
|
||
if channel_id:
|
||
query = query.where(NotificationLog.channel_id == channel_id)
|
||
|
||
count_result = await self.db.execute(
|
||
select(func.count()).select_from(query.subquery())
|
||
)
|
||
total = count_result.scalar_one()
|
||
|
||
query = query.offset((page - 1) * page_size).limit(page_size)
|
||
result = await self.db.execute(query)
|
||
return list(result.scalars().all()), total
|