gizmoduck/core/views.py

from dataclasses import dataclass
from datetime import date
from decimal import Decimal

from django.contrib import messages
from django.core import signing
from django.core.paginator import Paginator
from django.db import DatabaseError, connection
from django.http import HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.utils import timezone
from django.views.decorators.http import require_POST

from finance.models import Account, Category, Payee, Transaction
from finance.exports import build_export
from finance.queries import (
    QueryInputError, TransactionFilters, filtered_transactions,
    listed_accounts, listed_categories, listed_payees,
)
from finance.reconciliation import (
    checkpoint_history,
    latest_checkpoint_statuses,
    ledger_balance_through,
)
from finance.reports import account_balances, available_currencies, monthly_summary, spending_by_category
from finance.services import (
    Actor, DomainError, annotate_transaction, archive, create_account,
    create_category, create_payee, create_reconciliation_checkpoint,
    record_transaction,
)

from .forms import (
    AccountForm, AnnotationForm, CategoryForm, ManualTransactionForm, PayeeForm,
    ReconciliationForm, ReportSelectionForm,
)


LOCAL_ACTOR = Actor(kind="HUMAN", identifier="local-owner")
DOMAIN_MESSAGES = {
    "archived_reference": "That reference is archived and cannot be selected.",
    "category_kind_mismatch": "The category does not match the transaction kind.",
    "conflict": "A record with these values already exists.",
    "invalid_amount": "Enter a valid amount with at most two decimal places.",
    "invalid_name": "Enter a name.",
    "not_found": "The requested record was not found.",
    "validation_error": "The values do not satisfy the financial rules.",
    "zero_amount": "The amount cannot be zero.",
}
UNKNOWN_DOMAIN_MESSAGE = "Unable to save this change."


@dataclass(frozen=True)
class ReconciliationPreview:
    account: Account
    as_of_date: date
    institution_balance: Decimal
    ledger_balance: Decimal

    @property
    def difference(self) -> Decimal:
        return self.institution_balance - self.ledger_balance


def health(request):
    try:
        with connection.cursor() as cursor:
            cursor.execute("SELECT 1")
            cursor.fetchone()
    except DatabaseError:
        return JsonResponse({"status": "unavailable"}, status=503)
    return JsonResponse({"status": "ok"})


def home(request):
    return _monthly_report(request, "core/dashboard.html")


def monthly_report(request):
    return _monthly_report(request, "core/reports/monthly.html")


def reconciliation_index(request):
    form = ReconciliationForm(request.POST or None)
    preview = None
    preview_token = None
    if request.method == "POST" and form.is_valid():
        action = request.POST.get("action")
        values = form.cleaned_data
        preview_inputs = {
            "account": str(values["account"]),
            "as_of_date": values["as_of_date"].isoformat(),
            "institution_balance": format(values["institution_balance"], ".2f"),
        }
        if action == "preview":
            selected_account = get_object_or_404(
                Account,
                pk=values["account"],
                archived=False,
            )
            preview = ReconciliationPreview(
                account=selected_account,
                as_of_date=values["as_of_date"],
                institution_balance=values["institution_balance"],
                ledger_balance=ledger_balance_through(
                    selected_account.pk,
                    values["as_of_date"],
                ),
            )
            preview_token = signing.dumps(preview_inputs, salt="reconciliation-preview")
        elif action == "confirm":
            try:
                confirmed_inputs = signing.loads(
                    request.POST.get("preview_token", ""),
                    salt="reconciliation-preview",
                )
            except signing.BadSignature:
                confirmed_inputs = None
            if confirmed_inputs != preview_inputs:
                form.add_error(None, "Preview these values before confirming the checkpoint.")
            else:
                try:
                    saved = create_reconciliation_checkpoint(
                        account_id=values["account"],
                        as_of_date=values["as_of_date"],
                        institution_balance=values["institution_balance"],
                        actor=LOCAL_ACTOR,
                    )
                except DomainError as exc:
                    form.add_error(None, _domain_message(exc))
                else:
                    messages.success(request, "Reconciliation checkpoint recorded.")
                    return redirect("reconciliation-history", account_id=saved.account_id)
    return render(request, "core/reconciliations/index.html", {
        "form": form,
        "preview": preview,
        "preview_token": preview_token,
        "latest_statuses": _latest_reconciliation_statuses(),
    })


def reconciliation_history(request, account_id):
    selected_account = get_object_or_404(Account, pk=account_id)
    return render(request, "core/reconciliations/history.html", {
        "account": selected_account,
        "history": checkpoint_history(selected_account.pk),
    })


def export_index(request):
    return render(request, "core/exports.html")


@require_POST
def export_download(request):
    artifact = build_export()
    response = HttpResponse(artifact.content, content_type="application/zip")
    response["Content-Disposition"] = f'attachment; filename="{artifact.filename}"'
    return response


def transaction_list(request):
    filter_errors = {}
    try:
        filters = TransactionFilters.from_mapping(request.GET)
        rows = filtered_transactions(filters)
    except QueryInputError as exc:
        filters = TransactionFilters()
        rows = Transaction.objects.none()
        filter_errors = {field: _query_message(code) for field, code in exc.errors.items()}
    page = Paginator(rows, 50).get_page(request.GET.get("page"))
    pagination_parameters = request.GET.copy()
    pagination_parameters.pop("page", None)
    return render(request, "core/transactions/list.html", {
        "page": page,
        "filters": filters,
        "filter_errors": filter_errors,
        "accounts": listed_accounts(),
        "categories": listed_categories(),
        "payees": listed_payees(),
        "kinds": Transaction.Kind.choices,
        "pagination_query": pagination_parameters.urlencode(),
    })


def transaction_detail(request, transaction_id):
    row = get_object_or_404(
        Transaction.objects.select_related(
            "account", "payee", "category", "statement", "refund_of",
            "installment_plan", "outgoing_transfer__incoming__account",
            "incoming_transfer__outgoing__account",
        ).prefetch_related("source_records", "refunds"),
        pk=transaction_id,
    )
    return render(request, "core/transactions/detail.html", {"transaction": row})


def transaction_create(request):
    form = ManualTransactionForm(request.POST or None)
    if request.method == "POST" and form.is_valid():
        values = form.cleaned_data
        try:
            row = record_transaction(
                account_id=values["account"], amount=values["amount"],
                posting_date=values["posting_date"], occurrence_date=values["occurrence_date"],
                kind=values["kind"], description=values["description"],
                payee_id=values["payee"], category_id=values["category"], actor=LOCAL_ACTOR,
            )
        except DomainError as exc:
            form.add_error(None, _domain_message(exc))
        else:
            messages.success(request, "Transaction recorded.")
            return redirect("transaction-detail", transaction_id=row.pk)
    return render(request, "core/transactions/form.html", {"form": form, "title": "New transaction"})


def transaction_annotate(request, transaction_id):
    row = get_object_or_404(Transaction.objects.select_related("payee", "category"), pk=transaction_id)
    form = AnnotationForm(request.POST or None, transaction=row)
    if request.method == "POST" and form.is_valid():
        changes = form.changes()
        if not changes:
            return redirect("transaction-detail", transaction_id=row.pk)
        try:
            annotate_transaction(transaction_id=row.pk, actor=LOCAL_ACTOR, **changes)
        except DomainError as exc:
            form.add_error(None, _domain_message(exc))
        else:
            messages.success(request, "Transaction annotations updated.")
            return redirect("transaction-detail", transaction_id=row.pk)
    return render(request, "core/transactions/form.html", {
        "form": form, "title": "Edit transaction annotations", "transaction": row,
    })


def account_list(request):
    return _reference_list(request, listed_accounts(), "Accounts", "account-create", "account-archive")


def category_list(request):
    return _reference_list(request, listed_categories(), "Categories", "category-create", "category-archive")


def payee_list(request):
    return _reference_list(request, listed_payees(), "Payees", "payee-create", "payee-archive")


def account_create(request):
    return _reference_create(request, AccountForm, create_account, "New account", "account-list")


def category_create(request):
    return _reference_create(request, CategoryForm, create_category, "New category", "category-list")


def payee_create(request):
    return _reference_create(request, PayeeForm, create_payee, "New payee", "payee-list")


@require_POST
def account_archive(request, reference_id):
    return _reference_archive(request, Account, reference_id, "account-list")


@require_POST
def category_archive(request, reference_id):
    return _reference_archive(request, Category, reference_id, "category-list")


@require_POST
def payee_archive(request, reference_id):
    return _reference_archive(request, Payee, reference_id, "payee-list")


def _reference_list(request, rows, title, create_url, archive_url):
    return render(request, "core/references/list.html", {
        "title": title,
        "rows": rows,
        "create_url": create_url,
        "archive_url": archive_url,
    })


def _reference_create(request, form_class, service, title, redirect_name):
    form = form_class(request.POST or None)
    if request.method == "POST" and form.is_valid():
        try:
            service(actor=LOCAL_ACTOR, **form.cleaned_data)
        except DomainError as exc:
            form.add_error(None, _domain_message(exc))
        else:
            messages.success(request, f"{title.removeprefix('New ')} created.")
            return redirect(redirect_name)
    return render(request, "core/references/form.html", {"form": form, "title": title})


def _reference_archive(request, model, reference_id, redirect_name):
    item = get_object_or_404(model, pk=reference_id)
    try:
        archive(item, actor=LOCAL_ACTOR)
    except DomainError as exc:
        messages.error(request, _domain_message(exc))
    else:
        messages.success(request, "Reference archived.")
    return redirect(redirect_name)


def _domain_message(error):
    return DOMAIN_MESSAGES.get(error.code, UNKNOWN_DOMAIN_MESSAGE)


def _query_message(code):
    return {
        "invalid_date": "Enter a valid date.",
        "before_date_from": "The end date must be on or after the start date.",
        "invalid_uuid": "Select a valid reference.",
        "invalid_choice": "Select a valid choice.",
        "invalid_currency": "Enter a three-letter uppercase currency code.",
    }.get(code, "Invalid filter.")


def _monthly_report(request, template_name):
    currencies = available_currencies() or ("BRL",)
    parameters = {
        "month": request.GET.get("month", timezone.localdate().strftime("%Y-%m")),
        "currency": request.GET.get("currency", currencies[0]),
    }
    form = ReportSelectionForm(parameters, currencies=currencies)
    context = {"selection_form": form}
    if template_name == "core/dashboard.html":
        context["reconciliation_statuses"] = latest_checkpoint_statuses("BRL")
    status = 200
    if form.is_valid():
        month = form.cleaned_data["month"]
        currency = form.cleaned_data["currency"]
        context.update({
            "summary": monthly_summary(month, currency),
            "spending_rows": spending_by_category(month, currency),
            "balances": account_balances(month, currency),
        })
    else:
        status = 400
    return render(request, template_name, context, status=status)


def _latest_reconciliation_statuses():
    currencies = (
        Account.objects.filter(archived=False)
        .order_by("currency")
        .values_list("currency", flat=True)
        .distinct()
    )
    return [
        status
        for currency in currencies
        for status in latest_checkpoint_statuses(currency)
    ]