880fa5811d
The customer's ruling of 24.09.2026 is that every value present in the XML must reach the report, taking into account which block holds it. Two defects broke that for operation code 5016 with transfer types 4 and 5. Wrong paths in mapping_table.json. Columns 48-63 spelled the block as "СведенияПереводыДс" and columns 49-59 spelled the address as "АдрМестоПриемаВыдача"; the schema uses "СведенияПереводыДС" and "АдрМестаПриемаВыдача". Those columns carry allow_short_lookup=false, so they can only match on a fully scoped path and never matched at all: the whole cash receipt block (AV-BI) and the transfer status (BJ) stayed empty on every file. The existing test passed only because it fed the same misspellings. Rules that deleted data instead of filling it: - Rules 38, 44-45 and 46 wiped the account columns for code 5016. The XML of those records carries СчетБанкПолучатель = 00000000000000000000, which is exactly the value the customer asked for. The rules now only substitute the placeholder into an empty column, and never for 5016. - Rule 44 also wiped the correspondent accounts for operator types 3 and 5 even when the XML had them. - The cashless rule wiped columns 50, 61, 65, 76, 84, 95, 99 and 101 for transfer types 1-4, 7 and 10-14, although type 4 is by definition a transfer that accepts cash and must carry the receipt block. - Rule 62 limited the transfer status to types 2, 5 and 8. - Rule 30 limited the metal name to code C99, while the available exports only ever carry code A90. Verified on the sample exports: all 16 records with code 5016 now carry the zeros in column 45, the real account in column 46, and no invented placeholders in columns 38 and 44.
1155 lines
43 KiB
Python
1155 lines
43 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
|
||
from .column_constants import (
|
||
ACCOUNT_PLACEHOLDER,
|
||
BLOCK_CP,
|
||
BLOCK_EIO,
|
||
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_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_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,
|
||
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_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, codes)
|
||
|
||
_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_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: заглушка счёта плательщика для отдельных видов перевода.
|
||
|
||
Значение, пришедшее из XML, не затирается никогда: заказчик 24.09.2026
|
||
подтвердил, что в отчёт должны попадать все значения из файла. Правило
|
||
только подставляет заглушку в пустую колонку, а для операций с кодом 5016
|
||
(п.6 правок) не делает и этого — своего блока номеров счетов у 5016 нет.
|
||
"""
|
||
if TERRITORY_CODE in codes:
|
||
return
|
||
if transfer_type in {"1", "2", "3", "10"} and not _get(row, COL_PAYER_ACCOUNT):
|
||
_set(row, COL_PAYER_ACCOUNT, ACCOUNT_PLACEHOLDER)
|
||
|
||
|
||
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: заглушки корсчетов зависят от оператора и вида перевода.
|
||
|
||
Как и в правиле 38, пришедшее из XML значение сохраняется всегда. Это
|
||
важно именно для кода 5016: в таких записях в XML реально лежит
|
||
СчетБанкПолучатель = 00000000000000000000, и заказчик 24.09.2026 отдельно
|
||
указал, что эти нули обязаны попадать в отчёт. Раньше их затирала ветка
|
||
по коду 5016, а корсчёт плательщика — ветка по типам оператора 3 и 5.
|
||
"""
|
||
if TERRITORY_CODE in codes:
|
||
return
|
||
if not operator_type and not transfer_type:
|
||
return
|
||
|
||
if (
|
||
operator_type not in {"3", "5"}
|
||
and transfer_type not in {"12", "13"}
|
||
and not _get(row, COL_PAYER_BANK_ACCOUNT)
|
||
):
|
||
_set(row, COL_PAYER_BANK_ACCOUNT, ACCOUNT_PLACEHOLDER)
|
||
|
||
if (
|
||
operator_type not in {"3", "5"}
|
||
and transfer_type not in {"10", "11"}
|
||
and 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, codes: set[str]
|
||
) -> None:
|
||
"""Правило 46: заглушка счёта получателя для отдельных видов перевода.
|
||
|
||
Как и правила 38 и 44-45, значение из XML не затирает, а для кода 5016
|
||
заглушку не подставляет.
|
||
"""
|
||
if TERRITORY_CODE in codes:
|
||
return
|
||
if transfer_type in {"1", "4", "7", "12"} and not _get(row, COL_RECIPIENT_ACCOUNT):
|
||
_set(row, COL_RECIPIENT_ACCOUNT, ACCOUNT_PLACEHOLDER)
|
||
|
||
|
||
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, "")
|