From 5935eceadfb7a529075cbf715f3a37ca7d2ad8e4 Mon Sep 17 00:00:00 2001 From: Raykov-MS Date: Tue, 11 Aug 2026 15:38:11 +0300 Subject: [PATCH] optimize 11.8x --- app/pipeline/mapping.py | 312 ++++++++++++++++++++++--------------- tests/unit/test_mapping.py | 54 +++++++ 2 files changed, 240 insertions(+), 126 deletions(-) diff --git a/app/pipeline/mapping.py b/app/pipeline/mapping.py index 6458d23..43e71e6 100644 --- a/app/pipeline/mapping.py +++ b/app/pipeline/mapping.py @@ -82,18 +82,42 @@ from .column_constants import ( from .column_registry import COLUMNS, ColumnMeta +@dataclass(frozen=True) +class _CompiledCandidate: + raw: str + normalized: str + short: str + + @dataclass(frozen=True) class MappingRule: column_index: int column_id: str column_name: str - candidate_keys: tuple[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 @@ -128,6 +152,17 @@ def _bank_bik_candidate_keys(bank_block: str) -> tuple[str, ...]: return tuple(candidates) +def _compile_candidates(candidates: tuple[str, ...]) -> tuple[_CompiledCandidate, ...]: + return tuple( + _CompiledCandidate( + raw=candidate, + normalized=_INDEXED_PATH_SEGMENT.sub("", candidate), + short=candidate.rsplit(".", maxsplit=1)[-1], + ) + for candidate in candidates + ) + + def _build_rules(mapping_data: tuple[ColumnMeta, ...]) -> tuple[MappingRule, ...]: rules: list[MappingRule] = [] for meta in mapping_data: @@ -212,12 +247,13 @@ def _build_rules(mapping_data: tuple[ColumnMeta, ...]) -> tuple[MappingRule, ... 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, - candidate_keys=tuple(direct_candidates), + compiled_candidates=_compile_candidates(candidate_keys), source=meta.source_scope, allow_direct_mapping=meta.allow_direct_mapping, allow_short_lookup=meta.allow_short_lookup, @@ -321,11 +357,16 @@ def build_fixed_row_by_index( COL_FILE_NAME: file_name, COL_RECORD_ID: record_id, } - merged = _merge_fields(operation_fields, participant_fields) - merged_suffix_index = _build_suffix_index(merged) - operation_suffix_index = _build_suffix_index(operation_fields) - has_eio_block = _has_any_prefixed_key( - participant_fields, + 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, ( "УчастникЮЛ.СведЕИО.", "УчастникЮЛ.БенефициарЮЛ.", @@ -338,8 +379,8 @@ def build_fixed_row_by_index( "БенефициарИНБОЮЛ.", ), ) - has_cp_block = _has_any_prefixed_key( - operation_fields, + has_cp_block = _lookup_has_any_prefixed_key( + operation_lookup, ( "СведЦП.", "СообщОперКО.ИнформЧасть.СведКО.Операция.СведЦП.", @@ -352,34 +393,30 @@ def build_fixed_row_by_index( row[rule.column_index] = "" continue if rule.source == "participant": - source_payload = _scope_participant_payload( - participant_fields, - eio_only=rule.column_index in BLOCK_EIO, + source_lookup = ( + participant_eio_lookup + if rule.column_index in BLOCK_EIO + else participant_base_lookup ) - suffix_index = _build_suffix_index(source_payload) elif rule.source == "operation": - source_payload = operation_fields - suffix_index = operation_suffix_index + source_lookup = operation_lookup else: - source_payload = merged - suffix_index = merged_suffix_index + source_lookup = merged_lookup if rule.structured_value: row[rule.column_index] = _pick_structured_value( - source_payload=source_payload, - suffix_index=suffix_index, - candidates=rule.candidate_keys, + lookup=source_lookup, + candidates=rule.compiled_candidates, join_with_space="фио" in rule.column_name.lower(), allow_short_lookup=rule.allow_short_lookup, ) else: row[rule.column_index] = _pick_value( - source_payload, - suffix_index, - rule.candidate_keys, + source_lookup, + rule.compiled_candidates, allow_short_lookup=rule.allow_short_lookup, ) if not is_continuation: - _apply_participant_identity_rules(row, participant_fields) + _apply_participant_identity_rules(row, participant_lookup) # Сначала убираем запрещённые однострочные адреса, чтобы структурная # группа не очистила их компоненты до проверки резидентства. _rule_address_one_line(row) @@ -389,8 +426,8 @@ def build_fixed_row_by_index( row, has_eio_block=has_eio_block, has_cp_block=has_cp_block, - operation_fields=operation_fields, - participant_fields=participant_fields, + operation_lookup=operation_lookup, + participant_lookup=participant_lookup, ) else: _apply_continuation_validation_rules(row) @@ -455,149 +492,172 @@ def _merge_fields( return merged -def _has_payload_block(payload: dict[str, str], block_name: str) -> bool: - prefix = f"{block_name}." - nested_marker = f".{block_name}." - return any( - (normalized := _normalize_indexed_path(key)).startswith(prefix) - or nested_marker in normalized - for key in payload +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 _pick_scoped_payload_value(payload: dict[str, str], scoped_path: str) -> str: - nested_suffix = f".{scoped_path}" - for key, value in payload.items(): - normalized = _normalize_indexed_path(key) - if value and (normalized == scoped_path or normalized.endswith(nested_suffix)): - return str(value).strip() - return "" +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 _pick_direct_payload_value(payload: dict[str, str], field_name: str) -> str: +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-поиска.""" - for key, value in payload.items(): - if value and _normalize_indexed_path(key) == field_name: - return str(value).strip() - return "" + value = lookup.normalized_exact_first.get(field_name, "") + return str(value).strip() if value else "" def _apply_participant_identity_rules( - row: dict[int, str], participant_fields: dict[str, str] + row: dict[int, str], participant_lookup: _PayloadLookup ) -> None: participant_type = _normalize_code(_get(row, COL_PARTICIPANT_TYPE)) - has_legal_entity = _has_payload_block(participant_fields, "УчастникЮЛ") - has_physical_person = _has_payload_block(participant_fields, "УчастникФЛИП") - has_foreign_structure = _has_payload_block(participant_fields, "УчастникИНБОЮЛ") + 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 = _pick_scoped_payload_value( - participant_fields, "УчастникФЛИП.ИдентификацияФЛ" + 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 = _pick_scoped_payload_value( - participant_fields, "УчастникЮЛ.СведЮЛ.КППЮЛ" - ) + value = _lookup_pick_scoped_value(participant_lookup, "УчастникЮЛ.СведЮЛ.КППЮЛ") elif has_foreign_structure: - value = _pick_scoped_payload_value( - participant_fields, + value = _lookup_pick_scoped_value( + participant_lookup, "УчастникИНБОЮЛ.СведИНБОЮЛ.ПризнакОргФормаИНБОЮЛ", ) elif has_physical_person: - value = _pick_scoped_payload_value( - participant_fields, "УчастникФЛИП.ИдентификацияФЛ" + value = _lookup_pick_scoped_value( + participant_lookup, "УчастникФЛИП.ИдентификацияФЛ" ) else: value = "" _set(row, COL_KPP, value) -def _has_any_prefixed_key(payload: dict[str, str], prefixes: tuple[str, ...]) -> bool: +def _lookup_has_any_prefixed_key( + lookup: _PayloadLookup, prefixes: tuple[str, ...] +) -> bool: return any( - any(_normalize_indexed_path(key).startswith(prefix) for prefix in prefixes) - for key in payload + any(normalized_path.startswith(prefix) for prefix in prefixes) + for normalized_path in lookup.normalized_paths ) -def _normalize_indexed_path(path: str) -> str: - """Убирает индексы повторов, сохраняя логическую структуру XML-пути.""" - return _INDEXED_PATH_SEGMENT.sub("", path) - - -def _is_eio_path(path: str) -> bool: - normalized = _normalize_indexed_path(path) - return ".СведИНБОЮЛ.Учредитель." in f".{normalized}." or any( - f".{block_name}." in f".{normalized}." for block_name in _EIO_BLOCK_NAMES - ) - - -def _scope_participant_payload( - payload: dict[str, str], - *, - eio_only: bool, -) -> dict[str, str]: - """Изолирует поля участника от вложенного блока ЕИО/бенефициара.""" - return { - key: value for key, value in payload.items() if _is_eio_path(key) is eio_only - } - - -def _build_suffix_index(payload: dict[str, str]) -> dict[str, str]: - index: dict[str, str] = {} - for key, value in payload.items(): - if not value: - continue - short = key.rsplit(".", maxsplit=1)[-1] - index.setdefault(short, value) - return index - - def _pick_value( - payload: dict[str, str], - suffix_index: dict[str, str], - candidates: tuple[str, ...], + lookup: _PayloadLookup, + candidates: tuple[_CompiledCandidate, ...], *, allow_short_lookup: bool = True, ) -> str: - for key in candidates: - if value := payload.get(key): + for candidate in candidates: + if value := lookup.values.get(candidate.raw): + return value + if value := lookup.normalized_suffix_first.get(candidate.normalized): return value - normalized_key = _normalize_indexed_path(key) - for payload_key, value in payload.items(): - normalized_payload_key = _normalize_indexed_path(payload_key) - if value and ( - normalized_payload_key == normalized_key - or normalized_payload_key.endswith(f".{normalized_key}") - ): - return value if not allow_short_lookup: continue - short = key.rsplit(".", maxsplit=1)[-1] - if value := payload.get(short): + if value := lookup.values.get(candidate.short): + return value + if value := lookup.raw_short_first.get(candidate.short): return value - if short in suffix_index: - return suffix_index[short] return "" def _pick_structured_value( *, - source_payload: dict[str, str], - suffix_index: dict[str, str], - candidates: tuple[str, ...], + lookup: _PayloadLookup, + candidates: tuple[_CompiledCandidate, ...], join_with_space: bool, allow_short_lookup: bool = True, ) -> str: values: list[str] = [] - for key in candidates: + for candidate in candidates: value = _pick_value( - source_payload, - suffix_index, - (key,), + lookup, + (candidate,), allow_short_lookup=allow_short_lookup, ) if value and value not in values: @@ -643,8 +703,8 @@ def _apply_business_rules( *, has_eio_block: bool, has_cp_block: bool, - operation_fields: dict[str, str], - participant_fields: dict[str, str], + operation_lookup: _PayloadLookup, + participant_lookup: _PayloadLookup, ) -> None: codes = _operation_codes(row) operation_sign = _get(row, COL_OPERATION_SIGN) @@ -658,7 +718,7 @@ def _apply_business_rules( _rule_unusual_codes(row, codes) _rule_digital_rights_currency(row, operation_sign) _rule_sale_currency(row) - _rule_extra_info(row, codes, operation_fields, participant_fields) + _rule_extra_info(row, codes, operation_lookup, participant_lookup) _rule_suspicious_activity(row, codes) _rule_metal_name(row) _rule_item_type(row, codes) @@ -740,15 +800,15 @@ def _rule_sale_currency(row: dict[int, str]) -> None: def _rule_extra_info( row: dict[int, str], codes: set[str], - operation_fields: dict[str, str], - participant_fields: dict[str, str], + operation_lookup: _PayloadLookup, + participant_lookup: _PayloadLookup, ) -> None: """Правило 34: объединяет допустимые комментарии операции и участника.""" - operation_comment = _pick_direct_payload_value(operation_fields, "Коммент") + operation_comment = _lookup_pick_direct_value(operation_lookup, "Коммент") participant_comment = "" if codes & EXPORT_SUBSIDIARY_COMMENT_CODES: - participant_comment = _pick_direct_payload_value( - participant_fields, "КомментУчастник" + participant_comment = _lookup_pick_direct_value( + participant_lookup, "КомментУчастник" ) values = list( diff --git a/tests/unit/test_mapping.py b/tests/unit/test_mapping.py index 4e0177f..aff03f9 100644 --- a/tests/unit/test_mapping.py +++ b/tests/unit/test_mapping.py @@ -1,5 +1,6 @@ from __future__ import annotations +from app.pipeline import mapping from app.pipeline.mapping import ( FIXED_REPORT_COLUMNS, _extract_tag_candidate_keys, @@ -154,6 +155,38 @@ def test_build_fixed_row_skips_empty_suffix_values_in_index() -> None: assert row[_column_with("ИНН")] == "222222222222" +def test_mapping_normalizes_each_payload_path_only_constant_times(monkeypatch) -> None: + calls = 0 + original = mapping._normalize_indexed_path + + def counted(path: str) -> str: + nonlocal calls + calls += 1 + return original(path) + + monkeypatch.setattr(mapping, "_normalize_indexed_path", counted) + + operation_fields = { + "НомерОперация": "OP1", + "КодОперации": "6001", + } + participant_fields = { + "УчастникЮЛ.СведЮЛ.НаимЮЛ": "АО СИРИУС", + "УчастникЮЛ.СведЮЛ.ИННЮЛ": "2820000210", + "УчастникЮЛ.СведЕИО[0].ФЛЕИО.ФИОФЛИП.Фам": "Музалёв", + "УчастникЮЛ.СведЕИО[0].ФЛЕИО.ИННФЛИП": "410116812922", + } + build_fixed_row_by_index( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields=operation_fields, + participant_fields=participant_fields, + ) + + assert calls == 2 * (len(operation_fields) + len(participant_fields)) + + def test_structured_address_is_built_from_right_columns() -> None: row = build_fixed_row( file_name="f.xml", @@ -870,6 +903,27 @@ def test_participant_and_beneficiary_inn_are_scoped_regardless_of_field_order() assert "111111111111" not in row[157] +def test_legal_entity_and_eio_representative_inn_remain_strictly_scoped() -> None: + row = build_fixed_row_by_index( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "УчастникЮЛ.СведЮЛ.НаимЮЛ": "АО СИРИУС", + "УчастникЮЛ.СведЮЛ.ИННЮЛ": "2820000210", + "УчастникЮЛ.СведЕИО[0].ФЛЕИО.ФИОФЛИП.Фам": "Музалёв", + "УчастникЮЛ.СведЕИО[0].ФЛЕИО.ФИОФЛИП.Имя": "Александр", + "УчастникЮЛ.СведЕИО[0].ФЛЕИО.ФИОФЛИП.Отч": "Сергеевич", + "УчастникЮЛ.СведЕИО[0].ФЛЕИО.ИННФЛИП": "410116812922", + }, + ) + + assert row[120] == "2820000210" + assert row[157] == "Музалёв Александр Сергеевич" + assert row[162] == "410116812922" + + def test_flip_beneficiary_is_scoped_to_eio_columns() -> None: row = build_fixed_row_by_index( file_name="f.xml",