297 lines
15 KiB
Python
297 lines
15 KiB
Python
"""Фаза 1: чтение файлов `seed/` в память и проверка (обязательные поля, enum'ы, ссылки внутри наборов).
|
||
|
||
База данных здесь не используется; любая ошибка — `SeedError` с файлом и записью.
|
||
"""
|
||
|
||
import csv
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import yaml
|
||
|
||
from scripts.seed_lib.model import AccountSeed, PermissionSeed, RegionalOfficeSeed, RoleSeed, SeedData, SeedError
|
||
from src.models.enums import AccountKind, AccountRowType, 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:
|
||
"""Код права `<OBJECT>_<ACTION>`; при совпадении объекта и действия — одно слово (`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
|
||
|
||
|
||
CHART_COLUMNS = ["code", "parent_code", "name", "kind", "row_type", "is_off_balance"]
|
||
#: нормализованный код: NN или NNN (забалансовый), NN.MM, NN.MM.K (K без ведущего нуля: 76.07.1, как в ТЗ; 60.1 — нет)
|
||
_ACCOUNT_CODE = re.compile(r"^\d{2,3}(\.\d{2}(\.[1-9]\d?)?)?$")
|
||
_ROW_TYPE_BY_DEPTH = (AccountRowType.account, AccountRowType.subaccount, AccountRowType.analytic)
|
||
_OFF_BALANCE_ROOTS = {f"{n:03d}" for n in range(1, 12)}
|
||
_OFF_BALANCE_CODE_LENGTH = 3 # забалансовые счета — трёхзначные (001–011), балансовые синтетические — двузначные
|
||
|
||
|
||
def _chart_row(row: dict[str | None, Any], where: str, code: str) -> AccountSeed:
|
||
"""Проверка одной строки плана счетов, не требующая других строк."""
|
||
where = f"{where} ({code})"
|
||
if not _ACCOUNT_CODE.match(code):
|
||
raise SeedError(f"{where}: код не нормализован, ожидается вид 60, 001, 60.01 или 76.07.1")
|
||
root = code.split(".", maxsplit=1)[0]
|
||
off_balance_root = len(root) == _OFF_BALANCE_CODE_LENGTH
|
||
if off_balance_root and root not in _OFF_BALANCE_ROOTS:
|
||
raise SeedError(f"{where}: трёхзначные коды допустимы только для 001–011")
|
||
if root == "00":
|
||
raise SeedError(f"{where}: счёта 00 не существует")
|
||
kind_text = str(row["kind"]).strip()
|
||
try:
|
||
kind = AccountKind(kind_text) if kind_text else None
|
||
except ValueError:
|
||
raise SeedError(f"{where}: неизвестный kind «{kind_text}» (допустимы A, P, AP)") from None
|
||
row_type_text = str(row["row_type"]).strip()
|
||
try:
|
||
row_type = AccountRowType(row_type_text)
|
||
except ValueError:
|
||
raise SeedError(
|
||
f"{where}: неизвестный row_type «{row_type_text}» (допустимы account, subaccount, analytic)"
|
||
) from None
|
||
flag_text = str(row["is_off_balance"]).strip()
|
||
if flag_text not in ("true", "false"):
|
||
raise SeedError(f"{where}: is_off_balance должен быть true или false, получено «{flag_text}»")
|
||
is_off_balance = flag_text == "true"
|
||
expected_type = _ROW_TYPE_BY_DEPTH[code.count(".")]
|
||
if row_type is not expected_type:
|
||
raise SeedError(
|
||
f"{where}: row_type «{row_type.value}» не соответствует глубине кода (ожидался «{expected_type.value}»)"
|
||
)
|
||
if off_balance_root and not is_off_balance:
|
||
raise SeedError(f"{where}: забалансовый счёт должен иметь is_off_balance=true")
|
||
if not off_balance_root and is_off_balance:
|
||
raise SeedError(f"{where}: is_off_balance=true допустим только для счетов 001–011")
|
||
if row_type is AccountRowType.account and kind is None:
|
||
raise SeedError(f"{where}: у синтетического счёта должен быть задан kind")
|
||
name = str(row["name"]).strip()
|
||
if not name:
|
||
raise SeedError(f"{where}: пустое название счёта")
|
||
parent_text = str(row["parent_code"]).strip()
|
||
return AccountSeed(code, parent_text or None, name, kind, row_type, is_off_balance)
|
||
|
||
|
||
def load_chart_of_accounts(core: Path, seed_dir: Path) -> dict[str, AccountSeed]:
|
||
"""План счетов из `chart_of_accounts.csv` (UTF-8): проверка до записи, порядок файла сохраняется."""
|
||
path = core / "chart_of_accounts.csv"
|
||
if not path.is_file():
|
||
raise SeedError(f"{_where(path, seed_dir)}: файл не найден")
|
||
try:
|
||
text = path.read_text(encoding="utf-8")
|
||
except UnicodeDecodeError as error:
|
||
raise SeedError(f"{_where(path, seed_dir)}: не удалось прочитать файл как UTF-8: {error}") from error
|
||
reader = csv.DictReader(text.splitlines())
|
||
if reader.fieldnames != CHART_COLUMNS:
|
||
raise SeedError(f"{_where(path, seed_dir)}: заголовок должен быть «{','.join(CHART_COLUMNS)}»")
|
||
accounts: dict[str, AccountSeed] = {}
|
||
lines: dict[str, int] = {}
|
||
for line_no, row in enumerate(reader, start=2):
|
||
where = f"{_where(path, seed_dir)}, строка {line_no}"
|
||
if None in row or any(value is None for value in row.values()):
|
||
raise SeedError(f"{where}: число полей не совпадает с заголовком")
|
||
code = str(row["code"]).strip()
|
||
account = _chart_row(row, where, code)
|
||
if code in accounts:
|
||
raise SeedError(f"{where} ({code}): код уже встречался в строке {lines[code]}")
|
||
accounts[code] = account
|
||
lines[code] = line_no
|
||
for code, account in accounts.items():
|
||
where = f"{_where(path, seed_dir)}, строка {lines[code]} ({code})"
|
||
expected_parent = code.rsplit(".", 1)[0] if "." in code else None
|
||
if account.parent_code != expected_parent:
|
||
if expected_parent is None:
|
||
raise SeedError(f"{where}: у синтетического счёта parent_code должен быть пустым")
|
||
raise SeedError(f"{where}: родитель должен быть «{expected_parent}», указан «{account.parent_code or ''}»")
|
||
if account.parent_code is not None and account.parent_code not in accounts:
|
||
raise SeedError(f"{where}: родитель «{account.parent_code}» не найден в файле")
|
||
return accounts
|
||
|
||
|
||
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)
|
||
data.accounts = load_chart_of_accounts(core, seed_dir)
|
||
if dev:
|
||
_load_dev(data, seed_dir / "dev")
|
||
return data
|
||
|
||
|
||
def _load_dev(data: SeedData, dev_dir: Path) -> None:
|
||
"""Заглушка расширения: dev-данные (филиалы, пользователи по ролям) появятся в задаче 15."""
|