163 lines
5.3 KiB
Python
163 lines
5.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 .app_registry import APP_DEFINITIONS, ensure_portal_app_configs, normalize_portal_app_sort_orders
|
|
from .models import PortalAppConfig
|
|
|
|
|
|
EXPORT_FIELDS = (
|
|
'key',
|
|
'section',
|
|
'sort_order',
|
|
'is_enabled',
|
|
'visible_to_super_admin',
|
|
'visible_to_admin',
|
|
'visible_to_it_staff',
|
|
'visible_to_staff',
|
|
'title_override',
|
|
'title_override_en',
|
|
'description_override',
|
|
'description_override_en',
|
|
'action_label_override',
|
|
'action_label_override_en',
|
|
)
|
|
|
|
FORMAT_VERSION = 1
|
|
|
|
|
|
def _valid_keys() -> set[str]:
|
|
return {definition.key for definition in APP_DEFINITIONS}
|
|
|
|
|
|
def _valid_sections() -> set[str]:
|
|
return {key for key, _label in PortalAppConfig.SECTION_CHOICES}
|
|
|
|
|
|
def export_portal_app_config_payload() -> dict[str, object]:
|
|
ensure_portal_app_configs()
|
|
rows = list(
|
|
PortalAppConfig.objects.order_by('section', 'sort_order', 'key').values(*EXPORT_FIELDS)
|
|
)
|
|
return {
|
|
'format': 'portal_app_config',
|
|
'version': FORMAT_VERSION,
|
|
'generated_at': timezone.now(),
|
|
'items': rows,
|
|
}
|
|
|
|
|
|
def write_portal_app_config_export(path: str | Path) -> Path:
|
|
payload = export_portal_app_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_app_config_payload(path: str | Path) -> dict[str, object]:
|
|
payload = json.loads(Path(path).read_text(encoding='utf-8'))
|
|
if payload.get('format') != 'portal_app_config':
|
|
raise ValueError('Unsupported payload format.')
|
|
if payload.get('version') != FORMAT_VERSION:
|
|
raise ValueError(f"Unsupported payload version: {payload.get('version')}")
|
|
items = payload.get('items')
|
|
if not isinstance(items, list):
|
|
raise ValueError('Payload items must be a list.')
|
|
return payload
|
|
|
|
|
|
def validate_portal_app_config_payload(payload: dict[str, object]) -> list[dict[str, object]]:
|
|
valid_keys = _valid_keys()
|
|
valid_sections = _valid_sections()
|
|
items = payload['items']
|
|
seen_keys: set[str] = set()
|
|
normalized_items: list[dict[str, object]] = []
|
|
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
raise ValueError('Each item must be an object.')
|
|
key = item.get('key')
|
|
if not isinstance(key, str) or not key.strip():
|
|
raise ValueError('Each item must contain a non-empty key.')
|
|
if key not in valid_keys:
|
|
raise ValueError(f'Unknown app key: {key}')
|
|
if key in seen_keys:
|
|
raise ValueError(f'Duplicate app key: {key}')
|
|
seen_keys.add(key)
|
|
|
|
section = item.get('section')
|
|
if section not in valid_sections:
|
|
raise ValueError(f'Invalid section for {key}: {section}')
|
|
|
|
sort_order = item.get('sort_order')
|
|
if not isinstance(sort_order, int) or sort_order < 0:
|
|
raise ValueError(f'Invalid sort_order for {key}: {sort_order}')
|
|
|
|
normalized: dict[str, object] = {'key': key, 'section': section, 'sort_order': sort_order}
|
|
for field in EXPORT_FIELDS:
|
|
if field in {'key', 'section', 'sort_order'}:
|
|
continue
|
|
value = item.get(field, False if field.startswith('visible_to_') or field == 'is_enabled' else '')
|
|
if field == 'is_enabled' or field.startswith('visible_to_'):
|
|
if not isinstance(value, bool):
|
|
raise ValueError(f'Invalid boolean for {key}: {field}')
|
|
else:
|
|
if value is None:
|
|
value = ''
|
|
if not isinstance(value, str):
|
|
raise ValueError(f'Invalid string for {key}: {field}')
|
|
normalized[field] = value
|
|
normalized_items.append(normalized)
|
|
|
|
missing = valid_keys.difference(seen_keys)
|
|
if missing:
|
|
raise ValueError(f'Missing app keys: {", ".join(sorted(missing))}')
|
|
return normalized_items
|
|
|
|
|
|
@transaction.atomic
|
|
def import_portal_app_config_payload(payload: dict[str, object], dry_run: bool = False) -> dict[str, int]:
|
|
ensure_portal_app_configs()
|
|
items = validate_portal_app_config_payload(payload)
|
|
updated = 0
|
|
created = 0
|
|
|
|
existing_map = {config.key: config for config in PortalAppConfig.objects.all()}
|
|
for item in items:
|
|
key = str(item['key'])
|
|
config = existing_map.get(key)
|
|
if config is None:
|
|
config = PortalAppConfig(key=key)
|
|
created += 1
|
|
|
|
changed = False
|
|
for field in EXPORT_FIELDS:
|
|
value = item[field]
|
|
if getattr(config, field) != value:
|
|
setattr(config, field, value)
|
|
changed = True
|
|
if config.pk is None:
|
|
changed = True
|
|
if changed:
|
|
updated += 1
|
|
if not dry_run:
|
|
config.save()
|
|
|
|
if not dry_run:
|
|
normalize_portal_app_sort_orders()
|
|
else:
|
|
transaction.set_rollback(True)
|
|
|
|
return {
|
|
'created': created,
|
|
'updated': updated,
|
|
'total': len(items),
|
|
'dry_run': 1 if dry_run else 0,
|
|
}
|