459 lines
15 KiB
Python
459 lines
15 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, status, UploadFile, File
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from sqlalchemy.orm import Session, joinedload
|
|
from typing import List
|
|
import os
|
|
import shutil
|
|
from datetime import datetime, timedelta
|
|
|
|
from app.core.database import engine, get_db, Base
|
|
from app.core.security import verify_password, get_password_hash, create_access_token, decode_access_token
|
|
from app import models, schemas
|
|
|
|
# Crear tablas
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
app = FastAPI(title="Checklist Inteligente API", version="1.0.0")
|
|
|
|
# CORS
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["http://localhost:5173", "http://localhost:3000"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
security = HTTPBearer()
|
|
|
|
# Dependency para obtener usuario actual
|
|
def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
token = credentials.credentials
|
|
payload = decode_access_token(token)
|
|
print(f"Token payload: {payload}") # Debug
|
|
if payload is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Token inválido o expirado"
|
|
)
|
|
|
|
user_id = int(payload.get("sub"))
|
|
print(f"Looking for user ID: {user_id}") # Debug
|
|
user = db.query(models.User).filter(models.User.id == user_id).first()
|
|
if user is None:
|
|
print(f"User not found with ID: {user_id}") # Debug
|
|
raise HTTPException(status_code=404, detail="Usuario no encontrado")
|
|
|
|
return user
|
|
|
|
|
|
# ============= AUTH ENDPOINTS =============
|
|
@app.post("/api/auth/register", response_model=schemas.User)
|
|
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
|
# Verificar si usuario existe
|
|
db_user = db.query(models.User).filter(models.User.username == user.username).first()
|
|
if db_user:
|
|
raise HTTPException(status_code=400, detail="Usuario ya existe")
|
|
|
|
# Crear usuario
|
|
hashed_password = get_password_hash(user.password)
|
|
db_user = models.User(
|
|
username=user.username,
|
|
email=user.email,
|
|
full_name=user.full_name,
|
|
role=user.role,
|
|
password_hash=hashed_password
|
|
)
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return db_user
|
|
|
|
|
|
@app.post("/api/auth/login", response_model=schemas.Token)
|
|
def login(user_login: schemas.UserLogin, db: Session = Depends(get_db)):
|
|
user = db.query(models.User).filter(models.User.username == user_login.username).first()
|
|
if not user or not verify_password(user_login.password, user.password_hash):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Usuario o contraseña incorrectos"
|
|
)
|
|
|
|
access_token = create_access_token(data={"sub": str(user.id), "role": user.role})
|
|
return {
|
|
"access_token": access_token,
|
|
"token_type": "bearer",
|
|
"user": user
|
|
}
|
|
|
|
|
|
@app.get("/api/auth/me", response_model=schemas.User)
|
|
def get_me(current_user: models.User = Depends(get_current_user)):
|
|
return current_user
|
|
|
|
|
|
# ============= CHECKLIST ENDPOINTS =============
|
|
@app.get("/api/checklists", response_model=List[schemas.Checklist])
|
|
def get_checklists(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
active_only: bool = False,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
query = db.query(models.Checklist)
|
|
if active_only:
|
|
query = query.filter(models.Checklist.is_active == True)
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
|
|
@app.get("/api/checklists/{checklist_id}", response_model=schemas.ChecklistWithQuestions)
|
|
def get_checklist(checklist_id: int, db: Session = Depends(get_db)):
|
|
checklist = db.query(models.Checklist).options(
|
|
joinedload(models.Checklist.questions)
|
|
).filter(models.Checklist.id == checklist_id).first()
|
|
|
|
if not checklist:
|
|
raise HTTPException(status_code=404, detail="Checklist no encontrado")
|
|
|
|
return checklist
|
|
|
|
|
|
@app.post("/api/checklists", response_model=schemas.Checklist)
|
|
def create_checklist(
|
|
checklist: schemas.ChecklistCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="No autorizado")
|
|
|
|
db_checklist = models.Checklist(**checklist.dict(), created_by=current_user.id)
|
|
db.add(db_checklist)
|
|
db.commit()
|
|
db.refresh(db_checklist)
|
|
return db_checklist
|
|
|
|
|
|
@app.put("/api/checklists/{checklist_id}", response_model=schemas.Checklist)
|
|
def update_checklist(
|
|
checklist_id: int,
|
|
checklist: schemas.ChecklistUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="No autorizado")
|
|
|
|
db_checklist = db.query(models.Checklist).filter(models.Checklist.id == checklist_id).first()
|
|
if not db_checklist:
|
|
raise HTTPException(status_code=404, detail="Checklist no encontrado")
|
|
|
|
for key, value in checklist.dict(exclude_unset=True).items():
|
|
setattr(db_checklist, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_checklist)
|
|
return db_checklist
|
|
|
|
|
|
# ============= QUESTION ENDPOINTS =============
|
|
@app.post("/api/questions", response_model=schemas.Question)
|
|
def create_question(
|
|
question: schemas.QuestionCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="No autorizado")
|
|
|
|
db_question = models.Question(**question.dict())
|
|
db.add(db_question)
|
|
|
|
# Actualizar max_score del checklist
|
|
checklist = db.query(models.Checklist).filter(
|
|
models.Checklist.id == question.checklist_id
|
|
).first()
|
|
if checklist:
|
|
checklist.max_score += question.points
|
|
|
|
db.commit()
|
|
db.refresh(db_question)
|
|
return db_question
|
|
|
|
|
|
@app.put("/api/questions/{question_id}", response_model=schemas.Question)
|
|
def update_question(
|
|
question_id: int,
|
|
question: schemas.QuestionUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="No autorizado")
|
|
|
|
db_question = db.query(models.Question).filter(models.Question.id == question_id).first()
|
|
if not db_question:
|
|
raise HTTPException(status_code=404, detail="Pregunta no encontrada")
|
|
|
|
for key, value in question.dict(exclude_unset=True).items():
|
|
setattr(db_question, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_question)
|
|
return db_question
|
|
|
|
|
|
@app.delete("/api/questions/{question_id}")
|
|
def delete_question(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="No autorizado")
|
|
|
|
db_question = db.query(models.Question).filter(models.Question.id == question_id).first()
|
|
if not db_question:
|
|
raise HTTPException(status_code=404, detail="Pregunta no encontrada")
|
|
|
|
db.delete(db_question)
|
|
db.commit()
|
|
return {"message": "Pregunta eliminada"}
|
|
|
|
|
|
# ============= INSPECTION ENDPOINTS =============
|
|
@app.get("/api/inspections", response_model=List[schemas.Inspection])
|
|
def get_inspections(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
vehicle_plate: str = None,
|
|
status: str = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
query = db.query(models.Inspection)
|
|
|
|
# Mecánicos solo ven sus inspecciones
|
|
if current_user.role == "mechanic":
|
|
query = query.filter(models.Inspection.mechanic_id == current_user.id)
|
|
|
|
if vehicle_plate:
|
|
query = query.filter(models.Inspection.vehicle_plate.contains(vehicle_plate))
|
|
|
|
if status:
|
|
query = query.filter(models.Inspection.status == status)
|
|
|
|
return query.order_by(models.Inspection.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
|
|
@app.get("/api/inspections/{inspection_id}", response_model=schemas.InspectionDetail)
|
|
def get_inspection(
|
|
inspection_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
inspection = db.query(models.Inspection).options(
|
|
joinedload(models.Inspection.checklist).joinedload(models.Checklist.questions),
|
|
joinedload(models.Inspection.mechanic),
|
|
joinedload(models.Inspection.answers).joinedload(models.Answer.question),
|
|
joinedload(models.Inspection.answers).joinedload(models.Answer.media_files)
|
|
).filter(models.Inspection.id == inspection_id).first()
|
|
|
|
if not inspection:
|
|
raise HTTPException(status_code=404, detail="Inspección no encontrada")
|
|
|
|
return inspection
|
|
|
|
|
|
@app.post("/api/inspections", response_model=schemas.Inspection)
|
|
def create_inspection(
|
|
inspection: schemas.InspectionCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
# Obtener max_score del checklist
|
|
checklist = db.query(models.Checklist).filter(
|
|
models.Checklist.id == inspection.checklist_id
|
|
).first()
|
|
|
|
if not checklist:
|
|
raise HTTPException(status_code=404, detail="Checklist no encontrado")
|
|
|
|
db_inspection = models.Inspection(
|
|
**inspection.dict(),
|
|
mechanic_id=current_user.id,
|
|
max_score=checklist.max_score
|
|
)
|
|
db.add(db_inspection)
|
|
db.commit()
|
|
db.refresh(db_inspection)
|
|
return db_inspection
|
|
|
|
|
|
@app.put("/api/inspections/{inspection_id}", response_model=schemas.Inspection)
|
|
def update_inspection(
|
|
inspection_id: int,
|
|
inspection: schemas.InspectionUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
db_inspection = db.query(models.Inspection).filter(
|
|
models.Inspection.id == inspection_id
|
|
).first()
|
|
|
|
if not db_inspection:
|
|
raise HTTPException(status_code=404, detail="Inspección no encontrada")
|
|
|
|
for key, value in inspection.dict(exclude_unset=True).items():
|
|
setattr(db_inspection, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_inspection)
|
|
return db_inspection
|
|
|
|
|
|
@app.post("/api/inspections/{inspection_id}/complete", response_model=schemas.Inspection)
|
|
def complete_inspection(
|
|
inspection_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
inspection = db.query(models.Inspection).filter(
|
|
models.Inspection.id == inspection_id
|
|
).first()
|
|
|
|
if not inspection:
|
|
raise HTTPException(status_code=404, detail="Inspección no encontrada")
|
|
|
|
# Calcular score
|
|
answers = db.query(models.Answer).filter(models.Answer.inspection_id == inspection_id).all()
|
|
total_score = sum(a.points_earned for a in answers)
|
|
flagged_count = sum(1 for a in answers if a.is_flagged)
|
|
|
|
inspection.score = total_score
|
|
inspection.percentage = (total_score / inspection.max_score * 100) if inspection.max_score > 0 else 0
|
|
inspection.flagged_items_count = flagged_count
|
|
inspection.status = "completed"
|
|
inspection.completed_at = datetime.utcnow()
|
|
|
|
db.commit()
|
|
db.refresh(inspection)
|
|
return inspection
|
|
|
|
|
|
# ============= ANSWER ENDPOINTS =============
|
|
@app.post("/api/answers", response_model=schemas.Answer)
|
|
def create_answer(
|
|
answer: schemas.AnswerCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
# Obtener la pregunta para saber los puntos
|
|
question = db.query(models.Question).filter(models.Question.id == answer.question_id).first()
|
|
|
|
if not question:
|
|
raise HTTPException(status_code=404, detail="Pregunta no encontrada")
|
|
|
|
# Calcular puntos según status
|
|
points_earned = 0
|
|
if answer.status == "ok":
|
|
points_earned = question.points
|
|
elif answer.status == "warning":
|
|
points_earned = int(question.points * 0.5)
|
|
|
|
db_answer = models.Answer(
|
|
**answer.dict(),
|
|
points_earned=points_earned
|
|
)
|
|
db.add(db_answer)
|
|
db.commit()
|
|
db.refresh(db_answer)
|
|
return db_answer
|
|
|
|
|
|
@app.put("/api/answers/{answer_id}", response_model=schemas.Answer)
|
|
def update_answer(
|
|
answer_id: int,
|
|
answer: schemas.AnswerUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
db_answer = db.query(models.Answer).filter(models.Answer.id == answer_id).first()
|
|
|
|
if not db_answer:
|
|
raise HTTPException(status_code=404, detail="Respuesta no encontrada")
|
|
|
|
# Recalcular puntos si cambió el status
|
|
if answer.status and answer.status != db_answer.status:
|
|
question = db.query(models.Question).filter(
|
|
models.Question.id == db_answer.question_id
|
|
).first()
|
|
|
|
if answer.status == "ok":
|
|
db_answer.points_earned = question.points
|
|
elif answer.status == "warning":
|
|
db_answer.points_earned = int(question.points * 0.5)
|
|
else:
|
|
db_answer.points_earned = 0
|
|
|
|
for key, value in answer.dict(exclude_unset=True).items():
|
|
setattr(db_answer, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_answer)
|
|
return db_answer
|
|
|
|
|
|
# ============= MEDIA FILE ENDPOINTS =============
|
|
@app.post("/api/answers/{answer_id}/upload", response_model=schemas.MediaFile)
|
|
async def upload_photo(
|
|
answer_id: int,
|
|
file: UploadFile = File(...),
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
# Verificar que la respuesta existe
|
|
answer = db.query(models.Answer).filter(models.Answer.id == answer_id).first()
|
|
if not answer:
|
|
raise HTTPException(status_code=404, detail="Respuesta no encontrada")
|
|
|
|
# Crear directorio si no existe
|
|
upload_dir = f"uploads/inspection_{answer.inspection_id}"
|
|
os.makedirs(upload_dir, exist_ok=True)
|
|
|
|
# Guardar archivo
|
|
file_extension = file.filename.split(".")[-1]
|
|
file_name = f"answer_{answer_id}_{datetime.now().timestamp()}.{file_extension}"
|
|
file_path = os.path.join(upload_dir, file_name)
|
|
|
|
with open(file_path, "wb") as buffer:
|
|
shutil.copyfileobj(file.file, buffer)
|
|
|
|
# Crear registro en BD
|
|
media_file = models.MediaFile(
|
|
answer_id=answer_id,
|
|
file_path=file_path,
|
|
file_type="image"
|
|
)
|
|
db.add(media_file)
|
|
db.commit()
|
|
db.refresh(media_file)
|
|
|
|
return media_file
|
|
|
|
|
|
# ============= HEALTH CHECK =============
|
|
@app.get("/")
|
|
def root():
|
|
return {"message": "Checklist Inteligente API", "version": "1.0.0", "status": "running"}
|
|
|
|
|
|
@app.get("/health")
|
|
def health_check():
|
|
return {"status": "healthy"}
|