from django.shortcuts import render, redirect, get_object_or_404 from django.contrib.auth.decorators import login_required from django.contrib.auth.models import User from django.db.models import Q from django.http import JsonResponse from django.utils import timezone from django.contrib import messages from django.utils.translation import gettext_lazy as _ from django.views.decorators.http import require_POST, require_GET from urllib.parse import urlsplit, urlunsplit from .models import ( Notification, NotificationAttachment, NotificationRecipient, TeamNotificationTarget, TeamNotificationRead, Team, Discussion, Message, MessageAttachment, ) from .services import ( get_user_notifications, get_unread_count, notify_users, notify_team, post_message, mark_discussion_seen, ) from .forms import NotificationCreateForm, TeamForm # ────────────────────────────────────────────────────────────────────────────── # Inbox # ────────────────────────────────────────────────────────────────────────────── def _mobileize_action_url(action_url): """Convertit certaines URLs desktop en URLs mobile quand un équivalent existe.""" if not action_url or not isinstance(action_url, str): return action_url parsed = urlsplit(action_url) path = parsed.path or action_url normalized_path = path if path.startswith('/') else f'/{path}' if normalized_path.startswith('/mobile/'): return action_url prefixes = { '/interventions/': '/mobile/interventions/', '/observations/': '/mobile/observations/', '/observation/': '/mobile/observations/', '/projects/': '/mobile/projects/', '/controls/': '/mobile/controls/', '/stock/': '/mobile/stock/', '/assets/': '/mobile/assets/', '/notifications/': '/mobile/notifications/', } for desktop_prefix, mobile_prefix in prefixes.items(): desktop_no_slash = desktop_prefix.rstrip('/') if normalized_path == desktop_no_slash: mobile_path = mobile_prefix.rstrip('/') + '/' elif normalized_path.startswith(desktop_prefix): mobile_path = mobile_prefix + normalized_path[len(desktop_prefix):] else: continue if parsed.scheme or parsed.netloc: return urlunsplit((parsed.scheme, parsed.netloc, mobile_path, parsed.query, parsed.fragment)) return urlunsplit(('', '', mobile_path, parsed.query, parsed.fragment)) return action_url def _translate_notification_data(notification): from django.utils.translation import gettext as _gt from interventions.models import STATUS_CHOICES import re # 1. Translate action_label action_label = getattr(notification, 'action_label', None) if action_label: if action_label in ["Voir l'intervention", "View intervention", "Interventie bekijken"]: notification.action_label = _gt("Voir l'intervention") else: notification.action_label = _gt(action_label) # 2. Translate title title = getattr(notification, 'title', None) if title: code = None for prefix in ["Mise à jour de l'intervention ", "Update of intervention ", "Update van de interventie "]: if title.startswith(prefix): code = title[len(prefix):] break if code: notification.title = _gt("Mise à jour de l'intervention %(code)s") % {'code': code} else: notification.title = _gt(title) # 3. Translate content content = getattr(notification, 'content', None) if content: # Match French pattern m_fr = re.match(r"^L'intervention « (.*) » est passée du statut « (.*) » à « (.*) »\.$", content) # Match English pattern m_en = re.match(r"^Intervention « (.*) » status changed from « (.*) » to « (.*) »\.$", content) # Match Dutch pattern m_nl = re.match(r"^Interventie « (.*) » is veranderd van status « (.*) » naar « (.*) »\.$", content) matched = m_fr or m_en or m_nl if matched: int_title = matched.group(1) old_label = matched.group(2) new_label = matched.group(3) # Map labels to status choices keys STATUS_LABEL_MAP = { # French 'en préparation': 'in_preparation', 'a approuver': 'to_be_approved', 'a planifier': 'to_be_planned', 'a traiter': 'to_be_processed', 'pris en charge': 'assigned', 'démarré': 'in_progress', 'en pause': 'on_pause', 'terminé': 'finished', 'traité': 'processed', 'a corriger': 'to_be_corrected', 'corrigé': 'corrected', 'validé': 'validated', 'comptabilisé': 'invoiced', 'finalisé': 'closed', 'annulé': 'canceled', # English 'in preparation': 'in_preparation', 'to be approved': 'to_be_approved', 'to schedule': 'to_be_planned', 'to be planned': 'to_be_planned', 'to process': 'to_be_processed', 'to be processed': 'to_be_processed', 'taken over': 'assigned', 'started': 'in_progress', 'on pause': 'on_pause', 'finished': 'finished', 'processed': 'processed', 'to be corrected': 'to_be_corrected', 'corrected': 'corrected', 'validated': 'validated', 'invoiced': 'invoiced', 'closed': 'closed', 'canceled': 'canceled', # Dutch 'in voorbereiding': 'in_preparation', 'goed te keuren': 'to_be_approved', 'te plannen': 'to_be_planned', 'te behandelen': 'to_be_processed', 'in behandeling': 'assigned', 'gestart': 'in_progress', 'gepauzeerd': 'on_pause', 'beëindigd': 'finished', 'verwerkt': 'processed', 'te corrigeren': 'to_be_corrected', 'gecorrigeerd': 'corrected', 'gevalideerd': 'validated', 'geboekt': 'invoiced', 'afgerond': 'closed', 'geannuleerd': 'canceled', } old_key = STATUS_LABEL_MAP.get(old_label.strip().lower(), old_label) new_key = STATUS_LABEL_MAP.get(new_label.strip().lower(), new_label) old_trans = _gt(STATUS_CHOICES.get(old_key, old_key)) if old_key in STATUS_CHOICES else _gt(old_label) new_trans = _gt(STATUS_CHOICES.get(new_key, new_key)) if new_key in STATUS_CHOICES else _gt(new_label) notification.content = _gt("L'intervention « %(title)s » est passée du statut « %(old)s » à « %(new)s ».") % { 'title': int_title, 'old': old_trans, 'new': new_trans } def _get_inbox_context(request): """Construit le contexte de la boîte de réception (desktop/mobile).""" show_archived = request.GET.get('archived') == '1' notification_type = request.GET.get('type', '') all_notifications = get_user_notifications(request.user, include_archived=show_archived) if notification_type: all_notifications = [ n for n in all_notifications if n['notification'].notification_type == notification_type ] for item in all_notifications: notification = item.get('notification') _translate_notification_data(notification) action_url = getattr(notification, 'action_url', None) item['mobile_action_url'] = _mobileize_action_url(action_url) unread_count = sum(1 for n in all_notifications if not n['is_read'] and not n['is_archived']) return { 'notifications': all_notifications, 'unread_count': unread_count, 'show_archived': show_archived, 'filter_type': notification_type, 'notification_types': Notification.TYPES, } @login_required def inbox(request): """Boîte de réception des notifications de l'utilisateur connecté.""" return render(request, 'notifications/inbox.html', _get_inbox_context(request)) @login_required def inbox_mobile(request): """Boîte de réception des notifications optimisée pour le mobile.""" return render(request, 'notifications/inbox_mobile.html', _get_inbox_context(request)) # ────────────────────────────────────────────────────────────────────────────── # Mark read / archive # ────────────────────────────────────────────────────────────────────────────── @login_required @require_POST def mark_read(request, pk): """Marque une notification comme lue (personnelle ou d'équipe).""" recipient_type = request.POST.get('type', 'personal') if recipient_type == 'personal': nr = get_object_or_404(NotificationRecipient, pk=pk, user=request.user) nr.mark_read() elif recipient_type == 'team': tt = get_object_or_404(TeamNotificationTarget, pk=pk) # Verify user belongs to the team if not tt.team.members.filter(pk=request.user.pk).exists(): return JsonResponse({'error': 'forbidden'}, status=403) TeamNotificationRead.objects.get_or_create( team_notification=tt, user=request.user, ) if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return JsonResponse({'ok': True, 'unread_count': get_unread_count(request.user)}) return redirect('notifications:inbox') @login_required @require_POST def mark_all_read(request): """Marque toutes les notifications (personnelles + équipes) comme lues.""" # Personal NotificationRecipient.objects.filter( user=request.user, read_at__isnull=True, ).update(read_at=timezone.now()) # Team user_team_ids = request.user.notification_teams.values_list('id', flat=True) team_targets = TeamNotificationTarget.objects.filter(team_id__in=user_team_ids) already_read = TeamNotificationRead.objects.filter( user=request.user, ).values_list('team_notification_id', flat=True) unread_team = team_targets.exclude(pk__in=already_read) TeamNotificationRead.objects.bulk_create([ TeamNotificationRead(team_notification=tt, user=request.user) for tt in unread_team ], ignore_conflicts=True) if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return JsonResponse({'ok': True, 'unread_count': 0}) return redirect('notifications:inbox') @login_required @require_POST def archive_notification(request, pk): """Archive une notification personnelle.""" nr = get_object_or_404(NotificationRecipient, pk=pk, user=request.user) nr.archive() if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return JsonResponse({'ok': True, 'unread_count': get_unread_count(request.user)}) return redirect('notifications:inbox') # ────────────────────────────────────────────────────────────────────────────── # API # ────────────────────────────────────────────────────────────────────────────── @login_required def api_unread_count(request): """Endpoint JSON pour le badge de la cloche.""" return JsonResponse({'unread_count': get_unread_count(request.user)}) @login_required def api_recent(request): """Retourne les 5 dernières notifications non lues (pour le dropdown).""" notifs = get_user_notifications(request.user, include_archived=False) unread = [n for n in notifs if not n['is_read']][:5] for n in unread: _translate_notification_data(n['notification']) data = [ { 'id': n['recipient_id'], 'type': n['recipient_type'], 'title': n['notification'].title, 'content': n['notification'].content[:120], 'notification_type': n['notification'].notification_type, 'icon': n['notification'].type_icon, 'color': n['notification'].type_color, 'action_url': n['notification'].action_url, 'action_label': n['notification'].action_label, 'created_at': n['notification'].created_at.isoformat(), 'team_name': n['team_name'], } for n in unread ] return JsonResponse({ 'notifications': data, 'unread_count': get_unread_count(request.user), }) # ────────────────────────────────────────────────────────────────────────────── # Create (manual notifications — admins/managers) # ────────────────────────────────────────────────────────────────────────────── @login_required def create_notification(request): """Crée une notification manuelle.""" # Only admins/managers can create manual notifications user_config = getattr(request.user, 'config', None) if user_config and not ( request.user.is_superuser or user_config.has_role('admin') or user_config.has_role('manager') ): messages.error(request, _("Vous n'avez pas les droits pour créer une notification.")) return redirect('notifications:inbox') if request.method == 'POST': form = NotificationCreateForm(request.POST, request.FILES) if form.is_valid(): cd = form.cleaned_data create_kwargs = dict( title=cd['title'], content=cd['content'], notification_type=cd['notification_type'], action_url=cd['action_url'], action_label=cd['action_label'], source_app='manual', sender=request.user, expires_at=cd.get('expires_at'), ) notification = None if cd['recipient_users']: notification = notify_users(recipients=cd['recipient_users'], **create_kwargs) for team in cd['recipient_teams']: result = notify_team(team=team, **create_kwargs) if notification is None: notification = result.notification # Save attachments if notification: for f in request.FILES.getlist('attachments'): NotificationAttachment.objects.create( notification=notification, file=f, name=f.name, uploaded_by=request.user, ) messages.success(request, _("Notification envoyée avec succès.")) return redirect('notifications:inbox') else: form = NotificationCreateForm() return render(request, 'notifications/create.html', {'form': form}) # ────────────────────────────────────────────────────────────────────────────── # Teams # ────────────────────────────────────────────────────────────────────────────── @login_required def team_list(request): """Liste des équipes.""" user_config = getattr(request.user, 'config', None) is_admin = request.user.is_superuser or ( user_config and (user_config.has_role('admin') or user_config.has_role('manager')) ) if is_admin: teams = Team.objects.prefetch_related('members').all() else: teams = request.user.notification_teams.prefetch_related('members').all() return render(request, 'notifications/team_list.html', { 'teams': teams, 'is_admin': is_admin, }) @login_required def team_create(request): """Crée une équipe.""" user_config = getattr(request.user, 'config', None) if not (request.user.is_superuser or (user_config and ( user_config.has_role('admin') or user_config.has_role('manager') ))): messages.error(request, _("Vous n'avez pas les droits pour créer une équipe.")) return redirect('notifications:team_list') if request.method == 'POST': form = TeamForm(request.POST) if form.is_valid(): team = form.save(commit=False) team.created_by = request.user team.save() form.save_m2m() messages.success(request, _("Équipe créée avec succès.")) return redirect('notifications:team_list') else: form = TeamForm() return render(request, 'notifications/team_form.html', {'form': form, 'action': _('Créer')}) @login_required def team_edit(request, pk): """Modifie une équipe.""" team = get_object_or_404(Team, pk=pk) user_config = getattr(request.user, 'config', None) if not (request.user.is_superuser or (user_config and ( user_config.has_role('admin') or user_config.has_role('manager') ))): messages.error(request, _("Vous n'avez pas les droits pour modifier une équipe.")) return redirect('notifications:team_list') if request.method == 'POST': form = TeamForm(request.POST, instance=team) if form.is_valid(): form.save() messages.success(request, _("Équipe mise à jour.")) return redirect('notifications:team_list') else: form = TeamForm(instance=team) return render(request, 'notifications/team_form.html', {'form': form, 'team': team, 'action': _('Modifier')}) @login_required @require_POST def team_delete(request, pk): """Supprime une équipe.""" team = get_object_or_404(Team, pk=pk) user_config = getattr(request.user, 'config', None) if not (request.user.is_superuser or (user_config and user_config.has_role('admin'))): messages.error(request, _("Vous n'avez pas les droits pour supprimer une équipe.")) return redirect('notifications:team_list') team.delete() messages.success(request, _("Équipe supprimée.")) return redirect('notifications:team_list') # ───────────────────────────────────────────────────────────────────────────── # Discussions # ───────────────────────────────────────────────────────────────────────────── @login_required def discussion_list(request): """ Liste les discussions dont l'utilisateur est l'auteur (créateur de la discussion ou auteur d'un message) ou pour lesquelles il (ou une de ses équipes) a été explicitement taggé. """ user = request.user user_team_ids = list(user.notification_teams.values_list('id', flat=True)) mention_q = Q(messages__content__icontains=f'(u:{user.pk})') for tid in user_team_ids: mention_q |= Q(messages__content__icontains=f'(t:{tid})') discussion_filter = ( Q(messages__author=user, messages__is_deleted=False) | Q(notifications__recipients__user=user) | Q(notifications__team_targets__team_id__in=user_team_ids) | (mention_q & Q(messages__is_deleted=False)) | (Q(created_by=user) & Q(content_type__isnull=True)) ) discussions = ( Discussion.objects.filter(discussion_filter) .distinct() .prefetch_related('participants', 'messages') .select_related('created_by') .order_by('-updated_at') ) # Enrichir avec le compteur de non-lus et le dernier message items = [] for d in discussions: last_msg = d.messages.filter(is_deleted=False).order_by('-created_at').first() items.append({ 'discussion': d, 'last_message': last_msg, 'unread_count': d.unread_count_for(request.user), }) return render(request, 'notifications/discussion_list.html', {'items': items}) @login_required def discussion_detail(request, pk): """Affiche le fil de discussion et permet de répondre.""" discussion = get_object_or_404( Discussion.objects.prefetch_related('participants', 'messages__attachments'), pk=pk, ) # Vérifier que l'utilisateur est participant (ou admin / accès à l'objet lié) user_config = getattr(request.user, 'config', None) is_admin = request.user.is_superuser or ( user_config and (user_config.has_role('admin') or user_config.has_role('manager') or user_config.has_role('operator') or user_config.has_role('stock_manager')) ) is_participant = discussion.participants.filter(pk=request.user.pk).exists() has_access = is_admin or is_participant if not has_access and discussion.content_type_id and discussion.object_id: obj = discussion.content_object if obj: from stock.models import PreparationOrder from interventions.models import Intervention from observations.models import Observation if isinstance(obj, Intervention): from notifications.services import can_post_intervention_message has_access = can_post_intervention_message(request.user, obj) elif isinstance(obj, PreparationOrder): from notifications.services import can_post_preparation_order_message has_access = can_post_preparation_order_message(request.user, obj) elif isinstance(obj, Observation): from notifications.services import can_post_observation_message has_access = can_post_observation_message(request.user, obj) if not has_access: messages.error(request, _("Vous n'êtes pas participant de cette discussion.")) return redirect('notifications:discussion_list') if request.method == 'POST': content = request.POST.get('content', '').strip() if content: files = request.FILES.getlist('attachments') post_message( discussion=discussion, author=request.user, content=content, files=files or None, ) messages.success(request, _("Message envoyé.")) return redirect('notifications:discussion_detail', pk=pk) # Marquer les notifications de l'utilisateur courant comme lues mark_discussion_seen(discussion, user=request.user) thread_messages = discussion.messages.filter(is_deleted=False).prefetch_related('attachments').order_by('created_at') return render(request, 'notifications/discussion_detail.html', { 'discussion': discussion, 'thread_messages': thread_messages, }) @login_required def discussion_create(request): """ Crée une nouvelle discussion indépendante (sans objet lié). Poste le premier message et notifie les personnes/équipes mentionnées. """ if request.method == 'POST': title = request.POST.get('title', '').strip() content = request.POST.get('content', '').strip() files = request.FILES.getlist('attachments') if content: discussion = Discussion.objects.create( title=title or f"Discussion de {request.user.get_full_name() or request.user.username}", created_by=request.user, ) discussion.participants.add(request.user) post_message( discussion=discussion, author=request.user, content=content, files=files or None, ) messages.success(request, _("Discussion créée avec succès.")) return redirect('notifications:discussion_detail', pk=discussion.pk) else: messages.error(request, _("Le contenu du message ne peut pas être vide.")) return redirect('notifications:discussion_list') @login_required def notification_settings(request): """Permet à l'utilisateur de configurer ses rappels par e-mail pour les notifications non lues.""" from common.models import UserConfig from .forms import NotificationReminderSettingsForm config, _created = UserConfig.objects.get_or_create(user=request.user) if request.method == 'POST': form = NotificationReminderSettingsForm(request.POST, instance=config) if form.is_valid(): form.save() messages.success(request, _("Vos préférences de rappel par e-mail ont été enregistrées.")) return redirect('notifications:settings') else: form = NotificationReminderSettingsForm(instance=config) return render(request, 'notifications/notification_settings.html', { 'form': form, 'config': config, }) # ────────────────────────────────────────────────────────────────────────────── # Mention search API # ────────────────────────────────────────────────────────────────────────────── @login_required @require_GET def api_mention_search(request): """ Retourne les utilisateurs et équipes correspondant à la requête, pour l'autocomplétion @mention dans les discussions. Réponse JSON: { results: [{type, id, display}, ...] } """ q = (request.GET.get('q') or '').strip() results = [] # Users — recherche sur username, prénom, nom user_qs = User.objects.filter( Q(username__icontains=q) | Q(first_name__icontains=q) | Q(last_name__icontains=q), is_active=True, ).exclude(pk=request.user.pk).select_related('config__company').order_by('first_name', 'last_name')[:10] for u in user_qs: first = u.first_name or '' last = u.last_name or '' initial = (last[0].upper() + '.') if last else '' display = f"{first} {initial}".strip() or u.username try: company_name = u.config.company.name if u.config.company_id else '' except Exception: company_name = '' if company_name: display = f"{display} ({company_name})" results.append({'type': 'user', 'id': u.pk, 'display': display}) # Teams team_qs = Team.objects.filter(name__icontains=q).order_by('name')[:5] for t in team_qs: results.append({'type': 'team', 'id': t.pk, 'display': t.name}) return JsonResponse({'results': results})