RSHB/SFM
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

optimize 11.8x

Browse Source
This commit is contained in:
Raykov-MS 2026-08-11 15:38:11 +03:00
parent 510d668ab8
commit 5935eceadf
2 changed files with 240 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

312
app/pipeline/mapping.py
View File

@ -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(

54
tests/unit/test_mapping.py
View File

@ -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",