307 lines
12 KiB
Python
307 lines
12 KiB
Python
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import COOKIE, session_valid
|
|
from app.db import get_session
|
|
from app.engine.loans import add_months
|
|
from app.engine.recurrence import occurrences
|
|
from app.formats import de_label, eur
|
|
from app.models.tables import (Account, Category, PlannedItem,
|
|
ProjectionResult, RecurringItem,
|
|
ScenarioLoan, ScenarioModifier,
|
|
ScenarioPlannedItem, Transaction)
|
|
from app.routers.imports import list_imports as _list_imports
|
|
from app.routers.imports import preview as _preview_import
|
|
from app.routers.planning import list_loans as _list_loans
|
|
from app.routers.planning import list_planned as _list_planned
|
|
from app.routers.planning import list_recurring as _list_recurring
|
|
from app.routers.planning import recurring_suggestions as _recurring_suggestions
|
|
from app.routers.scenarios import list_scenarios as _list_scenarios
|
|
from app.routers.transactions import count_transactions, list_transactions
|
|
from app.services.balances import account_balance, total_balance
|
|
from app.version import get_version
|
|
|
|
PAGE_SIZE = 50
|
|
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
templates.env.globals["app_version"] = get_version()
|
|
templates.env.filters["eur"] = eur
|
|
templates.env.filters["de_label"] = de_label
|
|
router = APIRouter()
|
|
|
|
|
|
def gui_session(request: Request):
|
|
if not session_valid(request.cookies.get(COOKIE)):
|
|
raise _redirect()
|
|
|
|
|
|
def _redirect():
|
|
exc = HTTPException(302)
|
|
exc.headers = {"Location": "/login"}
|
|
return exc
|
|
|
|
|
|
def _upcoming_items(session: Session, today: date, days: int = 30, limit: int = 10) -> list[dict]:
|
|
"""Nächste Posten aus RecurringItems (Termine) und PlannedItems der nächsten `days` Tage."""
|
|
end = today + timedelta(days=days)
|
|
items: list[dict] = []
|
|
for r in session.execute(select(RecurringItem)).scalars():
|
|
for d in occurrences(r.rhythm, r.due_day, today, end, r.start_date, r.end_date):
|
|
items.append({"date": d, "name": r.name, "amount": Decimal(r.amount)})
|
|
for p in session.execute(
|
|
select(PlannedItem).where(PlannedItem.due >= today, PlannedItem.due <= end)
|
|
).scalars():
|
|
items.append({"date": p.due, "name": p.name, "amount": Decimal(p.amount)})
|
|
items.sort(key=lambda i: i["date"])
|
|
return items[:limit]
|
|
|
|
|
|
def _latest_projection(session: Session) -> ProjectionResult | None:
|
|
return session.execute(
|
|
select(ProjectionResult).order_by(ProjectionResult.computed_at.desc())
|
|
).scalars().first()
|
|
|
|
|
|
def _opt_int(value: str | None) -> int | None:
|
|
"""Wandelt einen GUI-Query-Parameter tolerant in int|None um. HTML-Formulare
|
|
senden bei nicht ausgefüllten Feldern leere Strings statt gar keinen
|
|
Parameter - das darf nie zu einem 422 fuehren, sondern degradiert zu None."""
|
|
if value is None or not value.strip():
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _opt_date(value: str | None) -> date | None:
|
|
"""Wie _opt_int, nur fuer date-Felder (ISO-Format aus <input type=date>)."""
|
|
if value is None or not value.strip():
|
|
return None
|
|
try:
|
|
return date.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _month_starts_desc(session: Session, today: date) -> list[date]:
|
|
"""Liefert die Monatsersten von "heute" rueckwaerts bis zum Monat der
|
|
fruehesten bestaetigten Buchung, absteigend sortiert (heute zuerst).
|
|
Ohne bestaetigte Buchungen: leere Liste."""
|
|
earliest = session.execute(
|
|
select(func.min(Transaction.booking_date)).where(Transaction.status == "confirmed")
|
|
).scalar_one_or_none()
|
|
if earliest is None:
|
|
return []
|
|
start = date(earliest.year, earliest.month, 1)
|
|
current = date(today.year, today.month, 1)
|
|
months = []
|
|
m = current
|
|
while m >= start:
|
|
months.append(m)
|
|
m = add_months(m, -1)
|
|
return months
|
|
|
|
|
|
@router.get("/", dependencies=[Depends(gui_session)])
|
|
def index(request: Request, session: Session = Depends(get_session)):
|
|
accounts = session.execute(select(Account)).scalars().all()
|
|
balances = [(a, account_balance(session, a)) for a in accounts]
|
|
return templates.TemplateResponse(request, "index.html", {
|
|
"balances": balances,
|
|
"total": total_balance(session),
|
|
"upcoming": _upcoming_items(session, date.today()),
|
|
"projection": _latest_projection(session),
|
|
})
|
|
|
|
|
|
@router.get("/buchungen", dependencies=[Depends(gui_session)])
|
|
def buchungen(
|
|
request: Request,
|
|
account_id: str | None = None,
|
|
date_from: str | None = None,
|
|
date_to: str | None = None,
|
|
category_id: str | None = None,
|
|
q: str | None = None,
|
|
status: str = "confirmed",
|
|
page: int = 1,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
# HTML-Filterformular sendet leere Strings fuer nicht ausgefuellte Felder -
|
|
# daher hier bewusst str|None statt int|None/date|None und tolerante
|
|
# Konvertierung, statt FastAPI vor dem Handler mit 422 abbrechen zu lassen.
|
|
# Die API-Routen (/api/transactions) bleiben strikt typisiert.
|
|
account_id = _opt_int(account_id)
|
|
date_from = _opt_date(date_from)
|
|
date_to = _opt_date(date_to)
|
|
category_id = _opt_int(category_id)
|
|
page = max(1, page)
|
|
total = count_transactions(
|
|
account_id=account_id,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
category_id=category_id,
|
|
q=q,
|
|
status=status,
|
|
session=session,
|
|
)
|
|
total_pages = max(1, -(-total // PAGE_SIZE)) # ceil division
|
|
page = min(page, total_pages)
|
|
txs = list_transactions(
|
|
account_id=account_id,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
category_id=category_id,
|
|
q=q,
|
|
status=status,
|
|
limit=PAGE_SIZE,
|
|
offset=(page - 1) * PAGE_SIZE,
|
|
session=session,
|
|
)
|
|
accounts = session.execute(select(Account)).scalars().all()
|
|
return templates.TemplateResponse(request, "transactions.html", {
|
|
"transactions": txs,
|
|
"accounts": accounts,
|
|
"account_names": {a.id: a.name for a in accounts},
|
|
"categories": session.execute(select(Category)).scalars().all(),
|
|
"filters": {
|
|
"account_id": account_id,
|
|
"date_from": date_from,
|
|
"date_to": date_to,
|
|
"category_id": category_id,
|
|
"q": q or "",
|
|
"status": status,
|
|
},
|
|
"page": page,
|
|
"total_pages": total_pages,
|
|
})
|
|
|
|
|
|
@router.get("/salden", dependencies=[Depends(gui_session)])
|
|
def salden_page(
|
|
request: Request,
|
|
stichtag: str | None = None,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
# Lenient wie die Filter auf /buchungen: ein leerer oder kaputter
|
|
# stichtag-Parameter fuehrt nie zu 422, sondern faellt auf "heute" zurueck.
|
|
today = date.today()
|
|
parsed_stichtag = _opt_date(stichtag) or today
|
|
|
|
accounts = session.execute(select(Account)).scalars().all()
|
|
balances = [(a, account_balance(session, a, at=parsed_stichtag)) for a in accounts]
|
|
total = sum((b for _, b in balances), Decimal("0"))
|
|
|
|
month_rows = []
|
|
for month in _month_starts_desc(session, today):
|
|
# Saldo am Monatsanfang = Stand am ENDE DES VORTAGS (letzter Tag des
|
|
# Vormonats), NICHT account_balance(at=month) - sonst wuerden
|
|
# Buchungen AM 1. bereits in den "Anfangs"-Saldo einfliessen und den
|
|
# Ausweis verzerren. Siehe Fussnote im Template ("Stand: Ende des
|
|
# Vortags").
|
|
vortag = month - timedelta(days=1)
|
|
values = [(a, account_balance(session, a, at=vortag)) for a in accounts]
|
|
month_rows.append({
|
|
"month": month,
|
|
"values": values,
|
|
"total": sum((v for _, v in values), Decimal("0")),
|
|
})
|
|
|
|
accounts_without_anchor = [
|
|
a for a in accounts if a.anchor_date is None or a.anchor_balance is None
|
|
]
|
|
|
|
return templates.TemplateResponse(request, "salden.html", {
|
|
"stichtag": parsed_stichtag,
|
|
"balances": balances,
|
|
"total": total,
|
|
"accounts": accounts,
|
|
"months": month_rows,
|
|
"accounts_without_anchor": accounts_without_anchor,
|
|
})
|
|
|
|
|
|
@router.get("/import", dependencies=[Depends(gui_session)])
|
|
def import_page(request: Request, session: Session = Depends(get_session)):
|
|
return templates.TemplateResponse(request, "import.html", {
|
|
"statements": _list_imports(session=session),
|
|
})
|
|
|
|
|
|
@router.get("/import/list", dependencies=[Depends(gui_session)])
|
|
def import_list_fragment(request: Request, session: Session = Depends(get_session)):
|
|
return templates.TemplateResponse(request, "_import_list.html", {
|
|
"statements": _list_imports(session=session),
|
|
})
|
|
|
|
|
|
@router.get("/import/{statement_id}/preview-fragment", dependencies=[Depends(gui_session)])
|
|
def import_preview_fragment(statement_id: int, request: Request,
|
|
session: Session = Depends(get_session)):
|
|
result = _preview_import(statement_id, session=session)
|
|
return templates.TemplateResponse(request, "_preview_table.html", {
|
|
"statement": result.statement,
|
|
"transactions": result.transactions,
|
|
"balance_ok": result.balance_ok,
|
|
"duplicates": result.duplicates,
|
|
})
|
|
|
|
|
|
def _scenario_rows(session: Session, scenarios: list, loans: list) -> list[dict]:
|
|
"""Baut je Szenario den Kontext fürs Template: zugeordnete Kredite,
|
|
Modifikatoren und (falls vorhanden) das letzte Projektionsergebnis."""
|
|
rows = []
|
|
for sc in scenarios:
|
|
assigned_loan_ids = {
|
|
sl.loan_id for sl in session.execute(
|
|
select(ScenarioLoan).where(ScenarioLoan.scenario_id == sc.id)
|
|
).scalars()
|
|
}
|
|
modifiers = session.execute(
|
|
select(ScenarioModifier).where(ScenarioModifier.scenario_id == sc.id)
|
|
).scalars().all()
|
|
planned_items = session.execute(
|
|
select(ScenarioPlannedItem)
|
|
.where(ScenarioPlannedItem.scenario_id == sc.id)
|
|
.order_by(ScenarioPlannedItem.due)).scalars().all()
|
|
rows.append({
|
|
"scenario": sc,
|
|
"assigned_loan_ids": assigned_loan_ids,
|
|
"modifiers": modifiers,
|
|
"planned_items": planned_items,
|
|
"result": session.get(ProjectionResult, sc.id),
|
|
})
|
|
return rows
|
|
|
|
|
|
@router.get("/hilfe", dependencies=[Depends(gui_session)])
|
|
def hilfe_page(request: Request):
|
|
return templates.TemplateResponse(request, "hilfe.html", {})
|
|
|
|
|
|
@router.get("/planung", dependencies=[Depends(gui_session)])
|
|
def planung_page(request: Request, session: Session = Depends(get_session)):
|
|
categories = session.execute(select(Category)).scalars().all()
|
|
recurring = _list_recurring(session=session)
|
|
planned = _list_planned(session=session)
|
|
loans = _list_loans(session=session)
|
|
scenarios = _list_scenarios(session=session)
|
|
return templates.TemplateResponse(request, "planning.html", {
|
|
"categories": categories,
|
|
"category_names": {c.id: c.name for c in categories},
|
|
"recurring": recurring,
|
|
"recurring_names": {r.id: r.name for r in recurring},
|
|
"suggestions": _recurring_suggestions(session=session),
|
|
"planned": planned,
|
|
"loans": loans,
|
|
"scenario_rows": _scenario_rows(session, scenarios, loans),
|
|
"rhythms": ["monthly", "quarterly", "yearly"],
|
|
"repayment_types": ["annuity", "bullet"],
|
|
"modifier_kinds": ["percent", "absolute", "remove", "ende"],
|
|
})
|