""" Web Push отправитель для browser push notifications """ import json from datetime import timedelta from urllib.parse import urlencode from django.conf import settings from django.utils import timezone from push_notifications.models import WebPushDevice from notifications.chat_suppression import suppress_native_for_active_chat from notifications.push_actions import ( build_push_avatar_token, build_push_read_token, ) from ..config import get from ..models import NotificationDeliveryRouting, NotificationInstallation from ..routing import APP_MAIN, APP_MESSENGER, normalize_routes, notification_category from .base import BaseNotificationSender class PushNotificationSender(BaseNotificationSender): """ Отправитель Web Push уведомлений через django-push-notifications. Доставляет уведомления в браузер даже когда вкладка закрыта. """ CHAT_NOTIFICATION_VERBS = { "chat_new_message", "chat_mention", "chat_reply", "announcement_new_message", } DEFAULT_TITLE = "Новое уведомление" def can_send(self, notification, user_preferences) -> bool: """Проверяет, включены ли push уведомления""" if not user_preferences.push_enabled: self.log_skip(notification, "push_enabled=False") return False return True def _get_actor_name(self, notification) -> str: actor = notification.actor if not actor: return "" return str(actor).strip() def _get_data_title(self, notification) -> str: data = notification.data if isinstance(notification.data, dict) else {} title = data.get("title") if not isinstance(title, str): return "" return title.strip() def _get_site_url(self) -> str: site_url = getattr(settings, "SITE_URL", "").strip() if site_url: return site_url.rstrip("/") if settings.ALLOWED_HOSTS and settings.ALLOWED_HOSTS[0] != "*": protocol = ( "https" if getattr(settings, "SECURE_SSL_REDIRECT", False) else "http" ) return f"{protocol}://{settings.ALLOWED_HOSTS[0]}" return "http://localhost:9000" def _get_actor_avatar_url(self, notification) -> str: actor = notification.actor if not actor: return "" try: avatar = getattr(actor, "avatar", None) if not avatar: return "" avatar_path = avatar.name except Exception: return "" if not avatar_path: return "" token = build_push_avatar_token(avatar_path) query = urlencode({"token": token}) return f"{self._get_site_url()}/api/v1/notifications/push/avatar/?{query}" def _build_title(self, notification, source_data: dict) -> str: if notification.verb in self.CHAT_NOTIFICATION_VERBS: chat_type = source_data.get("chat_type") chat_name = source_data.get("chat_name") if chat_type and chat_type != "private" and chat_name: return str(chat_name).strip() return self._get_actor_name(notification) or self.DEFAULT_TITLE return self._get_data_title(notification) or self.DEFAULT_TITLE def _chat_preferences(self, notification): data = notification.data if isinstance(notification.data, dict) else {} chat_id = data.get("chat_id") if not chat_id or notification.verb not in self.CHAT_NOTIFICATION_VERBS: return None from communications.models import ChatUserSettings return ChatUserSettings.objects.filter( chat_id=chat_id, user_id=notification.recipient_id, ).first() def build_message_data(self, notification) -> dict: """Build a browser payload with stable chat identity and privacy flags.""" source_data = notification.data if isinstance(notification.data, dict) else {} chat_id = source_data.get("chat_id") message_id = source_data.get("message_id") preferences = self._chat_preferences(notification) show_preview = preferences is None or preferences.show_notification_preview actor_name = self._get_actor_name(notification) title = self._build_title(notification, source_data) body = notification.description or "" if not show_preview and notification.verb in self.CHAT_NOTIFICATION_VERBS: title = source_data.get("chat_name") or self.DEFAULT_TITLE body = "Новое сообщение" elif ( notification.verb in self.CHAT_NOTIFICATION_VERBS and source_data.get("chat_type") not in (None, "private") and actor_name ): body = f"{actor_name}: {body}" if len(body) > 300: body = body[:297] + "..." message_data = { "title": title, "head": title, "body": body, "url": notification.action_url or "/", "tag": f"chat-{chat_id}" if chat_id else f"notification-{notification.id}", "requireInteraction": False, "silent": bool( preferences is not None and not preferences.notification_sound_enabled ), "data": { "url": notification.action_url or "/", "notification_id": notification.id, "chat_id": chat_id, "message_id": message_id, "chat_name": source_data.get("chat_name"), "chat_type": source_data.get("chat_type"), "actor_name": actor_name, }, } if notification.verb in self.CHAT_NOTIFICATION_VERBS and chat_id and message_id: try: read_token = build_push_read_token( notification_id=notification.id, user_id=notification.recipient_id, chat_id=int(chat_id), message_id=int(message_id), ) except (TypeError, ValueError): read_token = "" if read_token: message_data["actions"] = [ { "action": "mark-read", "type": "button", "title": "Прочитано", } ] message_data["data"].update( { "mark_read_url": "/api/v1/notifications/push/action/read/", "mark_read_token": read_token, } ) actor_icon = ( self._get_actor_avatar_url(notification) if notification.verb in self.CHAT_NOTIFICATION_VERBS and show_preview else "" ) default_icon = get("PUSH_DEFAULT_ICON") default_badge = get("PUSH_DEFAULT_BADGE") if actor_icon: message_data["icon"] = actor_icon elif default_icon: message_data["icon"] = default_icon if default_badge: message_data["badge"] = default_badge return message_data def _is_permanent_device_error(self, error: Exception) -> bool: error_text = str(error).lower() permanent_markers = ( "expired", "unregistered", "invalid subscription", "invalidregistration", "not found", "gone", "404", "410", "400 bad request", ) return any(marker in error_text for marker in permanent_markers) def _deactivate_device(self, device, reason: str) -> None: if device.active: device.active = False device.save(update_fields=["active"]) NotificationInstallation.objects.filter( push_device_id=device.id, ).update(endpoint=None, push_device_id=None) self.logger.info( "Push device deactivated after permanent delivery error", extra={ "event": "notification.push.device_deactivated", "device_id": device.id, "reason": "permanent_delivery_error", }, ) @staticmethod def _push_result_succeeded(result) -> bool: if not isinstance(result, dict): return True return bool(result.get("success", 0)) and not result.get("failure", 0) def _delivery_devices(self, notification): user = notification.recipient active_devices = list(WebPushDevice.objects.filter(user=user, active=True)) if not active_devices: return [], {} devices_by_endpoint = { device.registration_id: device for device in active_devices } installations = list( NotificationInstallation.objects.filter( user=user, active=True, endpoint__in=devices_by_endpoint, ) ) represented_endpoints = { installation.endpoint for installation in installations } selected = [ device for device in active_devices if device.registration_id not in represented_endpoints ] fallback_by_device_id = {} category = notification_category(notification.verb) routing_by_group = { str(routing.device_group_id): normalize_routes(routing.routes) for routing in NotificationDeliveryRouting.objects.filter( user=user, device_group_id__in={ installation.device_group_id for installation in installations }, ) } grouped = {} messenger_cutoff = timezone.now() - timedelta( days=getattr(settings, "MESSENGER_PUSH_LEASE_DAYS", 30), ) for installation in installations: if ( installation.app_kind == APP_MESSENGER and installation.last_seen_at < messenger_cutoff ): continue grouped.setdefault(str(installation.device_group_id), {}).setdefault( installation.app_kind, [], ).append(devices_by_endpoint[installation.endpoint]) for group_id, app_devices in grouped.items(): routes = routing_by_group.get(group_id, normalize_routes({})) destinations = routes[category] main_devices = app_devices.get(APP_MAIN, []) messenger_devices = app_devices.get(APP_MESSENGER, []) if APP_MAIN in destinations: selected.extend(main_devices) if APP_MESSENGER in destinations: if messenger_devices: selected.extend(messenger_devices) if APP_MAIN not in destinations and main_devices: for messenger_device in messenger_devices: fallback_by_device_id[messenger_device.id] = main_devices elif APP_MAIN not in destinations: selected.extend(main_devices) unique_selected = {device.id: device for device in selected} return list(unique_selected.values()), fallback_by_device_id def send(self, notification, **kwargs) -> bool: """ Отправляет Web Push уведомление. Args: notification: Объект Notification **kwargs: Дополнительные параметры Returns: True если отправлено успешно, False иначе """ try: if notification.deleted or not notification.unread: self.log_skip(notification, "notification no longer active") return False if suppress_native_for_active_chat(notification): self.log_skip(notification, "recipient is viewing chat") return False chat_preferences = self._chat_preferences(notification) if ( chat_preferences is not None and not chat_preferences.allows_notification(notification.verb) ): self.log_skip(notification, "chat notification policy") return False devices, fallback_by_device_id = self._delivery_devices(notification) if not devices: self.log_skip(notification, "no active push devices") return False message_data = self.build_message_data(notification) # Преобразуем в JSON-строку для send_message() message = json.dumps(message_data) sent_count = 0 failed_count = 0 # Отправляем через все устройства пользователя sent_device_ids = set() def send_to_device(device): nonlocal sent_count, failed_count try: result = device.send_message(message) if not self._push_result_succeeded(result): failed_count += 1 device.refresh_from_db(fields=["active"]) if not device.active: self._deactivate_device( device, "provider rejected subscription" ) self.logger.warning( "Push provider rejected delivery", extra={ "event": "notification.push.provider_rejected", "device_id": device.id, "notification_id": notification.id, }, ) return False sent_count += 1 sent_device_ids.add(device.id) self.logger.debug( "Push delivered to device", extra={ "event": "notification.push.device_delivered", "device_id": device.id, "browser": device.browser, }, ) return True except Exception as e: failed_count += 1 # Вычисляем размер payload для диагностики message_size = len(message.encode("utf-8")) error_text = str(e) self.logger.warning( "Push delivery to device failed", extra={ "event": "notification.push.device_failed", "device_id": device.id, "payload_size_bytes": message_size, "error_type": type(e).__name__, }, exc_info=True, ) if self._is_permanent_device_error(e): self._deactivate_device(device, error_text) return False for device in devices: delivered = send_to_device(device) if delivered: continue for fallback_device in fallback_by_device_id.get(device.id, []): if fallback_device.id in sent_device_ids: continue if send_to_device(fallback_device): break if sent_count > 0: self.log_success(notification, f"{sent_count} devices") if failed_count > 0: self.logger.warning( "Push delivered partially", extra={ "event": "notification.push.partially_delivered", "notification_id": notification.id, "sent_count": sent_count, "failed_count": failed_count, }, ) return True else: self.log_skip( notification, f"no successful sends ({failed_count} failures)", ) return False except ImportError: self.log_skip(notification, "django-push-notifications not installed") return False except Exception as e: self.log_error(notification, e, f"user_{notification.recipient.id}") return False