""" The dashboard's numbers in one call: lists and people, mailboxes and their health, recent sends and how they went, forms and presets. Read-only, and every count comes from the same stores the other pages read, so the dashboard never disagrees with them. """ from __future__ import annotations from datetime import datetime, timedelta, timezone from backend import collect, databases, mailboxes, orgs, presets, sending _RECENT_DAYS = 30 def _parse(ts: str | None) -> datetime | None: if not ts: return None try: return datetime.fromisoformat(ts) except ValueError: return None def get_overview(org_id: str, actor_uid: str) -> dict: orgs.require_permission(org_id, actor_uid) lists = [{"database_id": did, **d} for did, d in databases.list_databases(org_id).items()] boxes = list(mailboxes.list_mailboxes(org_id).values()) sends = sending.list_sends(org_id, actor_uid, limit=500) forms = [k for k in collect.list_keys(org_id, actor_uid) if not k.get("revoked")] saved_presets = presets.list_presets(org_id, actor_uid) cutoff = datetime.now(timezone.utc) - timedelta(days=_RECENT_DAYS) recent = [s for s in sends if (_parse(s["sent_at"]) or cutoff) > cutoff] delivered = sum(s["sent"] for s in recent) failed = sum(s["failed"] for s in recent) # One point per day for the last 30 days, oldest first, for the activity chart. by_day: dict[str, int] = {} for s in recent: day = (_parse(s["sent_at"]) or cutoff).date().isoformat() by_day[day] = by_day.get(day, 0) + s["sent"] today = datetime.now(timezone.utc).date() activity = [ {"day": (today - timedelta(days=i)).isoformat(), "sent": by_day.get((today - timedelta(days=i)).isoformat(), 0)} for i in range(_RECENT_DAYS - 1, -1, -1) ] names = {d["database_id"]: d["name"] for d in lists} return { "people": sum(d.get("row_count", 0) for d in lists if not d.get("disabled")), "lists": len(lists), "mailboxes": { "total": len(boxes), "verified": sum(1 for b in boxes if b.get("status") == "verified"), "needs_attention": sum(1 for b in boxes if b.get("status") not in ("verified", "pending_verification") and not b.get("dismissed")), }, "sends_30d": {"campaigns": len(recent), "delivered": delivered, "failed": failed}, "forms": len(forms), "presets": len(saved_presets), "activity": activity, "recent_sends": [s | {"list_name": names.get(s["database_id"], "Deleted list")} for s in sends[:8]], "largest_lists": sorted( ({"database_id": d["database_id"], "name": d["name"], "row_count": d.get("row_count", 0)} for d in lists), key=lambda d: d["row_count"], reverse=True, )[:5], }