crm-system/app/services/import_export_service.py

"""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.company import Company
from app.models.contact import Contact
from app.services.company_service import _company_to_dict
from app.services.contact_service import _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()