from pathlib import Path from datetime import timedelta import base64 import mimetypes import re from celery import shared_task from django.contrib.auth import get_user_model from django.conf import settings from django.utils import timezone from django.utils.translation import gettext as _, get_language, override from jinja2 import Template from pypdf import PageObject, PdfReader, PdfWriter from xhtml2pdf import pisa from .branding import get_default_notification_templates, get_portal_letterhead_path from .models import EmployeeProfile, IntroChecklistItem, NotificationRule, NotificationTemplate, OffboardingRequest, OnboardingIntroductionSession, OnboardingRequest, ScheduledWelcomeEmail, WorkflowConfig from .emailing import send_system_email from .services import upload_to_nextcloud from .services import get_email_test_redirect, is_email_test_mode from .forms import ( ACCESS_CHOICES, DEVICE_CHOICES, HARDWARE_EXTRA_CHOICES, OnboardingRequestForm, RESOURCE_CHOICES, SOFTWARE_CHOICES, SOFTWARE_EXTRA_CHOICES, WORKSPACE_GROUP_CHOICES, ) DEFAULT_NOTIFICATION_TEMPLATES = { 'onboarding_it': { 'subject': '[Onboarding] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[Onboarding] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Neue Onboarding-Anfrage für {{ FULL_NAME }}.\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Vertragsbeginn: {{ CONTRACT_START }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' 'Bitte IT-Setup vorbereiten.' ), 'body_en': ( 'New onboarding request for {{ FULL_NAME }}.\n' 'Department: {{ DEPARTMENT }}\n' 'Contract start: {{ CONTRACT_START }}\n' 'Requested by: {{ REQUESTED_BY }}\n' 'Please prepare the IT setup.' ), }, 'onboarding_general_info': { 'subject': '[Info Onboarding] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[Onboarding Info] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Hallo,\n\n' '{{ FULL_NAME }} wird onboarded.\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Vertragsbeginn: {{ CONTRACT_START }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' ), 'body_en': ( 'Hello,\n\n' '{{ FULL_NAME }} is being onboarded.\n' 'Department: {{ DEPARTMENT }}\n' 'Contract start: {{ CONTRACT_START }}\n' 'Requested by: {{ REQUESTED_BY }}\n' ), }, 'onboarding_business_card': { 'subject': '[Visitenkarte] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[Business Card] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Hallo,\n\n' 'bitte Visitenkarten erstellen:\n' 'Name: {{ BUSINESS_CARD_NAME }}\n' 'Titel: {{ BUSINESS_CARD_TITLE }}\n' 'E-Mail: {{ BUSINESS_CARD_EMAIL }}\n' 'Telefon: {{ BUSINESS_CARD_PHONE }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' ), 'body_en': ( 'Hello,\n\n' 'please create business cards:\n' 'Name: {{ BUSINESS_CARD_NAME }}\n' 'Title: {{ BUSINESS_CARD_TITLE }}\n' 'Email: {{ BUSINESS_CARD_EMAIL }}\n' 'Phone: {{ BUSINESS_CARD_PHONE }}\n' 'Requested by: {{ REQUESTED_BY }}\n' ), }, 'onboarding_hr_works': { 'subject': '[HR Works] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[HR Works] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Hello Stefanie,\n\n' 'Es ist wieder soweit. Zuwachs!\n\n' 'Könntest du deshalb bitte ein HR Works Konto mit den folgenden Daten erstellen:\n\n' 'Name: {{ VORNAME }} {{ NACHNAME }}\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Vertragsbeginn: {{ CONTRACT_START }}\n' 'E-Mail-Adresse: {{ EMAIL }}\n\n' '{% if PDF_LINK %}In 2 Minuten findest du alle Infos über den Mitarbeiter als PDF unter diesem Link: {{ PDF_LINK }}\n\n{% endif %}' 'Falls du noch irgendwelche anderen Informationen benötigen solltest, kannst du dich bei der it@tub.co melden!\n\n' 'Vielen Dank und schöne Grüße,\n' 'Die IT.' ), 'body_en': ( 'Hello Stefanie,\n\n' 'we have a new team member joining.\n\n' 'Could you please create an HR Works account with the following details:\n\n' 'Name: {{ VORNAME }} {{ NACHNAME }}\n' 'Department: {{ DEPARTMENT }}\n' 'Contract start: {{ CONTRACT_START }}\n' 'Email address: {{ EMAIL }}\n\n' '{% if PDF_LINK %}You will find the employee PDF here in about 2 minutes: {{ PDF_LINK }}\n\n{% endif %}' 'If you need any other information, please contact it@tub.co.\n\n' 'Thank you and best regards,\n' 'IT' ), }, 'onboarding_key': { 'subject': '[Schlüssel] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[Key] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Hallo,\n\n' 'bitte Schlüssel vorbereiten für:\n' 'Name: {{ FULL_NAME }}\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Vertragsbeginn: {{ CONTRACT_START }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' ), 'body_en': ( 'Hello,\n\n' 'please prepare keys for:\n' 'Name: {{ FULL_NAME }}\n' 'Department: {{ DEPARTMENT }}\n' 'Contract start: {{ CONTRACT_START }}\n' 'Requested by: {{ REQUESTED_BY }}\n' ), }, 'onboarding_reference': { 'subject': '[Referenz Onboarding] {{ FULL_NAME }} | Ihre Anfrage', 'subject_en': '[Onboarding Reference] {{ FULL_NAME }} | Your Request', 'body': ( 'Diese E-Mail dient als Referenz für Ihre Onboarding-Anfrage.\n' 'Name: {{ FULL_NAME }}\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Vertragsbeginn: {{ CONTRACT_START }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' ), 'body_en': ( 'This email is your reference copy for the onboarding request.\n' 'Name: {{ FULL_NAME }}\n' 'Department: {{ DEPARTMENT }}\n' 'Contract start: {{ CONTRACT_START }}\n' 'Requested by: {{ REQUESTED_BY }}\n' ), }, 'onboarding_welcome': { 'subject': 'Willkommen bei TUB/CO, {{ VORNAME }}', 'subject_en': 'Welcome to TUB/CO, {{ VORNAME }}', 'body': ( 'Hallo {{ FULL_NAME }},\n\n' 'herzlich willkommen bei TUB/CO.\n' 'Wir freuen uns sehr, dass du ab dem {{ CONTRACT_START }} unser Team in der Abteilung {{ DEPARTMENT }} verstärkst.\n\n' 'Deine dienstliche E-Mail-Adresse lautet: {{ EMAIL }}.\n' 'Im Anhang findest du deine Onboarding-Unterlagen als PDF.\n\n' 'Wenn du Fragen hast, melde dich gerne jederzeit.\n\n' 'Viele Grüße\n' 'TUB/CO IT' ), 'body_en': ( 'Hello {{ FULL_NAME }},\n\n' 'welcome to TUB/CO.\n' 'We are very happy that you will join our {{ DEPARTMENT }} team starting on {{ CONTRACT_START }}.\n\n' 'Your work email address is: {{ EMAIL }}.\n' 'You will find your onboarding documents attached as a PDF.\n\n' 'If you have any questions, feel free to contact us anytime.\n\n' 'Best regards,\n' 'TUB/CO IT' ), }, 'offboarding_it': { 'subject': '[Offboarding] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[Offboarding] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Neue Offboarding-Anfrage für {{ FULL_NAME }}.\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Letzter Arbeitstag: {{ LAST_WORKING_DAY }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' 'Bitte IT-Offboarding durchführen.' ), 'body_en': ( 'New offboarding request for {{ FULL_NAME }}.\n' 'Department: {{ DEPARTMENT }}\n' 'Last working day: {{ LAST_WORKING_DAY }}\n' 'Requested by: {{ REQUESTED_BY }}\n' 'Please complete the IT offboarding.' ), }, 'offboarding_general_info': { 'subject': '[Info Offboarding] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[Offboarding Info] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Neue Offboarding-Anfrage für {{ FULL_NAME }}.\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Letzter Arbeitstag: {{ LAST_WORKING_DAY }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' ), 'body_en': ( 'New offboarding request for {{ FULL_NAME }}.\n' 'Department: {{ DEPARTMENT }}\n' 'Last working day: {{ LAST_WORKING_DAY }}\n' 'Requested by: {{ REQUESTED_BY }}\n' ), }, 'offboarding_hr_works_disable': { 'subject': '[HR Works Deaktivierung] {{ FULL_NAME }} | Anfrage von {{ REQUESTED_BY }}', 'subject_en': '[HR Works Disable] {{ FULL_NAME }} | Requested by {{ REQUESTED_BY }}', 'body': ( 'Bitte HR Works Zugriff deaktivieren für {{ FULL_NAME }} ({{ EMAIL }}) zum {{ LAST_WORKING_DAY }}.\n' 'Angefordert von: {{ REQUESTED_BY }}\n' ), 'body_en': ( 'Please disable HR Works access for {{ FULL_NAME }} ({{ EMAIL }}) effective {{ LAST_WORKING_DAY }}.\n' 'Requested by: {{ REQUESTED_BY }}\n' ), }, 'offboarding_reference': { 'subject': '[Referenz Offboarding] {{ FULL_NAME }} | Ihre Anfrage', 'subject_en': '[Offboarding Reference] {{ FULL_NAME }} | Your Request', 'body': ( 'Diese E-Mail dient als Referenz für Ihre Offboarding-Anfrage.\n' 'Name: {{ FULL_NAME }}\n' 'Abteilung: {{ DEPARTMENT }}\n' 'Letzter Arbeitstag: {{ LAST_WORKING_DAY }}\n' 'Angefordert von: {{ REQUESTED_BY }}\n' ), 'body_en': ( 'This email is your reference copy for the offboarding request.\n' 'Name: {{ FULL_NAME }}\n' 'Department: {{ DEPARTMENT }}\n' 'Last working day: {{ LAST_WORKING_DAY }}\n' 'Requested by: {{ REQUESTED_BY }}\n' ), }, } def _split_name(full_name: str) -> tuple[str, str]: parts = full_name.split() if not parts: return '', '' return parts[0], ' '.join(parts[1:]) def _safe_filename_fragment(text: str, fallback: str = 'document') -> str: value = re.sub(r'[^A-Za-z0-9._-]+', '_', (text or '').strip()).strip('._') return value[:120] if value else fallback def _resolve_user_display_name(email: str) -> str: email = (email or '').strip().lower() if not email: return '' user_model = get_user_model() user = user_model.objects.filter(email__iexact=email).first() if not user: return '' first_name = (getattr(user, 'first_name', '') or '').strip() last_name = (getattr(user, 'last_name', '') or '').strip() full_name = f'{first_name} {last_name}'.strip() if full_name: return full_name return (getattr(user, 'username', '') or '').strip() def _chunk_list(data_list: list[str], chunk_size: int = 3) -> list[list[str]]: items = [i.strip() for i in data_list if i and i.strip()] chunks = [] for i in range(0, len(items), chunk_size): chunks.append(items[i : i + chunk_size]) return chunks def _split_multiline(text: str) -> list[str]: return [line.strip() for line in (text or '').split('\n') if line.strip()] def _chunk_choice_labels(choices: list[tuple[str, str]], chunk_size: int = 3) -> list[list[str]]: labels = [label for _, label in choices] return _chunk_list(labels, chunk_size=chunk_size) def _normalized_lang(language_code: str | None) -> str: return (language_code or 'de').split('-')[0].lower() or 'de' def _pdf_texts(language_code: str | None = None) -> dict[str, str]: lang = _normalized_lang(language_code) texts = { 'de': { 'lang': 'de', 'not_available': 'Keine Angabe', 'not_available_short': '-', 'yes': 'Ja', 'no': 'Nein', 'onboarding_title': 'Onboarding-Unterlagen', 'onboarding_staff_data': 'Personaldaten', 'name': 'Name', 'department': 'Abteilung', 'job_title': 'Berufsbezeichnung', 'work_email': 'Dienstliche E-Mail', 'employment_type': 'Beschäftigungsverhältnis', 'contract_start': 'Vertragsbeginn', 'contract_end': 'Vertragsende', 'handover_date': 'Übergabedatum', 'equipment_access': 'Ausstattung und Zugänge', 'devices': 'Benötigte Geräte und Gegenstände', 'workspace_groups': 'Benötigte Gruppen im Workspace', 'software': 'Benötigte Software', 'accesses': 'Benötigte Zugänge', 'resources': 'Benötigte Ressourcen', 'group_mailboxes_required': 'Gruppenpostfächer erforderlich', 'additional_hardware_needed': 'Darüber hinaus wird weitere Hardware benötigt', 'additional_software_needed': 'Wird zusätzliche Software benötigt', 'additional_access_needed': 'Darüber hinaus werden weitere Zugänge benötigt', 'additional_details': 'Zusätzliche Angaben', 'business_cards': 'Visitenkarten', 'email': 'E-Mail', 'phone': 'Telefon', 'additional_hardware_other': 'Weitere Hardware (Freitext)', 'successor_phone': 'Nachfolge und Telefon', 'successor_of': 'Nachfolge von', 'inherit_phone_number': 'Telefon von Vorgänger übernehmen', 'direct_extension': 'Direktwahl', 'notes': 'Notizen', 'confirmation': 'Bestätigung', 'requested_by_name': 'Angefordert von (Name)', 'requested_by_email': 'Angefordert von (E-Mail)', 'signature': 'Unterschrift', 'signature_alt': 'Unterschrift', 'onboarding_note': 'Hinweis: Dieses Formular dient als interne Prozessgrundlage für das Onboarding.', 'offboarding_title': 'Offboarding-Unterlagen', 'employee_info': 'Mitarbeitenden-Informationen', 'last_working_day': 'Letzter Arbeitstag', 'offboarding_requester': 'Offboarding-Anfordernde Person', 'it_hardware_status': 'IT-Hardware-Status (aus Onboarding)', 'hardware_check': 'Hardware-Check', 'no_onboarding_hardware': 'Keine Onboarding-Hardwaredaten gefunden.', 'manual_return_overview': 'Manuelle Rückgabeübersicht', 'manual_return_note': 'Es wurden keine gespeicherten Onboarding-Daten zu dieser Person gefunden. Die folgenden Listen dienen als manuelle Rückgabe- und Prüfübersicht.', 'returned_devices': 'Zurückgegebene Geräte und Artikel', 'returned_software': 'Zurückgegebene bzw. deaktivierte Software', 'removed_workspace_groups': 'Entfernte Gruppen im Workspace', 'removed_accesses': 'Entfernte Zugänge', 'returned_extra_it': 'Zurückgegebene zusätzliche Hardware / Software', 'it_signatures': 'IT-Section: Signaturen', 'it_checked_by': 'IT geprüft am durch:', 'it_signature': 'IT-Unterschrift:', 'return_complete': 'Rückgabe vollständig:', 'offboarding_note': 'Hinweis: Dieses Formular dient als interne Prozessgrundlage für das Offboarding.', 'intro_title': 'Einweisungs- und Übergabeprotokoll', 'intro_sub': 'Gesprächsleitfaden für die persönliche Einführung neuer Mitarbeitender.', 'base_data': 'Basisdaten', 'start_date': 'Startdatum', 'introduced_by': 'Einweisung durch', 'intro_note': 'Dieses Dokument dient als Gesprächsleitfaden für die persönliche Einweisung. Die Felder können während des Termins manuell abgehakt und anschließend unterschrieben werden.', 'employee_signature': 'Unterschrift Mitarbeitende Person:', 'trainer_signature': 'Unterschrift Einweisende Person:', 'intro_completed_at': 'Einweisung durchgeführt am:', 'open_questions': 'Rückfragen offen / Nacharbeit erforderlich:', 'live_intro_title': 'Einweisungsprotokoll', 'live_intro_sub': 'Export des aktuellen Live-Status aus der webbasierten Einweisung.', 'employment_start': 'Vertragsbeginn', 'employee_signature_block': 'Unterschrift Mitarbeitende Person', }, 'en': { 'lang': 'en', 'not_available': 'Not provided', 'not_available_short': '-', 'yes': 'Yes', 'no': 'No', 'onboarding_title': 'Onboarding Documents', 'onboarding_staff_data': 'Employee Details', 'name': 'Name', 'department': 'Department', 'job_title': 'Job title', 'work_email': 'Work email', 'employment_type': 'Employment type', 'contract_start': 'Contract start', 'contract_end': 'Contract end', 'handover_date': 'Handover date', 'equipment_access': 'Equipment and access', 'devices': 'Required devices and items', 'workspace_groups': 'Required workspace groups', 'software': 'Required software', 'accesses': 'Required accesses', 'resources': 'Required resources', 'group_mailboxes_required': 'Group mailboxes required', 'additional_hardware_needed': 'Additional hardware required', 'additional_software_needed': 'Additional software required', 'additional_access_needed': 'Additional accesses required', 'additional_details': 'Additional details', 'business_cards': 'Business cards', 'email': 'Email', 'phone': 'Phone', 'additional_hardware_other': 'Additional hardware (free text)', 'successor_phone': 'Successor and phone', 'successor_of': 'Successor to', 'inherit_phone_number': 'Take over predecessor phone number', 'direct_extension': 'Direct extension', 'notes': 'Notes', 'confirmation': 'Confirmation', 'requested_by_name': 'Requested by (name)', 'requested_by_email': 'Requested by (email)', 'signature': 'Signature', 'signature_alt': 'Signature', 'onboarding_note': 'Note: This form serves as the internal process basis for onboarding.', 'offboarding_title': 'Offboarding Documents', 'employee_info': 'Employee information', 'last_working_day': 'Last working day', 'offboarding_requester': 'Offboarding requester', 'it_hardware_status': 'IT hardware status (from onboarding)', 'hardware_check': 'Hardware check', 'no_onboarding_hardware': 'No onboarding hardware data found.', 'manual_return_overview': 'Manual return overview', 'manual_return_note': 'No stored onboarding data was found for this person. The following lists serve as a manual return and review overview.', 'returned_devices': 'Returned devices and items', 'returned_software': 'Returned or disabled software', 'removed_workspace_groups': 'Removed workspace groups', 'removed_accesses': 'Removed accesses', 'returned_extra_it': 'Returned additional hardware / software', 'it_signatures': 'IT section: signatures', 'it_checked_by': 'Checked by IT on:', 'it_signature': 'IT signature:', 'return_complete': 'Return complete:', 'offboarding_note': 'Note: This form serves as the internal process basis for offboarding.', 'intro_title': 'Introduction and Handover Protocol', 'intro_sub': 'Conversation guide for the personal introduction of new employees.', 'base_data': 'Basic data', 'start_date': 'Start date', 'introduced_by': 'Introduction by', 'intro_note': 'This document serves as a conversation guide for the personal introduction. The fields can be checked manually during the meeting and signed afterwards.', 'employee_signature': 'Employee signature:', 'trainer_signature': 'Trainer signature:', 'intro_completed_at': 'Introduction completed on:', 'open_questions': 'Open questions / follow-up required:', 'live_intro_title': 'Introduction Protocol', 'live_intro_sub': 'Export of the current live status from the web-based introduction.', 'employment_start': 'Contract start', 'employee_signature_block': 'Employee signature', }, } return texts.get(lang, texts['de']) MANUAL_ONBOARDING_FIELD_SECTIONS = [ ( 'Stammdaten', [ 'gender', 'first_name', 'last_name', 'job_title', 'department', 'work_email', 'order_business_cards', 'business_card_name', 'business_card_title', 'business_card_email', 'business_card_phone', ], ), ( 'Vertrag', [ 'contract_start', 'employment_type', 'employment_end_date', 'handover_date', ], ), ( 'IT-Setup', [ 'group_mailboxes_required_choice', 'group_mailboxes', 'additional_hardware_needed_choice', 'additional_hardware_other', 'additional_software_needed_choice', 'additional_software', 'additional_access_needed_choice', 'additional_access_text', 'successor_required_choice', 'successor_name', 'inherit_phone_number_choice', 'phone_number_choice', ], ), ( 'Abschluss', [ 'additional_notes', 'signature_image', 'agreement_confirm', ], ), ] def _manual_onboarding_field_sections() -> list[dict]: fields = OnboardingRequestForm.base_fields sections = [] for title, field_names in MANUAL_ONBOARDING_FIELD_SECTIONS: labels = [str(fields[name].label or name) for name in field_names if name in fields] if not labels: continue sections.append({'title': title, 'rows': _chunk_list(labels, chunk_size=2)}) return sections def _resolve_workflow_emails() -> tuple[str, str, str, str, str]: config = WorkflowConfig.objects.order_by('id').first() it_email = (config.it_onboarding_email if config and config.it_onboarding_email else settings.IT_ONBOARDING_NOTIFICATION_EMAIL) general_info_email = (config.general_info_email if config and config.general_info_email else settings.GENERAL_INFO_NOTIFICATION_EMAIL) business_card_email = (config.business_card_email if config and config.business_card_email else settings.BUSINESS_CARD_NOTIFICATION_EMAIL) hr_works_email = (config.hr_works_email if config and config.hr_works_email else settings.HR_WORKS_NOTIFICATION_EMAIL) key_email = (config.key_notification_email if config and config.key_notification_email else settings.KEY_NOTIFICATION_EMAIL) return it_email, general_info_email, business_card_email, hr_works_email, key_email def _matches_intro_condition(request_obj: OnboardingRequest, item: IntroChecklistItem) -> bool: operator = (item.condition_operator or 'always').strip() field_name = (item.condition_field or '').strip() expected = (item.condition_value or '').strip() if operator == 'always' or not field_name: return True raw_value = getattr(request_obj, field_name, '') if raw_value is None: raw_value = '' if operator == 'is_true': return bool(raw_value) if operator == 'is_false': return not bool(raw_value) text_value = str(raw_value).strip() if operator == 'equals': return text_value.lower() == expected.lower() if operator == 'contains': return expected.lower() in text_value.lower() return True def _build_intro_sections_from_admin(request_obj: OnboardingRequest, language_code: str | None = None) -> dict[str, list[str]]: items = list(IntroChecklistItem.objects.filter(is_active=True).order_by('section', 'sort_order', 'label')) if not items: return {} section_map = {key: [] for key, _label in IntroChecklistItem.SECTION_CHOICES} for item in items: if item.section not in section_map: continue if _matches_intro_condition(request_obj, item): section_map[item.section].append(item.translated_label(language_code)) return {key: values for key, values in section_map.items() if values} def build_intro_sections_for_request(request_obj: OnboardingRequest, language_code: str | None = None) -> list[dict]: lang = _normalized_lang(language_code or get_language()) with override(lang): section_titles = { 'workplace': _('Geräte und Arbeitsplatz'), 'accounts': _('Konten und Berechtigungen'), 'software': _('Software und Tools'), 'process': _('Prozesse und Hinweise'), } devices = _split_multiline(request_obj.needed_devices) software = _split_multiline(request_obj.needed_software) accesses = _split_multiline(request_obj.needed_accesses) groups = _split_multiline(request_obj.needed_workspace_groups) resources = _split_multiline(request_obj.needed_resources) extra_hardware = _split_multiline(request_obj.additional_hardware) extra_software = _split_multiline(request_obj.additional_software) group_mailboxes = _split_multiline(request_obj.group_mailboxes) workplace_items = [] for item in devices: workplace_items.append(_('%(item)s übergeben und Grundfunktionen erklärt') % {'item': item}) for item in resources: workplace_items.append(_('%(item)s gezeigt bzw. Nutzung erklärt') % {'item': item}) if request_obj.phone_number: workplace_items.append(_('Telefonnummer / Direktwahl erklärt: %(value)s') % {'value': request_obj.phone_number}) if not workplace_items: workplace_items.append(_('Arbeitsplatz, Geräte und allgemeine Nutzung besprochen')) account_items = [_('%(item)s Zugang erklärt') % {'item': item} for item in accesses] account_items.extend([_('%(item)s Gruppe / Berechtigung erläutert') % {'item': item} for item in groups]) if request_obj.work_email: account_items.insert(0, _('Dienstliche E-Mail-Adresse erläutert: %(value)s') % {'value': request_obj.work_email}) if group_mailboxes: account_items.extend([_('Gruppenpostfach erklärt: %(item)s') % {'item': item} for item in group_mailboxes]) if not account_items: account_items.append(_('Zugänge, Konten und Anmeldelogik besprochen')) software_items = [_('%(item)s Einführung durchgeführt') % {'item': item} for item in software] software_items.extend([_('%(item)s zusätzlich besprochen') % {'item': item} for item in extra_software]) if not software_items: software_items.append(_('Benötigte Standardsoftware und tägliche Nutzung erklärt')) process_items = [ _('Passwortregeln und sicherer Umgang besprochen'), _('Dateiablage, Nextcloud und Freigaben erklärt'), _('Kommunikationswege und Support-Prozess erklärt'), ] if extra_hardware: process_items.extend([_('%(item)s als zusätzliche Ausstattung besprochen') % {'item': item} for item in extra_hardware]) if request_obj.additional_access_text: process_items.extend([_('Zusätzlicher Zugang besprochen: %(item)s') % {'item': item} for item in _split_multiline(request_obj.additional_access_text)]) if request_obj.successor_name: process_items.append(_('Übergabe-/Nachfolgekontext besprochen: %(value)s') % {'value': request_obj.successor_name}) custom_intro_items = _build_intro_sections_from_admin(request_obj, lang) intro_sections_raw = [ ('workplace', section_titles['workplace'], workplace_items), ('accounts', section_titles['accounts'], account_items), ('software', section_titles['software'], software_items), ('process', section_titles['process'], process_items), ] sections = [] for key, title, default_items in intro_sections_raw: merged_items = list(default_items) merged_items.extend(custom_intro_items.get(key, [])) section_items = [] for idx, label in enumerate(merged_items, start=1): section_items.append({'id': f'{key}_{idx}', 'label': label}) if section_items: sections.append({'key': key, 'title': title, 'items': section_items}) return sections def _send_workflow_email( subject: str, body: str, to: list[str], attachments: list[Path] | None = None, from_email: str | None = None, ) -> None: recipients = [r for r in to if r] if not recipients: return effective_to = recipients effective_body = body if is_email_test_mode(): effective_to = [get_email_test_redirect()] effective_body = ( "[TEST MODE] Diese E-Mail wurde umgeleitet.\n" f"Originale Empfänger: {', '.join(recipients)}\n\n{body}" ) send_system_email( subject=subject, body=effective_body, to=effective_to, attachments=[str(a) for a in (attachments or [])], from_email=from_email, ) def _render_notification_template(template_key: str, context: dict, language_code: str | None = None) -> tuple[str, str]: lang = (language_code or 'de').split('-')[0] db_template = NotificationTemplate.objects.filter(key=template_key, is_active=True).first() if db_template: subject_template = db_template.translated_subject_template(lang) body_template = db_template.translated_body_template(lang) else: fallback = get_default_notification_templates()[template_key] subject_template = fallback.get(f'subject_{lang}', '') or fallback['subject'] body_template = fallback.get(f'body_{lang}', '') or fallback['body'] subject = Template(subject_template).render(context).strip() body = Template(body_template).render(context).strip() return subject, body def _parse_recipients(raw: str) -> list[str]: if not raw: return [] cleaned = raw.replace(';', ',').replace('\n', ',') return [x.strip() for x in cleaned.split(',') if x.strip()] def _as_bool(value) -> bool: if isinstance(value, bool): return value if value is None: return False text = str(value).strip().lower() return text in {'1', 'true', 'ja', 'yes', 'on', 'aktiv'} def _rule_matches(rule: NotificationRule, request_obj) -> bool: if rule.operator == 'always': return True raw_value = getattr(request_obj, rule.field_name, '') actual = '' if raw_value is None else str(raw_value) expected = (rule.expected_value or '').strip() if rule.operator == 'contains': return expected.lower() in actual.lower() if rule.operator == 'equals': return actual.strip().lower() == expected.lower() if rule.operator == 'is_true': return _as_bool(raw_value) if rule.operator == 'is_false': return not _as_bool(raw_value) return False def _apply_notification_rules( event_type: str, request_obj, context: dict, pdf_path: Path | None = None, ) -> None: language_code = (getattr(request_obj, 'preferred_language', '') or 'de').split('-')[0] rules = NotificationRule.objects.filter(event_type=event_type, is_active=True).order_by('sort_order', 'id') for rule in rules: if not _rule_matches(rule, request_obj): continue recipients = _parse_recipients(rule.recipients) if not recipients: continue attachments = [pdf_path] if (pdf_path and rule.include_pdf_attachment) else None template_key = (rule.template_key or '').strip() known_keys = {k for k, _ in NotificationTemplate.TEMPLATE_CHOICES} if template_key and template_key in known_keys: _send_templated_email( template_key=template_key, context=context, to=recipients, attachments=attachments, language_code=language_code, ) continue subject = rule.translated_custom_subject(language_code) body = rule.translated_custom_body(language_code) if not subject and not body: continue subject_rendered = Template(subject or f'[{event_type}] Regelmail').render(context).strip() body_rendered = Template(body or '-').render(context).strip() _send_workflow_email( subject=subject_rendered, body=body_rendered, to=recipients, attachments=attachments, ) def _schedule_welcome_email(request_obj: OnboardingRequest) -> None: recipient = (request_obj.work_email or '').strip().lower() if not recipient: return config = WorkflowConfig.objects.order_by('id').first() delay_days = 5 if config: delay_days = max(0, int(config.welcome_email_delay_days or 5)) send_at = timezone.now() + timedelta(days=delay_days) scheduled, _ = ScheduledWelcomeEmail.objects.update_or_create( onboarding_request=request_obj, defaults={ 'recipient_email': recipient, 'send_at': send_at, 'status': 'scheduled', 'last_error': '', 'sent_at': None, }, ) try: async_result = send_scheduled_welcome_email.apply_async(args=[scheduled.id], eta=send_at) scheduled.celery_task_id = async_result.id or '' scheduled.save(update_fields=['celery_task_id', 'updated_at']) except Exception as exc: scheduled.status = 'failed' scheduled.last_error = f'Scheduling failed: {exc}' scheduled.save(update_fields=['status', 'last_error', 'updated_at']) def _send_templated_email( template_key: str, to: list[str], context: dict, attachments: list[Path] | None = None, from_email: str | None = None, language_code: str | None = None, ) -> None: subject, body = _render_notification_template(template_key, context, language_code=language_code) _send_workflow_email(subject=subject, body=body, to=to, attachments=attachments, from_email=from_email) def _render_html(template_path: Path, context: dict) -> str: with template_path.open('r', encoding='utf-8') as handle: template = Template(handle.read()) return template.render(context) def _generate_content_pdf(html_content: str, output_pdf: Path) -> None: page_style = ( '' ) if '' in html_content: html_content = html_content.replace('', f'{page_style}', 1) else: html_content = page_style + html_content output_pdf.parent.mkdir(parents=True, exist_ok=True) with output_pdf.open('wb') as fp: result = pisa.CreatePDF( src=html_content, dest=fp, encoding='utf-8', ) if result.err: raise RuntimeError(f'Failed to render PDF content for {output_pdf.name}') def _overlay_with_letterhead(content_pdf: Path, letterhead_pdf: Path, output_pdf: Path) -> None: letterhead_reader = PdfReader(str(letterhead_pdf)) content_reader = PdfReader(str(content_pdf)) writer = PdfWriter() letterhead_page = letterhead_reader.pages[0] for page in content_reader.pages: merged = PageObject.create_blank_page( width=letterhead_page.mediabox.width, height=letterhead_page.mediabox.height, ) merged.merge_page(letterhead_page) merged.merge_page(page) writer.add_page(merged) with output_pdf.open('wb') as fp: writer.write(fp) def _generate_onboarding_pdf(request_obj: OnboardingRequest) -> Path: lang = _normalized_lang(request_obj.preferred_language) t = _pdf_texts(lang) first_name, last_name = _split_name(request_obj.full_name) safe_name = _safe_filename_fragment(request_obj.full_name, fallback=f'onboarding_{request_obj.id}') output_pdf = settings.PDF_OUTPUT_DIR / f'onboarding_letter_{safe_name}.pdf' temp_pdf = settings.PDF_OUTPUT_DIR / f'_temp_onboarding_{safe_name}.pdf' template_path = settings.PDF_TEMPLATES_DIR / 'onboarding_template.html' letterhead_path = get_portal_letterhead_path() devices = _split_multiline(request_obj.needed_devices) software = _split_multiline(request_obj.needed_software) accesses = _split_multiline(request_obj.needed_accesses) groups = _split_multiline(request_obj.needed_workspace_groups) resources = _split_multiline(request_obj.needed_resources) group_mailboxes_list = _split_multiline(request_obj.group_mailboxes or '') additional_hardware_list = _split_multiline(request_obj.additional_hardware or '') additional_software_list = _split_multiline(request_obj.additional_software or '') additional_access_list = _split_multiline(request_obj.additional_access_text or '') signature_src = '' signature_note = t['not_available_short'] if getattr(request_obj, 'signature_image', None): try: signature_path = Path(request_obj.signature_image.path).resolve() with signature_path.open('rb') as sig_fp: encoded = base64.b64encode(sig_fp.read()).decode('ascii') mime_type = mimetypes.guess_type(signature_path.name)[0] or 'image/png' signature_src = f"data:{mime_type};base64,{encoded}" signature_note = 'Digital signature stored as image file.' if lang == 'en' else 'Digitale Signatur als Bilddatei hinterlegt.' except Exception: signature_src = '' signature_note = request_obj.signature_url or t['not_available_short'] elif request_obj.signature_url: signature_note = request_obj.signature_url requester_email = request_obj.onboarded_by_email or t['not_available_short'] requester_name = request_obj.onboarded_by_name or _resolve_user_display_name(request_obj.onboarded_by_email) or t['not_available_short'] gender = (request_obj.get_gender_display() or t['not_available_short']).strip() or t['not_available_short'] employment_type = (request_obj.employment_type or t['not_available_short']).strip() or t['not_available_short'] employment_end = request_obj.employment_end_date or t['not_available_short'] order_business_cards = bool(request_obj.order_business_cards) group_mailboxes = (request_obj.group_mailboxes or '').strip() additional_hardware_other = (request_obj.additional_hardware_other or '').strip() additional_hardware = (request_obj.additional_hardware or '').strip() additional_software = (request_obj.additional_software or '').strip() additional_access_text = (request_obj.additional_access_text or '').strip() successor_name = (request_obj.successor_name or '').strip() additional_notes = (request_obj.additional_notes or '').strip() phone_number = (request_obj.phone_number or '').strip() display_name = f"{gender} {first_name} {last_name}".strip() if gender and gender != '-' else f"{first_name} {last_name}".strip() context = { 'T': t, 'PDF_LANG': lang, 'VORNAME': first_name, 'NACHNAME': last_name, 'DISPLAY_NAME': display_name or request_obj.full_name, 'ANREDE': gender, 'BERUFSBEZEICHNUNG': request_obj.job_title or t['not_available'], 'ABTEILUNG': request_obj.department or t['not_available'], 'EMAIL': request_obj.work_email or t['not_available'], 'VERTRAGSBEGINN': request_obj.contract_start, 'BESCHAEFTIGUNG': employment_type, 'VERTRAGSENDE': employment_end, 'UEBERGABEDATUM': request_obj.handover_date or t['not_available_short'], 'ARBEITSGERAETE_TEXT': ' | '.join(devices) if devices else t['not_available'], 'WORKSPACE_GROUPS_TEXT': ' | '.join(groups) if groups else t['not_available'], 'SOFTWARE_TEXT': ' | '.join(software) if software else t['not_available'], 'ZUGAENGE_TEXT': ' | '.join(accesses) if accesses else t['not_available'], 'RESSOURCEN_TEXT': ' | '.join(resources) if resources else t['not_available'], 'VISITENKARTE_BESTELLT': order_business_cards, 'HAS_VISITENKARTE_DATEN': order_business_cards and any( [ (request_obj.business_card_name or '').strip(), (request_obj.business_card_title or '').strip(), (request_obj.business_card_email or '').strip(), (request_obj.business_card_phone or '').strip(), ] ), 'VISITENKARTE_NAME': request_obj.business_card_name or t['not_available_short'], 'VISITENKARTE_TITEL': request_obj.business_card_title or t['not_available_short'], 'VISITENKARTE_EMAIL': request_obj.business_card_email or t['not_available_short'], 'VISITENKARTE_TELEFON': request_obj.business_card_phone or t['not_available_short'], 'GROUP_MAILBOXES': group_mailboxes or t['not_available'], 'ADDITIONAL_HARDWARE_OTHER': additional_hardware_other or t['not_available'], 'ADDITIONAL_HARDWARE': additional_hardware or t['not_available'], 'ADDITIONAL_SOFTWARE': additional_software or t['not_available'], 'ADDITIONAL_ACCESS_TEXT': additional_access_text or t['not_available'], 'SUCCESSOR_NAME': successor_name or t['not_available'], 'PHONE_NUMBER': phone_number or t['not_available_short'], 'INHERIT_PHONE_NUMBER': t['yes'] if request_obj.inherit_phone_number else t['no'], 'ADDITIONAL_NOTES': additional_notes or t['not_available'], 'GROUP_MAILBOXES_REQUIRED': bool(request_obj.group_mailboxes_required), 'ADDITIONAL_HARDWARE_NEEDED': bool(request_obj.additional_hardware_needed), 'ADDITIONAL_SOFTWARE_NEEDED': bool(request_obj.additional_software_needed), 'ADDITIONAL_ACCESS_NEEDED': bool(request_obj.additional_access_needed), 'HAS_DEVICES': bool(devices), 'HAS_GROUPS': bool(groups), 'HAS_SOFTWARE': bool(software), 'HAS_ACCESSES': bool(accesses), 'HAS_RESOURCES': bool(resources), 'HAS_GROUP_MAILBOXES': bool(group_mailboxes_list), 'HAS_ADDITIONAL_HARDWARE': bool(additional_hardware_list), 'HAS_ADDITIONAL_SOFTWARE': bool(additional_software_list), 'HAS_ADDITIONAL_ACCESS': bool(additional_access_list), 'HAS_ADDITIONAL_HARDWARE_OTHER': bool(additional_hardware_other), 'HAS_SUCCESSOR_INFO': bool(successor_name) or bool(request_obj.inherit_phone_number) or bool(phone_number), 'HAS_ADDITIONAL_NOTES': bool(additional_notes), 'GROUP_MAILBOXES_LIST': _chunk_list(group_mailboxes_list), 'ADDITIONAL_HARDWARE_LIST': _chunk_list(additional_hardware_list), 'ADDITIONAL_SOFTWARE_LIST': _chunk_list(additional_software_list), 'ADDITIONAL_ACCESS_LIST': _chunk_list(additional_access_list), 'ZUGAENGE_LIST': _chunk_list(groups), 'ARBEITSGERÄTE_LIST': _chunk_list(devices), 'SOFTWARE_LIST': _chunk_list(software), 'ACCOUNT_LIST': _chunk_list(accesses), 'STANDARD_RESSOURCEN': _chunk_list(resources), 'UNTERSCHRIFT': signature_src, 'UNTERSCHRIFT_HINWEIS': signature_note, 'REQUESTED_BY_NAME': requester_name, 'REQUESTED_BY_EMAIL': requester_email, } html = _render_html(template_path, context) _generate_content_pdf(html, temp_pdf) _overlay_with_letterhead(temp_pdf, letterhead_path, output_pdf) if temp_pdf.exists(): temp_pdf.unlink(missing_ok=True) return output_pdf def _generate_onboarding_intro_pdf(request_obj: OnboardingRequest, language_code: str | None = None) -> Path: lang = _normalized_lang(language_code or request_obj.preferred_language) t = _pdf_texts(lang) safe_name = _safe_filename_fragment(request_obj.full_name, fallback=f'onboarding_intro_{request_obj.id}') output_pdf = settings.PDF_OUTPUT_DIR / f'onboarding_intro_{safe_name}.pdf' temp_pdf = settings.PDF_OUTPUT_DIR / f'_temp_onboarding_intro_{safe_name}.pdf' template_path = settings.PDF_TEMPLATES_DIR / 'onboarding_intro_template.html' letterhead_path = get_portal_letterhead_path() salutation = (request_obj.get_gender_display() or '').strip() display_name = f"{salutation} {request_obj.full_name}".strip() if salutation else request_obj.full_name intro_sections = [ { 'title': section['title'], 'rows': _chunk_list([item['label'] for item in section['items']], chunk_size=2), } for section in build_intro_sections_for_request(request_obj, language_code=language_code) ] requester_email = request_obj.onboarded_by_email or '-' requester_name = request_obj.onboarded_by_name or _resolve_user_display_name(request_obj.onboarded_by_email) or '-' context = { 'T': t, 'DISPLAY_NAME': display_name, 'ABTEILUNG': request_obj.department or t['not_available_short'], 'BERUFSBEZEICHNUNG': request_obj.job_title or t['not_available_short'], 'VERTRAGSBEGINN': request_obj.contract_start, 'EMAIL': request_obj.work_email or t['not_available_short'], 'REQUESTED_BY_NAME': requester_name, 'REQUESTED_BY_EMAIL': requester_email, 'INTRO_SECTIONS': intro_sections, } html = _render_html(template_path, context) _generate_content_pdf(html, temp_pdf) _overlay_with_letterhead(temp_pdf, letterhead_path, output_pdf) if temp_pdf.exists(): temp_pdf.unlink(missing_ok=True) return output_pdf def _generate_onboarding_intro_session_pdf( session: OnboardingIntroductionSession, admin_signature_name: str = '-', language_code: str | None = None, ) -> Path: request_obj = session.onboarding_request lang = _normalized_lang(language_code or request_obj.preferred_language) t = _pdf_texts(lang) safe_name = _safe_filename_fragment(request_obj.full_name, fallback=f'onboarding_intro_session_{request_obj.id}') version = timezone.now().strftime('%Y%m%d%H%M%S') output_pdf = settings.PDF_OUTPUT_DIR / f'onboarding_intro_session_{safe_name}_{version}.pdf' temp_pdf = settings.PDF_OUTPUT_DIR / f'_temp_onboarding_intro_session_{safe_name}_{version}.pdf' template_path = settings.PDF_TEMPLATES_DIR / 'onboarding_intro_session_pdf.html' letterhead_path = get_portal_letterhead_path() salutation = (request_obj.get_gender_display() or '').strip() display_name = f"{salutation} {request_obj.full_name}".strip() if salutation else request_obj.full_name raw_sections = build_intro_sections_for_request(request_obj, language_code=language_code) checked_map = session.checklist_state or {} exported_sections = [] checked_count = 0 total_count = 0 for section in raw_sections: checked_items = [] for item in section['items']: checked = bool(checked_map.get(item['id'])) total_count += 1 if checked: checked_count += 1 checked_items.append({'label': item['label']}) if checked_items: exported_sections.append({ 'title': section['title'], 'rows': [checked_items[i:i + 2] for i in range(0, len(checked_items), 2)], }) requester_email = request_obj.onboarded_by_email or '-' requester_name = request_obj.onboarded_by_name or _resolve_user_display_name(request_obj.onboarded_by_email) or '-' context = { 'T': t, 'DISPLAY_NAME': display_name, 'ABTEILUNG': request_obj.department or t['not_available_short'], 'BERUFSBEZEICHNUNG': request_obj.job_title or t['not_available_short'], 'VERTRAGSBEGINN': request_obj.contract_start, 'EMAIL': request_obj.work_email or t['not_available_short'], 'REQUESTED_BY_NAME': requester_name, 'REQUESTED_BY_EMAIL': requester_email, 'SESSION_STATUS': session.get_status_display(), 'SESSION_COMPLETED_BY': session.completed_by_name or '-', 'SESSION_COMPLETED_AT': session.completed_at or '-', 'SESSION_UPDATED_AT': session.updated_at, 'SESSION_NOTES': session.notes or t['not_available_short'], 'INTRO_SECTIONS': exported_sections, } html = _render_html(template_path, context) _generate_content_pdf(html, temp_pdf) _overlay_with_letterhead(temp_pdf, letterhead_path, output_pdf) if temp_pdf.exists(): temp_pdf.unlink(missing_ok=True) return output_pdf def _generate_offboarding_pdf(request_obj: OffboardingRequest) -> Path: lang = _normalized_lang(request_obj.preferred_language) t = _pdf_texts(lang) safe_name = _safe_filename_fragment(request_obj.full_name, fallback=f'offboarding_{request_obj.id}') output_pdf = settings.PDF_OUTPUT_DIR / f'offboarding_letter_{safe_name}.pdf' temp_pdf = settings.PDF_OUTPUT_DIR / f'_temp_offboarding_{safe_name}.pdf' template_path = settings.PDF_TEMPLATES_DIR / 'offboarding_template.html' letterhead_path = get_portal_letterhead_path() latest_onboarding = ( OnboardingRequest.objects.filter(work_email=request_obj.work_email) .order_by('-created_at') .first() ) has_onboarding_data = latest_onboarding is not None onboarding_hardware = _split_multiline(latest_onboarding.needed_devices) if latest_onboarding else [] selected_set = {item.lower() for item in onboarding_hardware} hardware_catalog = [ 'Laptop', 'Docking-Station', 'Tastatur und Maus', 'Kopfhörer', 'Tragetasche', 'Monitor', 'Schlüssel', 'Tischtelefon', ] checklist = [{'label': item, 'selected': item.lower() in selected_set} for item in hardware_catalog] extra_selected = [item for item in onboarding_hardware if item.lower() not in {x.lower() for x in hardware_catalog}] for item in extra_selected: checklist.append({'label': item, 'selected': True}) requester_email = request_obj.requested_by_email or t['not_available_short'] requester_name = request_obj.requested_by_name or _resolve_user_display_name(request_obj.requested_by_email) or t['not_available_short'] context = { 'T': t, 'FULL_NAME': request_obj.full_name, 'EMAIL': request_obj.work_email, 'DEPARTMENT': request_obj.department or t['not_available_short'], 'JOB_TITLE': request_obj.job_title or t['not_available_short'], 'LAST_WORKING_DAY': request_obj.last_working_day, 'NOTES': request_obj.notes or t['not_available_short'], 'REQUESTED_BY': requester_email, 'REQUESTED_BY_NAME': requester_name, 'HAS_ONBOARDING_DATA': has_onboarding_data, 'ONBOARDING_HARDWARE': onboarding_hardware, 'HARDWARE_CHECKLIST': checklist, 'MANUAL_FIELD_SECTIONS': _manual_onboarding_field_sections(), 'MANUAL_DEVICES': _chunk_choice_labels(DEVICE_CHOICES), 'MANUAL_SOFTWARE': _chunk_choice_labels(SOFTWARE_CHOICES), 'MANUAL_ACCESSES': _chunk_choice_labels(ACCESS_CHOICES), 'MANUAL_WORKSPACE_GROUPS': _chunk_choice_labels(WORKSPACE_GROUP_CHOICES), 'MANUAL_RESOURCES': _chunk_choice_labels(RESOURCE_CHOICES), 'MANUAL_EXTRA_HARDWARE': _chunk_choice_labels(HARDWARE_EXTRA_CHOICES), 'MANUAL_EXTRA_SOFTWARE': _chunk_choice_labels(SOFTWARE_EXTRA_CHOICES), } html = _render_html(template_path, context) _generate_content_pdf(html, temp_pdf) _overlay_with_letterhead(temp_pdf, letterhead_path, output_pdf) if temp_pdf.exists(): temp_pdf.unlink(missing_ok=True) return output_pdf @shared_task def process_onboarding_request(onboarding_request_id: int) -> None: request_obj = OnboardingRequest.objects.get(id=onboarding_request_id) request_obj.processing_status = 'processing' request_obj.last_error = '' request_obj.save(update_fields=['processing_status', 'last_error']) try: it_email, general_info_email, business_card_email, hr_works_email, key_email = _resolve_workflow_emails() salutation = (request_obj.get_gender_display() or '').strip() display_name = f"{salutation} {request_obj.full_name}".strip() first_name, last_name = _split_name(request_obj.full_name) EmployeeProfile.objects.update_or_create( work_email=request_obj.work_email, defaults={ 'full_name': request_obj.full_name, 'first_name': first_name, 'last_name': last_name, 'department': request_obj.department, 'job_title': request_obj.job_title, }, ) pdf_path = _generate_onboarding_pdf(request_obj) request_obj.generated_pdf_path = str(pdf_path) request_obj.save(update_fields=['generated_pdf_path']) email_context = { 'FULL_NAME': display_name, 'VORNAME': first_name, 'NACHNAME': last_name, 'DEPARTMENT': request_obj.department or '-', 'CONTRACT_START': request_obj.contract_start, 'EMAIL': request_obj.work_email, 'REQUESTED_BY': request_obj.onboarded_by_email or '-', 'BUSINESS_CARD_NAME': request_obj.business_card_name or display_name, 'BUSINESS_CARD_TITLE': request_obj.business_card_title or '-', 'BUSINESS_CARD_EMAIL': request_obj.business_card_email or request_obj.work_email, 'BUSINESS_CARD_PHONE': request_obj.business_card_phone or '-', 'PDF_LINK': settings.ONBOARDING_SHARED_PDF_LINK, } _send_templated_email( template_key='onboarding_it', context=email_context, to=[it_email], attachments=[pdf_path], language_code=request_obj.preferred_language, ) _send_templated_email( template_key='onboarding_general_info', context=email_context, to=[general_info_email], language_code=request_obj.preferred_language, ) if request_obj.order_business_cards: _send_templated_email( template_key='onboarding_business_card', context=email_context, to=[business_card_email], language_code=request_obj.preferred_language, ) if 'HR Works' in request_obj.needed_accesses: _send_templated_email( template_key='onboarding_hr_works', context=email_context, to=[hr_works_email], language_code=request_obj.preferred_language, ) if 'Schlüssel' in request_obj.needed_devices: _send_templated_email( template_key='onboarding_key', context=email_context, to=[key_email], language_code=request_obj.preferred_language, ) if request_obj.onboarded_by_email: _send_templated_email( template_key='onboarding_reference', context=email_context, to=[request_obj.onboarded_by_email], attachments=[pdf_path], language_code=request_obj.preferred_language, ) _apply_notification_rules( event_type='onboarding', request_obj=request_obj, context=email_context, pdf_path=pdf_path, ) _schedule_welcome_email(request_obj) upload_to_nextcloud(pdf_path, Path(pdf_path).name) request_obj.processing_status = 'completed' request_obj.last_error = '' request_obj.save(update_fields=['processing_status', 'last_error']) except Exception as exc: request_obj.processing_status = 'failed' request_obj.last_error = str(exc) request_obj.save(update_fields=['processing_status', 'last_error']) raise @shared_task def process_offboarding_request(offboarding_request_id: int) -> None: request_obj = OffboardingRequest.objects.get(id=offboarding_request_id) request_obj.processing_status = 'processing' request_obj.last_error = '' request_obj.save(update_fields=['processing_status', 'last_error']) try: it_email, general_info_email, _, hr_works_email, _ = _resolve_workflow_emails() pdf_path = _generate_offboarding_pdf(request_obj) request_obj.generated_pdf_path = str(pdf_path) request_obj.save(update_fields=['generated_pdf_path']) email_context = { 'FULL_NAME': request_obj.full_name, 'DEPARTMENT': request_obj.department or '-', 'LAST_WORKING_DAY': request_obj.last_working_day, 'REQUESTED_BY': request_obj.requested_by_email, 'EMAIL': request_obj.work_email, } _send_templated_email( template_key='offboarding_it', context=email_context, to=[it_email], attachments=[pdf_path], language_code=request_obj.preferred_language, ) _send_templated_email( template_key='offboarding_general_info', context=email_context, to=[general_info_email], language_code=request_obj.preferred_language, ) had_hr_works = OnboardingRequest.objects.filter( work_email=request_obj.work_email, needed_accesses__icontains='HR Works', ).exists() if had_hr_works: _send_templated_email( template_key='offboarding_hr_works_disable', context=email_context, to=[hr_works_email], language_code=request_obj.preferred_language, ) _send_templated_email( template_key='offboarding_reference', context=email_context, to=[request_obj.requested_by_email], attachments=[pdf_path], language_code=request_obj.preferred_language, ) _apply_notification_rules( event_type='offboarding', request_obj=request_obj, context=email_context, pdf_path=pdf_path, ) upload_to_nextcloud(pdf_path, Path(pdf_path).name) request_obj.processing_status = 'completed' request_obj.last_error = '' request_obj.save(update_fields=['processing_status', 'last_error']) except Exception as exc: request_obj.processing_status = 'failed' request_obj.last_error = str(exc) request_obj.save(update_fields=['processing_status', 'last_error']) raise @shared_task def send_scheduled_welcome_email(scheduled_email_id: int, force_now: bool = False) -> None: scheduled = ScheduledWelcomeEmail.objects.select_related('onboarding_request').filter(id=scheduled_email_id).first() if not scheduled: return if scheduled.status in {'sent', 'cancelled'} and not force_now: return if scheduled.status == 'paused' and not force_now: return if not force_now and timezone.now() < scheduled.send_at: async_result = send_scheduled_welcome_email.apply_async(args=[scheduled.id], eta=scheduled.send_at) scheduled.celery_task_id = async_result.id or scheduled.celery_task_id scheduled.save(update_fields=['celery_task_id', 'updated_at']) return request_obj = scheduled.onboarding_request first_name, last_name = _split_name(request_obj.full_name) salutation = (request_obj.get_gender_display() or '').strip() display_name = f"{salutation} {request_obj.full_name}".strip() email_context = { 'FULL_NAME': display_name, 'VORNAME': first_name, 'NACHNAME': last_name, 'DEPARTMENT': request_obj.department or '-', 'CONTRACT_START': request_obj.contract_start, 'EMAIL': request_obj.work_email, 'REQUESTED_BY': request_obj.onboarded_by_email or '-', } config = WorkflowConfig.objects.order_by('id').first() include_pdf = True if not config else bool(config.welcome_include_pdf) from_email = '' if config: from_email = (config.welcome_sender_email or config.email_account or '').strip() attachments = [] if include_pdf and request_obj.generated_pdf_path: pdf_path = Path(request_obj.generated_pdf_path) if pdf_path.exists(): attachments = [pdf_path] try: _send_templated_email( template_key='onboarding_welcome', context=email_context, to=[scheduled.recipient_email], attachments=attachments, from_email=from_email or None, language_code=request_obj.preferred_language, ) scheduled.status = 'sent' scheduled.sent_at = timezone.now() scheduled.last_error = '' except Exception as exc: scheduled.status = 'failed' scheduled.last_error = str(exc) raise finally: scheduled.save(update_fields=['status', 'sent_at', 'last_error', 'updated_at'])