From 6668858b6f30b4ba4586c3b64267e96d128c5fde Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Fri, 9 Oct 2026 19:14:43 +0400 Subject: [PATCH 1/2] Pick report column values through a reverse path index build_fixed_row_by_index evaluated ~2 400 candidate paths per report row (7 million _pick_value calls per 13 800 rows). A reverse index "path -> (rule, candidate, probe)" is built once at import; per row we walk only the record's own fields and keep the best hit per rule, with the same candidate/probe ordering as _pick_value. Differential test compares the index with _pick_value on random and parsed payloads. Also: merged lookup reuses the operation lookup when the participant payload is empty, and index-free paths skip the regex normalisation. scripts/bench_report.py reproduces the measurement (0.42 -> 0.26 ms/row). Refs #1 Co-Authored-By: Claude Fable 5.1 --- app/pipeline/mapping.py | 205 ++++++++++++++++++++---- scripts/bench_report.py | 176 ++++++++++++++++++++ tests/unit/test_mapping_index.py | 187 +++++++++++++++++++++ tests/unit/test_mapping_lookup_cache.py | 51 ++++++ 4 files changed, 592 insertions(+), 27 deletions(-) create mode 100644 scripts/bench_report.py create mode 100644 tests/unit/test_mapping_index.py create mode 100644 tests/unit/test_mapping_lookup_cache.py 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/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/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 -- 2.47.2 From 27a1f22be4c0edfb8e31ea0d2fc506fd7fd954e7 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Fri, 9 Oct 2026 19:14:43 +0400 Subject: [PATCH 2/2] Stream uploaded attachments to disk and cover the progress callback UploadedFile already holds the whole attachment in memory; getvalue() made a second full copy of each of the 230 files before writing. Copy through the file interface instead. Smoke tests pin the progress callback contract: once per XML file, including unreadable ones. Refs #1 Co-Authored-By: Claude Fable 5.1 --- main.py | 8 +++- tests/smoke/test_batch_progress.py | 59 +++++++++++++++++++++++++++++ tests/unit/test_main_attachments.py | 50 ++++++++++++++++++++++++ 3 files changed, 116 insertions(+), 1 deletion(-) create mode 100644 tests/smoke/test_batch_progress.py create mode 100644 tests/unit/test_main_attachments.py 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/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" -- 2.47.2