Spaces:
Sleeping
Sleeping
File size: 3,570 Bytes
67c8aca | 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 | from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from sqlalchemy.orm import Session
from sqlalchemy import desc
import json
import os
from backend.db import database, models, schemas
from backend.logic.deterministic import process_pipeline
from backend.services.llm_advisor import generate_explanation
from backend.services.hf_sync import sync_db_to_hf
# Create tables
models.Base.metadata.create_all(bind=database.engine)
app = FastAPI(title="Loan Prediction System API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allow all for demo
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.post("/api/predict", response_model=schemas.LoanApplicationResponse)
def create_prediction(app_in: schemas.LoanApplicationCreate, background_tasks: BackgroundTasks, db: Session = Depends(database.get_db)):
# 1. Convert to dictionary
input_data = app_in.model_dump()
# 2. Run Deterministic Layer (Prediction + Wait-If Simulation)
packet = process_pipeline(input_data)
# 3. Pass Packet to LLM Layer
explanation = generate_explanation(packet)
# 4. Save to DB
db_record = models.LoanApplication(
applicant_name=app_in.applicant_name,
gender=app_in.gender,
married=app_in.married,
dependents=app_in.dependents,
education=app_in.education,
self_employed=app_in.self_employed,
applicant_income=app_in.applicant_income,
coapplicant_income=app_in.coapplicant_income,
loan_amount=app_in.loan_amount,
loan_amount_term=app_in.loan_amount_term,
credit_history=app_in.credit_history,
property_area=app_in.property_area,
prediction=packet["prediction"],
confidence=packet["confidence"],
dti_ratio=packet["dti_ratio"],
explanation_text=explanation,
optimized_suggestion=packet["optimized_suggestion"],
feature_importance_json=json.dumps(packet["feature_importances"]),
benchmarks_json=json.dumps(packet["benchmarks"])
)
db.add(db_record)
db.commit()
db.refresh(db_record)
# Trigger HF Backup in background
background_tasks.add_task(sync_db_to_hf)
return db_record
@app.get("/api/history", response_model=list[schemas.LoanApplicationResponse])
def get_history(limit: int = 10, db: Session = Depends(database.get_db)):
records = db.query(models.LoanApplication).order_by(desc(models.LoanApplication.created_at)).limit(limit).all()
return records
@app.delete("/api/history")
def clear_history(background_tasks: BackgroundTasks, db: Session = Depends(database.get_db)):
db.query(models.LoanApplication).delete()
db.commit()
# Trigger HF Backup in background
background_tasks.add_task(sync_db_to_hf)
return {"message": "History cleared successfully"}
# Serve Frontend
frontend_build_path = os.path.join(os.path.dirname(__file__), "..", "frontend", "dist")
if os.path.isdir(frontend_build_path):
app.mount("/assets", StaticFiles(directory=os.path.join(frontend_build_path, "assets")), name="assets")
@app.get("/{full_path:path}")
def catch_all(full_path: str):
index_path = os.path.join(frontend_build_path, "index.html")
if os.path.isfile(index_path):
return FileResponse(index_path)
return {"error": "Frontend build not found."}
|