Agent Zero
0eb6d7621e
Check Cross-Plugin Imports / check (push) Has been cancelled
P18: require_permission zu forgejo_error_reporter und ai_ui_control routes hinzugefügt P19: Cross-Tenant Permission-Cache-Invalidierung bei Rollenänderungen P20: Session/Permission-Cache-Invalidierung bei Gruppen-Änderungen P21: ENTITY_MODELS Registry um fehlende Plugin-Modelle erweitert P22: Entity-Links prüfen verknüpfte Entity-Permissions P23: authStore persist Middleware entfernt (kein localStorage mehr) P24: 5xx Retry nur noch für GET-Requests P25: KI-Kommentar in address.py (bekannte Inkonsistenz) P26: DeletionLog in EntityHistory gemerged (action=delete) P27: KI-Kommentar in entity_policy.py (ABAC nicht aktiv genutzt) P28: db.commit() aus bulk_permission_service entfernt P29: CSV-Export in export_service.py ausgelagert P30: plugins.py Business-Logik in plugin_install_service.py ausgelagert P31: KI-Kommentar in session.py (Dual-System dokumentiert) P32: Migration 0115: crm_platform_admin Role droppen P33: Cross-Plugin Imports über contracts.py behoben (10 Violations → 0)
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()
|