"""Import/export service — CSV/JSON/XLSX import with dry-run preview, partial-failure, CSV/XLSX export. This service is Contact-specific — it handles import/export for Contacts and Companies (both use the Contact model). It is NOT a generic import/export service. If other entity types need import/export in the future, a separate service or a plugin-based interface should be created. Declared as Contact-specific (P1-23 fix): no pretense of being generic. Uses shared helpers from import_export_helpers.py for parsing, writing, mapping, and validation. Implements partial-failure semantics: valid rows are committed while invalid rows are collected into a structured error report. Uses unified Contact model fields: firstname, surname, email_1, phone_1, phone_2, function. Company import creates Contact with type='company'. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.audit import log_audit from app.core.visibility import apply_visibility_filter from app.models.contact import Contact from app.services import import_export_helpers as helpers from app.services.contact_service import _serialize_contact as _contact_to_dict logger = logging.getLogger(__name__) # Expected CSV columns for each entity type # Company import creates Contact with type='company' using name field COMPANY_COLUMNS = ["name", "industry", "phone", "email", "website"] # Contact import uses unified Contact fields CONTACT_COLUMNS = ["firstname", "surname", "email", "phone", "mobile", "function", "department"] # Target fields for mapping suggestions CONTACT_TARGET_FIELDS = ["firstname", "surname", "email", "phone", "mobile", "function", "department"] COMPANY_TARGET_FIELDS = ["name", "industry", "phone", "email", "website"] # Validators for each entity type CONTACT_VALIDATORS: dict[str, dict] = { "email": {"type": "email"}, } COMPANY_VALIDATORS: dict[str, dict] = { "email": {"type": "email"}, "website": {"type": "url"}, } # Threshold for background job processing (rows above this go to ARQ) BACKGROUND_JOB_THRESHOLD = 1000 def _parse_csv(content: str | bytes) -> list[dict[str, str]]: """Parse file content into list of dicts (delegates to helpers with format detection). Despite the name, this handles CSV, JSON, and XLSX formats by detecting from content. This maintains backward compatibility with existing callers. """ if isinstance(content, str): content = content.encode("utf-8") # Detect format from content magic bytes fmt = helpers.detect_format("upload", content) if fmt == "json": return helpers.parse_json(content) elif fmt == "xlsx": return helpers.parse_xlsx(content) return helpers.parse_csv(content) def _validate_row(row: dict[str, str], required: list[str]) -> list[str]: """Validate a single row (delegates to helpers for backward compat).""" return helpers.validate_row(row, required) def _normalize_contact_row(row: dict[str, str]) -> dict[str, str]: """Normalize contact row to unified field names.""" firstname = (row.get("firstname") or row.get("first_name") or "").strip() surname = (row.get("surname") or row.get("last_name") or "").strip() row["firstname"] = firstname row["surname"] = surname # Map old column names to unified ones if "email" not in row and "email_address" in row: row["email"] = row["email_address"] if "mobile" not in row and "phone_2" in row: row["mobile"] = row["phone_2"] if "function" not in row and "position" in row: row["function"] = row["position"] return row def _normalize_company_row(row: dict[str, str]) -> dict[str, str]: """Normalize company row to unified field names.""" name = (row.get("name") or row.get("company") or row.get("company_name") or "").strip() row["name"] = name if "email" not in row and "email_address" in row: row["email"] = row["email_address"] if "website" not in row and "url" in row: row["website"] = row["url"] if "website" not in row and "homepage" in row: row["website"] = row["homepage"] return row async def import_companies( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, csv_content: str | bytes, dry_run: bool = False, field_mapping: dict[str, str] | None = None, ) -> dict[str, Any]: """Import companies from CSV as Contact with type='company'. Uses unified Contact model: name field for company name, email_1/phone_1 for contact info. Implements partial-failure: valid rows committed, invalid rows collected in error report. """ rows = _parse_csv(csv_content) total = len(rows) errors: list[dict] = [] valid_rows: list[dict] = [] for idx, row in enumerate(rows, start=1): # Apply field mapping if provided if field_mapping: row = helpers.map_fields(row, field_mapping) row = _normalize_company_row(row) row_errors = helpers.validate_row(row, ["name"], COMPANY_VALIDATORS) if row_errors: for e in row_errors: errors.append({"row": idx, "field": "", "message": e}) else: valid_rows.append(row) # Count unique failed rows (a row may have multiple errors) failed_row_count = len({e["row"] for e in errors}) if dry_run: return helpers.build_import_result( total=total, succeeded=len(valid_rows), failed=failed_row_count, errors=errors, dry_run=True, ) created: list[dict] = [] failed: list[dict] = [] succeeded = 0 for idx, row in enumerate(valid_rows, start=1): try: contact = Contact( tenant_id=tenant_id, type="company", name=row["name"].strip(), displayname=row["name"].strip(), email_1=row.get("email", "").strip() or None, phone_1=row.get("phone", "").strip() or None, website=row.get("website", "").strip() or None, owner_id=user_id, created_by=user_id, updated_by=user_id, ) db.add(contact) await db.flush() await log_audit( db, tenant_id, user_id, "import", "contact", contact.id, changes={"name": contact.name, "type": "company"}, ) created.append(_contact_to_dict(contact)) succeeded += 1 except Exception as exc: logger.warning("Import companies: row %d failed: %s", idx, exc) await db.rollback() failed.append({"row": idx, "field": "", "message": str(exc)}) if succeeded > 0: try: await db.commit() except Exception as exc: logger.error("Import companies: commit failed: %s", exc) await db.rollback() return helpers.build_import_result( total=total, succeeded=0, failed=len(valid_rows), errors=errors + [{"row": 0, "field": "", "message": f"Commit failed: {exc}"}], created=[], ) # Count unique failed rows from validation errors + runtime failures all_failed_rows = {e["row"] for e in errors} | {f["row"] for f in failed} return helpers.build_import_result( total=total, succeeded=succeeded, failed=len(all_failed_rows), errors=errors + failed, created=created, ) async def import_contacts( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, csv_content: str | bytes, dry_run: bool = False, field_mapping: dict[str, str] | None = None, ) -> dict[str, Any]: """Import contacts from CSV using unified Contact model fields. CSV columns: firstname, surname, email, phone, mobile, function, department. Maps to Contact fields: firstname, surname, email_1, phone_1, phone_2, function. Implements partial-failure: valid rows committed, invalid rows collected in error report. """ rows = _parse_csv(csv_content) total = len(rows) errors: list[dict] = [] valid_rows: list[dict] = [] for idx, row in enumerate(rows, start=1): # Apply field mapping if provided if field_mapping: row = helpers.map_fields(row, field_mapping) row = _normalize_contact_row(row) # Validate: at least one of firstname or surname required if not row["firstname"] and not row["surname"]: errors.append({"row": idx, "field": "", "message": "Missing required field: firstname or surname"}) continue row_errors = helpers.validate_row(row, [], CONTACT_VALIDATORS) if row_errors: for e in row_errors: errors.append({"row": idx, "field": "", "message": e}) else: valid_rows.append(row) # Count unique failed rows (a row may have multiple errors) failed_row_count = len({e["row"] for e in errors}) if dry_run: return helpers.build_import_result( total=total, succeeded=len(valid_rows), failed=failed_row_count, errors=errors, dry_run=True, ) created: list[dict] = [] failed: list[dict] = [] succeeded = 0 for idx, row in enumerate(valid_rows, start=1): try: contact = Contact( tenant_id=tenant_id, type="person", firstname=row["firstname"].strip() or None, surname=row["surname"].strip() or None, displayname=f"{row['firstname']} {row['surname']}".strip(), email_1=row.get("email", "").strip() or None, phone_1=row.get("phone", "").strip() or None, phone_2=row.get("mobile", "").strip() or None, owner_id=user_id, created_by=user_id, updated_by=user_id, ) db.add(contact) await db.flush() await log_audit( db, tenant_id, user_id, "import", "contact", contact.id, changes={"firstname": contact.firstname, "surname": contact.surname}, ) created.append(_contact_to_dict(contact)) succeeded += 1 except Exception as exc: logger.warning("Import contacts: row %d failed: %s", idx, exc) await db.rollback() failed.append({"row": idx, "field": "", "message": str(exc)}) if succeeded > 0: try: await db.commit() except Exception as exc: logger.error("Import contacts: commit failed: %s", exc) await db.rollback() return helpers.build_import_result( total=total, succeeded=0, failed=len(valid_rows), errors=errors + [{"row": 0, "field": "", "message": f"Commit failed: {exc}"}], created=[], ) # Count unique failed rows from validation errors + runtime failures all_failed_rows = {e["row"] for e in errors} | {f["row"] for f in failed} return helpers.build_import_result( total=total, succeeded=succeeded, failed=len(all_failed_rows), errors=errors + failed, created=created, ) async def import_csv( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, csv_content: str | bytes, entity_type: str, dry_run: bool = False, field_mapping: dict[str, str] | None = None, ) -> dict[str, Any]: """Generic CSV import dispatcher based on entity_type ('companies' or 'contacts').""" if entity_type == "companies": return await import_companies(db, tenant_id, user_id, csv_content, dry_run=dry_run, field_mapping=field_mapping) elif entity_type == "contacts": return await import_contacts(db, tenant_id, user_id, csv_content, dry_run=dry_run, field_mapping=field_mapping) else: return helpers.build_import_result( total=0, succeeded=0, failed=1, errors=[{"row": 0, "field": "", "message": f"Unknown entity_type: {entity_type}"}], ) # ─── Export functions ──────────────────────────────────────────────────────── async def export_contacts_csv( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None = None, is_system_admin: bool = False, ) -> str: """Export contacts as CSV string using unified Contact model fields.""" q = ( select(Contact) .where( Contact.tenant_id == tenant_id, Contact.deleted_at.is_(None), ) .order_by(Contact.surname, Contact.firstname) ) if user_id: q = await apply_visibility_filter( db, q, "contact", Contact, user_id, tenant_id, is_system_admin ) result = await db.execute(q) contacts = result.scalars().all() headers = ["id", "type", "firstname", "surname", "name", "email", "phone", "mobile", "city", "postalcode", "country"] rows = [ { "id": str(c.id), "type": c.type or "person", "firstname": c.firstname or "", "surname": c.surname or "", "name": c.name or "", "email": c.email_1 or "", "phone": c.phone_1 or "", "mobile": c.phone_2 or "", "city": c.mailing_city or "", "postalcode": c.mailing_postalcode or "", "country": c.mailing_country or "", } for c in contacts ] return helpers.write_csv(rows, headers).decode("utf-8") async def export_companies_csv( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None = None, is_system_admin: bool = False, ) -> str: """Export companies (Contact.type == 'company') as CSV string.""" q = ( select(Contact) .where( Contact.tenant_id == tenant_id, Contact.deleted_at.is_(None), Contact.type == "company", ) .order_by(Contact.name) ) if user_id: q = await apply_visibility_filter( db, q, "contact", Contact, user_id, tenant_id, is_system_admin ) result = await db.execute(q) companies = result.scalars().all() headers = ["id", "type", "name", "email", "phone", "website", "city", "postalcode", "country"] rows = [ { "id": str(c.id), "type": c.type or "company", "name": c.name or "", "email": c.email_1 or "", "phone": c.phone_1 or "", "website": c.website or "", "city": c.mailing_city or "", "postalcode": c.mailing_postalcode or "", "country": c.mailing_country or "", } for c in companies ] return helpers.write_csv(rows, headers).decode("utf-8") # ─── Preview & Validate ────────────────────────────────────────────────────── def preview_import( filename: str, content: bytes, entity_type: str, ) -> dict[str, Any]: """Parse a file and return preview data (first 10 rows) + mapping suggestion. Does NOT touch the database. """ rows = helpers.parse_file(filename, content) columns = list(rows[0].keys()) if rows else [] preview_rows = rows[:10] if entity_type == "contacts": target_fields = CONTACT_TARGET_FIELDS elif entity_type == "companies": target_fields = COMPANY_TARGET_FIELDS else: target_fields = [] mapping_suggestion = helpers.suggest_mapping(columns, target_fields) return { "total_rows": len(rows), "columns": columns, "preview_rows": preview_rows, "mapping_suggestion": mapping_suggestion, "target_fields": target_fields, } def validate_import( filename: str, content: bytes, entity_type: str, field_mapping: dict[str, str] | None = None, ) -> dict[str, Any]: """Validate all rows against mapping and return error report without importing.""" rows = helpers.parse_file(filename, content) total = len(rows) errors: list[dict] = [] valid_count = 0 for idx, row in enumerate(rows, start=1): if field_mapping: row = helpers.map_fields(row, field_mapping) if entity_type == "contacts": row = _normalize_contact_row(row) if not row["firstname"] and not row["surname"]: errors.append({"row": idx, "field": "", "message": "Missing required field: firstname or surname"}) continue row_errors = helpers.validate_row(row, [], CONTACT_VALIDATORS) elif entity_type == "companies": row = _normalize_company_row(row) row_errors = helpers.validate_row(row, ["name"], COMPANY_VALIDATORS) else: errors.append({"row": idx, "field": "", "message": f"Unknown entity_type: {entity_type}"}) continue if row_errors: for e in row_errors: errors.append({"row": idx, "field": "", "message": e}) else: valid_count += 1 # Count unique failed rows failed_row_count = len({e["row"] for e in errors}) return helpers.build_import_result( total=total, succeeded=valid_count, failed=failed_row_count, errors=errors, dry_run=True, )