diff --git a/expenses_manager/expenses/csv_export.py b/expenses_manager/expenses/csv_export.py new file mode 100644 index 0000000..abb4032 --- /dev/null +++ b/expenses_manager/expenses/csv_export.py @@ -0,0 +1,201 @@ +"""Dialectos CSV, formateo de celdas y volcado de filas en streaming. + +Va en módulo propio y no en views.py (ya anotado como deuda por tamaño): así el +dialecto y el formateo se prueban sin levantar un cliente HTTP. +""" + +import csv +from dataclasses import dataclass +from datetime import date as date_cls + +from django.db.models import Q + +from .models import Expense, Income, Transfer + + +class _Echo: + """Objeto tipo fichero que devuelve lo que le escriben, para csv.writer.""" + + def write(self, value): + return value + + +@dataclass(frozen=True) +class CsvDialect: + delimiter: str + decimal_sep: str + date_format: str + bool_true: str + bool_false: str + bom: bool + + +# Abre con doble clic en el Excel español. +EXCEL_ES = CsvDialect(";", ",", "%d/%m/%Y", "Sí", "No", bom=True) +# Salida RFC-4180 para pandas o para reimportar. +STANDARD = CsvDialect(",", ".", "%Y-%m-%d", "true", "false", bom=False) + +DIALECTS = {"excel": EXCEL_ES, "std": STANDARD} +DEFAULT_DIALECT = "excel" + + +def fmt_amount(value, dialect): + """Decimal -> texto plano, sin separador de miles a propósito. + + El formateo por locale de Django metería "1.234,56" y partiría la fila en + dos en el dialecto de Excel. Aquí no se usa floatformat ni + django.utils.formats. None sale como celda vacía: "no hay comisión" no es + lo mismo que una comisión de cero. + """ + if value is None: + return "" + + return f"{value:.2f}".replace(".", dialect.decimal_sep) + + +def fmt_date(value, dialect): + return value.strftime(dialect.date_format) if value else "" + + +def fmt_bool(value, dialect): + return dialect.bool_true if value else dialect.bool_false + + +def stream_csv(header, rows, dialect): + """Va soltando el CSV fila a fila, para StreamingHttpResponse.""" + writer = csv.writer(_Echo(), delimiter=dialect.delimiter, quoting=csv.QUOTE_MINIMAL) + + # El BOM se emite como texto y lo codifica la respuesta con su charset. + if dialect.bom: + yield "" + + yield writer.writerow(header) + for row in rows: + yield writer.writerow(row) + + +def csv_filename(stem, date_from, date_to): + """gastos_2026-01-01_2026-10-01.csv, o con la fecha de hoy si no hay rango.""" + span = [d.isoformat() for d in (date_from, date_to) if d] + + return "_".join([stem, *(span or [date_cls.today().isoformat()])]) + ".csv" + + +def _by_date(queryset, date_from, date_to): + """Recorta el rango y ordena. + + El desempate por id no es adorno: un order_by solo por fecha puede salir en + distinto orden en SQLite local y en el PostgreSQL del NAS. + """ + if date_from: + queryset = queryset.filter(date__gte=date_from) + if date_to: + queryset = queryset.filter(date__lte=date_to) + + return queryset.order_by("date", "id") + + +# chunk_size es obligatorio al combinar iterator() con prefetch_related: sin él +# el prefetch se ignora y las etiquetas saldrían vacías o con un N+1. +CHUNK = 2000 + + +def expenses_export(user, date_from, date_to, account, dialect): + queryset = ( + Expense.objects.filter(owner=user) + .select_related("account", "category", "category__parent") + .prefetch_related("tags") + ) + if account: + queryset = queryset.filter(account=account) + + header = [ + "Fecha", + "Importe", + "Categoría", + "Categoría padre", + "Cuenta", + "Concepto", + "Etiquetas", + "Traspaso", + ] + + def rows(): + for expense in _by_date(queryset, date_from, date_to).iterator( + chunk_size=CHUNK + ): + category = expense.category + yield [ + fmt_date(expense.date, dialect), + fmt_amount(expense.amount, dialect), + category.name, + # La categoría padre sale del JOIN, no recorriendo la jerarquía + # por fila. Si algún día se quiere la ruta completa a cualquier + # profundidad, eso es un dict en memoria, nunca más SQL. + category.parent.name if category.parent else "", + expense.account.name, + expense.description, + # Unidas con | porque la coma y el punto y coma son + # delimitadores en uno u otro dialecto. + "|".join(tag.name for tag in expense.tags.all()), + fmt_bool(expense.is_transfer, dialect), + ] + + return header, rows() + + +def incomes_export(user, date_from, date_to, account, dialect): + queryset = Income.objects.filter(owner=user).select_related("account") + if account: + queryset = queryset.filter(account=account) + + header = ["Fecha", "Importe", "Cuenta", "Concepto", "Traspaso"] + + def rows(): + for income in _by_date(queryset, date_from, date_to).iterator( + chunk_size=CHUNK + ): + yield [ + fmt_date(income.date, dialect), + fmt_amount(income.amount, dialect), + income.account.name, + income.name, + fmt_bool(income.is_transfer, dialect), + ] + + return header, rows() + + +def transfers_export(user, date_from, date_to, account, dialect): + queryset = Transfer.objects.filter(owner=user).select_related( + "from_account", "to_account" + ) + # Filtrar por cuenta es "traspasos en los que participa": da igual si entra + # o sale, el movimiento está en el extracto de esa cuenta. + if account: + queryset = queryset.filter(Q(from_account=account) | Q(to_account=account)) + + header = ["Fecha", "Importe", "Cuenta origen", "Cuenta destino", "Comisión"] + + def rows(): + for transfer in _by_date(queryset, date_from, date_to).iterator( + chunk_size=CHUNK + ): + yield [ + fmt_date(transfer.date, dialect), + fmt_amount(transfer.amount, dialect), + transfer.from_account.name, + transfer.to_account.name, + fmt_amount(transfer.fee_amount, dialect), + ] + + return header, rows() + + +# kind -> (nombre de fichero, constructor). Los nombres salen en Descargas, así +# que van en español. +EXPORTS = { + "expenses": ("gastos", expenses_export), + "incomes": ("ingresos", incomes_export), + "transfers": ("traspasos", transfers_export), +}