Python Разработчик-Эксперт

Модуль сверки банковских выписок «Нордвей Логистика»

1. Разбор задачи и допущения

Задача раскладывается на потоковый конвейер: чтение CSV → разбор и валидация → фильтр по датам → дедупликация → категоризация → агрегация → экспорт. Каждый этап реализован отдельной небольшой единицей и тестируется независимо.

Допущения (при необходимости легко меняются):

• Знак суммы. Отрицательная сумма означает списание, положительная — поступление. В отчёте по каждой категории выводятся поступления, списания и сальдо. Если выводить одну знаковую сумму, поступления и списания внутри категории будут взаимно гаситься.
• Валюты не складываются. Агрегация идёт по ключу (месяц, категория, валюта), итоги выводятся отдельно по каждой валюте. Складывать RUB с USD нельзя.
• Ключ дубликата — (дата, сумма, контрагент, назначение), строго по ТЗ. Сравнение точное, после обрезки пробелов по краям. Остаётся первое вхождение, остальные попадают в duplicates со ссылкой на номер строки оригинала.
• Сопоставление ключевых слов. Ищется подстрока без учёта регистра, «ё» приравнивается к «е». Побеждает самое длинное ключевое слово, при равной длине — то, что раньше в словаре. Например, «аренда автомобиля» имеет приоритет над «аренд».
• Формат файла. Заголовок — первая строка файла. Отбраковываются также строки с некорректной датой, пустой валютой и неверным числом колонок: последнее защищает от неэкранированного ; в назначении платежа, который сдвигает колонки. Суммы в отчёте выводятся с десятичной запятой и без разделителя тысяч (удобно для русского Excel), перевод строки — \n.

2. Основной код — bank_reconciliation.py

"""Сверка банковских выписок при закрытии месяца.

Модуль читает CSV-выгрузку из интернет-банка (формат, близкий к Тинькофф Бизнес),
валидирует и дедуплицирует транзакции, раскладывает их по категориям и строит
сводный отчёт по месяцам и категориям с экспортом в CSV.

Поток данных::

    строки CSV -> iter_csv_records -> parse_record -> фильтр по датам
               -> Deduplicator -> Categorizer -> ReportBuilder -> StatementReport

Пример::

    result = process_statement_file("statement.csv", {"аренд": "Аренда офиса"})
    export_report_csv(result.report, "report.csv")
"""

from __future__ import annotations

import csv
import datetime as dt
import io
import logging
import os
import re
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from decimal import Decimal
from typing import Final, Protocol, TextIO, TypeAlias, TypeVar

__all__ = [
    "ALL_PERIODS_LABEL",
    "DEFAULT_CATEGORY",
    "REPORT_COLUMNS",
    "REQUIRED_COLUMNS",
    "TOTAL_LABEL",
    "CategorizedTransaction",
    "Categorizer",
    "Deduplicator",
    "DuplicateRow",
    "FieldValidationError",
    "KeywordCategorizer",
    "PeriodReport",
    "ProcessingResult",
    "RawRecord",
    "RejectedRow",
    "ReportBuilder",
    "ReportLine",
    "RowValidationError",
    "StatementError",
    "StatementFormatError",
    "StatementReport",
    "Transaction",
    "export_report_csv",
    "format_amount",
    "iter_csv_records",
    "parse_amount",
    "parse_date",
    "parse_fields",
    "parse_record",
    "process_statement",
    "process_statement_file",
    "report_to_csv_text",
    "validate_inn",
    "write_report_csv",
]

logger = logging.getLogger(__name__)

# --------------------------------------------------------------------------- #
# Константы формата
# --------------------------------------------------------------------------- #

SOURCE_ENCODING: Final = "cp1251"
REPORT_ENCODING: Final = "utf-8"
CSV_DELIMITER: Final = ";"

COL_DATE: Final = "Дата операции"
COL_AMOUNT: Final = "Сумма"
COL_CURRENCY: Final = "Валюта"
COL_PURPOSE: Final = "Назначение платежа"
COL_COUNTERPARTY: Final = "Контрагент"
COL_INN: Final = "ИНН контрагента"
REQUIRED_COLUMNS: Final[tuple[str, ...]] = (
    COL_DATE,
    COL_AMOUNT,
    COL_CURRENCY,
    COL_PURPOSE,
    COL_COUNTERPARTY,
    COL_INN,
)

DEFAULT_CATEGORY: Final = "Прочее"
TOTAL_LABEL: Final = "Итого"
ALL_PERIODS_LABEL: Final = "Весь период"

REPORT_COLUMNS: Final[tuple[str, ...]] = (
    "Период",
    "Категория",
    "Валюта",
    "Количество",
    "Поступления",
    "Списания",
    "Сальдо",
)

_ZERO: Final = Decimal("0.00")
_CENT: Final = Decimal("0.01")
_DATE_FORMAT: Final = "%d.%m.%Y"
# До 15 цифр в целой части (с запасом покрывает любые реальные суммы и
# гарантирует, что quantize не выйдет за точность контекста Decimal),
# дробная часть — 1–2 знака через запятую или точку.
_AMOUNT_RE: Final = re.compile(r"[+-]?[0-9]{1,15}(?:[.,][0-9]{1,2})?")
# Явно [0-9], а не \d: \d пропускает «цифры» других алфавитов.
_INN_RE: Final = re.compile(r"[0-9]{10}|[0-9]{12}")
# Пробел, неразрывный пробел, узкий неразрывный и тонкий пробел —
# встречающиеся в банковских выгрузках разделители тысяч.
_THOUSANDS_SEPARATORS: Final = str.maketrans("", "", " \u00a0\u202f\u2009")

StrPath: TypeAlias = str | os.PathLike[str]
DuplicateKey: TypeAlias = tuple[dt.date, Decimal, str, str]
_T = TypeVar("_T")


# --------------------------------------------------------------------------- #
# Исключения
# --------------------------------------------------------------------------- #


class StatementError(Exception):
    """Базовое исключение модуля."""


class StatementFormatError(StatementError):
    """Файл выписки невозможно обработать целиком.

    Например: пустой файл, отсутствуют обязательные колонки, неверная кодировка
    или повреждённая CSV-структура.
    """


class FieldValidationError(StatementError, ValueError):
    """Значение отдельного поля не прошло валидацию."""


class RowValidationError(StatementError):
    """Строка выписки некорректна; содержит все найденные причины.

    Attributes:
        reasons: Причины отбраковки в порядке следования колонок.
    """

    def __init__(self, reasons: Sequence[str]) -> None:
        """Создаёт исключение.

        Args:
            reasons: Непустой список причин отбраковки.
        """
        self.reasons: tuple[str, ...] = tuple(reasons)
        super().__init__("; ".join(self.reasons))


# --------------------------------------------------------------------------- #
# Модели данных
# --------------------------------------------------------------------------- #


@dataclass(frozen=True, slots=True)
class Transaction:
    """Валидная транзакция из выписки.

    Attributes:
        line_no: Номер строки исходного файла (начало записи, с 1).
        date: Дата операции.
        amount: Сумма; отрицательная — списание, положительная — поступление.
        currency: Код валюты в верхнем регистре (например, "RUB").
        purpose: Назначение платежа.
        counterparty: Наименование контрагента (может быть пустым).
        inn: ИНН контрагента (10 или 12 цифр).
    """

    line_no: int
    date: dt.date
    amount: Decimal
    currency: str
    purpose: str
    counterparty: str
    inn: str


@dataclass(frozen=True, slots=True)
class CategorizedTransaction(Transaction):
    """Транзакция с присвоенной категорией.

    Структура плоская, чтобы её было удобно превратить в DataFrame
    через ``dataclasses.asdict``.

    Attributes:
        category: Категория по правилам категоризации.
    """

    category: str

    @classmethod
    def from_transaction(
        cls, transaction: Transaction, category: str
    ) -> CategorizedTransaction:
        """Создаёт категоризированную транзакцию из исходной.

        Args:
            transaction: Исходная транзакция.
            category: Присвоенная категория.

        Returns:
            Новый объект с теми же полями и категорией.
        """
        return cls(
            line_no=transaction.line_no,
            date=transaction.date,
            amount=transaction.amount,
            currency=transaction.currency,
            purpose=transaction.purpose,
            counterparty=transaction.counterparty,
            inn=transaction.inn,
            category=category,
        )


@dataclass(frozen=True, slots=True)
class RawRecord:
    """Сырая запись CSV, сопоставленная с заголовком.

    Attributes:
        line_no: Номер строки файла, с которой начинается запись.
        fields: Значения по именам колонок.
        column_error: Описание проблемы с числом колонок или None.
    """

    line_no: int
    fields: dict[str, str]
    column_error: str | None = None


@dataclass(frozen=True, slots=True)
class RejectedRow:
    """Отбракованная строка выписки.

    Attributes:
        line_no: Номер строки исходного файла.
        reasons: Все причины отбраковки.
        raw: Исходные значения колонок — для ручного разбора.
    """

    line_no: int
    reasons: tuple[str, ...]
    raw: dict[str, str]

    @property
    def reason(self) -> str:
        """Причины отбраковки одной строкой."""
        return "; ".join(self.reasons)


@dataclass(frozen=True, slots=True)
class DuplicateRow:
    """Строка, признанная дубликатом ранее встреченной.

    Attributes:
        line_no: Номер строки дубликата.
        original_line_no: Номер строки первого вхождения.
        transaction: Разобранная транзакция-дубликат.
    """

    line_no: int
    original_line_no: int
    transaction: Transaction


@dataclass(frozen=True, slots=True)
class ReportLine:
    """Строка сводного отчёта.

    Attributes:
        period: Месяц "ГГГГ-ММ" или ALL_PERIODS_LABEL для итога за весь период.
        category: Категория или TOTAL_LABEL для итоговой строки.
        currency: Код валюты.
        count: Количество транзакций.
        income: Сумма поступлений (>= 0).
        expense: Сумма списаний (<= 0).
        net: Сальдо: income + expense.
        is_total: True для итоговых строк — удобно для фильтрации в дашборде.
    """

    period: str
    category: str
    currency: str
    count: int
    income: Decimal
    expense: Decimal
    net: Decimal
    is_total: bool = False


@dataclass(frozen=True, slots=True)
class PeriodReport:
    """Сводка за один период.

    Attributes:
        period: Месяц "ГГГГ-ММ" или ALL_PERIODS_LABEL.
        categories: Строки по (категория, валюта), отсортированные.
        totals: Итоги периода по каждой валюте, отсортированные по валюте.
    """

    period: str
    categories: tuple[ReportLine, ...]
    totals: tuple[ReportLine, ...]


@dataclass(frozen=True, slots=True)
class StatementReport:
    """Итоговый отчёт по выписке.

    Attributes:
        periods: Сводки по месяцам в хронологическом порядке.
        overall: Сводка за весь период.
        transaction_count: Число учтённых транзакций.
        first_date: Дата первой учтённой транзакции или None.
        last_date: Дата последней учтённой транзакции или None.
    """

    periods: tuple[PeriodReport, ...]
    overall: PeriodReport
    transaction_count: int
    first_date: dt.date | None
    last_date: dt.date | None

    def lines(self) -> Iterator[ReportLine]:
        """Возвращает все строки отчёта в порядке экспорта.

        Порядок: для каждого месяца — категории, затем итоги по валютам;
        в конце — то же самое за весь период.

        Yields:
            Строки отчёта.
        """
        for period in (*self.periods, self.overall):
            yield from period.categories
            yield from period.totals


@dataclass(frozen=True, slots=True)
class ProcessingResult:
    """Результат обработки выписки.

    Attributes:
        transactions: Валидные уникальные категоризированные транзакции
            в порядке следования в файле.
        rejected: Отбракованные строки.
        duplicates: Строки-дубликаты.
        filtered_out: Число валидных строк вне диапазона дат.
        report: Сводный отчёт.
    """

    transactions: list[CategorizedTransaction]
    rejected: list[RejectedRow]
    duplicates: list[DuplicateRow]
    filtered_out: int
    report: StatementReport


# --------------------------------------------------------------------------- #
# Разбор отдельных полей
# --------------------------------------------------------------------------- #


def parse_amount(raw: str) -> Decimal:
    """Разбирает сумму вида "12 500,50" в Decimal с точностью до копеек.

    Допускаются знак, пробелы (в т.ч. неразрывные) как разделители тысяч,
    запятая или точка как десятичный разделитель и не более двух знаков
    после него.

    Args:
        raw: Строковое значение суммы.

    Returns:
        Сумма, приведённая к двум знакам после запятой. Отрицательный ноль
        нормализуется в 0.00.

    Raises:
        FieldValidationError: Если строка не является корректной суммой.
    """
    compact = raw.translate(_THOUSANDS_SEPARATORS)
    if not _AMOUNT_RE.fullmatch(compact):
        raise FieldValidationError(f"некорректная сумма: {raw!r}")
    amount = Decimal(compact.replace(",", ".")).quantize(_CENT)
    return amount if amount != 0 else _ZERO


def parse_date(raw: str) -> dt.date:
    """Разбирает дату в формате ДД.ММ.ГГГГ.

    Args:
        raw: Строковое значение даты.

    Returns:
        Дата операции.

    Raises:
        FieldValidationError: Если строка не является корректной датой.
    """
    try:
        return dt.datetime.strptime(raw.strip(), _DATE_FORMAT).date()
    except ValueError:
        raise FieldValidationError(f"некорректная дата: {raw!r}") from None


def validate_inn(raw: str) -> str:
    """Проверяет ИНН по длине: 10 цифр (юрлицо) или 12 цифр (ИП/физлицо).

    Контрольная сумма не проверяется.

    Args:
        raw: Строковое значение ИНН.

    Returns:
        ИНН без пробелов по краям.

    Raises:
        FieldValidationError: Если ИНН не состоит из 10 или 12 цифр.
    """
    inn = raw.strip()
    if not _INN_RE.fullmatch(inn):
        raise FieldValidationError(f"ИНН должен содержать 10 или 12 цифр: {raw!r}")
    return inn


def _try_parse(parser: Callable[[str], _T], raw: str, reasons: list[str]) -> _T | None:
    """Вызывает парсер поля; при ошибке добавляет её текст в reasons."""
    try:
        return parser(raw)
    except FieldValidationError as exc:
        reasons.append(str(exc))
        return None


def parse_fields(fields: Mapping[str, str], line_no: int) -> Transaction:
    """Строит Transaction из значений колонок, собирая все ошибки строки.

    Args:
        fields: Значения по именам колонок; отсутствующие считаются пустыми.
        line_no: Номер строки исходного файла.

    Returns:
        Валидная транзакция.

    Raises:
        RowValidationError: Если хотя бы одно поле некорректно; содержит все
            найденные причины, а не только первую.
    """

    def value(column: str) -> str:
        return fields.get(column, "").strip()

    reasons: list[str] = []
    op_date = _try_parse(parse_date, value(COL_DATE), reasons)
    amount = _try_parse(parse_amount, value(COL_AMOUNT), reasons)
    currency = value(COL_CURRENCY).upper()
    if not currency:
        reasons.append("не указана валюта")
    purpose = value(COL_PURPOSE)
    if not purpose:
        reasons.append("пустое назначение платежа")
    inn = _try_parse(validate_inn, value(COL_INN), reasons)

    if op_date is None or amount is None or inn is None or reasons:
        raise RowValidationError(reasons)

    return Transaction(
        line_no=line_no,
        date=op_date,
        amount=amount,
        currency=currency,
        purpose=purpose,
        counterparty=value(COL_COUNTERPARTY),
        inn=inn,
    )


def parse_record(record: RawRecord) -> Transaction:
    """Разбирает сырую запись CSV в транзакцию.

    Args:
        record: Запись, полученная из iter_csv_records.

    Returns:
        Валидная транзакция.

    Raises:
        RowValidationError: Если число колонок неверно или поля некорректны.
    """
    if record.column_error is not None:
        raise RowValidationError([record.column_error])
    return parse_fields(record.fields, record.line_no)


# --------------------------------------------------------------------------- #
# Потоковое чтение CSV
# --------------------------------------------------------------------------- #


def _normalize_header(header: Sequence[str]) -> list[str]:
    """Очищает имена колонок: BOM, пробелы, хвостовые пустые колонки."""
    columns = [name.replace("\ufeff", "").strip() for name in header]
    while columns and not columns[-1]:
        columns.pop()
    return columns


def _make_record(line_no: int, columns: Sequence[str], values: Sequence[str]) -> RawRecord:
    """Сопоставляет значения строки с заголовком и проверяет число колонок."""
    expected, actual = len(columns), len(values)
    column_error: str | None = None
    # Лишние пустые значения в хвосте (строка заканчивается на ";") допустимы.
    if actual < expected or (
        actual > expected and any(v.strip() for v in values[expected:])
    ):
        column_error = (
            f"неверное число колонок: ожидалось {expected}, получено {actual}"
        )
    fields = {name: value for name, value in zip(columns, values) if name}
    return RawRecord(line_no=line_no, fields=fields, column_error=column_error)


def iter_csv_records(lines: Iterable[str]) -> Iterator[RawRecord]:
    """Потоково читает CSV-выписку и выдаёт записи, сопоставленные с заголовком.

    Файл не загружается в память целиком: строки читаются по одной.
    Пустые строки пропускаются. Номер строки — физический номер строки файла,
    с которой начинается запись (корректно для многострочных полей в кавычках).

    Args:
        lines: Источник строк: открытый текстовый файл (с newline=""),
            io.StringIO или любой итерируемый объект строк.

    Yields:
        Записи RawRecord.

    Raises:
        StatementFormatError: Если файл пуст, в заголовке нет обязательных
            колонок (или они повторяются) либо CSV повреждён.
    """
    reader = csv.reader(lines, delimiter=CSV_DELIMITER)
    try:
        header = next(reader, None)
        if header is None:
            raise StatementFormatError("файл выписки пуст: нет строки заголовка")
        columns = _normalize_header(header)
        missing = [c for c in REQUIRED_COLUMNS if c not in columns]
        if missing:
            raise StatementFormatError(
                f"в заголовке отсутствуют обязательные колонки: {', '.join(missing)}"
            )
        repeated = [c for c in REQUIRED_COLUMNS if columns.count(c) > 1]
        if repeated:
            raise StatementFormatError(
                f"в заголовке повторяются колонки: {', '.join(repeated)}"
            )

        previous_line = reader.line_num
        for values in reader:
            line_no = previous_line + 1
            previous_line = reader.line_num
            if not any(value.strip() for value in values):
                continue
            yield _make_record(line_no, columns, values)
    except csv.Error as exc:
        raise StatementFormatError(
            f"ошибка разбора CSV около строки {reader.line_num}: {exc}"
        ) from exc


# --------------------------------------------------------------------------- #
# Категоризация
# --------------------------------------------------------------------------- #


class Categorizer(Protocol):
    """Контракт категоризатора: любая реализация с методом categorize."""

    def categorize(self, transaction: Transaction) -> str:
        """Возвращает категорию транзакции.

        Args:
            transaction: Транзакция для категоризации.

        Returns:
            Название категории.
        """
        ...


def _normalize_text(text: str) -> str:
    """Приводит текст к виду для сопоставления: регистр, «ё», пробелы."""
    return " ".join(text.casefold().replace("ё", "е").split())


class KeywordCategorizer:
    """Категоризатор по правилам «ключевое слово в назначении -> категория».

    Сопоставление — поиск подстроки без учёта регистра, «ё» приравнивается
    к «е», повторные пробелы схлопываются. Побеждает самое длинное подходящее
    ключевое слово (более специфичное правило), при равной длине — то, что
    раньше в словаре. Для учёта словоформ используйте основы: "аренд" найдёт
    «аренда», «аренды», «арендная плата».
    """

    def __init__(self, rules: Mapping[str, str], default: str = DEFAULT_CATEGORY) -> None:
        """Создаёт категоризатор.

        Args:
            rules: Словарь "ключевое слово -> категория".
            default: Категория, если ни одно правило не подошло.

        Raises:
            ValueError: Если ключевое слово или категория пустые либо два
                ключевых слова совпадают после нормализации.
        """
        if not default.strip():
            raise ValueError("категория по умолчанию не может быть пустой")
        normalized: list[tuple[str, str]] = []
        seen: set[str] = set()
        for keyword, category in rules.items():
            key = _normalize_text(keyword)
            if not key:
                raise ValueError("ключевое слово не может быть пустым")
            if not category.strip():
                raise ValueError(f"пустая категория для ключевого слова {keyword!r}")
            if key in seen:
                raise ValueError(f"ключевое слово {keyword!r} задано повторно")
            seen.add(key)
            normalized.append((key, category.strip()))
        # sorted() стабилен: при равной длине сохраняется порядок словаря.
        self._rules: tuple[tuple[str, str], ...] = tuple(
            sorted(normalized, key=lambda rule: -len(rule[0]))
        )
        self._default = default.strip()

    @property
    def rules(self) -> tuple[tuple[str, str], ...]:
        """Нормализованные правила в порядке применения."""
        return self._rules

    def categorize(self, transaction: Transaction) -> str:
        """Возвращает категорию по назначению платежа.

        Args:
            transaction: Транзакция для категоризации.

        Returns:
            Категория первого (самого длинного) подошедшего правила или
            категория по умолчанию.
        """
        text = _normalize_text(transaction.purpose)
        for keyword, category in self._rules:
            if keyword in text:
                return category
        return self._default


# --------------------------------------------------------------------------- #
# Дедупликация
# --------------------------------------------------------------------------- #


class Deduplicator:
    """Отслеживает дубликаты в пределах одного запуска обработки.

    Ключ дубликата — (дата, сумма, контрагент, назначение платежа).
    """

    def __init__(self) -> None:
        """Создаёт пустой реестр."""
        self._first_seen: dict[DuplicateKey, int] = {}

    @staticmethod
    def key(transaction: Transaction) -> DuplicateKey:
        """Возвращает ключ дубликата транзакции.

        Args:
            transaction: Транзакция.

        Returns:
            Кортеж (дата, сумма, контрагент, назначение).
        """
        return (
            transaction.date,
            transaction.amount,
            transaction.counterparty,
            transaction.purpose,
        )

    def register(self, transaction: Transaction) -> int | None:
        """Регистрирует транзакцию.

        Args:
            transaction: Транзакция.

        Returns:
            Номер строки первого вхождения, если транзакция — дубликат,
            иначе None (транзакция запоминается как оригинал).
        """
        key = self.key(transaction)
        original = self._first_seen.get(key)
        if original is None:
            self._first_seen[key] = transaction.line_no
        return original

    def __len__(self) -> int:
        """Число уникальных транзакций."""
        return len(self._first_seen)


# --------------------------------------------------------------------------- #
# Построение отчёта
# --------------------------------------------------------------------------- #


@dataclass(slots=True)
class _Bucket:
    """Изменяемый аккумулятор сумм для одной группы."""

    count: int = 0
    income: Decimal = _ZERO
    expense: Decimal = _ZERO

    def add(self, amount: Decimal) -> None:
        self.count += 1
        if amount > 0:
            self.income += amount
        elif amount < 0:
            self.expense += amount

    def merge(self, other: _Bucket) -> None:
        self.count += other.count
        self.income += other.income
        self.expense += other.expense

    def to_line(
        self, period: str, category: str, currency: str, *, is_total: bool
    ) -> ReportLine:
        return ReportLine(
            period=period,
            category=category,
            currency=currency,
            count=self.count,
            income=self.income,
            expense=self.expense,
            net=self.income + self.expense,
            is_total=is_total,
        )


def _period_of(value: dt.date) -> str:
    """Возвращает месяц в формате ГГГГ-ММ (лексикографически сортируемый)."""
    return f"{value.year:04d}-{value.month:02d}"


def _build_period(period: str, buckets: Mapping[tuple[str, str], _Bucket]) -> PeriodReport:
    """Собирает детерминированно отсортированную сводку за период."""
    categories = tuple(
        bucket.to_line(period, category, currency, is_total=False)
        for (category, currency), bucket in sorted(buckets.items(), key=lambda i: i[0])
    )
    per_currency: dict[str, _Bucket] = {}
    for (_, currency), bucket in buckets.items():
        per_currency.setdefault(currency, _Bucket()).merge(bucket)
    totals = tuple(
        per_currency[currency].to_line(period, TOTAL_LABEL, currency, is_total=True)
        for currency in sorted(per_currency)
    )
    return PeriodReport(period=period, categories=categories, totals=totals)


class ReportBuilder:
    """Инкрементально накапливает агрегаты и строит StatementReport.

    Память — O(месяцы × категории × валюты), независимо от числа транзакций.
    """

    def __init__(self) -> None:
        """Создаёт пустой построитель."""
        self._buckets: dict[tuple[str, str, str], _Bucket] = {}
        self._count = 0
        self._first_date: dt.date | None = None
        self._last_date: dt.date | None = None

    def add(self, transaction: CategorizedTransaction) -> None:
        """Учитывает транзакцию в агрегатах.

        Args:
            transaction: Категоризированная транзакция.
        """
        key = (_period_of(transaction.date), transaction.category, transaction.currency)
        bucket = self._buckets.get(key)
        if bucket is None:
            bucket = self._buckets[key] = _Bucket()
        bucket.add(transaction.amount)
        self._count += 1
        if self._first_date is None or transaction.date < self._first_date:
            self._first_date = transaction.date
        if self._last_date is None or transaction.date > self._last_date:
            self._last_date = transaction.date

    def build(self) -> StatementReport:
        """Строит итоговый отчёт.

        Returns:
            Отчёт; все коллекции отсортированы, поэтому результат не зависит
            от порядка поступления транзакций.
        """
        by_period: dict[str, dict[tuple[str, str], _Bucket]] = {}
        overall: dict[tuple[str, str], _Bucket] = {}
        for (period, category, currency), bucket in self._buckets.items():
            by_period.setdefault(period, {})[(category, currency)] = bucket
            overall.setdefault((category, currency), _Bucket()).merge(bucket)

        return StatementReport(
            periods=tuple(_build_period(p, by_period[p]) for p in sorted(by_period)),
            overall=_build_period(ALL_PERIODS_LABEL, overall),
            transaction_count=self._count,
            first_date=self._first_date,
            last_date=self._last_date,
        )


# --------------------------------------------------------------------------- #
# Оркестрация
# --------------------------------------------------------------------------- #


def _in_range(value: dt.date, date_from: dt.date | None, date_to: dt.date | None) -> bool:
    """Проверяет попадание даты в диапазон (обе границы включительно)."""
    if date_from is not None and value < date_from:
        return False
    return date_to is None or value <= date_to


def process_statement(
    lines: Iterable[str],
    rules: Categorizer | Mapping[str, str],
    *,
    date_from: dt.date | None = None,
    date_to: dt.date | None = None,
    keep_transactions: bool = True,
) -> ProcessingResult:
    """Обрабатывает выписку: разбор, валидация, фильтр, дедупликация, отчёт.

    Некорректные строки не прерывают обработку: они попадают в
    ``ProcessingResult.rejected`` и логируются на уровне WARNING.

    Args:
        lines: Источник строк CSV (файл, открытый с newline="", StringIO и т.п.).
        rules: Словарь "ключевое слово -> категория" или готовый Categorizer.
        date_from: Начало периода включительно; None — без ограничения.
        date_to: Конец периода включительно; None — без ограничения.
        keep_transactions: Сохранять ли список транзакций. При False отчёт
            всё равно строится, а память не растёт с размером файла
            (кроме реестра дубликатов).

    Returns:
        Результат обработки.

    Raises:
        ValueError: Если date_from > date_to или правила некорректны.
        StatementFormatError: Если файл невозможно обработать целиком.
    """
    if date_from is not None and date_to is not None and date_from > date_to:
        raise ValueError(f"date_from ({date_from}) позже date_to ({date_to})")
    categorizer: Categorizer = (
        KeywordCategorizer(rules) if isinstance(rules, Mapping) else rules
    )

    transactions: list[CategorizedTransaction] = []
    rejected: list[RejectedRow] = []
    duplicates: list[DuplicateRow] = []
    filtered_out = 0
    deduplicator = Deduplicator()
    builder = ReportBuilder()

    for record in iter_csv_records(lines):
        try:
            transaction = parse_record(record)
        except RowValidationError as exc:
            row = RejectedRow(line_no=record.line_no, reasons=exc.reasons, raw=dict(record.fields))
            logger.warning("Строка %d отброшена: %s", row.line_no, row.reason)
            rejected.append(row)
            continue

        if not _in_range(transaction.date, date_from, date_to):
            filtered_out += 1
            continue

        original_line = deduplicator.register(transaction)
        if original_line is not None:
            duplicates.append(
                DuplicateRow(
                    line_no=transaction.line_no,
                    original_line_no=original_line,
                    transaction=transaction,
                )
            )
            logger.info(
                "Строка %d — дубликат строки %d", transaction.line_no, original_line
            )
            continue

        categorized = CategorizedTransaction.from_transaction(
            transaction, categorizer.categorize(transaction)
        )
        builder.add(categorized)
        if keep_transactions:
            transactions.append(categorized)

    report = builder.build()
    logger.info(
        "Выписка обработана: учтено %d, отброшено %d, дублей %d, вне периода %d",
        report.transaction_count,
        len(rejected),
        len(duplicates),
        filtered_out,
    )
    return ProcessingResult(
        transactions=transactions,
        rejected=rejected,
        duplicates=duplicates,
        filtered_out=filtered_out,
        report=report,
    )


def process_statement_file(
    path: StrPath,
    rules: Categorizer | Mapping[str, str],
    *,
    date_from: dt.date | None = None,
    date_to: dt.date | None = None,
    encoding: str = SOURCE_ENCODING,
    keep_transactions: bool = True,
) -> ProcessingResult:
    """Обрабатывает файл выписки с диска (потоково).

    Args:
        path: Путь к CSV-файлу.
        rules: Словарь "ключевое слово -> категория" или Categorizer.
        date_from: Начало периода включительно.
        date_to: Конец периода включительно.
        encoding: Кодировка файла (по умолчанию cp1251).
        keep_transactions: См. process_statement.

    Returns:
        Результат обработки.

    Raises:
        OSError: Если файл не удаётся открыть.
        StatementFormatError: Если файл повреждён или не в указанной кодировке.
        ValueError: Если date_from > date_to или правила некорректны.
    """
    try:
        with open(path, encoding=encoding, newline="") as source:
            return process_statement(
                source,
                rules,
                date_from=date_from,
                date_to=date_to,
                keep_transactions=keep_transactions,
            )
    except UnicodeDecodeError as exc:
        raise StatementFormatError(
            f"файл {os.fspath(path)!r} не соответствует кодировке {encoding}: {exc}"
        ) from exc


# --------------------------------------------------------------------------- #
# Экспорт
# --------------------------------------------------------------------------- #


def format_amount(value: Decimal) -> str:
    """Форматирует сумму для CSV: два знака, десятичная запятая.

    Args:
        value: Сумма.

    Returns:
        Строка вида "-12500,50" (без разделителя тысяч, без "-0,00").
    """
    quantized = value.quantize(_CENT)
    if quantized == 0:
        quantized = _ZERO
    return format(quantized, "f").replace(".", ",")


def write_report_csv(report: StatementReport, target: TextIO) -> None:
    """Пишет отчёт в текстовый поток в формате CSV с разделителем ";".

    Args:
        report: Отчёт.
        target: Текстовый поток (для файла — открытый с newline="").
    """
    writer = csv.writer(target, delimiter=CSV_DELIMITER, lineterminator="\n")
    writer.writerow(REPORT_COLUMNS)
    for line in report.lines():
        writer.writerow(
            (
                line.period,
                line.category,
                line.currency,
                str(line.count),
                format_amount(line.income),
                format_amount(line.expense),
                format_amount(line.net),
            )
        )


def report_to_csv_text(report: StatementReport) -> str:
    """Возвращает отчёт в виде CSV-строки (удобно для st.download_button).

    Args:
        report: Отчёт.

    Returns:
        Содержимое CSV.
    """
    buffer = io.StringIO(newline="")
    write_report_csv(report, buffer)
    return buffer.getvalue()


def export_report_csv(
    report: StatementReport, path: StrPath, *, encoding: str = REPORT_ENCODING
) -> None:
    """Сохраняет отчёт в CSV-файл.

    Args:
        report: Отчёт.
        path: Путь к создаваемому файлу (перезаписывается).
        encoding: Кодировка; "utf-8" по ТЗ, "utf-8-sig" — если файл будут
            открывать двойным кликом в Excel.

    Raises:
        OSError: Если файл не удаётся записать.
    """
    with open(path, "w", encoding=encoding, newline="") as target:
        write_report_csv(report, target)

3. Unit-тесты — test_bank_reconciliation.py

"""Unit-тесты модуля bank_reconciliation."""

from __future__ import annotations

import csv
import datetime as dt
import io
import logging
from collections.abc import Iterable, Sequence
from decimal import Decimal
from pathlib import Path

import pytest

from bank_reconciliation import (
    ALL_PERIODS_LABEL,
    REQUIRED_COLUMNS,
    TOTAL_LABEL,
    Deduplicator,
    FieldValidationError,
    KeywordCategorizer,
    RowValidationError,
    StatementFormatError,
    Transaction,
    export_report_csv,
    format_amount,
    iter_csv_records,
    parse_amount,
    parse_date,
    parse_fields,
    process_statement,
    process_statement_file,
    report_to_csv_text,
    validate_inn,
)

RULES = {"аренд": "Аренда офиса", "зарплат": "ФОТ", "топлив": "Транспорт"}

DEFAULTS = {
    "date": "05.03.2026",
    "amount": "-100 000,00",
    "currency": "RUB",
    "purpose": "Аренда офиса за март",
    "counterparty": "ООО Альфа",
    "inn": "7701234567",
}


def row(**overrides: str) -> list[str]:
    """Строка выписки в порядке REQUIRED_COLUMNS со значениями по умолчанию."""
    values = {**DEFAULTS, **overrides}
    return [values[k] for k in ("date", "amount", "currency", "purpose", "counterparty", "inn")]


def make_source(
    rows: Iterable[Sequence[str]], header: Sequence[str] = REQUIRED_COLUMNS
) -> io.StringIO:
    """Собирает CSV в памяти через csv.writer (корректное экранирование)."""
    buffer = io.StringIO(newline="")
    writer = csv.writer(buffer, delimiter=";", lineterminator="\n")
    writer.writerow(header)
    writer.writerows(rows)
    buffer.seek(0)
    return buffer


def make_tx(**overrides: object) -> Transaction:
    values: dict[str, object] = {
        "line_no": 2,
        "date": dt.date(2026, 3, 5),
        "amount": Decimal("-100.00"),
        "currency": "RUB",
        "purpose": "Аренда офиса",
        "counterparty": "ООО Альфа",
        "inn": "7701234567",
    }
    values.update(overrides)
    return Transaction(**values)  # type: ignore[arg-type]


# ----------------------------- parse_amount -------------------------------- #


@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        ("12 500,50", Decimal("12500.50")),
        ("-1 000,00", Decimal("-1000.00")),
        ("+1 000,5", Decimal("1000.50")),
        ("100", Decimal("100.00")),
        ("\u00a012\u00a0500,5", Decimal("12500.50")),
        ("1\u202f000,01", Decimal("1000.01")),
        ("99.99", Decimal("99.99")),
        (" 0,10 ", Decimal("0.10")),
    ],
)
def test_parse_amount_valid(raw: str, expected: Decimal) -> None:
    result = parse_amount(raw)
    assert result == expected
    assert isinstance(result, Decimal)
    assert result.as_tuple().exponent == -2


def test_parse_amount_negative_zero_is_normalized() -> None:
    assert str(parse_amount("-0,00")) == "0.00"


@pytest.mark.parametrize(
    "raw",
    ["", "abc", "12,345", "1.000,00", "12,", ",5", "1e5", "NaN", "Infinity",
     "--5", "12 500,50 руб", "1234567890123456"],
)
def test_parse_amount_invalid(raw: str) -> None:
    with pytest.raises(FieldValidationError, match="некорректная сумма"):
        parse_amount(raw)


# ------------------------------ parse_date --------------------------------- #


def test_parse_date_valid() -> None:
    assert parse_date(" 31.03.2026 ") == dt.date(2026, 3, 31)


@pytest.mark.parametrize("raw", ["", "31.02.2026", "2026-03-05", "05/03/2026", "завтра"])
def test_parse_date_invalid(raw: str) -> None:
    with pytest.raises(FieldValidationError, match="некорректная дата"):
        parse_date(raw)


# ----------------------------- validate_inn -------------------------------- #


@pytest.mark.parametrize("raw", ["7701234567", "770123456789", " 7701234567 "])
def test_validate_inn_valid(raw: str) -> None:
    assert validate_inn(raw) == raw.strip()


@pytest.mark.parametrize(
    "raw", ["", "123", "77012345678", "7701234567890", "77O1234567", "７７０１２３４５６７"]
)
def test_validate_inn_invalid(raw: str) -> None:
    with pytest.raises(FieldValidationError, match="ИНН"):
        validate_inn(raw)


# ----------------------------- parse_fields -------------------------------- #


def test_parse_fields_builds_transaction() -> None:
    fields = dict(zip(REQUIRED_COLUMNS, row(currency="rub")))
    tx = parse_fields(fields, line_no=7)
    assert tx == Transaction(
        line_no=7,
        date=dt.date(2026, 3, 5),
        amount=Decimal("-100000.00"),
        currency="RUB",
        purpose="Аренда офиса за март",
        counterparty="ООО Альфа",
        inn="7701234567",
    )


def test_parse_fields_collects_all_reasons() -> None:
    fields = dict(zip(REQUIRED_COLUMNS, row(amount="abc", purpose="  ", inn="123")))
    with pytest.raises(RowValidationError) as exc_info:
        parse_fields(fields, line_no=2)
    reasons = exc_info.value.reasons
    assert len(reasons) == 3
    assert "некорректная сумма" in reasons[0]
    assert reasons[1] == "пустое назначение платежа"
    assert "ИНН" in reasons[2]


def test_parse_fields_missing_columns_treated_as_empty() -> None:
    with pytest.raises(RowValidationError) as exc_info:
        parse_fields({}, line_no=2)
    assert len(exc_info.value.reasons) == 5  # всё, кроме контрагента


# --------------------------- iter_csv_records ------------------------------ #


def test_empty_file_raises() -> None:
    with pytest.raises(StatementFormatError, match="пуст"):
        list(iter_csv_records(io.StringIO("")))


def test_missing_required_column_raises() -> None:
    header = [c for c in REQUIRED_COLUMNS if c != "ИНН контрагента"]
    with pytest.raises(StatementFormatError, match="ИНН контрагента"):
        list(iter_csv_records(make_source([], header=header)))


def test_header_is_normalized_and_order_does_not_matter() -> None:
    header = ["\ufeff Контрагент ", *[c for c in REQUIRED_COLUMNS if c != "Контрагент"], ""]
    values = ["ООО Бета", *row()[:4], row()[5], ""]
    records = list(iter_csv_records(make_source([values], header=header)))
    assert records[0].fields["Контрагент"] == "ООО Бета"
    assert records[0].column_error is None


def test_line_numbers_with_blank_lines_and_multiline_fields() -> None:
    text = (
        ";".join(REQUIRED_COLUMNS) + "\n"
        + ";".join(row()) + "\n"                                       # строка 2
        + "\n"                                                          # строка 3
        + '05.03.2026;-1,00;RUB;"Оплата\nаренды";ООО;7701234567\n'      # строки 4–5
        + ";".join(row(inn="1")) + "\n"                                 # строка 6
    )
    records = list(iter_csv_records(io.StringIO(text, newline="")))
    assert [r.line_no for r in records] == [2, 4, 6]
    assert records[1].fields["Назначение платежа"] == "Оплата\nаренды"


def test_extra_non_empty_column_is_column_error() -> None:
    text = ";".join(REQUIRED_COLUMNS) + "\n05.03.2026;-1,00;RUB;Оплата; аренды;ООО;7701234567\n"
    (record,) = iter_csv_records(io.StringIO(text))
    assert record.column_error is not None
    assert "ожидалось 6, получено 7" in record.column_error


def test_trailing_semicolon_is_tolerated() -> None:
    text = ";".join(REQUIRED_COLUMNS) + "\n" + ";".join(row()) + ";\n"
    (record,) = iter_csv_records(io.StringIO(text))
    assert record.column_error is None


def test_short_row_is_column_error() -> None:
    text = ";".join(REQUIRED_COLUMNS) + "\n05.03.2026;-1,00\n"
    (record,) = iter_csv_records(io.StringIO(text))
    assert record.column_error is not None


# --------------------------- KeywordCategorizer ---------------------------- #


@pytest.mark.parametrize(
    ("purpose", "expected"),
    [
        ("АРЕНДА офиса за март", "Аренда офиса"),
        ("Перечисление зарплаты за февраль", "ФОТ"),
        ("Оплата за ТОПЛИВО по картам", "Транспорт"),
        ("Оплата услуг связи", "Прочее"),
    ],
)
def test_keyword_categorizer(purpose: str, expected: str) -> None:
    assert KeywordCategorizer(RULES).categorize(make_tx(purpose=purpose)) == expected


def test_categorizer_treats_yo_as_ye() -> None:
    categorizer = KeywordCategorizer({"расчет": "Расчёты"})
    assert categorizer.categorize(make_tx(purpose="Расчёт по договору")) == "Расчёты"


def test_longest_keyword_wins_regardless_of_order() -> None:
    categorizer = KeywordCategorizer({"аренд": "Аренда офиса", "аренда автомобиля": "Транспорт"})
    assert categorizer.categorize(make_tx(purpose="Аренда  автомобиля, март")) == "Транспорт"


def test_equal_length_keywords_use_dict_order() -> None:
    categorizer = KeywordCategorizer({"связь": "Связь", "налог": "Налоги"})
    assert categorizer.categorize(make_tx(purpose="Налог на связь")) == "Связь"


@pytest.mark.parametrize(
    "rules", [{"": "X"}, {"  ": "X"}, {"аренд": " "}, {"Аренд": "A", "аренд ": "B"}]
)
def test_categorizer_rejects_invalid_rules(rules: dict[str, str]) -> None:
    with pytest.raises(ValueError):
        KeywordCategorizer(rules)


def test_custom_categorizer_via_protocol() -> None:
    class ByInn:
        def categorize(self, transaction: Transaction) -> str:
            return "Налоги" if transaction.inn == "7727406020" else "Прочее"

    source = make_source([row(inn="7727406020", purpose="Уплата НДС")])
    result = process_statement(source, ByInn())
    assert result.transactions[0].category == "Налоги"


# ------------------------------ Deduplicator ------------------------------- #


def test_deduplicator_returns_original_line() -> None:
    dedup = Deduplicator()
    assert dedup.register(make_tx(line_no=2)) is None
    assert dedup.register(make_tx(line_no=5)) == 2
    assert len(dedup) == 1


# --------------------------- process_statement ----------------------------- #


def test_invalid_rows_are_rejected_and_logged(caplog: pytest.LogCaptureFixture) -> None:
    source = make_source([row(amount="сто рублей"), row(), row(purpose=""), row(inn="12345")])
    with caplog.at_level(logging.WARNING, logger="bank_reconciliation"):
        result = process_statement(source, RULES)
    assert [r.line_no for r in result.rejected] == [2, 4, 5]
    assert len(result.transactions) == 1
    assert result.rejected[0].raw["Сумма"] == "сто рублей"
    assert "Строка 2 отброшена" in caplog.text
    assert all(rec.levelno == logging.WARNING for rec in caplog.records)


def test_duplicates_are_removed() -> None:
    source = make_source([
        row(),
        row(),                        # точный дубль
        row(inn="770123456789"),       # отличается только ИНН — по ТЗ всё равно дубль
        row(amount="-100 000,01"),     # другая сумма — не дубль
        row(amount="-100000"),         # то же число в другой записи — дубль
    ])
    result = process_statement(source, RULES)
    assert [t.line_no for t in result.transactions] == [2, 5]
    assert [(d.line_no, d.original_line_no) for d in result.duplicates] == [(3, 2), (4, 2), (6, 2)]
    assert result.report.transaction_count == 2


def test_date_filter_is_inclusive() -> None:
    source = make_source([row(date="31.03.2026"), row(date="01.04.2026"), row(date="30.04.2026"),
                          row(date="01.05.2026")])
    result = process_statement(
        source, RULES, date_from=dt.date(2026, 4, 1), date_to=dt.date(2026, 4, 30)
    )
    assert [t.date.day for t in result.transactions] == [1, 30]
    assert result.filtered_out == 2


def test_invalid_date_range_raises() -> None:
    with pytest.raises(ValueError, match="позже"):
        process_statement(make_source([]), RULES,
                          date_from=dt.date(2026, 5, 1), date_to=dt.date(2026, 4, 1))


def test_currencies_are_not_mixed() -> None:
    source = make_source([row(), row(currency="USD", amount="-1 000,00", date="06.03.2026")])
    march = process_statement(source, RULES).report.periods[0]
    assert [(line.currency, line.expense) for line in march.categories] == [
        ("RUB", Decimal("-100000.00")),
        ("USD", Decimal("-1000.00")),
    ]
    assert [line.currency for line in march.totals] == ["RUB", "USD"]
    assert all(line.is_total and line.category == TOTAL_LABEL for line in march.totals)


def test_report_by_months_and_categories() -> None:
    source = make_source([
        row(date="02.04.2026", amount="-3 500,50", purpose="Топливо АИ-95"),
        row(),
        row(date="15.03.2026", amount="500 000,00", purpose="Оплата перевозки"),
        row(date="10.03.2026", amount="-250 000,00", purpose="Зарплата за февраль"),
    ])
    report = process_statement(source, RULES).report

    assert [p.period for p in report.periods] == ["2026-03", "2026-04"]
    march = report.periods[0]
    assert [(line.category, line.count, line.net) for line in march.categories] == [
        ("Аренда офиса", 1, Decimal("-100000.00")),
        ("Прочее", 1, Decimal("500000.00")),
        ("ФОТ", 1, Decimal("-250000.00")),
    ]
    (march_total,) = march.totals
    assert march_total.income == Decimal("500000.00")
    assert march_total.expense == Decimal("-350000.00")

    assert report.overall.period == ALL_PERIODS_LABEL
    (grand_total,) = report.overall.totals
    assert grand_total.count == 4
    assert grand_total.net == Decimal("146499.50")
    assert (report.first_date, report.last_date) == (dt.date(2026, 3, 5), dt.date(2026, 4, 2))


def test_amounts_never_become_float() -> None:
    result = process_statement(make_source([row(amount="0,10"), row(amount="0,20", date="06.03.2026")]),
                               RULES)
    total = result.report.overall.totals[0]
    assert isinstance(total.income, Decimal)
    assert total.income == Decimal("0.30")  # 0.1 + 0.2 во float дало бы 0.30000000000000004


def test_keep_transactions_false_still_builds_report() -> None:
    result = process_statement(make_source([row(), row(date="06.03.2026")]), RULES,
                               keep_transactions=False)
    assert result.transactions == []
    assert result.report.transaction_count == 2


def test_empty_statement_gives_empty_report() -> None:
    report = process_statement(make_source([]), RULES).report
    assert report.periods == ()
    assert report.overall.categories == () and report.overall.totals == ()
    assert report.first_date is None
    assert report_to_csv_text(report) == "Период;Категория;Валюта;Количество;Поступления;Списания;Сальдо\n"


# --------------------------------- Экспорт --------------------------------- #


@pytest.mark.parametrize(
    ("value", "expected"),
    [(Decimal("-12500.5"), "-12500,50"), (Decimal("0"), "0,00"),
     (Decimal("-0.00"), "0,00"), (Decimal("1E+3"), "1000,00")],
)
def test_format_amount(value: Decimal, expected: str) -> None:
    assert format_amount(value) == expected


EXPECTED_CSV = """\
Период;Категория;Валюта;Количество;Поступления;Списания;Сальдо
2026-03;Аренда офиса;RUB;1;0,00;-100000,00;-100000,00
2026-03;Прочее;RUB;1;500000,00;0,00;500000,00
2026-03;ФОТ;RUB;1;0,00;-250000,00;-250000,00
2026-03;Итого;RUB;3;500000,00;-350000,00;150000,00
2026-04;Транспорт;RUB;1;0,00;-3500,50;-3500,50
2026-04;Итого;RUB;1;0,00;-3500,50;-3500,50
Весь период;Аренда офиса;RUB;1;0,00;-100000,00;-100000,00
Весь период;Прочее;RUB;1;500000,00;0,00;500000,00
Весь период;Транспорт;RUB;1;0,00;-3500,50;-3500,50
Весь период;ФОТ;RUB;1;0,00;-250000,00;-250000,00
Весь период;Итого;RUB;4;500000,00;-353500,50;146499,50
"""

SAMPLE_ROWS = [
    row(),
    row(date="10.03.2026", amount="-250 000,00", purpose="Зарплата за февраль", inn="770123456789"),
    row(date="15.03.2026", amount="500 000,00", purpose="Оплата перевозки", inn="7802345678"),
    row(date="02.04.2026", amount="-3 500,50", purpose="Топливо АИ-95", inn="7708004767"),
]


def test_report_csv_exact_content() -> None:
    result = process_statement(make_source(SAMPLE_ROWS), RULES)
    assert report_to_csv_text(result.report) == EXPECTED_CSV


def test_report_is_deterministic_and_order_independent() -> None:
    first = process_statement(make_source(SAMPLE_ROWS), RULES).report
    second = process_statement(make_source(SAMPLE_ROWS), RULES).report
    shuffled = process_statement(make_source(list(reversed(SAMPLE_ROWS))), RULES).report
    assert report_to_csv_text(first).encode() == report_to_csv_text(second).encode()
    assert report_to_csv_text(first) == report_to_csv_text(shuffled)


def test_file_roundtrip_cp1251_to_utf8(tmp_path: Path) -> None:
    statement = tmp_path / "statement.csv"
    statement.write_bytes(make_source(SAMPLE_ROWS).getvalue().encode("cp1251"))
    report_path = tmp_path / "report.csv"

    result = process_statement_file(statement, RULES)
    export_report_csv(result.report, report_path)

    assert report_path.read_bytes().decode("utf-8") == EXPECTED_CSV


def test_file_with_wrong_encoding_raises(tmp_path: Path) -> None:
    statement = tmp_path / "broken.csv"
    header = ";".join(REQUIRED_COLUMNS).encode("cp1251")
    statement.write_bytes(header + b"\n05.03.2026;1,00;RUB;\x98;OOO;7701234567\n")
    with pytest.raises(StatementFormatError, match="cp1251"):
        process_statement_file(statement, RULES)

Запуск тестов и проверок:

pytest -q
mypy --strict bank_reconciliation.py
ruff check . && ruff format --check .

4. Пример использования

"""demo.py — обработка выписки за март–апрель 2026."""

import datetime as dt
import logging
from pathlib import Path

from bank_reconciliation import export_report_csv, process_statement_file

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")

SAMPLE = """\
Дата операции;Сумма;Валюта;Назначение платежа;Контрагент;ИНН контрагента
01.03.2026;-185 000,00;RUB;Арендная плата за март 2026 по дог. 14/А;ООО "БЦ Северный";7813456789
05.03.2026;-1 240 500,00;RUB;Зарплата за февраль 2026, реестр 3;АО "ТБанк";7710140679
05.03.2026;-1 240 500,00;RUB;Зарплата за февраль 2026, реестр 3;АО "ТБанк";7710140679
12.03.2026;-48 730,20;RUB;Топливо по топливным картам, сч. 1123;ООО "Газпромнефть-Корп";7728348420
14.03.2026;2 310 000,00;RUB;Оплата за перевозку груза по сч. 88;ООО "Альфа Трейд";5029123456
18.03.2026;abc;RUB;Оплата связи;ПАО "МТС";7740000076
20.03.2026;-12 000,00;RUB;;ИП Смирнов А.В.;502912345678
02.04.2026;-3 500,50;RUB;Топливо АИ-95;ООО "Лукойл-Интер-Кард";7705503750
"""

RULES = {
    "аренд": "Аренда офиса",
    "зарплат": "ФОТ",
    "топлив": "Транспорт",
    "перевозк": "Выручка",
    "связ": "Связь",
}

path = Path("statement_2026_03.csv")
path.write_bytes(SAMPLE.encode("cp1251"))  # имитация выгрузки из банка

result = process_statement_file(path, RULES, date_from=dt.date(2026, 3, 1))

print(f"Учтено: {len(result.transactions)}, отброшено: {len(result.rejected)}, "
      f"дублей: {len(result.duplicates)}")
for rejected in result.rejected:
    print(f"  строка {rejected.line_no}: {rejected.reason}")
for line in result.report.lines():
    marker = "==" if line.is_total else "  "
    print(f"{marker} {line.period:12} {line.category:14} {line.currency} "
          f"{line.count:>3} {line.net:>15}")

export_report_csv(result.report, "report_2026_03.csv")

Ожидаемый вывод (логи опущены):

Учтено: 5, отброшено: 2, дублей: 1
  строка 7: некорректная сумма: 'abc'
  строка 8: пустое назначение платежа
   2026-03      Аренда офиса   RUB   1      -185000.00
   2026-03      Выручка        RUB   1      2310000.00
   2026-03      Транспорт      RUB   1       -48730.20
   2026-03      ФОТ            RUB   1     -1240500.00
== 2026-03      Итого          RUB   4       835769.80
   2026-04      Транспорт      RUB   1        -3500.50
== 2026-04      Итого          RUB   1        -3500.50
   Весь период  Аренда офиса   RUB   1      -185000.00
   ...
== Весь период  Итого          RUB   5       832269.30

Интеграция со Streamlit:

import dataclasses
import io

import pandas as pd
import streamlit as st

from bank_reconciliation import process_statement, report_to_csv_text

uploaded = st.file_uploader("Выписка (CSV, cp1251)", type="csv")
if uploaded is not None:
    result = process_statement(io.TextIOWrapper(uploaded, encoding="cp1251", newline=""), RULES)
    st.dataframe(pd.DataFrame([dataclasses.asdict(line) for line in result.report.lines()]))
    if result.rejected:
        st.warning(f"Отброшено строк: {len(result.rejected)}")
        st.dataframe(pd.DataFrame(
            [{"строка": r.line_no, "причина": r.reason} for r in result.rejected]
        ))
    st.download_button("Скачать отчёт", report_to_csv_text(result.report).encode("utf-8"),
                       file_name="report.csv", mime="text/csv")

5. requirements.txt

Для самого модуля внешние зависимости не нужны, только стандартная библиотека. Для разработки понадобятся pytest, mypy и ruff. Для пригодности pandas и streamlit — это зависимости дашборда, а не модуля.

6. Пояснение логики и дальнейшие улучшения

Архитектура и поток данных

Нетривиальные решения

• Потоковая обработка. Все функции работают с Iterable[str], поэтому в тестах используется io.StringIO, а в проде — файл или загрузка из Streamlit. Память ReportBuilder зависит только от числа групп (месяц × категория × валюта), а не от числа строк.
  • Список транзакций по ТЗ всё же хранится: на 200 тыс. строк это десятки мегабайт, что приемлемо.
  • При keep_transactions=False остаётся только реестр ключей дублей.
• Номера строк. csv.reader.line_num указывает на последнюю физическую строку записи. Поэтому номер начала записи вычисляется как предыдущий line_num + 1. Так номера корректны и для многострочных назначений в кавычках: оператор откроет файл и сразу найдёт нужную строку.
• Контроль числа колонок. Неэкранированный ; в назначении платежа сдвигает колонки. Без этой проверки ИНН мог бы «случайно» оказаться валидным, и в учёт попали бы неверные данные.
• Decimal. Строгая регулярка не пропускает NaN, Infinity, экспоненту и неоднозначные 1.000,00 / 12,345. Ограничение в 15 цифр гарантирует, что quantize никогда не выйдет за точность контекста. Все суммы имеют ровно 2 знака, поэтому сложение точное и не зависит от порядка. Отрицательный ноль нормализуется, чтобы в отчёте не появлялось -0,00.
• Детерминизм.
  • Все коллекции отчёта явно отсортированы (месяц ГГГГ-ММ сортируется лексикографически).
  • В CSV зафиксирован lineterminator="\n", нет временных меток.
  • Тест test_report_is_deterministic_and_order_independent проверяет, что отчёт не меняется даже при перестановке строк во входном файле.
• Расширяемость категоризации.
  • Новое правило — это новая запись в словаре.
  • Новый тип правил (по ИНН, по знаку суммы, по регулярке) — это класс с методом categorize. Парсинг при этом не меняется, пример — test_custom_categorizer_via_protocol.
  • Приоритет «самое длинное ключевое слово» делает результат независимым от случайного порядка ключей в конфиге.
• Ошибки.
  • Фатальные проблемы всего файла (StatementFormatError) отделены от проблем отдельной строки (RowValidationError): первые прерывают обработку, вторые собираются в rejected и логируются на уровне WARNING.
  • Дубли логируются на уровне INFO: это не ошибка данных.

Компромиссы и риски

• Легитимные одинаковые платежи. Ключ дубликата по ТЗ может «съесть» два реально разных платежа: например, два одинаковых перевода одному контрагенту в один день. Поэтому все дубли возвращаются в result.duplicates со ссылкой на оригинал, чтобы бухгалтер мог их проверить. Если в выгрузке есть номер документа или ID операции, его стоит добавить в ключ — это самое ценное улучшение.
• Сопоставление подстрокой. Оно даёт O(правила × длина текста) на строку, что для десятков правил несущественно. При тысячах правил стоит перейти на алгоритм Ахо — Корасик.
• Пустой ИНН. Строки с пустым ИНН (например, банковские комиссии) по ТЗ отбраковываются. Если это нежелательно, можно добавить отдельную категорию, но это вопрос к финотделу.

Что можно улучшить дальше

1. Добавить в ключ дубликата номер документа или ID операции из выгрузки.
2. Проверять контрольную сумму ИНН — по той же схеме, как отдельный валидатор.
3. Поддержать «шапку» выписки (реквизиты счёта над таблицей): автоматически искать строку заголовка.
4. Сверять остатки: входящий остаток плюс обороты должны давать исходящий остаток из выписки.
5. Загружать правила из JSON/TOML-конфига с валидацией, чтобы финотдел мог менять их без релиза.
6. Добавить property-based тесты (hypothesis) для parse_amount и инвариантов отчёта: сумма по категориям равна итогу, итог за весь период равен сумме месячных итогов.