trustgraph/trustgraph-cli/trustgraph/cli/load_knowledge.py

"""
Loads triples and entity contexts into the knowledge graph.
"""

import asyncio
import argparse
import os
import time
import rdflib
import json
from websockets.asyncio.client import connect
from typing import List, Dict, Any

from trustgraph.log_level import LogLevel

default_url = os.getenv("TRUSTGRAPH_URL", 'ws://localhost:8088/')
default_user = 'trustgraph'
default_collection = 'default'

class KnowledgeLoader:

    def __init__(
            self,
            files,
            flow,
            user,
            collection,
            document_id,
            url = default_url,
    ):

        if not url.endswith("/"):
            url += "/"

        self.triples_url = url + f"api/v1/flow/{flow}/import/triples"
        self.entity_contexts_url = url + f"api/v1/flow/{flow}/import/entity-contexts"

        self.files = files
        self.user = user
        self.collection = collection
        self.document_id = document_id

    async def run(self):

        try:
            # Load triples first
            async with connect(self.triples_url) as ws:
                for file in self.files:
                    await self.load_triples(file, ws)

            # Then load entity contexts
            async with connect(self.entity_contexts_url) as ws:
                for file in self.files:
                    await self.load_entity_contexts(file, ws)

        except Exception as e:
            print(e, flush=True)

    async def load_triples(self, file, ws):

        g = rdflib.Graph()
        g.parse(file, format="turtle")

        def Value(value, is_uri):
            return { "v": value, "e": is_uri }

        for e in g:
            s = Value(value=str(e[0]), is_uri=True)
            p = Value(value=str(e[1]), is_uri=True)
            if type(e[2]) == rdflib.term.URIRef:
                o = Value(value=str(e[2]), is_uri=True)
            else:
                o = Value(value=str(e[2]), is_uri=False)

            req = {
                "metadata": {
                    "id": self.document_id,
                    "metadata": [],
                    "user": self.user,
                    "collection": self.collection
                },
                "triples": [
                    {
                        "s": s,
                        "p": p,
                        "o": o,
                    }
                ]
            }

            await ws.send(json.dumps(req))

    async def load_entity_contexts(self, file, ws):
        """
        Load entity contexts by extracting entities from the RDF graph
        and generating contextual descriptions based on their relationships.
        """

        g = rdflib.Graph()
        g.parse(file, format="turtle")

        for s, p, o in g:
            # If object is a URI, do nothing
            if isinstance(o, rdflib.term.URIRef):
                continue
                
            # If object is a literal, create entity context for subject with literal as context
            s_str = str(s)
            o_str = str(o)
            
            req = {
                "metadata": {
                    "id": self.document_id,
                    "metadata": [],
                    "user": self.user,
                    "collection": self.collection
                },
                "entities": [
                    {
                        "entity": {
                            "v": s_str,
                            "e": True
                        },
                        "context": o_str
                    }
                ]
            }

            await ws.send(json.dumps(req))


def main():

    parser = argparse.ArgumentParser(
        prog='tg-load-knowledge',
        description=__doc__,
    )

    parser.add_argument(
        '-u', '--api-url',
        default=default_url,
        help=f'API URL (default: {default_url})',
    )

    parser.add_argument(
        '-i', '--document-id',
        required=True,
        help=f'Document ID)',
    )

    parser.add_argument(
        '-f', '--flow-id',
        default="default",
        help=f'Flow ID (default: default)'
    )

    parser.add_argument(
        '-U', '--user',
        default=default_user,
        help=f'User ID (default: {default_user})'
    )

    parser.add_argument(
        '-C', '--collection',
        default=default_collection,
        help=f'Collection ID (default: {default_collection})'
    )


    parser.add_argument(
        'files', nargs='+',
        help=f'Turtle files to load'
    )

    args = parser.parse_args()

    while True:

        try:
            loader = KnowledgeLoader(
                document_id = args.document_id,
                url = args.api_url,
                flow = args.flow_id,
                files = args.files,
                user = args.user,
                collection = args.collection,
            )

            asyncio.run(loader.run())

            print("Triples and entity contexts loaded.")
            break

        except Exception as e:

            print("Exception:", e, flush=True)
            print("Will retry...", flush=True)

        time.sleep(10)

if __name__ == "__main__":
    main()