Files
workdock-platform/backend/workflows/trial.py
2026-03-27 01:11:29 +01:00

64 lines
2.5 KiB
Python

from __future__ import annotations
import shutil
from pathlib import Path
from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.sessions.models import Session
from django.utils import timezone
from .branding import get_portal_trial_config, is_trial_expired
from .models import AdminAuditLog, EmployeeProfile, OffboardingRequest, OnboardingIntroductionSession, OnboardingRequest, ScheduledWelcomeEmail
from .roles import ROLE_PLATFORM_OWNER, get_user_role_key
# Trial cleanup is intentionally destructive but platform-scoped.
# It preserves the platform-owner account so expired demo environments remain
# recoverable and inspectable by the product operator.
def cleanup_trial_workspace_data() -> dict[str, int]:
user_model = get_user_model()
deleted_counts = {
'onboarding_requests': OnboardingRequest.objects.count(),
'offboarding_requests': OffboardingRequest.objects.count(),
'intro_sessions': OnboardingIntroductionSession.objects.count(),
'scheduled_welcome_emails': ScheduledWelcomeEmail.objects.count(),
'employee_profiles': EmployeeProfile.objects.count(),
'audit_logs': AdminAuditLog.objects.count(),
'sessions': Session.objects.count(),
'users_removed': 0,
'media_paths_removed': 0,
}
OnboardingIntroductionSession.objects.all().delete()
ScheduledWelcomeEmail.objects.all().delete()
OnboardingRequest.objects.all().delete()
OffboardingRequest.objects.all().delete()
EmployeeProfile.objects.all().delete()
AdminAuditLog.objects.all().delete()
Session.objects.all().delete()
for user in user_model.objects.all():
if get_user_role_key(user) == ROLE_PLATFORM_OWNER:
continue
user.delete()
deleted_counts['users_removed'] += 1
for path in (settings.MEDIA_ROOT / 'pdfs', settings.MEDIA_ROOT / 'signatures'):
candidate = Path(path)
if candidate.exists():
shutil.rmtree(candidate, ignore_errors=True)
candidate.mkdir(parents=True, exist_ok=True)
deleted_counts['media_paths_removed'] += 1
trial_config = get_portal_trial_config()
trial_config.last_cleanup_at = timezone.now()
trial_config.save(update_fields=['last_cleanup_at', 'updated_at'])
return deleted_counts
def trial_cleanup_is_due() -> bool:
trial_config = get_portal_trial_config()
return bool(trial_config.is_trial_mode and trial_config.auto_cleanup_enabled and is_trial_expired())