processor: read_file_content

Author: SteveLauC

coco.yml

@@ -1,5 +1,5 @@
 env:
-  ES_ENDPOINT: https://localhost:9200
+  ES_ENDPOINT: http://192.168.3.119:9200
   ES_USERNAME: admin
   ES_PASSWORD: $[[keystore.ES_PASSWORD]]
   WEB_BINDING: 0.0.0.0:9000
@@ -311,7 +311,7 @@ web:
 ##background jobs
 pipeline:
   - name: enrich_documents
-    auto_start: false
+    auto_start: true
     keep_running: true
     processor:
       - consumer:
@@ -323,21 +323,25 @@ pipeline:
             group: enriched_documents
             fetch_max_messages: 10
           processor:
-            - document_summarization:
-                model: $[[env.ENRICHMENT_MODEL]]
-                input_queue: "indexing_documents"
-                min_input_document_length: 500
+            - read_file_content:
                 output_queue:
                   name: "enriched_documents"
-                  label:
-                    tag: "enriched"
+
+            # - document_summarization:
+            #     model: $[[env.ENRICHMENT_MODEL]]
+            #     input_queue: "indexing_documents"
+            #     min_input_document_length: 500
+            #     output_queue:
+            #       name: "enriched_documents"
+            #       label:
+            #         tag: "enriched"
 
   - name: merge_documents
     auto_start: true
     keep_running: true
     processor:
       - indexing_merge:
-          input_queue: "indexing_documents"
+          input_queue: "enriched_documents"
           idle_timeout_in_seconds: 1
           elasticsearch: "prod"
           index_name: "coco_document-v2"
@@ -362,6 +366,7 @@ pipeline:
           queues:
             type: indexing_merge
             tag: "merged"
+
   - name: connector_dispatcher
     auto_start: true
     keep_running: true

plugins/connectors/local_fs/plugin.go

@@ -154,9 +154,10 @@ func (p *Plugin) saveDocument(ctx *pipeline.Context, currentPath, basePath strin
 		Source:   core.DataSourceReference{ID: datasource.ID, Type: "connector", Name: datasource.Name},
 		Type:     connectors.TypeFile,
 		Category: filepath.Dir(currentPath),
-		Content:  "", // skip content
-		URL:      currentPath,
-		Size:     int(fileInfo.Size()),
+		// skip content here, which will be popluated by the `read_file_content` processor
+		Content: "",
+		URL:     currentPath,
+		Size:    int(fileInfo.Size()),
 	}
 	doc.System = datasource.System
 	if doc.System == nil {

plugins/processors/embedding/processor.go

@@ -147,7 +147,7 @@ func (processor *DocumentSummaryProcessor) Process(ctx *pipeline.Context) error
 
 				if !processor.config.IncludeSkippedDocumentToOutputQueue {
 					continue
-				}+
+				}
 
 				outputBytes = pop
 			}

plugins/processors/read_file_content/processor.go

@@ -0,0 +1,96 @@
+package read_file_content
+
+import (
+	"fmt"
+	"os"
+
+	log "github.com/cihub/seelog"
+	"infini.sh/coco/core"
+	"infini.sh/coco/plugins/connectors"
+	"infini.sh/framework/core/config"
+	"infini.sh/framework/core/param"
+	"infini.sh/framework/core/pipeline"
+	"infini.sh/framework/core/queue"
+	"infini.sh/framework/core/util"
+)
+
+func init() {
+	pipeline.RegisterProcessorPlugin("read_file_content", New)
+}
+
+type ReadFileContentProcessor struct {
+	config      *Config
+	outputQueue *queue.QueueConfig
+}
+
+type Config struct {
+	MessageField param.ParaKey      `config:"message_field"`
+	OutputQueue  *queue.QueueConfig `config:"output_queue"`
+}
+
+func New(c *config.Config) (pipeline.Processor, error) {
+	fmt.Printf("DBG: read_file_content.New invoked\n")
+
+	cfg := Config{
+		MessageField: "messages",
+	}
+	if err := c.Unpack(&cfg); err != nil {
+		return nil, err
+	}
+
+	p := &ReadFileContentProcessor{config: &cfg}
+
+	if cfg.OutputQueue != nil {
+		p.outputQueue = queue.SmartGetOrInitConfig(cfg.OutputQueue)
+	}
+
+	return p, nil
+}
+
+func (p *ReadFileContentProcessor) Name() string {
+	return "read_file_content"
+}
+
+func (p *ReadFileContentProcessor) Process(ctx *pipeline.Context) error {
+	fmt.Printf("DBG: read_file_content.Process invoked.\n")
+
+	obj := ctx.Get(p.config.MessageField)
+	if obj == nil {
+		fmt.Printf("DBG: read_file_content.Process obj is nil for field: %s\n", p.config.MessageField)
+		return nil
+	}
+
+	messages, ok := obj.([]queue.Message)
+	if !ok {
+		return nil
+	}
+
+	for _, msg := range messages {
+		doc := core.Document{}
+
+		docBytes := msg.Data
+		err := util.FromJSONBytes(docBytes, &doc)
+		if err != nil {
+			log.Error("error on handle document:", err)
+			continue
+		}
+
+		if doc.Type == connectors.TypeFile {
+			content, err := os.ReadFile(doc.URL)
+			if err != nil {
+				log.Errorf("failed to read file content from %s: %v", doc.URL, err)
+				continue
+			}
+			doc.Content = string(content)
+			updatedDocBytes := util.MustToJSONBytes(doc)
+			msg.Data = updatedDocBytes
+		}
+
+		if p.outputQueue != nil {
+			if err := queue.Push(p.outputQueue, msg.Data); err != nil {
+				log.Error("failed to push document to [%s]'s output queue: %v\n", p.Name(), err)
+			}
+		}
+	}
+	return nil
+}