diff --git a/app/pipeline/mapping.py b/app/pipeline/mapping.py
index 6de30a4..4c53313 100644
--- a/app/pipeline/mapping.py
+++ b/app/pipeline/mapping.py
@@ -457,8 +457,11 @@ def build_fixed_row_by_index(
participant_base_lookup, participant_eio_lookup = _partition_participant_lookup(
participant_lookup
)
- merged_lookup = _build_payload_lookup(
- _merge_fields(operation_fields, participant_fields)
+ # Без полей участника объединённый payload совпадает с payload операции.
+ merged_lookup = (
+ operation_lookup
+ if not participant_fields
+ else _build_payload_lookup(_merge_fields(operation_fields, participant_fields))
)
has_eio_block = _lookup_has_any_prefixed_key(
participant_lookup,
@@ -481,36 +484,24 @@ def build_fixed_row_by_index(
"СообщОперКО.ИнформЧасть.СведКО.Операция.СведЦП.",
),
)
- for rule in REPORT_MAPPING_RULES:
+ # Значения прямых правил выбираем через обратный индекс «путь -> правила»
+ # (см. pick_rule_values_indexed): один проход по полям записи вместо
+ # перебора ~2 400 кандидатов на каждую строку отчёта.
+ values_by_rule: dict[int, str] = {}
+ for source, source_lookup in (
+ ("operation", operation_lookup),
+ ("participant_base", participant_base_lookup),
+ ("participant_eio", participant_eio_lookup),
+ ("merged", merged_lookup),
+ ):
+ values_by_rule.update(pick_rule_values_indexed(source_lookup, source))
+ for position, rule in enumerate(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,
- )
+ row[rule.column_index] = values_by_rule[position]
_set(
row,
COL_OPERATION_DOCUMENTS,
@@ -612,6 +603,8 @@ def _merge_fields(
def _normalize_indexed_path(path: str) -> str:
"""Убирает индексы повторов, сохраняя логическую структуру XML-пути."""
+ if "[" not in path:
+ return path
return _INDEXED_PATH_SEGMENT.sub("", path)
@@ -851,6 +844,164 @@ def _pick_structured_value(
return separator.join(values)
+# --- Индексная выборка значений прямых правил -------------------------------
+#
+# _pick_value / _pick_structured_value выше — эталонная семантика выбора
+# значения колонки: по порядку кандидатов, внутри кандидата по порядку проб
+# (raw -> суффикс normalized -> короткое имя / составные суффиксы). На каждую
+# строку отчёта это ~2 400 кандидатов × до 4 проб — главный потребитель CPU.
+#
+# Обратный индекс строится один раз при импорте: для каждого источника
+# (operation / participant_base / participant_eio / merged) и каждого вида
+# словаря _PayloadLookup — «ключ -> какие (правило, кандидат, проба) на него
+# претендуют». За строку мы проходим только по полям самой записи и для
+# каждого правила запоминаем лучшую (по порядку кандидата и пробы) находку.
+# Результат совпадает с эталоном; это проверяет tests/unit/test_mapping_index.py.
+
+LOOKUP_SOURCES: tuple[str, ...] = (
+ "operation",
+ "participant_base",
+ "participant_eio",
+ "merged",
+)
+
+# Порядок проб внутри кандидата — как в _pick_value.
+_PROBE_RAW = 0
+_PROBE_SUFFIX = 1
+_PROBE_DOTTED = 2 # только для правил без короткого поиска
+_PROBE_SHORT_RAW = 2 # только для правил с коротким поиском
+_PROBE_SHORT_FIRST = 3
+
+# (позиция правила, позиция кандидата, номер пробы)
+_IndexHit = tuple[int, int, int]
+# вид словаря -> ключ -> находки
+_ReverseIndex = dict[str, dict[str, tuple[_IndexHit, ...]]]
+
+
+def _rule_lookup_source(rule: MappingRule) -> str:
+ if rule.source == "participant":
+ return (
+ "participant_eio" if rule.column_index in BLOCK_EIO else "participant_base"
+ )
+ if rule.source == "operation":
+ return "operation"
+ return "merged"
+
+
+def _rule_allows_short_lookup(rule: MappingRule) -> bool:
+ return rule.allow_short_lookup or rule.column_index in BLOCK_EIO
+
+
+def _build_reverse_index(
+ rules: tuple[MappingRule, ...],
+) -> dict[str, tuple[_ReverseIndex, tuple[int, ...]]]:
+ """source -> (обратный индекс, позиции прямых правил этого источника)."""
+ index: dict[str, dict[str, dict[str, list[_IndexHit]]]] = {
+ source: {"values": {}, "suffix": {}, "dotted": {}, "short": {}}
+ for source in LOOKUP_SOURCES
+ }
+ rule_positions: dict[str, list[int]] = {source: [] for source in LOOKUP_SOURCES}
+ for rule_position, rule in enumerate(rules):
+ if not rule.allow_direct_mapping:
+ continue
+ source = _rule_lookup_source(rule)
+ rule_positions[source].append(rule_position)
+ by_kind = index[source]
+ allow_short_lookup = _rule_allows_short_lookup(rule)
+ for candidate_position, candidate in enumerate(rule.compiled_candidates):
+ if not allow_short_lookup and not candidate.strict_scoped:
+ continue
+ hit = (rule_position, candidate_position)
+ by_kind["values"].setdefault(candidate.raw, []).append((*hit, _PROBE_RAW))
+ by_kind["suffix"].setdefault(candidate.normalized, []).append(
+ (*hit, _PROBE_SUFFIX)
+ )
+ if not allow_short_lookup:
+ for suffix in candidate.dotted_suffixes:
+ by_kind["dotted"].setdefault(suffix, []).append(
+ (*hit, _PROBE_DOTTED)
+ )
+ continue
+ by_kind["values"].setdefault(candidate.short, []).append(
+ (*hit, _PROBE_SHORT_RAW)
+ )
+ by_kind["short"].setdefault(candidate.short, []).append(
+ (*hit, _PROBE_SHORT_FIRST)
+ )
+ return {
+ source: (
+ {
+ kind: {key: tuple(hits) for key, hits in keys.items()}
+ for kind, keys in index[source].items()
+ },
+ tuple(rule_positions[source]),
+ )
+ for source in LOOKUP_SOURCES
+ }
+
+
+_REVERSE_INDEX = _build_reverse_index(REPORT_MAPPING_RULES)
+
+
+def pick_rule_values_indexed(lookup: _PayloadLookup, source: str) -> dict[int, str]:
+ """Значения всех прямых правил источника: позиция правила -> значение.
+
+ Эквивалентно вызову _pick_value / _pick_structured_value для каждого
+ правила, но за один проход по полям записи.
+ """
+ index, rule_positions = _REVERSE_INDEX[source]
+ # правило -> кандидат -> ((проба, позиция поля), значение). Позиция поля
+ # различает несколько составных суффиксов одного кандидата — как в
+ # _pick_value, побеждает самое раннее поле записи.
+ best: dict[int, dict[int, tuple[tuple[int, int], str]]] = {}
+
+ def _offer(hits: tuple[_IndexHit, ...], field_position: int, value: str) -> None:
+ for rule_position, candidate_position, probe in hits:
+ candidates = best.setdefault(rule_position, {})
+ current = candidates.get(candidate_position)
+ rank = (probe, field_position)
+ if current is None or rank < current[0]:
+ candidates[candidate_position] = (rank, value)
+
+ values_index = index["values"]
+ for key, value in lookup.values.items():
+ if value and (hits := values_index.get(key)):
+ _offer(hits, 0, value)
+ suffix_index = index["suffix"]
+ for key, value in lookup.normalized_suffix_first.items():
+ if hits := suffix_index.get(key):
+ _offer(hits, 0, value)
+ dotted_index = index["dotted"]
+ if dotted_index:
+ for key, (field_position, value) in lookup.dotted_path_first.items():
+ if hits := dotted_index.get(key):
+ _offer(hits, field_position, value)
+ short_index = index["short"]
+ if short_index:
+ for key, value in lookup.raw_short_first.items():
+ if hits := short_index.get(key):
+ _offer(hits, 0, value)
+
+ result: dict[int, str] = {}
+ for rule_position in rule_positions:
+ candidates = best.get(rule_position)
+ if not candidates:
+ result[rule_position] = ""
+ continue
+ rule = REPORT_MAPPING_RULES[rule_position]
+ if rule.structured_value:
+ values: list[str] = []
+ for candidate_position in sorted(candidates):
+ value = candidates[candidate_position][1]
+ if value not in values:
+ values.append(value)
+ separator = " " if "фио" in rule.column_name.lower() else ", "
+ result[rule_position] = separator.join(values)
+ else:
+ result[rule_position] = candidates[min(candidates)][1]
+ return result
+
+
def _get(row: dict[int, str], column: int) -> str:
return str(row.get(column, "")).strip()
diff --git a/main.py b/main.py
index 2cbf404..be2ec27 100644
--- a/main.py
+++ b/main.py
@@ -1,4 +1,5 @@
import asyncio
+import shutil
import tempfile
from pathlib import Path
@@ -122,7 +123,12 @@ def _save_attachments(uploaded_files, target_dir: Path) -> list[Path]:
holder = target_dir / str(index)
holder.mkdir(parents=True, exist_ok=True)
path = holder / uploaded.name
- path.write_bytes(uploaded.getvalue())
+ # Потоковое копирование: UploadedFile уже целиком лежит в памяти
+ # Streamlit, а getvalue() делал бы вторую полную копию каждого
+ # вложения перед записью на диск.
+ uploaded.seek(0)
+ with path.open("wb") as target:
+ shutil.copyfileobj(uploaded, target)
paths.append(path)
return paths
diff --git a/scripts/bench_report.py b/scripts/bench_report.py
new file mode 100644
index 0000000..d59b5d5
--- /dev/null
+++ b/scripts/bench_report.py
@@ -0,0 +1,176 @@
+"""Воспроизводимый замер скорости формирования отчёта (без Streamlit и OpenCloud).
+
+Генерирует N синтетических XML-файлов по M операций с двумя участниками
+(ФЛИП с адресом и документом, ЮЛ с ЕИО и бенефициаром), прогоняет их через
+parse_xml_content + StreamingReportWriter и печатает время по стадиям и
+стоимость одной строки отчёта. Это тот же путь, что и в run_batch, минус
+распаковка вложений и загрузка результата.
+
+Пример:
+ python scripts/bench_report.py # 230 файлов × 20 операций
+ python scripts/bench_report.py --files 50 --ops 100
+ python scripts/bench_report.py --profile # топ функций по cProfile
+
+Ориентир (Apple M-серия, Python 3.13): ~0.25 мс на строку отчёта. На сервере
+с квотой 0.25 CPU (raisa_config.yml, профиль «Малый») ожидайте в 8–12 раз
+медленнее.
+"""
+
+from __future__ import annotations
+
+import argparse
+import cProfile
+import pstats
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+ sys.path.insert(0, str(ROOT))
+
+from app.pipeline.parser import parse_xml_content # noqa: E402
+from app.pipeline.report import ReportRow, StreamingReportWriter # noqa: E402
+
+
+def _operation(index: int) -> str:
+ return f"""
+<Операция>
+ <ИдентификаторЗаписи>OP{index}
+ <КодОперации>6001
+ <ДатаОперации>2026-06-16
+ <СуммаОперации>12345.67
+ <ВалютаОперации>643
+ <СуммаОперацииРуб>12345.67
+ <НазначениеПлатежа>Оплата по договору №{index} от 01.06.2026
+ <ХарактеристикаОперации>Перевод денежных средств клиента
+ <ОснованиеОп><КодДок>21<НомДок>{index}<ДатаДок>2026-06-01
+ <ОснованиеОп><КодДок>22<НомДок>{index}-b<ДатаДок>2026-06-02
+ <СведПереводДС>
+ <ВидПеревода>1
+ <СведБанкПлательщик><БИККО>044525111<НаимБанк>АО Банк<НомерСчета>40702810000000000001
+ <СведБанкПолучатель><БИККО>044525222<НаимБанк>ПАО Банк2<НомерСчета>40702810000000000002
+
+ <УчастникОп>
+ <ТипУчастника>2
+ <УчастникФЛИП>
+ <СведФЛИП>
+ <ФИО><Фамилия>Иванов<Имя>Иван<Отчество>Иванович
+ <ИННФЛИП>123456789012
+ <ДатаРождения>1980-01-01
+ <КодОКСМ>643
+ <СведДокУдЛичн><КодДок>21<СерияДок>4510<НомДок>123456<ДатаДок>2010-01-01<КемВыдан>ОУФМС России по г. Москве
+ <АдрРег Индекс="674500" КодОКСМ="643" КодСубъектаПоОКАТО="45" Район="р-н" Пункт="Москва г" Улица="Ленина ул" Дом="1" Корпус="2" Квартира="3"/>
+
+
+
+ <УчастникОп>
+ <ТипУчастника>1
+ <УчастникЮЛ>
+ <СведЮЛ>
+ <НаимЮЛ>ООО Ромашка<ИННЮЛ>7701234567<КППЮЛ>770101001<ОГРН>1027700000001
+ <АдрРегЮЛ Индекс="675520" КодОКСМ="643" КодСубъектаПоОКАТО="10" Район="Благовещенский р-н" Пункт="Чигири с" Улица="Зеленая ул" Дом="1"/>
+
+ <СведЕИО><ФЛЕИО><ФИО><Фамилия>Петров<Имя>Петр<ИННФЛИП>111111111111
+ <БенефициарЮЛ><ФЛЕИО><ФИО><Фамилия>Сидоров<Имя>Сидор<ИННФЛИП>222222222222
+
+
+"""
+
+
+def make_xml(operations: int) -> bytes:
+ body = "".join(_operation(index) for index in range(operations))
+ return f"""
+<СообщОперКО xmlns="urn:test">
+ <СлужЧасть><ИдФайл>ID1<ДатаФайла>2026-06-16
+ <ИнформЧасть>
+ <ИнфБанк><БИК>044525111<НаимКО>АО Банк
+ <СведКО>
+ <ИнфФилиал><НомерФилиала>01
+ {body}
+
+
+
+""".encode(
+ "utf-8"
+ )
+
+
+def run(files: int, operations: int) -> dict[str, float]:
+ xml = make_xml(operations)
+ started = time.perf_counter()
+ writer = StreamingReportWriter()
+ rows = 0
+ parse_seconds = 0.0
+ append_seconds = 0.0
+ for number in range(files):
+ file_name = f"SKO115FZ_01_044525111_20260616_{number:06d}.xml"
+ stage = time.perf_counter()
+ result = parse_xml_content(file_name=file_name, xml_content=xml)
+ parse_seconds += time.perf_counter() - stage
+ stage = time.perf_counter()
+ for row in result.rows:
+ writer.append_row(
+ ReportRow(
+ file_name=row.file_name,
+ record_id=row.record_id,
+ operation_index=row.operation_index,
+ operation_fields=row.operation_fields,
+ participant_fields=row.participant_fields,
+ is_continuation=row.is_continuation,
+ )
+ )
+ append_seconds += time.perf_counter() - stage
+ rows += len(result.rows)
+ with tempfile.TemporaryDirectory() as temp_dir:
+ stage = time.perf_counter()
+ report_path = Path(temp_dir) / "report.xlsx"
+ writer.save(report_path)
+ save_seconds = time.perf_counter() - stage
+ report_size_mb = report_path.stat().st_size / 2**20
+ return {
+ "files": files,
+ "operations_per_file": operations,
+ "xml_kb": len(xml) / 1024,
+ "rows": rows,
+ "total_seconds": time.perf_counter() - started,
+ "parse_seconds": parse_seconds,
+ "append_seconds": append_seconds,
+ "save_seconds": save_seconds,
+ "ms_per_row": append_seconds / rows * 1000 if rows else 0.0,
+ "report_mb": report_size_mb,
+ }
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+ parser.add_argument("--files", type=int, default=230)
+ parser.add_argument("--ops", type=int, default=20, help="операций в файле")
+ parser.add_argument("--profile", action="store_true", help="печатать cProfile")
+ args = parser.parse_args()
+
+ profiler = cProfile.Profile() if args.profile else None
+ if profiler is not None:
+ profiler.enable()
+ stats = run(args.files, args.ops)
+ if profiler is not None:
+ profiler.disable()
+
+ print(
+ f"files={stats['files']} ops/file={stats['operations_per_file']} "
+ f"xml={stats['xml_kb']:.0f} KB rows={stats['rows']}"
+ )
+ print(
+ f"total={stats['total_seconds']:.1f}s parse={stats['parse_seconds']:.1f}s "
+ f"append={stats['append_seconds']:.1f}s save={stats['save_seconds']:.1f}s "
+ f"report={stats['report_mb']:.1f} MB"
+ )
+ print(f"per row (parse excluded): {stats['ms_per_row']:.3f} ms")
+ if profiler is not None:
+ pstats.Stats(profiler).sort_stats("cumulative").print_stats(25)
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests/smoke/test_batch_progress.py b/tests/smoke/test_batch_progress.py
new file mode 100644
index 0000000..abf2cb3
--- /dev/null
+++ b/tests/smoke/test_batch_progress.py
@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import zipfile
+from pathlib import Path
+
+from app.pipeline import service
+from app.pipeline.service import run_batch
+from tests.smoke.test_batch_smoke import _patch_opencloud, _valid_xml
+
+
+def test_progress_callback_is_called_once_per_xml_file(
+ tmp_path: Path, monkeypatch
+) -> None:
+ _patch_opencloud(monkeypatch)
+ zip_path = tmp_path / "batch.zip"
+ names = [f"SKO115FZ_01_123456789_20260616_X0000{i}.xml" for i in range(1, 4)]
+ with zipfile.ZipFile(zip_path, "w") as archive:
+ for name in names:
+ archive.writestr(name, _valid_xml())
+ loose = tmp_path / "SKO115FZ_01_123456789_20260616_X00009.xml"
+ loose.write_text(_valid_xml(), encoding="utf-8")
+
+ calls: list[tuple[int, int, str]] = []
+ result = run_batch(
+ input_paths=[zip_path, loose],
+ launcher="tester",
+ opencloud_username="test-user",
+ opencloud_access_token="token-123",
+ progress_callback=lambda done, total, name: calls.append((done, total, name)),
+ )
+
+ assert result.status == "success"
+ assert [done for done, _, _ in calls] == [1, 2, 3, 4]
+ assert {total for _, total, _ in calls} == {4}
+ assert sorted(name for _, _, name in calls) == sorted([*names, loose.name])
+
+
+def test_progress_callback_is_called_for_unreadable_file_too(
+ tmp_path: Path, monkeypatch
+) -> None:
+ _patch_opencloud(monkeypatch)
+ loose = tmp_path / "SKO115FZ_01_123456789_20260616_X00001.xml"
+ loose.write_text(_valid_xml(), encoding="utf-8")
+
+ def _boom(self: Path) -> bytes:
+ raise OSError("disk gone")
+
+ monkeypatch.setattr(service.Path, "read_bytes", _boom)
+ calls: list[tuple[int, int, str]] = []
+ result = run_batch(
+ input_paths=[loose],
+ launcher="tester",
+ opencloud_username="test-user",
+ opencloud_access_token="token-123",
+ progress_callback=lambda done, total, name: calls.append((done, total, name)),
+ )
+
+ assert result.status == "error"
+ assert calls == [(1, 1, loose.name)]
diff --git a/tests/unit/test_main_attachments.py b/tests/unit/test_main_attachments.py
new file mode 100644
index 0000000..a05c37e
--- /dev/null
+++ b/tests/unit/test_main_attachments.py
@@ -0,0 +1,50 @@
+"""Сохранение вложений Streamlit на диск не должно удваивать их в памяти.
+
+UploadedFile — это BytesIO: Streamlit уже держит всё содержимое вложения в
+памяти. getvalue() делает ещё одну полную копию каждого файла перед записью
+на диск; при 230 вложениях в контейнере на 1 ГБ это лишняя пиковая нагрузка.
+Копировать нужно потоково, через файловый интерфейс, не трогая getvalue().
+"""
+
+from __future__ import annotations
+
+import io
+from pathlib import Path
+
+from main import _save_attachments
+
+
+class _UploadWithoutGetValue(io.BytesIO):
+ """Вложение, у которого есть только файловый интерфейс и имя."""
+
+ def __init__(self, name: str, payload: bytes) -> None:
+ super().__init__(payload)
+ self.name = name
+
+ def getvalue(self) -> bytes: # type: ignore[override]
+ raise AssertionError("getvalue() копирует всё вложение в память")
+
+
+def test_save_attachments_streams_without_getvalue(tmp_path: Path) -> None:
+ uploads = [
+ _UploadWithoutGetValue("a.xml", b""),
+ _UploadWithoutGetValue("a.xml", b"" * 1000),
+ ]
+
+ paths = _save_attachments(uploads, tmp_path)
+
+ assert [path.name for path in paths] == ["a.xml", "a.xml"]
+ assert paths[0] != paths[1]
+ assert paths[0].read_bytes() == b""
+ assert paths[1].read_bytes() == b"" * 1000
+
+
+def test_save_attachments_copies_from_start_even_if_upload_was_read(
+ tmp_path: Path,
+) -> None:
+ upload = _UploadWithoutGetValue("f.xml", b"0123456789")
+ upload.read(4)
+
+ (path,) = _save_attachments([upload], tmp_path)
+
+ assert path.read_bytes() == b"0123456789"
diff --git a/tests/unit/test_mapping_index.py b/tests/unit/test_mapping_index.py
new file mode 100644
index 0000000..ba08d90
--- /dev/null
+++ b/tests/unit/test_mapping_index.py
@@ -0,0 +1,187 @@
+"""Индексная выборка значений колонок должна давать то же, что _pick_value.
+
+_pick_value / _pick_structured_value — эталонная (медленная) семантика:
+на каждую строку отчёта перебираются все кандидаты каждого правила. Быстрый
+путь строит обратный индекс «путь -> правила» один раз и за строку проходит
+только по полям самой записи. Тест сверяет оба пути на случайных и реальных
+payload по всем правилам и всем источникам (operation / participant / merged).
+"""
+
+from __future__ import annotations
+
+import random
+
+from app.pipeline import mapping
+from app.pipeline.column_constants import BLOCK_EIO
+from app.pipeline.mapping import (
+ REPORT_MAPPING_RULES,
+ _build_payload_lookup,
+ _merge_fields,
+ _partition_participant_lookup,
+ _pick_structured_value,
+ _pick_value,
+ _rule_lookup_source,
+ pick_rule_values_indexed,
+)
+from app.pipeline.parser import parse_xml_content
+
+_PREFIXES = ("", "УчастникОп.", "СообщОперКО.ИнформЧасть.СведКО.Операция.")
+
+
+def _candidate_pool() -> list[str]:
+ pool: set[str] = set()
+ for rule in REPORT_MAPPING_RULES:
+ for candidate in rule.compiled_candidates:
+ pool.add(candidate.raw)
+ pool.add(candidate.short)
+ parts = candidate.raw.split(".")
+ for start in range(len(parts)):
+ pool.add(".".join(parts[start:]))
+ return sorted(pool)
+
+
+def _random_payload(rng: random.Random, pool: list[str]) -> dict[str, str]:
+ payload: dict[str, str] = {}
+ for _ in range(rng.randint(1, 40)):
+ key = rng.choice(pool)
+ prefix = rng.choice(_PREFIXES)
+ if rng.random() < 0.25:
+ parts = key.split(".")
+ position = rng.randrange(len(parts))
+ parts[position] = f"{parts[position]}[{rng.randint(0, 2)}]"
+ key = ".".join(parts)
+ key = f"{prefix}{key}"
+ value = "" if rng.random() < 0.1 else f"v{rng.randint(1, 5)}"
+ payload[key] = value
+ if rng.random() < 0.3:
+ payload[f"Шум{rng.randint(1, 9)}.Поле"] = "шум"
+ return payload
+
+
+def _reference_values(lookup, source: str) -> dict[int, str]:
+ expected: dict[int, str] = {}
+ for position, rule in enumerate(REPORT_MAPPING_RULES):
+ if not rule.allow_direct_mapping or _rule_lookup_source(rule) != source:
+ continue
+ allow_short_lookup = rule.allow_short_lookup or rule.column_index in BLOCK_EIO
+ if rule.structured_value:
+ expected[position] = _pick_structured_value(
+ lookup=lookup,
+ candidates=rule.compiled_candidates,
+ join_with_space="фио" in rule.column_name.lower(),
+ allow_short_lookup=allow_short_lookup,
+ )
+ else:
+ expected[position] = _pick_value(
+ lookup,
+ rule.compiled_candidates,
+ allow_short_lookup=allow_short_lookup,
+ )
+ return expected
+
+
+def _lookups_for(operation_fields, participant_fields):
+ operation_lookup = _build_payload_lookup(operation_fields)
+ participant_lookup = _build_payload_lookup(participant_fields)
+ base_lookup, eio_lookup = _partition_participant_lookup(participant_lookup)
+ merged_lookup = _build_payload_lookup(
+ _merge_fields(operation_fields, participant_fields)
+ )
+ return {
+ "operation": operation_lookup,
+ "participant_base": base_lookup,
+ "participant_eio": eio_lookup,
+ "merged": merged_lookup,
+ }
+
+
+def _assert_same_values(operation_fields, participant_fields, label: str) -> None:
+ for source, lookup in _lookups_for(operation_fields, participant_fields).items():
+ expected = _reference_values(lookup, source)
+ actual = pick_rule_values_indexed(lookup, source)
+ for position, expected_value in expected.items():
+ rule = REPORT_MAPPING_RULES[position]
+ assert actual.get(position, "") == expected_value, (
+ f"{label}: source={source}, колонка {rule.column_index} "
+ f"({rule.column_id}): ожидалось {expected_value!r}, "
+ f"получено {actual.get(position, '')!r}"
+ )
+ extra = set(actual) - set(expected)
+ assert not extra, f"{label}: лишние правила в индексной выборке: {extra}"
+
+
+def test_indexed_values_match_pick_value_on_random_payloads() -> None:
+ rng = random.Random(20261009)
+ pool = _candidate_pool()
+ for sample in range(400):
+ operation_fields = _random_payload(rng, pool)
+ participant_fields = _random_payload(rng, pool)
+ _assert_same_values(operation_fields, participant_fields, f"sample {sample}")
+
+
+def test_indexed_values_match_pick_value_on_parsed_xml() -> None:
+ xml = """
+<СообщОперКО xmlns="urn:test">
+ <СлужЧасть><ИдФайл>ID1<ДатаФайла>2026-06-16
+ <ИнформЧасть>
+ <ИнфБанк><БИК>044525111<НаимКО>АО Банк
+ <СведКО>
+ <ИнфФилиал><НомерФилиала>01
+ <Операция>
+ <ИдентификаторЗаписи>OP1
+ <КодОперации>6001
+ <ДатаОперации>2026-06-16
+ <СуммаОперации>12345.67
+ <ВалютаОперации>643
+ <НазначениеПлатежа>Оплата по договору
+ <ОснованиеОп><КодДок>21<НомДок>1<ДатаДок>2026-06-01
+ <ОснованиеОп><КодДок>22<НомДок>2<ДатаДок>2026-06-02
+ <СведПереводДС>
+ <ВидПеревода>1
+ <СведБанкПлательщик><БИККО>044525111<НаимБанк>АО Банк<НомерСчета>40702810000000000001
+ <СведБанкПолучатель><БИККО>044525222<НаимБанк>ПАО Банк2<НомерСчета>40702810000000000002
+
+ <УчастникОп>
+ <ТипУчастника>2
+ <УчастникФЛИП>
+ <СведФЛИП>
+ <ФИО><Фамилия>Иванов<Имя>Иван<Отчество>Иванович
+ <ИННФЛИП>123456789012
+ <ДатаРождения>1980-01-01
+ <КодОКСМ>643
+ <СведДокУдЛичн><КодДок>21<СерияДок>4510<НомДок>123456<ДатаДок>2010-01-01<КемВыдан>ОУФМС
+ <АдрРег Индекс="674500" КодОКСМ="643" КодСубъектаПоОКАТО="45" Пункт="Москва г" Улица="Ленина ул" Дом="1" Корпус="2" Квартира="3"/>
+
+
+
+ <УчастникОп>
+ <ТипУчастника>1
+ <УчастникЮЛ>
+ <СведЮЛ>
+ <НаимЮЛ>ООО Ромашка<ИННЮЛ>7701234567<КППЮЛ>770101001<ОГРН>1027700000001
+ <АдрРегЮЛ Индекс="675520" КодОКСМ="643" КодСубъектаПоОКАТО="10" Район="Благовещенский р-н" Пункт="Чигири с" Улица="Зеленая ул" Дом="1"/>
+
+ <СведЕИО><ФЛЕИО><ФИО><Фамилия>Петров<Имя>Петр<ИННФЛИП>111111111111
+ <БенефициарЮЛ><ФЛЕИО><ФИО><Фамилия>Сидоров<Имя>Сидор<ИННФЛИП>222222222222
+
+
+
+
+
+
+"""
+ result = parse_xml_content("SKO115FZ_01_044525111_20260616_000001.xml", xml)
+ assert not result.is_fatal
+ assert result.rows
+ for row in result.rows:
+ _assert_same_values(
+ row.operation_fields,
+ row.participant_fields,
+ f"row {row.operation_index}/{row.is_continuation}",
+ )
+
+
+def test_every_direct_rule_has_exactly_one_lookup_source() -> None:
+ for rule in REPORT_MAPPING_RULES:
+ source = _rule_lookup_source(rule)
+ assert source in mapping.LOOKUP_SOURCES, (rule.column_id, source)
diff --git a/tests/unit/test_mapping_lookup_cache.py b/tests/unit/test_mapping_lookup_cache.py
new file mode 100644
index 0000000..3a21263
--- /dev/null
+++ b/tests/unit/test_mapping_lookup_cache.py
@@ -0,0 +1,51 @@
+"""Переиспользование lookup операции между строками одной операции.
+
+Все строки одной операции (участники, continuation) приходят с одним и тем
+же словарём operation_fields. Его lookup можно не пересобирать, но только
+если результат не зависит от того, кешировался словарь или нет: изменение
+содержимого словаря между вызовами обязано отражаться в строке.
+"""
+
+from __future__ import annotations
+
+from app.pipeline.mapping import build_fixed_row
+
+_OPERATION_CODE_COLUMN = "Код вида операции"
+
+
+def _row(operation_fields: dict[str, str], participant_fields: dict[str, str]):
+ return build_fixed_row(
+ file_name="f.xml",
+ record_id="R1",
+ operation_index=1,
+ operation_fields=operation_fields,
+ participant_fields=participant_fields,
+ )
+
+
+def test_same_operation_dict_across_rows_gives_same_values_as_fresh_dict() -> None:
+ shared = {"КодОперации": "5007", "СумОперации": "10.00"}
+ first = _row(shared, {"ИННФЛ": "111111111111", "Тип": "01"})
+ second = _row(shared, {"ИННФЛ": "222222222222", "Тип": "01"})
+ fresh = _row(dict(shared), {"ИННФЛ": "222222222222", "Тип": "01"})
+
+ assert first[_OPERATION_CODE_COLUMN] == "5007"
+ assert second == fresh
+
+
+def test_mutating_operation_dict_between_rows_is_reflected() -> None:
+ operation = {"КодОперации": "5007"}
+ assert _row(operation, {})[_OPERATION_CODE_COLUMN] == "5007"
+
+ operation["КодОперации"] = "6001"
+ assert _row(operation, {})[_OPERATION_CODE_COLUMN] == "6001"
+
+ del operation["КодОперации"]
+ assert _row(operation, {})[_OPERATION_CODE_COLUMN] == ""
+
+
+def test_empty_participant_row_equals_row_built_without_participant_data() -> None:
+ operation = {"КодОперации": "5007", "ИННФЛ": "333333333333"}
+ with_empty = _row(operation, {})
+ fresh = _row(dict(operation), dict())
+ assert with_empty == fresh