from fastapi import FastAPI, File, UploadFile, Form, Depends, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.orm import Session, joinedload from sqlalchemy import func, case from typing import List, Optional import os import boto3 from botocore.client import Config import uuid from app.core import config as app_config 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 import shutil from datetime import datetime, timedelta BACKEND_VERSION = "1.0.25" app = FastAPI(title="Checklist Inteligente API", version=BACKEND_VERSION) # S3/MinIO configuration S3_ENDPOINT = app_config.MINIO_ENDPOINT S3_ACCESS_KEY = app_config.MINIO_ACCESS_KEY S3_SECRET_KEY = app_config.MINIO_SECRET_KEY S3_IMAGE_BUCKET = app_config.MINIO_IMAGE_BUCKET S3_PDF_BUCKET = app_config.MINIO_PDF_BUCKET s3_client = boto3.client( 's3', endpoint_url=S3_ENDPOINT, aws_access_key_id=S3_ACCESS_KEY, aws_secret_access_key=S3_SECRET_KEY, config=Config(signature_version='s3v4'), region_name='us-east-1' ) # Crear tablas Base.metadata.create_all(bind=engine) # Información visual al iniciar el backend print("\n================ BACKEND STARTUP INFO ================") print(f"Backend version: {BACKEND_VERSION}") print(f"Database URL: {app_config.settings.DATABASE_URL}") print(f"Environment: {app_config.settings.ENVIRONMENT}") print(f"MinIO endpoint: {app_config.MINIO_ENDPOINT}") print("====================================================\n", flush=True) # CORS app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173", "http://localhost:3000"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Simulación de modelos y autenticación para ejemplo class User: def __init__(self, role): self.role = role class AIConfiguration: is_active = True logo_url = "" class models: User = User AIConfiguration = AIConfiguration # Simulación de get_db y get_current_user def get_db(): # Aquí iría la lógica real de SQLAlchemy class DummyDB: def query(self, model): return self def filter(self, *args, **kwargs): return self def first(self): return models.AIConfiguration() def commit(self): pass def refresh(self, obj): pass return DummyDB() def get_current_user(): # Aquí iría la lógica real de autenticación return models.User(role="admin") # Endpoint para subir el logo @app.post("/api/config/logo", response_model=dict) async def upload_logo( file: UploadFile = File(...), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if current_user.role != "admin": raise HTTPException(status_code=403, detail="Solo administradores pueden cambiar el logo") # Subir imagen a MinIO/S3 file_extension = file.filename.split(".")[-1] now = datetime.now() folder = "logo" file_name = f"logo_{now.strftime('%Y%m%d_%H%M%S')}.{file_extension}" s3_key = f"{folder}/{file_name}" # s3_client.upload_fileobj(file.file, S3_IMAGE_BUCKET, s3_key, ExtraArgs={"ContentType": file.content_type}) logo_url = f"https://minio.example.com/bucket/{s3_key}" # Ajusta según tu config # Actualiza la configuración en la base de datos # config = db.query(models.AIConfiguration).filter(models.AIConfiguration.is_active == True).first() # if config: # config.logo_url = logo_url # db.commit() return {"logo_url": logo_url} # Endpoint para obtener el logo @app.get("/api/config/logo", response_model=dict) def get_logo_url(db: Session = Depends(get_db)): # config = db.query(models.AIConfiguration).filter(models.AIConfiguration.is_active == True).first() # if config and getattr(config, "logo_url", None): # return {"logo_url": config.logo_url} # return {"logo_url": "https://minio.example.com/bucket/logo/default_logo.png"} return {"logo_url": "https://minio.example.com/bucket/logo/default_logo.png"}