# app/main.py from dotenv import load_dotenv import os, requests from fastapi import FastAPI, Query, HTTPException, Form, UploadFile, File, Header from fastapi.responses import PlainTextResponse,JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from connection.connection import db from connection.querys import query from models.LivenessResponse import LivenessResponse import base64 load_dotenv() FACETEC_SERVER = os.getenv("FACETEC_SERVER") API_KEY = os.getenv("API_KEY") DEVICE_KEY = os.getenv("FACETEC_DEVICE_KEY") FIXED_TOKEN = os.getenv("FIXED_TOKEN") origins = [ "http://localhost:5233" ] class LivenessPayload(BaseModel): cedula: str faceScan: str auditTrailImage: str | None = None class MatchPayload(BaseModel): imageBase64: str minMatchLevel: int = 6 transactionId: str if not FACETEC_SERVER or not DEVICE_KEY: raise RuntimeError("❌ Config: define FACETEC_SERVER y FACETEC_DEVICE_KEY") app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=origins, # Orígenes permitidos allow_credentials=True, allow_methods=["*"], # Métodos permitidos: GET, POST, etc. allow_headers=["*"], # Headers permitidos ) @app.post("/api/facetec/session-token", response_class=JSONResponse,) def post_session_token(x_device_key: str = Header(...)): headers = {"X-Device-Key": x_device_key} try: r = requests.get(f"{FACETEC_SERVER}/session-token", headers=headers, timeout=10) except requests.RequestException as exc: raise HTTPException(502, f"Conexión falló: {exc}") if r.status_code != 200: raise HTTPException(502, f"FaceTec {r.status_code}: {r.text}") data = r.json() if not data.get("success"): raise HTTPException(502, f"FaceTec no generó token: {data}") return JSONResponse(status_code=200, content={ "sessionToken": data.get("sessionToken"), }) @app.post("/api/v1/session-token", response_class=JSONResponse,) def post_session_token(x_device_key: str = Header(...)): headers = {"X-Device-Key": x_device_key} try: r = requests.get(f"{FACETEC_SERVER}/session-token", headers=headers, timeout=10) except requests.RequestException as exc: raise HTTPException(502, f"Conexión falló: {exc}") if r.status_code != 200: raise HTTPException(502, f"Error {r.status_code}: {r.text}") data = r.json() if not data.get("success"): raise HTTPException(502, f"error no se generó token: {data}") return JSONResponse(status_code=200, content={ "sessionToken": data.get("sessionToken"), }) @app.post("/api/v1/enrollment-liveness", response_class=JSONResponse) async def enrollment_liveness(data: LivenessResponse,x_device_key: str = Header(...)): try: success = await query.query.insert_liveness_result(data) if not success: return JSONResponse(status_code=500, content={"status": "error", "message": "No se pudo guardar el resultado de liveness"}) except Exception as e: return JSONResponse(status_code=500, content={"status": "error", "message": str(e)}) headers = { "X-Device-Key": x_device_key, "Content-Type": "application/json" } body = { "faceScan": data.faceScan, "auditTrailImage": data.auditTrailImage, "lowQualityAuditTrailImage": data.lowQualityAuditTrailImage, "sessionId": data.sessionId, "externalDatabaseRefID": data.transactionId, } r = requests.post(f"{FACETEC_SERVER}/enrollment-3d", json=body, headers=headers, timeout=15) res = r.json() # Ejemplo de respuesta del Server SDK: if not res.get("success"): print("Error en la respuesta de FaceTec:", res) raise HTTPException(401, "Liveness no superado") return res # return JSONResponse(status_code=200, content={ # "success": True, # "message": "Liveness registrado correctamente"}) """API PARA VALIDAR LIVENESS CON LA FOTO DEL REGISTRO CIVIL Args: data (LivenessPayload): Raises: HTTPException: Si la validación falla o hay un error de conexión. """ @app.post("/api/match-identification") def validar_liveness(data: MatchPayload, authorization: str = Header(None)): if authorization != f"Bearer {FIXED_TOKEN}": raise HTTPException(status_code=401, detail="Unauthorized") try: headers = { "X-Device-Key": DEVICE_KEY } body = { "externalDatabaseRefID": data.transactionId, "image": data.imageBase64, "minMatchLevel": data.minMatchLevel, } r = requests.post(f"{FACETEC_SERVER}/match-3d-2d-profile-pic", json=body, headers=headers, timeout=15) res = r.json() print("Respuesta de FaceTec:", res) if not res.get("success"): return { "success": False, "matchLevel": 0, "error": res.get("error", True), "errorMessage": res.get("errorMessage", "Error desconocido") } return { "success": res.get("success"), "matchLevel": res.get("matchLevel"), "error": res.get("error", False), } except Exception as e: raise HTTPException(500, { "success": False, "matchLevel": 0, "error": True, "errorMessage": str(e) }) """Webhook para recibir notificaciones de FaceTec""" @app.post("/api/webhook") async def webhook(request: dict, authorization: str = Header(None)): if authorization != f"Bearer {FIXED_TOKEN}": raise HTTPException(status_code=401, detail="Unauthorized") try: # Aquí puedes procesar la notificación recibida print("Webhook recibido:", request) # Por ejemplo, podrías guardar la información en tu base de datos # await db.save_webhook_data(request) return {"success": True, "message": "Webhook recibido correctamente"} except Exception as e: print("Error en el webhook:", str(e)) return JSONResponse(status_code=500, content={"success": False, "errorMessage": str(e)}) @app.on_event("startup") async def startup(): await db.connect() @app.on_event("shutdown") async def shutdown(): await db.disconnect() @app.post("/registrar-imagen") async def registrar_imagen( identificacion: str = Form(...), imagen: UploadFile = File(...) ): try: contenido = await imagen.read() imagen_base64 = base64.b64encode(contenido).decode("utf-8") success = await query.query.insertar_imagen(identificacion, imagen_base64) if success: return {"status": "ok", "message": "Imagen registrada o actualizada correctamente"} else: return JSONResponse(status_code=500, content={"status": "error", "message": "No se pudo guardar la imagen"}) except Exception as e: return JSONResponse(status_code=500, content={"status": "error", "message": str(e)}) @app.get("/obtener-imagen/{identificacion}") async def obtener_imagen(identificacion: str): resultado = await query.query.obtener_imagen_por_identificacion(identificacion) if resultado: return { # "id": resultado["id"], "identificacion": resultado["identificacion"], "imagen_base64": resultado["imagen_base64"], "fecha_registro": str(resultado["fecha_registro"]) } else: raise HTTPException(status_code=404, detail="Imagen no encontrada")