from django.shortcuts import render from django.views import View from core.models import ( AIRequest, AIResult, AIResultSignal, Chat, ChatSession, Group, MemoryItem, Message, MessageEvent, PatternArtifactExport, PatternMitigationAutoSettings, PatternMitigationCorrection, PatternMitigationGame, PatternMitigationMessage, PatternMitigationPlan, PatternMitigationRule, Person, PersonIdentifier, Persona, QueuedMessage, WorkspaceConversation, WorkspaceMetricSnapshot, ) from core.views.manage.permissions import SuperUserRequiredMixin class SystemSettings(SuperUserRequiredMixin, View): template_name = "pages/system-settings.html" def _counts(self, user): return { "chat_sessions": ChatSession.objects.filter(user=user).count(), "messages": Message.objects.filter(user=user).count(), "queued_messages": QueuedMessage.objects.filter(user=user).count(), "message_events": MessageEvent.objects.filter(user=user).count(), "workspace_conversations": WorkspaceConversation.objects.filter(user=user).count(), "workspace_snapshots": WorkspaceMetricSnapshot.objects.filter( conversation__user=user ).count(), "ai_requests": AIRequest.objects.filter(user=user).count(), "ai_results": AIResult.objects.filter(user=user).count(), "ai_result_signals": AIResultSignal.objects.filter(user=user).count(), "memory_items": MemoryItem.objects.filter(user=user).count(), "mitigation_plans": PatternMitigationPlan.objects.filter(user=user).count(), "mitigation_rules": PatternMitigationRule.objects.filter(user=user).count(), "mitigation_games": PatternMitigationGame.objects.filter(user=user).count(), "mitigation_corrections": PatternMitigationCorrection.objects.filter( user=user ).count(), "mitigation_messages": PatternMitigationMessage.objects.filter( user=user ).count(), "mitigation_auto_settings": PatternMitigationAutoSettings.objects.filter( user=user ).count(), "mitigation_exports": PatternArtifactExport.objects.filter(user=user).count(), "osint_people": Person.objects.filter(user=user).count(), "osint_identifiers": PersonIdentifier.objects.filter(user=user).count(), "osint_groups": Group.objects.filter(user=user).count(), "osint_personas": Persona.objects.filter(user=user).count(), } def _purge_non_osint(self, user): deleted = 0 deleted += PatternArtifactExport.objects.filter(user=user).delete()[0] deleted += PatternMitigationMessage.objects.filter(user=user).delete()[0] deleted += PatternMitigationCorrection.objects.filter(user=user).delete()[0] deleted += PatternMitigationGame.objects.filter(user=user).delete()[0] deleted += PatternMitigationRule.objects.filter(user=user).delete()[0] deleted += PatternMitigationAutoSettings.objects.filter(user=user).delete()[0] deleted += PatternMitigationPlan.objects.filter(user=user).delete()[0] deleted += AIResultSignal.objects.filter(user=user).delete()[0] deleted += AIResult.objects.filter(user=user).delete()[0] deleted += AIRequest.objects.filter(user=user).delete()[0] deleted += MemoryItem.objects.filter(user=user).delete()[0] deleted += WorkspaceMetricSnapshot.objects.filter(conversation__user=user).delete()[0] deleted += MessageEvent.objects.filter(user=user).delete()[0] deleted += Message.objects.filter(user=user).delete()[0] deleted += QueuedMessage.objects.filter(user=user).delete()[0] deleted += WorkspaceConversation.objects.filter(user=user).delete()[0] deleted += ChatSession.objects.filter(user=user).delete()[0] # Chat rows are legacy Signal cache rows and are not user-scoped. deleted += Chat.objects.all().delete()[0] return deleted def _purge_osint_people(self, user): return Person.objects.filter(user=user).delete()[0] def _purge_osint_identifiers(self, user): return PersonIdentifier.objects.filter(user=user).delete()[0] def _purge_osint_groups(self, user): return Group.objects.filter(user=user).delete()[0] def _purge_osint_personas(self, user): return Persona.objects.filter(user=user).delete()[0] def _handle_action(self, request): action = str(request.POST.get("action") or "").strip().lower() if action == "purge_non_osint": return ( "success", f"Purged {self._purge_non_osint(request.user)} non-OSINT row(s).", ) if action == "purge_osint_people": return ( "warning", f"Purged {self._purge_osint_people(request.user)} OSINT people row(s).", ) if action == "purge_osint_identifiers": return ( "warning", f"Purged {self._purge_osint_identifiers(request.user)} OSINT identifier row(s).", ) if action == "purge_osint_groups": return ( "warning", f"Purged {self._purge_osint_groups(request.user)} OSINT group row(s).", ) if action == "purge_osint_personas": return ( "warning", f"Purged {self._purge_osint_personas(request.user)} OSINT persona row(s).", ) return ("danger", "Unknown action.") def get(self, request): return render( request, self.template_name, { "counts": self._counts(request.user), "notice_level": "", "notice_message": "", }, ) def post(self, request): notice_level, notice_message = self._handle_action(request) return render( request, self.template_name, { "counts": self._counts(request.user), "notice_level": notice_level, "notice_message": notice_message, }, )