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 ingestion import guided_imports from .forms import ( AccountForm, AnnotationForm, CategoryForm, ManualTransactionForm, PayeeForm, ImportMappingForm, 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 imports_index(request): if request.method == "POST": action = request.POST.get("action") if action == "scan": discovered = guided_imports.discover_imports() files = [{ "name": row.name, "profile": row.profile, "sections": list(row.sections), "state": "NEEDS_ATTENTION" if row.code else "MAPPING_REQUIRED", "code": row.code, "counts": None, } for row in discovered] scannable = [_import_metadata(row) for row in discovered if not row.code] return _render_imports(request, files=files, scan_token=signing.dumps( scannable, salt="guided-imports-scan", compress=True, ) if scannable else None) elif action in {"prepare", "commit"}: try: payload = _load_import_token(request, prepared=action == "commit") except (signing.BadSignature, ValueError, TypeError): return render(request, "core/imports/index.html", { "error": "This import state is invalid or expired. Scan again before continuing.", }, status=400) if action == "prepare": return _prepare_import_files(request, payload) else: return _commit_import_files(request, payload) else: return render(request, "core/imports/index.html", { "error": "Choose an import action. Scan again before continuing.", }, status=400) return _render_imports(request, files=[]) def _render_imports(request, *, files, **context): """Only immediate form responses hold signed workflow metadata.""" displayed = [] mapping_index = 0 for row in files: row = dict(row) if row["state"] == "MAPPING_REQUIRED": row["form"] = ImportMappingForm(sections=row["sections"], prefix=f"file-{mapping_index}") mapping_index += 1 displayed.append(row) return render(request, "core/imports/index.html", {**context, "files": displayed}) def _import_metadata(row, *, prepared=False): metadata = { "name": row.name, "document_hash": row.document_hash, "profile": row.profile, "sections": list(row.sections), } if prepared: metadata.update(batch_id=row.batch_id, revision=row.revision) return metadata def _load_import_token(request, *, prepared): field, salt = ("prepared_token", "guided-imports-prepared") if prepared else ("scan_token", "guided-imports-scan") payload = signing.loads(request.POST.get(field, ""), salt=salt, max_age=900) keys = {"name", "document_hash", "profile", "sections"} if prepared: keys |= {"batch_id", "revision"} if not isinstance(payload, list) or not payload: raise ValueError for row in payload: if not isinstance(row, dict) or set(row) != keys: raise ValueError if any(not isinstance(row[key], str) or not row[key] for key in ("name", "document_hash", "profile")): raise ValueError if not isinstance(row["sections"], list) or not row["sections"] or any( not isinstance(section, str) or not section for section in row["sections"] ): raise ValueError if prepared and (not isinstance(row["batch_id"], str) or type(row["revision"]) is not int): raise ValueError return payload def _import_summary(row): return { "name": row.name, "profile": row.profile, "sections": list(row.sections), "state": row.state, "code": row.code, "counts": row.counts, } def _import_failure(metadata, code="IMPORT_FAILED"): return { "name": metadata["name"], "profile": metadata["profile"], "sections": metadata["sections"], "state": "NEEDS_ATTENTION", "code": code, "counts": {"created": 0, "matched": 0, "skipped": 0}, } def _prepare_import_files(request, payload): results, ready = [], [] for index, metadata in enumerate(payload): form = ImportMappingForm(request.POST, sections=metadata["sections"], prefix=f"file-{index}") if not form.is_valid(): results.append(_import_failure(metadata, "INVALID_SECTION_MAPPING")) continue discovered = guided_imports.DiscoveredFile( **{**metadata, "sections": tuple(metadata["sections"])}, code=None, ) try: result = guided_imports.prepare_import( discovered=discovered, section_mapping=form.section_mapping(), actor=LOCAL_ACTOR, ) except Exception: # Source-bearing errors must stay local and cannot stop another file. results.append(_import_failure(metadata)) continue results.append(_import_summary(result)) if result.state in {"READY", "ALREADY_IMPORTED"}: ready.append(_import_metadata(result, prepared=True)) return _render_imports(request, files=results, prepared_token=signing.dumps( ready, salt="guided-imports-prepared", compress=True, ) if ready else None) def _commit_import_files(request, payload): results = [] for metadata in payload: prepared = guided_imports.PreparedImport( **{**metadata, "sections": tuple(metadata["sections"])}, state="READY", counts={"created": 0, "matched": 0, "skipped": 0}, code=None, staged=True, ) try: result = guided_imports.commit_prepared_import(prepared=prepared, actor=LOCAL_ACTOR) except Exception: results.append(_import_failure(metadata)) continue results.append(_import_summary(result)) return _render_imports(request, files=results) 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) ]