from pathlib import Path from datetime import timedelta import base64 import mimetypes import re from celery import current_task, 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_branding_email_copy, get_company_contact_copy, get_portal_letterhead_path from . import email_workflows, notification_dispatch, pdf_rendering from .models import AsyncTaskLog, EmployeeProfile, IntroChecklistItem, NotificationRule, NotificationTemplate, OffboardingRequest, OnboardingIntroductionSession, OnboardingRequest, ScheduledWelcomeEmail, WorkflowConfig 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, ) from .pdf_sections import build_pdf_sections # These templates are the product-level defaults for fresh deployments. # Runtime branding and company config can override the company-facing identity # without changing the workflow/task logic itself. 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 {{ SUPPORT_EMAIL }} 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 {{ SUPPORT_EMAIL }}.\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 {{ COMPANY_NAME }}, {{ VORNAME }}', 'subject_en': 'Welcome to {{ COMPANY_NAME }}, {{ VORNAME }}', 'body': ( 'Hallo {{ FULL_NAME }},\n\n' 'herzlich willkommen bei {{ COMPANY_NAME }}.\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' '{{ COMPANY_NAME }} IT' ), 'body_en': ( 'Hello {{ FULL_NAME }},\n\n' 'welcome to {{ COMPANY_NAME }}.\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' '{{ COMPANY_NAME }} 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 _notify_request_result(*, recipient_email: str, title: str, body: str, level: str, event_key: str) -> None: return notification_dispatch.notify_request_result( recipient_email=recipient_email, title=title, body=body, level=level, event_key=event_key, ) def _notify_welcome_email_result(*, recipient_email: str, full_name: str, body: str, level: str, event_key: str) -> None: return notification_dispatch.notify_welcome_email_result( recipient_email=recipient_email, full_name=full_name, body=body, level=level, event_key=event_key, ) def _start_task_log(task_name: str, *, target_type: str = '', target_id: int | None = None, target_label: str = '') -> AsyncTaskLog: task_request = getattr(current_task, 'request', None) return AsyncTaskLog.objects.create( task_name=task_name, task_id=getattr(task_request, 'id', '') or '', target_type=target_type, target_id=target_id, target_label=target_label, status='started', ) def _finish_task_log(task_log: AsyncTaskLog | None, *, status: str, error_message: str = '') -> None: if not task_log: return task_log.status = status task_log.error_message = error_message task_log.finished_at = timezone.now() task_log.save(update_fields=['status', 'error_message', 'finished_at']) def _split_name(full_name: str) -> tuple[str, str]: return pdf_rendering._split_name(full_name) def _safe_filename_fragment(text: str, fallback: str = 'document') -> str: return pdf_rendering._safe_filename_fragment(text, fallback=fallback) def _resolve_user_display_name(email: str) -> str: return pdf_rendering._resolve_user_display_name(email) def _chunk_list(data_list: list[str], chunk_size: int = 3) -> list[list[str]]: return pdf_rendering._chunk_list(data_list, chunk_size=chunk_size) def _split_multiline(text: str) -> list[str]: return pdf_rendering._split_multiline(text) def _chunk_choice_labels(choices: list[tuple[str, str]], chunk_size: int = 3) -> list[list[str]]: return pdf_rendering._chunk_choice_labels(choices, chunk_size=chunk_size) def _normalized_lang(language_code: str | None) -> str: return pdf_rendering._normalized_lang(language_code) def _pdf_texts(language_code: str | None = None) -> dict[str, str]: return pdf_rendering._pdf_texts(language_code) 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]: return pdf_rendering._manual_onboarding_field_sections() def _resolve_workflow_emails() -> tuple[str, str, str, str, str]: return email_workflows.resolve_workflow_emails() def _matches_intro_condition(request_obj: OnboardingRequest, item: IntroChecklistItem) -> bool: return pdf_rendering._matches_intro_condition(request_obj, item) def _build_intro_sections_from_admin(request_obj: OnboardingRequest, language_code: str | None = None) -> dict[str, list[str]]: return pdf_rendering._build_intro_sections_from_admin(request_obj, language_code=language_code) def build_intro_sections_for_request(request_obj: OnboardingRequest, language_code: str | None = None) -> list[dict]: return pdf_rendering.build_intro_sections_for_request(request_obj, language_code=language_code) def _send_workflow_email( subject: str, body: str, to: list[str], attachments: list[Path] | None = None, from_email: str | None = None, ) -> None: return email_workflows.send_workflow_email( subject=subject, body=body, to=to, attachments=attachments, from_email=from_email, ) def _render_notification_template(template_key: str, context: dict, language_code: str | None = None) -> tuple[str, str]: return email_workflows.render_notification_template( template_key, context, language_code=language_code, ) def _parse_recipients(raw: str) -> list[str]: return email_workflows.parse_recipients(raw) def _as_bool(value) -> bool: return email_workflows.as_bool(value) def _rule_matches(rule: NotificationRule, request_obj) -> bool: return email_workflows.rule_matches(rule, request_obj) 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: return email_workflows.schedule_welcome_email( request_obj, send_scheduled_welcome_email_task=send_scheduled_welcome_email, ) 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: return pdf_rendering._render_html(template_path, context) def _generate_content_pdf(html_content: str, output_pdf: Path) -> None: return pdf_rendering._generate_content_pdf(html_content, output_pdf) def _overlay_with_letterhead(content_pdf: Path, letterhead_pdf: Path, output_pdf: Path) -> None: return pdf_rendering._overlay_with_letterhead(content_pdf, letterhead_pdf, output_pdf) def _generate_onboarding_pdf(request_obj: OnboardingRequest) -> Path: return pdf_rendering._generate_onboarding_pdf(request_obj) def _generate_onboarding_intro_pdf(request_obj: OnboardingRequest, language_code: str | None = None) -> Path: return pdf_rendering._generate_onboarding_intro_pdf(request_obj, language_code=language_code) def _generate_onboarding_intro_session_pdf( session: OnboardingIntroductionSession, admin_signature_name: str = '-', language_code: str | None = None, ) -> Path: return pdf_rendering._generate_onboarding_intro_session_pdf( session, admin_signature_name=admin_signature_name, language_code=language_code, ) def _generate_offboarding_pdf(request_obj: OffboardingRequest) -> Path: return pdf_rendering._generate_offboarding_pdf(request_obj) @shared_task def process_onboarding_request(onboarding_request_id: int) -> None: request_obj = OnboardingRequest.objects.get(id=onboarding_request_id) task_log = _start_task_log( 'process_onboarding_request', target_type='onboarding_request', target_id=request_obj.id, target_label=request_obj.full_name, ) request_obj.processing_status = 'processing' request_obj.last_error = '' request_obj.save(update_fields=['processing_status', 'last_error']) try: branding_copy = get_branding_email_copy() company_contact = get_company_contact_copy() 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 '-', 'SUPPORT_EMAIL': company_contact['it_contact_email'] or branding_copy['support_email'] or f"it@{branding_copy['company_domain']}", 'IT_CONTACT_EMAIL': company_contact['it_contact_email'], 'HR_CONTACT_EMAIL': company_contact['hr_contact_email'], 'OPERATIONS_CONTACT_EMAIL': company_contact['operations_contact_email'], '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']) _notify_request_result( recipient_email=request_obj.onboarded_by_email, title=_('Onboarding abgeschlossen: %(name)s') % {'name': request_obj.full_name}, body=_('Die Onboarding-Anfrage wurde erfolgreich verarbeitet.'), level='success', event_key='onboarding_success', ) _finish_task_log(task_log, status='succeeded') except Exception as exc: request_obj.processing_status = 'failed' request_obj.last_error = str(exc) request_obj.save(update_fields=['processing_status', 'last_error']) _notify_request_result( recipient_email=request_obj.onboarded_by_email, title=_('Onboarding fehlgeschlagen: %(name)s') % {'name': request_obj.full_name}, body=str(exc), level='error', event_key='onboarding_failure', ) _finish_task_log(task_log, status='failed', error_message=str(exc)) raise @shared_task def process_offboarding_request(offboarding_request_id: int) -> None: request_obj = OffboardingRequest.objects.get(id=offboarding_request_id) task_log = _start_task_log( 'process_offboarding_request', target_type='offboarding_request', target_id=request_obj.id, target_label=request_obj.full_name, ) request_obj.processing_status = 'processing' request_obj.last_error = '' request_obj.save(update_fields=['processing_status', 'last_error']) try: branding_copy = get_branding_email_copy() company_contact = get_company_contact_copy() it_email, general_info_email, business_card_email_unused, hr_works_email, key_email_unused = _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, 'SUPPORT_EMAIL': company_contact['it_contact_email'] or branding_copy['support_email'] or f"it@{branding_copy['company_domain']}", 'IT_CONTACT_EMAIL': company_contact['it_contact_email'], 'HR_CONTACT_EMAIL': company_contact['hr_contact_email'], 'OPERATIONS_CONTACT_EMAIL': company_contact['operations_contact_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']) _notify_request_result( recipient_email=request_obj.requested_by_email, title=_('Offboarding abgeschlossen: %(name)s') % {'name': request_obj.full_name}, body=_('Die Offboarding-Anfrage wurde erfolgreich verarbeitet.'), level='success', event_key='offboarding_success', ) _finish_task_log(task_log, status='succeeded') except Exception as exc: request_obj.processing_status = 'failed' request_obj.last_error = str(exc) request_obj.save(update_fields=['processing_status', 'last_error']) _notify_request_result( recipient_email=request_obj.requested_by_email, title=_('Offboarding fehlgeschlagen: %(name)s') % {'name': request_obj.full_name}, body=str(exc), level='error', event_key='offboarding_failure', ) _finish_task_log(task_log, status='failed', error_message=str(exc)) 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 task_log = _start_task_log( 'send_scheduled_welcome_email', target_type='scheduled_welcome_email', target_id=scheduled.id, target_label=scheduled.recipient_email, ) if scheduled.status in {'sent', 'cancelled'} and not force_now: _finish_task_log(task_log, status='succeeded') return if scheduled.status == 'paused' and not force_now: _finish_task_log(task_log, status='succeeded') 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']) _finish_task_log(task_log, status='succeeded') 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 = '' _notify_welcome_email_result( recipient_email=request_obj.onboarded_by_email, full_name=request_obj.full_name, body=_('Die geplante Welcome E-Mail wurde erfolgreich versendet.'), level='success', event_key='welcome_email_success', ) _finish_task_log(task_log, status='succeeded') except Exception as exc: scheduled.status = 'failed' scheduled.last_error = str(exc) _notify_welcome_email_result( recipient_email=request_obj.onboarded_by_email, full_name=request_obj.full_name, body=str(exc), level='error', event_key='welcome_email_failure', ) _finish_task_log(task_log, status='failed', error_message=str(exc)) raise finally: scheduled.save(update_fields=['status', 'sent_at', 'last_error', 'updated_at'])