"""CSV import parsing and export streaming (tasks 3.8, 3.9; spec csv-export). Two rules earn their keep here: * Import is per-row tolerant. One malformed row out of 800 must not abort the batch — it gets reported with a reason while the other 799 import. * Export is UTF-8 **with BOM**. Without it, Excel renders Bangla names as mojibake, which makes the caterer list useless to the person who actually needs it. """ import csv import io from collections.abc import AsyncIterator, Iterable, Sequence from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from app.models.enums import InvitationType, Locale, PreferredChannel from app.services.phone import InvalidPhoneNumberError, normalize_email, normalize_phone #: Fixed column order (FR-5.2). Consumers rely on it, so it is not derived from the model. EXPORT_COLUMNS = [ "guest_id", "full_name", "email", "phone", "whatsapp", "side", "group_tags", "event", "status", "party_size", "adults", "children", "dietary_notes", "message_to_couple", "invited_via", "invited_at", "opened_at", "responded_at", "cancelled_at", "last_reminder_sent_at", "notes", ] CATERER_COLUMNS = ["full_name", "party_size", "dietary_notes"] IMPORT_TEMPLATE_COLUMNS = [ "full_name", "email", "phone", "whatsapp", "side", "group_tags", "invitation_type", "max_guests", "locale", "preferred_channel", ] UTF8_BOM = "" #: Ceiling applied when an import row leaves max_guests blank or unreadable. Matches the #: 1-4 range the RSVP form offers, so a blank column never truncates a guest's answer. DEFAULT_MAX_GUESTS = 4 @dataclass class ImportRow: """One parsed row, valid or not. Invalid rows carry their reason to the UI.""" line_number: int raw: dict[str, str] full_name: str = "" email: str | None = None phone: str | None = None whatsapp: str | None = None side: str = "common" group_tags: list[str] = field(default_factory=list) invitation_type: InvitationType = InvitationType.SINGLE max_guests: int = DEFAULT_MAX_GUESTS locale: Locale = Locale.EN preferred_channel: PreferredChannel = PreferredChannel.AUTO errors: list[str] = field(default_factory=list) @property def is_valid(self) -> bool: return not self.errors def parse_import_csv(content: bytes, mapping: dict[str, str] | None = None) -> list[ImportRow]: """Parse an upload into rows, validating each independently. `mapping` renames incoming headers to canonical ones, so a spreadsheet with "Mobile" instead of "phone" does not need editing by hand first. """ text = content.decode("utf-8-sig", errors="replace") reader = csv.DictReader(io.StringIO(text)) rows: list[ImportRow] = [] for index, raw in enumerate(reader, start=2): # line 1 is the header clean = { (mapping or {}).get(k.strip(), k.strip()).lower(): (v or "").strip() for k, v in raw.items() if k } row = ImportRow(line_number=index, raw=clean) row.full_name = clean.get("full_name", "") if not row.full_name: row.errors.append("full_name is required") raw_phone = clean.get("phone") or "" if raw_phone: try: row.phone = normalize_phone(raw_phone) except InvalidPhoneNumberError as exc: row.errors.append(str(exc)) row.email = normalize_email(clean.get("email")) if row.email and "@" not in row.email: row.errors.append(f"invalid email: {row.email}") row.email = None # PRD §14.2: a row with neither phone nor email is rejected with a reason, # because there would be no way to ever contact that guest. if not row.phone and not row.email: row.errors.append("row has neither a usable phone nor an email") raw_whatsapp = clean.get("whatsapp") or "" if raw_whatsapp: try: row.whatsapp = normalize_phone(raw_whatsapp) except InvalidPhoneNumberError: row.whatsapp = row.phone # fall back rather than fail the row else: row.whatsapp = row.phone side = (clean.get("side") or "common").lower() row.side = side if side in {"bride", "groom", "common"} else "common" tags = clean.get("group_tags") or "" row.group_tags = [t.strip() for t in tags.split(",") if t.strip()] # The `events` column is deliberately not read. The destination is the event in the # request URL (design D11), so a sheet naming its own events could only contradict # the list the admin is looking at. A leftover column in an old file is ignored # rather than rejected, so last season's spreadsheet still imports. invitation_type = (clean.get("invitation_type") or "single").strip().lower() try: row.invitation_type = InvitationType(invitation_type) except ValueError: # Unreadable rather than absent, so it is worth saying: silently seating a # family as a single is the kind of error nobody notices until the day. row.errors.append(f"unknown invitation type: {invitation_type}") try: row.max_guests = max(1, int(clean.get("max_guests") or DEFAULT_MAX_GUESTS)) except ValueError: row.max_guests = DEFAULT_MAX_GUESTS locale = (clean.get("locale") or "en").lower() row.locale = Locale(locale) if locale in {"bn", "en"} else Locale.EN channel = (clean.get("preferred_channel") or "auto").lower() try: row.preferred_channel = PreferredChannel(channel) except ValueError: row.preferred_channel = PreferredChannel.AUTO rows.append(row) return rows def export_filename(event: str | None, status: str | None) -> str: """guests_{event}_{status}_{YYYY-MM-DD_HHmm}.csv (FR-5.4).""" stamp = datetime.now(UTC).strftime("%Y-%m-%d_%H%M") return f"guests_{event or 'all'}_{status or 'all'}_{stamp}.csv" async def stream_csv(columns: Sequence[str], rows: Iterable[Sequence[Any]]) -> AsyncIterator[str]: """Yield CSV incrementally so a 5,000-row export never materialises in memory (FR-5.6). The BOM goes out first, alone, so Excel detects UTF-8 before it reads the header. """ yield UTF8_BOM buffer = io.StringIO() writer = csv.writer(buffer, lineterminator="\n") writer.writerow(columns) yield buffer.getvalue() buffer.seek(0) buffer.truncate(0) for row in rows: writer.writerow(["" if v is None else v for v in row]) yield buffer.getvalue() buffer.seek(0) buffer.truncate(0)