113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
import os
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy import Column, String, Table, MetaData
|
|
from pgvector.sqlalchemy import Vector
|
|
from sqlalchemy.orm import registry, Session
|
|
from src.TextManager.nota import Nota
|
|
from src.ConexionSql.Base_conexion import ConexionBase
|
|
|
|
# ----------------------
|
|
# Cargar .env
|
|
# ----------------------
|
|
from entrypoint import ENV_PATH
|
|
load_dotenv(ENV_PATH)
|
|
|
|
# ----------------------
|
|
# REGISTRO DINÁMICO PARA TABLAS
|
|
# ----------------------
|
|
mapper_registry = registry()
|
|
|
|
# ----------------------
|
|
# FUNCIONES AUXILIARES
|
|
# ----------------------
|
|
|
|
def generar_tabla_nota_para_biblioteca(biblioteca_nombre: str, vector_dim: int, metadata: MetaData):
|
|
nombre_tabla = biblioteca_nombre.lower().replace(" ", "_")
|
|
|
|
tabla = Table(
|
|
nombre_tabla,
|
|
metadata,
|
|
Column("id", String, primary_key=True),
|
|
Column("titulo", String, nullable=False),
|
|
Column("tags", String),
|
|
Column("conexiones", String),
|
|
Column("texto", String),
|
|
Column("resumen", String),
|
|
Column("vector", Vector(vector_dim), nullable=True),
|
|
Column("vector_resumen", Vector(vector_dim), nullable=True),
|
|
)
|
|
|
|
class NotaModel:
|
|
def __init__(self, nota: Nota):
|
|
self.id = nota.id
|
|
self.titulo = nota.titulo
|
|
self.tags = ",".join(nota.tags)
|
|
self.conexiones = ",".join(nota.conexiones)
|
|
self.texto = nota.texto
|
|
self.resumen = nota.resumen
|
|
self.vector = nota.vector
|
|
self.vector_resumen = getattr(nota, "vector_resumen", None)
|
|
|
|
# Evitar mapear dos veces la misma clase
|
|
if NotaModel not in mapper_registry._class_registry.values():
|
|
mapper_registry.map_imperatively(NotaModel, tabla)
|
|
|
|
return tabla, NotaModel
|
|
|
|
|
|
# ----------------------
|
|
# MAPPER
|
|
# ----------------------
|
|
|
|
class NotaMapper:
|
|
@staticmethod
|
|
def to_dict(nota: Nota) -> dict:
|
|
return {
|
|
"id": nota.id,
|
|
"titulo": nota.titulo,
|
|
"tags": ",".join(nota.tags),
|
|
"conexiones": ",".join(nota.conexiones),
|
|
"texto": nota.texto,
|
|
"resumen": nota.resumen,
|
|
"vector": nota.vector,
|
|
"vector_resumen": nota.vector_resumen
|
|
}
|
|
|
|
@staticmethod
|
|
def from_model(model) -> Nota:
|
|
return Nota(
|
|
id=model.id,
|
|
titulo=model.titulo,
|
|
tags=model.tags.split(",") if model.tags else [],
|
|
conexiones=model.conexiones.split(",") if model.conexiones else [],
|
|
texto=model.texto,
|
|
resumen=model.resumen,
|
|
vector=list(model.vector) if model.vector is not None else None,
|
|
vector_resumen=list(model.vector_resumen) if model.vector_resumen is not None else None
|
|
)
|
|
|
|
# ----------------------
|
|
# REPO
|
|
# ----------------------
|
|
|
|
class NotaRepo:
|
|
def __init__(self, session: Session, modelo_nota):
|
|
if modelo_nota is None:
|
|
raise ValueError("No se puede instanciar NotaRepo sin un modelo válido de nota.")
|
|
self.session = session
|
|
self.ModeloNota = modelo_nota
|
|
|
|
def add(self, nota: Nota) -> str:
|
|
model = self.ModeloNota(nota)
|
|
self.session.add(model)
|
|
self.session.commit()
|
|
return model.id
|
|
|
|
def get_by_id(self, id_: str) -> Nota | None:
|
|
model = self.session.get(self.ModeloNota, id_)
|
|
return NotaMapper.from_model(model) if model else None
|
|
|
|
def get_all(self) -> list[Nota]:
|
|
models = self.session.query(self.ModeloNota).all()
|
|
return [NotaMapper.from_model(m) for m in models]
|