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\d+)\])?\." r"(?PКодДок|ИноеНаимДок|ДатаДок|НомДок|СодДок)$" ) _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, "") # Эталонный отчёт отделяет номер документа запятой, а не точкой: # "1, 13, , , 05/08/2026, ...". Расхождение затрагивало почти каждую # строку отчёта, поэтому приводим к виду эталона. 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")