diff --git a/services/search/pkg/config/content.go b/services/search/pkg/config/content.go index 515db03b63..1ee17b53b1 100644 --- a/services/search/pkg/config/content.go +++ b/services/search/pkg/config/content.go @@ -11,4 +11,5 @@ type Extractor struct { type ExtractorTika struct { TikaURL string `yaml:"tika_url" env:"SEARCH_EXTRACTOR_TIKA_TIKA_URL" desc:"URL of the tika server." introductionVersion:"1.0.0"` CleanStopWords bool `yaml:"clean_stop_words" env:"SEARCH_EXTRACTOR_TIKA_CLEAN_STOP_WORDS" desc:"Defines if stop words should be cleaned or not. See the documentation for more details." introductionVersion:"1.0.0"` + MaxWorkers int `yaml:"max_workers" env:"SEARCH_EXTRACTOR_TIKA_MAX_WORKERS" desc:"Maximum number of parallel extraction workers. Only effective with open_taki v2. Defaults to 8." introductionVersion:"7.1.0"` } diff --git a/services/search/pkg/config/defaults/defaultconfig.go b/services/search/pkg/config/defaults/defaultconfig.go index 3e504350fa..1e823c249f 100644 --- a/services/search/pkg/config/defaults/defaultconfig.go +++ b/services/search/pkg/config/defaults/defaultconfig.go @@ -51,6 +51,7 @@ func DefaultConfig() *config.Config { Tika: config.ExtractorTika{ TikaURL: "http://127.0.0.1:9998", CleanStopWords: false, + MaxWorkers: 8, }, }, Events: config.Events{ diff --git a/services/search/pkg/content/content.go b/services/search/pkg/content/content.go index e185351678..5c7ead934c 100644 --- a/services/search/pkg/content/content.go +++ b/services/search/pkg/content/content.go @@ -26,6 +26,31 @@ type Document struct { Image *libregraph.Image `json:"image,omitempty"` Location *libregraph.GeoCoordinates `json:"location,omitempty"` Photo *libregraph.Photo `json:"photo,omitempty"` + + // Taki v2 extended fields (populated when extractor is open_taki with v2 protocol) + Taki *TakiExtraction `json:"taki,omitempty"` +} + +// TakiExtraction holds extended extraction results from open_taki v2 protocol. +type TakiExtraction struct { + Method string `json:"method,omitempty"` + Summary string `json:"summary,omitempty"` + Entities []Entity `json:"entities,omitempty"` + Embed []float64 `json:"embedding,omitempty"` + Routing *Routing `json:"routing,omitempty"` +} + +// Entity represents a named entity extracted by the LLM. +type Entity struct { + Type string `json:"type"` + Value string `json:"value"` +} + +// Routing tells the search service where to store content. +type Routing struct { + ContentTarget string `json:"content_target"` // bleve, vector, both, none + MetaTarget string `json:"meta_target"` // always bleve + VectorTarget string `json:"vector_target"` // always vector } func CleanString(content, langCode string) string { diff --git a/services/search/pkg/content/tika.go b/services/search/pkg/content/tika.go index 8b6ac5e60b..b97c39297f 100644 --- a/services/search/pkg/content/tika.go +++ b/services/search/pkg/content/tika.go @@ -1,9 +1,13 @@ package content import ( + "bytes" "context" + "encoding/json" "fmt" + "io" "math" + "net/http" "strconv" "strings" "time" @@ -12,20 +16,30 @@ import ( provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/google/go-tika/tika" "github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool" + "github.com/opencloud-eu/reva/v2/pkg/storagespace" libregraph "github.com/opencloud-eu/libre-graph-api-go" "github.com/opencloud-eu/opencloud/pkg/log" "github.com/opencloud-eu/opencloud/services/search/pkg/config" ) -// Tika is used to extract content from a resource, -// it uses apache tika to retrieve all the data. +// Tika is used to extract content from a resource. +// Supports both Apache Tika and open_taki (drop-in replacement). +// When open_taki is detected, uses the v2 protocol for enhanced extraction. type Tika struct { *Basic Retriever tika *tika.Client + tikaURL string + httpClient *http.Client ContentExtractionSizeLimit uint64 CleanStopWords bool + isTaki bool +} + +// IsTaki returns true if open_taki was detected as the extraction backend. +func (t *Tika) IsTaki() bool { + return t.isTaki } // NewTikaExtractor creates a new Tika instance. @@ -35,23 +49,46 @@ func NewTikaExtractor(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], return nil, err } - tk := tika.NewClient(nil, cfg.Extractor.Tika.TikaURL) + tikaURL := cfg.Extractor.Tika.TikaURL + + tk := tika.NewClient(nil, tikaURL) tkv, err := tk.Version(context.Background()) if err != nil { return nil, err } - logger.Info().Msgf("Tika version: %s", tkv) + + // Detect open_taki by checking the health endpoint + isTaki := false + hc := &http.Client{Timeout: 5 * time.Second} + if resp, err := hc.Get(tikaURL + "/tika"); err == nil { + defer resp.Body.Close() + var health map[string]interface{} + if json.NewDecoder(resp.Body).Decode(&health) == nil { + if name, ok := health["name"].(string); ok && name == "open_taki" { + isTaki = true + logger.Info().Msgf("open_taki detected (version: %v), using v2 protocol", health["version"]) + } + } + } + + if !isTaki { + logger.Info().Msgf("Tika version: %s", tkv) + } return &Tika{ Basic: basic, Retriever: newCS3Retriever(gatewaySelector, logger, cfg.Extractor.CS3AllowInsecure), - tika: tika.NewClient(nil, cfg.Extractor.Tika.TikaURL), + tika: tk, + tikaURL: tikaURL, + httpClient: &http.Client{Timeout: 5 * time.Minute}, ContentExtractionSizeLimit: cfg.ContentExtractionSizeLimit, CleanStopWords: cfg.Extractor.Tika.CleanStopWords, + isTaki: isTaki, }, nil } -// Extract loads a resource from its underlying storage, passes it to tika and processes the result into a Document. +// Extract loads a resource from its underlying storage, passes it to tika/taki +// and processes the result into a Document. func (t Tika) Extract(ctx context.Context, ri *provider.ResourceInfo) (Document, error) { doc, err := t.Basic.Extract(ctx, ri) if err != nil { @@ -77,6 +114,14 @@ func (t Tika) Extract(ctx context.Context, ri *provider.ResourceInfo) (Document, } defer data.Close() + if t.isTaki { + return t.extractTakiV2(ctx, ri, data, doc) + } + return t.extractTikaV1(ctx, ri, data, doc) +} + +// extractTikaV1 is the original Tika extraction path (unchanged behavior). +func (t Tika) extractTikaV1(ctx context.Context, ri *provider.ResourceInfo, data io.ReadCloser, doc Document) (Document, error) { metas, err := t.tika.MetaRecursive(ctx, data) if err != nil { return doc, err @@ -107,6 +152,107 @@ func (t Tika) Extract(ctx context.Context, ri *provider.ResourceInfo) (Document, return doc, nil } +// takiV2Response matches open_taki's v2 JSON response format. +type takiV2Response struct { + Content string `json:"X-TIKA:content"` + ContentType string `json:"Content-Type"` + Method string `json:"X-TAKI:method"` + Meta *struct { + Title string `json:"title"` + Author string `json:"author"` + Created string `json:"created"` + Language string `json:"language"` + DocType string `json:"doc_type"` + } `json:"X-TAKI:meta"` + Entities []Entity `json:"X-TAKI:entities"` + Summary string `json:"X-TAKI:summary"` + Embed []float64 `json:"X-TAKI:embedding"` + Routing *struct { + ContentTarget string `json:"content_target"` + MetaTarget string `json:"meta_target"` + VectorTarget string `json:"vector_target"` + SourceRef string `json:"source_ref"` + } `json:"X-TAKI:routing"` +} + +// extractTakiV2 uses the open_taki v2 protocol for enhanced extraction. +func (t Tika) extractTakiV2(ctx context.Context, ri *provider.ResourceInfo, data io.ReadCloser, doc Document) (Document, error) { + body, err := io.ReadAll(data) + if err != nil { + return doc, fmt.Errorf("reading resource data: %w", err) + } + + sourceRef := storagespace.FormatResourceID(ri.Id) + + req, err := http.NewRequestWithContext(ctx, "PUT", + t.tikaURL+"/rmeta/text", bytes.NewReader(body)) + if err != nil { + return doc, err + } + + req.Header.Set("Content-Type", ri.MimeType) + req.Header.Set("X-Taki-Protocol", "v2") + req.Header.Set("X-Taki-Features", "meta,entities,summary,embedding") + req.Header.Set("X-Taki-Source-Ref", sourceRef) + + resp, err := t.httpClient.Do(req) + if err != nil { + t.logger.Warn().Err(err).Msg("taki v2 request failed, falling back to v1") + return t.extractTikaV1(ctx, ri, io.NopCloser(bytes.NewReader(body)), doc) + } + defer resp.Body.Close() + + var results []takiV2Response + if err := json.NewDecoder(resp.Body).Decode(&results); err != nil { + t.logger.Warn().Err(err).Msg("taki v2 response parse failed, falling back to v1") + return t.extractTikaV1(ctx, ri, io.NopCloser(bytes.NewReader(body)), doc) + } + + if len(results) == 0 { + return doc, nil + } + + r := results[0] + + // Content — always populate (bleve needs it for fulltext) + doc.Content = strings.TrimSpace(r.Content) + + // Title — prefer taki meta over Tika's title field + if r.Meta != nil && r.Meta.Title != "" { + doc.Title = r.Meta.Title + } + + // Build TakiExtraction with routing + enrichments + taki := &TakiExtraction{ + Method: r.Method, + Summary: r.Summary, + } + + if r.Entities != nil { + taki.Entities = r.Entities + } + + if r.Embed != nil { + taki.Embed = r.Embed + } + + if r.Routing != nil { + taki.Routing = &Routing{ + ContentTarget: r.Routing.ContentTarget, + MetaTarget: r.Routing.MetaTarget, + VectorTarget: r.Routing.VectorTarget, + } + } + + doc.Taki = taki + + if t.CleanStopWords && r.Meta != nil && r.Meta.Language != "" { + doc.Content = CleanString(doc.Content, r.Meta.Language) + } + + return doc, nil +} + func (t Tika) getImage(meta map[string][]string) *libregraph.Image { var image *libregraph.Image initImage := func() { diff --git a/services/search/pkg/search/service.go b/services/search/pkg/search/service.go index a13ccd3b3b..44754ac67d 100644 --- a/services/search/pkg/search/service.go +++ b/services/search/pkg/search/service.go @@ -7,6 +7,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1" @@ -64,6 +65,7 @@ type Service struct { engine Engine extractor content.Extractor metrics *metrics.Metrics + cfg *config.Config serviceAccountID string serviceAccountSecret string @@ -81,6 +83,7 @@ func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng E logger: logger, extractor: extractor, metrics: metrics, + cfg: cfg, serviceAccountID: cfg.ServiceAccount.ServiceAccountID, serviceAccountSecret: cfg.ServiceAccount.ServiceAccountSecret, @@ -485,6 +488,34 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, forceRescan bool) } logDocCount(s.engine, s.logger) }() + + // Parallel extraction when taki v2 is detected (LLM benefits from concurrent calls) + isTaki := false + if tikaExtractor, ok := s.extractor.(*content.Tika); ok { + isTaki = tikaExtractor.IsTaki() + } + + indexWorkers := s.cfg.Extractor.Tika.MaxWorkers + if indexWorkers < 1 { + indexWorkers = 8 + } + var workCh chan *provider.Reference + var indexWg sync.WaitGroup + + if isTaki { + s.logger.Info().Int("workers", indexWorkers).Msg("taki v2 detected: parallel indexing enabled") + workCh = make(chan *provider.Reference, indexWorkers*2) + for i := 0; i < indexWorkers; i++ { + indexWg.Add(1) + go func() { + defer indexWg.Done() + for ref := range workCh { + s.doUpsertItem(ref, nil) + } + }() + } + } + err = w.Walk(ownerCtx, &rootID, func(wd string, info *provider.ResourceInfo, err error) error { if err != nil { s.logger.Error().Err(err).Msg("error walking the tree") @@ -502,7 +533,11 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, forceRescan bool) s.logger.Debug().Str("path", ref.Path).Msg("Walking tree") if forceRescan { - s.doUpsertItem(ref, batch) + if isTaki { + workCh <- ref + } else { + s.doUpsertItem(ref, batch) + } return nil } @@ -519,11 +554,20 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, forceRescan bool) return nil } - s.doUpsertItem(ref, batch) + if isTaki { + workCh <- ref + } else { + s.doUpsertItem(ref, batch) + } return nil }) + if isTaki { + close(workCh) + indexWg.Wait() + } + if err != nil { return err } @@ -615,10 +659,29 @@ func (s *Service) doUpsertItem(ref *provider.Reference, batch BatchOperator) { r.ParentID = storagespace.FormatResourceID(parentID) } + // Taki v2 routing: decide what goes to bleve based on content_target + bleveDoc := doc + if doc.Taki != nil && doc.Taki.Routing != nil { + target := doc.Taki.Routing.ContentTarget + if target == "vector" || target == "none" { + // Content should NOT go to bleve — clear it for the bleve index + // but keep Title (meta always goes to bleve) + bleveDoc.Content = "" + s.logger.Debug(). + Str("name", doc.Name). + Str("target", target). + Str("method", doc.Taki.Method). + Msg("taki routing: content excluded from bleve index") + } + } + + bleveResource := r + bleveResource.Document = bleveDoc + if batch != nil { - err = batch.Upsert(r.ID, r) + err = batch.Upsert(r.ID, bleveResource) } else { - err = s.engine.Upsert(r.ID, r) + err = s.engine.Upsert(r.ID, bleveResource) } if err != nil { s.logger.Error().Err(err).Msg("error adding updating the resource in the index") @@ -626,6 +689,18 @@ func (s *Service) doUpsertItem(ref *provider.Reference, batch BatchOperator) { logDocCount(s.engine, s.logger) } + // Taki v2: log extraction summary + if doc.Taki != nil { + s.logger.Info(). + Str("name", doc.Name). + Str("method", doc.Taki.Method). + Int("content_len", len(doc.Content)). + Int("embedding_dims", len(doc.Taki.Embed)). + Int("entities", len(doc.Taki.Entities)). + Str("summary", doc.Taki.Summary). + Msg("taki v2 extraction complete") + } + // determine if metadata needs to be stored in storage as well metadata := map[string]string{} addAudioMetadata(metadata, doc.Audio)