From e0127e26c513fcf5f908ac2c8d75e7d36d337a47 Mon Sep 17 00:00:00 2001 From: Ashwin Date: Mon, 25 May 2026 17:16:37 +0530 Subject: [PATCH] feat: implement CSV contact import with two-step preview and validation flow --- backend/contacts/import_views.py | 133 ++++ backend/contacts/services/__init__.py | 0 backend/contacts/services/csv_import.py | 729 ++++++++++++++++++ backend/contacts/tests/test_import.py | 435 +++++++++++ backend/contacts/urls.py | 13 +- .../contacts/ContactImportDrawer.svelte | 478 ++++++++++++ .../src/routes/(app)/contacts/+page.server.js | 59 ++ .../src/routes/(app)/contacts/+page.svelte | 11 + 8 files changed, 1857 insertions(+), 1 deletion(-) create mode 100644 backend/contacts/import_views.py create mode 100644 backend/contacts/services/__init__.py create mode 100644 backend/contacts/services/csv_import.py create mode 100644 backend/contacts/tests/test_import.py create mode 100644 frontend/src/lib/components/contacts/ContactImportDrawer.svelte diff --git a/backend/contacts/import_views.py b/backend/contacts/import_views.py new file mode 100644 index 00000000..90088772 --- /dev/null +++ b/backend/contacts/import_views.py @@ -0,0 +1,133 @@ +"""CSV import endpoints for contacts. + +Preview reads the uploaded file and returns row-level validation results +without touching the DB. Commit re-runs validation and writes inside a single +transaction. Both endpoints are gated to ADMIN or sales-access users so +non-privileged members cannot mass-create contacts through this surface. +""" + +from drf_spectacular.utils import extend_schema, inline_serializer +from rest_framework import serializers, status +from rest_framework.parsers import MultiPartParser +from rest_framework.permissions import IsAuthenticated +from rest_framework.response import Response +from rest_framework.views import APIView + +from common.permissions import HasOrgContext +from contacts.services.csv_import import commit_rows, parse_and_validate + + +MAX_UPLOAD_BYTES = 5 * 1024 * 1024 # 5 MB; matches the UI hint + + +def _can_import(profile) -> bool: + """Mass-create requires admin or explicit sales-access permission.""" + if profile is None: + return False + if getattr(profile, "role", None) == "ADMIN": + return True + if getattr(profile, "is_admin", False): + return True + return bool(getattr(profile, "has_sales_access", False)) + + +def _read_upload(request): + """Return (file_bytes, error_response). One of the two will be None. + + The size cap is enforced against the actual bytes read, not against + `upload.size`. `upload.size` is derived from a client-supplied + Content-Length header for in-memory uploads and can be zero or absent + even when the body is large, so trusting it for the limit check creates + a bypass. + """ + upload = request.FILES.get("file") + if not upload: + return None, Response( + {"error": True, "message": "No file uploaded (expected field 'file')"}, + status=status.HTTP_400_BAD_REQUEST, + ) + name = (upload.name or "").lower() + if not name.endswith(".csv"): + return None, Response( + {"error": True, "message": "File must have a .csv extension"}, + status=status.HTTP_400_BAD_REQUEST, + ) + file_bytes = upload.read() + if len(file_bytes) > MAX_UPLOAD_BYTES: + return None, Response( + {"error": True, "message": "File exceeds the 5 MB upload limit"}, + status=status.HTTP_400_BAD_REQUEST, + ) + return file_bytes, None + + +class ContactImportPreviewView(APIView): + permission_classes = (IsAuthenticated, HasOrgContext) + parser_classes = (MultiPartParser,) + + @extend_schema( + tags=["contacts"], + request=inline_serializer( + name="ContactImportPreviewRequest", + fields={"file": serializers.FileField()}, + ), + responses={ + 200: inline_serializer( + name="ContactImportPreviewResponse", + fields={ + "header_error": serializers.CharField(allow_null=True), + "valid": serializers.ListField(child=serializers.DictField()), + "errors": serializers.ListField(child=serializers.DictField()), + "summary": serializers.DictField(), + }, + ) + }, + ) + def post(self, request, *args, **kwargs): + if not _can_import(request.profile): + return Response( + {"error": True, "message": "Permission denied"}, + status=status.HTTP_403_FORBIDDEN, + ) + file_bytes, err = _read_upload(request) + if err is not None: + return err + result = parse_and_validate(file_bytes, request.profile.org) + return Response(result.to_dict(), status=status.HTTP_200_OK) + + +class ContactImportCommitView(APIView): + permission_classes = (IsAuthenticated, HasOrgContext) + parser_classes = (MultiPartParser,) + + @extend_schema( + tags=["contacts"], + request=inline_serializer( + name="ContactImportCommitRequest", + fields={"file": serializers.FileField()}, + ), + responses={ + 200: inline_serializer( + name="ContactImportCommitResponse", + fields={ + "error": serializers.BooleanField(), + "created": serializers.IntegerField(), + "ids": serializers.ListField(child=serializers.CharField()), + }, + ) + }, + ) + def post(self, request, *args, **kwargs): + if not _can_import(request.profile): + return Response( + {"error": True, "message": "Permission denied"}, + status=status.HTTP_403_FORBIDDEN, + ) + file_bytes, err = _read_upload(request) + if err is not None: + return err + result = commit_rows(file_bytes, request.profile.org, request.profile) + http_status = ( + status.HTTP_400_BAD_REQUEST if result.get("error") else status.HTTP_200_OK + ) + return Response(result, status=http_status) diff --git a/backend/contacts/services/__init__.py b/backend/contacts/services/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/contacts/services/csv_import.py b/backend/contacts/services/csv_import.py new file mode 100644 index 00000000..15218720 --- /dev/null +++ b/backend/contacts/services/csv_import.py @@ -0,0 +1,729 @@ +"""CSV import for contacts. + +Two-phase: `parse_and_validate` reads + validates a CSV without writing to the +DB beyond read-only lookups; `commit_rows` writes inside a transaction. Both +phases re-run validation so the commit endpoint is safe even if called +directly. + +Duplicate policy (per org): + * email — hard error (DB enforces a per-org case-insensitive unique + constraint; we mirror it in the importer so users see a clean + row-level message instead of an IntegrityError at commit). + * phone — hard error. There is no DB constraint, but most callers + consider phone a unique identifier in practice. We normalize + using `common.validators.normalize_phone` (digits-only, last + 10) so "+1 (202) 555-1234" matches "2025551234". + * full name — hard error ONLY when the row has neither email nor phone to + disambiguate from an existing same-named contact. Two people + can legitimately share a name when other identifiers differ. + +All reference lookups (account, assignees, teams) are bulk-prefetched once per +call (one SELECT per reference type), scoped to the caller's org so a +malicious CSV cannot reach across tenants. +""" + +from __future__ import annotations + +import csv +import io +import re +from dataclasses import dataclass, field +from typing import Any, Iterable + +from django.core.exceptions import ValidationError as DjangoValidationError +from django.core.validators import URLValidator +from django.db import IntegrityError, transaction +from django.db.models.functions import Lower +from django.utils.text import slugify + +from accounts.models import Account +from common.models import Profile, Tags, Teams +from common.utils import COUNTRIES +from common.validators import normalize_phone +from contacts.models import Contact + + +REQUIRED_HEADERS = ("first_name", "last_name") +OPTIONAL_HEADERS = ( + "email", + "phone", + "organization", + "title", + "department", + "do_not_call", + "linkedin_url", + "address_line", + "city", + "state", + "postcode", + "country", + "description", + "account_name", + "assigned_emails", + "team_names", + "tags", +) +KNOWN_HEADERS = REQUIRED_HEADERS + OPTIONAL_HEADERS + +MAX_ROWS = 5000 +NAME_MAX_LEN = 255 +EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") +PHONE_RE = re.compile(r"^[\d\s\-\(\)\+\.]{7,25}$") +_url_validator = URLValidator(schemes=("http", "https")) +TRUTHY = {"1", "true", "yes", "y", "t"} +FALSY = {"", "0", "false", "no", "n", "f"} +COUNTRY_CODES = {code.upper() for code, _label in COUNTRIES} + + +@dataclass +class RowError: + row: int # 1-based, matching CSV line numbers (header is row 0) + field: str + message: str + + def to_dict(self) -> dict[str, Any]: + return {"row": self.row, "field": self.field, "message": self.message} + + +@dataclass +class ValidatedRow: + """A row that passed validation; reference fields resolved to UUID strings.""" + + row: int + first_name: str + last_name: str + email: str | None = None + phone: str | None = None + organization: str | None = None + title: str | None = None + department: str | None = None + do_not_call: bool = False + linkedin_url: str | None = None + address_line: str | None = None + city: str | None = None + state: str | None = None + postcode: str | None = None + country: str | None = None + description: str | None = None + account_id: str | None = None + assigned_ids: list[str] = field(default_factory=list) + team_ids: list[str] = field(default_factory=list) + tag_names: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "row": self.row, + "first_name": self.first_name, + "last_name": self.last_name, + "email": self.email, + "phone": self.phone, + "organization": self.organization, + "title": self.title, + "department": self.department, + "do_not_call": self.do_not_call, + "linkedin_url": self.linkedin_url, + "address_line": self.address_line, + "city": self.city, + "state": self.state, + "postcode": self.postcode, + "country": self.country, + "description": self.description, + "account_id": self.account_id, + "assigned_ids": list(self.assigned_ids), + "team_ids": list(self.team_ids), + "tag_names": list(self.tag_names), + } + + +@dataclass +class ImportResult: + valid: list[ValidatedRow] + errors: list[RowError] + header_error: str | None = None + + @property + def summary(self) -> dict[str, int]: + return { + "total": len(self.valid) + len({e.row for e in self.errors}), + "valid": len(self.valid), + "invalid": len({e.row for e in self.errors}), + } + + def to_dict(self) -> dict[str, Any]: + return { + "header_error": self.header_error, + "valid": [r.to_dict() for r in self.valid], + "errors": [e.to_dict() for e in self.errors], + "summary": self.summary, + } + + +@dataclass +class _RefMaps: + """Reference values needed to validate + resolve rows in one pass. + + Keys are lowercased; values are canonical UUID strings for the match. + The three "existing_*" sets exist only for duplicate detection. + """ + + accounts: dict[str, str] + profiles: dict[str, str] + teams: dict[str, str] + existing_emails: set[str] + existing_phones: set[str] + existing_full_names: set[str] + + +def _decode(file_bytes: bytes) -> str | None: + """Decode CSV bytes as UTF-8 (with or without BOM). + + Return None for non-UTF-8 so the caller can surface a "save as UTF-8" + header_error rather than producing mojibake that passes validation. + """ + for encoding in ("utf-8-sig", "utf-8"): + try: + return file_bytes.decode(encoding) + except UnicodeDecodeError: + continue + return None + + +def _normalize_headers(raw_headers: Iterable[str]) -> list[str]: + return [(h or "").strip().lower() for h in raw_headers] + + +def _split_multi(raw: str) -> list[str]: + if not raw: + return [] + return [part.strip() for part in raw.split(";") if part.strip()] + + +def _parse_bool(raw: str) -> bool | None: + if raw is None: + return False + val = raw.strip().lower() + if val in TRUTHY: + return True + if val in FALSY: + return False + return None # caller treats None as "invalid input" + + +def parse_and_validate(file_bytes: bytes, org) -> ImportResult: + """Parse a CSV byte string and validate every row against the given org. + + All reference and duplicate lookups are scoped to `org` so a malicious CSV + cannot reach across tenants. + """ + text = _decode(file_bytes) + if text is None: + return ImportResult( + valid=[], + errors=[], + header_error="File could not be decoded as UTF-8. Save your CSV as UTF-8 and try again.", + ) + reader = csv.reader(io.StringIO(text)) + rows = list(reader) + if not rows: + return ImportResult(valid=[], errors=[], header_error="CSV is empty") + + headers = _normalize_headers(rows[0]) + missing = [h for h in REQUIRED_HEADERS if h not in headers] + if missing: + return ImportResult( + valid=[], + errors=[], + header_error=f"Missing required header(s): {', '.join(missing)}", + ) + unknown = [h for h in headers if h and h not in KNOWN_HEADERS] + if unknown: + return ImportResult( + valid=[], + errors=[], + header_error=f"Unknown header(s): {', '.join(unknown)}", + ) + + data_rows = rows[1:] + if len(data_rows) > MAX_ROWS: + return ImportResult( + valid=[], + errors=[], + header_error=f"Too many rows ({len(data_rows)}); limit is {MAX_ROWS}", + ) + + # First pass: parse each row into a dict and skip blank rows. + parsed: list[tuple[int, dict[str, str]]] = [] + for idx, raw_row in enumerate(data_rows, start=1): + if not any((cell or "").strip() for cell in raw_row): + continue + record = {h: (raw_row[i].strip() if i < len(raw_row) else "") for i, h in enumerate(headers)} + parsed.append((idx, record)) + + ref_maps = _build_ref_maps(parsed, org) + + valid: list[ValidatedRow] = [] + errors: list[RowError] = [] + seen_emails: dict[str, int] = {} # email.lower() -> row number + seen_phones: dict[str, int] = {} # normalized phone -> row number + # Tracks rows that had NO email and NO phone — the same disambiguation + # rule used vs the DB. A row with an email/phone is exempt because the + # email/phone uniqueness checks already prevent silent duplicates. + seen_full_names: dict[str, int] = {} # "first|last".lower() -> row number + + for idx, record in parsed: + row_errors, validated = _validate_and_build( + idx, record, ref_maps, seen_emails, seen_phones, seen_full_names + ) + if row_errors: + errors.extend(row_errors) + continue + valid.append(validated) + if validated.email: + seen_emails[validated.email.lower()] = idx + if validated.phone: + normalized = normalize_phone(validated.phone) + if normalized: + seen_phones[normalized] = idx + if not validated.email and not validated.phone: + key = f"{validated.first_name.lower()}|{validated.last_name.lower()}" + seen_full_names[key] = idx + + return ImportResult(valid=valid, errors=errors) + + +def _build_ref_maps(parsed: list[tuple[int, dict[str, str]]], org) -> _RefMaps: + """Bulk-prefetch every reference value referenced anywhere in the file. + + One query per reference type, scoped to `org`. Keys are lowercased so + row-level lookups are O(1) and case-insensitive. The three "existing_*" + sets cover the duplicate-detection policy: email (DB-unique), phone + (normalized digits), and full_name (only used when row has no email or + phone for disambiguation). + """ + account_names: set[str] = set() + assigned_emails: set[str] = set() + team_names: set[str] = set() + candidate_emails: set[str] = set() + candidate_phones: set[str] = set() + candidate_full_names: set[str] = set() + + for _idx, record in parsed: + account_name = record.get("account_name", "") + if account_name: + account_names.add(account_name.lower()) + for email in _split_multi(record.get("assigned_emails", "")): + assigned_emails.add(email.lower()) + for team in _split_multi(record.get("team_names", "")): + team_names.add(team.lower()) + email = record.get("email", "") + if email: + candidate_emails.add(email.lower()) + phone = record.get("phone", "") + if phone: + normalized = normalize_phone(phone) + if normalized: + candidate_phones.add(normalized) + first = record.get("first_name", "") + last = record.get("last_name", "") + if first and last and not email and not phone: + candidate_full_names.add(f"{first.lower()}|{last.lower()}") + + accounts: dict[str, str] = {} + if account_names: + for pk, name_lower in ( + Account.objects.filter(org=org) + .annotate(name_lower=Lower("name")) + .filter(name_lower__in=account_names) + .values_list("id", "name_lower") + ): + accounts.setdefault(name_lower, str(pk)) + + profiles: dict[str, str] = {} + if assigned_emails: + for pk, email_lower in ( + Profile.objects.filter(org=org, is_active=True) + .annotate(email_lower=Lower("user__email")) + .filter(email_lower__in=assigned_emails) + .values_list("id", "email_lower") + ): + profiles.setdefault(email_lower, str(pk)) + + teams: dict[str, str] = {} + if team_names: + for pk, name_lower in ( + Teams.objects.filter(org=org) + .annotate(name_lower=Lower("name")) + .filter(name_lower__in=team_names) + .values_list("id", "name_lower") + ): + teams.setdefault(name_lower, str(pk)) + + existing_emails: set[str] = set() + if candidate_emails: + existing_emails = set( + Contact.objects.filter(org=org) + .annotate(email_lower=Lower("email")) + .filter(email_lower__in=candidate_emails) + .values_list("email_lower", flat=True) + ) + + existing_phones: set[str] = set() + if candidate_phones: + # Phone has no DB constraint and no normalized column; we have to scan + # all contacts that have a phone in this org and normalize in Python. + # Bounded by org size, not file size — fine for any reasonable tenant. + for raw_phone in Contact.objects.filter(org=org).exclude( + phone__isnull=True + ).exclude(phone="").values_list("phone", flat=True): + normalized = normalize_phone(raw_phone) + if normalized and normalized in candidate_phones: + existing_phones.add(normalized) + + existing_full_names: set[str] = set() + if candidate_full_names: + # Only run this query if any row has a full name but no email and no + # phone — that's the only case where full-name dedup applies. + for first, last in ( + Contact.objects.filter(org=org) + .annotate( + first_lower=Lower("first_name"), + last_lower=Lower("last_name"), + ) + .values_list("first_lower", "last_lower") + ): + key = f"{first}|{last}" + if key in candidate_full_names: + existing_full_names.add(key) + + return _RefMaps( + accounts=accounts, + profiles=profiles, + teams=teams, + existing_emails=existing_emails, + existing_phones=existing_phones, + existing_full_names=existing_full_names, + ) + + +def _validate_and_build( + idx: int, + record: dict, + refs: _RefMaps, + seen_emails: dict[str, int], + seen_phones: dict[str, int], + seen_full_names: dict[str, int], +) -> tuple[list[RowError], ValidatedRow | None]: + """Validate a single row and, on success, return the resolved ValidatedRow. + + All reference resolution uses the prefetched maps — no DB calls here. + """ + errors: list[RowError] = [] + + first_name = record.get("first_name", "") + last_name = record.get("last_name", "") + if not first_name: + errors.append(RowError(idx, "first_name", "First name is required")) + elif len(first_name) > NAME_MAX_LEN: + errors.append( + RowError(idx, "first_name", f"First name exceeds {NAME_MAX_LEN} characters") + ) + if not last_name: + errors.append(RowError(idx, "last_name", "Last name is required")) + elif len(last_name) > NAME_MAX_LEN: + errors.append( + RowError(idx, "last_name", f"Last name exceeds {NAME_MAX_LEN} characters") + ) + + email = record.get("email", "") + if email: + if not EMAIL_RE.match(email): + errors.append(RowError(idx, "email", f"'{email}' is not a valid email")) + else: + email_lower = email.lower() + prior = seen_emails.get(email_lower) + if prior: + errors.append( + RowError( + idx, + "email", + f"Duplicate email also used by row {prior} in this file", + ) + ) + elif email_lower in refs.existing_emails: + errors.append( + RowError( + idx, + "email", + "A contact with this email already exists in your organization", + ) + ) + + phone = record.get("phone", "") + normalized_phone: str | None = None + if phone: + if not PHONE_RE.match(phone): + errors.append( + RowError( + idx, + "phone", + "Phone must be 7-25 characters of digits and separators (+ - ( ) . space)", + ) + ) + else: + normalized_phone = normalize_phone(phone) + if normalized_phone: + prior = seen_phones.get(normalized_phone) + if prior: + errors.append( + RowError( + idx, + "phone", + f"Duplicate phone number also used by row {prior} in this file", + ) + ) + elif normalized_phone in refs.existing_phones: + errors.append( + RowError( + idx, + "phone", + "A contact with this phone number already exists in your organization", + ) + ) + + # Full-name dedup only kicks in when there's nothing else to tell two + # same-named people apart. If the row has an email or phone, we trust + # those as the disambiguator (and have already checked them above). + if first_name and last_name and not email and not phone: + key = f"{first_name.lower()}|{last_name.lower()}" + prior = seen_full_names.get(key) + if prior: + errors.append( + RowError( + idx, + "first_name", + f"Same name as row {prior} in this file; add an email or phone to either row to confirm they're different people", + ) + ) + elif key in refs.existing_full_names: + errors.append( + RowError( + idx, + "first_name", + "A contact with this name already exists; add an email or phone to confirm it's a different person", + ) + ) + + linkedin_url = record.get("linkedin_url", "") + if linkedin_url: + try: + _url_validator(linkedin_url) + except DjangoValidationError: + errors.append( + RowError( + idx, + "linkedin_url", + "Must be a valid URL starting with http:// or https://", + ) + ) + + country_raw = record.get("country", "") + country: str | None = None + if country_raw: + country = country_raw.strip().upper() + if country not in COUNTRY_CODES: + errors.append( + RowError(idx, "country", f"'{country_raw}' is not a known country code") + ) + country = None + + do_not_call_raw = record.get("do_not_call", "") + do_not_call = _parse_bool(do_not_call_raw) + if do_not_call is None: + errors.append( + RowError(idx, "do_not_call", "Use yes/no, true/false, or 1/0") + ) + do_not_call = False + + account_id: str | None = None + account_name = record.get("account_name", "") + if account_name: + account_id = refs.accounts.get(account_name.lower()) + if account_id is None: + errors.append( + RowError(idx, "account_name", f"No account named '{account_name}'") + ) + + assigned_ids: list[str] = [] + for assigned_email in _split_multi(record.get("assigned_emails", "")): + if not EMAIL_RE.match(assigned_email): + errors.append( + RowError( + idx, + "assigned_emails", + f"'{assigned_email}' is not a valid email", + ) + ) + continue + resolved = refs.profiles.get(assigned_email.lower()) + if resolved is None: + errors.append( + RowError( + idx, + "assigned_emails", + f"No active member with email '{assigned_email}'", + ) + ) + else: + assigned_ids.append(resolved) + + team_ids: list[str] = [] + for team_name in _split_multi(record.get("team_names", "")): + resolved = refs.teams.get(team_name.lower()) + if resolved is None: + errors.append( + RowError(idx, "team_names", f"No team named '{team_name}'") + ) + else: + team_ids.append(resolved) + + for length_field in ( + "organization", + "title", + "department", + "address_line", + "city", + "state", + ): + val = record.get(length_field, "") + if val and len(val) > NAME_MAX_LEN: + errors.append( + RowError(idx, length_field, f"Exceeds {NAME_MAX_LEN} characters") + ) + + postcode = record.get("postcode", "") + if postcode and len(postcode) > 64: + errors.append(RowError(idx, "postcode", "Exceeds 64 characters")) + + if errors: + return errors, None + + return [], ValidatedRow( + row=idx, + first_name=first_name, + last_name=last_name, + email=email or None, + phone=phone or None, + organization=record.get("organization") or None, + title=record.get("title") or None, + department=record.get("department") or None, + do_not_call=do_not_call, + linkedin_url=linkedin_url or None, + address_line=record.get("address_line") or None, + city=record.get("city") or None, + state=record.get("state") or None, + postcode=postcode or None, + country=country, + description=record.get("description") or None, + account_id=account_id, + assigned_ids=assigned_ids, + team_ids=team_ids, + tag_names=_split_multi(record.get("tags", "")), + ) + + +def commit_rows(file_bytes: bytes, org, profile) -> dict[str, Any]: + """Re-parse the uploaded CSV and create contacts in a single transaction. + + Returning structured counts lets the UI render a per-row outcome strip. + Any unexpected error rolls back the whole batch. IntegrityError races + (e.g. a contact with the same email created between preview and commit + by another request) are caught and reported as a 400-equivalent payload + so the user sees an actionable message instead of a generic 500. + """ + result = parse_and_validate(file_bytes, org) + if result.header_error: + return { + "error": True, + "header_error": result.header_error, + "created": 0, + } + if result.errors: + # Refuse to write anything if any row failed; users fix the file first. + return { + "error": True, + "message": "Fix the invalid rows before importing", + "errors": [e.to_dict() for e in result.errors], + "created": 0, + } + + try: + return _commit_validated(result.valid, org, profile) + except IntegrityError as exc: + # The atomic block rolled back; surface a friendly message and ask + # the user to re-preview so the conflict shows up as a row error. + return { + "error": True, + "message": ( + "A contact was created concurrently that conflicts with this " + "import (likely a duplicate email). Re-run preview and try again." + ), + "detail": str(exc), + "created": 0, + } + + +@transaction.atomic +def _commit_validated(rows: list[ValidatedRow], org, profile) -> dict[str, Any]: + created_ids: list[str] = [] + tag_cache: dict[str, Tags] = {} + + for vr in rows: + contact = Contact.objects.create( + first_name=vr.first_name, + last_name=vr.last_name, + email=vr.email, + phone=vr.phone, + organization=vr.organization, + title=vr.title, + department=vr.department, + do_not_call=vr.do_not_call, + linkedin_url=vr.linkedin_url, + address_line=vr.address_line, + city=vr.city, + state=vr.state, + postcode=vr.postcode, + country=vr.country, + description=vr.description, + account_id=vr.account_id, + org=org, + created_by=profile.user, + ) + if vr.assigned_ids: + contact.assigned_to.set(vr.assigned_ids) + if vr.team_ids: + contact.teams.set(vr.team_ids) + if vr.tag_names: + tag_objs = [_get_or_create_tag(name, org, tag_cache) for name in vr.tag_names] + contact.tags.set(tag_objs) + created_ids.append(str(contact.id)) + + return { + "error": False, + "created": len(created_ids), + "ids": created_ids, + } + + +def _get_or_create_tag(name: str, org, cache: dict[str, Tags]) -> Tags: + # slug+org is the unique key on Tags; using it for get_or_create collapses + # the SELECT-then-INSERT race when concurrent imports reference the same tag. + slug = slugify(name) or name.lower() + if slug in cache: + return cache[slug] + tag, _ = Tags.objects.get_or_create( + slug=slug, + org=org, + defaults={"name": name}, + ) + cache[slug] = tag + return tag diff --git a/backend/contacts/tests/test_import.py b/backend/contacts/tests/test_import.py new file mode 100644 index 00000000..10516fdf --- /dev/null +++ b/backend/contacts/tests/test_import.py @@ -0,0 +1,435 @@ +"""Tests for the contacts CSV import endpoints (preview + commit). + +Edge cases the policy specifically targets: + * email dup vs DB and within-file (DB-enforced unique per org) + * phone dup vs DB and within-file (normalized to last-10 digits) + * full-name dup vs DB, but only when the row has no email and no phone + * permission gate (non-admin user blocked) +""" + +import csv +import io + +import pytest +from django.core.files.uploadedfile import SimpleUploadedFile + +from accounts.models import Account +from common.models import Profile, Tags, Teams +from contacts.models import Contact + + +def _csv(headers: list[str], rows: list[list[str]]) -> SimpleUploadedFile: + """Build an in-memory CSV upload from header + row lists.""" + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow(headers) + for row in rows: + writer.writerow(row) + return SimpleUploadedFile( + "contacts.csv", + buf.getvalue().encode("utf-8"), + content_type="text/csv", + ) + + +@pytest.fixture +def existing_contact(org_a, admin_user): + """A contact already in org_a — used for vs-DB duplicate tests.""" + return Contact.objects.create( + first_name="Pat", + last_name="Lee", + email="pat@acme.test", + phone="+1 (202) 555-1234", + org=org_a, + created_by=admin_user, + ) + + +@pytest.fixture +def account_a(org_a, admin_user): + return Account.objects.create(name="Acme Corp", org=org_a, created_by=admin_user) + + +@pytest.fixture +def team_a(org_a): + return Teams.objects.create(name="Sales", description="", org=org_a) + + +@pytest.mark.django_db +class TestImportPreview: + def test_happy_path(self, admin_client, org_a, account_a, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "email", "phone", "account_name", "tags"], + [["Alice", "Smith", "alice@example.com", "+1 555 111 2222", "Acme Corp", "vip;newsletter"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + assert response.status_code == 200, response.json() + body = response.json() + assert body["header_error"] is None + assert body["summary"] == {"total": 1, "valid": 1, "invalid": 0} + row = body["valid"][0] + assert row["first_name"] == "Alice" + assert row["last_name"] == "Smith" + assert row["account_id"] == str(account_a.id) + assert row["tag_names"] == ["vip", "newsletter"] + # Preview must not write anything + assert Contact.objects.filter(org=org_a).count() == 0 + + def test_missing_required_header(self, admin_client, admin_profile): + csv_file = _csv(["first_name"], [["Alice"]]) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + assert response.status_code == 200 + body = response.json() + assert "last_name" in (body["header_error"] or "") + + def test_unknown_header_rejected(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "totally_made_up"], + [["Alice", "Smith", "x"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["header_error"] and "totally_made_up" in body["header_error"] + + def test_email_duplicate_vs_db( + self, admin_client, existing_contact, admin_profile + ): + csv_file = _csv( + ["first_name", "last_name", "email"], + [["Different", "Person", "pat@acme.test"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + fields = {(e["row"], e["field"]) for e in body["errors"]} + assert (1, "email") in fields + + def test_email_duplicate_case_insensitive( + self, admin_client, existing_contact, admin_profile + ): + csv_file = _csv( + ["first_name", "last_name", "email"], + [["Different", "Person", "PAT@ACME.TEST"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + + def test_email_duplicate_within_file(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "email"], + [ + ["A", "One", "dup@example.com"], + ["B", "Two", "dup@example.com"], + ], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["valid"] == 1 + assert body["summary"]["invalid"] == 1 + # The second row is the one that's flagged + err = body["errors"][0] + assert err["row"] == 2 + assert err["field"] == "email" + assert "row 1" in err["message"] + + def test_phone_duplicate_vs_db_normalized( + self, admin_client, existing_contact, admin_profile + ): + # existing_contact.phone == "+1 (202) 555-1234"; this is the same + # number written differently — normalizer strips to "2025551234". + csv_file = _csv( + ["first_name", "last_name", "phone"], + [["Different", "Person", "202-555-1234"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + fields = {(e["row"], e["field"]) for e in body["errors"]} + assert (1, "phone") in fields + + def test_phone_duplicate_within_file(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "phone"], + [ + ["A", "One", "+1 555 000 1111"], + ["B", "Two", "5550001111"], + ], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["valid"] == 1 + assert body["summary"]["invalid"] == 1 + err = body["errors"][0] + assert err["row"] == 2 and err["field"] == "phone" + + def test_full_name_dup_blocked_without_disambiguator( + self, admin_client, existing_contact, admin_profile + ): + # existing_contact is "Pat Lee" with pat@acme.test + a phone. Row has + # the same name but NO email and NO phone → can't disambiguate. + csv_file = _csv( + ["first_name", "last_name"], + [["Pat", "Lee"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + fields = {(e["row"], e["field"]) for e in body["errors"]} + assert (1, "first_name") in fields + + def test_full_name_dup_allowed_with_different_email( + self, admin_client, existing_contact, admin_profile + ): + # Same name, different email → allowed; could be a different person. + csv_file = _csv( + ["first_name", "last_name", "email"], + [["Pat", "Lee", "other-pat@acme.test"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"] == {"total": 1, "valid": 1, "invalid": 0} + + def test_full_name_dup_within_file_blocked( + self, admin_client, admin_profile + ): + # Two name-only rows with no email/phone — second row must be flagged. + csv_file = _csv( + ["first_name", "last_name"], + [ + ["Same", "Name"], + ["Same", "Name"], + ], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["valid"] == 1 + assert body["summary"]["invalid"] == 1 + err = body["errors"][0] + assert err["row"] == 2 and err["field"] == "first_name" + assert "row 1" in err["message"] + + def test_linkedin_url_with_only_scheme_rejected( + self, admin_client, admin_profile + ): + # "https://" passes the old regex but is not a valid URL — Django's + # URLValidator catches it. + csv_file = _csv( + ["first_name", "last_name", "linkedin_url"], + [["A", "B", "https://"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + assert any(e["field"] == "linkedin_url" for e in body["errors"]) + + def test_invalid_email_format(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "email"], + [["A", "B", "not-an-email"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + assert any(e["field"] == "email" for e in body["errors"]) + + def test_invalid_phone_format(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "phone"], + [["A", "B", "abc"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + assert any(e["field"] == "phone" for e in body["errors"]) + + def test_invalid_country_code(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "country"], + [["A", "B", "ZZ"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + assert any(e["field"] == "country" for e in body["errors"]) + + def test_invalid_linkedin_url(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "linkedin_url"], + [["A", "B", "linkedin.com/in/a"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + assert any(e["field"] == "linkedin_url" for e in body["errors"]) + + def test_unknown_account_ref(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "account_name"], + [["A", "B", "Nonexistent Co"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert body["summary"]["invalid"] == 1 + assert any(e["field"] == "account_name" for e in body["errors"]) + + def test_unknown_assigned_email_ref(self, admin_client, admin_profile): + csv_file = _csv( + ["first_name", "last_name", "assigned_emails"], + [["A", "B", "ghost@nowhere.test"]], + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + body = response.json() + assert any(e["field"] == "assigned_emails" for e in body["errors"]) + + def test_permission_denied_for_regular_user( + self, user_client, user_profile + ): + csv_file = _csv( + ["first_name", "last_name"], + [["A", "B"]], + ) + response = user_client.post( + "/api/contacts/import/preview/", {"file": csv_file}, format="multipart" + ) + assert response.status_code == 403 + + def test_non_csv_extension_rejected(self, admin_client, admin_profile): + upload = SimpleUploadedFile( + "contacts.xlsx", + b"first_name,last_name\nA,B\n", + content_type="text/csv", + ) + response = admin_client.post( + "/api/contacts/import/preview/", {"file": upload}, format="multipart" + ) + assert response.status_code == 400 + + +@pytest.mark.django_db +class TestImportCommit: + def test_commits_in_one_transaction( + self, admin_client, org_a, account_a, team_a, admin_profile + ): + csv_file = _csv( + ["first_name", "last_name", "email", "phone", "account_name", "team_names", "tags"], + [ + ["Alice", "Smith", "alice@example.com", "+1 555 111 2222", "Acme Corp", "Sales", "vip"], + ["Bob", "Jones", "bob@example.com", "+1 555 333 4444", "Acme Corp", "Sales", "vip;newsletter"], + ], + ) + response = admin_client.post( + "/api/contacts/import/commit/", {"file": csv_file}, format="multipart" + ) + assert response.status_code == 200, response.json() + body = response.json() + assert body["error"] is False + assert body["created"] == 2 + contacts = Contact.objects.filter(org=org_a) + assert contacts.count() == 2 + alice = contacts.get(email="alice@example.com") + assert alice.account_id == account_a.id + assert list(alice.teams.values_list("name", flat=True)) == ["Sales"] + assert set(alice.tags.values_list("name", flat=True)) == {"vip"} + # Tag was reused across both rows + assert Tags.objects.filter(name="vip", org=org_a).count() == 1 + + def test_commit_refuses_if_any_row_invalid( + self, admin_client, org_a, existing_contact, admin_profile + ): + # Row 1 is valid; row 2 collides on email with the existing contact. + # The whole batch must be rejected — atomic = all-or-nothing. + csv_file = _csv( + ["first_name", "last_name", "email"], + [ + ["New", "Person", "newp@example.com"], + ["Other", "Pat", "pat@acme.test"], + ], + ) + response = admin_client.post( + "/api/contacts/import/commit/", {"file": csv_file}, format="multipart" + ) + assert response.status_code == 400 + body = response.json() + assert body["error"] is True + assert body["created"] == 0 + # Only the original contact exists + assert Contact.objects.filter(org=org_a).count() == 1 + + def test_permission_denied_for_regular_user(self, user_client, user_profile): + csv_file = _csv( + ["first_name", "last_name"], + [["A", "B"]], + ) + response = user_client.post( + "/api/contacts/import/commit/", {"file": csv_file}, format="multipart" + ) + assert response.status_code == 403 + + def test_integrity_error_race_returns_400_not_500( + self, admin_client, org_a, admin_user, admin_profile, monkeypatch + ): + # Simulate the case where another request inserts a colliding contact + # AFTER parse_and_validate has read the DB but BEFORE the create runs. + # The DB then raises IntegrityError on the unique_contact_email_per_org + # constraint — this must surface as a 400 with a friendly message, not + # a 500 from Django's exception middleware. + from django.db import IntegrityError + + from contacts.services import csv_import + + original_create = csv_import.Contact.objects.create + + def racing_create(**kwargs): + raise IntegrityError("duplicate key value violates unique constraint") + + monkeypatch.setattr(csv_import.Contact.objects, "create", racing_create) + + csv_file = _csv( + ["first_name", "last_name", "email"], + [["Race", "Condition", "race@example.com"]], + ) + response = admin_client.post( + "/api/contacts/import/commit/", {"file": csv_file}, format="multipart" + ) + assert response.status_code == 400 + body = response.json() + assert body["error"] is True + assert body["created"] == 0 + assert "concurrently" in body["message"].lower() diff --git a/backend/contacts/urls.py b/backend/contacts/urls.py index b9ec8b92..af4c7189 100644 --- a/backend/contacts/urls.py +++ b/backend/contacts/urls.py @@ -1,11 +1,22 @@ from django.urls import path -from contacts import views +from contacts import import_views, views app_name = "api_contacts" urlpatterns = [ path("", views.ContactsListView.as_view()), + # CSV import (must be before / to avoid being captured as an ID) + path( + "import/preview/", + import_views.ContactImportPreviewView.as_view(), + name="contacts_import_preview", + ), + path( + "import/commit/", + import_views.ContactImportCommitView.as_view(), + name="contacts_import_commit", + ), path("/", views.ContactDetailView.as_view()), path("comment//", views.ContactCommentView.as_view()), path("attachment//", views.ContactAttachmentView.as_view()), diff --git a/frontend/src/lib/components/contacts/ContactImportDrawer.svelte b/frontend/src/lib/components/contacts/ContactImportDrawer.svelte new file mode 100644 index 00000000..a2be6972 --- /dev/null +++ b/frontend/src/lib/components/contacts/ContactImportDrawer.svelte @@ -0,0 +1,478 @@ + + + onOpenChange?.(v)}> + + + + + Import contacts from CSV + + + Upload a CSV of contacts. We'll validate every row — including duplicates + by email, phone, and name — before writing anything. + + + + {#if step === 'select'} +
+ + + +
+

CSV format

+

+ Required headers: first_name, + last_name. + Optional: email, phone, organization, title, department, do_not_call (yes/no), + linkedin_url, address_line, city, state, postcode, country (2-letter code), + description, account_name, assigned_emails, team_names, tags + (semicolon-separated for the last three). +

+

+ Duplicates: rows with a previously-used email or phone (in your org or earlier in + the file) are flagged. Two people with the same name are allowed only if at least + one has an email or phone to tell them apart. +

+ +
+ + {#if formError} +
+ + {formError} +
+ {/if} +
+ + + + + + {:else if step === 'preview' && preview} + {@const hasErrors = preview.errors.length > 0} +
+
+ + Total: {preview.summary.total} + + + Valid: {preview.summary.valid} + + {#if preview.summary.invalid > 0} + + Invalid: {preview.summary.invalid} + + {/if} +
+ + {#if preview.valid.length > 0} +
+ + + + + + + + + + + + {#each preview.valid.slice(0, 20) as row (row.row)} + + + + + + + + {/each} + +
#NameEmailPhoneCompany
{row.row} + {row.first_name} {row.last_name} + + {row.email || '—'} + + {row.phone || '—'} + + {row.organization || '—'} +
+ {#if preview.valid.length > 20} +
+ Showing first 20 of {preview.valid.length} valid rows +
+ {/if} +
+ {/if} + + {#if hasErrors} +
+
+

+ {preview.errors.length} error{preview.errors.length === 1 ? '' : 's'} — fix the CSV before importing +

+ +
+
+ + + + + + + + + + {#each preview.errors as err (`${err.row}-${err.field}-${err.message}`)} + + + + + + {/each} + +
RowFieldProblem
{err.row}{err.field}{err.message}
+
+
+ {/if} + + {#if commitErrors.length > 0} +
+
+

+ Server rejected {commitErrors.length} row{commitErrors.length === 1 ? '' : 's'} during import — the file may have changed since preview +

+ +
+
+ + + + + + + + + + {#each commitErrors as err (`${err.row}-${err.field}-${err.message}`)} + + + + + + {/each} + +
RowFieldProblem
{err.row}{err.field}{err.message}
+
+
+ {/if} + + {#if formError} +
+ + {formError} +
+ {/if} +
+ + + + + + {:else if step === 'done' && commitResult} +
+
+ +
+

+ Imported {commitResult.created} contact{commitResult.created === 1 ? '' : 's'} +

+

+ The contacts list has been refreshed. +

+
+
+
+ + + + {/if} + + diff --git a/frontend/src/routes/(app)/contacts/+page.server.js b/frontend/src/routes/(app)/contacts/+page.server.js index 16733142..e0b9d649 100644 --- a/frontend/src/routes/(app)/contacts/+page.server.js +++ b/frontend/src/routes/(app)/contacts/+page.server.js @@ -10,8 +10,38 @@ */ import { error, fail } from '@sveltejs/kit'; +import { env } from '$env/dynamic/public'; import { apiRequest, buildQueryParams } from '$lib/api-helpers.js'; +const API_BASE_URL = `${env.PUBLIC_DJANGO_API_URL}/api`; + +/** + * Forward a multipart upload to Django, preserving the auth cookie. + * apiRequest only handles JSON, so file uploads need a direct fetch. + */ +async function forwardMultipart(endpoint, file, cookies) { + const accessToken = cookies?.get?.('jwt_access'); + const fd = new FormData(); + fd.append('file', file, file.name || 'upload.csv'); + const response = await fetch(`${API_BASE_URL}${endpoint}`, { + method: 'POST', + headers: accessToken ? { Authorization: `Bearer ${accessToken}` } : {}, + body: fd + }); + const raw = await response.text(); + let body; + try { + body = raw ? JSON.parse(raw) : {}; + } catch (err) { + console.error( + `forwardMultipart: non-JSON response from ${endpoint} (status ${response.status}):`, + raw.slice(0, 500) + ); + body = {}; + } + return { status: response.status, body }; +} + /** @type {import('./$types').PageServerLoad} */ export async function load({ url, locals, cookies }) { const org = locals.org; @@ -384,5 +414,34 @@ export const actions = { console.error('Error deleting contact:', err); return fail(400, { error: err.message || 'Failed to delete contact' }); } + }, + + importPreview: async ({ request, cookies }) => { + const form = await request.formData(); + const file = form.get('file'); + if (!(file instanceof File) || file.size === 0) { + return fail(400, { importError: 'Please choose a CSV file.' }); + } + const { status, body } = await forwardMultipart('/contacts/import/preview/', file, cookies); + if (status !== 200) { + return fail(status, { importError: body?.message || 'Preview failed' }); + } + return { importPreview: body }; + }, + + importCommit: async ({ request, cookies }) => { + const form = await request.formData(); + const file = form.get('file'); + if (!(file instanceof File) || file.size === 0) { + return fail(400, { importError: 'Please choose a CSV file.' }); + } + const { status, body } = await forwardMultipart('/contacts/import/commit/', file, cookies); + if (status !== 200) { + return fail(status, { + importError: body?.message || body?.header_error || 'Import failed', + importErrors: body?.errors || [] + }); + } + return { importCommit: body }; } }; diff --git a/frontend/src/routes/(app)/contacts/+page.svelte b/frontend/src/routes/(app)/contacts/+page.svelte index b3a8d6f4..1136917d 100644 --- a/frontend/src/routes/(app)/contacts/+page.svelte +++ b/frontend/src/routes/(app)/contacts/+page.svelte @@ -7,6 +7,7 @@ import { Plus, Eye, + Upload, User, Mail, Phone, @@ -23,6 +24,7 @@ import { CrmDrawer } from '$lib/components/ui/crm-drawer'; import { CustomFieldsPanel } from '$lib/components/custom-fields'; import { CommentSection } from '$lib/components/ui/comment-section'; + import ContactImportDrawer from '$lib/components/contacts/ContactImportDrawer.svelte'; import { getCurrentUser } from '$lib/api.js'; import { Button } from '$lib/components/ui/button/index.js'; import * as DropdownMenu from '$lib/components/ui/dropdown-menu/index.js'; @@ -99,6 +101,9 @@ let dropdownOptionsLoaded = $state(false); let dropdownOptionsLoading = $state(false); + // CSV bulk-import drawer + let importOpen = $state(false); + // Account from URL param (for quick action from account page) let accountFromUrl = $state(false); let accountName = $state(''); @@ -793,6 +798,10 @@ {/each} +