112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
import os
|
|
import base64
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy import Column, String, Integer
|
|
|
|
from src.ArquitectureLayer.Mapper import Mapper_base
|
|
from src.ArquitectureLayer.Model import Model_base
|
|
from src.ArquitectureLayer.Repo import Repo_base
|
|
|
|
from src.ConexionSql.Base_conexion import ConexionBase
|
|
from src.base import Base
|
|
from src.Security.Encriptar import Encriptar_fernet
|
|
from src.Security.GenerarIDs import GeneradorIDUnico
|
|
from src.Llms.Embedders.Base_Embedder import EmbedderABC
|
|
from src.TextManager.biblioteca import Biblioteca # Suponiendo que defines la clase lógica Biblioteca aquí
|
|
|
|
# ----------------------
|
|
# Cargar clave maestra
|
|
# ----------------------
|
|
from entrypoint import ENV_PATH
|
|
load_dotenv(ENV_PATH)
|
|
pssword = os.getenv('MASTER_PASSWORD')
|
|
if pssword is None:
|
|
raise ValueError("MASTER_PASSWORD no está definida en el archivo .env")
|
|
|
|
# ----------------------
|
|
# MODELO (SQLAlchemy)
|
|
# ----------------------
|
|
|
|
class BibliotecaModel(Base, Model_base):
|
|
__tablename__ = "bibliotecas"
|
|
|
|
id = Column(String, primary_key=True, unique=True)
|
|
nombre = Column(String, nullable=False, unique=True)
|
|
|
|
descripcion = Column(String, nullable=False, default="")
|
|
|
|
vector_dim = Column(Integer, nullable=False)
|
|
|
|
embedder_info = Column(String, nullable=True) # Nombre de clase, ID de configuración, o info encriptada del embedder
|
|
|
|
# ----------------------
|
|
# MAPPER
|
|
# ----------------------
|
|
|
|
class BibliotecaMapper(Mapper_base[Biblioteca, BibliotecaModel]):
|
|
|
|
@staticmethod
|
|
def to_model(obj: Biblioteca) -> BibliotecaModel:
|
|
return BibliotecaModel(
|
|
id=obj.id,
|
|
nombre=obj.nombre,
|
|
descripcion=obj.descripcion,
|
|
vector_dim=obj.vector_dim
|
|
# embedder no se serializa en el modelo, se maneja por separado
|
|
)
|
|
|
|
@staticmethod
|
|
def from_model(model: BibliotecaModel) -> Biblioteca:
|
|
return Biblioteca(
|
|
id=model.id,
|
|
nombre=model.nombre,
|
|
descripcion=model.descripcion,
|
|
vector_dim=model.vector_dim,
|
|
embedder=None # se deja para inyección posterior si es necesario
|
|
)
|
|
|
|
@staticmethod
|
|
def to_dict(obj: Biblioteca) -> dict:
|
|
embedder_info = type(obj.embedder).__name__ if obj.embedder else None
|
|
return {
|
|
"id": obj.id,
|
|
"nombre": obj.nombre,
|
|
"descripcion": obj.descripcion,
|
|
"vector_dim": obj.vector_dim,
|
|
"embedder_info": embedder_info
|
|
}
|
|
|
|
@staticmethod
|
|
def from_dict(data: dict) -> Biblioteca:
|
|
return Biblioteca(
|
|
id=data["id"],
|
|
nombre=data["nombre"],
|
|
descripcion=data["descripcion"],
|
|
vector_dim=data["vector_dim"],
|
|
embedder=None # inyección manual si se desea
|
|
)
|
|
|
|
|
|
# ----------------------
|
|
# REPO
|
|
# ----------------------
|
|
|
|
class BibliotecaRepo(Repo_base[BibliotecaModel, Biblioteca]):
|
|
def __init__(self, conexion: ConexionBase):
|
|
super().__init__(
|
|
session=conexion.get_session(),
|
|
modelo=BibliotecaModel,
|
|
mapper=BibliotecaMapper
|
|
)
|
|
|
|
def add(self, biblioteca: Biblioteca, created_by: str = None, notes: str = None) -> str:
|
|
# Lógica personalizada: prevenir duplicados por nombre
|
|
existente = self.session.query(self.Modelo).filter_by(nombre=biblioteca.nombre).first()
|
|
if existente:
|
|
raise ValueError(f"Ya existe una biblioteca con el nombre '{biblioteca.nombre}'")
|
|
|
|
return super().add(biblioteca, created_by=created_by, notes=notes)
|
|
|
|
def get_by_nombre(self, nombre: str) -> Biblioteca | None:
|
|
model = self.session.query(self.Modelo).filter_by(nombre=nombre, sys_deleted_at=None).first()
|
|
return self.Mapper.from_model(model) if model else None |