leocrm/app/services/export_service.py

"""Export service — CSV and other format exports for CRM entities."""

from __future__ import annotations

import csv
import io
import uuid
from typing import Any

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.sensitive_data import get_sensitive_fields
from app.models.contact import Contact


class ExportService:
    """Handles export operations for CRM entities."""

    @staticmethod
    async def export_contacts_csv(
        db: AsyncSession,
        tenant_id: uuid.UUID,
        contact_type: str | None = None,
        search: str | None = None,
        user_id: uuid.UUID | None = None,
        is_system_admin: bool = False,
    ) -> str:
        """Export contacts as CSV string. Only exports visible contacts."""
        from app.core.visibility import apply_visibility_filter

        base = select(Contact).where(
            Contact.tenant_id == tenant_id,
            Contact.deleted_at.is_(None),
        )
        if contact_type:
            base = base.where(Contact.type == contact_type)
        if search:
            base = base.where(Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search)))

        # Apply visibility filter
        if user_id and not is_system_admin:
            base = await apply_visibility_filter(
                db, base, "contact", Contact, user_id, tenant_id, is_system_admin
            )

        base = base.order_by(Contact.displayname)

        result = await db.execute(base)
        contacts = result.scalars().all()

        # Exclude sensitive fields that must never appear in exports
        sensitive = get_sensitive_fields("contact")

        all_headers = [
            "id", "type", "displayname", "name", "firstname", "surname", "code",
            "email_1", "email_2", "phone_1", "phone_2", "website",
            "mailing_city", "mailing_postalcode", "mailing_country",
            "vat_code", "tags",
        ]
        # Drop headers for sensitive fields (e.g. password_hash would never be
        # in a contact row, but this is a safety net).
        export_headers = [h for h in all_headers if h not in sensitive]

        output = io.StringIO()
        writer = csv.writer(output)
        writer.writerow(export_headers)
        for c in contacts:
            row = []
            for h in export_headers:
                if h in sensitive:
                    row.append("")
                else:
                    row.append(getattr(c, h, None) or "")
            writer.writerow(row)
        return output.getvalue()


export_service = ExportService()