178 lines
9.4 KiB
Python
178 lines
9.4 KiB
Python
"""Фаза 1: правила элиминации (§4) и чек-лист обязательных документов — чтение и проверка без БД.
|
||
|
||
Ссылки на счета проверяются по коду плана счетов того же набора (`account_codes`): неизвестный счёт — `SeedError`
|
||
с номером правила / подходом чек-листа до первой записи в БД.
|
||
"""
|
||
|
||
import datetime
|
||
from collections.abc import Collection
|
||
from pathlib import Path
|
||
from typing import Any, TypeVar
|
||
|
||
from scripts.seed_lib.model import EliminationRuleSeed, RequiredDocumentRuleSeed, RuleSideSeed, SeedError
|
||
from scripts.seed_lib.readers import bool_field, checked_records, place, read_yaml, str_field
|
||
from src.models.enums import ConsolidationApproach, DocumentType, StatementType
|
||
|
||
RULES_FILE = "elimination_rules_v1.yaml"
|
||
CHECKLIST_FILE = "required_document_rules.yaml"
|
||
_SIDES = (1, 2)
|
||
_SIDE_1_VALUES = ("D", "K")
|
||
_MEASURES = ("closing", "turnover")
|
||
_MODES = ("min", "full", "manual")
|
||
_RULE_FIELDS_REQUIRED = {
|
||
"rule_no",
|
||
"article_1_line",
|
||
"article_2_line",
|
||
"side_1",
|
||
"measure",
|
||
"elimination_mode",
|
||
"company_1",
|
||
"company_2",
|
||
}
|
||
_RULE_FIELDS = _RULE_FIELDS_REQUIRED | {"vat_handling", "excise_adjust", "comment"}
|
||
|
||
|
||
_E = TypeVar("_E", StatementType, ConsolidationApproach, DocumentType)
|
||
|
||
|
||
def _enum_value(enum_type: type[_E], value: Any, what: str, where: str) -> _E:
|
||
try:
|
||
return enum_type(value)
|
||
except ValueError:
|
||
allowed = ", ".join(member.value for member in enum_type)
|
||
raise SeedError(f"{where}: неизвестное значение {what} «{value}» (допустимы {allowed})") from None
|
||
|
||
|
||
def _choice(record: dict[str, Any], key: str, allowed: tuple[str, ...], where: str) -> str:
|
||
value = str_field(record, key, where)
|
||
if value not in allowed:
|
||
raise SeedError(f"{where}: поле «{key}» должно быть одним из {', '.join(allowed)}, получено «{value}»")
|
||
return str(value)
|
||
|
||
|
||
def _codes(value: Any, key: str, where: str, known: Collection[str]) -> tuple[str, ...]:
|
||
"""Список кодов счетов (строки в кавычках), существующих в плане счетов; без дублей."""
|
||
if value is None:
|
||
return ()
|
||
if not isinstance(value, list):
|
||
raise SeedError(f"{where}: «{key}» — ожидается список кодов счетов")
|
||
result: list[str] = []
|
||
for code in value:
|
||
if not isinstance(code, str):
|
||
raise SeedError(
|
||
f'{where}: код счёта {code!r} в «{key}» — не строка; коды счетов пишутся в кавычках ("01", "60.02")'
|
||
)
|
||
if code not in known:
|
||
raise SeedError(f"{where}: счёт {code} не найден в плане счетов")
|
||
if code in result:
|
||
raise SeedError(f"{where}: счёт {code} указан в «{key}» больше одного раза")
|
||
result.append(code)
|
||
return tuple(result)
|
||
|
||
|
||
def _side(record: dict[str, Any], company_no: int, where: str, known: Collection[str]) -> RuleSideSeed:
|
||
key = f"company_{company_no}"
|
||
side_where = f"{where} (сторона {company_no})"
|
||
body = record[key]
|
||
if not isinstance(body, dict):
|
||
raise SeedError(f"{side_where}: «{key}» — ожидается словарь с полями accounts и excluded")
|
||
for extra in sorted(set(body) - {"accounts", "excluded"}):
|
||
raise SeedError(f"{side_where}: неизвестное поле «{extra}» (допустимы accounts, excluded)")
|
||
accounts = _codes(body.get("accounts"), "accounts", side_where, known)
|
||
excluded = _codes(body.get("excluded"), "excluded", side_where, known)
|
||
if not accounts:
|
||
raise SeedError(f"{side_where}: нет счетов стороны, у правила должны быть счета обеих сторон")
|
||
for code in excluded:
|
||
if code in accounts:
|
||
raise SeedError(f"{side_where}: счёт {code} одновременно включён и исключён")
|
||
return RuleSideSeed(accounts, excluded)
|
||
|
||
|
||
def load_elimination_rules(core: Path, seed_dir: Path, known_accounts: Collection[str]) -> list[EliminationRuleSeed]:
|
||
"""Правила элиминации из `elimination_rules_v1.yaml`: общие поля файла + список правил со счетами сторон."""
|
||
path = core / RULES_FILE
|
||
data = read_yaml(path, seed_dir)
|
||
file_place = place(path, seed_dir)
|
||
if not isinstance(data, dict):
|
||
raise SeedError(f"{file_place}: ожидается словарь (statement_type, version_no, valid_from, rules)")
|
||
for key in sorted(set(data) - {"statement_type", "version_no", "valid_from", "rules"}):
|
||
raise SeedError(f"{file_place}: неизвестное поле «{key}»")
|
||
for key in ("statement_type", "version_no", "valid_from", "rules"):
|
||
if key not in data:
|
||
raise SeedError(f"{file_place}: нет обязательного поля «{key}»")
|
||
statement_type = _enum_value(StatementType, data["statement_type"], "statement_type", file_place)
|
||
version_no = data["version_no"]
|
||
if not isinstance(version_no, int) or isinstance(version_no, bool) or version_no < 1:
|
||
raise SeedError(f"{file_place}: поле «version_no» должно быть целым числом от 1")
|
||
valid_from = data["valid_from"]
|
||
if not isinstance(valid_from, datetime.date) or isinstance(valid_from, datetime.datetime):
|
||
raise SeedError(f"{file_place}: поле «valid_from» должно быть датой ГГГГ-ММ-ДД")
|
||
records = checked_records(data["rules"], _RULE_FIELDS, _RULE_FIELDS_REQUIRED | {"rule_no"}, path, seed_dir)
|
||
rules: list[EliminationRuleSeed] = []
|
||
seen: set[int] = set()
|
||
for index, record in enumerate(records):
|
||
rule_no = record["rule_no"]
|
||
where = f"{place(path, seed_dir, index)}: правило № {rule_no}"
|
||
if not isinstance(rule_no, int) or isinstance(rule_no, bool) or rule_no < 1:
|
||
raise SeedError(f"{place(path, seed_dir, index)}: поле «rule_no» должно быть целым числом от 1")
|
||
if rule_no in seen:
|
||
raise SeedError(f"{file_place}: правило № {rule_no} встречается больше одного раза")
|
||
seen.add(rule_no)
|
||
vat = str_field(record, "vat_handling", where, optional=True)
|
||
rules.append(
|
||
EliminationRuleSeed(
|
||
statement_type=statement_type,
|
||
rule_no=rule_no,
|
||
version_no=version_no,
|
||
article_1_line=str(str_field(record, "article_1_line", where)),
|
||
article_2_line=str(str_field(record, "article_2_line", where)),
|
||
side_1=_choice(record, "side_1", _SIDE_1_VALUES, where),
|
||
measure=_choice(record, "measure", _MEASURES, where),
|
||
elimination_mode=_choice(record, "elimination_mode", _MODES, where),
|
||
vat_handling=vat,
|
||
excise_adjust=bool_field(record, "excise_adjust", where, default=False),
|
||
comment=str_field(record, "comment", where, optional=True),
|
||
valid_from=valid_from,
|
||
sides={no: _side(record, no, where, known_accounts) for no in _SIDES},
|
||
)
|
||
)
|
||
return rules
|
||
|
||
|
||
_CHECKLIST_FIELDS = {"approach", "document_type", "account", "is_mandatory"}
|
||
|
||
|
||
def load_required_document_rules(
|
||
core: Path, seed_dir: Path, known_accounts: Collection[str]
|
||
) -> list[RequiredDocumentRuleSeed]:
|
||
"""Чек-лист из `required_document_rules.yaml`: OSV — со счётом из плана, прочие типы — без счёта; дублей нет."""
|
||
path = core / CHECKLIST_FILE
|
||
records = checked_records(
|
||
read_yaml(path, seed_dir), _CHECKLIST_FIELDS, {"approach", "document_type"}, path, seed_dir
|
||
)
|
||
rules: list[RequiredDocumentRuleSeed] = []
|
||
seen: set[tuple[ConsolidationApproach, DocumentType, str | None]] = set()
|
||
for index, record in enumerate(records):
|
||
where = place(path, seed_dir, index)
|
||
approach = _enum_value(ConsolidationApproach, record["approach"], "approach", where)
|
||
document_type = _enum_value(DocumentType, record["document_type"], "document_type", where)
|
||
where = f"{place(path, seed_dir, index)}: чек-лист ({approach.value})"
|
||
account = record.get("account")
|
||
if document_type is DocumentType.OSV:
|
||
if not isinstance(account, str) or not account.strip():
|
||
raise SeedError(f"{where}: для OSV обязателен account — код счёта строкой в кавычках")
|
||
if account not in known_accounts:
|
||
raise SeedError(f"{where}: счёт {account} не найден в плане счетов")
|
||
elif account is not None:
|
||
raise SeedError(f"{where}: account допустим только для OSV, у {document_type.value} счёта нет")
|
||
key = (approach, document_type, account)
|
||
if key in seen:
|
||
raise SeedError(f"{where}: строка {document_type.value} (счёт {account}) встречается больше одного раза")
|
||
seen.add(key)
|
||
rules.append(
|
||
RequiredDocumentRuleSeed(
|
||
approach, document_type, account, bool_field(record, "is_mandatory", where, default=True)
|
||
)
|
||
)
|
||
return rules
|