snapshot: preserve trial lifecycle and product-grade expiry UX

backend/workflows/trial.py

@@ -0,0 +1,59 @@
+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
+
+
+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())