import time from django.contrib.auth.mixins import LoginRequiredMixin from django.http import HttpResponseRedirect, JsonResponse from django.shortcuts import render from django.urls import reverse from django.views import View from core.clients import transport from core.models import ( AdapterHealthEvent, AIRequest, AIResult, AIResultSignal, Chat, ChatSession, ConversationEvent, Group, MemoryItem, Message, MessageEvent, PatternArtifactExport, PatternMitigationAutoSettings, PatternMitigationCorrection, PatternMitigationGame, PatternMitigationMessage, PatternMitigationPlan, PatternMitigationRule, Person, Persona, PersonIdentifier, QueuedMessage, CommandSecurityPolicy, UserAccessibilitySettings, UserXmppOmemoState, UserXmppSecuritySettings, WorkspaceConversation, WorkspaceMetricSnapshot, ) from core.events.projection import shadow_compare_session from core.memory.search_backend import backend_status, get_memory_search_backend from core.transports.capabilities import capability_snapshot 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(), "conversation_events": ConversationEvent.objects.filter(user=user).count(), "adapter_health_events": AdapterHealthEvent.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 += ConversationEvent.objects.filter(user=user).delete()[0] deleted += AdapterHealthEvent.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 _diagnostics_options(self, user): session_rows = list( ChatSession.objects.filter(user=user) .select_related("identifier", "identifier__person") .order_by("-last_interaction", "-id")[:120] ) session_options = [] for row in session_rows: identifier = getattr(row, "identifier", None) person = getattr(identifier, "person", None) if identifier else None session_options.append( { "id": str(row.id), "label": " | ".join( [ str(getattr(person, "name", "") or "-"), str(row.id), str(getattr(identifier, "service", "") or "-"), str(getattr(identifier, "identifier", "") or "-"), ] ), } ) trace_options = [] seen_trace_ids = set() for trace_id in ( ConversationEvent.objects.filter(user=user) .exclude(trace_id="") .order_by("-ts") .values_list("trace_id", flat=True)[:400] ): value = str(trace_id or "").strip() if not value or value in seen_trace_ids: continue seen_trace_ids.add(value) trace_options.append(value) if len(trace_options) >= 120: break service_candidates = {"signal", "whatsapp", "xmpp", "instagram", "web"} service_candidates.update( str(item or "").strip().lower() for item in ConversationEvent.objects.filter(user=user) .exclude(origin_transport="") .values_list("origin_transport", flat=True) .distinct()[:50] ) service_options = sorted(value for value in service_candidates if value) event_type_candidates = { "message_created", "reaction_added", "reaction_removed", "read_receipt", "message_updated", "message_deleted", } event_type_candidates.update( str(item or "").strip().lower() for item in ConversationEvent.objects.filter(user=user) .exclude(event_type="") .values_list("event_type", flat=True) .distinct()[:80] ) event_type_options = sorted(value for value in event_type_candidates if value) return { "sessions": session_options, "trace_ids": trace_options, "services": service_options, "event_types": event_type_options, } def _render_page(self, request, notice_level="", notice_message=""): return render( request, self.template_name, { "counts": self._counts(request.user), "notice_level": notice_level, "notice_message": notice_message, "diagnostics_options": self._diagnostics_options(request.user), }, ) def get(self, request): return self._render_page(request) def post(self, request): notice_level, notice_message = self._handle_action(request) return self._render_page( request, notice_level=notice_level, notice_message=notice_message, ) class ServiceCapabilitySnapshotAPI(SuperUserRequiredMixin, View): def get(self, request): service = str(request.GET.get("service") or "").strip().lower() return JsonResponse( { "ok": True, "data": capability_snapshot(service), } ) class AdapterHealthSummaryAPI(SuperUserRequiredMixin, View): def get(self, request): latest_by_service = {} rows = AdapterHealthEvent.objects.order_by("service", "-ts")[:200] for row in rows: key = str(row.service or "").strip().lower() if key in latest_by_service: continue latest_by_service[key] = { "status": str(row.status or ""), "reason": str(row.reason or ""), "ts": int(row.ts or 0), "created_at": row.created_at.isoformat(), } return JsonResponse({"ok": True, "services": latest_by_service}) class TraceDiagnosticsAPI(SuperUserRequiredMixin, View): def get(self, request): trace_id = str(request.GET.get("trace_id") or "").strip() if not trace_id: return JsonResponse( {"ok": False, "error": "trace_id_required"}, status=400, ) rows = list( ConversationEvent.objects.filter( user=request.user, trace_id=trace_id, ) .select_related("session") .order_by("ts", "created_at")[:500] ) related_session_ids = [] seen_sessions = set() for row in rows: session_id = str(row.session_id or "").strip() if not session_id or session_id in seen_sessions: continue seen_sessions.add(session_id) related_session_ids.append(session_id) return JsonResponse( { "ok": True, "trace_id": trace_id, "count": len(rows), "related_session_ids": related_session_ids, "projection_shadow_urls": [ f"{reverse('system_projection_shadow')}?session_id={session_id}" for session_id in related_session_ids ], "events": [ { "id": str(row.id), "ts": int(row.ts or 0), "event_type": str(row.event_type or ""), "direction": str(row.direction or ""), "session_id": str(row.session_id or ""), "projection_shadow_url": ( f"{reverse('system_projection_shadow')}?session_id={str(row.session_id or '').strip()}" if str(row.session_id or "").strip() else "" ), "origin_transport": str(row.origin_transport or ""), "origin_message_id": str(row.origin_message_id or ""), "payload": dict(row.payload or {}), } for row in rows ], } ) class EventProjectionShadowAPI(SuperUserRequiredMixin, View): def get(self, request): session_id = str(request.GET.get("session_id") or "").strip() if not session_id: return JsonResponse( {"ok": False, "error": "session_id_required"}, status=400, ) detail_limit = int(request.GET.get("detail_limit") or 25) session = ChatSession.objects.filter( id=session_id, user=request.user, ).first() if session is None: return JsonResponse( {"ok": False, "error": "session_not_found"}, status=404, ) compared = shadow_compare_session(session, detail_limit=max(0, detail_limit)) return JsonResponse( { "ok": True, "result": compared, "cause_summary": dict(compared.get("cause_counts") or {}), "cause_samples": dict(compared.get("cause_samples") or {}), } ) class EventLedgerSmokeAPI(SuperUserRequiredMixin, View): def get(self, request): minutes = max(1, int(request.GET.get("minutes") or 120)) service = str(request.GET.get("service") or "").strip().lower() user_id = str(request.GET.get("user_id") or "").strip() or str(request.user.id) limit = max(1, min(500, int(request.GET.get("limit") or 200))) require_types_raw = str(request.GET.get("require_types") or "").strip() required_types = [ item.strip().lower() for item in require_types_raw.split(",") if item.strip() ] cutoff_ts = int(time.time() * 1000) - (minutes * 60 * 1000) queryset = ConversationEvent.objects.filter(ts__gte=cutoff_ts).order_by("-ts") if service: queryset = queryset.filter(origin_transport=service) if user_id: queryset = queryset.filter(user_id=user_id) rows = list( queryset.values( "id", "user_id", "session_id", "ts", "event_type", "direction", "origin_transport", "trace_id", )[:limit] ) event_type_counts = {} for row in rows: key = str(row.get("event_type") or "") event_type_counts[key] = int(event_type_counts.get(key) or 0) + 1 missing_required_types = [ event_type for event_type in required_types if int(event_type_counts.get(event_type) or 0) <= 0 ] return JsonResponse( { "ok": True, "minutes": minutes, "service": service, "user_id": user_id, "count": len(rows), "event_type_counts": event_type_counts, "required_types": required_types, "missing_required_types": missing_required_types, "sample": rows[:25], } ) class MemorySearchStatusAPI(SuperUserRequiredMixin, View): def get(self, request): return JsonResponse({"ok": True, "status": backend_status()}) class MemorySearchQueryAPI(SuperUserRequiredMixin, View): def get(self, request): query = str(request.GET.get("q") or "").strip() user_id = int(request.GET.get("user_id") or request.user.id) conversation_id = str(request.GET.get("conversation_id") or "").strip() limit = max(1, min(50, int(request.GET.get("limit") or 20))) statuses = tuple( item.strip().lower() for item in str(request.GET.get("statuses") or "active").split(",") if item.strip() ) if not query: return JsonResponse({"ok": False, "error": "query_required"}, status=400) backend = get_memory_search_backend() hits = backend.search( user_id=user_id, query=query, conversation_id=conversation_id, limit=limit, include_statuses=statuses, ) return JsonResponse( { "ok": True, "backend": getattr(backend, "name", "unknown"), "query": query, "count": len(hits), "hits": [ { "memory_id": item.memory_id, "score": item.score, "summary": item.summary, "payload": item.payload, } for item in hits ], } ) def _parse_xmpp_jid(jid_str: str) -> dict: """Split a full JID (localpart@domain/resource) into components.""" raw = str(jid_str or "").strip() bare, _, resource = raw.partition("/") localpart, _, domain = bare.partition("@") return {"full": raw, "bare": bare, "localpart": localpart, "domain": domain, "resource": resource} def _to_bool(value, default=False): if value is None: return bool(default) text = str(value).strip().lower() if text in {"1", "true", "yes", "on", "y"}: return True if text in {"0", "false", "no", "off", "n"}: return False return bool(default) class SecurityPage(LoginRequiredMixin, View): """Security settings page for OMEMO and command-scope policy controls.""" template_name = "pages/security.html" page_mode = "encryption" GLOBAL_SCOPE_KEY = "global.override" # Allowed Services list used by both Global Scope Override and local scopes. # Keep this in sync with the UI text on the Security page. POLICY_SERVICES = ["xmpp", "whatsapp", "signal", "instagram", "web"] # Override mode names as shown in the interface: # - per_scope: local scope controls remain editable # - on/off: global override forces each local scope value OVERRIDE_OPTIONS = ("per_scope", "on", "off") GLOBAL_OVERRIDE_FIELDS = ( "scope_enabled", "require_omemo", "require_trusted_fingerprint", ) POLICY_SCOPES = [ ("gateway.tasks", "Gateway .tasks commands", "Handles .tasks list/show/complete/undo over gateway channels."), ("gateway.approval", "Gateway approval commands", "Handles .approval/.codex/.claude approve/deny over gateway channels."), ("gateway.totp", "Gateway TOTP enrollment", "Controls TOTP enrollment/status commands over gateway channels."), ("tasks.submit", "Task submissions from chat", "Controls automatic task creation from inbound messages."), ("tasks.commands", "Task command verbs (.task/.undo/.epic)", "Controls explicit task command verbs."), ("command.bp", "Business plan command", "Controls Business Plan command execution."), ("command.codex", "Codex command", "Controls Codex command execution."), ("command.claude", "Claude command", "Controls Claude command execution."), ] POLICY_GROUP_LABELS = { "gateway": "Gateway", "tasks": "Tasks", "command": "Commands", "agentic": "Agentic", "other": "Other", } def _show_encryption(self) -> bool: return str(getattr(self, "page_mode", "encryption")).strip().lower() in { "encryption", "all", } def _show_permission(self) -> bool: return str(getattr(self, "page_mode", "encryption")).strip().lower() in { "permission", "all", } def _security_settings(self, request): row, _ = UserXmppSecuritySettings.objects.get_or_create(user=request.user) return row def _parse_override_value(self, value): option = str(value or "").strip().lower() if option == "inherit": # Backward-compat for existing persisted values. option = "per_scope" if option in self.OVERRIDE_OPTIONS: return option return "per_scope" def _global_override_payload(self, request): row, _ = CommandSecurityPolicy.objects.get_or_create( user=request.user, scope_key=self.GLOBAL_SCOPE_KEY, defaults={ "enabled": True, "allowed_services": [], "allowed_channels": {}, "settings": {}, }, ) settings_payload = dict(row.settings or {}) values = { "scope_enabled": self._parse_override_value( settings_payload.get("scope_enabled") ), "require_omemo": self._parse_override_value( settings_payload.get("require_omemo") ), "require_trusted_fingerprint": self._parse_override_value( settings_payload.get("require_trusted_fingerprint") ), } allowed_services = [ str(value or "").strip().lower() for value in (row.allowed_services or []) if str(value or "").strip() ] channel_rules = self._channel_rules_from_map(dict(row.allowed_channels or {})) if not channel_rules: channel_rules = [{"service": "xmpp", "pattern": ""}] return { "row": row, "values": values, "allowed_services": allowed_services, "channel_rules": channel_rules, } def _apply_global_override(self, current_value: bool, option: str) -> bool: normalized = self._parse_override_value(option) if normalized == "on": return True if normalized == "off": return False return bool(current_value) def _channel_rules_from_map(self, source_map): rows = [] raw = dict(source_map or {}) for service_key, patterns in raw.items(): service_name = str(service_key or "").strip().lower() if not service_name: continue if isinstance(patterns, list): for pattern in patterns: pattern_text = str(pattern or "").strip() if pattern_text: rows.append({ "service": service_name, "pattern": pattern_text, }) return rows def _channels_map_from_post(self, request): channel_services = request.POST.getlist("allowed_channel_service") channel_patterns = request.POST.getlist("allowed_channel_pattern") allowed_channels: dict[str, list[str]] = {} for idx, raw_pattern in enumerate(channel_patterns): pattern = str(raw_pattern or "").strip() if not pattern: continue service_name = str( channel_services[idx] if idx < len(channel_services) else "" ).strip().lower() if not service_name: service_name = "*" allowed_channels.setdefault(service_name, []) if pattern not in allowed_channels[service_name]: allowed_channels[service_name].append(pattern) return allowed_channels def _scope_rows(self, request): global_overrides = self._global_override_payload(request)["values"] security_settings = self._security_settings(request) rows = { str(item.scope_key or "").strip().lower(): item for item in CommandSecurityPolicy.objects.filter(user=request.user).exclude( scope_key=self.GLOBAL_SCOPE_KEY ) } payload = [] for scope_key, label, description in self.POLICY_SCOPES: key = str(scope_key or "").strip().lower() item = rows.get(key) raw_allowed_services = [ str(value or "").strip().lower() for value in (getattr(item, "allowed_services", []) or []) if str(value or "").strip() ] channel_rules = self._channel_rules_from_map( dict(getattr(item, "allowed_channels", {}) or {}) ) if not channel_rules: channel_rules = [{"service": "xmpp", "pattern": ""}] enabled_locked = global_overrides["scope_enabled"] != "per_scope" require_omemo_locked = ( global_overrides["require_omemo"] != "per_scope" or bool(security_settings.require_omemo) ) require_trusted_locked = ( global_overrides["require_trusted_fingerprint"] != "per_scope" ) payload.append({ "scope_key": key, "label": label, "description": description, "enabled": self._apply_global_override( bool(getattr(item, "enabled", True)), global_overrides["scope_enabled"], ), "require_omemo": self._apply_global_override( bool(getattr(item, "require_omemo", False)), global_overrides["require_omemo"], ), "require_trusted_fingerprint": self._apply_global_override( bool(getattr(item, "require_trusted_omemo_fingerprint", False)), global_overrides["require_trusted_fingerprint"], ), "enabled_locked": enabled_locked, "require_omemo_locked": require_omemo_locked, "require_trusted_fingerprint_locked": require_trusted_locked, "lock_help": "Set this field to 'Per Scope' in Global Scope Override to edit it here.", "require_omemo_lock_help": ( "Disable 'Require OMEMO encryption' in Encryption settings to edit this field." if bool(security_settings.require_omemo) else "Set this field to 'Per Scope' in Global Scope Override to edit it here." ), "allowed_services": raw_allowed_services, "channel_rules": channel_rules, }) return payload def _scope_group_key(self, scope_key: str) -> str: key = str(scope_key or "").strip().lower() if key in {"tasks.commands", "gateway.tasks"}: return "tasks" if key in {"command.codex", "command.claude"}: return "agentic" if key.startswith("gateway."): return "command" if key.startswith("tasks."): if key == "tasks.submit": return "tasks" return "command" if key.startswith("command."): return "command" if ".commands" in key: return "command" if ".approval" in key: return "command" if ".totp" in key: return "command" if ".task" in key: return "tasks" return "other" def _grouped_scope_rows(self, request): rows = self._scope_rows(request) grouped: dict[str, list[dict]] = {key: [] for key in self.POLICY_GROUP_LABELS} for row in rows: group_key = self._scope_group_key(row.get("scope_key")) grouped.setdefault(group_key, []) grouped[group_key].append(row) payload = [] for group_key in ("tasks", "command", "agentic", "other"): items = grouped.get(group_key) or [] if not items: continue payload.append({ "key": group_key, "label": self.POLICY_GROUP_LABELS.get(group_key, group_key.title()), "rows": items, }) return payload def post(self, request): row = self._security_settings(request) if "require_omemo" in request.POST: row.require_omemo = _to_bool(request.POST.get("require_omemo"), False) row.save(update_fields=["require_omemo", "updated_at"]) redirect_to = HttpResponseRedirect(request.path) scope_key = str(request.POST.get("scope_key") or "").strip().lower() if not self._show_permission(): return redirect_to if scope_key == self.GLOBAL_SCOPE_KEY: global_row = self._global_override_payload(request)["row"] security_settings = self._security_settings(request) settings_payload = dict(global_row.settings or {}) for field in self.GLOBAL_OVERRIDE_FIELDS: if field == "require_omemo" and bool(security_settings.require_omemo): continue settings_payload[field] = self._parse_override_value( request.POST.get(f"global_{field}") ) global_row.allowed_services = [ str(item or "").strip().lower() for item in request.POST.getlist("allowed_services") if str(item or "").strip() ] global_row.allowed_channels = self._channels_map_from_post(request) global_row.settings = settings_payload global_row.save( update_fields=[ "settings", "allowed_services", "allowed_channels", "updated_at", ] ) return redirect_to if scope_key: if str(request.POST.get("scope_change_mode") or "").strip() != "1": return redirect_to global_overrides = self._global_override_payload(request)["values"] security_settings = self._security_settings(request) allowed_services = [ str(item or "").strip().lower() for item in request.POST.getlist("allowed_services") if str(item or "").strip() ] allowed_channels = self._channels_map_from_post(request) policy, _ = CommandSecurityPolicy.objects.get_or_create( user=request.user, scope_key=scope_key, ) policy.allowed_services = allowed_services policy.allowed_channels = allowed_channels if global_overrides["scope_enabled"] == "per_scope": policy.enabled = _to_bool(request.POST.get("policy_enabled"), True) if ( global_overrides["require_omemo"] == "per_scope" and not bool(security_settings.require_omemo) ): policy.require_omemo = _to_bool( request.POST.get("policy_require_omemo"), False ) if global_overrides["require_trusted_fingerprint"] == "per_scope": policy.require_trusted_omemo_fingerprint = _to_bool( request.POST.get("policy_require_trusted_fingerprint"), False, ) policy.save( update_fields=[ "enabled", "require_omemo", "require_trusted_omemo_fingerprint", "allowed_services", "allowed_channels", "updated_at", ] ) return redirect_to def get(self, request): show_encryption = self._show_encryption() show_permission = self._show_permission() xmpp_state = transport.get_runtime_state("xmpp") if show_encryption else {} omemo_row = None if show_encryption: try: omemo_row = UserXmppOmemoState.objects.get(user=request.user) except UserXmppOmemoState.DoesNotExist: omemo_row = None security_settings = self._security_settings(request) sender_jid = _parse_xmpp_jid(getattr(omemo_row, "last_sender_jid", "") or "") omemo_plan = [] if show_encryption: omemo_plan = [ { "label": "Component OMEMO active", "done": bool(xmpp_state.get("omemo_enabled")), "hint": "The gateway's OMEMO plugin must be loaded and initialised.", }, { "label": "OMEMO observed from your client", "done": omemo_row is not None and omemo_row.status == "detected", "hint": "Send any message with OMEMO enabled in your XMPP client.", }, { "label": "Client key on file", "done": bool(getattr(omemo_row, "latest_client_key", "")), "hint": "A device key (sid/rid) must be recorded from your client.", }, { "label": "Encryption required", "done": security_settings.require_omemo, "hint": "Enable 'Require OMEMO encryption' in Security Policy above to enforce this policy.", }, ] return render(request, self.template_name, { "xmpp_state": xmpp_state, "omemo_row": omemo_row, "security_settings": security_settings, "global_override": self._global_override_payload(request), "policy_services": self.POLICY_SERVICES, "policy_rows": self._scope_rows(request), "policy_groups": self._grouped_scope_rows(request), "sender_jid": sender_jid, "omemo_plan": omemo_plan, "show_encryption": show_encryption, "show_permission": show_permission, }) class AccessibilitySettings(LoginRequiredMixin, View): template_name = "pages/accessibility-settings.html" def _row(self, request): row, _ = UserAccessibilitySettings.objects.get_or_create(user=request.user) return row def get(self, request): return render(request, self.template_name, { "accessibility_settings": self._row(request), }) def post(self, request): row = self._row(request) row.disable_animations = _to_bool(request.POST.get("disable_animations"), False) row.save(update_fields=["disable_animations", "updated_at"]) return HttpResponseRedirect(reverse("accessibility_settings")) class _SettingsCategoryPage(LoginRequiredMixin, View): template_name = "pages/settings-category.html" category_key = "general" category_title = "General" category_description = "" tabs = () def _tab_rows(self): current_path = str(getattr(self.request, "path", "") or "") rows = [] for label, href in self.tabs: rows.append({ "label": label, "href": href, "active": current_path == href, }) return rows def get(self, request): return render(request, self.template_name, { "category_key": self.category_key, "category_title": self.category_title, "category_description": self.category_description, "category_tabs": self._tab_rows(), }) class AISettingsPage(LoginRequiredMixin, View): def get(self, request): return HttpResponseRedirect(reverse("ai_models")) class ModulesSettingsPage(_SettingsCategoryPage): def get(self, request): return HttpResponseRedirect(reverse("command_routing"))