k1nq
970fe9dacf
Gitea Actions CI / detect (push) Successful in 3s
Gitea Actions CI / Compose & nginx validation (push) Failing after 14s
Gitea Actions CI / Frontend (lint / type / test / build) (push) Successful in 45s
Gitea Actions CI / Backend (lint / type / test / build) (push) Successful in 1m31s
108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
import uuid
|
|
from collections import defaultdict
|
|
from datetime import date
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.budgeting.models import PeriodClosure
|
|
from app.budgeting.service import compute_period_state, get_or_create_period
|
|
from app.categorization.service import resolve_category_name, resolve_category_name_key
|
|
from app.ledger.models import Account, Category, Operation
|
|
from app.ledger.service import account_balance
|
|
|
|
|
|
def category_spend(
|
|
db: Session, *, space_id: uuid.UUID, date_from: date, date_to: date
|
|
) -> list[dict]:
|
|
expenses = (
|
|
db.query(Operation)
|
|
.filter(
|
|
Operation.space_id == space_id,
|
|
Operation.status == "confirmed",
|
|
Operation.type == "expense",
|
|
Operation.occurred_on >= date_from,
|
|
Operation.occurred_on <= date_to,
|
|
)
|
|
.all()
|
|
)
|
|
totals: dict[uuid.UUID | None, int] = defaultdict(int)
|
|
for expense in expenses:
|
|
refunded = (
|
|
db.query(func.coalesce(func.sum(Operation.amount_kopecks), 0))
|
|
.filter(
|
|
Operation.space_id == space_id,
|
|
Operation.status == "confirmed",
|
|
Operation.type == "refund",
|
|
Operation.parent_operation_id == expense.id,
|
|
)
|
|
.scalar()
|
|
)
|
|
totals[expense.category_id] += expense.amount_kopecks - int(refunded or 0)
|
|
categories = {item.id: item for item in db.query(Category).filter_by(space_id=space_id).all()}
|
|
return [
|
|
{
|
|
"category_id": category_id,
|
|
"name": (
|
|
resolve_category_name(categories[category_id])
|
|
if category_id is not None and category_id in categories
|
|
else None
|
|
),
|
|
"name_key": (
|
|
resolve_category_name_key(categories[category_id])
|
|
if category_id is not None and category_id in categories
|
|
else "uncategorized"
|
|
),
|
|
"amount_kopecks": amount,
|
|
}
|
|
for category_id, amount in sorted(totals.items(), key=lambda item: item[1], reverse=True)
|
|
]
|
|
|
|
|
|
def cash_flow(db: Session, *, space_id: uuid.UUID, date_from: date, date_to: date) -> list[dict]:
|
|
operations = (
|
|
db.query(Operation)
|
|
.filter(Operation.space_id == space_id, Operation.status == "confirmed")
|
|
.all()
|
|
)
|
|
totals: dict[tuple[date, str], int] = defaultdict(int)
|
|
for operation in operations:
|
|
effective = operation.posted_on or operation.occurred_on
|
|
if not date_from <= effective <= date_to or operation.type in {"transfer", "adjustment"}:
|
|
continue
|
|
signed = (
|
|
operation.amount_kopecks
|
|
if operation.type in {"income", "refund"}
|
|
else -operation.amount_kopecks
|
|
)
|
|
totals[(effective, operation.type)] += signed
|
|
return [
|
|
{"date": day, "type": kind, "amount_kopecks": amount}
|
|
for (day, kind), amount in sorted(totals.items())
|
|
]
|
|
|
|
|
|
def plan_fact(db: Session, *, space_id: uuid.UUID, year: int, month: int) -> dict:
|
|
period = get_or_create_period(db, space_id=space_id, year=year, month=month)
|
|
if period.status == "closed":
|
|
closure = db.query(PeriodClosure).filter_by(period_id=period.id).one_or_none()
|
|
if closure is not None:
|
|
return closure.snapshot
|
|
return compute_period_state(db, space_id=space_id, period=period)
|
|
|
|
|
|
def balances(db: Session, *, space_id: uuid.UUID) -> list[dict]:
|
|
result = []
|
|
for account in db.query(Account).filter_by(space_id=space_id).order_by(Account.created_at):
|
|
balance = account_balance(db, account)
|
|
result.append(
|
|
{
|
|
"account_id": account.id,
|
|
"account": account.name,
|
|
"kind": account.kind,
|
|
"balance_kopecks": balance,
|
|
"debt_kopecks": max(-balance, 0) if account.kind == "credit" else 0,
|
|
}
|
|
)
|
|
return result
|