"""Фаза 1: чтение файлов `seed/` в память и проверка (обязательные поля, enum'ы, ссылки внутри наборов). База данных здесь не используется; любая ошибка — `SeedError` с файлом и записью. """ from pathlib import Path from typing import Any import yaml from scripts.seed_lib.model import PermissionSeed, RegionalOfficeSeed, RoleSeed, SeedData, SeedError from src.models.enums import PermissionAction, PermissionObject DEFAULT_SEED_DIR = Path(__file__).resolve().parent.parent.parent / "seed" def _where(path: Path, seed_dir: Path, index: int | None = None, key: str | None = None) -> str: place = str(path.relative_to(seed_dir)) if path.is_relative_to(seed_dir) else str(path) if index is not None: place += f", запись №{index + 1}" if key is not None: place += f" ({key})" return place def _read_yaml(path: Path, seed_dir: Path) -> Any: if not path.is_file(): raise SeedError(f"{_where(path, seed_dir)}: файл не найден") try: return yaml.safe_load(path.read_text(encoding="utf-8")) except (yaml.YAMLError, UnicodeDecodeError) as error: raise SeedError(f"{_where(path, seed_dir)}: не удалось разобрать файл: {error}") from error def _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"{_where(path, seed_dir)}: ожидается список записей") for index, record in enumerate(data): if not isinstance(record, dict): raise SeedError(f"{_where(path, seed_dir, index)}: ожидается словарь полей") for key in sorted(set(record) - allowed): raise SeedError(f"{_where(path, seed_dir, index)}: неизвестное поле «{key}»") for key in sorted(required - set(record)): raise SeedError(f"{_where(path, seed_dir, index)}: нет обязательного поля «{key}»") return data def _str(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(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 _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"{_where(path, seed_dir)}: {what} «{value}» встречается больше одного раза") seen.add(value) def _load_regional_offices(core: Path, seed_dir: Path) -> list[RegionalOfficeSeed]: path = core / "regional_office.yaml" records = _records(_read_yaml(path, seed_dir), {"name", "is_head_office", "is_active"}, {"name"}, path, seed_dir) result = [] for index, record in enumerate(records): where = _where(path, seed_dir, index) result.append( RegionalOfficeSeed( name=str(_str(record, "name", where)), is_head_office=_bool(record, "is_head_office", where, default=False), is_active=_bool(record, "is_active", where, default=True), ) ) _unique([o.name for o in result], "филиал", path, seed_dir) return result def _load_roles(core: Path, seed_dir: Path) -> list[RoleSeed]: path = core / "roles.yaml" records = _records( _read_yaml(path, seed_dir), {"code", "name", "description", "is_system", "is_active"}, {"code", "name"}, path, seed_dir, ) result = [] for index, record in enumerate(records): where = _where(path, seed_dir, index, str(record.get("code"))) result.append( RoleSeed( code=str(_str(record, "code", where)), name=str(_str(record, "name", where)), description=_str(record, "description", where, optional=True), is_system=_bool(record, "is_system", where, default=False), is_active=_bool(record, "is_active", where, default=True), ) ) _unique([r.code for r in result], "роль", path, seed_dir) return result def _expected_permission_code(object_type: PermissionObject, action: PermissionAction) -> str: """Код права `_`; при совпадении объекта и действия — одно слово (`EXPORT`).""" if object_type.value == action.value: return action.value return f"{object_type.value}_{action.value}" def _load_permissions(core: Path, seed_dir: Path) -> list[PermissionSeed]: path = core / "permissions.yaml" records = _records( _read_yaml(path, seed_dir), {"code", "object", "action", "name", "description"}, {"code", "object", "action", "name"}, path, seed_dir, ) result = [] for index, record in enumerate(records): where = _where(path, seed_dir, index, str(record.get("code"))) try: object_type = PermissionObject(record["object"]) except ValueError: raise SeedError(f"{where}: неизвестный объект права «{record['object']}»") from None try: action = PermissionAction(record["action"]) except ValueError: raise SeedError(f"{where}: неизвестное действие права «{record['action']}»") from None code = str(_str(record, "code", where)) if code != _expected_permission_code(object_type, action): raise SeedError( f"{where}: код права «{code}» не соответствует объекту и действию, " f"ожидался «{_expected_permission_code(object_type, action)}»" ) result.append( PermissionSeed( code=code, object_type=object_type, action=action, name=str(_str(record, "name", where)), description=_str(record, "description", where, optional=True), ) ) _unique([p.code for p in result], "право", path, seed_dir) pairs = [f"{p.object_type.value}.{p.action.value}" for p in result] _unique(pairs, "сочетание объект.действие", path, seed_dir) return result def _load_role_permissions( core: Path, seed_dir: Path, role_codes: set[str], permission_codes: set[str] ) -> dict[str, list[str]]: path = core / "role_permissions.yaml" data = _read_yaml(path, seed_dir) if not isinstance(data, dict): raise SeedError(f"{_where(path, seed_dir)}: ожидается словарь «код роли -> список кодов прав»") result: dict[str, list[str]] = {} for role_code, codes in data.items(): where = _where(path, seed_dir, key=str(role_code)) if role_code not in role_codes: raise SeedError(f"{where}: роль «{role_code}» не описана в roles.yaml") if not isinstance(codes, list) or not all(isinstance(c, str) for c in codes): raise SeedError(f"{where}: ожидается список кодов прав") for code in codes: if code not in permission_codes: raise SeedError(f"{where}: право «{code}» не описано в permissions.yaml") duplicates = sorted({c for c in codes if codes.count(c) > 1}) if duplicates: raise SeedError(f"{where}: право «{duplicates[0]}» указано больше одного раза") result[role_code] = list(codes) return result def load_seed(seed_dir: Path | None = None, *, dev: bool = False) -> SeedData: """Фаза 1: прочитать и проверить все файлы. Бросает `SeedError`; к БД не обращается.""" seed_dir = seed_dir or DEFAULT_SEED_DIR core = seed_dir / "core" data = SeedData() data.roles = _load_roles(core, seed_dir) data.permissions = _load_permissions(core, seed_dir) data.role_permissions = _load_role_permissions( core, seed_dir, {r.code for r in data.roles}, {p.code for p in data.permissions} ) data.regional_offices = _load_regional_offices(core, seed_dir) if dev: _load_dev(data, seed_dir / "dev") return data def _load_dev(data: SeedData, dev_dir: Path) -> None: """Заглушка расширения: dev-данные (филиалы, пользователи по ролям) появятся в задаче 15."""