70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
|
"""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.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()
|
||
|
|
|
||
|
|
output = io.StringIO()
|
||
|
|
writer = csv.writer(output)
|
||
|
|
writer.writerow([
|
||
|
|
"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",
|
||
|
|
])
|
||
|
|
for c in contacts:
|
||
|
|
writer.writerow([
|
||
|
|
str(c.id), c.type, c.displayname, c.name or "", c.firstname or "", c.surname or "",
|
||
|
|
c.code or "", c.email_1 or "", c.email_2 or "", c.phone_1 or "", c.phone_2 or "",
|
||
|
|
c.website or "", c.mailing_city or "", c.mailing_postalcode or "",
|
||
|
|
c.mailing_country or "", c.vat_code or "", c.tags or "",
|
||
|
|
])
|
||
|
|
return output.getvalue()
|
||
|
|
|
||
|
|
|
||
|
|
export_service = ExportService()
|