#!/usr/bin/env python3 """ Loads a PDF document into TrustGraph processing. """ import pulsar from pulsar.schema import JsonSchema from trustgraph.schema import Document, Source, document_ingest_queue import base64 import hashlib import argparse import os import time from trustgraph.log_level import LogLevel class Loader: def __init__( self, pulsar_host, output_queue, log_level, file, ): self.client = pulsar.Client( pulsar_host, logger=pulsar.ConsoleLogger(log_level.to_pulsar()) ) self.producer = self.client.create_producer( topic=output_queue, schema=JsonSchema(Document), chunking_enabled=True, ) self.file = file def run(self): try: path = self.file data = open(path, "rb").read() id = hashlib.sha256(path.encode("utf-8")).hexdigest()[0:8] r = Document( source=Source( source=path, title=path, id=id, ), data=base64.b64encode(data), ) self.producer.send(r) except Exception as e: print(e, flush=True) def __del__(self): self.client.close() def main(): parser = argparse.ArgumentParser( prog='loader', description=__doc__, ) default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650') default_output_queue = document_ingest_queue parser.add_argument( '-p', '--pulsar-host', default=default_pulsar_host, help=f'Pulsar host (default: {default_pulsar_host})', ) parser.add_argument( '-o', '--output-queue', default=default_output_queue, help=f'Output queue (default: {default_output_queue})' ) parser.add_argument( '-l', '--log-level', type=LogLevel, default=LogLevel.ERROR, choices=list(LogLevel), help=f'Output queue (default: info)' ) parser.add_argument( '-f', '--file', required=True, help=f'File to load' ) args = parser.parse_args() while True: try: p = Loader( pulsar_host=args.pulsar_host, output_queue=args.output_queue, log_level=args.log_level, file=args.file, ) p.run() print("File loaded.") break except Exception as e: print("Exception:", e, flush=True) print("Will retry...", flush=True) time.sleep(10) main()