|
|
from datetime import datetime, timezone |
|
|
from typing import Any, Dict, List, Optional |
|
|
|
|
|
from fastapi import HTTPException, status |
|
|
from bson import ObjectId |
|
|
|
|
|
from ..database.connection import get_collection |
|
|
from ..models.monthly_record import ( |
|
|
MonthlyRecord, MonthlyRecordCreate, MonthlyRecordUpdate, |
|
|
ExpenseCategory, month_key_from, normalize_month_name |
|
|
) |
|
|
|
|
|
class RecordService: |
|
|
def __init__(self): |
|
|
self.col = get_collection() |
|
|
|
|
|
async def ensure_indexes(self) -> None: |
|
|
|
|
|
await self.col.create_index("month_key", unique=True) |
|
|
|
|
|
def _calc_totals(self, salary: float, expenses: List[ExpenseCategory]) -> Dict[str, float]: |
|
|
total_expenses = round(sum(e.amount for e in expenses), 2) |
|
|
remaining = round(salary - total_expenses, 2) |
|
|
return {"total_expenses": total_expenses, "remaining": remaining} |
|
|
|
|
|
def _serialize(self, doc: Dict[str, Any]) -> Dict[str, Any]: |
|
|
if not doc: |
|
|
return doc |
|
|
doc["_id"] = str(doc["_id"]) |
|
|
return doc |
|
|
|
|
|
async def get_by_month_key(self, month_key: str) -> Dict[str, Any]: |
|
|
doc = await self.col.find_one({"month_key": month_key}) |
|
|
if not doc: |
|
|
raise HTTPException(status_code=404, detail="No record found for this month") |
|
|
return self._serialize(doc) |
|
|
|
|
|
async def list(self, limit: int = 12, skip: int = 0) -> List[Dict[str, Any]]: |
|
|
cursor = self.col.find({}).sort("month_key", 1).skip(skip).limit(limit) |
|
|
return [self._serialize(d) async for d in cursor] |
|
|
|
|
|
async def create(self, payload: MonthlyRecordCreate) -> Dict[str, Any]: |
|
|
month = normalize_month_name(payload.month) |
|
|
month_key = month_key_from(month, payload.year) |
|
|
now = datetime.now(timezone.utc) |
|
|
|
|
|
totals = self._calc_totals(payload.salary, payload.expenses) |
|
|
|
|
|
doc = { |
|
|
"month": month, |
|
|
"year": payload.year, |
|
|
"month_key": month_key, |
|
|
"salary": float(payload.salary), |
|
|
"expenses": [e.model_dump() for e in payload.expenses], |
|
|
"total_expenses": totals["total_expenses"], |
|
|
"remaining": totals["remaining"], |
|
|
"created_at": now, |
|
|
"updated_at": now, |
|
|
} |
|
|
|
|
|
try: |
|
|
result = await self.col.insert_one(doc) |
|
|
except Exception as e: |
|
|
|
|
|
raise HTTPException(status_code=400, detail=str(e)) |
|
|
|
|
|
created = await self.col.find_one({"_id": result.inserted_id}) |
|
|
return self._serialize(created) |
|
|
|
|
|
async def update(self, month_key: str, payload: MonthlyRecordUpdate) -> Dict[str, Any]: |
|
|
existing = await self.col.find_one({"month_key": month_key}) |
|
|
if not existing: |
|
|
raise HTTPException(status_code=404, detail="No record found for this month") |
|
|
|
|
|
salary = payload.salary if payload.salary is not None else existing["salary"] |
|
|
expenses = payload.expenses if payload.expenses is not None else existing["expenses"] |
|
|
|
|
|
expenses_list = [e.model_dump() if hasattr(e, "model_dump") else e for e in expenses] |
|
|
|
|
|
totals = self._calc_totals(salary, [ExpenseCategory(**e) for e in expenses_list]) |
|
|
|
|
|
update_doc = { |
|
|
"$set": { |
|
|
"salary": float(salary), |
|
|
"expenses": expenses_list, |
|
|
"total_expenses": totals["total_expenses"], |
|
|
"remaining": totals["remaining"], |
|
|
"updated_at": datetime.now(timezone.utc), |
|
|
} |
|
|
} |
|
|
|
|
|
await self.col.update_one({"month_key": month_key}, update_doc) |
|
|
updated = await self.col.find_one({"month_key": month_key}) |
|
|
return self._serialize(updated) |
|
|
|
|
|
async def delete(self, month_key: str) -> None: |
|
|
result = await self.col.delete_one({"month_key": month_key}) |
|
|
if result.deleted_count == 0: |
|
|
raise HTTPException(status_code=404, detail="No record found for this month") |
|
|
|
