140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
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,
|
|
}
|