import variables env_path = variables.variables_entorno() from fastapi import FastAPI, Form, Request from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware import logging import valormas import valormas_ws import os from datetime import datetime import base64 import binascii # Importante para capturar errores base64 import mimetypes from typing import Optional from dotenv import load_dotenv import valormas_gemini from valormas_llama import generar_respuesta import valormas_anthropic_v2 # Cargar variables de entorno desde archivo .env current_dir = os.path.dirname(os.path.abspath(__file__)) env_file_path = os.path.join(current_dir, ".env") # Verificar que el archivo .env existe if not os.path.exists(env_file_path): logging.error(f"Archivo .env no encontrado en: {env_file_path}") raise FileNotFoundError(f"No se encontró el archivo .env en {env_file_path}") # Cargar el archivo .env load_dotenv(env_file_path) # Obtener la variable de entorno con valor por defecto CARPETA_DESTINO = os.getenv("CARPETA_DESTINO") if not CARPETA_DESTINO: CARPETA_DESTINO = "/var/www/cobra/DESARROLLO/valormas/archivos_ciudadanos" logging.warning("Usando valor por defecto para CARPETA_DESTINO") else: logging.info(f"CARPETA_DESTINO cargada: {CARPETA_DESTINO}") # Verificar si la carpeta existe y crearla si es necesario if not os.path.exists(CARPETA_DESTINO): os.makedirs(CARPETA_DESTINO, exist_ok=True) logging.info(f"Directorio creado: {CARPETA_DESTINO}") app = FastAPI(debug=True) # Configurar CORS para permitir solicitudes desde los orígenes necesarios app.add_middleware( CORSMiddleware, allow_origins=[ "http://localhost:4200", "http://10.51.20.5", "https://dev.crm.ayudacatastro.co", "https://core.cic-ware.com", "https://dev.ayudacatastro.co", "https://dev.miperfil.ayudacatastro.co", "https://pre.crm.ayudacatastro.co", "https://pre.ayudacatastro.co", "https://pre.miperfil.ayudacatastro.co", "https://evolution.api.cic-ware.com/message/sendContact/t", "https://demo.agencycic.com", "https://www.agencycic.com", "https://agencycic.com", "https://pro.ayudacatastro.co", "https://pro.crm.ayudacatastro.co", "https://pro.miperfil.ayudacatastro.co", "https://ayuda.catastroantioquia-mas.com", "https://miperfil.catastroantioquia-mas.com", "https://crm.catastroantioquia-mas.com", ], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Configurar logging para capturar y mostrar errores logging.basicConfig(level=logging.INFO) # Nuevo endpoint en `/catia` @app.post("/catia") def catia_endpoint( asistente: str = Form(...), pregunta: str = Form(...), idcliente: str = Form(...), volume_up: str = Form(...), human_on: str = Form(...), thread_id: str = Form(...), instrucciones: str = Form(...), id_asistente: str = Form(...), archivo: Optional[str] = Form(None), ): try: archivo_path = "no" if archivo and archivo != "no": try: tipo_mime = "application/octet-stream" if "," in archivo: partes = archivo.split(",", 1) encabezado, contenido_base64 = partes[0], partes[1] if "data:" in encabezado and ";base64" in encabezado: tipo_mime = encabezado.split(":")[1].split(";")[0] archivo = contenido_base64 else: archivo = archivo # Limpiar base64 archivo = archivo.strip().replace("\n", "").replace("\r", "") archivo += "=" * (-len(archivo) % 4) try: contenido = base64.b64decode(archivo) except (binascii.Error, ValueError) as e: logging.error(f"Base64 malformado: {e}") return { "status": "error", "message": "El archivo enviado no tiene un formato base64 válido.", } # Obtener extensión desde el tipo MIME extension = mimetypes.guess_extension(tipo_mime) or ".bin" # Crear nombre del archivo con extensión timestamp = datetime.now().strftime("%Y%m%d%H%M%S") nombre_archivo = f"{idcliente}-{timestamp}{extension}" # Guardar el archivo usando la variable global os.makedirs(CARPETA_DESTINO, exist_ok=True) archivo_path = os.path.join(CARPETA_DESTINO, nombre_archivo) with open(archivo_path, "wb") as f: f.write(contenido) logging.info(f"Archivo guardado exitosamente en: {archivo_path}") except Exception as e: logging.error(f"Error procesando archivo base64: {str(e)}") return { "status": "error", "message": f"Error al procesar el archivo: {str(e)}", } # Ejecutar función principal resultado = valormas.main( asistente, thread_id, pregunta, archivo_path, idcliente, instrucciones, id_asistente, "no", human_on, volume_up, ) logging.info(f"Resultado de ValorPlus.main(): {resultado}") return resultado except Exception as e: logging.error(f"Error ejecutando ValorPlus.main(): {str(e)}") return {"status": "error", "message": str(e)} @app.post("/catia_ws") async def catia_endpoint(request: Request): try: data = await request.json() asistente = data.get("asistente", "") pregunta = data.get("pregunta", "") volume_up = data.get("volume_up", "") instrucciones = data.get("instrucciones", "") id_asistente = data.get("id_asistente", "") telefono = data.get("telefono", "") # Llamamos a la función `main()` con los parámetros extraídos resultado = valormas_ws.main_telefono( asistente, pregunta, instrucciones, id_asistente, volume_up, telefono ) # logging.info(f"Resultado de main(): {resultado}") return resultado except Exception as e: logging.error(f"Error ejecutando main(): {str(e)}") return {"status": "error", "message": str(e), "resultado": resultado} @app.post("/catia-llama") async def catia_llama_endpoint( thread_id: str = Form(...), pregunta: str = Form(...), idcliente: str = Form(...), id_asistente: str = Form(...), volume_up: str = Form(...), ): try: respuesta = await generar_respuesta( hilo_conversacion=thread_id, prompt=pregunta, idcliente=idcliente, id_asistente=id_asistente, volume_up=volume_up, ) return respuesta except Exception as e: logging.error(f"Error ejecutando Valormas Llama: {str(e)}") return {"status": "error", "message": str(e)} @app.post("/catia-gemini") async def catia_gemini_endpoint( thread_id: str = Form(...), pregunta: str = Form(...), idcliente: int = Form(...), asistente: str = Form(...), volume_up: str = Form(...), id_asistente: int = Form(...), ): try: resultado = valormas_gemini.main( thread_id=thread_id, pregunta=pregunta, idcliente=idcliente, asistente=asistente, volume_up=volume_up, id_asistente=id_asistente, ) return JSONResponse(content=resultado) except Exception as e: return JSONResponse( status_code=500, content={"status": "error", "message": str(e)} ) @app.post("/catia-anthropic-v2") async def catia_anthropic_endpoint( thread_id: str = Form(...), pregunta: str = Form(...), idcliente: int = Form(...), asistente: str = Form(...), volume_up: str = Form(...), id_asistente: int = Form(...), ): try: # Solo llamamos a la función main pasándole estas variables resultado = valormas_anthropic_v2.main( thread_id=thread_id, pregunta=pregunta, idcliente=idcliente, asistente=asistente, volume_up=volume_up, id_asistente=id_asistente, ) return JSONResponse(content=resultado) except Exception as e: return JSONResponse( status_code=500, content={"status": "error", "message": str(e)} ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=5000)