mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-20 18:51:03 +02:00
Working on librarian flow
This commit is contained in:
Cyber MacGeddon
parent
bae112ee99
commit
385b02021f
8 changed files with 106 additions and 87 deletions
|
|
@ -4,6 +4,7 @@ import requests
|
|||
import json
|
||||
import sys
|
||||
import base64
|
||||
import time
|
||||
|
||||
url = "http://localhost:8088/api/v1/"
|
||||
|
||||
|
|
@ -11,13 +12,17 @@ url = "http://localhost:8088/api/v1/"
|
|||
|
||||
id = "http://trustgraph.ai/doc/12345678"
|
||||
|
||||
with open("docs/README.cats") as f:
|
||||
doc = base64.b64encode(f.read().encode("utf-8")).decode("utf-8")
|
||||
with open("docs/README.cats", "rb") as f:
|
||||
doc = base64.b64encode(f.read()).decode("utf-8")
|
||||
|
||||
input = {
|
||||
"operation": "add",
|
||||
"document": {
|
||||
"operation": "add-document",
|
||||
"document-metadata": {
|
||||
"id": id,
|
||||
"time": int(time.time()),
|
||||
"kind": "text/plain",
|
||||
"title": "Mark's cats",
|
||||
"comments": "Test doc taken from the TrustGraph repo",
|
||||
"metadata": [
|
||||
{
|
||||
"s": {
|
||||
|
|
@ -46,13 +51,10 @@ input = {
|
|||
},
|
||||
},
|
||||
],
|
||||
"document": doc,
|
||||
"kind": "text/plain",
|
||||
"user": "trustgraph",
|
||||
"collection": "default",
|
||||
"title": "Mark's cats",
|
||||
"comments": "Test doc taken from the TrustGraph repo",
|
||||
}
|
||||
"tags": ["mark", "cats"],
|
||||
},
|
||||
"content": doc,
|
||||
}
|
||||
|
||||
resp = requests.post(
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import requests
|
|||
import json
|
||||
import sys
|
||||
import base64
|
||||
import time
|
||||
|
||||
url = "http://localhost:8088/api/v1/"
|
||||
|
||||
|
|
@ -17,9 +18,13 @@ with open(source, "rb") as f:
|
|||
doc = base64.b64encode(f.read()).decode("utf-8")
|
||||
|
||||
input = {
|
||||
"operation": "add",
|
||||
"id": id,
|
||||
"document": {
|
||||
"operation": "add-document",
|
||||
"document-metadata": {
|
||||
"id": id,
|
||||
"time": int(time.time()),
|
||||
"kind": "application/pdf",
|
||||
"title": "Application of SAE ARP4754A to Flight Critical Systems",
|
||||
"comments": "Application of federal safety standards to NASA spacecraft",
|
||||
"metadata": [
|
||||
{
|
||||
"s": {
|
||||
|
|
@ -61,11 +66,10 @@ input = {
|
|||
},
|
||||
},
|
||||
],
|
||||
"document": doc,
|
||||
"kind": "application/pdf",
|
||||
"user": "trustgraph",
|
||||
"collection": "default",
|
||||
}
|
||||
"tags": ["nasa", "safety-engineering"],
|
||||
},
|
||||
"content": doc,
|
||||
}
|
||||
|
||||
resp = requests.post(
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -73,6 +73,7 @@ class DocumentMetadata(Record):
|
|||
tags = Array(String())
|
||||
|
||||
class ProcessingMetadata(Record):
|
||||
id = String()
|
||||
document_id = String()
|
||||
time = Long()
|
||||
flow = String()
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
|
||||
import base64
|
||||
|
||||
from ... schema import LibrarianRequest, LibrarianResponse
|
||||
from ... schema import librarian_request_queue
|
||||
from ... schema import librarian_response_queue
|
||||
|
|
@ -25,6 +27,7 @@ class LibrarianRequestor(ServiceRequestor):
|
|||
|
||||
def to_request(self, body):
|
||||
|
||||
print("to_request")
|
||||
if "document-metadata" in body:
|
||||
dm = to_document_metadata(body["document-metadata"])
|
||||
else:
|
||||
|
|
@ -40,13 +43,20 @@ class LibrarianRequestor(ServiceRequestor):
|
|||
else:
|
||||
criteria = None
|
||||
|
||||
print("a")
|
||||
print(type(body["content"]))
|
||||
|
||||
if "content" in body:
|
||||
content = base64.b64decode(
|
||||
body["content"].decode("utf-8")
|
||||
).encode("utf-8")
|
||||
body["content"].encode("utf-8")
|
||||
)
|
||||
else:
|
||||
content = None
|
||||
|
||||
print("b")
|
||||
|
||||
print(type(content))
|
||||
|
||||
return LibrarianRequest(
|
||||
operation = body.get("operation", None),
|
||||
document_id = body.get("document-id", None),
|
||||
|
|
@ -61,6 +71,8 @@ class LibrarianRequestor(ServiceRequestor):
|
|||
|
||||
def from_response(self, message):
|
||||
|
||||
print("from_response")
|
||||
|
||||
response = {}
|
||||
|
||||
if message.document_metadata:
|
||||
|
|
@ -70,8 +82,8 @@ class LibrarianRequestor(ServiceRequestor):
|
|||
|
||||
if message.content:
|
||||
response["content"] = base64.b64encode(
|
||||
message["content"].decode("utf-8")
|
||||
).encode("utf-8")
|
||||
message["content"]
|
||||
).decode("utf-8")
|
||||
|
||||
if message.document_metadatas:
|
||||
response["document-metadatas"] = [
|
||||
|
|
|
|||
|
|
@ -84,8 +84,8 @@ def serialize_document_metadata(message):
|
|||
|
||||
ret = {}
|
||||
|
||||
# if message.id:
|
||||
# ret["id"] = message.id
|
||||
if message.id:
|
||||
ret["id"] = message.id
|
||||
|
||||
if message.time:
|
||||
ret["time"] = message.time
|
||||
|
|
@ -114,6 +114,9 @@ def serialize_processing_metadata(message):
|
|||
|
||||
ret = {}
|
||||
|
||||
if message.id:
|
||||
ret["id"] = message.id
|
||||
|
||||
if message.id:
|
||||
ret["document_id"] = message.document_id
|
||||
|
||||
|
|
@ -137,7 +140,7 @@ def serialize_processing_metadata(message):
|
|||
def to_document_metadata(x):
|
||||
|
||||
return DocumentMetadata(
|
||||
# id = x.get("id", None),
|
||||
id = x.get("id", None),
|
||||
time = x.get("time", None),
|
||||
kind = x.get("kind", None),
|
||||
title = x.get("title", None),
|
||||
|
|
@ -150,6 +153,7 @@ def to_document_metadata(x):
|
|||
def to_processing_metadata(x):
|
||||
|
||||
return ProcessingMetadata(
|
||||
id = x.get("id", None),
|
||||
document_id = x.get("document_id", None),
|
||||
time = x.get("time", None),
|
||||
flow = x.get("flow", None),
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
|
||||
from .. schema import LibrarianRequest, LibrarianResponse, Error, Triple
|
||||
from .. knowledge import hash
|
||||
from .. exceptions import RequestError
|
||||
|
|
@ -27,33 +28,41 @@ class Librarian:
|
|||
self.load_text = load_text
|
||||
|
||||
async def add_document(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
if document.kind not in (
|
||||
print(request)
|
||||
|
||||
if request.document_metadata.kind not in (
|
||||
"text/plain", "application/pdf"
|
||||
):
|
||||
raise RequestError("Invalid document kind: " + document.kind)
|
||||
raise RequestError(
|
||||
"Invalid document kind: " + request.document_metadata.kind
|
||||
)
|
||||
|
||||
# Create object ID for blob
|
||||
object_id = str(uuid.uuid4())
|
||||
object_id = uuid.uuid4()
|
||||
|
||||
print("HERE")
|
||||
self.blob_store.add(object_id, request.content,
|
||||
request.document_metadata.kind)
|
||||
|
||||
self.table_store.add(object_id, document)
|
||||
self.table_store.add_document(request.document_metadata, object_id)
|
||||
print("HERE2")
|
||||
|
||||
if document.kind == "application/pdf":
|
||||
await self.load_document(document)
|
||||
elif document.kind == "text/plain":
|
||||
await self.load_text(document)
|
||||
# if document.kind == "application/pdf":
|
||||
# await self.load_document(document)
|
||||
# elif document.kind == "text/plain":
|
||||
# await self.load_text(document)
|
||||
|
||||
print("Add complete", flush=True)
|
||||
|
||||
return LibrarianResponse(
|
||||
error = None,
|
||||
document = None,
|
||||
info = None,
|
||||
document_metadata = None,
|
||||
content = None,
|
||||
document_metadatas = None,
|
||||
processing_metadatas = None,
|
||||
)
|
||||
|
||||
async def remove_document(self, request):
|
||||
raise RuntimeError("Not implemented")
|
||||
|
||||
|
|
|
|||
|
|
@ -206,6 +206,7 @@ class Processor(AsyncProcessor):
|
|||
if v.operation not in impls:
|
||||
raise RequestError(f"Invalid operation: {v.operation}")
|
||||
|
||||
print("HANDLING...")
|
||||
return await impls[v.operation](v)
|
||||
|
||||
async def on_librarian_request(self, msg, consumer, flow):
|
||||
|
|
@ -221,7 +222,15 @@ class Processor(AsyncProcessor):
|
|||
print(f"Handling input {id}...", flush=True)
|
||||
|
||||
try:
|
||||
func = await self.process_request(v)
|
||||
|
||||
resp = await self.process_request(v)
|
||||
|
||||
await self.librarian_response_producer.send(
|
||||
resp, properties={"id": id}
|
||||
)
|
||||
|
||||
return
|
||||
|
||||
except RequestError as e:
|
||||
resp = LibrarianResponse(
|
||||
error = Error(
|
||||
|
|
@ -249,43 +258,6 @@ class Processor(AsyncProcessor):
|
|||
|
||||
return
|
||||
|
||||
try:
|
||||
resp = await func()
|
||||
print("->", resp)
|
||||
except RequestError as e:
|
||||
resp = LibrarianResponse(
|
||||
error = Error(
|
||||
type = "request-error",
|
||||
message = str(e),
|
||||
)
|
||||
)
|
||||
|
||||
await self.librarian_response_producer.send(
|
||||
resp, properties={"id": id}
|
||||
)
|
||||
|
||||
return
|
||||
except Exception as e:
|
||||
print("Exception:", e, flush=True)
|
||||
resp = LibrarianResponse(
|
||||
error = Error(
|
||||
type = "processing-error",
|
||||
message = "Unhandled error: " + str(e),
|
||||
)
|
||||
)
|
||||
|
||||
await self.librarian_response_producer.send(
|
||||
resp, properties={"id": id}
|
||||
)
|
||||
|
||||
return
|
||||
|
||||
print("Send response..!.", flush=True)
|
||||
|
||||
await self.librarian_response_producer.send(
|
||||
resp, properties={"id": id}
|
||||
)
|
||||
|
||||
print("Done.", flush=True)
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -85,6 +85,21 @@ class TableStore:
|
|||
ON document (object_id)
|
||||
""");
|
||||
|
||||
print("processing table...", flush=True)
|
||||
|
||||
self.cassandra.execute("""
|
||||
CREATE TABLE IF NOT EXISTS processing (
|
||||
id text,
|
||||
document_id text,
|
||||
time timestamp,
|
||||
flow text,
|
||||
user text,
|
||||
collection text,
|
||||
tags list<text>,
|
||||
PRIMARY KEY (user, id)
|
||||
);
|
||||
""");
|
||||
|
||||
return
|
||||
|
||||
print("triples table...", flush=True)
|
||||
|
|
@ -157,8 +172,9 @@ class TableStore:
|
|||
self.insert_document_stmt = self.cassandra.prepare("""
|
||||
INSERT INTO document
|
||||
(
|
||||
id, user, time, kind, title, comments, metadata, tags,
|
||||
object_id
|
||||
id, user, time,
|
||||
kind, title, comments,
|
||||
metadata, tags, object_id
|
||||
)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""")
|
||||
|
|
@ -207,17 +223,17 @@ class TableStore:
|
|||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
""")
|
||||
|
||||
def add(self, object_id, document):
|
||||
def add_document(self, document, object_id):
|
||||
|
||||
if document.kind not in (
|
||||
"text/plain", "application/pdf"
|
||||
):
|
||||
raise RequestError("Invalid document kind: " + document.kind)
|
||||
# if document.kind not in (
|
||||
# "text/plain", "application/pdf"
|
||||
# ):
|
||||
# raise RequestError("Invalid document kind: " + document.kind)
|
||||
|
||||
# Create random doc ID
|
||||
# Create timestamp
|
||||
when = int(time.time() * 1000)
|
||||
|
||||
print("Adding", document.id, object_id)
|
||||
print("Adding document", document.id, object_id)
|
||||
|
||||
metadata = [
|
||||
(
|
||||
|
|
@ -234,10 +250,9 @@ class TableStore:
|
|||
resp = self.cassandra.execute(
|
||||
self.insert_document_stmt,
|
||||
(
|
||||
document.id, document.user, document.collection,
|
||||
document.kind, object_id, when,
|
||||
document.title, document.comments,
|
||||
metadata
|
||||
document.id, document.user, when,
|
||||
document.kind, document.title, document.comments,
|
||||
metadata, document.tags, object_id
|
||||
)
|
||||
)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue