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),
+}