"""Bulk CSV import for organization members. TODO: Move large imports to an async batch job with webhook completion callbacks. """ from __future__ import annotations import re from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.models import Account from app.schemas import ( OrgMemberCreate, OrgMemberImportRequest, OrgMemberImportResponse, OrgMemberImportRowResult, OrgMemberRead, ) from app.services.org_members import create_org_member MAX_IMPORT_ROWS = 300 EMAIL_LOCAL_PATTERN = re.compile( r"^[A-Za-z0-9](?:[A-Za-z0-9._-]{0,62}[A-Za-z0-9])?$" ) def _password_error(value: str) -> str | None: if len(value) < 8: return "Password must be at least 8 characters." if not ( re.search(r"[a-z]", value) and re.search(r"[A-Z]", value) and re.search(r"\d", value) ): return "Password must include uppercase, lowercase, and a number." return None def _normalize_work_email(raw: str, company_domain: str) -> str: value = raw.strip().lower() if "@" in value: return value return f"{value}@{company_domain}" async def import_org_members( session: AsyncSession, *, manager: Account, payload: OrgMemberImportRequest, ) -> OrgMemberImportResponse: company_domain = manager.company_domain if company_domain is None: raise ValueError("Company domain is required.") rows = payload.rows if not rows: raise ValueError("Add at least one person to import.") if len(rows) > MAX_IMPORT_ROWS: raise ValueError(f"You can import up to {MAX_IMPORT_ROWS} people at a time.") errors: list[OrgMemberImportRowResult] = [] normalized_rows: list[tuple[int, OrgMemberCreate]] = [] seen_emails: set[str] = set() seen_invites: set[str] = set() for index, row in enumerate(rows, start=1): row_errors: list[str] = [] first_name = row.first_name.strip() last_name = row.last_name.strip() if not first_name: row_errors.append("First name is required.") if not last_name: row_errors.append("Last name is required.") work_email = _normalize_work_email(row.work_email, company_domain) local_part = work_email.split("@", 1)[0] if not work_email.endswith(f"@{company_domain}"): row_errors.append(f"Work email must use @{company_domain}.") elif not EMAIL_LOCAL_PATTERN.match(local_part): row_errors.append("Work email format is invalid.") if work_email in seen_emails: row_errors.append("Duplicate work email in this file.") seen_emails.add(work_email) password = row.password.strip() password_issue = _password_error(password) if password_issue: row_errors.append(password_issue) invitation_email = ( str(row.invitation_email).lower().strip() if row.invitation_email else None ) if payload.send_invitation: if not invitation_email: row_errors.append( "Invitation email is required when sending invitations." ) elif invitation_email.endswith(f"@{company_domain}"): row_errors.append( "Use a personal email for invitations, not your company domain." ) elif invitation_email in seen_invites: row_errors.append("Duplicate invitation email in this file.") else: seen_invites.add(invitation_email) if row_errors: errors.append( OrgMemberImportRowResult( row=index, email=work_email, success=False, message="; ".join(row_errors), ) ) continue normalized_rows.append( ( index, OrgMemberCreate( first_name=first_name, last_name=last_name, email=work_email, invitation_email=invitation_email, org_role=payload.org_role, send_invitation=payload.send_invitation, force_password_change=payload.force_password_change, password=password, ), ) ) if normalized_rows: existing = await session.scalars( select(Account.email).where( func.lower(Account.email).in_( [item[1].email.lower() for item in normalized_rows] ) ) ) existing_emails = {email.lower() for email in existing.all()} if existing_emails: filtered: list[tuple[int, OrgMemberCreate]] = [] for row_number, member_payload in normalized_rows: if member_payload.email.lower() in existing_emails: errors.append( OrgMemberImportRowResult( row=row_number, email=member_payload.email, success=False, message="That work email is already registered.", ) ) else: filtered.append((row_number, member_payload)) normalized_rows = filtered if errors and not normalized_rows: return OrgMemberImportResponse( created=[], errors=errors, created_count=0, error_count=len(errors), ) created: list[OrgMemberRead] = [] for row_number, member_payload in normalized_rows: try: member = await create_org_member( session, manager=manager, payload=member_payload, ) created.append(member) errors.append( OrgMemberImportRowResult( row=row_number, email=member.email, success=True, message="Created", invitation_sent=member.invitation_sent, ) ) except ValueError as error: errors.append( OrgMemberImportRowResult( row=row_number, email=member_payload.email, success=False, message=str(error), ) ) created_count = len(created) error_count = sum(1 for item in errors if not item.success) return OrgMemberImportResponse( created=created, errors=errors, created_count=created_count, error_count=error_count, )