import json from pathlib import Path from tempfile import NamedTemporaryFile from django.conf import settings from django.contrib import messages from django.shortcuts import redirect, render from django.utils import timezone from django.utils.translation import gettext as _ from .models import NotificationRule, NotificationTemplate, SystemEmailConfig, UserNotification, UserProfile, WorkflowConfig from .notifications import notify_user from .services import get_email_test_redirect, is_email_test_mode, is_nextcloud_enabled, upload_to_nextcloud from .branding import get_default_notification_templates from .emailing import send_system_email def integrations_setup_page_impl(request): config, _ = WorkflowConfig.objects.get_or_create(name='Default') kind = (request.GET.get('kind') or 'nextcloud').strip().lower() if kind not in {'nextcloud', 'mail', 'emails', 'rules', 'backup'}: kind = 'nextcloud' templates = list(NotificationTemplate.objects.all().order_by('key')) system_email_config = ( SystemEmailConfig.objects.filter(is_active=True).order_by('-updated_at').first() or SystemEmailConfig.objects.filter(name='Default SMTP').first() ) return render( request, 'workflows/integrations_setup.html', { 'workflow_config': config, 'system_email_config': system_email_config, 'nextcloud_enabled': is_nextcloud_enabled(), 'email_test_mode': is_email_test_mode(), 'kind': kind, 'templates': templates, 'notification_rules': NotificationRule.objects.all().order_by('event_type', 'sort_order', 'id'), 'rule_event_choices': NotificationRule.EVENT_CHOICES, 'rule_operator_choices': NotificationRule.OPERATOR_CHOICES, 'template_choices': NotificationTemplate.TEMPLATE_CHOICES, 'remote_backup_target_choices': WorkflowConfig.REMOTE_BACKUP_TARGET_CHOICES, }, ) def send_test_email_impl(request, *, audit_fn, redirect_back_fn): mode = 'TEST_MODE_ON' if is_email_test_mode() else 'TEST_MODE_OFF' redirect_email = get_email_test_redirect() try: send_system_email( subject=f'SMTP test from onboarding/offboarding v2 ({mode})', body=( 'This is a test email. If you see this, SMTP is configured correctly.\n' f'EMAIL_TEST_MODE={is_email_test_mode()}\n' f'EMAIL_TEST_REDIRECT={redirect_email}\n' ), to=[settings.TEST_NOTIFICATION_EMAIL], ) audit_fn(request, 'smtp_test_sent', target_type='system_email', target_label=settings.TEST_NOTIFICATION_EMAIL, details={'email_test_mode': is_email_test_mode()}) notify_user( user=request.user, title=_('SMTP-Test erfolgreich'), body=_('Die SMTP-Testmail wurde erfolgreich gesendet.'), level=UserNotification.LEVEL_SUCCESS, link_url='/admin-tools/integrations/', event_key=UserProfile.NOTIFICATION_SYSTEM_ALERTS, ) messages.success(request, f'SMTP-Testmail wurde gesendet ({mode}).') except Exception as exc: notify_user( user=request.user, title=_('SMTP-Test fehlgeschlagen'), body=str(exc), level=UserNotification.LEVEL_ERROR, link_url='/admin-tools/integrations/', event_key=UserProfile.NOTIFICATION_SYSTEM_ALERTS, ) messages.error(request, _('SMTP-Testmail konnte nicht gesendet werden: %(error)s') % {'error': exc}) return redirect_back_fn(request, 'home') def nextcloud_test_upload_impl(request, *, audit_fn, redirect_back_fn): filename = f"nextcloud_test_{timezone.now().strftime('%Y%m%d_%H%M%S')}.txt" content = ( "Nextcloud test upload from onboarding/offboarding system.\n" f"Time: {timezone.now().isoformat()}\n" f"User: {request.user.username}\n" ) temp_path = None try: with NamedTemporaryFile('w', suffix='.txt', delete=False, encoding='utf-8') as tf: tf.write(content) temp_path = Path(tf.name) ok = upload_to_nextcloud(temp_path, filename) if ok: audit_fn(request, 'nextcloud_test_upload', target_type='nextcloud', target_label=filename, details={'result': 'success'}) notify_user( user=request.user, title=_('Nextcloud-Test erfolgreich'), body=_('Der Testupload nach Nextcloud war erfolgreich.'), level=UserNotification.LEVEL_SUCCESS, link_url='/admin-tools/integrations/', event_key=UserProfile.NOTIFICATION_SYSTEM_ALERTS, ) messages.success(request, f'Nextcloud-Testupload erfolgreich: {filename}') else: audit_fn(request, 'nextcloud_test_upload', target_type='nextcloud', target_label=filename, details={'result': 'error'}) notify_user( user=request.user, title=_('Nextcloud-Test fehlgeschlagen'), body=_('Der Testupload nach Nextcloud ist fehlgeschlagen.'), level=UserNotification.LEVEL_ERROR, link_url='/admin-tools/integrations/', event_key=UserProfile.NOTIFICATION_SYSTEM_ALERTS, ) messages.error(request, 'Nextcloud-Testupload fehlgeschlagen. Bitte Konfiguration prüfen.') except Exception as exc: notify_user( user=request.user, title=_('Nextcloud-Test fehlgeschlagen'), body=str(exc), level=UserNotification.LEVEL_ERROR, link_url='/admin-tools/integrations/', event_key=UserProfile.NOTIFICATION_SYSTEM_ALERTS, ) messages.error(request, f'Nextcloud-Testupload fehlgeschlagen: {exc}') finally: if temp_path and temp_path.exists(): temp_path.unlink(missing_ok=True) return redirect_back_fn(request, 'home') def toggle_nextcloud_enabled_impl(request, *, audit_fn, redirect_back_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') currently_enabled = is_nextcloud_enabled() config.nextcloud_enabled_override = not currently_enabled config.save(update_fields=['nextcloud_enabled_override']) audit_fn(request, 'nextcloud_mode_toggled', target_type='workflow_config', target_label='nextcloud', details={'enabled': config.nextcloud_enabled_override}) state = 'aktiviert' if config.nextcloud_enabled_override else 'deaktiviert' messages.success(request, f'Nextcloud Upload wurde {state}.') return redirect_back_fn(request, 'home') def toggle_email_mode_impl(request, *, audit_fn, redirect_back_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') currently_test_mode = is_email_test_mode() config.email_test_mode_override = not currently_test_mode config.save(update_fields=['email_test_mode_override']) audit_fn(request, 'email_mode_toggled', target_type='workflow_config', target_label='email_mode', details={'test_mode': config.email_test_mode_override}) state = 'Testmodus (Umleitung)' if config.email_test_mode_override else 'Produktionsmodus' messages.success(request, f'E-Mail-Modus wurde auf {state} gesetzt.') return redirect_back_fn(request, 'home') def save_integrations_settings_impl(request, *, audit_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') try: sync_interval = int(request.POST.get('sync_interval_seconds', config.sync_interval_seconds or 60)) smtp_port = int(request.POST.get('smtp_port', config.smtp_port or 465)) except ValueError: messages.error(request, 'Ungültige Zahl bei Sync-Intervall oder SMTP Port.') return redirect('home') config.nextcloud_base_url_override = request.POST.get('nextcloud_base_url_override', '').strip() config.nextcloud_username_override = request.POST.get('nextcloud_username_override', '').strip() config.nextcloud_directory_override = request.POST.get('nextcloud_directory_override', '').strip() config.sync_interval_seconds = max(10, sync_interval) config.imap_server = request.POST.get('imap_server', '').strip() config.mailbox = request.POST.get('mailbox', '').strip() or 'INBOX' config.smtp_server = request.POST.get('smtp_server', '').strip() config.smtp_port = max(1, smtp_port) config.email_account = request.POST.get('email_account', '').strip() config.smtp_use_ssl = request.POST.get('smtp_use_ssl') == 'on' config.smtp_use_tls = request.POST.get('smtp_use_tls') == 'on' nextcloud_password = request.POST.get('nextcloud_password_override', '').strip() if nextcloud_password: config.nextcloud_password_override = nextcloud_password email_password = request.POST.get('email_password', '').strip() if email_password: config.email_password = email_password config.save() audit_fn(request, 'integrations_saved', target_type='workflow_config', target_label='all_integrations') messages.success(request, 'Integrations-Einstellungen wurden gespeichert.') return redirect('home') def save_nextcloud_settings_impl(request, *, audit_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') try: sync_interval = int(request.POST.get('sync_interval_seconds', config.sync_interval_seconds or 60)) except ValueError: messages.error(request, 'Ungültige Zahl beim Sync-Intervall.') return redirect('home') config.nextcloud_base_url_override = request.POST.get('nextcloud_base_url_override', '').strip() config.nextcloud_username_override = request.POST.get('nextcloud_username_override', '').strip() config.nextcloud_directory_override = request.POST.get('nextcloud_directory_override', '').strip() config.sync_interval_seconds = max(10, sync_interval) nextcloud_password = request.POST.get('nextcloud_password_override', '').strip() if nextcloud_password: config.nextcloud_password_override = nextcloud_password config.save() audit_fn(request, 'nextcloud_settings_saved', target_type='workflow_config', target_label='nextcloud') messages.success(request, 'Nextcloud-Einstellungen wurden gespeichert.') return redirect('/admin-tools/integrations/?kind=nextcloud') def save_workflow_rules_impl(request, *, audit_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') try: handover_lead_days = int( request.POST.get( 'device_handover_lead_days', config.device_handover_lead_days or 5, ) ) except ValueError: messages.error(request, 'Ungültige Zahl beim Hardware-Vorlauf.') return redirect('/admin-tools/integrations/?kind=rules') config.device_handover_lead_days = max(0, handover_lead_days) config.save(update_fields=['device_handover_lead_days']) audit_fn( request, 'workflow_rules_saved', target_type='workflow_config', target_label='workflow_rules', details={ 'device_handover_lead_days': config.device_handover_lead_days, }, ) messages.success(request, 'Workflow-Regeln wurden gespeichert.') return redirect('/admin-tools/integrations/?kind=rules') def save_backup_settings_impl(request, *, audit_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') target_type = (request.POST.get('remote_backup_target_type') or config.remote_backup_target_type or 'nextcloud').strip().lower() if target_type not in {choice for choice, _ in WorkflowConfig.REMOTE_BACKUP_TARGET_CHOICES}: target_type = 'nextcloud' remote_backup_enabled = request.POST.get('remote_backup_enabled') == 'on' remote_backup_nextcloud_directory = request.POST.get('remote_backup_nextcloud_directory', '').strip() primary_nextcloud_directory = ( (config.nextcloud_directory_override or '').strip() or settings.NEXTCLOUD_DIRECTORY.strip() ).strip('/') if remote_backup_enabled and target_type == 'nextcloud': if not remote_backup_nextcloud_directory: messages.error(request, 'Bitte ein separates Nextcloud Backup-Verzeichnis angeben.') return redirect('/admin-tools/integrations/?kind=backup') if remote_backup_nextcloud_directory.strip('/') == primary_nextcloud_directory: messages.error(request, 'Das Backup-Verzeichnis muss vom normalen Nextcloud Dokumentenordner getrennt sein.') return redirect('/admin-tools/integrations/?kind=backup') config.remote_backup_enabled = remote_backup_enabled config.remote_backup_target_type = target_type config.remote_backup_nextcloud_directory = remote_backup_nextcloud_directory config.remote_backup_s3_bucket = request.POST.get('remote_backup_s3_bucket', '').strip() config.remote_backup_nfs_path = request.POST.get('remote_backup_nfs_path', '').strip() config.save( update_fields=[ 'device_handover_lead_days', 'remote_backup_enabled', 'remote_backup_target_type', 'remote_backup_nextcloud_directory', 'remote_backup_s3_bucket', 'remote_backup_nfs_path', ] ) audit_fn( request, 'backup_settings_saved', target_type='workflow_config', target_label='backup_settings', details={ 'remote_backup_enabled': config.remote_backup_enabled, 'remote_backup_target_type': config.remote_backup_target_type, 'remote_backup_nextcloud_directory': config.remote_backup_nextcloud_directory, }, ) messages.success(request, 'Backup-Einstellungen wurden gespeichert.') return redirect('/admin-tools/integrations/?kind=backup') def save_mail_settings_impl(request, *, audit_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') try: smtp_port = int(request.POST.get('smtp_port', config.smtp_port or 465)) except ValueError: messages.error(request, 'Ungültige Zahl beim SMTP Port.') return redirect('home') config.imap_server = request.POST.get('imap_server', '').strip() config.mailbox = request.POST.get('mailbox', '').strip() or 'INBOX' config.smtp_server = request.POST.get('smtp_server', '').strip() config.smtp_port = max(1, smtp_port) config.email_account = request.POST.get('email_account', '').strip() config.smtp_use_ssl = request.POST.get('smtp_use_ssl') == 'on' config.smtp_use_tls = request.POST.get('smtp_use_tls') == 'on' email_password = request.POST.get('email_password', '').strip() if email_password: config.email_password = email_password config.save() smtp_cfg, _ = SystemEmailConfig.objects.get_or_create(name='Default SMTP') SystemEmailConfig.objects.exclude(id=smtp_cfg.id).update(is_active=False) smtp_cfg.is_active = True smtp_cfg.host = config.smtp_server smtp_cfg.port = config.smtp_port smtp_cfg.username = config.email_account if email_password: smtp_cfg.password = email_password smtp_cfg.use_ssl = config.smtp_use_ssl smtp_cfg.use_tls = config.smtp_use_tls smtp_cfg.from_email = request.POST.get('from_email', '').strip() smtp_cfg.save() audit_fn(request, 'mail_settings_saved', target_type='workflow_config', target_label='mail') messages.success(request, 'Mail-Einstellungen wurden gespeichert.') return redirect('/admin-tools/integrations/?kind=mail') def save_email_routing_settings_impl(request, *, audit_fn): config, _ = WorkflowConfig.objects.get_or_create(name='Default') config.it_onboarding_email = request.POST.get('it_onboarding_email', '').strip() config.general_info_email = request.POST.get('general_info_email', '').strip() config.business_card_email = request.POST.get('business_card_email', '').strip() config.hr_works_email = request.POST.get('hr_works_email', '').strip() config.key_notification_email = request.POST.get('key_notification_email', '').strip() config.save( update_fields=[ 'it_onboarding_email', 'general_info_email', 'business_card_email', 'hr_works_email', 'key_notification_email', ] ) known_keys = {k for k, _ in NotificationTemplate.TEMPLATE_CHOICES} for key in known_keys: subject = request.POST.get(f'subject_{key}') body = request.POST.get(f'body_{key}') subject_en = request.POST.get(f'subject_en_{key}') body_en = request.POST.get(f'body_en_{key}') if subject is None and body is None and subject_en is None and body_en is None: continue subject = (subject or '').strip() body = (body or '').strip() subject_en = (subject_en or '').strip() body_en = (body_en or '').strip() if not subject and not body and not subject_en and not body_en: continue obj, _ = NotificationTemplate.objects.get_or_create( key=key, defaults={ 'subject_template': subject or f'[{key}]', 'body_template': body or '-', 'subject_template_en': subject_en, 'body_template_en': body_en, 'is_active': True, }, ) changed = [] if subject and obj.subject_template != subject: obj.subject_template = subject changed.append('subject_template') if body and obj.body_template != body: obj.body_template = body changed.append('body_template') if obj.subject_template_en != subject_en: obj.subject_template_en = subject_en changed.append('subject_template_en') if obj.body_template_en != body_en: obj.body_template_en = body_en changed.append('body_template_en') if not obj.is_active: obj.is_active = True changed.append('is_active') if changed: obj.save(update_fields=changed) audit_fn(request, 'email_routing_saved', target_type='workflow_config', target_label='email_routing') messages.success(request, 'E-Mail Routing und Vorlagen wurden gespeichert.') return redirect('/admin-tools/integrations/?kind=emails') def save_notification_rules_impl(request, *, audit_fn): rule_ids = request.POST.getlist('rule_ids') for position, raw_id in enumerate(rule_ids): rule = NotificationRule.objects.filter(id=raw_id).first() if not rule: continue if request.POST.get(f'delete_{rule.id}') == 'on': rule.delete() continue rule.name = request.POST.get(f'name_{rule.id}', '').strip() or rule.name event_type = request.POST.get(f'event_type_{rule.id}', '').strip() if event_type in {'onboarding', 'offboarding'}: rule.event_type = event_type operator = request.POST.get(f'operator_{rule.id}', '').strip() if operator in {x[0] for x in NotificationRule.OPERATOR_CHOICES}: rule.operator = operator rule.field_name = request.POST.get(f'field_name_{rule.id}', '').strip() rule.expected_value = request.POST.get(f'expected_value_{rule.id}', '').strip() rule.recipients = request.POST.get(f'recipients_{rule.id}', '').strip() rule.template_key = request.POST.get(f'template_key_{rule.id}', '').strip() rule.custom_subject = request.POST.get(f'custom_subject_{rule.id}', '').strip() rule.custom_body = request.POST.get(f'custom_body_{rule.id}', '').strip() rule.custom_subject_en = request.POST.get(f'custom_subject_en_{rule.id}', '').strip() rule.custom_body_en = request.POST.get(f'custom_body_en_{rule.id}', '').strip() rule.include_pdf_attachment = request.POST.get(f'include_pdf_{rule.id}') == 'on' rule.is_active = request.POST.get(f'active_{rule.id}') == 'on' rule.sort_order = position rule.save() new_name = request.POST.get('new_name', '').strip() new_recipients = request.POST.get('new_recipients', '').strip() if new_name and new_recipients: new_event = request.POST.get('new_event_type', 'onboarding').strip() if new_event not in {'onboarding', 'offboarding'}: new_event = 'onboarding' new_operator = request.POST.get('new_operator', 'always').strip() if new_operator not in {x[0] for x in NotificationRule.OPERATOR_CHOICES}: new_operator = 'always' NotificationRule.objects.create( name=new_name, event_type=new_event, field_name=request.POST.get('new_field_name', '').strip(), operator=new_operator, expected_value=request.POST.get('new_expected_value', '').strip(), recipients=new_recipients, template_key=request.POST.get('new_template_key', '').strip(), custom_subject=request.POST.get('new_custom_subject', '').strip(), custom_body=request.POST.get('new_custom_body', '').strip(), custom_subject_en=request.POST.get('new_custom_subject_en', '').strip(), custom_body_en=request.POST.get('new_custom_body_en', '').strip(), include_pdf_attachment=request.POST.get('new_include_pdf') == 'on', is_active=True, sort_order=NotificationRule.objects.filter(event_type=new_event).count() + 1, ) audit_fn(request, 'notification_rules_saved', target_type='notification_rule') messages.success(request, 'Benachrichtigungsregeln wurden gespeichert.') return redirect('/admin-tools/integrations/?kind=emails')