65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
"""Expenses CSV reader and summary."""
|
|
|
|
import csv
|
|
from collections import defaultdict
|
|
from fastapi import APIRouter, Query
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from lib_bridge import EXPENSES_CSV
|
|
|
|
router = APIRouter(tags=["expenses"])
|
|
|
|
|
|
def _read_expenses() -> list[dict]:
|
|
"""Read all expenses from CSV."""
|
|
if not EXPENSES_CSV.exists():
|
|
return []
|
|
with open(EXPENSES_CSV, "r", newline="") as f:
|
|
return list(csv.DictReader(f))
|
|
|
|
|
|
@router.get("/expenses")
|
|
def list_expenses(month: str | None = Query(None, description="Filter by YYYY-MM")):
|
|
"""List expenses, optionally filtered by month."""
|
|
expenses = _read_expenses()
|
|
if month:
|
|
expenses = [e for e in expenses if e.get("date", "").startswith(month)]
|
|
return expenses
|
|
|
|
|
|
@router.get("/expenses/summary")
|
|
def expenses_summary(month: str | None = Query(None, description="Filter by YYYY-MM")):
|
|
"""Monthly total, count, top 10 vendors by amount."""
|
|
from datetime import date
|
|
if not month:
|
|
month = date.today().strftime("%Y-%m")
|
|
expenses = _read_expenses()
|
|
all_time_count = len(expenses)
|
|
expenses = [e for e in expenses if e.get("date", "").startswith(month)]
|
|
|
|
if not expenses:
|
|
return {"total": 0, "count": 0, "all_time": all_time_count, "top_vendors": [], "month": month}
|
|
|
|
total = 0.0
|
|
vendor_totals = defaultdict(float)
|
|
for e in expenses:
|
|
try:
|
|
amount = float(e.get("amount", 0))
|
|
except (ValueError, TypeError):
|
|
amount = 0.0
|
|
total += amount
|
|
vendor = e.get("vendor", "unknown")
|
|
vendor_totals[vendor] += amount
|
|
|
|
top_vendors = sorted(vendor_totals.items(), key=lambda x: x[1], reverse=True)[:10]
|
|
|
|
return {
|
|
"total": round(total, 2),
|
|
"count": len(expenses),
|
|
"top_vendors": [{"vendor": v, "amount": round(a, 2)} for v, a in top_vendors],
|
|
"month": month,
|
|
}
|