d07eedfafb
The АдрМестаПриемаВыдача block under СведенияКартаИнБанк (code 1009) has no single АдресСтрока tag, unlike the same-named block under СведенияВнесениеПолучениеНалДС — column 99 was mapped directly onto that nonexistent tag and stayed empty. It now assembles the value from the individual address components (Индекс, КодОКСМ, КодСубъектаПоОКАТО, Район, Пункт, Улица, Дом, Корп, Оф), the same way column 153 already does for registration place. Also disabled allow_short_lookup for this column, since it was previously able to pick up a same-named tag (e.g. Улица) from an unrelated address elsewhere in the file. Added regression tests at both the mapping level and a full parse_xml_content -> build_fixed_row_by_index pipeline level with a synthetic code-1009 XML.
1177 lines
44 KiB
Python
1177 lines
44 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
|
||
from .column_constants import (
|
||
ACCOUNT_PLACEHOLDER,
|
||
BLOCK_CP,
|
||
BLOCK_EIO,
|
||
CASHLESS_CLEARED_COLUMNS,
|
||
CASHLESS_TRANSFER_TYPES,
|
||
COL_ADDRESS_ONE_LINE,
|
||
COL_ADDRESS_STRUCTURED,
|
||
COL_AMOUNT_CURRENCY,
|
||
COL_AMOUNT_RUB,
|
||
COL_BIRTH_REG_DATE,
|
||
COL_BRANCH_SIGN,
|
||
COL_CARD_HOLDER_INFO,
|
||
COL_CLIENT_SIGN,
|
||
COL_COUNTRY_CODE,
|
||
COL_CP_ISSUER_INN,
|
||
COL_CURRENCY,
|
||
COL_EIO_ADDRESS_ONE_LINE,
|
||
COL_EIO_INN,
|
||
COL_EIO_OTHER_INFO,
|
||
COL_EIO_REG_PLACE_ONE_LINE,
|
||
COL_EIO_REG_PLACE_STRUCTURED,
|
||
COL_EMPLOYEE_SIGN,
|
||
COL_ESP_SIGN,
|
||
COL_EXTRA_CODES,
|
||
COL_EXTRA_INFO,
|
||
COL_FILE_NAME,
|
||
COL_FOREIGN_BANK_NAME,
|
||
COL_FOREIGN_CARD_ADDRESS_STRUCTURED,
|
||
COL_FTR_SIGN,
|
||
COL_IDENT_FL,
|
||
COL_INN,
|
||
COL_INSURANCE_NUMBER,
|
||
COL_ITEM_TYPE,
|
||
COL_KPP,
|
||
COL_METAL_CODE,
|
||
COL_METAL_NAME,
|
||
COL_OPERATION_CODE,
|
||
COL_OPERATION_DOCUMENTS,
|
||
COL_OPERATION_SIGN,
|
||
COL_OPERATOR_TYPE,
|
||
COL_PARTICIPANT_TYPE,
|
||
COL_PAYER_ACCOUNT,
|
||
COL_PAYER_BANK_ACCOUNT,
|
||
COL_PAYER_BANK_BIK,
|
||
COL_PAYER_BANK_NAME,
|
||
COL_PAYER_ESP,
|
||
COL_RECIPIENT_ACCOUNT,
|
||
COL_RECIPIENT_BANK_ACCOUNT,
|
||
COL_RECIPIENT_BANK_BIK,
|
||
COL_RECIPIENT_BANK_NAME,
|
||
COL_RECIPIENT_ESP,
|
||
COL_RECORD_ID,
|
||
COL_REG_PLACE_ONE_LINE,
|
||
COL_REG_PLACE_STRUCTURED,
|
||
COL_RESIDENT_SIGN,
|
||
COL_SALE_AMOUNT,
|
||
COL_SALE_CURRENCY,
|
||
COL_STAY_DOCUMENT_TYPE,
|
||
COL_SUSPENSION_BASIS,
|
||
COL_SUSPICIOUS_ID,
|
||
COL_TERRITORY_CODE,
|
||
COL_TRANSFER_STATUS,
|
||
COL_TRANSFER_TYPE,
|
||
COL_UNUSUAL_CODES,
|
||
EMPLOYEE_ABSENT_CODE,
|
||
EXPORT_SUBSIDIARY_COMMENT_CODES,
|
||
OKATO_COUNTRY_PAIRS,
|
||
PARTICIPANT_ADDRESS_COMPONENT_COLUMNS,
|
||
PARTICIPANT_TYPE_FL,
|
||
PARTICIPANT_TYPE_FLCHP,
|
||
PARTICIPANT_TYPE_IP,
|
||
PARTICIPANT_TYPE_UL,
|
||
PRECIOUS_ITEM_CODES,
|
||
PRECIOUS_METAL_CODE,
|
||
RUSSIA_COUNTRY_CODE,
|
||
SUSPICIOUS_CODE,
|
||
TERRITORY_CODE,
|
||
VALID_ESP_SIGNS,
|
||
VALID_FTR_SIGNS,
|
||
VALID_SUSPENSION_BASES,
|
||
)
|
||
from .column_registry import COLUMNS, ColumnMeta
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _CompiledCandidate:
|
||
raw: str
|
||
normalized: str
|
||
short: str
|
||
strict_scoped: bool
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MappingRule:
|
||
column_index: int
|
||
column_id: str
|
||
column_name: str
|
||
compiled_candidates: tuple[_CompiledCandidate, ...]
|
||
source: str = "any"
|
||
allow_direct_mapping: bool = True
|
||
allow_short_lookup: bool = True
|
||
structured_value: bool = False
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class _LookupEntry:
|
||
raw_path: str
|
||
normalized_path: str
|
||
value: str
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class _PayloadLookup:
|
||
values: dict[str, str]
|
||
entries: tuple[_LookupEntry, ...]
|
||
normalized_paths: tuple[str, ...]
|
||
normalized_exact_first: dict[str, str]
|
||
normalized_suffix_first: dict[str, str]
|
||
raw_short_first: dict[str, str]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StructuredGroupRule:
|
||
structured_column: int
|
||
one_line_columns: tuple[int, ...]
|
||
component_columns: tuple[int, ...]
|
||
join_with_space: bool = False
|
||
preserve_empty_positions: bool = False
|
||
|
||
|
||
_INVALID_TAGS = {"", "-", "Источник", "путь", "подразумеваются", "тэга", "нашла"}
|
||
_INDEXED_PATH_SEGMENT = re.compile(r"\[\d+\]")
|
||
_OPERATION_DOCUMENT_FIELDS = (
|
||
"КодДок",
|
||
"ИноеНаимДок",
|
||
"ДатаДок",
|
||
"НомДок",
|
||
"СодДок",
|
||
)
|
||
_OPERATION_DOCUMENT_FIELD = re.compile(
|
||
r"(?:^|\.)ОснованиеОп(?:\[(?P<index>\d+)\])?\."
|
||
r"(?P<field>КодДок|ИноеНаимДок|ДатаДок|НомДок|СодДок)$"
|
||
)
|
||
_EIO_PERSON_BASE_PATHS = ("УчастникЮЛ.СведЕИО.ФЛЕИО",)
|
||
_EIO_OTHER_INFO_FIELDS = (
|
||
("СНИЛС", "СНИЛСФЛ"),
|
||
("ОМС", "ПолисНомер"),
|
||
("Номер телефона", "ТелефонНомер"),
|
||
("Дата рождения", "ДатаРождения"),
|
||
("Тип ФЛ, занимающегося частной практикой", "ТипФЛЧастнаяПрактика"),
|
||
("Регистрационный номер", "РегнНомер"),
|
||
)
|
||
_EIO_BLOCK_NAMES = (
|
||
"СведЕИО",
|
||
"БенефициарЮЛ",
|
||
"БенефициарФЛИП",
|
||
"БенефициарИНБОЮЛ",
|
||
)
|
||
_PARTICIPANT_ADDRESS_BASE_PATHS = (
|
||
"УчастникЮЛ.СведЮЛ.АдрРегЮЛ",
|
||
"УчастникФЛИП.СведФЛИП.АдрРег",
|
||
"УчастникИНБОЮЛ.СведИНБОЮЛ.МестоДеятельностьИНБОЮЛ",
|
||
)
|
||
_ONE_LINE_STRUCTURED_PAIRS = (
|
||
(COL_REG_PLACE_ONE_LINE, COL_REG_PLACE_STRUCTURED),
|
||
(COL_EIO_REG_PLACE_ONE_LINE, COL_EIO_REG_PLACE_STRUCTURED),
|
||
)
|
||
|
||
|
||
def _bank_bik_candidate_keys(bank_block: str) -> tuple[str, ...]:
|
||
candidates: list[str] = []
|
||
for transfers_block in ("СведенияПереводыДС", "СвединияПереводыДС"):
|
||
path = (
|
||
"/СообщОперКО/ИнформЧасть/СведКО/Операция/"
|
||
f"{transfers_block}/{bank_block}/БИККО"
|
||
)
|
||
for candidate in _extract_path_candidate_keys(path):
|
||
if (
|
||
candidate.endswith(f"{bank_block}.БИККО")
|
||
and candidate not in candidates
|
||
):
|
||
candidates.append(candidate)
|
||
return tuple(candidates)
|
||
|
||
|
||
def _compile_candidates(candidates: tuple[str, ...]) -> tuple[_CompiledCandidate, ...]:
|
||
normalized_candidates = tuple(
|
||
_INDEXED_PATH_SEGMENT.sub("", candidate) for candidate in candidates
|
||
)
|
||
return tuple(
|
||
_CompiledCandidate(
|
||
raw=candidate,
|
||
normalized=normalized,
|
||
short=candidate.rsplit(".", maxsplit=1)[-1],
|
||
strict_scoped="." in normalized
|
||
and not any(
|
||
other != normalized and other.endswith(f".{normalized}")
|
||
for other in normalized_candidates
|
||
),
|
||
)
|
||
for candidate, normalized in zip(candidates, normalized_candidates)
|
||
)
|
||
|
||
|
||
def _build_rules(mapping_data: tuple[ColumnMeta, ...]) -> tuple[MappingRule, ...]:
|
||
rules: list[MappingRule] = []
|
||
for meta in mapping_data:
|
||
index = meta.index
|
||
column = meta.name
|
||
tag = meta.xml_tag
|
||
path = meta.xml_path
|
||
if not column or tag in _INVALID_TAGS:
|
||
continue
|
||
candidates = list(_extract_tag_candidate_keys(tag))
|
||
if index == COL_INN:
|
||
candidates = ["ИННФЛИП", "ИННФЛ", "ИННЮЛ", "ИНН", *candidates]
|
||
if column == "Тип участника":
|
||
candidates = ["Тип", "ТипУчастника", *candidates]
|
||
if column == "Отчество":
|
||
# В некоторых выгрузках встречается опечатка тега "Очт".
|
||
candidates = ["Отч", "Очт", *candidates]
|
||
path_candidates = list(_extract_path_candidate_keys(path)) if path else []
|
||
if path:
|
||
for path_tag in path_candidates:
|
||
if path_tag not in candidates:
|
||
candidates.append(path_tag)
|
||
if index == COL_PAYER_BANK_BIK:
|
||
candidates = list(_bank_bik_candidate_keys("СведБанкПлательщик"))
|
||
elif index == COL_RECIPIENT_BANK_BIK:
|
||
candidates = list(_bank_bik_candidate_keys("СведБанкПолучатель"))
|
||
elif index == COL_COUNTRY_CODE:
|
||
candidates = [
|
||
candidate
|
||
for candidate in [*path_candidates, *candidates]
|
||
if candidate.endswith("СведФЛИП.КодОКСМ")
|
||
]
|
||
elif index == COL_IDENT_FL:
|
||
candidates = [
|
||
candidate
|
||
for candidate in [*path_candidates, *candidates]
|
||
if candidate.endswith("ИдентификацияФЛ")
|
||
]
|
||
elif index == COL_BIRTH_REG_DATE:
|
||
candidates = [
|
||
candidate
|
||
for candidate in [*path_candidates, *candidates]
|
||
if candidate.endswith(("ДатаРегЮЛ", "ДатаРождения"))
|
||
]
|
||
elif index == COL_STAY_DOCUMENT_TYPE:
|
||
candidates = [
|
||
candidate
|
||
for candidate in path_candidates
|
||
if candidate.endswith("УчастникФЛИП.СведФЛИП.СведДокПреб.ВидДокКод")
|
||
]
|
||
elif (
|
||
index == COL_ADDRESS_STRUCTURED
|
||
or index in PARTICIPANT_ADDRESS_COMPONENT_COLUMNS
|
||
):
|
||
candidates = [
|
||
f"{base_path}.{address_tag}"
|
||
for base_path in _PARTICIPANT_ADDRESS_BASE_PATHS
|
||
for address_tag in _extract_tag_candidate_keys(tag)
|
||
]
|
||
elif index == COL_REG_PLACE_ONE_LINE:
|
||
candidates = [
|
||
"УчастникЮЛ.СведЮЛ.АдрРегЮЛ.АдресСтрока",
|
||
"СообщОперКО.ИнформЧасть.СведКО.Операция.УчастникОП.УчастникЮЛ.СведЮЛ.АдрРегЮЛ.АдресСтрока",
|
||
]
|
||
elif index == COL_REG_PLACE_STRUCTURED:
|
||
base_paths = (
|
||
"УчастникЮЛ.СведЮЛ.АдрРегЮЛ",
|
||
"СообщОперКО.ИнформЧасть.СведКО.Операция.УчастникОП.УчастникЮЛ.СведЮЛ.АдрРегЮЛ",
|
||
)
|
||
address_tags = (
|
||
"Индекс",
|
||
"КодОКСМ",
|
||
"КодСубъектаПоОКАТО",
|
||
"Район",
|
||
"Пункт",
|
||
"Улица",
|
||
"Дом",
|
||
"Корп",
|
||
"Оф",
|
||
)
|
||
candidates = [
|
||
f"{base_path}.{tag}" for base_path in base_paths for tag in address_tags
|
||
]
|
||
elif index == COL_FOREIGN_CARD_ADDRESS_STRUCTURED:
|
||
# legacy_099: адрес места приёма/выдачи по карте иностранного
|
||
# банка (код 1009) приходит в XML разложенным по тем же тегам
|
||
# АдрМестаПриемаВыдача, что и другие адресные блоки (см.
|
||
# COL_REG_PLACE_STRUCTURED выше) — АдресСтрока как единого поля
|
||
# для этого блока в схеме нет, поэтому колонку нужно собирать
|
||
# из компонентов, а не мапить напрямую на несуществующий тег.
|
||
base_paths = (
|
||
"СведенияКартаИнБанк.СведМестоОперация.АдрМестаПриемаВыдача",
|
||
"СообщОперКО.ИнформЧасть.СведКО.Операция.СведенияКартаИнБанк."
|
||
"СведМестоОперация.АдрМестаПриемаВыдача",
|
||
)
|
||
address_tags = (
|
||
"Индекс",
|
||
"КодОКСМ",
|
||
"КодСубъектаПоОКАТО",
|
||
"Район",
|
||
"Пункт",
|
||
"Улица",
|
||
"Дом",
|
||
"Корп",
|
||
"Оф",
|
||
)
|
||
candidates = [
|
||
f"{base_path}.{tag}" for base_path in base_paths for tag in address_tags
|
||
]
|
||
elif index == COL_CP_ISSUER_INN:
|
||
candidates = [
|
||
candidate
|
||
for candidate in [*path_candidates, *candidates]
|
||
if candidate.endswith("ИННЭмитентЦП")
|
||
]
|
||
if not candidates:
|
||
candidates = list(_extract_tag_candidate_keys(tag))
|
||
non_one_line_candidates = [
|
||
candidate
|
||
for candidate in candidates
|
||
if not _is_one_line_candidate(candidate)
|
||
]
|
||
direct_candidates = candidates
|
||
if meta.structured_value and non_one_line_candidates:
|
||
direct_candidates = non_one_line_candidates
|
||
candidate_keys = tuple(direct_candidates)
|
||
rules.append(
|
||
MappingRule(
|
||
column_index=index,
|
||
column_id=meta.column_id,
|
||
column_name=column,
|
||
compiled_candidates=_compile_candidates(candidate_keys),
|
||
source=meta.source_scope,
|
||
allow_direct_mapping=meta.allow_direct_mapping,
|
||
allow_short_lookup=meta.allow_short_lookup,
|
||
structured_value=meta.structured_value,
|
||
)
|
||
)
|
||
return tuple(rules)
|
||
|
||
|
||
def _is_one_line_candidate(candidate: str) -> bool:
|
||
return "строка" in candidate.lower()
|
||
|
||
|
||
def _build_structured_group_rules(
|
||
mapping_data: tuple[ColumnMeta, ...],
|
||
) -> tuple[StructuredGroupRule, ...]:
|
||
grouped: dict[str, list[ColumnMeta]] = {}
|
||
for meta in mapping_data:
|
||
if meta.structured_group:
|
||
grouped.setdefault(meta.structured_group, []).append(meta)
|
||
|
||
rules: list[StructuredGroupRule] = []
|
||
for group_id, members in grouped.items():
|
||
aggregates = [
|
||
member for member in members if member.structured_role == "aggregate"
|
||
]
|
||
if len(aggregates) != 1:
|
||
raise ValueError(
|
||
f"Структурная группа {group_id!r} должна иметь один aggregate"
|
||
)
|
||
aggregate = aggregates[0]
|
||
one_line_columns = tuple(
|
||
member.index
|
||
for member in sorted(
|
||
(item for item in members if item.structured_role == "one_line"),
|
||
key=lambda item: item.structured_order,
|
||
)
|
||
)
|
||
component_columns = tuple(
|
||
member.index
|
||
for member in sorted(
|
||
(item for item in members if item.structured_role == "component"),
|
||
key=lambda item: item.structured_order,
|
||
)
|
||
)
|
||
if not component_columns:
|
||
raise ValueError(f"Структурная группа {group_id!r} не содержит components")
|
||
rules.append(
|
||
StructuredGroupRule(
|
||
structured_column=aggregate.index,
|
||
one_line_columns=one_line_columns,
|
||
component_columns=component_columns,
|
||
join_with_space=aggregate.join_with_space,
|
||
preserve_empty_positions="адрес" in aggregate.name.lower(),
|
||
)
|
||
)
|
||
return tuple(sorted(rules, key=lambda rule: rule.structured_column))
|
||
|
||
|
||
def _extract_tag_candidate_keys(tag: str) -> tuple[str, ...]:
|
||
candidates: list[str] = []
|
||
for raw_tag in re.split(r"[|,]", tag):
|
||
cleaned_tag = raw_tag.strip()
|
||
if not cleaned_tag or cleaned_tag in _INVALID_TAGS:
|
||
continue
|
||
if cleaned_tag not in candidates:
|
||
candidates.append(cleaned_tag)
|
||
return tuple(candidates)
|
||
|
||
|
||
def _extract_path_candidate_keys(path: str) -> tuple[str, ...]:
|
||
candidates: list[str] = []
|
||
for raw_path in re.findall(r"/[^\s|]+", path):
|
||
parts = [part for part in raw_path.strip("/").split("/") if part]
|
||
if not parts:
|
||
continue
|
||
for start_index in range(len(parts)):
|
||
dotted_path = ".".join(parts[start_index:])
|
||
if dotted_path and dotted_path not in candidates:
|
||
candidates.append(dotted_path)
|
||
return tuple(candidates)
|
||
|
||
|
||
_MAPPING_DATA = COLUMNS.ordered_columns()
|
||
FIXED_REPORT_COLUMNS: tuple[str, ...] = COLUMNS.all_columns()
|
||
REPORT_MAPPING_RULES: tuple[MappingRule, ...] = _build_rules(_MAPPING_DATA)
|
||
STRUCTURED_GROUP_RULES: tuple[StructuredGroupRule, ...] = _build_structured_group_rules(
|
||
_MAPPING_DATA
|
||
)
|
||
|
||
|
||
def build_fixed_row_by_index(
|
||
*,
|
||
file_name: str,
|
||
record_id: str,
|
||
operation_index: int,
|
||
operation_fields: dict[str, str],
|
||
participant_fields: dict[str, str],
|
||
is_continuation: bool = False,
|
||
) -> dict[int, str]:
|
||
row: dict[int, str] = {
|
||
COL_FILE_NAME: file_name,
|
||
COL_RECORD_ID: record_id,
|
||
}
|
||
operation_lookup = _build_payload_lookup(operation_fields)
|
||
participant_lookup = _build_payload_lookup(participant_fields)
|
||
participant_base_lookup, participant_eio_lookup = _partition_participant_lookup(
|
||
participant_lookup
|
||
)
|
||
merged_lookup = _build_payload_lookup(
|
||
_merge_fields(operation_fields, participant_fields)
|
||
)
|
||
has_eio_block = _lookup_has_any_prefixed_key(
|
||
participant_lookup,
|
||
(
|
||
"УчастникЮЛ.СведЕИО.",
|
||
"УчастникЮЛ.БенефициарЮЛ.",
|
||
"УчастникФЛИП.БенефициарФЛИП.",
|
||
"УчастникИНБОЮЛ.БенефициарИНБОЮЛ.",
|
||
"УчастникИНБОЮЛ.СведИНБОЮЛ.Учредитель.",
|
||
"СведЕИО.",
|
||
"БенефициарЮЛ.",
|
||
"БенефициарФЛИП.",
|
||
"БенефициарИНБОЮЛ.",
|
||
),
|
||
)
|
||
has_cp_block = _lookup_has_any_prefixed_key(
|
||
operation_lookup,
|
||
(
|
||
"СведЦП.",
|
||
"СообщОперКО.ИнформЧасть.СведКО.Операция.СведЦП.",
|
||
),
|
||
)
|
||
for rule in REPORT_MAPPING_RULES:
|
||
if rule.column_index in row:
|
||
continue
|
||
if not rule.allow_direct_mapping:
|
||
row[rule.column_index] = ""
|
||
continue
|
||
if rule.source == "participant":
|
||
source_lookup = (
|
||
participant_eio_lookup
|
||
if rule.column_index in BLOCK_EIO
|
||
else participant_base_lookup
|
||
)
|
||
elif rule.source == "operation":
|
||
source_lookup = operation_lookup
|
||
else:
|
||
source_lookup = merged_lookup
|
||
allow_short_lookup = rule.allow_short_lookup or rule.column_index in BLOCK_EIO
|
||
if rule.structured_value:
|
||
row[rule.column_index] = _pick_structured_value(
|
||
lookup=source_lookup,
|
||
candidates=rule.compiled_candidates,
|
||
join_with_space="фио" in rule.column_name.lower(),
|
||
allow_short_lookup=allow_short_lookup,
|
||
)
|
||
else:
|
||
row[rule.column_index] = _pick_value(
|
||
source_lookup,
|
||
rule.compiled_candidates,
|
||
allow_short_lookup=allow_short_lookup,
|
||
)
|
||
_set(
|
||
row,
|
||
COL_OPERATION_DOCUMENTS,
|
||
_format_operation_documents(operation_lookup),
|
||
)
|
||
_set(
|
||
row,
|
||
COL_EIO_OTHER_INFO,
|
||
_format_eio_other_info(participant_eio_lookup),
|
||
)
|
||
if not is_continuation:
|
||
_apply_participant_identity_rules(row, participant_lookup)
|
||
# Сначала убираем запрещённые однострочные адреса, чтобы структурная
|
||
# группа не очистила их компоненты до проверки резидентства.
|
||
_rule_address_one_line(row)
|
||
_apply_structured_group_rules(row)
|
||
_apply_one_line_structured_pairs(row)
|
||
if not is_continuation:
|
||
_apply_business_rules(
|
||
row,
|
||
has_eio_block=has_eio_block,
|
||
has_cp_block=has_cp_block,
|
||
operation_lookup=operation_lookup,
|
||
participant_lookup=participant_lookup,
|
||
)
|
||
else:
|
||
_apply_continuation_validation_rules(row)
|
||
for meta in COLUMNS.ordered_columns():
|
||
row.setdefault(meta.index, "")
|
||
return row
|
||
|
||
|
||
def build_fixed_row(
|
||
*,
|
||
file_name: str,
|
||
record_id: str,
|
||
operation_index: int,
|
||
operation_fields: dict[str, str],
|
||
participant_fields: dict[str, str],
|
||
is_continuation: bool = False,
|
||
) -> dict[str, str]:
|
||
by_index = build_fixed_row_by_index(
|
||
file_name=file_name,
|
||
record_id=record_id,
|
||
operation_index=operation_index,
|
||
operation_fields=operation_fields,
|
||
participant_fields=participant_fields,
|
||
is_continuation=is_continuation,
|
||
)
|
||
by_name: dict[str, str] = {}
|
||
for meta in COLUMNS.ordered_columns():
|
||
by_name.setdefault(meta.name, by_index.get(meta.index, ""))
|
||
return by_name
|
||
|
||
|
||
def _apply_structured_group_rules(row: dict[int, str]) -> None:
|
||
for rule in STRUCTURED_GROUP_RULES:
|
||
one_line_has_value = any(
|
||
_get(row, column_index).strip() for column_index in rule.one_line_columns
|
||
)
|
||
if one_line_has_value:
|
||
_set(row, rule.structured_column, "")
|
||
for component_column_index in rule.component_columns:
|
||
_set(row, component_column_index, "")
|
||
continue
|
||
|
||
component_values = [
|
||
_get(row, component_column_index).strip()
|
||
for component_column_index in rule.component_columns
|
||
]
|
||
if not any(component_values):
|
||
continue
|
||
|
||
values = (
|
||
component_values
|
||
if rule.preserve_empty_positions
|
||
else [value for value in component_values if value]
|
||
)
|
||
separator = " " if rule.join_with_space else ", "
|
||
_set(row, rule.structured_column, separator.join(values))
|
||
|
||
|
||
def _apply_one_line_structured_pairs(row: dict[int, str]) -> None:
|
||
"""Оставляет только однострочное значение, если заполнены обе версии поля."""
|
||
for one_line_column, structured_column in _ONE_LINE_STRUCTURED_PAIRS:
|
||
if _get(row, one_line_column):
|
||
_set(row, structured_column, "")
|
||
|
||
|
||
def _merge_fields(
|
||
operation_fields: dict[str, str],
|
||
participant_fields: dict[str, str],
|
||
) -> dict[str, str]:
|
||
merged: dict[str, str] = {}
|
||
merged.update(operation_fields)
|
||
merged.update(participant_fields)
|
||
return merged
|
||
|
||
|
||
def _normalize_indexed_path(path: str) -> str:
|
||
"""Убирает индексы повторов, сохраняя логическую структуру XML-пути."""
|
||
return _INDEXED_PATH_SEGMENT.sub("", path)
|
||
|
||
|
||
def _build_payload_lookup(payload: dict[str, str]) -> _PayloadLookup:
|
||
entries = tuple(
|
||
_LookupEntry(
|
||
raw_path=key,
|
||
normalized_path=_normalize_indexed_path(key),
|
||
value=value,
|
||
)
|
||
for key, value in payload.items()
|
||
)
|
||
return _build_payload_lookup_from_entries(entries)
|
||
|
||
|
||
def _build_payload_lookup_from_entries(
|
||
entries: tuple[_LookupEntry, ...],
|
||
) -> _PayloadLookup:
|
||
values = {entry.raw_path: entry.value for entry in entries}
|
||
normalized_exact_first: dict[str, str] = {}
|
||
normalized_suffix_first: dict[str, str] = {}
|
||
raw_short_first: dict[str, str] = {}
|
||
|
||
for entry in entries:
|
||
if not entry.value:
|
||
continue
|
||
normalized_exact_first.setdefault(entry.normalized_path, entry.value)
|
||
path_parts = entry.normalized_path.split(".")
|
||
for start_index in range(len(path_parts)):
|
||
suffix = ".".join(path_parts[start_index:])
|
||
normalized_suffix_first.setdefault(suffix, entry.value)
|
||
raw_short = entry.raw_path.rsplit(".", maxsplit=1)[-1]
|
||
raw_short_first.setdefault(raw_short, entry.value)
|
||
|
||
return _PayloadLookup(
|
||
values=values,
|
||
entries=entries,
|
||
normalized_paths=tuple(entry.normalized_path for entry in entries),
|
||
normalized_exact_first=normalized_exact_first,
|
||
normalized_suffix_first=normalized_suffix_first,
|
||
raw_short_first=raw_short_first,
|
||
)
|
||
|
||
|
||
def _is_normalized_eio_path(normalized_path: str) -> bool:
|
||
surrounded_path = f".{normalized_path}."
|
||
return ".СведИНБОЮЛ.Учредитель." in surrounded_path or any(
|
||
f".{block_name}." in surrounded_path for block_name in _EIO_BLOCK_NAMES
|
||
)
|
||
|
||
|
||
def _partition_participant_lookup(
|
||
participant_lookup: _PayloadLookup,
|
||
) -> tuple[_PayloadLookup, _PayloadLookup]:
|
||
base_entries: list[_LookupEntry] = []
|
||
eio_entries: list[_LookupEntry] = []
|
||
for entry in participant_lookup.entries:
|
||
target = (
|
||
eio_entries
|
||
if _is_normalized_eio_path(entry.normalized_path)
|
||
else base_entries
|
||
)
|
||
target.append(entry)
|
||
return (
|
||
_build_payload_lookup_from_entries(tuple(base_entries)),
|
||
_build_payload_lookup_from_entries(tuple(eio_entries)),
|
||
)
|
||
|
||
|
||
def _lookup_has_payload_block(lookup: _PayloadLookup, block_name: str) -> bool:
|
||
prefix = f"{block_name}."
|
||
nested_marker = f".{block_name}."
|
||
return any(
|
||
normalized_path.startswith(prefix) or nested_marker in normalized_path
|
||
for normalized_path in lookup.normalized_paths
|
||
)
|
||
|
||
|
||
def _lookup_pick_scoped_value(lookup: _PayloadLookup, scoped_path: str) -> str:
|
||
value = lookup.normalized_suffix_first.get(scoped_path, "")
|
||
return str(value).strip() if value else ""
|
||
|
||
|
||
def _lookup_pick_direct_value(lookup: _PayloadLookup, field_name: str) -> str:
|
||
"""Возвращает только прямое поле текущего XML-блока без suffix-поиска."""
|
||
value = lookup.normalized_exact_first.get(field_name, "")
|
||
return str(value).strip() if value else ""
|
||
|
||
|
||
def _format_operation_documents(lookup: _PayloadLookup) -> str:
|
||
indexed_documents: dict[int, dict[str, str]] = {}
|
||
single_document: dict[str, str] = {}
|
||
for entry in lookup.entries:
|
||
match = _OPERATION_DOCUMENT_FIELD.search(entry.raw_path)
|
||
if match is None:
|
||
continue
|
||
field_name = match.group("field")
|
||
value = str(entry.value).strip()
|
||
document_index = match.group("index")
|
||
if document_index is None:
|
||
single_document[field_name] = value
|
||
continue
|
||
indexed_documents.setdefault(int(document_index), {})[field_name] = value
|
||
|
||
documents = (
|
||
[indexed_documents[index] for index in sorted(indexed_documents)]
|
||
if indexed_documents
|
||
else [single_document] if single_document else []
|
||
)
|
||
formatted_documents = []
|
||
for position, document in enumerate(documents, start=1):
|
||
fields = [
|
||
document.get(field_name, "") for field_name in _OPERATION_DOCUMENT_FIELDS
|
||
]
|
||
fields.insert(1, "")
|
||
formatted_documents.append(f"{position}. {', '.join(fields)};")
|
||
return "\n".join(formatted_documents)
|
||
|
||
|
||
def _format_eio_other_info(lookup: _PayloadLookup) -> str:
|
||
values: list[tuple[str, str]] = []
|
||
for label, field_name in _EIO_OTHER_INFO_FIELDS:
|
||
value = ""
|
||
for base_path in _EIO_PERSON_BASE_PATHS:
|
||
value = _lookup_pick_scoped_value(lookup, f"{base_path}.{field_name}")
|
||
if value:
|
||
break
|
||
values.append((label, value))
|
||
|
||
if not any(value for _, value in values):
|
||
return ""
|
||
lines = [f"{label}: {value}" for label, value in values]
|
||
lines.append("Дата регистрации: ")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _apply_participant_identity_rules(
|
||
row: dict[int, str], participant_lookup: _PayloadLookup
|
||
) -> None:
|
||
participant_type = _normalize_code(_get(row, COL_PARTICIPANT_TYPE))
|
||
has_legal_entity = _lookup_has_payload_block(participant_lookup, "УчастникЮЛ")
|
||
has_physical_person = _lookup_has_payload_block(participant_lookup, "УчастникФЛИП")
|
||
has_foreign_structure = _lookup_has_payload_block(
|
||
participant_lookup, "УчастникИНБОЮЛ"
|
||
)
|
||
|
||
if has_physical_person and not has_legal_entity:
|
||
identification = _lookup_pick_scoped_value(
|
||
participant_lookup, "УчастникФЛИП.ИдентификацияФЛ"
|
||
)
|
||
_set(row, COL_IDENT_FL, identification)
|
||
else:
|
||
_set(row, COL_IDENT_FL, "")
|
||
|
||
if has_legal_entity or participant_type == PARTICIPANT_TYPE_UL:
|
||
value = _lookup_pick_scoped_value(participant_lookup, "УчастникЮЛ.СведЮЛ.КППЮЛ")
|
||
elif has_foreign_structure:
|
||
value = _lookup_pick_scoped_value(
|
||
participant_lookup,
|
||
"УчастникИНБОЮЛ.СведИНБОЮЛ.ПризнакОргФормаИНБОЮЛ",
|
||
)
|
||
elif has_physical_person:
|
||
value = _lookup_pick_scoped_value(
|
||
participant_lookup, "УчастникФЛИП.ИдентификацияФЛ"
|
||
)
|
||
else:
|
||
value = ""
|
||
_set(row, COL_KPP, value)
|
||
|
||
|
||
def _lookup_has_any_prefixed_key(
|
||
lookup: _PayloadLookup, prefixes: tuple[str, ...]
|
||
) -> bool:
|
||
return any(
|
||
any(normalized_path.startswith(prefix) for prefix in prefixes)
|
||
for normalized_path in lookup.normalized_paths
|
||
)
|
||
|
||
|
||
def _pick_value(
|
||
lookup: _PayloadLookup,
|
||
candidates: tuple[_CompiledCandidate, ...],
|
||
*,
|
||
allow_short_lookup: bool = True,
|
||
) -> str:
|
||
for candidate in candidates:
|
||
if not allow_short_lookup and not candidate.strict_scoped:
|
||
continue
|
||
if value := lookup.values.get(candidate.raw):
|
||
return value
|
||
if value := lookup.normalized_suffix_first.get(candidate.normalized):
|
||
return value
|
||
if not allow_short_lookup:
|
||
for entry in lookup.entries:
|
||
if (
|
||
entry.value
|
||
and "." in entry.normalized_path
|
||
and candidate.normalized.endswith(f".{entry.normalized_path}")
|
||
):
|
||
return entry.value
|
||
continue
|
||
if value := lookup.values.get(candidate.short):
|
||
return value
|
||
if value := lookup.raw_short_first.get(candidate.short):
|
||
return value
|
||
return ""
|
||
|
||
|
||
def _pick_structured_value(
|
||
*,
|
||
lookup: _PayloadLookup,
|
||
candidates: tuple[_CompiledCandidate, ...],
|
||
join_with_space: bool,
|
||
allow_short_lookup: bool = True,
|
||
) -> str:
|
||
values: list[str] = []
|
||
for candidate in candidates:
|
||
value = _pick_value(
|
||
lookup,
|
||
(candidate,),
|
||
allow_short_lookup=allow_short_lookup,
|
||
)
|
||
if value and value not in values:
|
||
values.append(value)
|
||
if not values:
|
||
return ""
|
||
separator = " " if join_with_space else ", "
|
||
return separator.join(values)
|
||
|
||
|
||
def _get(row: dict[int, str], column: int) -> str:
|
||
return str(row.get(column, "")).strip()
|
||
|
||
|
||
def _normalize_code(value: str) -> str:
|
||
normalized = value.strip()
|
||
if re.fullmatch(r"\d+", normalized):
|
||
return normalized.lstrip("0") or "0"
|
||
return normalized
|
||
|
||
|
||
def _set(row: dict[int, str], column: int, value: str) -> None:
|
||
row[column] = value
|
||
|
||
|
||
def _clear_columns(row: dict[int, str], columns: tuple[int, ...]) -> None:
|
||
for column in columns:
|
||
_set(row, column, "")
|
||
|
||
|
||
def _extract_numeric_codes(value: str) -> set[str]:
|
||
return {match for match in re.findall(r"\d+", value)}
|
||
|
||
|
||
def _operation_codes(row: dict[int, str]) -> set[str]:
|
||
op_code = _get(row, COL_OPERATION_CODE)
|
||
extra_codes = _get(row, COL_EXTRA_CODES)
|
||
return _extract_numeric_codes(op_code) | _extract_numeric_codes(extra_codes)
|
||
|
||
|
||
def _apply_business_rules(
|
||
row: dict[int, str],
|
||
*,
|
||
has_eio_block: bool,
|
||
has_cp_block: bool,
|
||
operation_lookup: _PayloadLookup,
|
||
participant_lookup: _PayloadLookup,
|
||
) -> None:
|
||
codes = _operation_codes(row)
|
||
operation_sign = _get(row, COL_OPERATION_SIGN)
|
||
transfer_type = _get(row, COL_TRANSFER_TYPE)
|
||
operator_type = _get(row, COL_OPERATOR_TYPE)
|
||
participant_type = _normalize_code(_get(row, COL_PARTICIPANT_TYPE))
|
||
|
||
_rule_ftr_sign(row)
|
||
_rule_suspension_basis(row)
|
||
_rule_esp_sign(row)
|
||
_rule_unusual_codes(row, codes)
|
||
_rule_digital_rights_currency(row, operation_sign)
|
||
_rule_sale_currency(row)
|
||
_rule_extra_info(row, codes, operation_lookup, participant_lookup)
|
||
_rule_suspicious_activity(row, codes)
|
||
_rule_metal_name(row)
|
||
_rule_item_type(row, codes)
|
||
_rule_territory_codes(row, codes)
|
||
|
||
_rule_payer_account(row, transfer_type, codes)
|
||
_rule_esp_identifiers(row, operation_sign)
|
||
_rule_payer_bank(row, operator_type)
|
||
_rule_recipient_bank(row, operator_type)
|
||
_rule_correspondent_accounts(row, operator_type, transfer_type, codes)
|
||
_rule_recipient_account(row, transfer_type)
|
||
_rule_transfer_status(row, transfer_type)
|
||
|
||
_rule_cashless_clears_cash_columns(row, transfer_type)
|
||
_rule_card_without_employee(row)
|
||
|
||
_rule_participant_identity(row, participant_type)
|
||
_rule_branch_sign(row, participant_type)
|
||
_rule_snils_for_ip(row, participant_type)
|
||
_rule_address_one_line(row)
|
||
_rule_okato_by_country(row)
|
||
_rule_eio_inn_for_physical_person(row, participant_type)
|
||
|
||
if not has_eio_block:
|
||
_clear_columns(row, BLOCK_EIO)
|
||
if not has_cp_block:
|
||
_clear_columns(row, BLOCK_CP)
|
||
|
||
|
||
def _apply_continuation_validation_rules(row: dict[int, str]) -> None:
|
||
"""Проверяет локальные коды повтора без подстановки значений основной строки."""
|
||
_rule_ftr_sign(row)
|
||
_rule_suspension_basis(row)
|
||
_rule_esp_sign(row)
|
||
|
||
|
||
def _rule_ftr_sign(row: dict[int, str]) -> None:
|
||
"""Правило 10: оставляем только допустимые значения признака ФТР."""
|
||
if _get(row, COL_FTR_SIGN) not in VALID_FTR_SIGNS:
|
||
_set(row, COL_FTR_SIGN, "")
|
||
|
||
|
||
def _rule_suspension_basis(row: dict[int, str]) -> None:
|
||
"""Правило 11: основание приостановления должно быть допустимым."""
|
||
if _get(row, COL_SUSPENSION_BASIS) not in VALID_SUSPENSION_BASES:
|
||
_set(row, COL_SUSPENSION_BASIS, "")
|
||
|
||
|
||
def _rule_esp_sign(row: dict[int, str]) -> None:
|
||
"""Правило 15: признак ЭСП должен быть допустимым."""
|
||
if _get(row, COL_ESP_SIGN) not in VALID_ESP_SIGNS:
|
||
_set(row, COL_ESP_SIGN, "")
|
||
|
||
|
||
def _rule_unusual_codes(row: dict[int, str], codes: set[str]) -> None:
|
||
"""Правило 21: коды необычной операции относятся только к 6001."""
|
||
if SUSPICIOUS_CODE not in codes:
|
||
_set(row, COL_UNUSUAL_CODES, "")
|
||
|
||
|
||
def _rule_digital_rights_currency(row: dict[int, str], operation_sign: str) -> None:
|
||
"""Правила 22-24: для цифровых прав валютные суммы не выводятся."""
|
||
if operation_sign == "8":
|
||
_set(row, COL_CURRENCY, "")
|
||
_set(row, COL_AMOUNT_CURRENCY, "")
|
||
_set(row, COL_AMOUNT_RUB, "")
|
||
|
||
|
||
def _rule_sale_currency(row: dict[int, str]) -> None:
|
||
"""Правило 26: нулевая сумма или отсутствие конверсии дают пустое поле."""
|
||
conversion_amount = _get(row, COL_SALE_AMOUNT)
|
||
normalized_amount = conversion_amount.replace(",", ".")
|
||
if not _get(row, COL_SALE_CURRENCY) or re.fullmatch(
|
||
r"[+-]?0+(?:\.0+)?", normalized_amount
|
||
):
|
||
_set(row, COL_SALE_AMOUNT, "")
|
||
|
||
|
||
def _rule_extra_info(
|
||
row: dict[int, str],
|
||
codes: set[str],
|
||
operation_lookup: _PayloadLookup,
|
||
participant_lookup: _PayloadLookup,
|
||
) -> None:
|
||
"""Правило 34: объединяет допустимые комментарии операции и участника."""
|
||
operation_comment = _lookup_pick_direct_value(operation_lookup, "Коммент")
|
||
participant_comment = ""
|
||
if codes & EXPORT_SUBSIDIARY_COMMENT_CODES:
|
||
participant_comment = _lookup_pick_direct_value(
|
||
participant_lookup, "КомментУчастник"
|
||
)
|
||
|
||
values = list(
|
||
dict.fromkeys(
|
||
value for value in (operation_comment, participant_comment) if value
|
||
)
|
||
)
|
||
_set(row, COL_EXTRA_INFO, "; ".join(values))
|
||
|
||
|
||
def _rule_suspicious_activity(row: dict[int, str], codes: set[str]) -> None:
|
||
"""Правило 28: идентификатор подозрительной деятельности только для 6001."""
|
||
if SUSPICIOUS_CODE not in codes:
|
||
_set(row, COL_SUSPICIOUS_ID, "")
|
||
|
||
|
||
def _rule_metal_name(row: dict[int, str]) -> None:
|
||
"""Правило 30: наименование металла только для кода C99."""
|
||
if _get(row, COL_METAL_CODE).upper() != PRECIOUS_METAL_CODE:
|
||
_set(row, COL_METAL_NAME, "")
|
||
|
||
|
||
def _rule_item_type(row: dict[int, str], codes: set[str]) -> None:
|
||
"""Правило 31: предмет операции только для кодов 5020-5023."""
|
||
if not codes & PRECIOUS_ITEM_CODES:
|
||
_set(row, COL_ITEM_TYPE, "")
|
||
|
||
|
||
def _rule_territory_codes(row: dict[int, str], codes: set[str]) -> None:
|
||
"""Правило 36: код территории (КодТерИнГос) — только для операции 5016.
|
||
|
||
Колонка 96 (КодТерИнБанк, признак 96) сюда НЕ относится — это другой
|
||
одноимённый показатель официальной спецификации (п.59), обязательный
|
||
для блока СведенияКартаИнБанк (код операции 1009), а не для 5016.
|
||
Раньше обе колонки ошибочно чистились по одному условию (TERRITORY_CODE
|
||
not in codes), из-за чего колонка 96 всегда обнулялась для 1009-строк —
|
||
именно 5016-код там никогда не бывает. Для колонки 96 отдельная очистка
|
||
не нужна: движок маппинга и так подставляет в неё значение только тогда,
|
||
когда в XML реально есть СведенияКартаИнБанк/КодТерИнБанк.
|
||
"""
|
||
if TERRITORY_CODE not in codes:
|
||
_set(row, COL_TERRITORY_CODE, "")
|
||
|
||
|
||
def _rule_payer_account(
|
||
row: dict[int, str], transfer_type: str, codes: set[str]
|
||
) -> None:
|
||
"""Правило 38: счёт плательщика зависит от вида перевода.
|
||
|
||
Для операций с кодом 5016 (п.6 правок) заглушку не подставляем — такого
|
||
блока (номера счетов) для 5016 в спецификации нет.
|
||
"""
|
||
if TERRITORY_CODE in codes:
|
||
_set(row, COL_PAYER_ACCOUNT, "")
|
||
return
|
||
if transfer_type in {"1", "2", "3", "10"}:
|
||
if not _get(row, COL_PAYER_ACCOUNT):
|
||
_set(row, COL_PAYER_ACCOUNT, ACCOUNT_PLACEHOLDER)
|
||
elif transfer_type not in {"12", "13", "14"}:
|
||
_set(row, COL_PAYER_ACCOUNT, "")
|
||
|
||
|
||
def _rule_esp_identifiers(row: dict[int, str], operation_sign: str) -> None:
|
||
"""Правила 39 и 47: идентификаторы ЭСП отсутствуют для признака 9."""
|
||
if operation_sign == "9":
|
||
_set(row, COL_PAYER_ESP, "")
|
||
_set(row, COL_RECIPIENT_ESP, "")
|
||
|
||
|
||
def _rule_payer_bank(row: dict[int, str], operator_type: str) -> None:
|
||
"""Правило 40: банк плательщика только для типов оператора 2 и 4."""
|
||
if operator_type not in {"2", "4"}:
|
||
_set(row, COL_PAYER_BANK_NAME, "")
|
||
|
||
|
||
def _rule_recipient_bank(row: dict[int, str], operator_type: str) -> None:
|
||
"""Правила 42-43: банк получателя только для типов оператора 1 и 4."""
|
||
if operator_type not in {"1", "4"}:
|
||
_set(row, COL_RECIPIENT_BANK_NAME, "")
|
||
_set(row, COL_RECIPIENT_BANK_BIK, "")
|
||
|
||
|
||
def _rule_correspondent_accounts(
|
||
row: dict[int, str], operator_type: str, transfer_type: str, codes: set[str]
|
||
) -> None:
|
||
"""Правила 44-45: корсчета зависят от оператора и вида перевода.
|
||
|
||
Для операций с кодом 5016 (п.6 правок) заглушку не подставляем — такого
|
||
блока (номера счетов) для 5016 в спецификации нет.
|
||
"""
|
||
if TERRITORY_CODE in codes:
|
||
_set(row, COL_PAYER_BANK_ACCOUNT, "")
|
||
_set(row, COL_RECIPIENT_BANK_ACCOUNT, "")
|
||
return
|
||
|
||
if not operator_type and not transfer_type:
|
||
_set(row, COL_PAYER_BANK_ACCOUNT, "")
|
||
_set(row, COL_RECIPIENT_BANK_ACCOUNT, "")
|
||
return
|
||
|
||
if operator_type in {"3", "5"} or transfer_type in {"12", "13"}:
|
||
_set(row, COL_PAYER_BANK_ACCOUNT, "")
|
||
elif not _get(row, COL_PAYER_BANK_ACCOUNT):
|
||
_set(row, COL_PAYER_BANK_ACCOUNT, ACCOUNT_PLACEHOLDER)
|
||
|
||
if operator_type in {"3", "5"} or transfer_type in {"10", "11"}:
|
||
_set(row, COL_RECIPIENT_BANK_ACCOUNT, "")
|
||
elif not _get(row, COL_RECIPIENT_BANK_ACCOUNT):
|
||
_set(row, COL_RECIPIENT_BANK_ACCOUNT, ACCOUNT_PLACEHOLDER)
|
||
|
||
|
||
def _rule_recipient_account(row: dict[int, str], transfer_type: str) -> None:
|
||
"""Правило 46: счёт получателя зависит от вида перевода."""
|
||
if transfer_type in {"1", "4", "7", "12"}:
|
||
if not _get(row, COL_RECIPIENT_ACCOUNT):
|
||
_set(row, COL_RECIPIENT_ACCOUNT, ACCOUNT_PLACEHOLDER)
|
||
elif transfer_type not in {"10", "11", "14"}:
|
||
_set(row, COL_RECIPIENT_ACCOUNT, "")
|
||
|
||
|
||
def _rule_transfer_status(row: dict[int, str], transfer_type: str) -> None:
|
||
"""Правило 62: статус перевода только для видов 2, 5 и 8."""
|
||
if transfer_type not in {"2", "5", "8"}:
|
||
_set(row, COL_TRANSFER_STATUS, "")
|
||
|
||
|
||
def _rule_cashless_clears_cash_columns(row: dict[int, str], transfer_type: str) -> None:
|
||
"""Для безналичных переводов очищаем согласованные наличные поля."""
|
||
if transfer_type in CASHLESS_TRANSFER_TYPES:
|
||
_clear_columns(row, CASHLESS_CLEARED_COLUMNS)
|
||
|
||
|
||
def _rule_card_without_employee(row: dict[int, str]) -> None:
|
||
"""Правила 103 и 105: подставляем заглушку при отсутствии сотрудника."""
|
||
if _get(row, COL_EMPLOYEE_SIGN) != EMPLOYEE_ABSENT_CODE:
|
||
return
|
||
if not _get(row, COL_CARD_HOLDER_INFO):
|
||
_set(row, COL_CARD_HOLDER_INFO, "Информация отсутствует")
|
||
if not _get(row, COL_FOREIGN_BANK_NAME):
|
||
_set(row, COL_FOREIGN_BANK_NAME, "Информация отсутствует")
|
||
|
||
|
||
def _rule_participant_identity(row: dict[int, str], participant_type: str) -> None:
|
||
"""Правила 119 и 122: идентификация и КПП зависят от типа участника."""
|
||
if participant_type in {PARTICIPANT_TYPE_UL, PARTICIPANT_TYPE_FL}:
|
||
_set(row, COL_IDENT_FL, "")
|
||
if participant_type == PARTICIPANT_TYPE_FL:
|
||
_set(row, COL_KPP, "")
|
||
|
||
|
||
def _rule_branch_sign(row: dict[int, str], participant_type: str) -> None:
|
||
"""Правило 124: показатель отсутствует для ФЛ, ИП и ФЛЧП."""
|
||
if participant_type in {
|
||
PARTICIPANT_TYPE_FL,
|
||
PARTICIPANT_TYPE_IP,
|
||
PARTICIPANT_TYPE_FLCHP,
|
||
}:
|
||
_set(row, COL_BRANCH_SIGN, "")
|
||
|
||
|
||
def _rule_snils_for_ip(row: dict[int, str], participant_type: str) -> None:
|
||
"""Правило 125: СНИЛС не выводится для ИП."""
|
||
if participant_type != PARTICIPANT_TYPE_IP:
|
||
return
|
||
insurance_value = _get(row, COL_INSURANCE_NUMBER)
|
||
if re.fullmatch(r"\d{3}-?\d{3}-?\d{3}\s?\d{2}", insurance_value):
|
||
_set(row, COL_INSURANCE_NUMBER, "")
|
||
|
||
|
||
def _allow_one_line_address(row: dict[int, str]) -> bool:
|
||
resident_flag = _get(row, COL_RESIDENT_SIGN)
|
||
client_flag = _get(row, COL_CLIENT_SIGN)
|
||
if not resident_flag and not client_flag:
|
||
return True
|
||
return resident_flag in {"0", "9"} or (resident_flag == "1" and client_flag == "0")
|
||
|
||
|
||
def _rule_address_one_line(row: dict[int, str]) -> None:
|
||
"""Правила 141, 152, 167 и 178: однострочные адреса только для нерезидентов."""
|
||
if not _allow_one_line_address(row):
|
||
_set(row, COL_ADDRESS_ONE_LINE, "")
|
||
_set(row, COL_REG_PLACE_ONE_LINE, "")
|
||
_set(row, COL_EIO_ADDRESS_ONE_LINE, "")
|
||
_set(row, COL_EIO_REG_PLACE_ONE_LINE, "")
|
||
|
||
|
||
def _rule_okato_by_country(row: dict[int, str]) -> None:
|
||
"""ОКАТО адресного блока заполняется только для России."""
|
||
for country_column, okato_column in OKATO_COUNTRY_PAIRS:
|
||
if _get(row, country_column) != RUSSIA_COUNTRY_CODE:
|
||
_set(row, okato_column, "")
|
||
|
||
|
||
def _rule_eio_inn_for_physical_person(
|
||
row: dict[int, str], participant_type: str
|
||
) -> None:
|
||
"""Правило 163: показатель ЕИО отсутствует для ФЛ."""
|
||
if participant_type == PARTICIPANT_TYPE_FL:
|
||
_set(row, COL_EIO_INN, "")
|