Enhance config loading and data processing

Added support for loading configuration files from a URL and improved data processing by implementing comprehensive Skolemization and graph association. Enhanced bulkLoader with a new flag for archiving, updated documentation, and incremented the version.

Author: fils

VERSION

@@ -1 +1 @@
-2.0.18-df-development
+2.0.19-df-development

docs/httpSPARQL.md

@@ -23,4 +23,15 @@ curl -H 'Accept: application/sparql-results+json' http://coreos.lan:9090/blazegr
 
 ```bash
 curl -H 'Accept: application/sparql-results+json' http://coreos.lan:9090/blazegraph/namespace/iow/sparql --data-urlencode 'query=SELECT (COUNT(DISTINCT ?graph) AS ?namedGraphsCount)(COUNT(*) AS ?triplesCount)WHERE {GRAPH ?graph {?subject ?predicate ?object}}'
+```
+
+
+### Oxigraph
+
+```bash
+curl -i -X PUT  -H 'Content-Type:text/x-nquads'    --data-binary @veupathdb_release.nq  http://localhost:7878/store
+```
+
+```bash
+curl -i -X POST  -H 'Content-Type:text/x-nquads'    --data-binary @veupathdb_release.nq  http://localhost:7878/store
 ```

internal/objects/pipecopy.go

@@ -47,10 +47,6 @@ func PipeCopy(v1 *viper.Viper, mc *minio.Client, name, bucket, prefix, destprefi
 			}
 		}(pw)
 
-		// Set and use a "single file flag" to bypass skolimaization since if it is a single file
-		// the JSON-LD to RDF will correctly map blank nodes.
-		// NOTE:  with a background context we can't get the len(channel) so we have to iterate it.
-		// This is fast, but it means we have to do the ListObjects twice
 		clen := 0
 		sf := false
 		ctx, cancel := context.WithCancel(context.Background())
@@ -67,25 +63,23 @@ func PipeCopy(v1 *viper.Viper, mc *minio.Client, name, bucket, prefix, destprefi
 
 		objectCh := mc.ListObjects(context.Background(), bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: isRecursive})
 
-		// for object := range mc.ListObjects(context.Background(), bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: isRecursive}, doneCh) {
+		lastProcessed := false
+		idList := make([]string, 0)
 		for object := range objectCh {
 			fo, err := mc.GetObject(context.Background(), bucket, object.Key, minio.GetObjectOptions{})
 			if err != nil {
 				fmt.Println(err)
+				continue
 			}
-
 			var b bytes.Buffer
 			bw := bufio.NewWriter(&b)
-
 			_, err = io.Copy(bw, fo)
 			if err != nil {
 				log.Println(err)
+				continue
 			}
-
 			s := string(b.Bytes())
-
 			nq := ""
-			//log.Println("Calling JSONLDtoNQ")
 			if strings.HasSuffix(object.Key, ".nq") {
 				nq = s
 			} else {
@@ -95,33 +89,59 @@ func PipeCopy(v1 *viper.Viper, mc *minio.Client, name, bucket, prefix, destprefi
 					return
 				}
 			}
-
 			var snq string
-
 			if sf {
-				snq = nq //  just pass through the RDF without trying to Skolemize since we ar a single fil
+				snq = nq
 			} else {
 				snq, err = graph.Skolemization(nq, object.Key)
 				if err != nil {
 					return
 				}
 			}
-
-			// 1) get graph URI
 			ctx, err := graph.MakeURN(v1, object.Key)
 			if err != nil {
 				return
 			}
-			// 2) convert NT to NQ
 			csnq, err := graph.NtToNq(snq, ctx)
 			if err != nil {
 				return
 			}
-
 			_, err = pw.Write([]byte(csnq))
 			if err != nil {
 				return
 			}
+			idList = append(idList, ctx)
+			lastProcessed = true
+		}
+
+		// Once we are done with the loop, put in the triples to associate all the graphURIs with the org.
+		if lastProcessed {
+
+			data := `_:b0 <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://schema.org/DataCatalog> .
+_:b0 <https://schema.org/dateCreated> "2024-09-20" .
+_:b0 <https://schema.org/description> "This is an example data catalog containing various datasets from this organization" .
+_:b0 <https://schema.org/provider> _:b1 .
+_:b0 <https://schema.org/publisher> _:b2 .
+_:b1 <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://schema.org/Organization> .
+_:b1 <https://schema.org/name> "Provider XYZ" .
+_:b2 <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://schema.org/Organization> .
+_:b2 <https://schema.org/name> "DeCoder" .
+`
+			for _, item := range idList {
+				data += `_:b0 <https://schema.org/dataset> <` + item + `> .` + "\n"
+			}
+
+			sdata, err := graph.Skolemization(data, "release graph prov for ORG")
+			if err != nil {
+				log.Println(err)
+			}
+
+			// Perform the final write to the pipe here
+			// ilstr := strings.Join(idList, ",")
+			_, err = pw.Write([]byte(sdata))
+			if err != nil {
+				log.Println(err)
+			}
 		}
 	}()
 

internal/services/releases/bulkLoader.go

@@ -57,6 +57,7 @@ func BulkRelease(v1 *viper.Viper, mc *minio.Client) error {
 			return err
 		}
 
+		// TODO  Should this be optional / controlled by flag?
 		// Copy the "latest" graph just made to archive with a date
 		// This means the graph in latests is a duplicate of the most recently dated version in archive/{provider}
 		const layout = "2006-01-02-15-04-05"

pkg/cli/root.go

@@ -18,7 +18,7 @@ import (
 	"github.com/spf13/viper"
 )
 
-var cfgFile, cfgName, cfgPath, nabuConfName string
+var cfgFile, cfgURL, cfgName, cfgPath, nabuConfName string
 var minioVal, portVal, accessVal, secretVal, bucketVal string
 var sslVal, dangerousVal bool
 var viperVal *viper.Viper
@@ -75,6 +75,7 @@ func init() {
 	// Enpoint Server setting var
 	rootCmd.PersistentFlags().StringVar(&endpointVal, "endpoint", "", "end point server set for the SPARQL endpoints")
 
+	rootCmd.PersistentFlags().StringVar(&cfgURL, "cfgURL", "configs", "URL location for config file")
 	rootCmd.PersistentFlags().StringVar(&cfgPath, "cfgPath", "configs", "base location for config files (default is configs/)")
 	rootCmd.PersistentFlags().StringVar(&cfgName, "cfgName", "local", "config file (default is local so configs/local)")
 	rootCmd.PersistentFlags().StringVar(&nabuConfName, "nabuConfName", "nabu", "config file (default is local so configs/local)")
@@ -105,6 +106,11 @@ func initConfig() {
 		if err != nil {
 			log.Fatal("cannot read config %s", err)
 		}
+	} else if cfgURL != "" {
+		viperVal, err = config.ReadNabuConfigURL(cfgURL)
+		if err != nil {
+			log.Fatal("cannot read config URL %s", err)
+		}
 	} else {
 		// Find home directory.
 		//home, err := os.UserHomeDir()

pkg/config/nabuConfig.go

@@ -1,6 +1,12 @@
 package config
 
-import "github.com/spf13/viper"
+import (
+	"fmt"
+	"github.com/spf13/viper"
+	"io"
+	"net/http"
+	"strings"
+)
 
 var nabuTemplate = map[string]interface{}{
 	"minio":   MinioTemplate,
@@ -28,3 +34,58 @@ func ReadNabuConfig(filename string, cfgPath string) (*viper.Viper, error) {
 	err := v.ReadInConfig()
 	return v, err
 }
+
+func ReadNabuConfigURL(configURL string) (*viper.Viper, error) {
+	v := viper.New()
+	for key, value := range nabuTemplate {
+		v.SetDefault(key, value)
+	}
+
+	fmt.Printf("Reading config from URL: %v\n", configURL)
+
+	resp, err := http.Get(configURL)
+	if err != nil {
+		return v, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return v, fmt.Errorf("HTTP request failed with status code %v", resp.StatusCode)
+	}
+
+	// Read the content of the config file
+	configData, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return v, err
+	}
+
+	// Convert configData to a string
+	configString := string(configData)
+
+	// Convert the string to an io.Reader
+	reader := strings.NewReader(configString)
+
+	//v.SetConfigName(fileNameWithoutExtTrimSuffix(filename))
+	//v.AddConfigPath(cfgPath)
+	v.SetConfigType("yaml")
+	//v.BindEnv("headless", "GLEANER_HEADLESS_ENDPOINT")
+	v.BindEnv("minio.address", "MINIO_ADDRESS")
+	v.BindEnv("minio.port", "MINIO_PORT")
+	v.BindEnv("minio.ssl", "MINIO_USE_SSL")
+	v.BindEnv("minio.accesskey", "MINIO_ACCESS_KEY")
+	v.BindEnv("minio.secretkey", "MINIO_SECRET_KEY")
+	v.BindEnv("minio.bucket", "MINIO_BUCKET")
+	v.AutomaticEnv()
+
+	err = v.ReadConfig(reader)
+	if err != nil {
+		fmt.Printf("Error reading config from URL: %v\n", err)
+		return v, err
+	}
+
+	fmt.Printf("Config read from URL: %v\n", v.AllSettings())
+
+	//err = v.ReadInConfig()
+
+	return v, err
+}