mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-20 18:51:03 +02:00
Implementing new schema
This commit is contained in:
Cyber MacGeddon
parent
0a95fc1263
commit
4b63806c4a
6 changed files with 136 additions and 116 deletions
|
|
@ -63,7 +63,7 @@ from . documents import Document, TextDocument
|
|||
# search(<key,op,value>[]) : id[]
|
||||
|
||||
class DocumentMetadata(Record):
|
||||
id = String()
|
||||
# id = String()
|
||||
time = Long()
|
||||
kind = String()
|
||||
title = String()
|
||||
|
|
@ -77,17 +77,7 @@ class ProcessingMetadata(Record):
|
|||
flow = String()
|
||||
user = String()
|
||||
collection = String()
|
||||
|
||||
class DocumentInfo(Record):
|
||||
id = String()
|
||||
flow = String()
|
||||
kind = String()
|
||||
user = String()
|
||||
collection = String()
|
||||
title = String()
|
||||
comments = String()
|
||||
time = Long()
|
||||
metadata = Array(Triple())
|
||||
tags = Array(String())
|
||||
|
||||
class Criteria(Record):
|
||||
key = String()
|
||||
|
|
@ -128,8 +118,10 @@ class LibrarianRequest(Record):
|
|||
|
||||
class LibrarianResponse(Record):
|
||||
error = Error()
|
||||
document = DocumentPackage()
|
||||
info = Array(DocumentInfo())
|
||||
document_metadata = DocumentMetadata()
|
||||
content = Bytes()
|
||||
document_metadatas = Array(DocumentMetadata())
|
||||
processing_metadatas = Array(ProcessingMetadata())
|
||||
|
||||
librarian_request_queue = topic(
|
||||
'librarian', kind='non-persistent', namespace='request'
|
||||
|
|
|
|||
|
|
@ -23,21 +23,35 @@ class LibrarianRequestor(ServiceRequestor):
|
|||
|
||||
def to_request(self, body):
|
||||
|
||||
if "document" in body:
|
||||
dp = to_document_package(body["document"])
|
||||
if "document-metadata" in body:
|
||||
dm = to_document_metadata(body["document-metadata"])
|
||||
else:
|
||||
dp = None
|
||||
dm = None
|
||||
|
||||
if "processing-metadata" in body:
|
||||
pm = to_processing_metadata(body["processing-metadata"])
|
||||
else:
|
||||
pm = None
|
||||
|
||||
if "criteria" in body:
|
||||
criteria = to_criteria(body["criteria"])
|
||||
else:
|
||||
criteria = None
|
||||
|
||||
if "content" in body:
|
||||
content = base64.b64decode(
|
||||
body["content"].decode("utf-8")
|
||||
).encode("utf-8")
|
||||
else:
|
||||
content = None
|
||||
|
||||
return LibrarianRequest(
|
||||
operation = body.get("operation", None),
|
||||
id = body.get("id", None),
|
||||
flow = body.get("flow", None),
|
||||
document = dp,
|
||||
document_id = body.get("document-id", None),
|
||||
processing_id = body.get("processing-id", None),
|
||||
document_metadata = dm,
|
||||
processing_metadata = pm,
|
||||
content = content
|
||||
user = body.get("user", None),
|
||||
collection = body.get("collection", None),
|
||||
criteria = criteria,
|
||||
|
|
@ -47,13 +61,26 @@ class LibrarianRequestor(ServiceRequestor):
|
|||
|
||||
response = {}
|
||||
|
||||
if message.document:
|
||||
response["document"] = serialize_document_package(message.document)
|
||||
if message.document_metadata:
|
||||
response["document-metadata"] = serialize_document_metadata(
|
||||
message.document_metadata
|
||||
)
|
||||
|
||||
if message.info:
|
||||
response["info"] = [
|
||||
serialize_document_info(v)
|
||||
for v in message.info
|
||||
if message.content:
|
||||
response["content"] = base64.b64encode(
|
||||
message["content"].decode("utf-8")
|
||||
).encode("utf-8")
|
||||
|
||||
if message.document_metadatas:
|
||||
response["document-metadatas"] = [
|
||||
serialize_document_metadata(v)
|
||||
for v in message.document_metadatas
|
||||
]
|
||||
|
||||
if message.processing_metadatas:
|
||||
response["processing-metadatas"] = [
|
||||
serialize_processing_metadata(v)
|
||||
for v in message.processing_metadatas
|
||||
]
|
||||
|
||||
return response, True
|
||||
|
|
|
|||
|
|
@ -80,96 +80,78 @@ def serialize_document_embeddings(message):
|
|||
],
|
||||
}
|
||||
|
||||
def serialize_document_package(message):
|
||||
def serialize_document_metadata(message):
|
||||
|
||||
ret = {}
|
||||
|
||||
if message.id:
|
||||
ret["id"] = message.id
|
||||
# if message.id:
|
||||
# ret["id"] = message.id
|
||||
|
||||
if message.metadata:
|
||||
ret["metadata"] = serialize_subgraph(message.metdata)
|
||||
|
||||
if message.document:
|
||||
blob = base64.b64encode(
|
||||
message.document.encode("utf-8")
|
||||
).decode("utf-8")
|
||||
ret["document"] = blob
|
||||
if message.time:
|
||||
ret["time"] = message.time
|
||||
|
||||
if message.kind:
|
||||
ret["kind"] = message.kind
|
||||
|
||||
if message.flow:
|
||||
ret["flow"] = message.flow
|
||||
|
||||
if message.user:
|
||||
ret["user"] = message.user
|
||||
|
||||
if message.collection:
|
||||
ret["collection"] = message.collection
|
||||
|
||||
return ret
|
||||
|
||||
def serialize_document_info(message):
|
||||
|
||||
ret = {}
|
||||
|
||||
if message.id:
|
||||
ret["id"] = message.id
|
||||
|
||||
if message.kind:
|
||||
ret["kind"] = message.kind
|
||||
|
||||
if message.flow:
|
||||
ret["flow"] = message.flow
|
||||
|
||||
if message.user:
|
||||
ret["user"] = message.user
|
||||
|
||||
if message.collection:
|
||||
ret["collection"] = message.collection
|
||||
|
||||
if message.title:
|
||||
ret["title"] = message.title
|
||||
|
||||
if message.comments:
|
||||
ret["comments"] = message.comments
|
||||
|
||||
if message.time:
|
||||
ret["time"] = message.time
|
||||
|
||||
if message.metadata:
|
||||
ret["metadata"] = serialize_subgraph(message.metadata)
|
||||
|
||||
if message.tags:
|
||||
ret["tags"] = message.tags
|
||||
|
||||
return ret
|
||||
|
||||
def to_document_package(x):
|
||||
def serialize_processing_metadata(message):
|
||||
|
||||
ret = {}
|
||||
|
||||
if message.id:
|
||||
ret["document_id"] = message.document_id
|
||||
|
||||
if message.time:
|
||||
ret["time"] = message.time
|
||||
|
||||
if message.kind:
|
||||
ret["flow"] = message.flow
|
||||
|
||||
if message.user:
|
||||
ret["user"] = message.user
|
||||
|
||||
if message.collection:
|
||||
ret["collection"] = message.collection
|
||||
|
||||
if message.tags:
|
||||
ret["tags"] = message.tags
|
||||
|
||||
return ret
|
||||
|
||||
def to_document_metadata(x):
|
||||
|
||||
return DocumentPackage(
|
||||
id = x.get("id", None),
|
||||
flow = x.get("flow", None),
|
||||
# id = x.get("id", None),
|
||||
time = x.get("time", None),
|
||||
kind = x.get("kind", None),
|
||||
user = x.get("user", None),
|
||||
collection = x.get("collection", None),
|
||||
title = x.get("title", None),
|
||||
comments = x.get("comments", None),
|
||||
time = x.get("time", None),
|
||||
document = x.get("document", None),
|
||||
metadata = to_subgraph(x["metadata"]),
|
||||
tags = x.get("tags", None),
|
||||
)
|
||||
|
||||
def to_document_info(x):
|
||||
def to_processing_metadata(x):
|
||||
|
||||
return DocumentInfo(
|
||||
id = x.get("id", None),
|
||||
document_id = x.get("document_id", None),
|
||||
time = x.get("time", None),
|
||||
flow = x.get("flow", None),
|
||||
kind = x.get("kind", None),
|
||||
user = x.get("user", None),
|
||||
collection = x.get("collection", None),
|
||||
title = x.get("title", None),
|
||||
comments = x.get("comments", None),
|
||||
time = x.get("time", None),
|
||||
metadata = to_subgraph(x["metadata"]),
|
||||
tags = x.get("tags", None),
|
||||
)
|
||||
|
||||
def to_criteria(x):
|
||||
|
|
|
|||
|
|
@ -26,17 +26,19 @@ class Librarian:
|
|||
self.load_document = load_document
|
||||
self.load_text = load_text
|
||||
|
||||
async def add(self, document):
|
||||
async def add_document(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
if document.kind not in (
|
||||
"text/plain", "application/pdf"
|
||||
):
|
||||
raise RequestError("Invalid document kind: " + document.kind)
|
||||
|
||||
# Create object ID as a hash of the document
|
||||
object_id = uuid.UUID(hash(document.document))
|
||||
# Create object ID for blob
|
||||
object_id = str(uuid.uuid4())
|
||||
|
||||
self.blob_store.add(object_id, document.document, document.kind)
|
||||
self.blob_store.add(object_id, request.content,
|
||||
request.document_metadata.kind)
|
||||
|
||||
self.table_store.add(object_id, document)
|
||||
|
||||
|
|
@ -52,9 +54,36 @@ class Librarian:
|
|||
document = None,
|
||||
info = None,
|
||||
)
|
||||
async def remove_document(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def update_document(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def get_document_metadata(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def get_document_content(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def add_processing(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def remove_processing(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def list_documents(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def list_processing(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
async def list(self, user, collection):
|
||||
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
|
||||
|
||||
print("list")
|
||||
|
||||
info = self.table_store.list(user, collection)
|
||||
|
|
|
|||
|
|
@ -184,39 +184,29 @@ class Processor(AsyncProcessor):
|
|||
|
||||
self.text_load.send(None, doc)
|
||||
|
||||
def parse_request(self, v):
|
||||
def process_request(self, v):
|
||||
|
||||
if v.operation is None:
|
||||
raise RequestError("Null operation")
|
||||
|
||||
print("op", v.operation)
|
||||
|
||||
if v.operation == "add":
|
||||
if (
|
||||
v.document and v.document.id and v.document.metadata and
|
||||
v.document.document and v.document.kind
|
||||
):
|
||||
return partial(
|
||||
self.librarian.add,
|
||||
document = v.document,
|
||||
)
|
||||
else:
|
||||
raise RequestError("Invalid call")
|
||||
impls = {
|
||||
"add-document": self.librarian.add_document,
|
||||
"remove-document": self.librarian.remove_document,
|
||||
"update-document": self.librarian.update_document,
|
||||
"get-document-metadata": self.librarian.get_document_metadata,
|
||||
"get-document-content": self.librarian.get_document_content,
|
||||
"add-processing": self.librarian.add_processing,
|
||||
"remove-processing": self.librarian.remove_processing,
|
||||
"list-documents": self.librarian.list_documents,
|
||||
"list-processing": self.librarian.list_processing,
|
||||
}
|
||||
|
||||
if v.operation == "list":
|
||||
print("list", v)
|
||||
print(v.user)
|
||||
if v.user:
|
||||
return partial(
|
||||
self.librarian.list,
|
||||
user = v.user,
|
||||
collection = v.collection,
|
||||
)
|
||||
else:
|
||||
print("User not specified")
|
||||
raise RequestError("Invalid call")
|
||||
if v.operation not in impls:
|
||||
raise RequestError(f"Invalid operation: {v.operation}")
|
||||
|
||||
raise RequestError("Invalid operation: " + v.operation)
|
||||
return impls[v.operation](v)
|
||||
|
||||
async def on_librarian_request(self, msg, consumer, flow):
|
||||
|
||||
|
|
@ -231,7 +221,7 @@ class Processor(AsyncProcessor):
|
|||
print(f"Handling input {id}...", flush=True)
|
||||
|
||||
try:
|
||||
func = self.parse_request(v)
|
||||
func = self.process_request(v)
|
||||
except RequestError as e:
|
||||
resp = LibrarianResponse(
|
||||
error = Error(
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
from .. schema import LibrarianRequest, LibrarianResponse
|
||||
from .. schema import DocumentInfo, Error, Triple, Value
|
||||
from .. schema import DocumentMetadata, Error, Triple, Value
|
||||
from .. knowledge import hash
|
||||
from .. exceptions import RequestError
|
||||
|
||||
|
|
@ -325,7 +325,7 @@ class TableStore:
|
|||
print("OK2")
|
||||
|
||||
info = [
|
||||
DocumentInfo(
|
||||
DocumentMetadata(
|
||||
id = row[0],
|
||||
kind = row[1],
|
||||
flow = row[2],
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue