Spaces:
Sleeping
Sleeping
File size: 4,042 Bytes
45df06e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
import os
from dotenv import load_dotenv
from typing import Any, Dict
load_dotenv()
# Local module imports (we'll implement these next)
from backend import ai_agent
from backend import expense_manager
from backend import budget_allocator
from backend import reminder_system
# App init
app = FastAPI(title="FinSync AI Backend")
# Allow CORS from Streamlit / local dev
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Pydantic models
class AnalyzeRequest(BaseModel):
user_id: str
text: str
class ExpenseRequest(BaseModel):
user_id: str
name: str
amount: float
category: str = ""
date: str = ""
class ReminderRequest(BaseModel):
user_id: str
name: str
date: str
amount: float = 0.0
notes: str = ""
# Simple health check
@app.get("/health")
async def health():
return {"status": "ok", "service": "finsync-backend"}
# Analyze / natural language endpoint
@app.post("/analyze")
async def analyze(req: AnalyzeRequest):
"""
Send user natural command to AI agent; AI responds with action/intention and human-readable reply.
Example use-cases:
- "Show my expenses this month"
- "Set reminder to pay rent on 2025-11-10"
- "Add expense Rent 500"
"""
try:
response: Dict[str, Any] = ai_agent.parse_and_respond(user_id=req.user_id, text=req.text)
return {"ok": True, "result": response}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Add expense endpoint
@app.post("/add_expense")
async def add_expense(req: ExpenseRequest):
"""
Add an expense into the database.json (demo DB).
"""
try:
entry = {
"user_id": req.user_id,
"name": req.name,
"amount": float(req.amount),
"category": req.category or "Uncategorized",
"date": req.date or expense_manager.today_iso(),
}
expense_manager.add_expense(entry)
return {"ok": True, "expense": entry}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# List expenses for a user
@app.get("/expenses")
async def get_expenses(user_id: str):
try:
items = expense_manager.list_expenses(user_id=user_id)
return {"ok": True, "expenses": items}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Get budget suggestion
@app.get("/budget")
async def get_budget(user_id: str):
try:
expenses = expense_manager.list_expenses(user_id=user_id)
suggestion = budget_allocator.suggest_budget(user_id=user_id, expenses=expenses)
return {"ok": True, "budget": suggestion}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Add reminder
@app.post("/reminder")
async def add_reminder(req: ReminderRequest):
try:
reminder = {
"user_id": req.user_id,
"name": req.name,
"date": req.date,
"amount": float(req.amount),
"notes": req.notes
}
reminder_system.add_reminder(reminder)
return {"ok": True, "reminder": reminder}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# List reminders
@app.get("/reminders")
async def list_reminders(user_id: str):
try:
items = reminder_system.list_reminders(user_id=user_id)
return {"ok": True, "reminders": items}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
port = int(os.environ.get("BACKEND_PORT", 8000))
uvicorn.run("backend.main:app", host="0.0.0.0", port=port, reload=True) |