""" Notification services — point d'entrée pour créer des notifications depuis n'importe quelle app. Usage depuis une autre app: from notifications.services import notify_user, notify_team, notify_users # Notification personnelle notify_user( recipient=user, title="Nouvelle intervention assignée", content="L'intervention #123 vous a été assignée.", notification_type='task', action_url='/interventions/123/', action_label="Voir l'intervention", source_app='interventions', sender=request.user, ) # Notification à une équipe from notifications.models import Team team = Team.objects.get(name='Équipe technique') notify_team( team=team, title="Stock bas", content="Le niveau de stock du produit X est critique.", notification_type='alert', source_app='stock', ) """ from django.contrib.auth.models import User from django.utils import timezone import re as _re # Mention pattern: @[Display Name](u:123) or @[Team Name](t:456) _MENTION_PATTERN = _re.compile(r'@\[([^\]]+)\]\(([ut]):(\d+)\)') def _create_notification( title, content='', notification_type='info', action_url='', action_label='', source_app='', sender=None, expires_at=None, is_system=False, is_single_claim=False, ): from .models import Notification return Notification.objects.create( title=title, content=content, notification_type=notification_type, action_url=action_url, action_label=action_label, source_app=source_app, sender=sender, expires_at=expires_at, is_system=is_system, is_single_claim=is_single_claim, ) def notify_user( recipient, title, content='', notification_type='info', action_url='', action_label='', source_app='', sender=None, expires_at=None, is_system=False, ): """ Envoie une notification personnelle à un utilisateur. Retourne le NotificationRecipient créé. """ from .models import NotificationRecipient notification = _create_notification( title=title, content=content, notification_type=notification_type, action_url=action_url, action_label=action_label, source_app=source_app, sender=sender, expires_at=expires_at, is_system=is_system, ) recipient_obj = NotificationRecipient.objects.create( notification=notification, user=recipient, ) return recipient_obj def notify_users( recipients, title, content='', notification_type='info', action_url='', action_label='', source_app='', sender=None, expires_at=None, is_system=False, ): """ Envoie une notification à plusieurs utilisateurs (crée un seul objet Notification avec plusieurs NotificationRecipient). Retourne la Notification créée. """ from .models import Notification, NotificationRecipient notification = _create_notification( title=title, content=content, notification_type=notification_type, action_url=action_url, action_label=action_label, source_app=source_app, sender=sender, expires_at=expires_at, is_system=is_system, ) NotificationRecipient.objects.bulk_create([ NotificationRecipient(notification=notification, user=user) for user in recipients ], ignore_conflicts=True) return notification def notify_team( team, title, content='', notification_type='info', action_url='', action_label='', source_app='', sender=None, expires_at=None, is_system=False, ): """ Envoie une notification à toute une équipe. Retourne la TeamNotificationTarget créée. """ from .models import TeamNotificationTarget notification = _create_notification( title=title, content=content, notification_type=notification_type, action_url=action_url, action_label=action_label, source_app=source_app, sender=sender, expires_at=expires_at, is_system=is_system, ) target = TeamNotificationTarget.objects.create( notification=notification, team=team, ) return target def get_user_notifications(user, include_archived=False): """ Retourne une liste unifiée de notifications pour un utilisateur, combinant notifications personnelles et notifications d'équipe. Chaque élément est un dict avec: - notification: Notification - is_read: bool - is_archived: bool - recipient_type: 'personal' | 'team' - team_name: str (si team) - recipient_id: int (NotificationRecipient.pk ou TeamNotificationTarget.pk) """ from .models import NotificationRecipient, TeamNotificationTarget, TeamNotificationRead results = [] # --- Personal notifications --- personal_qs = NotificationRecipient.objects.select_related( 'notification', 'notification__sender' ).prefetch_related('notification__attachments').filter(user=user) if not include_archived: personal_qs = personal_qs.filter(archived_at__isnull=True) for nr in personal_qs: if nr.notification.is_expired: continue results.append({ 'notification': nr.notification, 'is_read': nr.is_read, 'is_archived': nr.is_archived, 'recipient_type': 'personal', 'team_name': None, 'recipient_id': nr.pk, }) # --- Team notifications --- user_team_ids = user.notification_teams.values_list('id', flat=True) team_targets = TeamNotificationTarget.objects.select_related( 'notification', 'notification__sender', 'team' ).prefetch_related('notification__attachments').filter(team_id__in=user_team_ids) read_ids = TeamNotificationRead.objects.filter( user=user, team_notification__in=team_targets, ).values_list('team_notification_id', flat=True) for tt in team_targets: if tt.notification.is_expired: continue results.append({ 'notification': tt.notification, 'is_read': tt.pk in read_ids, 'is_archived': False, 'recipient_type': 'team', 'team_name': tt.team.name, 'recipient_id': tt.pk, }) # Sort by created_at descending results.sort(key=lambda x: x['notification'].created_at, reverse=True) return results def get_unread_count(user): """Retourne le nombre de notifications non lues pour un utilisateur.""" from .models import NotificationRecipient, TeamNotificationTarget, TeamNotificationRead personal_unread = NotificationRecipient.objects.filter( user=user, read_at__isnull=True, archived_at__isnull=True, ).count() user_team_ids = user.notification_teams.values_list('id', flat=True) team_targets = TeamNotificationTarget.objects.filter(team_id__in=user_team_ids) team_read_ids = TeamNotificationRead.objects.filter( user=user, team_notification__in=team_targets, ).values_list('team_notification_id', flat=True) team_unread = team_targets.exclude(pk__in=team_read_ids).count() return personal_unread + team_unread # ────────────────────────────────────────────────────────────────────────────── # Discussion / Messaging services # ────────────────────────────────────────────────────────────────────────────── def post_message(discussion, author, content, files=None): """ Poste un message dans une discussion. - Crée l'objet Message (+ pièces jointes). - Efface les notifications non lues liées à cette discussion pour tous les participants (la réponse annule les alertes précédentes). - Crée une nouvelle notification pour tous les participants sauf l'auteur. - Met à jour discussion.updated_at pour la faire remonter dans les listes. Retourne le Message créé. """ from .models import Discussion, Message, MessageAttachment, NotificationRecipient message = Message.objects.create(discussion=discussion, author=author, content=content) if files: for f in files: MessageAttachment.objects.create(message=message, file=f, name=f.name) # Effacer les anciennes notifications non lues de cette discussion NotificationRecipient.objects.filter( notification__source_discussion=discussion, read_at__isnull=True, ).update(read_at=timezone.now()) # Bump discussion.updated_at sans déclencher auto_now Discussion.objects.filter(pk=discussion.pk).update(updated_at=timezone.now()) # Notifier les utilisateurs et équipes mentionnés dans le message _notify_mentions(discussion, author, content) return message def _notify_mentions(discussion, author, content): """ Parse les mentions @[Name](u:id) et @[Name](t:id) dans le contenu d'un message et envoie des notifications ciblées aux personnes/équipes mentionnées. Ajoute également les utilisateurs mentionnés comme participants de la discussion. """ from .models import Notification, NotificationRecipient, TeamNotificationTarget, Team mentioned_user_ids = set() mentioned_team_ids = set() for m in _MENTION_PATTERN.finditer(content): kind = m.group(2) obj_id = int(m.group(3)) if kind == 'u': mentioned_user_ids.add(obj_id) else: mentioned_team_ids.add(obj_id) if not mentioned_user_ids and not mentioned_team_ids: return action_url = discussion.object_url or discussion.get_absolute_url() action_label = discussion.object_label or str(discussion) author_name = author.get_full_name().strip() or author.username if author else "Système" exclude_pk = author.pk if author else -1 title = f"{author_name} vous a mentionné — {discussion.title or str(discussion)}" msg_preview = content[:200] # Mentions d'utilisateurs individuels users_to_notify = list( User.objects.filter(pk__in=mentioned_user_ids).exclude(pk=exclude_pk) ) if users_to_notify: notif = Notification.objects.create( title=title, content=msg_preview, notification_type=Notification.TYPE_MESSAGE, sender=author, action_url=action_url, action_label=action_label, source_app='discussions', source_discussion=discussion, ) NotificationRecipient.objects.bulk_create([ NotificationRecipient(notification=notif, user=u) for u in users_to_notify ], ignore_conflicts=True) # Ajouter les utilisateurs mentionnés comme participants discussion.participants.add(*users_to_notify) # Mentions d'équipes if mentioned_team_ids: teams = Team.objects.filter(pk__in=mentioned_team_ids).prefetch_related('members') for team in teams: notif = Notification.objects.create( title=title, content=msg_preview, notification_type=Notification.TYPE_MESSAGE, sender=author, action_url=action_url, action_label=action_label, source_app='discussions', source_discussion=discussion, ) TeamNotificationTarget.objects.create(notification=notif, team=team) # Ajouter les membres de l'équipe comme participants discussion.participants.add(*team.members.all()) def mark_discussion_seen(discussion, user=None): """ Marque comme lues les notifications non lues liées à cette discussion. Si `user` est fourni, seules les notifications de cet utilisateur sont marquées. """ from .models import NotificationRecipient qs = NotificationRecipient.objects.filter( notification__source_discussion=discussion, read_at__isnull=True, ) if user is not None: qs = qs.filter(user=user) qs.update(read_at=timezone.now()) # ────────────────────────────────────────────────────────────────────────────── # Règles de notification automatiques # ────────────────────────────────────────────────────────────────────────────── def process_notification_rules(event, context, source_app=''): """ Déclenche les règles de notification actives correspondant à *event*. Arguments --------- event : str Identifiant de l'événement, ex. ``'intervention_canceled'``. context : dict Données contextuelles utilisées pour : - filtrer les règles (ex. ``cancellation_type``) ; - interpoler les modèles de titre/contenu (format Python ``{var}``) ; - fournir ``action_url`` et ``action_label`` pour le lien. source_app : str Application source enregistrée sur les notifications créées. Exemple d'appel (depuis interventions) :: from notifications.services import process_notification_rules process_notification_rules( event='intervention_canceled', context={ 'intervention_code': intervention.code, 'intervention_title': intervention.title, 'cancellation_type': 'to_redirect', 'cancellation_type_label': 'A rediriger', 'cancellation_reason': reason_text, 'action_url': '/interventions/42/', 'action_label': "Voir l'intervention", }, source_app='interventions', ) """ from .models import NotificationRule, NotificationRecipient rules = NotificationRule.objects.filter( is_active=True, trigger_event=event ).prefetch_related('recipient_teams', 'recipient_users') cancellation_type = context.get('cancellation_type', '') for rule in rules: # Filtre conditionnel : si la règle précise un type d'annulation, # on ne déclenche que si ça correspond. if rule.trigger_cancellation_type and rule.trigger_cancellation_type != cancellation_type: continue # Préparer les variables sécurisées pour l'interpolation safe_ctx = {k: (v or '') for k, v in context.items()} try: title = rule.title_template.format(**safe_ctx) except (KeyError, ValueError): title = rule.title_template try: content = rule.content_template.format(**safe_ctx) if rule.content_template else '' except (KeyError, ValueError): content = rule.content_template notification = _create_notification( title=title, content=content, notification_type=rule.notification_type, action_url=safe_ctx.get('action_url', ''), action_label=safe_ctx.get('action_label', ''), source_app=source_app, is_system=True, is_single_claim=rule.single_claim, ) # Collecter tous les utilisateurs destinataires : # - membres (CompanyMember.user) des équipes CompanyTeam configurées # - utilisateurs individuels configurés from contracts.models import CompanyMember recipient_user_ids = set() for team in rule.recipient_teams.all(): team_user_ids = CompanyMember.objects.filter( teams=team, user__isnull=False ).values_list('user_id', flat=True) recipient_user_ids.update(team_user_ids) for u in rule.recipient_users.all(): recipient_user_ids.add(u.pk) if recipient_user_ids: NotificationRecipient.objects.bulk_create( [NotificationRecipient(notification=notification, user_id=uid) for uid in recipient_user_ids], ignore_conflicts=True, ) def can_post_observation_message(user, observation): """ Retourne True si l'utilisateur peut poster un message dans la discussion d'une observation. Critères : - C'est le créateur de l'observation, OU - Il est admin/superuser/operator, OU - Il peut traiter l'observation (process_thematic_ids), OU - Il est membre d'une équipe gestionnaire d'assets (is_assets_manager=True) qui gère la catégorie de l'observation. Cohérence : tout utilisateur disposant des actions peut aussi poster un message. """ if not user or not user.is_authenticated: return False if user.is_superuser: return True user_config = getattr(user, 'config', None) if user_config and hasattr(user_config, 'has_role'): if user_config.has_role('admin') or user_config.has_role('operator'): return True if observation.created_by_id and observation.created_by_id == user.pk: return True # Check if user can process this observation via process_thematic_ids try: from observations.permissions import get_observation_access_context access_context = get_observation_access_context(user, user_config=user_config) if access_context and access_context.can_process_observation(observation): return True except Exception: pass if observation.category_id: try: from contracts.models import CompanyMember return CompanyMember.objects.filter( user=user, teams__is_assets_manager=True, teams__asset_categories=observation.category, ).exists() except Exception: pass return False def get_or_create_observation_discussion(observation): """ Trouve ou crée la discussion associée à une observation. Les participants sont : le créateur de l'observation + les membres des équipes gestionnaires d'assets (is_assets_manager=True) pour la catégorie concernée. Retourne (discussion, created). """ from .models import Discussion from django.contrib.contenttypes.models import ContentType as CT from django.urls import reverse ct = CT.objects.get_for_model(observation) discussion = Discussion.objects.filter( content_type=ct, object_id=observation.pk ).first() if discussion: return discussion, False try: obj_url = reverse('observations:observation_detail', args=[observation.pk]) except Exception: obj_url = '' obj_label = f"Observation {observation.code}" discussion = Discussion.objects.create( title=obj_label, created_by=observation.created_by, content_type=ct, object_id=observation.pk, object_url=obj_url, object_label=obj_label, ) participants = set() if observation.created_by_id: participants.add(observation.created_by) if observation.category_id: try: from contracts.models import CompanyMember from django.contrib.auth.models import User as AuthUser manager_users = AuthUser.objects.filter( companymember__teams__is_assets_manager=True, companymember__teams__asset_categories=observation.category, ).distinct() participants.update(manager_users) except Exception: pass discussion.participants.set(participants) return discussion, True def can_post_intervention_message(user, intervention): """ Retourne True si l'utilisateur peut poster un message dans la discussion d'une intervention. Critères : - Il est superuser, OU - Il est admin/operator, OU - Il peut voir l'intervention (can_view_intervention). """ if not user or not user.is_authenticated: return False if user.is_superuser: return True user_config = getattr(user, 'config', None) if user_config and hasattr(user_config, 'has_role'): if user_config.has_role('admin') or user_config.has_role('operator'): return True try: from interventions.permissions import can_view_intervention return can_view_intervention(user, intervention) except Exception: return False def get_or_create_intervention_discussion(intervention): """ Trouve ou crée la discussion associée à une intervention. Les participants initiaux sont : le créateur de l'intervention et le gestionnaire (si présents). Les autres intervenants sont ajoutés lorsqu'ils répondent ou sont mentionnés. Retourne (discussion, created). """ from .models import Discussion from django.contrib.contenttypes.models import ContentType as CT from django.urls import reverse ct = CT.objects.get_for_model(intervention) discussion = Discussion.objects.filter( content_type=ct, object_id=intervention.pk ).first() if discussion: return discussion, False try: obj_url = reverse('interventions:intervention_detail', args=[intervention.pk]) except Exception: obj_url = '' obj_label = f"Intervention {intervention.code}" discussion = Discussion.objects.create( title=obj_label, created_by=intervention.created_by, content_type=ct, object_id=intervention.pk, object_url=obj_url, object_label=obj_label, ) participants = set() if hasattr(intervention, 'intervention_manager') and intervention.intervention_manager: participants.add(intervention.intervention_manager) if participants: discussion.participants.set(participants) return discussion, True def can_post_preparation_order_message(user, preparation_order): """ Retourne True si l'utilisateur peut poster un message dans la discussion d'un bon de préparation. Critères : - Il est superuser, OU - Il est admin/operator/stock_manager, OU - Il a la permission can_manage_preparation_orders_any, OU - L'intervention liée peut être vue (can_view_intervention). """ if not user or not user.is_authenticated: return False if user.is_superuser: return True user_config = getattr(user, 'config', None) if user_config and hasattr(user_config, 'has_role'): if user_config.has_role('admin') or user_config.has_role('operator') or user_config.has_role('stock_manager'): return True if hasattr(user_config, 'can_manage_preparation_orders_any') and user_config.can_manage_preparation_orders_any(): return True if preparation_order and preparation_order.intervention: try: from interventions.permissions import can_view_intervention if can_view_intervention(user, preparation_order.intervention): return True except Exception: pass return False def get_or_create_preparation_order_discussion(preparation_order): """ Trouve ou crée la discussion associée à un bon de préparation. """ from .models import Discussion from django.contrib.contenttypes.models import ContentType as CT from django.urls import reverse ct = CT.objects.get_for_model(preparation_order) discussion = Discussion.objects.filter( content_type=ct, object_id=preparation_order.pk ).first() if discussion: return discussion, False try: obj_url = reverse('stock:preparation_order_detail', args=[preparation_order.pk]) except Exception: obj_url = '' obj_label = f"Bon de préparation {preparation_order.order_number}" created_by_user = None if preparation_order.intervention and hasattr(preparation_order.intervention, 'created_by'): created_by_user = preparation_order.intervention.created_by discussion = Discussion.objects.create( title=obj_label, created_by=created_by_user, content_type=ct, object_id=preparation_order.pk, object_url=obj_url, object_label=obj_label, ) participants = set() if preparation_order.intervention: interv = preparation_order.intervention if hasattr(interv, 'intervention_manager') and interv.intervention_manager: participants.add(interv.intervention_manager) if participants: discussion.participants.set(participants) return discussion, True def cleanup_linked_discussion_participants(discussion): """ Conserve uniquement comme participants le gestionnaire d'intervention, ainsi que les auteurs de messages et personnes mentionnées. Ne conserve pas l'auteur/créateur de l'intervention par défaut. """ if not discussion.content_type_id: return obj = discussion.content_object if not obj: return keep_user_ids = set() if hasattr(obj, 'intervention_manager_id') and obj.intervention_manager_id: keep_user_ids.add(obj.intervention_manager_id) msg_author_ids = discussion.messages.values_list('author_id', flat=True) keep_user_ids.update(msg_author_ids) for msg in discussion.messages.all(): for m in _MENTION_PATTERN.finditer(msg.content or ''): if m.group(2) == 'u': keep_user_ids.add(int(m.group(3))) discussion.participants.set(keep_user_ids)