cceee02243
Every rule that emptied a column because the operation code, transfer type or participant type was not in an allowed list is gone. What is left never loses data: it assembles a column from several tags, substitutes the zero placeholder into an empty account column, and clears the ЕИО/ЦП blocks when the record has no such block at all. Removed: rules 10, 11, 15 (code validity), 21, 22-24, 26, 28, 31, 36, 39/47, 40, 42-43, 119/122, 124, 125, 141/152/167/178, the ОКАТО-by-country rule, 163, and the continuation-row validation, which only re-ran the three code checks. Removing them exposed three defects the rules had been masking, all of them the other half of the customer's requirement — a value must land in the column of its own block: - Columns 40 and 42 pulled the bank name from a foreign block. The tag НаимКО sits in six blocks, and short lookup matched the first one, so a record with only СведБанкПолучатель filled the payer's column too. 4180 and 6906 wrong values on the 03.02.2023 export. Fixed by forbidding short lookup, the same way columns 41 and 43 already do it. - Column 141 pulled the participant address from the cash place block for the same reason. All 380 values it held were wrong; the 311 that remain now come from the participant's own address block, verified one to one against the XML. - The one-line and structured address versions were treated as mutually exclusive: a filled one-line column wiped the structured column and every component. That cost 50 structured addresses on this export. Now both are kept, and only an exact duplicate of the one-line value is suppressed, so a single tag matching both columns is not echoed twice. Three path typos in mapping_table.json also came out, checked against the schema: УчастникФлип, СвебИНБОЮЛ and полноеНаимИНБОЮЛ. The first made the participant one-line address unreachable by its scoped path, which is why it depended on the leaking short lookup. On the 03.02.2023 export only 12 columns change and nothing but column 141 goes down. The 5016 cross-check still reports no losses: every value in the cash receipt, cash payment, status and account columns reaches the report.
972 lines
37 KiB
Python
972 lines
37 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
|
||
from .column_constants import (
|
||
ACCOUNT_PLACEHOLDER,
|
||
BLOCK_CP,
|
||
BLOCK_EIO,
|
||
COL_ADDRESS_STRUCTURED,
|
||
COL_BIRTH_REG_DATE,
|
||
COL_CARD_HOLDER_INFO,
|
||
COL_CLIENT_SIGN,
|
||
COL_COUNTRY_CODE,
|
||
COL_CP_ISSUER_INN,
|
||
COL_EIO_OTHER_INFO,
|
||
COL_EIO_REG_PLACE_ONE_LINE,
|
||
COL_EIO_REG_PLACE_STRUCTURED,
|
||
COL_EMPLOYEE_SIGN,
|
||
COL_EXTRA_CODES,
|
||
COL_EXTRA_INFO,
|
||
COL_FILE_NAME,
|
||
COL_FOREIGN_BANK_NAME,
|
||
COL_FOREIGN_CARD_ADDRESS_STRUCTURED,
|
||
COL_IDENT_FL,
|
||
COL_INN,
|
||
COL_KPP,
|
||
COL_OPERATION_CODE,
|
||
COL_OPERATION_DOCUMENTS,
|
||
COL_OPERATOR_TYPE,
|
||
COL_PARTICIPANT_TYPE,
|
||
COL_PAYER_ACCOUNT,
|
||
COL_PAYER_BANK_ACCOUNT,
|
||
COL_PAYER_BANK_BIK,
|
||
COL_RECIPIENT_ACCOUNT,
|
||
COL_RECIPIENT_BANK_ACCOUNT,
|
||
COL_RECIPIENT_BANK_BIK,
|
||
COL_RECORD_ID,
|
||
COL_REG_PLACE_ONE_LINE,
|
||
COL_REG_PLACE_STRUCTURED,
|
||
COL_RESIDENT_SIGN,
|
||
COL_STAY_DOCUMENT_TYPE,
|
||
COL_TRANSFER_TYPE,
|
||
EMPLOYEE_ABSENT_CODE,
|
||
EXPORT_SUBSIDIARY_COMMENT_CODES,
|
||
PARTICIPANT_ADDRESS_COMPONENT_COLUMNS,
|
||
PARTICIPANT_TYPE_UL,
|
||
TERRITORY_CODE,
|
||
)
|
||
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 = (
|
||
"СведЕИО",
|
||
"БенефициарЮЛ",
|
||
"БенефициарФЛИП",
|
||
"БенефициарИНБОЮЛ",
|
||
)
|
||
_ONE_LINE_STRUCTURED_PAIRS = (
|
||
(COL_REG_PLACE_ONE_LINE, COL_REG_PLACE_STRUCTURED),
|
||
(COL_EIO_REG_PLACE_ONE_LINE, COL_EIO_REG_PLACE_STRUCTURED),
|
||
)
|
||
_PARTICIPANT_ADDRESS_BASE_PATHS = (
|
||
"УчастникЮЛ.СведЮЛ.АдрРегЮЛ",
|
||
"УчастникФЛИП.СведФЛИП.АдрРег",
|
||
"УчастникИНБОЮЛ.СведИНБОЮЛ.МестоДеятельностьИНБОЮЛ",
|
||
)
|
||
|
||
|
||
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)
|
||
_apply_structured_group_rules(row)
|
||
_drop_structured_duplicates_of_one_line(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,
|
||
)
|
||
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:
|
||
"""Собирает структурную колонку адреса из её компонентов.
|
||
|
||
Однострочная версия адреса больше не отменяет структурную: раньше при
|
||
заполненной одной строке структурная колонка и все её компоненты
|
||
обнулялись, то есть отчёт показывал либо одно, либо другое. Заказчик
|
||
24.09.2026 потребовал выводить все значения из XML, а в файлах бывают
|
||
заполнены обе версии сразу.
|
||
"""
|
||
for rule in STRUCTURED_GROUP_RULES:
|
||
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 _drop_structured_duplicates_of_one_line(row: dict[int, str]) -> None:
|
||
"""Убирает структурную колонку, если она дословно повторяет однострочную.
|
||
|
||
Такое бывает, когда один и тот же тег XML подходит под кандидатов обеих
|
||
колонок: значение одно, а колонок две. Именно это, а не взаимное
|
||
исключение версий адреса, и должно отсекаться — если структура собрана из
|
||
своих компонентов и отличается от одной строки, обе колонки остаются
|
||
заполненными, потому что оба значения есть в файле.
|
||
"""
|
||
for rule in STRUCTURED_GROUP_RULES:
|
||
structured_value = _get(row, rule.structured_column).strip()
|
||
if not structured_value:
|
||
continue
|
||
if any(
|
||
_get(row, column_index).strip() == structured_value
|
||
for column_index in rule.one_line_columns
|
||
):
|
||
_set(row, rule.structured_column, "")
|
||
for one_line_column, structured_column in _ONE_LINE_STRUCTURED_PAIRS:
|
||
structured_value = _get(row, structured_column).strip()
|
||
if structured_value and _get(row, one_line_column).strip() == structured_value:
|
||
_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)
|
||
transfer_type = _get(row, COL_TRANSFER_TYPE)
|
||
operator_type = _get(row, COL_OPERATOR_TYPE)
|
||
|
||
# Заказчик 24.09.2026: в отчёт попадают все значения из XML. Поэтому здесь
|
||
# остались только правила, которые ничего не удаляют: сборка колонки из
|
||
# нескольких тегов и подстановка заглушек в пустые колонки. Правила,
|
||
# которые обнуляли колонку, если код операции, вид перевода или тип
|
||
# участника не входил в разрешённый список, убраны — они выбрасывали
|
||
# данные, реально присутствующие в файле.
|
||
_rule_extra_info(row, codes, operation_lookup, participant_lookup)
|
||
_rule_payer_account(row, transfer_type, codes)
|
||
_rule_correspondent_accounts(row, operator_type, transfer_type, codes)
|
||
_rule_recipient_account(row, transfer_type, codes)
|
||
_rule_card_without_employee(row)
|
||
|
||
# Это не фильтр по бизнес-логике, а обратная сторона требования "принимая
|
||
# во внимание, в каком блоке содержатся": без блока ЕИО/ЦП в записи
|
||
# значения в этих колонках могли появиться только из чужого блока.
|
||
if not has_eio_block:
|
||
_clear_columns(row, BLOCK_EIO)
|
||
if not has_cp_block:
|
||
_clear_columns(row, BLOCK_CP)
|
||
|
||
|
||
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_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_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 _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")
|