#!/usr/bin/env python3 """ Loads a PDF document into TrustGraph processing. """ import pulsar from pulsar.schema import JsonSchema from trustgraph.schema import Document, document_ingest_queue, Metadata import base64 import hashlib import argparse import os import time import uuid from trustgraph.log_level import LogLevel default_user = 'trustgraph' default_collection = 'default' class MetadataSource: def __init__(self, args): self.source = args def id_hash(self, data): # Create a SHA256 hash from the data id = hashlib.sha256(data).hexdigest() # Convert into a UUID, 64-byte hash becomes 32-byte UUID id = str(uuid.UUID(id[::2])) return id def add(self, id, add): source = self.source add(id, IS_A, DIGITAL_DOCUMENT) if source.name: add(id, LABEL, source.name) add(id, NAME, source.name) if source.description: add(id, DESCRIPTION, source.description) if source.copyright_notice: add(id, COPYRIGHT_NOTICE, source.copyright_notice) if source.copyright_holder: add(id, COPYRIGHT_HOLDER, source.copyright_holder) if source.copyright_year: add(id, COPYRIGHT_YEAR, source.copyright_year) if source.license: add(id, LICENSE, source.license) if source.publication_organization: pub_id = self.id_hash( source.publication_organization.encode("utf-8") ) add(id, IS_A, PUBLICATION_EVENT) add(id, PUBLICATION, pub_id) if source.publication_organization: add(pub_id, PUBLISHED_BY, source.publication_organization) if source.publication_description: add(pub_id, DESCRIPTION, source.publication_description) if source.publication_date: add(pub_id, START_DATE, source.publication_date) if source.publication_date: add(pub_id, END_DATE, source.publication_date) if source.url: add(id, URL, source.url) if source.identifier: add(id, IDENTIFIER, source.identifier) if source.keyword: for k in source.keyword: add(id, KEYWORD, k) IS_A = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type' LABEL = 'http://www.w3.org/2000/01/rdf-schema#label' DIGITAL_DOCUMENT = 'https://schema.org/DigitalDocument' PUBLICATION_EVENT = 'https://schema.org/PublicationEvent' NAME = 'https://schema.org/name' DESCRIPTION = 'https://schema.org/description' COPYRIGHT_NOTICE = 'https://schema.org/copyrightNotice' COPYRIGHT_HOLDER = 'https://schema.org/copyrightHolder' COPYRIGHT_YEAR = 'https://schema.org/copyrightYear' LICENSE = 'https://schema.org/license' PUBLICATION = 'https://schema.org/publication' START_DATE = 'https://schema.org/startDate' END_DATE = 'https://schema.org/endDate' PUBLISHED_BY = 'https://schema.org/publishedBy' DATE_PUBLISHED = 'https://schema.org/datePublished' PUBLICATION = 'https://schema.org/publication' DATE_PUBLISHED = 'https://schema.org/datePublished' URL = 'https://schema.org/url' IDENTIFIER = 'https://schema.org/identifier' KEYWORD = 'https://schema.org/keywords' class Loader: def __init__( self, pulsar_host, output_queue, user, collection, log_level, metadata, ): # 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, # ) class Bunch: def close(self): print("CLOSE") self.client = Bunch() self.user = user self.collection = collection self.metadata = metadata def load(self, files): for file in files: self.load_file(file) def load_file(self, file): try: path = file data = open(path, "rb").read() # Create a SHA256 hash from the data id = hashlib.sha256(data).hexdigest() # Convert into a UUID, 64-byte hash becomes 32-byte UUID id = str(uuid.UUID(id[::2])) triples = [] def add(s, p, o): triples.append((s, p, o)) self.metadata.add(id, add) for t in triples: print(t) return r = Document( metadata=Metadata( source=path, title=path, id=id, user=self.user, collection=self.collection, ), data=base64.b64encode(data), ) self.producer.send(r) print(f"{file}: Loaded successfully.") except Exception as e: print(f"{file}: Failed: {str(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( '-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( '--name', help=f'Document name' ) parser.add_argument( '--description', help=f'Document description' ) parser.add_argument( '--copyright-notice', help=f'Copyright notice' ) parser.add_argument( '--copyright-holder', help=f'Copyright holder' ) parser.add_argument( '--copyright-year', help=f'Copyright year' ) parser.add_argument( '--license', help=f'Copyright license' ) parser.add_argument( '--publication-organization', help=f'Publication organization' ) parser.add_argument( '--publication-description', help=f'Publication description' ) parser.add_argument( '--publication-date', help=f'Publication date' ) parser.add_argument( '--url', help=f'Document URL' ) parser.add_argument( '--keyword', nargs='+', help=f'Keyword' ) parser.add_argument( '--identifier', '--id', help=f'Document ID' ) parser.add_argument( '-l', '--log-level', type=LogLevel, default=LogLevel.ERROR, choices=list(LogLevel), help=f'Output queue (default: info)' ) parser.add_argument( 'files', nargs='+', help=f'File to load' ) args = parser.parse_args() while True: try: p = Loader( pulsar_host=args.pulsar_host, output_queue=args.output_queue, user=args.user, collection=args.collection, log_level=args.log_level, metadata=MetadataSource(args), ) p.load(args.files) print("All done.") break except Exception as e: print("Exception:", e, flush=True) print("Will retry...", flush=True) time.sleep(10) main()