processor: extract_file_text

Author: SteveLauC

coco.yml

@@ -323,7 +323,8 @@ pipeline:
             group: enriched_documents
             fetch_max_messages: 10
           processor:
-            - read_file_content:
+            - read_file_content: {}
+            - extract_file_text:
                 output_queue:
                   name: "enriched_documents"
 

core/document.go

@@ -43,6 +43,7 @@ type Document struct {
 
 	Lang    string `json:"lang,omitempty" elastic_mapping:"lang:{type:keyword,copy_to:combined_fulltext}"`    // Language code (e.g., "en", "fr")
 	Content string `json:"content,omitempty" elastic_mapping:"content:{type:text,copy_to:combined_fulltext}"` // Document content for full-text indexing
+	Text    string `json:"text,omitempty" elastic_mapping:"content:{type:text,copy_to:combined_fulltext}"`    // Document content in text for full-text indexing
 
 	Icon      string `json:"icon,omitempty" elastic_mapping:"icon:{enabled:false}"`           // Icon Key, need work with datasource's assets to get the icon url, if it is a full url, then use it directly
 	Thumbnail string `json:"thumbnail,omitempty" elastic_mapping:"thumbnail:{enabled:false}"` // Thumbnail image URL, for preview purposes

plugins/processors/extract_file_text/processor.go

@@ -0,0 +1,98 @@
+package extract_file_text
+
+import (
+	"fmt"
+	"os/exec"
+
+	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("extract_file_text", New)
+}
+
+type ExtractFileTextProcessor 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) {
+	cfg := Config{
+		MessageField: "messages",
+	}
+	if err := c.Unpack(&cfg); err != nil {
+		return nil, err
+	}
+
+	p := &ExtractFileTextProcessor{config: &cfg}
+
+	if cfg.OutputQueue != nil {
+		p.outputQueue = queue.SmartGetOrInitConfig(cfg.OutputQueue)
+	}
+
+	return p, nil
+}
+
+func (p *ExtractFileTextProcessor) Name() string {
+	return "extract_file_text"
+}
+
+func (p *ExtractFileTextProcessor) Process(ctx *pipeline.Context) error {
+	obj := ctx.Get(p.config.MessageField)
+	if obj == nil {
+		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 {
+			// Call extract-cli <doc.Url>
+			cmd := exec.Command("extract-cli", doc.URL)
+			output, err := cmd.Output()
+			if err != nil {
+				log.Errorf("failed to extract text from %s: %v", doc.URL, err)
+				// We might want to continue even if extraction fails,
+				// or maybe we just don't update the Text field.
+			} else {
+				doc.Text = string(output)
+				fmt.Printf("DBG: extracted text length: %d\n", len(doc.Text))
+
+				// Update msg.Data with the new document content
+				updatedDocBytes := util.MustToJSONBytes(doc)
+				msg.Data = updatedDocBytes
+			}
+		}
+
+		if p.outputQueue != nil {
+			if err := queue.Push(p.outputQueue, msg.Data); err != nil {
+				log.Errorf("failed to push document to [%s]'s output queue: %v", p.Name(), err)
+			}
+		}
+	}
+	return nil
+}