fix(security): 16 mittlere Probleme behoben (P18-P33)

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)
2026-08-06 13:23:58 +02:00
parent 9f79107fa7
commit 0eb6d7621e
32 changed files with 593 additions and 307 deletions
@@ -0,0 +1,69 @@
+"""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()