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

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
 

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"""<?xml version="1.0" encoding="UTF-8"?>
+<СообщОперКО 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())

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)]

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"<a/>"),
+        _UploadWithoutGetValue("a.xml", b"<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"<a/>"
+    assert paths[1].read_bytes() == b"<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"

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 = """<?xml version="1.0" encoding="UTF-8"?>
+<СообщОперКО 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)

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