138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
from __future__ import annotations
|
|
from typing import Optional
|
|
from sqlalchemy import Column, String, Text, JSON
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import Session
|
|
from pgvector.sqlalchemy import Vector # ✅ importación correcta para pgvector
|
|
from domains.arquitecture_layer.Model import Model_base
|
|
from domains.arquitecture_layer.Mapper import Mapper_base
|
|
from domains.arquitecture_layer.Repo import Repo_base
|
|
|
|
from .embedder import NomicEmbedder
|
|
|
|
Base = declarative_base()
|
|
|
|
class notaDom:
|
|
def __init__(
|
|
self,
|
|
titulo: str,
|
|
contenido: str,
|
|
embedder: list[float], # ✅ sigue siendo list[float] en Python
|
|
tags: Optional[dict] = None,
|
|
relaciones: Optional[dict] = None,
|
|
propiedades: Optional[dict] = None, # ✅ nueva propiedad
|
|
):
|
|
self.titulo = titulo
|
|
self.contenido = contenido
|
|
self.embedder = embedder
|
|
self.tags = tags
|
|
self.relaciones = relaciones
|
|
self.propiedades = propiedades
|
|
|
|
# ------------------------
|
|
|
|
class notaModel(Model_base):
|
|
__tablename__ = "nota"
|
|
__table_args__ = ({"schema": "public"})
|
|
|
|
titulo = Column(String(255), nullable=False, comment="Título de la nota")
|
|
contenido = Column(Text, nullable=True, comment="Contenido de la nota")
|
|
embedder = Column(Vector(768), nullable=True, comment="Vectores de embedder de tamaño 768") # ✅ ahora pgvector
|
|
tags = Column(JSON, nullable=True, default=None, comment="Tags en formato JSON")
|
|
relaciones = Column(JSON, nullable=True, default=None, comment="Relaciones en formato JSON")
|
|
propiedades = Column(JSON, nullable=True, default=None, comment="Propiedades adicionales en formato JSON")
|
|
|
|
# ------------------
|
|
|
|
class notaMapper(Mapper_base[notaDom, notaModel]):
|
|
emb = NomicEmbedder.get_instance()
|
|
|
|
@staticmethod
|
|
def to_model(obj: notaDom) -> notaModel:
|
|
# Auto-generate embedding if not provided
|
|
if obj.embedder is None:
|
|
obj.embedder = notaMapper.emb.embed(obj.contenido)
|
|
|
|
return notaModel(
|
|
titulo=obj.titulo,
|
|
contenido=obj.contenido,
|
|
embedder=obj.embedder,
|
|
tags=obj.tags,
|
|
relaciones=obj.relaciones,
|
|
propiedades=obj.propiedades,
|
|
)
|
|
|
|
@staticmethod
|
|
def from_model(model: notaModel) -> notaDom:
|
|
return notaDom(
|
|
titulo=model.titulo,
|
|
contenido=model.contenido,
|
|
embedder=model.embedder,
|
|
tags=model.tags,
|
|
relaciones=model.relaciones,
|
|
propiedades=model.propiedades,
|
|
)
|
|
|
|
@staticmethod
|
|
def from_dict(d: dict) -> notaDom:
|
|
# Auto-generate embedding if missing
|
|
emb = d.get("embedder")
|
|
if emb is None and "contenido" in d:
|
|
emb = notaMapper.emb.embed(d["contenido"])
|
|
|
|
return notaDom(
|
|
titulo=d["titulo"],
|
|
contenido=d["contenido"],
|
|
embedder=emb,
|
|
tags=d.get("tags"),
|
|
relaciones=d.get("relaciones"),
|
|
propiedades=d.get("propiedades"),
|
|
)
|
|
|
|
|
|
# -----------------
|
|
|
|
class notaRepo(Repo_base[notaModel, notaDom]):
|
|
def __init__(self, session: Session):
|
|
super().__init__(session=session, modelo=notaModel, mapper=notaMapper)
|
|
|
|
def update_contenido(self, nota: notaModel, nuevo_contenido: str) -> notaDom:
|
|
emb = notaMapper.emb.embed(nuevo_contenido)
|
|
nota.contenido = nuevo_contenido
|
|
nota.embedder = emb
|
|
self.session.commit()
|
|
self.session.refresh(nota)
|
|
return self.Mapper.from_model(nota)
|
|
|
|
def get_by_titulo(self, titulo: str) -> Optional[notaDom]:
|
|
model = (
|
|
self.session.query(self.Modelo)
|
|
.filter_by(titulo=titulo, sys_deleted_at=None)
|
|
.first()
|
|
)
|
|
return self.Mapper.from_model(model) if model else None
|
|
|
|
def buscar_por_contenido(self, contenido: str) -> list[notaDom]:
|
|
models = (
|
|
self.session.query(self.Modelo)
|
|
.filter(
|
|
self.Modelo.contenido.ilike(f"%{contenido}%"),
|
|
self.Modelo.sys_deleted_at.is_(None),
|
|
)
|
|
.all()
|
|
)
|
|
return self.Mapper.from_model_list(models)
|
|
|
|
def get_paginated_by_tags(self, tags: dict, offset: int = 0, limit: int = 10) -> list[notaDom]:
|
|
models = (
|
|
self.session.query(self.Modelo)
|
|
.filter(
|
|
self.Modelo.tags.contains(tags),
|
|
self.Modelo.sys_deleted_at.is_(None),
|
|
)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return self.Mapper.from_model_list(models)
|