"""Общие помощники чтения файлов 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)