from __future__ import annotations import base64 import json import logging import os from pathlib import Path from typing import Any from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec from app.core.config import get_settings from app.services.tenant_store import SessionIdentity, get_tenant_store try: from pywebpush import WebPushException, webpush except ImportError: # pragma: no cover - runtime safety if dependency is unavailable. WebPushException = Exception # type: ignore[assignment] webpush = None # type: ignore[assignment] logger = logging.getLogger(__name__) def _urlsafe_base64(data: bytes) -> str: return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") def _private_key_path() -> Path: return Path(get_settings().push_vapid_private_key_file) def _load_or_create_private_key(): path = _private_key_path() if path.exists(): return serialization.load_pem_private_key(path.read_bytes(), password=None) path.parent.mkdir(parents=True, exist_ok=True) private_key = ec.generate_private_key(ec.SECP256R1()) pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), ) path.write_bytes(pem) try: os.chmod(path, 0o600) except OSError: logger.warning("Impossibile impostare i permessi del file VAPID %s", path) return private_key def get_push_public_key() -> str | None: if webpush is None: return None private_key = _load_or_create_private_key() public_numbers = private_key.public_key().public_numbers() raw_public_key = b"\x04" + public_numbers.x.to_bytes(32, "big") + public_numbers.y.to_bytes(32, "big") return _urlsafe_base64(raw_public_key) def _push_payload_for_report(report: dict[str, object]) -> dict[str, object]: title = str(report.get("title") or "Nuova segnalazione").strip() reporter = ( str(report.get("reporter_name") or report.get("reporter_username") or report.get("reporter_email") or "Dipendente") .strip() ) body = f"{reporter}: {title}" if len(body) > 160: body = f"{body[:157]}..." return { "title": "Nuova segnalazione", "body": body, "url": "/reports", "tag": f"report-{report.get('id') or ''}", "data": { "type": "report_created", "report_id": report.get("id"), }, } def _push_payload_for_fiscal_document_discrepancy(document: Any, order_match: dict[str, object]) -> dict[str, object]: document_id = str(getattr(document, "id", "") or "") supplier_name = str(getattr(document, "supplier_name", "") or "Fornitore non rilevato").strip() display_name = str(getattr(document, "display_name", "") or getattr(document, "original_name", "") or "documento fiscale").strip() matched_batch_id = order_match.get("matched_batch_id") missing_count = int(order_match.get("missing_line_count") or 0) extra_count = int(order_match.get("extra_line_count") or 0) differences: list[str] = [] if missing_count: missing_label = "riga mancante o parziale" if missing_count == 1 else "righe mancanti o parziali" differences.append(f"{missing_count} {missing_label}") if extra_count: extra_label = "riga extra" if extra_count == 1 else "righe extra" differences.append(f"{extra_count} {extra_label}") difference_label = ", ".join(differences) or "differenze da verificare" order_label = f"ordine #{matched_batch_id}" if matched_batch_id else "ordine abbinato" body = f"{supplier_name}: {difference_label} su {order_label}" if display_name: body = f"{body} ({display_name})" if len(body) > 180: body = f"{body[:177]}..." return { "title": "Discrepanza documento fiscale", "body": body, "url": "/fiscal-documents", "tag": f"fiscal-discrepancy-{document_id or matched_batch_id or 'document'}", "data": { "type": "fiscal_document_discrepancy", "document_id": document_id or None, "matched_batch_id": matched_batch_id, }, } def send_web_push(subscription: dict[str, Any], payload: dict[str, object]) -> tuple[bool, str | None, bool]: if webpush is None: return False, "pywebpush non disponibile", False endpoint = str(subscription.get("endpoint") or "").strip() keys = subscription.get("keys") if isinstance(subscription.get("keys"), dict) else {} if not endpoint or not keys.get("p256dh") or not keys.get("auth"): return False, "Sottoscrizione push incompleta", True try: response = webpush( subscription_info={ "endpoint": endpoint, "keys": { "p256dh": str(keys["p256dh"]), "auth": str(keys["auth"]), }, }, data=json.dumps(payload, ensure_ascii=False), vapid_private_key=str(_private_key_path()), vapid_claims={"sub": get_settings().push_vapid_subject}, timeout=10, ttl=86400, headers={"Urgency": "high"}, ) status_code = getattr(response, "status_code", None) logger.info("Notifica push inviata a %s con status %s", endpoint[:48], status_code or "ok") except WebPushException as exc: response = getattr(exc, "response", None) status_code = getattr(response, "status_code", None) disable = status_code in {404, 410} return False, str(exc), disable except Exception as exc: # pragma: no cover - defensive boundary for background tasks. logger.exception("Invio notifica push non riuscito") return False, str(exc), False return True, None, False def send_report_created_notification(session: SessionIdentity, report: dict[str, object]) -> None: store = get_tenant_store() payload = _push_payload_for_report(report) for subscription in store.list_admin_push_subscriptions(session): ok, error, disable = send_web_push(subscription, payload) if not ok: store.mark_push_subscription_error( session, str(subscription.get("endpoint") or ""), error or "Invio notifica push non riuscito", disable=disable, ) def send_fiscal_document_discrepancy_notification( session: SessionIdentity, document: Any, order_match: dict[str, object], ) -> None: store = get_tenant_store() payload = _push_payload_for_fiscal_document_discrepancy(document, order_match) for subscription in store.list_admin_push_subscriptions(session): ok, error, disable = send_web_push(subscription, payload) if not ok: store.mark_push_subscription_error( session, str(subscription.get("endpoint") or ""), error or "Invio notifica push non riuscito", disable=disable, )