69fd329f78
Verified our report against the Комита reference export by joining on the record id. Our report_20260923_5.xlsx covers the 114 files of 30.01.2023 and all 5266 of its records are inside the reference period, so the two can be compared row by row. Fifteen of the twenty-one shared columns already match exactly. The largest remaining gap was the separator in the code lists: the reference prints "1101,1106,1499" while the parser joins repeated tags with "; ", which put 336 of 5266 records out of step. Only the two code-list columns are brought to the reference's separator; "; " stays in the free-text columns, where it reads better and where the reference has no equivalent. What the comparison also showed, for the record: - The empty branch БИК and number on head-office records are correct. Those files carry no ИнфФилиал block at all and the reference leaves the cells empty too. - Of the 105 address differences, 94 are the column 141 leak fixed in cceee02, and the remaining 11 are values that contain a comma, such as Пункт = "Южно-Сахалинск г, Хомутово пл.р-н", which the reference cannot represent in one comma-joined cell. - Column 121 shows РегнНомер for people in private practice, where the reference's ОГРН column stays empty. That is the mapping table's fourth source for the column, not a defect: 3 records out of 13168.
991 lines
38 KiB
Python
991 lines
38 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,
|
||
COL_UNUSUAL_CODES,
|
||
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_code_list_separator(row)
|
||
_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)
|
||
|
||
|
||
# Перечисления кодов эталонный отчёт АРМ «Комита» печатает через запятую без
|
||
# пробела ("1101,1106,1499"), а парсер склеивает повторяющиеся теги через
|
||
# "; " (см. _append_value в parser.py). Приводим к виду эталона только
|
||
# колонки-перечисления кодов: в текстовых полях "; " читается лучше.
|
||
_CODE_LIST_COLUMNS = (COL_EXTRA_CODES, COL_UNUSUAL_CODES)
|
||
_CODE_LIST_SEPARATOR = ","
|
||
|
||
|
||
def _rule_code_list_separator(row: dict[int, str]) -> None:
|
||
for column in _CODE_LIST_COLUMNS:
|
||
value = _get(row, column)
|
||
if "; " not in value:
|
||
continue
|
||
codes = [part.strip() for part in value.split(";") if part.strip()]
|
||
_set(row, column, _CODE_LIST_SEPARATOR.join(codes))
|
||
|
||
|
||
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")
|