67 lines
2.9 KiB
Python
67 lines
2.9 KiB
Python
"""Общие помощники чтения файлов seed: путь в сообщениях, yaml, проверка записей и полей."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from scripts.seed_lib.model import SeedError
|
|
|
|
|
|
def place(path: Path, seed_dir: Path, index: int | None = None, key: str | None = None) -> str:
|
|
text = str(path.relative_to(seed_dir)) if path.is_relative_to(seed_dir) else str(path)
|
|
if index is not None:
|
|
text += f", запись №{index + 1}"
|
|
if key is not None:
|
|
text += f" ({key})"
|
|
return text
|
|
|
|
|
|
def read_yaml(path: Path, seed_dir: Path) -> Any:
|
|
if not path.is_file():
|
|
raise SeedError(f"{place(path, seed_dir)}: файл не найден")
|
|
try:
|
|
return yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except (yaml.YAMLError, UnicodeDecodeError) as error:
|
|
raise SeedError(f"{place(path, seed_dir)}: не удалось разобрать файл: {error}") from error
|
|
|
|
|
|
def checked_records(
|
|
data: Any, allowed: set[str], required: set[str], path: Path, seed_dir: Path
|
|
) -> list[dict[str, Any]]:
|
|
"""Список записей-словарей: тип, обязательные поля, отсутствие лишних ключей."""
|
|
if not isinstance(data, list):
|
|
raise SeedError(f"{place(path, seed_dir)}: ожидается список записей")
|
|
for index, record in enumerate(data):
|
|
if not isinstance(record, dict):
|
|
raise SeedError(f"{place(path, seed_dir, index)}: ожидается словарь полей")
|
|
for key in sorted(set(record) - allowed):
|
|
raise SeedError(f"{place(path, seed_dir, index)}: неизвестное поле «{key}»")
|
|
for key in sorted(required - set(record)):
|
|
raise SeedError(f"{place(path, seed_dir, index)}: нет обязательного поля «{key}»")
|
|
return data
|
|
|
|
|
|
def str_field(record: dict[str, Any], key: str, where: str, *, optional: bool = False) -> str | None:
|
|
value = record.get(key)
|
|
if value is None and optional:
|
|
return None
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise SeedError(f"{where}: поле «{key}» должно быть непустой строкой")
|
|
return value.strip()
|
|
|
|
|
|
def bool_field(record: dict[str, Any], key: str, where: str, *, default: bool) -> bool:
|
|
value = record.get(key, default)
|
|
if not isinstance(value, bool):
|
|
raise SeedError(f"{where}: поле «{key}» должно быть true или false, получено {value!r}")
|
|
return value
|
|
|
|
|
|
def ensure_unique(values: list[str], what: str, path: Path, seed_dir: Path) -> None:
|
|
seen: set[str] = set()
|
|
for value in values:
|
|
if value in seen:
|
|
raise SeedError(f"{place(path, seed_dir)}: {what} «{value}» встречается больше одного раза")
|
|
seen.add(value)
|