167 lines
6.1 KiB
Python
167 lines
6.1 KiB
Python
from __future__ import annotations
|
|
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.shortcuts import render
|
|
from django.views import View
|
|
|
|
from core.events.manticore import get_behavioral_availability_stats
|
|
from core.events.shadow import get_shadow_behavioral_availability_stats
|
|
from core.models import ContactAvailabilitySettings, Person
|
|
|
|
|
|
def _to_int(value, default=0):
|
|
try:
|
|
return int(value)
|
|
except Exception:
|
|
return int(default)
|
|
|
|
|
|
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 AvailabilitySettingsPage(LoginRequiredMixin, View):
|
|
template_name = "pages/availability-settings.html"
|
|
|
|
def _settings(self, request):
|
|
row, _ = ContactAvailabilitySettings.objects.get_or_create(user=request.user)
|
|
return row
|
|
|
|
def post(self, request):
|
|
row = self._settings(request)
|
|
row.enabled = _to_bool(request.POST.get("enabled"), row.enabled)
|
|
row.show_in_chat = _to_bool(request.POST.get("show_in_chat"), row.show_in_chat)
|
|
row.show_in_groups = _to_bool(
|
|
request.POST.get("show_in_groups"), row.show_in_groups
|
|
)
|
|
row.inference_enabled = _to_bool(
|
|
request.POST.get("inference_enabled"), row.inference_enabled
|
|
)
|
|
row.retention_days = max(1, _to_int(request.POST.get("retention_days"), 90))
|
|
row.fade_threshold_seconds = max(
|
|
30, _to_int(request.POST.get("fade_threshold_seconds"), 900)
|
|
)
|
|
row.save(
|
|
update_fields=[
|
|
"enabled",
|
|
"show_in_chat",
|
|
"show_in_groups",
|
|
"inference_enabled",
|
|
"retention_days",
|
|
"fade_threshold_seconds",
|
|
"updated_at",
|
|
]
|
|
)
|
|
return self.get(request)
|
|
|
|
def get(self, request):
|
|
behavioral_stats, stats_source = self._behavioral_stats(request.user)
|
|
transport_stats = self._transport_stats(behavioral_stats)
|
|
totals = self._totals(behavioral_stats)
|
|
context = {
|
|
"settings_row": self._settings(request),
|
|
"behavioral_stats": behavioral_stats,
|
|
"behavioral_stats_source": stats_source,
|
|
"transport_stats": transport_stats,
|
|
"behavioral_totals": totals,
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
def _behavioral_stats(self, user):
|
|
try:
|
|
person_map = {
|
|
str(row["id"]): str(row["name"] or "")
|
|
for row in Person.objects.filter(user=user).values("id", "name")
|
|
}
|
|
rows = []
|
|
for row in list(
|
|
get_behavioral_availability_stats(user_id=int(user.id)) or []
|
|
):
|
|
person_id = str(row.get("person_id") or "").strip()
|
|
rows.append(
|
|
{
|
|
"person_id": person_id,
|
|
"person_name": person_map.get(person_id, person_id or "-"),
|
|
"service": str(row.get("transport") or "").strip().lower(),
|
|
"total_events": _to_int(row.get("total_events"), 0),
|
|
"presence_events": _to_int(row.get("presence_events"), 0),
|
|
"read_events": _to_int(row.get("read_events"), 0),
|
|
"typing_events": _to_int(row.get("typing_events"), 0),
|
|
"message_events": _to_int(row.get("message_events"), 0),
|
|
"abandoned_events": _to_int(row.get("abandoned_events"), 0),
|
|
"last_event_ts": _to_int(row.get("last_event_ts"), 0),
|
|
}
|
|
)
|
|
if rows:
|
|
return rows, "manticore"
|
|
except Exception:
|
|
pass
|
|
return list(get_shadow_behavioral_availability_stats(user=user)), "conversation_event_shadow"
|
|
|
|
def _transport_stats(self, behavioral_stats: list[dict]) -> list[dict]:
|
|
by_transport = {}
|
|
for row in list(behavioral_stats or []):
|
|
service = str(row.get("service") or "").strip().lower() or "-"
|
|
state = by_transport.setdefault(
|
|
service,
|
|
{
|
|
"service": service,
|
|
"contacts": 0,
|
|
"total_events": 0,
|
|
"presence_events": 0,
|
|
"read_events": 0,
|
|
"typing_events": 0,
|
|
"message_events": 0,
|
|
"abandoned_events": 0,
|
|
"last_event_ts": 0,
|
|
},
|
|
)
|
|
state["contacts"] += 1
|
|
for key in (
|
|
"total_events",
|
|
"presence_events",
|
|
"read_events",
|
|
"typing_events",
|
|
"message_events",
|
|
"abandoned_events",
|
|
):
|
|
state[key] += _to_int(row.get(key), 0)
|
|
state["last_event_ts"] = max(
|
|
int(state.get("last_event_ts") or 0),
|
|
_to_int(row.get("last_event_ts"), 0),
|
|
)
|
|
return sorted(
|
|
by_transport.values(),
|
|
key=lambda row: (-int(row.get("total_events") or 0), str(row.get("service") or "")),
|
|
)
|
|
|
|
def _totals(self, behavioral_stats: list[dict]) -> dict:
|
|
totals = {
|
|
"contacts": 0,
|
|
"total_events": 0,
|
|
"presence_events": 0,
|
|
"read_events": 0,
|
|
"typing_events": 0,
|
|
"message_events": 0,
|
|
"abandoned_events": 0,
|
|
}
|
|
for row in list(behavioral_stats or []):
|
|
totals["contacts"] += 1
|
|
for key in (
|
|
"total_events",
|
|
"presence_events",
|
|
"read_events",
|
|
"typing_events",
|
|
"message_events",
|
|
"abandoned_events",
|
|
):
|
|
totals[key] += _to_int(row.get(key), 0)
|
|
return totals
|