Dmitry
2e742a093b
sources/reports/base.py — контракт ReportParser (плагины: sniff/parse, чистые функции без БД). registry.py выбирает парсер по содержимому файла, CSV последним: он узнаёт файл по набору колонок и иначе перехватил бы чужой формат. Комиссия капитализируется в сделку, отдельным событием не эмитится: и Сбер, и ВТБ печатают её дважды — колонками в сделках и строками в движении денег, суммы совпадают, второе прочтение задвоило бы её. Расчётные строки («Сделка от …», «Сальдо расчетов по сделкам») не эмитятся — это денежные ноги уже учтённых сделок. У Сбера таблица «Информация о зачислениях на ИИС» кумулятивна за календарный год и в леджер не идёт, иначе два пересекающихся отчёта задвоили бы пополнения. CSV Snowball сводный по всем брокерам — годится как сверка потоков и сделок на уровне портфеля, но не позиций по счетам; CUSTOM_HOLDING_PRICE не событие, а цена — уходит в meta для price_manual. Обезличивание — anonymize.py + scripts/anonymize_reports.py, секреты собираются по всему корпусу отчётов разом (Snowball цитирует номер договора Сбера в примечании к переводу). test_fixtures_anonymized.py падает, если в tests/fixtures/reports/ вне raw/ найдётся ИНН, ФИО или номер счёта — и по форме (работает в CI без raw/), и по фактическому содержимому raw/, когда оно на месте.
57 lines
2.6 KiB
Python
57 lines
2.6 KiB
Python
"""Реестр парсеров отчётов: какой файл достаётся какому парсеру (plan §2).
|
|
|
|
Проверка выглядит тривиальной, но ловит ровно ту ошибку, которую больше не поймает никто:
|
|
универсальный CSV узнаёт файл по набору колонок, то есть по определению шире любого
|
|
брокерского формата, и достаточно поставить его в списке первым, чтобы отчёт Сбера тихо
|
|
уехал в чужой парсер. Порядок в `PARSERS` — это поведение, а не оформление.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fintracker.sources.reports import registry
|
|
|
|
FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "reports"
|
|
|
|
CASES = [
|
|
(FIXTURES / "sber" / "S930W42_11022026_17092026.html", "report_sber"),
|
|
(FIXTURES / "sber" / "S930W42_01082026_31082026.html", "report_sber"),
|
|
(FIXTURES / "vtb" / "reportd0235f1e-55ac-fdc8-5c7b-279e4fe26e2b.xlsx", "report_vtb"),
|
|
(FIXTURES / "vtb" / "report8ef8347a-2deb-e880-959a-d9a8ca12f735.xlsx", "report_vtb"),
|
|
(next((FIXTURES / "snowball").glob("*.csv")), "csv"),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("path,expected", CASES, ids=lambda v: getattr(v, "name", v))
|
|
def test_pick_routes_each_fixture_to_its_parser(path: Path, expected: str) -> None:
|
|
chosen = registry.pick(path.read_bytes(), path.name)
|
|
assert chosen is not None, f"{path.name}: ни один парсер не узнал файл"
|
|
assert chosen.name == expected
|
|
|
|
|
|
def test_every_registered_parser_is_reachable_by_name() -> None:
|
|
assert set(registry.names()) == {"report_sber", "report_vtb", "csv"}
|
|
for name in registry.names():
|
|
assert registry.get(name).name == name
|
|
|
|
|
|
def test_unknown_format_is_a_none_not_an_exception() -> None:
|
|
"""`pick` возвращает None, и роутер превращает это в 415 — не в 500."""
|
|
assert registry.pick(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n", "statement.pdf") is None
|
|
assert registry.pick(b"", "empty.txt") is None
|
|
|
|
|
|
def test_parser_metadata_is_filled_in() -> None:
|
|
"""`parser_name` и `parser_version` уходят в `raw_report_file` — пустых быть не должно."""
|
|
for parser in registry.PARSERS:
|
|
assert parser.name and parser.broker and parser.version
|
|
assert parser.formats
|
|
|
|
|
|
def test_csv_is_last_so_it_cannot_shadow_a_broker_format() -> None:
|
|
names = registry.names()
|
|
assert names.index("csv") == len(names) - 1
|