"""Import/export service — CSV import with dry-run preview, CSV/XLSX export.""" from __future__ import annotations import csv import io 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.models.contact import Contact, ContactPerson as Company from app.models.contact import Contact, ContactPerson from app.services.company_service import _company_to_dict from app.services.contact_service import _serialize_contact as _contact_to_dict # Expected CSV columns for each entity type COMPANY_COLUMNS = ["name", "industry", "phone", "email", "website", "description"] CONTACT_COLUMNS = ["first_name", "last_name", "email", "phone", "mobile", "position", "department"] def _parse_csv(content: str) -> list[dict[str, str]]: """Parse CSV content into list of dicts.""" reader = csv.DictReader(io.StringIO(content)) return [dict(row) for row in reader] def _validate_row(row: dict[str, str], required: list[str]) -> list[str]: """Validate a single row. Returns list of error messages (empty if valid).""" errors = [] for col in required: val = row.get(col, "").strip() if not val: errors.append(f"Missing required field: {col}") return errors async def import_companies( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, csv_content: str, dry_run: bool = False, ) -> dict[str, Any]: """Import companies from CSV. If dry_run=True, no DB changes are made. Returns {total, valid, invalid, errors, created (empty in dry_run)}. """ rows = _parse_csv(csv_content) total = len(rows) valid_rows = [] errors = [] for idx, row in enumerate(rows, start=1): row_errors = _validate_row(row, ["name"]) if row_errors: for e in row_errors: errors.append({"row": idx, "error": e}) else: valid_rows.append(row) if dry_run: return { "total": total, "valid": len(valid_rows), "invalid": len(errors), "errors": errors, "created": [], "dry_run": True, } created = [] for row in valid_rows: company = Company( tenant_id=tenant_id, name=row["name"].strip(), industry=row.get("industry", "").strip() or None, phone=row.get("phone", "").strip() or None, email=row.get("email", "").strip() or None, website=row.get("website", "").strip() or None, description=row.get("description", "").strip() or None, created_by=user_id, updated_by=user_id, ) db.add(company) await db.flush() await log_audit( db, tenant_id, user_id, "import", "company", company.id, changes={"name": company.name}, ) created.append(_company_to_dict(company)) return { "total": total, "valid": len(valid_rows), "invalid": len(errors), "errors": errors, "created": created, "dry_run": False, } async def import_contacts( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, csv_content: str, dry_run: bool = False, ) -> dict[str, Any]: """Import contacts from CSV. If dry_run=True, no DB changes are made. Returns {total, valid, invalid, errors, created (empty in dry_run)}. """ rows = _parse_csv(csv_content) total = len(rows) valid_rows = [] errors = [] for idx, row in enumerate(rows, start=1): row_errors = _validate_row(row, ["first_name", "last_name"]) if row_errors: for e in row_errors: errors.append({"row": idx, "error": e}) else: valid_rows.append(row) if dry_run: return { "total": total, "valid": len(valid_rows), "invalid": len(errors), "errors": errors, "created": [], "dry_run": True, } created = [] for row in valid_rows: contact = Contact( tenant_id=tenant_id, first_name=row["first_name"].strip(), last_name=row["last_name"].strip(), email=row.get("email", "").strip() or None, phone=row.get("phone", "").strip() or None, mobile=row.get("mobile", "").strip() or None, position=row.get("position", "").strip() or None, department=row.get("department", "").strip() or None, 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={"first_name": contact.first_name, "last_name": contact.last_name}, ) created.append(_contact_to_dict(contact)) return { "total": total, "valid": len(valid_rows), "invalid": len(errors), "errors": errors, "created": created, "dry_run": False, } async def import_csv( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, csv_content: str, entity_type: str, dry_run: bool = False, ) -> 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) elif entity_type == "contacts": return await import_contacts(db, tenant_id, user_id, csv_content, dry_run=dry_run) else: return { "total": 0, "valid": 0, "invalid": 0, "errors": [{"row": 0, "error": f"Unknown entity_type: {entity_type}"}], "created": [], "dry_run": dry_run, } async def export_contacts_csv( db: AsyncSession, tenant_id: uuid.UUID, ) -> str: """Export contacts as CSV string.""" q = ( select(Contact) .where( Contact.tenant_id == tenant_id, Contact.deleted_at.is_(None), ) .order_by(Contact.last_name, Contact.first_name) ) result = await db.execute(q) contacts = result.scalars().all() output = io.StringIO() writer = csv.writer(output) writer.writerow( ["id", "first_name", "last_name", "email", "phone", "mobile", "position", "department"] ) for c in contacts: writer.writerow( [ str(c.id), c.first_name, c.last_name, c.email or "", c.phone or "", c.mobile or "", c.position or "", c.department or "", ] ) return output.getvalue()