ISO-27001-Risk-Management/risks/utils.py

from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.mail import send_mail
from django.utils.timezone import now
from django.utils.translation import gettext_lazy as _
from .models import AuditLog, Notification,NotificationRule, NotificationKind, Risk, ResidualRisk
from typing import Iterable, Optional

User = get_user_model()

def model_diff(old, new, fields=None):
    """
    Compare two model instances and return a dict of changed fields.
    - old: previous model instance (from DB)
    - new: updated model instance (unsaved)
    - fields: optional list of fields to check
    """
    changes = {}
    opts = new._meta

    if fields is None:
        fields = [f.name for f in opts.fields]

    for field_name in fields:
        old_value = getattr(old, field_name, None)
        new_value = getattr(new, field_name, None)

        if old_value != new_value:
            changes[field_name] = {"old": old_value, "new": new_value}

    return changes

def check_risk_followups():
    """
    Check if follow ups need attention and create notifications.
    Ensures no duplicate notifications per risk per day
    """
    today = now().date()
    risks = Risk.objects.filter(follow_up__lte=today).select_related("owner")

    for risk in risks:
        # Risk-Status auf review_required setzen (nicht überschreiben, wenn bereits closed)
        if risk.status != "closed" and risk.status != "review_required":
            Risk.objects.filter(pk=risk.pk).update(status="review_required")

        # ResidualRisk-Objekt sicherstellen und Review-Flag setzen
        resid, created = ResidualRisk.objects.get_or_create(risk=risk)
        if not resid.review_required:
            resid.review_required = True
            resid.save()

        # Notification an Stakeholder
        message = _("Follow-up reached: review required for risk '{t}'").format(t=risk.title)
        notification, created = Notification.objects.get_or_create(
            user=risk.owner,
            message=message,
            defaults={"read": False, "sent": False},
        )
        if created:
            AuditLog.objects.create(
                user=None, action="create", model="Notification", object_id=notification.pk,
                changes={"message": notification.message, "user": risk.owner.username if risk.owner else None},
            )

        notify_event(
            NotificationKind.RISK_REVIEW_REQUIRED,
            message=_("Follow-up reached: review required for risk '{t}'").format(t=risk.title),
            users=[risk.owner] if risk.owner_id else None,
        )

def _split_emails(value: str) -> list[str]:
    if not value:
        return []
    raw = value.replace("\n", ",").split(",")
    return [e.strip() for e in raw if "@" in e and e.strip()]

def notify_event(kind: str, *, message: str, users: Optional[Iterable[User]] = None):
    """
    Generates in-app notifications and/or emails depending on the rule.
    - users: Basic recipients (owner/responsible/reporter) – can be None.
    - staff/extra recipients are added from the rule.
    """
    rule = NotificationRule.objects.filter(kind=kind).first()

    # Fallback: without rule → only in-app
    enabled_in_app = True
    enabled_email = False
    to_staff = False
    extra_emails = []

    recipients_users = set()

    if users:
        for u in users:
            if u and getattr(u, "is_active", False):
                recipients_users.add(u)

    if rule:
        enabled_in_app = rule.enabled_in_app
        enabled_email = rule.enabled_email
        if rule.to_staff:
            to_staff = True
        extra_emails = _split_emails(rule.extra_recipients)

    if to_staff:
        for u in User.objects.filter(is_staff=True, is_active=True):
            recipients_users.add(u)

    # In-App
    if enabled_in_app:
        for u in recipients_users:
            Notification.objects.create(user=u, message=message)

    # E-Mail
    if enabled_email:
        emails = [u.email for u in recipients_users if u and u.email] + extra_emails
        emails = list(dict.fromkeys(emails))  # de-dupe, Reihenfolge erhalten
        if emails:
            subject = _("Notification")
            body = message
            send_mail(
                subject,
                body,
                getattr(settings, "DEFAULT_FROM_EMAIL", "webmaster@localhost"),
                emails,
                fail_silently=True,  # im Zweifel nicht crashen
            )