109 lines
4.0 KiB
Python
109 lines
4.0 KiB
Python
"""Versioned YAML import and export for reusable People lists."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Sequence
|
|
|
|
import yaml
|
|
|
|
from mka.application.meeting_service import ParticipantInput
|
|
|
|
PEOPLE_YAML_VERSION = 1
|
|
ATTENDANCE_VALUES = frozenset({"present", "mentioned_only"})
|
|
PARTICIPANT_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]*$")
|
|
|
|
|
|
class PeopleYamlError(ValueError):
|
|
"""Raised when a reusable People-list document is invalid."""
|
|
|
|
|
|
def export_people_yaml(people: Sequence[ParticipantInput]) -> str:
|
|
"""Serialize people deterministically without meeting-specific data."""
|
|
document = {
|
|
"version": PEOPLE_YAML_VERSION,
|
|
"people": [
|
|
{
|
|
"participant_id": person.participant_id,
|
|
"display_name": person.display_name,
|
|
"role": person.role,
|
|
"organization": person.organization,
|
|
"attendance_status": person.attendance_status,
|
|
}
|
|
for person in people
|
|
],
|
|
}
|
|
return yaml.safe_dump(
|
|
document,
|
|
allow_unicode=True,
|
|
sort_keys=False,
|
|
default_flow_style=False,
|
|
)
|
|
|
|
|
|
def import_people_yaml(content: str | bytes) -> list[ParticipantInput]:
|
|
"""Parse and validate a complete replacement People list."""
|
|
try:
|
|
if isinstance(content, bytes):
|
|
content = content.decode("utf-8")
|
|
document = yaml.safe_load(content)
|
|
except (yaml.YAMLError, UnicodeDecodeError) as exc:
|
|
raise PeopleYamlError(f"Malformed People YAML: {exc}") from exc
|
|
|
|
if not isinstance(document, dict):
|
|
raise PeopleYamlError("People YAML must contain a top-level mapping.")
|
|
version = document.get("version")
|
|
if type(version) is not int or version != PEOPLE_YAML_VERSION:
|
|
raise PeopleYamlError(
|
|
f"Unsupported People YAML version {version!r}; expected {PEOPLE_YAML_VERSION}."
|
|
)
|
|
entries = document.get("people")
|
|
if not isinstance(entries, list):
|
|
raise PeopleYamlError("People YAML must contain a top-level 'people' list.")
|
|
|
|
people: list[ParticipantInput] = []
|
|
seen_ids: set[str] = set()
|
|
for index, entry in enumerate(entries, start=1):
|
|
if not isinstance(entry, dict):
|
|
raise PeopleYamlError(f"Person {index} must be a mapping.")
|
|
participant_id = _required_text(entry, "participant_id", index)
|
|
if PARTICIPANT_ID_PATTERN.fullmatch(participant_id) is None:
|
|
raise PeopleYamlError(f"Person {index} has invalid participant_id {participant_id!r}.")
|
|
if participant_id in seen_ids:
|
|
raise PeopleYamlError(f"Duplicate participant_id: {participant_id!r}.")
|
|
seen_ids.add(participant_id)
|
|
|
|
display_name = _required_text(entry, "display_name", index)
|
|
role = _optional_text(entry, "role", index)
|
|
organization = _optional_text(entry, "organization", index)
|
|
attendance_status = entry.get("attendance_status")
|
|
if attendance_status not in ATTENDANCE_VALUES:
|
|
allowed = ", ".join(sorted(ATTENDANCE_VALUES))
|
|
raise PeopleYamlError(
|
|
f"Person {index} has invalid attendance_status; expected one of: {allowed}."
|
|
)
|
|
people.append(
|
|
ParticipantInput(
|
|
participant_id=participant_id,
|
|
display_name=display_name,
|
|
role=role,
|
|
organization=organization,
|
|
attendance_status=attendance_status,
|
|
)
|
|
)
|
|
return people
|
|
|
|
|
|
def _required_text(entry: dict[object, object], field: str, index: int) -> str:
|
|
value = entry.get(field)
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise PeopleYamlError(f"Person {index} requires a non-empty {field}.")
|
|
return value
|
|
|
|
|
|
def _optional_text(entry: dict[object, object], field: str, index: int) -> str:
|
|
value = entry.get(field, "")
|
|
if not isinstance(value, str):
|
|
raise PeopleYamlError(f"Person {index} field {field} must be text.")
|
|
return value
|