Spaces:
Sleeping
Sleeping
| """ | |
| 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], | |
| } | |