feat: add deployment config sync commands

backend/workflows/portal_config_sync.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from django.core.serializers.json import DjangoJSONEncoder
+from django.db import transaction
+from django.utils import timezone
+
+from .branding import get_portal_branding, get_portal_company_config
+
+BRANDING_FIELDS = (
+    'name',
+    'portal_title',
+    'company_name',
+    'company_domain',
+    'support_email',
+    'sender_display_name',
+    'login_subtitle',
+    'footer_text',
+    'footer_text_en',
+    'legal_notice',
+    'legal_notice_en',
+    'default_language',
+    'primary_color',
+    'secondary_color',
+)
+
+COMPANY_FIELDS = (
+    'name',
+    'legal_company_name',
+    'street_address',
+    'postal_code',
+    'city',
+    'country',
+    'website_url',
+    'imprint_url',
+    'privacy_url',
+    'hr_contact_email',
+    'it_contact_email',
+    'operations_contact_email',
+    'phone_number',
+    'vat_id',
+    'registration_number',
+)
+
+FORMAT_VERSION = 1
+
+
+def export_portal_deployment_config_payload() -> dict[str, object]:
+    branding = get_portal_branding()
+    company_config = get_portal_company_config()
+    return {
+        'format': 'portal_deployment_config',
+        'version': FORMAT_VERSION,
+        'generated_at': timezone.now(),
+        'branding': {field: getattr(branding, field) for field in BRANDING_FIELDS},
+        'company_config': {field: getattr(company_config, field) for field in COMPANY_FIELDS},
+        'notes': {
+            'file_fields_not_included': ['logo_image', 'pdf_letterhead', 'favicon_image'],
+        },
+    }
+
+
+def write_portal_deployment_config_export(path: str | Path) -> Path:
+    payload = export_portal_deployment_config_payload()
+    output_path = Path(path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(json.dumps(payload, indent=2, cls=DjangoJSONEncoder) + '\n', encoding='utf-8')
+    return output_path
+
+
+def load_portal_deployment_config_payload(path: str | Path) -> dict[str, object]:
+    payload = json.loads(Path(path).read_text(encoding='utf-8'))
+    if payload.get('format') != 'portal_deployment_config':
+        raise ValueError('Unsupported payload format.')
+    if payload.get('version') != FORMAT_VERSION:
+        raise ValueError(f"Unsupported payload version: {payload.get('version')}")
+    return payload
+
+
+def _validate_mapping(payload: dict[str, object], key: str, expected_fields: tuple[str, ...]) -> dict[str, str]:
+    value = payload.get(key)
+    if not isinstance(value, dict):
+        raise ValueError(f'{key} must be an object.')
+    fields = set(value.keys())
+    expected = set(expected_fields)
+    missing = expected.difference(fields)
+    extra = fields.difference(expected)
+    if missing:
+        raise ValueError(f'Missing {key} fields: {", ".join(sorted(missing))}')
+    if extra:
+        raise ValueError(f'Unexpected {key} fields: {", ".join(sorted(extra))}')
+
+    normalized: dict[str, str] = {}
+    for field in expected_fields:
+        field_value = value[field]
+        if field_value is None:
+            field_value = ''
+        if not isinstance(field_value, str):
+            raise ValueError(f'Invalid {key}.{field}: expected string')
+        normalized[field] = field_value
+    return normalized
+
+
+def validate_portal_deployment_config_payload(payload: dict[str, object]) -> dict[str, dict[str, str]]:
+    return {
+        'branding': _validate_mapping(payload, 'branding', BRANDING_FIELDS),
+        'company_config': _validate_mapping(payload, 'company_config', COMPANY_FIELDS),
+    }
+
+
+@transaction.atomic
+def import_portal_deployment_config_payload(payload: dict[str, object], dry_run: bool = False) -> dict[str, int]:
+    validated = validate_portal_deployment_config_payload(payload)
+    branding = get_portal_branding()
+    company_config = get_portal_company_config()
+    updated = 0
+
+    for field, value in validated['branding'].items():
+        if getattr(branding, field) != value:
+            setattr(branding, field, value)
+            updated += 1
+
+    for field, value in validated['company_config'].items():
+        if getattr(company_config, field) != value:
+            setattr(company_config, field, value)
+            updated += 1
+
+    if not dry_run:
+        branding.save(update_fields=list(BRANDING_FIELDS))
+        company_config.save(update_fields=list(COMPANY_FIELDS))
+    else:
+        transaction.set_rollback(True)
+
+    return {
+        'updated_fields': updated,
+        'dry_run': 1 if dry_run else 0,
+    }