Add reusable people list import and export
This commit is contained in:
@@ -0,0 +1,108 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user