Refactoring

Author: nichtich

app.py

@@ -1,6 +1,7 @@
 from flask import Flask, jsonify, request, render_template, send_from_directory, send_file
 from waitress import serve
-from lib import CollectionRegistry, TerminologyRegistry, MappingRegistry, ApiError, NotFound, ValidationError, TripleStore
+from lib import CollectionRegistry, TerminologyRegistry, MappingRegistry, \
+    ApiError, NotFound, ValidationError, TripleStore
 import argparse
 import os
 from pathlib import Path
@@ -18,7 +19,6 @@ def init(**config):
     global collections
     global terminologies
     global mappings
-    global sparql
 
     title = config.get('title', os.getenv('TITLE', 'N4O Graph Importer'))
 
@@ -67,9 +67,9 @@ def status():
     values = {key: str(val) for key, val in app.config.items() if key.islower()}
     try:
         sparql = TripleStore(app.config['sparql'])
-        sparql.insert(app.config['base']+'collection/', '')
+        sparql.insert(f"{app.config['base']}collection/", '')
         values['connected'] = True
-    except Exception as e:
+    except Exception:
         values['connected'] = False
     return values
 

extract-rdf.py

@@ -1,13 +1,11 @@
-from .walk import walk, zipwalk
-from .extract import extractRDF
 from .collections import CollectionRegistry
 from .terminologies import TerminologyRegistry
 from .mappings import MappingRegistry
 from .errors import ApiError, NotFound, NotAllowed, ValidationError, ServerError, ClientError
 from .utils import read_json, write_json
-from .rdf import TripleStore
+from .rdf import TripleStore, triple_iterator
 
 
-__all__ = [walk, zipwalk, extractRDF, CollectionRegistry, TerminologyRegistry,
+__all__ = [CollectionRegistry, TerminologyRegistry, triple_iterator,
            MappingRegistry, ApiError, NotFound, NotAllowed, read_json,
            write_json, ValidationError, ServerError, ClientError, TripleStore]

lib/__init__.py

@@ -25,7 +25,7 @@ class MappingRegistry(Registry):
     def __init__(self, **config):
         super().__init__("mappings", **config)
 
-    def process_received(self, id, original, fmt, log):
+    def preprocess_source(self, id, original, fmt, log):
         if fmt == "ndjson":
             log.append("Converting JSKOS mappings to RDF mapping triples")
             source = open(original)
@@ -41,10 +41,10 @@ def process_received(self, id, original, fmt, log):
                     if type(f) is not list or type(t) is not list or len(f) != 1 or len(t) != 1:
                         continue
                     target.write(f"<{f[0]['uri']}> <{prop}> <{t[0]['uri']}> .\n")
-            except Exception as e:
+            except Exception:
                 raise ValidationError("Failed to convert JSKOS mappings!")
             # TODO: log number of triples
 
         return original
 
-    # TODO: def process_received_rdf to further filter triples from graph
+    # TODO: def preprocess_source_rdf to further filter triples from graph

lib/extract.py

@@ -1,8 +1,10 @@
-from rdflib import Graph, URIRef, Literal, BNode
+from rdflib import URIRef, Literal, BNode
 from SPARQLWrapper import SPARQLWrapper
 import requests
 from pyld import jsonld
 from .errors import ServerError
+from .walk import walk
+import lightrdf
 
 
 def jsonld2nt(doc, context):
@@ -84,43 +86,38 @@ def sparql_to_rdf(binding):
             return Literal(binding['value'])
 
 
-def rdf_receive(source, path, log, namespaces={}, properties=[]):
-    namespaces = tuple(list(namespaces.values()))
-
-    graph = Graph()
-    graph.parse(source)
-    size = len(graph)
-    log.append(f"Parsed {size} unique triples")
-
-    checked = open(path / "checked.nt", "w")
-    removed = open(path / "removed.nt", "w")
+rdfparser = lightrdf.Parser()
+
+
+def triple_iterator(source, log):
+    """Recursively extract RDF triples from a file, directory and/or ZIP archive."""
+    for name, path, archive in walk(source):
+        format = None
+        if name.endswith(".ttl"):
+            format = "turtle"
+        elif name.endswith(".nt"):
+            format = "nt"
+        elif name.endswith(".owl"):
+            format = "owl"
+        elif name.endswith(".rdf"):
+            format = "xml"
+        elif name.endswith(".xml"):
+            # TODO: check whether it's RDF/XML?
+            format = "xml"
+        else:
+            continue
 
-    count = 0
-    for s, p, o in graph.triples((None, None, None)):
-        if str(s).startswith(namespaces):
-            removed.write(f"{s.n3()} {p.n3()} {o.n3()} .\n")
+        if archive:
+            file = archive.open(name)
+            base = f"file://{file.name}"
         else:
-            count = count + 1
-            # TODO: filter out namespaces
-            # if predicate.startswith(rdflib.RDFS)
-            checked.write(f"{s.n3()} {p.n3()} {o.n3()} .\n")
-
-    log.append(
-        f"Removed {size - count} triples, remaining {count} unique triples.")
-
-
-"""
-checked=$stage/checked.nt
-removed=$stage/removed.nt
-namespace=$(jq -r --arg uri "$uri" '.[$uri]' $STAGE/terminology/namespaces.json)
-
-export RDFFILTER_ALLOW_NAMESPACE=$namespace
-npm run --silent -- rdffilter $unique -o $checked --stats -f ./js/rdffilter.js
-# TODO: extend rdffilter for one-pass
-npm run --silent -- rdffilter $unique -o $removed -r -f ./js/rdffilter.js
-unset RDFFILTER_ALLOW_NAMESPACE
-
-wc -l $duplicated
-wc -l $removed
-wc -l $checked
-"""
+            file = f"{'/'.join(path)}/{name}"
+            base = f"file://{file}"
+
+        try:
+            log.append(f"Extracting RDF from {file} as {format}")
+            for triple in rdfparser.parse(file, format=format):
+                yield triple
+        except Exception as e:
+            log.append(f"Error parsing {base}: {e}")
+            continue

lib/mappings.py

@@ -3,7 +3,7 @@
 from shutil import copy, copyfileobj, rmtree
 import urllib
 from jsonschema import validate
-from .rdf import rdf_receive, jsonld2nt, TripleStore
+from .rdf import jsonld2nt, TripleStore, triple_iterator
 from .log import Log
 from .errors import NotFound, ClientError, ValidationError
 from .utils import read_json, write_json, access_location
@@ -105,7 +105,7 @@ def delete(self, id):
         rmtree(self.stage / str(id), ignore_errors=True)
 
     def purge(self):
-        for id in [t["uri"].split("/")[-1] for t in self.list()]:
+        for id in [t["id"] for t in self.list()]:
             self.delete(id)
 
     def load(self, id):
@@ -139,40 +139,13 @@ def forbidden_namespaces(self, id):
         return {}
 
     def receive(self, id, file=None):
-        file, fmt = self.get_source(id, file)
-        original, log = self.receive_source(id, file, fmt)
-        stage = self.stage / str(id)
-        file = self.process_received(id, original, fmt, log)
-        namespaces = self.forbidden_namespaces(id)
-        rdf_receive(file, stage, log, namespaces)
-
+        file, fmt = self.identify_source(id, file)
+        original, log = self.fetch_source(id, file, fmt)
+        file = self.preprocess_source(id, original, fmt, log)
+        self.receive_rdf(id, file, log)
         return log.done()
 
-    def receive_source(self, id, source, fmt):
-        stage = self.stage / str(id)
-        stage.mkdir(exist_ok=True)
-
-        original = stage / f"original.{fmt}"
-        log = Log(stage / "receive.json", f"Receiving {id} from {source}")
-
-        try:
-            if "/" not in source:
-                source = self.data / source
-                log.append(f"Retrieving source {source} from data directory")
-                copy(source, original)
-            else:
-                # TODO: source may be a DOI or similar identifier
-                # ./extract-rdf.py $download_dir $stage/triples.nt
-                log.append(f"Retrieving source from {source}")
-                with urllib.request.urlopen(source) as fsrc, open(original, 'wb') as fdst:
-                    copyfileobj(fsrc, fdst)
-        except Exception as e:
-            log.done(f"Retrieving failed: {e}")
-            raise NotFound(f"{source} not found")
-
-        return (original, log)
-
-    def get_source(self, id, source=None):
+    def identify_source(self, id, source=None):
         item = self.get(id)
         fmt = None
 
@@ -188,7 +161,7 @@ def get_source(self, id, source=None):
             elif fmt == "rdf/xml":
                 fmt = "xml"
 
-        # TODO: configure and extend this
+        # TODO: configure and extend this. Add support of .zip files
         if not fmt:
             if Path(source).suffix in [".nt", ".ttl"]:
                 fmt = "ttl"
@@ -202,5 +175,53 @@ def get_source(self, id, source=None):
 
         return source, fmt
 
-    def process_received(self, id, file, fmt, log):
+    def fetch_source(self, id, source, fmt):
+        stage = self.stage / str(id)
+        stage.mkdir(exist_ok=True)
+
+        original = stage / f"original.{fmt}"
+        log = Log(stage / "receive.json", f"Receiving {id} from {source}")
+
+        try:
+            if "/" not in source:
+                source = self.data / source
+                log.append(f"Retrieving source {source} from data directory")
+                copy(source, original)
+            else:
+                # TODO: source may be a DOI or similar identifier
+                # ./extract-rdf.py $download_dir $stage/triples.nt
+                log.append(f"Retrieving source from {source}")
+                with urllib.request.urlopen(source) as fsrc, open(original, 'wb') as fdst:
+                    copyfileobj(fsrc, fdst)
+        except Exception as e:
+            log.done(f"Retrieving failed: {e}")
+            raise NotFound(f"{source} not found")
+
+        return (original, log)
+
+    def preprocess_source(self, id, file, fmt, log):
         return file
+
+    def receive_rdf(self, id, source, log):
+        namespaces = tuple(list(self.forbidden_namespaces(id).values()))
+
+        stage = self.stage / str(id)
+        checked = open(stage / "checked.nt", "w")
+        removed = open(stage / "removed.nt", "w")
+
+        okCount, removedCount = 0, 0
+        for s, p, o in triple_iterator(source, log):
+            # TODO: implement more filtering and rewrite
+            if str(s)[1:].startswith(namespaces):
+                removedCount = removedCount + 1
+                removed.write(f"{s} {p} {o} .\n")
+            else:
+                okCount = okCount + 1
+                # TODO: filter out namespaces
+                # if predicate.startswith(rdflib.RDFS)
+                checked.write(f"{s} {p} {o} .\n")
+
+        log.append(
+            f"Removed {removedCount} triples, remaining {okCount} unique triples.")
+
+        # TODO: if okCount is zero, raise an error

lib/rdf.py

@@ -40,7 +40,7 @@ def register(self, item):
 
         return super().register(voc[0], id)
 
-    def process_received(self, id, original, fmt, log):
+    def preprocess_source(self, id, original, fmt, log):
         if fmt == "ndjson":
             log.append("Converting JSKOS to RDF")
             with open(original) as file:

lib/registry.py

@@ -1,10 +1,11 @@
 import os
 from zipfile import ZipFile, ZipExtFile
+from pathlib import Path
 
 
 def isZip(file) -> bool:
     """Check whether given file looks like a ZIP archive."""
-    return file.endswith(".zip") or file.endswith(".ZIP")
+    return Path(file).suffix == ".zip" or Path(file).suffix == ".ZIP"
 
 
 def zipwalk(file, path=None) -> list:

lib/terminologies.py

@@ -336,5 +336,3 @@ def test_mappings(client):
     # TODO: check contents
 
     assert client.get("/mappings/2/stage/").status_code == 200
-
-