Ускорить формирование отчёта: обратный индекс маппинга, потоковые вложения, тесты прогресса #2
@ -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()
|
||||
|
||||
|
||||
8
main.py
8
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
|
||||
|
||||
|
||||
176
scripts/bench_report.py
Normal file
176
scripts/bench_report.py
Normal file
@ -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())
|
||||
59
tests/smoke/test_batch_progress.py
Normal file
59
tests/smoke/test_batch_progress.py
Normal file
@ -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)]
|
||||
50
tests/unit/test_main_attachments.py
Normal file
50
tests/unit/test_main_attachments.py
Normal file
@ -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"
|
||||
187
tests/unit/test_mapping_index.py
Normal file
187
tests/unit/test_mapping_index.py
Normal file
@ -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)
|
||||
51
tests/unit/test_mapping_lookup_cache.py
Normal file
51
tests/unit/test_mapping_lookup_cache.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user