feat(reports): протокол и парсеры отчётов Сбера, ВТБ и Snowball CSV
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/, когда оно на месте.
This commit is contained in:
Dmitry
@@ -0,0 +1,168 @@
|
||||
"""Персданные не должны попадать в git (AGENTS.md).
|
||||
|
||||
Отчёты Сбера и ВТБ содержат ФИО, ИНН, номер соглашения и номер лицевого счёта. В
|
||||
репозиторий идут только обезличенные копии, и этот тест — то, что делает правило
|
||||
исполнимым, а не пожеланием: он падает, если в `tests/fixtures/reports/` (вне `raw/`)
|
||||
встречается что-то персональное.
|
||||
|
||||
Две проверки, и вторая важнее первой:
|
||||
|
||||
1. **По форме.** 12-значный ИНН, 20-значный счёт, ФИО кириллицей — ловятся регулярками и
|
||||
без доступа к оригиналу, то есть работают и в CI, где `raw/` нет.
|
||||
2. **По содержанию.** Если `raw/` на месте, из оригиналов извлекаются те же секреты, что
|
||||
вырезает `anonymize.py`, и каждый проверяется на отсутствие в каждой фикстуре и в её
|
||||
имени. Это ловит случай, который первая проверка пропустит: фикстуру, собранную руками
|
||||
мимо скрипта.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import re
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from fintracker.sources.reports.anonymize import (
|
||||
FAKE_NAME,
|
||||
FAKE_PATRONYMIC,
|
||||
FAKE_SURNAME,
|
||||
RE_ACCOUNT_20,
|
||||
RE_CAPS_FIO,
|
||||
RE_INN,
|
||||
RE_PLACEHOLDER_NUMBER,
|
||||
RE_SHORT_NAME,
|
||||
Secrets,
|
||||
collect_secrets,
|
||||
)
|
||||
|
||||
FIXTURES = Path(__file__).resolve().parent / "fixtures" / "reports"
|
||||
RAW = FIXTURES / "raw"
|
||||
|
||||
ALLOWED_NAMES = {
|
||||
f"{FAKE_SURNAME} {FAKE_NAME[0]}. {FAKE_PATRONYMIC[0]}.",
|
||||
f"{FAKE_SURNAME.upper()} {FAKE_NAME.upper()} {FAKE_PATRONYMIC.upper()}",
|
||||
}
|
||||
|
||||
|
||||
def committed_fixtures() -> list[Path]:
|
||||
"""Every fixture that git tracks: everything under reports/ except raw/."""
|
||||
if not FIXTURES.exists():
|
||||
return []
|
||||
return sorted(
|
||||
p
|
||||
for p in FIXTURES.rglob("*")
|
||||
if p.is_file() and RAW not in p.parents and p != RAW and not p.name.startswith(".")
|
||||
)
|
||||
|
||||
|
||||
def raw_reports() -> list[Path]:
|
||||
if not RAW.exists():
|
||||
return []
|
||||
return sorted(p for p in RAW.rglob("*") if p.is_file() and not p.name.startswith("."))
|
||||
|
||||
|
||||
def raw_secrets() -> Secrets:
|
||||
"""The same corpus-wide secret set the script uses — never a per-file one."""
|
||||
return collect_secrets((p.read_bytes(), p.name) for p in raw_reports())
|
||||
|
||||
|
||||
def readable_text(path: Path) -> str:
|
||||
"""The visible text of a fixture, whatever its container."""
|
||||
data = path.read_bytes()
|
||||
if path.suffix.lower() in {".xlsx", ".xlsm"}:
|
||||
archive = zipfile.ZipFile(io.BytesIO(data))
|
||||
parts = [
|
||||
archive.read(name).decode("utf-8", "replace")
|
||||
for name in archive.namelist()
|
||||
if name.startswith("xl/") and name.endswith(".xml")
|
||||
]
|
||||
return "\n".join(parts)
|
||||
return data.decode("utf-8", "replace")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", committed_fixtures(), ids=lambda p: str(p.name))
|
||||
def test_fixture_has_no_personal_shapes(path: Path) -> None:
|
||||
text = readable_text(path)
|
||||
|
||||
inns = [v for v in RE_INN.findall(text) if not RE_PLACEHOLDER_NUMBER.match(v)]
|
||||
assert not inns, f"{path.name}: похоже на ИНН физлица: {inns[:3]}"
|
||||
|
||||
accounts = [v for v in RE_ACCOUNT_20.findall(text) if not RE_PLACEHOLDER_NUMBER.match(v)]
|
||||
assert not accounts, f"{path.name}: похоже на номер счёта: {accounts[:3]}"
|
||||
|
||||
names = {m.group(0) for m in RE_SHORT_NAME.finditer(text)} - ALLOWED_NAMES
|
||||
assert not names, f"{path.name}: ФИО с инициалами: {sorted(names)[:3]}"
|
||||
|
||||
caps = {m.group(0) for m in RE_CAPS_FIO.finditer(text)} - ALLOWED_NAMES
|
||||
assert not caps, f"{path.name}: ФИО заглавными: {sorted(caps)[:3]}"
|
||||
|
||||
|
||||
@pytest.mark.skipif(not RAW.exists(), reason="реальных отчётов нет (CI) — проверка по форме")
|
||||
def test_no_secret_from_raw_leaks_into_fixtures() -> None:
|
||||
"""Ни один секрет, извлечённый из оригиналов, не встречается в коммитимых фикстурах."""
|
||||
secrets = set(raw_secrets().originals)
|
||||
if not secrets:
|
||||
pytest.skip("в оригиналах не нашлось персданных — нечего проверять")
|
||||
|
||||
leaks: list[str] = []
|
||||
for fixture in committed_fixtures():
|
||||
text = readable_text(fixture)
|
||||
for secret in secrets:
|
||||
if secret in text:
|
||||
leaks.append(f"{fixture.name}: содержимое содержит {secret!r}")
|
||||
if secret in fixture.name:
|
||||
leaks.append(f"{fixture.name}: имя файла содержит {secret!r}")
|
||||
assert not leaks, "утечка персданных в git:\n" + "\n".join(leaks)
|
||||
|
||||
|
||||
@pytest.mark.skipif(not RAW.exists(), reason="реальных отчётов нет")
|
||||
def test_fixtures_are_reproducible_from_the_script() -> None:
|
||||
"""Фикстуры совпадают с тем, что сейчас выдаёт скрипт: обезличивание воспроизводимо."""
|
||||
from fintracker.sources.reports.anonymize import anonymize, anonymize_filename
|
||||
|
||||
secrets = raw_secrets()
|
||||
mismatched: list[str] = []
|
||||
for source in raw_reports():
|
||||
clean, _ = anonymize(source.read_bytes(), source.name, secrets)
|
||||
relative = source.relative_to(RAW)
|
||||
target = FIXTURES / relative.parent / anonymize_filename(source.name, secrets)
|
||||
if not target.exists():
|
||||
mismatched.append(f"нет фикстуры {target.name} для {relative}")
|
||||
elif target.read_bytes() != clean:
|
||||
mismatched.append(f"{target.name} не совпадает с выводом скрипта")
|
||||
assert not mismatched, (
|
||||
"перегенерируйте: uv run python scripts/anonymize_reports.py\n" + "\n".join(mismatched)
|
||||
)
|
||||
|
||||
|
||||
def test_raw_reports_are_not_tracked_by_git() -> None:
|
||||
"""`raw/` в .gitignore — иначе всё остальное бессмысленно."""
|
||||
import subprocess
|
||||
|
||||
root = Path(__file__).resolve().parents[2]
|
||||
out = subprocess.run(
|
||||
["git", "ls-files", "backend/tests/fixtures/reports/raw"],
|
||||
cwd=root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
tracked = [
|
||||
line
|
||||
for line in out.stdout.splitlines()
|
||||
if line.strip() and not line.rsplit("/", 1)[-1].startswith(".")
|
||||
]
|
||||
assert not tracked, f"реальные отчёты под контролем версий: {tracked}"
|
||||
|
||||
|
||||
RE_ANY_CYRILLIC_FIO = re.compile(r"[А-ЯЁ][а-яё]+\s+[А-ЯЁ][а-яё]+(?:ович|евич|овна|евна)\b")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", committed_fixtures(), ids=lambda p: str(p.name))
|
||||
def test_fixture_has_no_full_patronymic(path: Path) -> None:
|
||||
"""«Иванов Пётр Сергеевич» в обычном регистре — отдельный случай, регистр не спасает."""
|
||||
found = {m.group(0) for m in RE_ANY_CYRILLIC_FIO.finditer(readable_text(path))}
|
||||
found -= {f"{FAKE_NAME} {FAKE_PATRONYMIC}", f"{FAKE_SURNAME} {FAKE_PATRONYMIC}"}
|
||||
assert not found, f"{path.name}: ФИО с отчеством: {sorted(found)[:3]}"
|
||||
Reference in New Issue
Block a user