Spaces:
Sleeping
Sleeping
| """ | |
| Audience "databases" — a parsed upload (CSV/XLSX/JSON), cleaned, classified, and normalized | |
| (see backend/normalize.py for exactly what "normalized" means here and why it stops short of | |
| blind full 5NF). Replaces the old flat backend/audiences.py contact list: a row now belongs to | |
| a named, typed database rather than floating in one flat per-org list. | |
| Column types: email, contact (phone), name, tag, bool, link, text (default). The auto- | |
| classifier runs once at upload time; a column's type can be changed later via | |
| update_column_type, which requires at least one existing value to conform to the new type's | |
| syntax and flags (not blocks) every row that doesn't. | |
| Encryption follows the same posture as the old contacts module (SRS FR-15): row data and | |
| lookup-table data are both encrypted at rest via backend.crypto_utils, same mechanism as | |
| mailbox credentials. | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import io | |
| import json | |
| import re | |
| import uuid | |
| from datetime import datetime, timezone | |
| from backend import firestore_client as db | |
| from backend import normalize | |
| from backend.templating import split_items | |
| from backend import orgs | |
| from backend.crypto_utils import decrypt, encrypt | |
| COLUMN_TYPES: tuple[str, ...] = ("email", "contact", "name", "tag", "bool", "link", "date", "text") | |
| EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") | |
| # Sheets exports often prefix phone numbers with a stray apostrophe ('+234...) to stop them | |
| # being read as numbers; tolerate it rather than call the whole column text. | |
| PHONE_RE = re.compile(r"^'?\+?[\d][\d\s\-().]{5,}$") | |
| # Dates and timestamps: 2004-05-17, 2024/03/09 14:22:01 GMT+1, 17/05/2004, 17 May 2004, | |
| # 2025-01-02T10:00:00.000Z. Checked before PHONE_RE, which would otherwise swallow digits-and-dashes. | |
| _TIME = r"\d{1,2}:\d{2}(:\d{2}(\.\d+)?)?(\s*(am|pm))?(\s*(z|gmt|utc|[a-z]{3})?\s*([+-]\d{1,2}(:?\d{2})?)?)?" | |
| DATE_RE = re.compile( | |
| rf"^(\d{{4}}[-/.]\d{{1,2}}[-/.]\d{{1,2}}([ T]{_TIME})?" | |
| rf"|\d{{1,2}}[-/.]\d{{1,2}}[-/.]\d{{2,4}}([ T]{_TIME})?" | |
| r"|\d{1,2}\s+[A-Za-z]{3,9}\.?,?\s+\d{4}" | |
| r"|[A-Za-z]{3,9}\.?\s+\d{1,2},?\s+\d{4})$", | |
| re.IGNORECASE, | |
| ) | |
| # "18 - 24", "25-34": a range of small numbers, not a phone number. | |
| RANGE_RE = re.compile(r"^\d{1,3}\s*(-|to|–)\s*\d{1,3}\+?$|^\d{1,3}\+$", re.IGNORECASE) | |
| URL_RE = re.compile(r"^(https?://|www\.)\S+$", re.IGNORECASE) | |
| BOOL_VALUES = {"true", "false", "yes", "no", "y", "n", "1", "0"} | |
| # None of these three ever raise for an empty cell — an unfilled field isn't a conformance | |
| # failure, only a filled-but-malformed one is. | |
| TYPE_VALIDATORS = { | |
| "email": lambda v: bool(EMAIL_RE.match(v.strip())), | |
| "contact": lambda v: bool(PHONE_RE.match(v.strip())) and not DATE_RE.match(v.strip()), | |
| "date": lambda v: bool(DATE_RE.match(v.strip())), | |
| "link": lambda v: bool(URL_RE.match(v.strip())), | |
| "bool": lambda v: v.strip().lower() in BOOL_VALUES, | |
| } | |
| class ValidationError(Exception): | |
| pass | |
| def _now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def _database_path(org_id: str, database_id: str) -> str: | |
| return f"orgs/{org_id}/databases/{database_id}" | |
| def _row_path(org_id: str, database_id: str, row_id: str) -> str: | |
| return f"orgs/{org_id}/databases/{database_id}/rows/{row_id}" | |
| def _rows_prefix(org_id: str, database_id: str) -> str: | |
| return f"orgs/{org_id}/databases/{database_id}/rows/" | |
| def _lookup_path(org_id: str, database_id: str, lookup_name: str) -> str: | |
| return f"orgs/{org_id}/databases/{database_id}/lookups/{lookup_name}" | |
| # --- Parsing ------------------------------------------------------------------ | |
| def _parse_csv(file_bytes: bytes) -> tuple[list[str], list[dict]]: | |
| text = file_bytes.decode("utf-8-sig") | |
| reader = csv.DictReader(io.StringIO(text)) | |
| headers = list(reader.fieldnames or []) | |
| return headers, [dict(row) for row in reader] | |
| def _parse_json(file_bytes: bytes) -> tuple[list[str], list[dict]]: | |
| data = json.loads(file_bytes.decode("utf-8")) | |
| if isinstance(data, dict): | |
| data = data.get("rows") or data.get("data") or list(data.values()) | |
| if not isinstance(data, list) or not data or not all(isinstance(r, dict) for r in data): | |
| raise ValidationError("JSON upload must be a non-empty array of objects") | |
| headers = list(dict.fromkeys(k for row in data for k in row.keys())) | |
| return headers, [{h: row.get(h) for h in headers} for row in data] | |
| def _parse_xlsx(file_bytes: bytes) -> tuple[list[str], list[dict]]: | |
| import openpyxl # only needed for this one branch — keep the import lazy | |
| workbook = openpyxl.load_workbook(io.BytesIO(file_bytes), read_only=True, data_only=True) | |
| sheet = workbook.active | |
| rows_iter = sheet.iter_rows(values_only=True) | |
| try: | |
| header_row = next(rows_iter) | |
| except StopIteration: | |
| raise ValidationError("Spreadsheet has no rows") | |
| headers = [str(h) if h is not None else "" for h in header_row] | |
| rows = [ | |
| {headers[i]: (r[i] if i < len(r) else None) for i in range(len(headers))} | |
| for r in rows_iter | |
| ] | |
| return headers, rows | |
| _PARSERS = {"csv": _parse_csv, "json": _parse_json, "xlsx": _parse_xlsx, "xls": _parse_xlsx} | |
| def _clean(headers: list[str], rows: list[dict]) -> tuple[list[str], list[dict]]: | |
| """Trims whitespace, drops blank/duplicate headers, drops fully-empty rows and columns.""" | |
| deduped: list[str] = [] | |
| seen: set[str] = set() | |
| for h in headers: | |
| h = (h or "").strip() | |
| if not h: | |
| continue | |
| base, i = h, 1 | |
| while h in seen: | |
| i += 1 | |
| h = f"{base}_{i}" | |
| seen.add(h) | |
| deduped.append(h) | |
| cleaned_rows = [] | |
| for row in rows: | |
| new_row = {h: (str(row.get(h)).strip() if row.get(h) not in (None, "") else "") for h in deduped} | |
| if any(new_row.values()): | |
| cleaned_rows.append(new_row) | |
| nonempty_cols = [h for h in deduped if any(r[h] for r in cleaned_rows)] | |
| cleaned_rows = [{h: r[h] for h in nonempty_cols} for r in cleaned_rows] | |
| return nonempty_cols, cleaned_rows | |
| # --- Classification ------------------------------------------------------------- | |
| def _match_ratio(values: list[str], predicate) -> float: | |
| nonempty = [v for v in values if v] | |
| if not nonempty: | |
| return 0.0 | |
| return sum(1 for v in nonempty if predicate(v)) / len(nonempty) | |
| _PERSON_HEADER_RE = re.compile(r"^(full |first |last |sur|given |your |participant |contact )?(name|surname|fullname)\b", re.IGNORECASE) | |
| def _looks_like_person_name_header(header: str) -> bool: | |
| """A header that merely *contains* "name" is often not a person ("Name of School / | |
| Institution", "Username", "Company name"): only headers that lead with name-ish words count.""" | |
| return bool(_PERSON_HEADER_RE.match(header.strip())) and not re.search( | |
| r"\b(of|school|institution|organi[sz]ation|company|business|user|file|team|group)\b", header, re.IGNORECASE | |
| ) | |
| def classify_column(header: str, values: list[str]) -> str: | |
| if _match_ratio(values, lambda v: bool(EMAIL_RE.match(v))) >= 0.6: | |
| return "email" | |
| if _match_ratio(values, lambda v: bool(URL_RE.match(v))) >= 0.6: | |
| return "link" | |
| if _match_ratio(values, lambda v: v.lower() in BOOL_VALUES) >= 0.9: | |
| return "bool" | |
| if _match_ratio(values, lambda v: bool(DATE_RE.match(v))) >= 0.6: | |
| return "date" | |
| if _match_ratio(values, lambda v: bool(PHONE_RE.match(v)) and not RANGE_RE.match(v)) >= 0.6: | |
| return "contact" | |
| if _match_ratio(values, lambda v: ";" in v) >= 0.3: | |
| return "tag" | |
| if _looks_like_person_name_header(header): | |
| return "name" | |
| return "text" | |
| LIST_DELIMITERS = (";", ",", "|", "\n", "mixed") | |
| def detect_list_delimiter(values: list[str]) -> str | None: | |
| """Looks for a separator that makes this column a list of short items ("Acme, Beta Ltd"). | |
| Returns the delimiter or None. Deliberately conservative: it only ever feeds a *suggestion* | |
| the user confirms, because commas also appear in names ("Smith, John") and sentences.""" | |
| nonempty = [v for v in values if v] | |
| if len(nonempty) < 2: | |
| return None | |
| def qualifies(delim: str) -> bool: | |
| parts = [split_items(v, delim) for v in nonempty if (delim in v if delim != "mixed" else re.search(r"[;,|\n]", v))] | |
| if len(parts) < 2 or len(parts) / len(nonempty) < 0.3: | |
| return False | |
| items = [i for p in parts for i in p] | |
| if not items or any(len(i) > 60 for i in items): | |
| return False | |
| # Sentences have long "items"; lists of names/events/tags have short ones. | |
| return sum(len(i.split()) for i in items) / len(items) <= 4 | |
| ratios = {d: sum(1 for v in nonempty if d in v) / len(nonempty) for d in (";", ",", "|", "\n")} | |
| if ratios[";"] >= 0.2 and ratios[","] >= 0.2 and qualifies("mixed"): | |
| return "mixed" | |
| for delim in sorted(ratios, key=lambda d: -ratios[d]): | |
| if ratios[delim] >= 0.3 and qualifies(delim): | |
| return delim | |
| return None | |
| def _has_conforming_value(values: list[str], new_type: str) -> bool: | |
| validator = TYPE_VALIDATORS.get(new_type) | |
| if validator is None: | |
| return True # name/tag/text have no syntax to conform to | |
| nonempty = [v for v in values if v] | |
| return not nonempty or any(validator(v) for v in nonempty) | |
| def _flagged_values(values: list[str], col_type: str) -> list[int]: | |
| validator = TYPE_VALIDATORS.get(col_type) | |
| if validator is None: | |
| return [] | |
| return [i for i, v in enumerate(values) if v and not validator(v)] | |
| def _assign_row_ids(rows: list[dict]) -> list[tuple[str, dict]]: | |
| return [(uuid.uuid4().hex, row) for row in rows] | |
| def _columns_with_flags(column_types: dict[str, str], headers: list[str], id_rows: list[tuple[str, dict]]) -> list[dict]: | |
| """Flags non-conforming values up front, at classification time — not only when a column's | |
| type is changed later. An auto-classified "email" column with one bad row should show that | |
| immediately, the same as a manual type change would.""" | |
| columns = [] | |
| for h in headers: | |
| values = [row.get(h, "") for _, row in id_rows] | |
| flagged_indices = _flagged_values(values, column_types[h]) | |
| col = { | |
| "name": h, | |
| "type": column_types[h], | |
| "flagged_row_ids": [id_rows[i][0] for i in flagged_indices], | |
| } | |
| if col["type"] == "tag": | |
| col["delimiter"] = detect_list_delimiter(values) or ";" | |
| elif col["type"] in ("text", "name"): | |
| suggested = detect_list_delimiter(values) | |
| if suggested: | |
| col["list_suggestion"] = suggested | |
| columns.append(col) | |
| return columns | |
| # --- Upload / create ------------------------------------------------------------ | |
| def create_database_from_upload( | |
| org_id: str, actor_uid: str, name: str, file_bytes: bytes, filename: str, | |
| save_copy=None, | |
| ) -> str: | |
| """save_copy(database_id, ext, file_bytes) -> None is injected rather than hardcoded so | |
| this stays unit-testable without touching disk; the real implementation lives in app.py's | |
| route (writes to uploads/{org_id}/{database_id}.{ext}).""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" | |
| parser = _PARSERS.get(ext) | |
| if parser is None: | |
| raise ValidationError(f"Unsupported file type {filename!r} — use .csv, .xlsx, or .json") | |
| headers, rows = parser(file_bytes) | |
| headers, rows = _clean(headers, rows) | |
| if not headers or not rows: | |
| raise ValidationError("No usable data found in that file after cleaning") | |
| column_types = {h: classify_column(h, [r[h] for r in rows]) for h in headers} | |
| identity_columns = {h for h in headers if column_types[h] in ("email", "contact")} | |
| result = normalize.normalize(rows, headers, exclude_determinants=identity_columns, auto_split=False) | |
| database_id = uuid.uuid4().hex | |
| if save_copy: | |
| save_copy(database_id, ext, file_bytes) | |
| id_rows = _assign_row_ids(result.root.rows) | |
| columns = _columns_with_flags(column_types, result.root.columns, id_rows) | |
| db.set_doc(_database_path(org_id, database_id), { | |
| "name": name, | |
| "source_filename": filename, | |
| "source_kind": "xlsx" if ext in ("xlsx", "xls") else ext, | |
| "created_at": _now(), | |
| "row_count": len(result.root.rows), | |
| "columns": columns, | |
| "lookup_tables": [{"name": t.name, "columns": t.columns} for t in result.lookups], | |
| "suggestions": result.suggestions, | |
| }) | |
| for lookup in result.lookups: | |
| db.set_doc(_lookup_path(org_id, database_id, lookup.name), { | |
| "columns": lookup.columns, | |
| "encrypted_rows": encrypt(json.dumps(lookup.rows)), | |
| }) | |
| for row_id, row in id_rows: | |
| db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(row))}) | |
| return database_id | |
| def get_database(org_id: str, database_id: str) -> dict: | |
| doc = db.get_doc(_database_path(org_id, database_id)) | |
| if doc is None: | |
| raise orgs.NotFoundError(f"No database {database_id!r} in org {org_id!r}") | |
| return doc | |
| def list_databases(org_id: str) -> dict[str, dict]: | |
| prefix = f"orgs/{org_id}/databases/" | |
| docs = db.query_prefix(prefix) | |
| return {path[len(prefix):]: data for path, data in docs.items()} | |
| def list_rows(org_id: str, database_id: str, search: str | None = None, sort: str | None = None) -> list[dict]: | |
| """Every row, decrypted, with its row_id — client-side-sized (prototype scale), so search | |
| and sort both happen here in Python rather than needing a real query engine.""" | |
| prefix = _rows_prefix(org_id, database_id) | |
| docs = db.query_prefix(prefix) | |
| rows = [] | |
| for path, data in docs.items(): | |
| row = json.loads(decrypt(data["encrypted_data"])) | |
| row["row_id"] = path[len(prefix):] | |
| rows.append(row) | |
| if search: | |
| needle = search.strip().lower() | |
| rows = [r for r in rows if any(needle in str(v).lower() for v in r.values())] | |
| if sort: | |
| rows.sort(key=lambda r: (r.get(sort) is None, str(r.get(sort, "")).lower())) | |
| return rows | |
| # Operators a segment condition can use. Which ones make sense depends on the column's type — | |
| # SEGMENT_OPERATORS_BY_TYPE below is what the frontend uses to only offer valid ones, but | |
| # _row_matches tolerates any operator against any column rather than erroring, since a changed | |
| # column type shouldn't break a condition written before the change. | |
| SEGMENT_OPERATORS_BY_TYPE: dict[str, tuple[str, ...]] = { | |
| "bool": ("is_true", "is_false", "is_empty", "is_not_empty"), | |
| "tag": ("has_tag", "lacks_tag", "is_empty", "is_not_empty"), | |
| } | |
| _DEFAULT_OPERATORS = ("equals", "not_equals", "contains", "not_contains", "is_empty", "is_not_empty") | |
| def _row_matches(row: dict, col_type: str, field: str, op: str, value: str, delimiter: str | None = None) -> bool: | |
| text = str(row.get(field, "") or "") | |
| if op == "is_empty": | |
| return text.strip() == "" | |
| if op == "is_not_empty": | |
| return text.strip() != "" | |
| if col_type == "bool": | |
| truthy = text.strip().lower() in {"true", "yes", "y", "1"} | |
| if op == "is_true": | |
| return truthy | |
| if op == "is_false": | |
| return not truthy | |
| if col_type == "tag": | |
| tags = {t.lower() for t in split_items(text, delimiter)} | |
| needle = value.strip().lower() | |
| if op == "has_tag": | |
| return needle in tags | |
| if op == "lacks_tag": | |
| return needle not in tags | |
| needle = value.strip().lower() | |
| hay = text.strip().lower() | |
| if op == "equals": | |
| return hay == needle | |
| if op == "not_equals": | |
| return hay != needle | |
| if op == "contains": | |
| return needle in hay | |
| if op == "not_contains": | |
| return needle not in hay | |
| return True | |
| def apply_segment(rows: list[dict], columns: list[dict], conditions: list[dict]) -> list[dict]: | |
| """Narrows `rows` to only those matching every condition (AND). Each condition is | |
| {field, op, value}; unknown fields match nothing (a stale condition from a since-removed | |
| column shouldn't silently widen a send back out to everyone).""" | |
| col_types = {c["name"]: c["type"] for c in columns} | |
| delimiters = {c["name"]: c.get("delimiter") for c in columns} | |
| matched = rows | |
| for cond in conditions: | |
| field = cond.get("field", "") | |
| if field not in col_types: | |
| return [] | |
| op = cond.get("op", "") | |
| value = str(cond.get("value", "")) | |
| col_type = col_types[field] | |
| matched = [r for r in matched if _row_matches(r, col_type, field, op, value, delimiters[field])] | |
| return matched | |
| def preview_segment(org_id: str, database_id: str, conditions: list[dict]) -> dict: | |
| """Live "how many recipients does this match" count for the campaign segment builder, plus | |
| a small sample so the UI can show who, not just how many.""" | |
| database = get_database(org_id, database_id) | |
| rows = list_rows(org_id, database_id) | |
| matched = apply_segment(rows, database["columns"], conditions) | |
| return {"matched": len(matched), "sample": matched[:5]} | |
| def add_row(org_id: str, actor_uid: str, database_id: str, values: dict) -> str: | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| return insert_row(org_id, database_id, values) | |
| def insert_row(org_id: str, database_id: str, values: dict) -> str: | |
| """add_row without the role check — for callers that have already authorized the insert | |
| some other way (backend/collect.py: a confirmed double opt-in, not a signed-in user).""" | |
| get_database(org_id, database_id) # raises NotFoundError if missing | |
| row_id = uuid.uuid4().hex | |
| db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(values))}) | |
| database = get_database(org_id, database_id) | |
| db.update_doc(_database_path(org_id, database_id), {"row_count": database["row_count"] + 1}) | |
| known = {c["name"] for c in database["columns"]} | |
| new_cols = [k for k in values if k not in known] | |
| if new_cols: | |
| updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| return row_id | |
| def update_row(org_id: str, actor_uid: str, database_id: str, row_id: str, tags: dict) -> None: | |
| """Merges `tags` into the row's existing values. Any key not already a known column is | |
| appended to the database's column list (type "text") — a manually-added tag becomes a | |
| real, placeholder-usable column for the whole database, matching how the CSV-derived | |
| columns work, not a side channel.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| path = _row_path(org_id, database_id, row_id) | |
| doc = db.get_doc(path) | |
| if doc is None: | |
| raise orgs.NotFoundError(f"No row {row_id!r} in database {database_id!r}") | |
| row = json.loads(decrypt(doc["encrypted_data"])) | |
| row.update(tags) | |
| db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))}) | |
| database = get_database(org_id, database_id) | |
| known = {c["name"] for c in database["columns"]} | |
| new_cols = [k for k in tags if k not in known] | |
| if new_cols: | |
| updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| def bulk_update_rows(org_id: str, actor_uid: str, database_id: str, row_ids: list[str], tags: dict) -> list[str]: | |
| """Same merge semantics as update_row, applied to many rows in one call — the mass "fill a | |
| column" / "add a tag to the selection" grid actions need this, since looping update_row | |
| client-side would mean one HTTP round trip per selected row (unusable at a few thousand | |
| rows). Fetches the database doc once up front and updates it once at the end, rather than | |
| once per row like update_row does. Silently skips a row_id that no longer exists (e.g. a | |
| stale selection from before a delete) instead of failing the whole batch over it.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| database = get_database(org_id, database_id) | |
| updated: list[str] = [] | |
| for row_id in row_ids: | |
| path = _row_path(org_id, database_id, row_id) | |
| doc = db.get_doc(path) | |
| if doc is None: | |
| continue | |
| row = json.loads(decrypt(doc["encrypted_data"])) | |
| row.update(tags) | |
| db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))}) | |
| updated.append(row_id) | |
| known = {c["name"] for c in database["columns"]} | |
| new_cols = [k for k in tags if k not in known] | |
| if new_cols: | |
| updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| return updated | |
| def bulk_append_tag( | |
| org_id: str, actor_uid: str, database_id: str, row_ids: list[str], tag: str, tag_column: str = "Tags", | |
| ) -> list[str]: | |
| """Appends `tag` to the existing semicolon-delimited tag_column value for each row, rather | |
| than overwriting it like bulk_update_rows does — each row keeps whatever tags it already | |
| had, so this can't be expressed as one uniform PATCH value the way a column "fill" can. | |
| Still a single HTTP call for the whole selection, looping server-side rather than once per | |
| row over the network.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| database = get_database(org_id, database_id) | |
| updated: list[str] = [] | |
| for row_id in row_ids: | |
| path = _row_path(org_id, database_id, row_id) | |
| doc = db.get_doc(path) | |
| if doc is None: | |
| continue | |
| row = json.loads(decrypt(doc["encrypted_data"])) | |
| existing = [t for t in str(row.get(tag_column, "")).split(";") if t] | |
| if tag not in existing: | |
| existing.append(tag) | |
| row[tag_column] = ";".join(existing) | |
| db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))}) | |
| updated.append(row_id) | |
| known = {c["name"] for c in database["columns"]} | |
| if tag_column not in known: | |
| updated_columns = database["columns"] + [{"name": tag_column, "type": "tag", "flagged_row_ids": []}] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| return updated | |
| def delete_rows(org_id: str, actor_uid: str, database_id: str, row_ids: list[str]) -> int: | |
| """Bulk delete for the grid's mass-selection "Delete" action — same one-call-not-N | |
| reasoning as bulk_update_rows. Returns the number actually deleted (stale ids are skipped, | |
| not errors) so the caller can show an accurate count.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| database = get_database(org_id, database_id) | |
| deleted = 0 | |
| for row_id in row_ids: | |
| path = _row_path(org_id, database_id, row_id) | |
| if db.get_doc(path) is None: | |
| continue | |
| db.delete_doc(path) | |
| deleted += 1 | |
| if deleted: | |
| db.update_doc(_database_path(org_id, database_id), {"row_count": max(0, database["row_count"] - deleted)}) | |
| return deleted | |
| def bulk_set_values(org_id: str, actor_uid: str, database_id: str, updates: dict[str, dict]) -> list[str]: | |
| """Like bulk_update_rows, but each row gets its OWN value map rather than one uniform | |
| value for everyone. This is what undoing a "fill column for N rows" needs — each row's | |
| prior value was (generally) different before the fill overwrote it, so restoring them | |
| can't be expressed as a single uniform PATCH the way the fill itself was.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| database = get_database(org_id, database_id) | |
| updated: list[str] = [] | |
| all_new_keys: set[str] = set() | |
| for row_id, values in updates.items(): | |
| path = _row_path(org_id, database_id, row_id) | |
| doc = db.get_doc(path) | |
| if doc is None: | |
| continue | |
| row = json.loads(decrypt(doc["encrypted_data"])) | |
| row.update(values) | |
| db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))}) | |
| updated.append(row_id) | |
| all_new_keys.update(values.keys()) | |
| known = {c["name"] for c in database["columns"]} | |
| new_cols = [k for k in all_new_keys if k not in known] | |
| if new_cols: | |
| updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| return updated | |
| def remove_column(org_id: str, actor_uid: str, database_id: str, column: str) -> dict: | |
| """Drops a column from the schema and strips it from every row. Returns a snapshot | |
| ({column def, values_by_row_id}) so the caller (the grid's undo stack) can restore it | |
| exactly via restore_column without a second round trip to reconstruct what was deleted.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| database = get_database(org_id, database_id) | |
| col_def = next((c for c in database["columns"] if c["name"] == column), None) | |
| if col_def is None: | |
| raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}") | |
| values_by_row_id: dict[str, str] = {} | |
| for row in list_rows(org_id, database_id): | |
| row_id = row["row_id"] | |
| if column not in row: | |
| continue | |
| values_by_row_id[row_id] = row[column] | |
| remaining = {k: v for k, v in row.items() if k not in (column, "row_id")} | |
| db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(remaining))}) | |
| updated_columns = [c for c in database["columns"] if c["name"] != column] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| return {"column": col_def, "values_by_row_id": values_by_row_id} | |
| def restore_column(org_id: str, actor_uid: str, database_id: str, column_def: dict, values_by_row_id: dict[str, str]) -> None: | |
| """The inverse of remove_column — undo target for the grid's "Remove column" action.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| database = get_database(org_id, database_id) | |
| for row_id, value in values_by_row_id.items(): | |
| path = _row_path(org_id, database_id, row_id) | |
| doc = db.get_doc(path) | |
| if doc is None: | |
| continue | |
| row = json.loads(decrypt(doc["encrypted_data"])) | |
| row[column_def["name"]] = value | |
| db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))}) | |
| if not any(c["name"] == column_def["name"] for c in database["columns"]): | |
| db.update_doc(_database_path(org_id, database_id), {"columns": database["columns"] + [column_def]}) | |
| def rename_column(org_id: str, actor_uid: str, database_id: str, column: str, new_name: str) -> str: | |
| """Renames a column in the schema and re-keys it in every row, keeping column order. No | |
| values change, so renaming back is an exact undo. Saved presets and campaigns that use | |
| {OldName} keep the old text; the UI warns about that.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| new_name = " ".join(new_name.split()) | |
| if not new_name: | |
| raise ValidationError("A column name can't be empty") | |
| if new_name == "row_id": | |
| raise ValidationError("That name is reserved") | |
| database = get_database(org_id, database_id) | |
| if not any(c["name"] == column for c in database["columns"]): | |
| raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}") | |
| if new_name == column: | |
| return new_name | |
| if any(c["name"].lower() == new_name.lower() for c in database["columns"] if c["name"] != column): | |
| raise ValidationError(f"There is already a column called {new_name!r}") | |
| for row in list_rows(org_id, database_id): | |
| if column not in row: | |
| continue | |
| rekeyed = {(new_name if k == column else k): v for k, v in row.items() if k != "row_id"} | |
| db.set_doc(_row_path(org_id, database_id, row["row_id"]), {"encrypted_data": encrypt(json.dumps(rekeyed))}) | |
| updated_columns = [{**c, "name": new_name} if c["name"] == column else c for c in database["columns"]] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| return new_name | |
| def column_rename_suggestions(org_id: str, database_id: str) -> list[dict]: | |
| from backend import rename_suggest | |
| return rename_suggest.suggest_renames(get_database(org_id, database_id)["columns"]) | |
| def column_type_suggestions(org_id: str, database_id: str) -> list[dict]: | |
| """Columns whose stored type no longer matches what their values say (e.g. a date column | |
| that an older classifier called "contact"). Only ever a suggestion; never applied unasked. | |
| Only recommends moving between the plain, low-stakes types, so a column the person | |
| deliberately set to email, list or yes/no is left alone.""" | |
| database = get_database(org_id, database_id) | |
| rows = list_rows(org_id, database_id) | |
| suggestions = [] | |
| for col in database["columns"]: | |
| if col["type"] not in ("text", "contact", "name"): | |
| continue | |
| guess = classify_column(col["name"], [r.get(col["name"], "") for r in rows]) | |
| if guess != col["type"] and guess in ("text", "date", "contact", "name", "email", "link", "bool"): | |
| suggestions.append({"column": col["name"], "current": col["type"], "suggested": guess}) | |
| return suggestions | |
| def rename_database(org_id: str, actor_uid: str, database_id: str, new_name: str) -> None: | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| get_database(org_id, database_id) | |
| if not new_name.strip(): | |
| raise ValidationError("Name can't be empty") | |
| db.update_doc(_database_path(org_id, database_id), {"name": new_name.strip()}) | |
| def set_database_disabled(org_id: str, actor_uid: str, database_id: str, disabled: bool) -> None: | |
| """"Disabled" means excluded from campaign sends only — still fully visible and editable | |
| here. Doesn't cascade into blocking merges or anything else; that would be a separate, | |
| more surprising rule than what the name implies.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| get_database(org_id, database_id) | |
| db.update_doc(_database_path(org_id, database_id), {"disabled": disabled}) | |
| def delete_database(org_id: str, actor_uid: str, database_id: str) -> None: | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| get_database(org_id, database_id) | |
| for path in list(db.query_prefix(_rows_prefix(org_id, database_id)).keys()): | |
| db.delete_doc(path) | |
| lookups_prefix = f"orgs/{org_id}/databases/{database_id}/lookups/" | |
| for path in list(db.query_prefix(lookups_prefix).keys()): | |
| db.delete_doc(path) | |
| db.delete_doc(_database_path(org_id, database_id)) | |
| def duplicate_database(org_id: str, actor_uid: str, database_id: str, new_name: str) -> str: | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| database = get_database(org_id, database_id) | |
| rows = list_rows(org_id, database_id) | |
| new_id = uuid.uuid4().hex | |
| id_map = {r["row_id"]: uuid.uuid4().hex for r in rows} | |
| db.set_doc(_database_path(org_id, new_id), {**database, "name": new_name, "created_at": _now()}) | |
| for r in rows: | |
| payload = {k: v for k, v in r.items() if k != "row_id"} | |
| db.set_doc(_row_path(org_id, new_id, id_map[r["row_id"]]), {"encrypted_data": encrypt(json.dumps(payload))}) | |
| lookups_prefix = f"orgs/{org_id}/databases/{database_id}/lookups/" | |
| for path, data in db.query_prefix(lookups_prefix).items(): | |
| db.set_doc(_lookup_path(org_id, new_id, path[len(lookups_prefix):]), data) | |
| # re-point flagged_row_ids at the new row ids — leaving them pointing at the OLD | |
| # database's ids would silently flag the wrong rows (or none, once those ids never match). | |
| updated_columns = [ | |
| {**c, "flagged_row_ids": [id_map[rid] for rid in c["flagged_row_ids"] if rid in id_map]} | |
| for c in database["columns"] | |
| ] | |
| db.update_doc(_database_path(org_id, new_id), {"columns": updated_columns}) | |
| return new_id | |
| def get_lookup_table(org_id: str, database_id: str, lookup_name: str) -> dict: | |
| """Read-only access to a normalized lookup table's actual rows — the database doc only | |
| ever stores the lookup's NAME and column list (see create_database_from_upload), never its | |
| rows, so the grid's lookup-table viewer needs a dedicated fetch rather than reading it off | |
| the database doc it already has.""" | |
| doc = db.get_doc(_lookup_path(org_id, database_id, lookup_name)) | |
| if doc is None: | |
| raise orgs.NotFoundError(f"No lookup table {lookup_name!r} in database {database_id!r}") | |
| return {"columns": doc["columns"], "rows": json.loads(decrypt(doc["encrypted_rows"]))} | |
| def set_column_list(org_id: str, actor_uid: str, database_id: str, column: str, delimiter: str | None) -> None: | |
| """Makes `column` a list column split on `delimiter` (or, with None, dismisses the | |
| suggestion to do so and leaves the column as it is).""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| if delimiter is not None and delimiter not in LIST_DELIMITERS: | |
| raise ValidationError(f"Unknown list separator {delimiter!r}") | |
| database = get_database(org_id, database_id) | |
| if not any(c["name"] == column for c in database["columns"]): | |
| raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}") | |
| updated = [] | |
| for c in database["columns"]: | |
| if c["name"] == column: | |
| c = {k: v for k, v in c.items() if k != "list_suggestion"} | |
| if delimiter is not None: | |
| c = {**c, "type": "tag", "delimiter": delimiter, "flagged_row_ids": []} | |
| updated.append(c) | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated}) | |
| def update_column_type(org_id: str, database_id: str, column: str, new_type: str) -> list[str]: | |
| if new_type not in COLUMN_TYPES: | |
| raise ValidationError(f"Unknown column type {new_type!r}") | |
| database = get_database(org_id, database_id) | |
| if not any(c["name"] == column for c in database["columns"]): | |
| raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}") | |
| rows = list_rows(org_id, database_id) | |
| values = [r.get(column, "") for r in rows] | |
| if not _has_conforming_value(values, new_type): | |
| raise ValidationError( | |
| f"No existing value in {column!r} looks like a valid {new_type} — refusing the change" | |
| ) | |
| flagged_indices = _flagged_values(values, new_type) | |
| flagged_row_ids = [rows[i]["row_id"] for i in flagged_indices] | |
| updated_columns = [ | |
| {**{k: v for k, v in c.items() if k != "list_suggestion"}, "type": new_type, "flagged_row_ids": flagged_row_ids, | |
| **({"delimiter": c.get("delimiter") or ";"} if new_type == "tag" else {})} | |
| if c["name"] == column else c | |
| for c in database["columns"] | |
| ] | |
| db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns}) | |
| return flagged_row_ids | |
| def merge_databases(org_id: str, actor_uid: str, database_id_a: str, database_id_b: str, new_name: str) -> str: | |
| """Concatenates both databases' rows under a brand-new database, reconciling columns by | |
| name (union) and re-running classification + normalization over the combined set — a | |
| merge isn't just a row concat, it's a fresh upload of the combined data.""" | |
| orgs.require_permission(org_id, actor_uid, "edit_lists") | |
| rows_a = list_rows(org_id, database_id_a) | |
| rows_b = list_rows(org_id, database_id_b) | |
| headers = list(dict.fromkeys([k for r in rows_a + rows_b for k in r if k != "row_id"])) | |
| # Exact-duplicate rows (e.g. the same contact present in both databases) are dropped | |
| # before normalizing — besides being the obviously-wanted merge behavior on its own, an | |
| # un-deduplicated duplicate row makes EVERY column pair look like a valid functional | |
| # dependency (they trivially agree with their own duplicate), which would otherwise send | |
| # the FD-discovery heuristic down a false trail. | |
| seen_rows: set[tuple] = set() | |
| combined: list[dict] = [] | |
| for r in rows_a + rows_b: | |
| row = {h: r.get(h, "") for h in headers} | |
| key = tuple(row[h] for h in headers) | |
| if key in seen_rows: | |
| continue | |
| seen_rows.add(key) | |
| combined.append(row) | |
| column_types = {h: classify_column(h, [r[h] for r in combined]) for h in headers} | |
| identity_columns = {h for h in headers if column_types[h] in ("email", "contact")} | |
| result = normalize.normalize(combined, headers, exclude_determinants=identity_columns, auto_split=False) | |
| database_id = uuid.uuid4().hex | |
| id_rows = _assign_row_ids(result.root.rows) | |
| columns = _columns_with_flags(column_types, result.root.columns, id_rows) | |
| db.set_doc(_database_path(org_id, database_id), { | |
| "name": new_name, | |
| "source_filename": f"merge({database_id_a[:8]}, {database_id_b[:8]})", | |
| "source_kind": "merge", | |
| "created_at": _now(), | |
| "row_count": len(result.root.rows), | |
| "columns": columns, | |
| "lookup_tables": [{"name": t.name, "columns": t.columns} for t in result.lookups], | |
| "suggestions": result.suggestions, | |
| }) | |
| for lookup in result.lookups: | |
| db.set_doc(_lookup_path(org_id, database_id, lookup.name), { | |
| "columns": lookup.columns, | |
| "encrypted_rows": encrypt(json.dumps(lookup.rows)), | |
| }) | |
| for row_id, row in id_rows: | |
| db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(row))}) | |
| return database_id | |