Spaces:
Sleeping
Sleeping
| """ | |
| Settings, in two scopes: | |
| - Personal (per person, per org): the name shown in the console and which notices they want. | |
| Any member edits their own. | |
| - Organisation: the org's name (owner or admin) and how its emails go out (operator or above): | |
| the sender name beside the address, a reply-to address, the mailbox the composer picks by | |
| default, and whether a copy of each email is filed in the mailbox's Sent folder. | |
| Defaults live here, so a setting nobody has touched still reads as a full set of values. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from datetime import datetime, timezone | |
| from backend import firestore_client as db | |
| from backend import mailboxes, orgs | |
| _EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") | |
| _MAX_NAME = 80 | |
| NOTIFICATION_KEYS = ("campaign_finished", "emails_turned_away", "mailbox_problem", "new_signup", "weekly_summary") | |
| DEFAULT_NOTIFICATIONS = { | |
| "campaign_finished": False, | |
| "emails_turned_away": True, | |
| "mailbox_problem": True, | |
| "new_signup": False, | |
| "weekly_summary": False, | |
| } | |
| DEFAULT_EMAILS = {"from_name": "", "reply_to": "", "default_mailbox_id": "", "save_sent_copy": True, "pause_seconds": 2} | |
| # Seconds M8ion waits between two emails of a campaign. Some mail servers accept a fast burst and | |
| # then never deliver it, so campaigns are spaced out unless the organisation turns this down. | |
| MAX_PAUSE_SECONDS = 30 | |
| ORG_PROFILE_FIELDS = ("website", "domain", "country", "state", "lga", "address", "postal_code", "timezone") | |
| _DOMAIN_RE = re.compile(r"^(?=.{3,253}$)([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}$") | |
| def _now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def _personal_path(org_id: str, uid: str) -> str: | |
| return f"orgs/{org_id}/personal_settings/{uid}" | |
| def _emails_path(org_id: str) -> str: | |
| return f"orgs/{org_id}/settings/emails" | |
| def _clean_name(value: str, what: str) -> str: | |
| value = " ".join(str(value or "").split()) | |
| if len(value) > _MAX_NAME: | |
| raise ValueError(f"Keep the {what} to {_MAX_NAME} characters or fewer") | |
| return value | |
| def get_settings(org_id: str, uid: str) -> dict: | |
| """Everything the Settings pages show, with defaults filled in.""" | |
| role = orgs.require_permission(org_id, uid) | |
| permissions = sorted(orgs.permissions_for(org_id, role)) | |
| personal = db.get_doc(_personal_path(org_id, uid)) or {} | |
| return { | |
| "role": role, | |
| "permissions": permissions, | |
| "profile": profile_of(org_id, uid, personal), | |
| "organisation": organisation(org_id), | |
| "notifications": DEFAULT_NOTIFICATIONS | {k: bool(v) for k, v in (personal.get("notifications") or {}).items() if k in NOTIFICATION_KEYS}, | |
| "emails": email_settings(org_id), | |
| } | |
| def email_settings(org_id: str) -> dict: | |
| """How the org's emails go out. Read by the send itself, so it takes no actor.""" | |
| return DEFAULT_EMAILS | {k: v for k, v in (db.get_doc(_emails_path(org_id)) or {}).items() if k in DEFAULT_EMAILS} | |
| PROFILE_EXTRAS = ("job_title", "phone", "about") | |
| def profile_of(org_id: str, uid: str, personal: dict | None = None) -> dict: | |
| """The person's own profile: what they chose to share, plus facts about their use of this organisation.""" | |
| personal = personal if personal is not None else (db.get_doc(_personal_path(org_id, uid)) or {}) | |
| member = db.get_doc(f"orgs/{org_id}/members/{uid}") or {} | |
| sends = [d for d in db.query_prefix(f"orgs/{org_id}/sends/").values() if d.get("created_by") == uid] | |
| return { | |
| "display_name": personal.get("display_name", ""), | |
| **{k: personal.get(k, "") for k in PROFILE_EXTRAS}, | |
| "active_since": member.get("added_at"), | |
| "campaigns_sent": len(sends), | |
| "last_active": max((d.get("created_at") or "" for d in sends), default="") or None, | |
| } | |
| def update_profile(org_id: str, uid: str, display_name: str, extras: dict | None = None) -> dict: | |
| orgs.require_permission(org_id, uid) | |
| name = _clean_name(display_name, "name") | |
| update = {"display_name": name} | |
| for key in PROFILE_EXTRAS: | |
| if extras and key in extras: | |
| limit = 500 if key == "about" else 80 | |
| value = " ".join(str(extras[key] or "").split()) if key != "about" else str(extras[key] or "").strip() | |
| if len(value) > limit: | |
| raise ValueError(f"Keep this to {limit} characters or fewer") | |
| update[key] = value | |
| _merge(_personal_path(org_id, uid), update) | |
| return profile_of(org_id, uid) | |
| def update_notifications(org_id: str, uid: str, values: dict) -> dict: | |
| orgs.require_permission(org_id, uid) | |
| unknown = set(values) - set(NOTIFICATION_KEYS) | |
| if unknown: | |
| raise ValueError(f"Unknown notice: {', '.join(sorted(unknown))}") | |
| current = get_settings(org_id, uid)["notifications"] | {k: bool(v) for k, v in values.items()} | |
| _merge(_personal_path(org_id, uid), {"notifications": current}) | |
| return current | |
| def organisation(org_id: str) -> dict: | |
| org = orgs.get_org(org_id) | |
| profile = org.get("profile") or {} | |
| return {"name": org["name"], **{k: profile.get(k, "") for k in ORG_PROFILE_FIELDS}} | |
| def rename_org(org_id: str, uid: str, name: str, profile: dict | None = None) -> dict: | |
| """Saves the org's name and its details: website, main domain, country, state, LGA, postal | |
| address and postal code (many countries' email laws ask for one in marketing emails) and time zone.""" | |
| orgs.require_permission(org_id, uid, "manage_settings") | |
| name = _clean_name(name, "organisation name") | |
| if not name: | |
| raise ValueError("Give the organisation a name") | |
| update: dict = {"name": name} | |
| if profile is not None: | |
| unknown = set(profile) - set(ORG_PROFILE_FIELDS) | |
| if unknown: | |
| raise ValueError(f"Unknown organisation detail: {', '.join(sorted(unknown))}") | |
| clean = {k: " ".join(str(profile.get(k) or "").split()) for k in ORG_PROFILE_FIELDS} | |
| if len(clean["address"]) > 300: | |
| raise ValueError("Keep the address to 300 characters or fewer") | |
| if len(clean["postal_code"]) > 20: | |
| raise ValueError("Keep the postal code to 20 characters or fewer") | |
| for key in ("country", "state", "lga"): | |
| if len(clean[key]) > 100: | |
| raise ValueError(f"Keep the {key} to 100 characters or fewer") | |
| clean["domain"] = clean["domain"].lower().removeprefix("https://").removeprefix("http://").removeprefix("www.").rstrip("/") | |
| if clean["domain"] and not _DOMAIN_RE.match(clean["domain"]): | |
| raise ValueError("Enter a domain like yourcompany.com") | |
| if clean["website"] and not clean["website"].startswith(("https://", "http://")): | |
| clean["website"] = "https://" + clean["website"] | |
| if clean["timezone"]: | |
| try: | |
| from zoneinfo import ZoneInfo | |
| ZoneInfo(clean["timezone"]) | |
| except Exception as exc: # noqa: BLE001 | |
| raise ValueError("Pick a time zone from the list") from exc | |
| update["profile"] = clean | |
| db.update_doc(f"orgs/{org_id}", update) | |
| return organisation(org_id) | |
| def update_emails(org_id: str, uid: str, values: dict) -> dict: | |
| orgs.require_permission(org_id, uid, "manage_settings") | |
| unknown = set(values) - set(DEFAULT_EMAILS) | |
| if unknown: | |
| raise ValueError(f"Unknown email setting: {', '.join(sorted(unknown))}") | |
| clean: dict = {} | |
| if "from_name" in values: | |
| clean["from_name"] = _clean_name(values["from_name"], "sender name").replace('"', "") | |
| if "reply_to" in values: | |
| reply_to = str(values["reply_to"] or "").strip() | |
| if reply_to and not _EMAIL_RE.match(reply_to): | |
| raise ValueError("The reply-to address does not look like an email address") | |
| clean["reply_to"] = reply_to | |
| if "default_mailbox_id" in values: | |
| mailbox_id = str(values["default_mailbox_id"] or "") | |
| if mailbox_id: | |
| mailboxes.get_mailbox(org_id, mailbox_id) # raises NotFoundError for another org's id | |
| clean["default_mailbox_id"] = mailbox_id | |
| if "save_sent_copy" in values: | |
| clean["save_sent_copy"] = bool(values["save_sent_copy"]) | |
| if "pause_seconds" in values: | |
| try: | |
| pause = float(values["pause_seconds"] or 0) | |
| except (TypeError, ValueError) as exc: | |
| raise ValueError("The pause must be a number of seconds") from exc | |
| if not 0 <= pause <= MAX_PAUSE_SECONDS: | |
| raise ValueError(f"The pause must be between 0 and {MAX_PAUSE_SECONDS} seconds") | |
| clean["pause_seconds"] = pause | |
| _merge(_emails_path(org_id), clean) | |
| return email_settings(org_id) | |
| def _merge(path: str, values: dict) -> None: | |
| doc = db.get_doc(path) or {} | |
| db.set_doc(path, doc | values | {"updated_at": _now()}) | |