Agent Zero
b231c2d0d3
Check Cross-Plugin Imports / check (push) Has been cancelled
B-SENS: app/core/sensitive_data.py (NEU) — zentrale Sensitive-Field-Verwaltung - SENSITIVE_FIELDS dict für contact/user/mail_account/system_settings - is_sensitive(), sanitize_dict(), register_sensitive_fields() - Integration: errors.py (Log-Redaction), audit.py (Audit-Masking), export_service.py (Export-Filter), embedding.py (Index-Filter) B-DATA-POL: AI/Data Exposure Policy - DATA_EXPOSURE_POLICY: pro Entity+Field welche Systeme erlaubt (llm_context/search/embeddings/rag/agent_memory/export) - filter_for_llm_context/search/embeddings/export/rag/agent_memory() B-AIPROV-COMP: AIProvider Compliance Metadata - Migration 0119: 7 neue Spalten an ai_providers (region, hosting_type, dpa_status, retention_policy, training_on_customer_data, transfer_notice, allowed_data_classes) - llm_client.py: get_provider_compliance() + check_data_class_allowed() B-PRIV-TEST: 76 Tests in test_sensitive_data.py — alle grün - Sensitive Fields, Exposure Policy, Provider Compliance, Secrets-always-blocked - Keine Regression: 39 LLM-Client Tests grün
80 lines
2.5 KiB
Python
80 lines
2.5 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.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()
|