Spaces:
Sleeping
Sleeping
File size: 37,886 Bytes
45dc2a1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 | """
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
|