76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
|
|
from domains.nota import notaDom, notaRepo
|
|
from domains.embedder import NomicEmbedder
|
|
|
|
|
|
class NotaService:
|
|
def __init__(self, session: Session):
|
|
self.repo = notaRepo(session)
|
|
self.emb = NomicEmbedder.get_instance()
|
|
|
|
# -------------------
|
|
# CRUD
|
|
# -------------------
|
|
def create(self, titulo: str, contenido: str, tags: Optional[dict] = None,
|
|
relaciones: Optional[dict] = None, propiedades: Optional[dict] = None) -> str:
|
|
embedding = self.emb.embed(contenido)
|
|
nota = notaDom(titulo, contenido, embedding, tags, relaciones, propiedades)
|
|
return self.repo.add(nota)
|
|
|
|
def get(self, id_: str) -> Optional[notaDom]:
|
|
return self.repo.get_by_id(id_)
|
|
|
|
def list_all(self) -> List[notaDom]:
|
|
return self.repo.get_all()
|
|
|
|
def update(self, id_: str, nuevo_contenido: Optional[str] = None,
|
|
tags: Optional[dict] = None, relaciones: Optional[dict] = None,
|
|
propiedades: Optional[dict] = None) -> bool:
|
|
new_data = {}
|
|
if nuevo_contenido:
|
|
new_data["contenido"] = nuevo_contenido
|
|
new_data["embedder"] = self.emb.embed(nuevo_contenido)
|
|
if tags: new_data["tags"] = tags
|
|
if relaciones: new_data["relaciones"] = relaciones
|
|
if propiedades: new_data["propiedades"] = propiedades
|
|
return self.repo.update(id_, new_data)
|
|
|
|
def delete(self, id_: str) -> bool:
|
|
return self.repo.soft_delete(id_)
|
|
|
|
# -------------------
|
|
# RAG: búsquedas
|
|
# -------------------
|
|
def search_by_text(self, texto: str, top_k: int = 5) -> List[notaDom]:
|
|
emb = self.emb.embed(texto)
|
|
sql = """
|
|
SELECT * FROM public.nota
|
|
WHERE sys_deleted_at IS NULL
|
|
ORDER BY embedder <-> :query_emb
|
|
LIMIT :top_k
|
|
"""
|
|
rows = self.repo.session.execute(sql, {"query_emb": emb, "top_k": top_k}).fetchall()
|
|
return [self.repo.Mapper.from_model(r) for r in rows]
|
|
|
|
def search_by_tags(self, tags: dict, offset: int = 0, limit: int = 10) -> List[notaDom]:
|
|
return self.repo.get_paginated_by_tags(tags, offset, limit)
|
|
|
|
def hybrid_search(self, texto: str, tags: Optional[dict] = None, top_k: int = 5) -> List[notaDom]:
|
|
emb = self.emb.embed(texto)
|
|
sql = """
|
|
SELECT * FROM public.nota
|
|
WHERE sys_deleted_at IS NULL
|
|
{tags_filter}
|
|
ORDER BY embedder <-> :query_emb
|
|
LIMIT :top_k
|
|
"""
|
|
tags_filter = "AND tags @> :tags" if tags else ""
|
|
query = sql.format(tags_filter=tags_filter)
|
|
params = {"query_emb": emb, "top_k": top_k}
|
|
if tags:
|
|
params["tags"] = tags
|
|
rows = self.repo.session.execute(query, params).fetchall()
|
|
return [self.repo.Mapper.from_model(r) for r in rows]
|