Aaron Liu
7 months ago
parent
ec9fdd33bc
commit
1e3b851e19
@ -1 +1 @@
|
|||||||
Subproject commit 21a98194339c89e375ac7e34fdb93d4aa0d213ef
|
Subproject commit 4a38a946cb40c3fe9878af0945c33a8b6987637b
|
||||||
@ -0,0 +1,529 @@
|
|||||||
|
package manager
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/application/dependency"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/ent"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/ent/task"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/inventory"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/inventory/types"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/filemanager/fs"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/filemanager/fs/dbfs"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/hashid"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/logging"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/queue"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/searcher"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/util"
|
||||||
|
"github.com/samber/lo"
|
||||||
|
)
|
||||||
|
|
||||||
|
type (
|
||||||
|
FullTextIndexTask struct {
|
||||||
|
*queue.DBTask
|
||||||
|
}
|
||||||
|
|
||||||
|
FullTextIndexTaskState struct {
|
||||||
|
Uri *fs.URI `json:"uri"`
|
||||||
|
EntityID int `json:"entity_id"`
|
||||||
|
FileID int `json:"file_id"`
|
||||||
|
OwnerID int `json:"owner_id"`
|
||||||
|
}
|
||||||
|
|
||||||
|
ftsFileInfo struct {
|
||||||
|
FileID int
|
||||||
|
OwnerID int
|
||||||
|
EntityID int
|
||||||
|
FileName string
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
func (m *manager) SearchFullText(ctx context.Context, query string, offset int) (*FullTextSearchResults, error) {
|
||||||
|
indexer := m.dep.SearchIndexer(ctx)
|
||||||
|
results, total, err := indexer.Search(ctx, m.user.ID, query, offset)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to search full text: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(results) == 0 {
|
||||||
|
// No results.
|
||||||
|
return &FullTextSearchResults{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Traverse each file in result
|
||||||
|
files := lo.FilterMap(results, func(result searcher.SearchResult, _ int) (FullTextSearchResult, bool) {
|
||||||
|
file, err := m.TraverseFile(ctx, result.FileID)
|
||||||
|
if err != nil {
|
||||||
|
m.l.Debug("Failed to traverse file %d for full text search: %s, skipping.", result.FileID, err)
|
||||||
|
return FullTextSearchResult{}, false
|
||||||
|
}
|
||||||
|
|
||||||
|
return FullTextSearchResult{
|
||||||
|
File: file,
|
||||||
|
Content: result.Text,
|
||||||
|
}, true
|
||||||
|
})
|
||||||
|
|
||||||
|
if len(files) == 0 {
|
||||||
|
// No valid files, run next offset
|
||||||
|
return m.SearchFullText(ctx, query, offset+len(results))
|
||||||
|
}
|
||||||
|
|
||||||
|
return &FullTextSearchResults{
|
||||||
|
Hits: files,
|
||||||
|
Total: total,
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
queue.RegisterResumableTaskFactory(queue.FullTextIndexTaskType, NewFullTextIndexTaskFromModel)
|
||||||
|
queue.RegisterResumableTaskFactory(queue.FullTextCopyTaskType, NewFullTextCopyTaskFromModel)
|
||||||
|
queue.RegisterResumableTaskFactory(queue.FullTextChangeOwnerTaskType, NewFullTextChangeOwnerTaskFromModel)
|
||||||
|
queue.RegisterResumableTaskFactory(queue.FullTextDeleteTaskType, NewFullTextDeleteTaskFromModel)
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewFullTextIndexTask(ctx context.Context, uri *fs.URI, entityID, fileID, ownerID int, creator *ent.User) (*FullTextIndexTask, error) {
|
||||||
|
state := &FullTextIndexTaskState{
|
||||||
|
Uri: uri,
|
||||||
|
EntityID: entityID,
|
||||||
|
FileID: fileID,
|
||||||
|
OwnerID: ownerID,
|
||||||
|
}
|
||||||
|
stateBytes, err := json.Marshal(state)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to marshal state: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &FullTextIndexTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
DirectOwner: creator,
|
||||||
|
Task: &ent.Task{
|
||||||
|
Type: queue.FullTextIndexTaskType,
|
||||||
|
CorrelationID: logging.CorrelationID(ctx),
|
||||||
|
PrivateState: string(stateBytes),
|
||||||
|
PublicState: &types.TaskPublicState{},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewFullTextIndexTaskFromModel(t *ent.Task) queue.Task {
|
||||||
|
return &FullTextIndexTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
Task: t,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type (
|
||||||
|
FullTextCopyTask struct {
|
||||||
|
*queue.DBTask
|
||||||
|
}
|
||||||
|
|
||||||
|
FullTextCopyTaskState struct {
|
||||||
|
Uri *fs.URI `json:"uri"`
|
||||||
|
OriginalFileID int `json:"original_file_id"`
|
||||||
|
FileID int `json:"file_id"`
|
||||||
|
OwnerID int `json:"owner_id"`
|
||||||
|
EntityID int `json:"entity_id"`
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
func NewFullTextCopyTask(ctx context.Context, uri *fs.URI, originalFileID, fileID, ownerID, entityID int, creator *ent.User) (*FullTextCopyTask, error) {
|
||||||
|
state := &FullTextCopyTaskState{
|
||||||
|
Uri: uri,
|
||||||
|
OriginalFileID: originalFileID,
|
||||||
|
FileID: fileID,
|
||||||
|
OwnerID: ownerID,
|
||||||
|
EntityID: entityID,
|
||||||
|
}
|
||||||
|
stateBytes, err := json.Marshal(state)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to marshal state: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &FullTextCopyTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
DirectOwner: creator,
|
||||||
|
Task: &ent.Task{
|
||||||
|
Type: queue.FullTextCopyTaskType,
|
||||||
|
CorrelationID: logging.CorrelationID(ctx),
|
||||||
|
PrivateState: string(stateBytes),
|
||||||
|
PublicState: &types.TaskPublicState{},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewFullTextCopyTaskFromModel(t *ent.Task) queue.Task {
|
||||||
|
return &FullTextCopyTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
Task: t,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *FullTextCopyTask) Do(ctx context.Context) (task.Status, error) {
|
||||||
|
dep := dependency.FromContext(ctx)
|
||||||
|
l := dep.Logger()
|
||||||
|
fm := NewFileManager(dep, inventory.UserFromContext(ctx)).(*manager)
|
||||||
|
|
||||||
|
if !fm.settings.FTSEnabled(ctx) {
|
||||||
|
l.Debug("FTS disabled, skipping full text copy task.")
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
var state FullTextCopyTaskState
|
||||||
|
if err := json.Unmarshal([]byte(t.State()), &state); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to unmarshal state: %s (%w)", err, queue.CriticalErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get fresh file to make sure task is not stale.
|
||||||
|
file, err := fm.Get(ctx, state.Uri, dbfs.WithFilePublicMetadata())
|
||||||
|
if err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to get latest file: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if file.PrimaryEntityID() != state.EntityID {
|
||||||
|
l.Debug("File %d entity changed, skipping copy index.", state.FileID)
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
indexer := dep.SearchIndexer(ctx)
|
||||||
|
if err := indexer.CopyByFileID(ctx, state.OriginalFileID, state.FileID, state.OwnerID, state.EntityID); err != nil {
|
||||||
|
l.Warning("Failed to copy index from file %d to %d, falling back to full indexing: %s", state.OriginalFileID, state.FileID, err)
|
||||||
|
return performIndexing(ctx, fm, state.Uri, state.EntityID, state.FileID, state.OwnerID, file.Name(), false)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Patch metadata to mark file as indexed.
|
||||||
|
if err := fm.fs.PatchMetadata(ctx, []*fs.URI{state.Uri}, fs.MetadataPatch{
|
||||||
|
Key: dbfs.FullTextIndexKey,
|
||||||
|
Value: hashid.EncodeEntityID(fm.hasher, state.EntityID),
|
||||||
|
}); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to patch metadata: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
l.Debug("Successfully copied index from file %d to %d.", state.OriginalFileID, state.FileID)
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type (
|
||||||
|
FullTextChangeOwnerTask struct {
|
||||||
|
*queue.DBTask
|
||||||
|
}
|
||||||
|
|
||||||
|
FullTextChangeOwnerTaskState struct {
|
||||||
|
Uri *fs.URI `json:"uri"`
|
||||||
|
EntityID int `json:"entity_id"`
|
||||||
|
FileID int `json:"file_id"`
|
||||||
|
OriginalOwnerID int `json:"original_owner_id"`
|
||||||
|
NewOwnerID int `json:"new_owner_id"`
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
func NewFullTextChangeOwnerTask(ctx context.Context, uri *fs.URI, entityID, fileID, originalOwnerID, newOwnerID int, creator *ent.User) (*FullTextChangeOwnerTask, error) {
|
||||||
|
state := &FullTextChangeOwnerTaskState{
|
||||||
|
Uri: uri,
|
||||||
|
EntityID: entityID,
|
||||||
|
FileID: fileID,
|
||||||
|
OriginalOwnerID: originalOwnerID,
|
||||||
|
NewOwnerID: newOwnerID,
|
||||||
|
}
|
||||||
|
stateBytes, err := json.Marshal(state)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to marshal state: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &FullTextChangeOwnerTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
DirectOwner: creator,
|
||||||
|
Task: &ent.Task{
|
||||||
|
Type: queue.FullTextChangeOwnerTaskType,
|
||||||
|
CorrelationID: logging.CorrelationID(ctx),
|
||||||
|
PrivateState: string(stateBytes),
|
||||||
|
PublicState: &types.TaskPublicState{},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewFullTextChangeOwnerTaskFromModel(t *ent.Task) queue.Task {
|
||||||
|
return &FullTextChangeOwnerTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
Task: t,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *FullTextChangeOwnerTask) Do(ctx context.Context) (task.Status, error) {
|
||||||
|
dep := dependency.FromContext(ctx)
|
||||||
|
l := dep.Logger()
|
||||||
|
fm := NewFileManager(dep, inventory.UserFromContext(ctx)).(*manager)
|
||||||
|
|
||||||
|
if !fm.settings.FTSEnabled(ctx) {
|
||||||
|
l.Debug("FTS disabled, skipping full text change owner task.")
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
var state FullTextChangeOwnerTaskState
|
||||||
|
if err := json.Unmarshal([]byte(t.State()), &state); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to unmarshal state: %s (%w)", err, queue.CriticalErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get fresh file to make sure task is not stale.
|
||||||
|
file, err := fm.Get(ctx, state.Uri, dbfs.WithFilePublicMetadata())
|
||||||
|
if err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to get latest file: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if file.PrimaryEntityID() != state.EntityID {
|
||||||
|
l.Debug("File %d entity changed, skipping owner change.", state.FileID)
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
indexer := dep.SearchIndexer(ctx)
|
||||||
|
if err := indexer.ChangeOwner(ctx, state.FileID, state.OriginalOwnerID, state.NewOwnerID); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to change owner for file %d: %w", state.FileID, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
l.Debug("Successfully changed index owner for file %d from %d to %d.", state.FileID, state.OriginalOwnerID, state.NewOwnerID)
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type (
|
||||||
|
FullTextDeleteTask struct {
|
||||||
|
*queue.DBTask
|
||||||
|
}
|
||||||
|
|
||||||
|
FullTextDeleteTaskState struct {
|
||||||
|
FileIDs []int `json:"file_ids"`
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
func NewFullTextDeleteTask(ctx context.Context, fileIDs []int, creator *ent.User) (*FullTextDeleteTask, error) {
|
||||||
|
state := &FullTextDeleteTaskState{
|
||||||
|
FileIDs: fileIDs,
|
||||||
|
}
|
||||||
|
stateBytes, err := json.Marshal(state)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to marshal state: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &FullTextDeleteTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
DirectOwner: creator,
|
||||||
|
Task: &ent.Task{
|
||||||
|
Type: queue.FullTextDeleteTaskType,
|
||||||
|
CorrelationID: logging.CorrelationID(ctx),
|
||||||
|
PrivateState: string(stateBytes),
|
||||||
|
PublicState: &types.TaskPublicState{},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewFullTextDeleteTaskFromModel(t *ent.Task) queue.Task {
|
||||||
|
return &FullTextDeleteTask{
|
||||||
|
DBTask: &queue.DBTask{
|
||||||
|
Task: t,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *FullTextDeleteTask) Do(ctx context.Context) (task.Status, error) {
|
||||||
|
dep := dependency.FromContext(ctx)
|
||||||
|
l := dep.Logger()
|
||||||
|
|
||||||
|
var state FullTextDeleteTaskState
|
||||||
|
if err := json.Unmarshal([]byte(t.State()), &state); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to unmarshal state: %s (%w)", err, queue.CriticalErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
indexer := dep.SearchIndexer(ctx)
|
||||||
|
if err := indexer.DeleteByFileIDs(ctx, state.FileIDs...); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to delete index for %d file(s): %w", len(state.FileIDs), err)
|
||||||
|
}
|
||||||
|
|
||||||
|
l.Debug("Successfully deleted index for %d file(s).", len(state.FileIDs))
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *FullTextIndexTask) Do(ctx context.Context) (task.Status, error) {
|
||||||
|
dep := dependency.FromContext(ctx)
|
||||||
|
l := dep.Logger()
|
||||||
|
fm := NewFileManager(dep, inventory.UserFromContext(ctx)).(*manager)
|
||||||
|
|
||||||
|
// Check FTS enabled
|
||||||
|
if !fm.settings.FTSEnabled(ctx) {
|
||||||
|
l.Debug("FTS disabled, skipping full text index task.")
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Unmarshal state
|
||||||
|
var state FullTextIndexTaskState
|
||||||
|
if err := json.Unmarshal([]byte(t.State()), &state); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to unmarshal state: %s (%w)", err, queue.CriticalErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get fresh file to make sure task is not stale
|
||||||
|
file, err := fm.Get(ctx, state.Uri, dbfs.WithFilePublicMetadata())
|
||||||
|
if err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to get latest file: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if file.PrimaryEntityID() != state.EntityID {
|
||||||
|
l.Debug("File %d is not the latest version, skipping indexing.", state.FileID)
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
deleteOldChunks := false
|
||||||
|
if _, ok := file.Metadata()[dbfs.FullTextIndexKey]; ok {
|
||||||
|
deleteOldChunks = true
|
||||||
|
}
|
||||||
|
|
||||||
|
return performIndexing(ctx, fm, state.Uri, state.EntityID, state.FileID, state.OwnerID, state.Uri.Name(), deleteOldChunks)
|
||||||
|
}
|
||||||
|
|
||||||
|
// performIndexing extracts text from the entity and indexes it. This is shared between
|
||||||
|
// the regular index task and the copy task (as a fallback when copy fails).
|
||||||
|
func performIndexing(ctx context.Context, fm *manager, uri *fs.URI, entityID, fileID, ownerID int, fileName string, deleteOldChunks bool) (task.Status, error) {
|
||||||
|
dep := fm.dep
|
||||||
|
l := dep.Logger()
|
||||||
|
|
||||||
|
// Get entity source
|
||||||
|
source, err := fm.GetEntitySource(ctx, entityID)
|
||||||
|
if err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to get entity source: %w", err)
|
||||||
|
}
|
||||||
|
defer source.Close()
|
||||||
|
|
||||||
|
// Extract text
|
||||||
|
var text string
|
||||||
|
if source.Entity().Size() > 0 {
|
||||||
|
extractor := dep.TextExtractor(ctx)
|
||||||
|
text, err = extractor.Extract(ctx, source)
|
||||||
|
if err != nil {
|
||||||
|
l.Warning("Failed to extract text for file %d: %s", fileID, err)
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
indexer := dep.SearchIndexer(ctx)
|
||||||
|
|
||||||
|
// Delete old chunks first so that stale chunks from a previously longer
|
||||||
|
// version of the file are removed before upserting the new (possibly fewer)
|
||||||
|
// chunks.
|
||||||
|
if deleteOldChunks {
|
||||||
|
if err := indexer.DeleteByFileIDs(ctx, fileID); err != nil {
|
||||||
|
l.Warning("Failed to delete old index chunks for file %d: %s", fileID, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Index via SearchIndexer
|
||||||
|
if err := indexer.IndexFile(ctx, ownerID, fileID, entityID, fileName, text); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to index file %d: %w", fileID, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Upsert metadata
|
||||||
|
if err := fm.fs.PatchMetadata(ctx, []*fs.URI{uri}, fs.MetadataPatch{
|
||||||
|
Key: dbfs.FullTextIndexKey,
|
||||||
|
Value: hashid.EncodeEntityID(fm.hasher, entityID),
|
||||||
|
}); err != nil {
|
||||||
|
return task.StatusError, fmt.Errorf("failed to patch metadata: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
l.Debug("Successfully indexed file %d for owner %d.", fileID, ownerID)
|
||||||
|
return task.StatusCompleted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// shouldIndexFullText checks if a file should be indexed for full-text search.
|
||||||
|
func (m *manager) shouldIndexFullText(ctx context.Context, fileName string, size int64) bool {
|
||||||
|
if !m.settings.FTSEnabled(ctx) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
extractor := m.dep.TextExtractor(ctx)
|
||||||
|
return util.IsInExtensionList(extractor.Exts(), fileName) && extractor.MaxFileSize() > size
|
||||||
|
}
|
||||||
|
|
||||||
|
// fullTextIndexForNewEntity creates and queues a full text index task for a newly uploaded entity.
|
||||||
|
func (m *manager) fullTextIndexForNewEntity(ctx context.Context, session *fs.UploadSession, owner int) {
|
||||||
|
if session.Props.EntityType != nil && *session.Props.EntityType != types.EntityTypeVersion {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if !m.shouldIndexFullText(ctx, session.Props.Uri.Name(), session.Props.Size) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
t, err := NewFullTextIndexTask(ctx, session.Props.Uri, session.EntityID, session.FileID, owner, m.user)
|
||||||
|
if err != nil {
|
||||||
|
m.l.Warning("Failed to create full text index task: %s", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := m.dep.MediaMetaQueue(ctx).QueueTask(ctx, t); err != nil {
|
||||||
|
m.l.Warning("Failed to queue full text index task: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *manager) processIndexDiff(ctx context.Context, diff *fs.IndexDiff) {
|
||||||
|
if diff == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, update := range diff.IndexToUpdate {
|
||||||
|
t, err := NewFullTextIndexTask(ctx, &update.Uri, update.EntityID, update.FileID, update.OwnerID, m.user)
|
||||||
|
if err != nil {
|
||||||
|
m.l.Warning("Failed to create full text update task: %s", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err := m.dep.MediaMetaQueue(ctx).QueueTask(ctx, t); err != nil {
|
||||||
|
m.l.Warning("Failed to queue full text update task: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, cp := range diff.IndexToCopy {
|
||||||
|
t, err := NewFullTextCopyTask(ctx, &cp.Uri, cp.OriginalFileID, cp.FileID, cp.OwnerID, cp.EntityID, m.user)
|
||||||
|
if err != nil {
|
||||||
|
m.l.Warning("Failed to create full text copy task: %s", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err := m.dep.MediaMetaQueue(ctx).QueueTask(ctx, t); err != nil {
|
||||||
|
m.l.Warning("Failed to queue full text copy task: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, change := range diff.IndexToChangeOwner {
|
||||||
|
t, err := NewFullTextChangeOwnerTask(ctx, &change.Uri, change.EntityID, change.FileID, change.OriginalOwnerID, change.NewOwnerID, m.user)
|
||||||
|
if err != nil {
|
||||||
|
m.l.Warning("Failed to create full text change owner task: %s", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err := m.dep.MediaMetaQueue(ctx).QueueTask(ctx, t); err != nil {
|
||||||
|
m.l.Warning("Failed to queue full text change owner task: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(diff.IndexToDelete) > 0 && m.dep.SettingProvider().FTSEnabled(ctx) {
|
||||||
|
t, err := NewFullTextDeleteTask(ctx, diff.IndexToDelete, m.user)
|
||||||
|
if err != nil {
|
||||||
|
m.l.Warning("Failed to create full text delete task: %s", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := m.dep.MediaMetaQueue(ctx).QueueTask(ctx, t); err != nil {
|
||||||
|
m.l.Warning("Failed to queue full text delete task: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx = context.WithoutCancel(ctx)
|
||||||
|
indexer := m.dep.SearchIndexer(ctx)
|
||||||
|
go func() {
|
||||||
|
for _, rename := range diff.IndexToRename {
|
||||||
|
if err := indexer.Rename(ctx, rename.FileID, rename.EntityID, rename.Uri.Name()); err != nil {
|
||||||
|
m.l.Warning("Failed to rename index for file %d: %s", rename.FileID, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
@ -0,0 +1,15 @@
|
|||||||
|
package extractor
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
)
|
||||||
|
|
||||||
|
// NoopExtractor is a no-op implementation of TextExtractor, used when text extraction is disabled.
|
||||||
|
type NoopExtractor struct{}
|
||||||
|
|
||||||
|
func (n *NoopExtractor) Exts() []string { return nil }
|
||||||
|
func (n *NoopExtractor) MaxFileSize() int64 { return 0 }
|
||||||
|
func (n *NoopExtractor) Extract(ctx context.Context, reader io.Reader) (string, error) {
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
@ -0,0 +1,82 @@
|
|||||||
|
package extractor
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/logging"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/request"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/setting"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TikaExtractor extracts text from documents using Apache Tika.
|
||||||
|
type TikaExtractor struct {
|
||||||
|
client request.Client
|
||||||
|
settings setting.Provider
|
||||||
|
l logging.Logger
|
||||||
|
exts []string
|
||||||
|
maxFileSize int64
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewTikaExtractor creates a new TikaExtractor.
|
||||||
|
func NewTikaExtractor(client request.Client, settings setting.Provider, l logging.Logger, cfg *setting.FTSTikaExtractorSetting) *TikaExtractor {
|
||||||
|
exts := cfg.Exts
|
||||||
|
return &TikaExtractor{
|
||||||
|
client: client,
|
||||||
|
settings: settings,
|
||||||
|
l: l,
|
||||||
|
exts: exts,
|
||||||
|
maxFileSize: cfg.MaxFileSize,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Exts returns the list of supported file extensions.
|
||||||
|
func (t *TikaExtractor) Exts() []string {
|
||||||
|
return t.exts
|
||||||
|
}
|
||||||
|
|
||||||
|
// MaxFileSize returns the maximum file size for text extraction.
|
||||||
|
func (t *TikaExtractor) MaxFileSize() int64 {
|
||||||
|
return t.maxFileSize
|
||||||
|
}
|
||||||
|
|
||||||
|
// Extract sends the document to Tika and returns the extracted plain text.
|
||||||
|
func (t *TikaExtractor) Extract(ctx context.Context, reader io.Reader) (string, error) {
|
||||||
|
tikaCfg := t.settings.FTSTikaExtractor(ctx)
|
||||||
|
if tikaCfg.Endpoint == "" {
|
||||||
|
return "", fmt.Errorf("tika endpoint not configured")
|
||||||
|
}
|
||||||
|
|
||||||
|
endpoint := strings.TrimRight(tikaCfg.Endpoint, "/") + "/tika"
|
||||||
|
resp := t.client.Request(
|
||||||
|
"PUT",
|
||||||
|
endpoint,
|
||||||
|
reader,
|
||||||
|
request.WithHeader(map[string][]string{
|
||||||
|
"Accept": {"text/plain"},
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
if resp.Err != nil {
|
||||||
|
return "", fmt.Errorf("tika request failed: %w", resp.Err)
|
||||||
|
}
|
||||||
|
defer resp.Response.Body.Close()
|
||||||
|
|
||||||
|
if resp.Response.StatusCode != 200 {
|
||||||
|
return "", fmt.Errorf("tika returned status %d", resp.Response.StatusCode)
|
||||||
|
}
|
||||||
|
|
||||||
|
maxSize := tikaCfg.MaxResponseSize
|
||||||
|
if maxSize <= 0 {
|
||||||
|
maxSize = 10 * 1024 * 1024 // default 10MB
|
||||||
|
}
|
||||||
|
|
||||||
|
limited := io.LimitReader(resp.Response.Body, maxSize)
|
||||||
|
body, err := io.ReadAll(limited)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("failed to read tika response: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return strings.TrimSpace(string(body)), nil
|
||||||
|
}
|
||||||
@ -0,0 +1,49 @@
|
|||||||
|
package searcher
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
)
|
||||||
|
|
||||||
|
type SearchDocument struct {
|
||||||
|
ID string `json:"id"`
|
||||||
|
FileID int `json:"file_id"`
|
||||||
|
OwnerID int `json:"owner_id"`
|
||||||
|
EntityID int `json:"entity_id"`
|
||||||
|
ChunkIdx int `json:"chunk_idx"`
|
||||||
|
FileName string `json:"file_name"`
|
||||||
|
Text string `json:"text"`
|
||||||
|
Formated *FormatedHit `json:"_formatted,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type FormatedHit struct {
|
||||||
|
Text string `json:"text"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type SearchResult struct {
|
||||||
|
FileID int `json:"file_id"`
|
||||||
|
OwnerID int `json:"owner_id"`
|
||||||
|
EntityID int `json:"entity_id"`
|
||||||
|
FileName string `json:"file_name"`
|
||||||
|
Text string `json:"text"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type SearchIndexer interface {
|
||||||
|
IndexFile(ctx context.Context, ownerID, fileID, entityID int, fileName, text string) error
|
||||||
|
DeleteByFileIDs(ctx context.Context, fileID ...int) error
|
||||||
|
ChangeOwner(ctx context.Context, fileID, oldOwnerID, newOwnerID int) error
|
||||||
|
CopyByFileID(ctx context.Context, srcFileID, dstFileID, dstOwnerID, dstEntityID int) error
|
||||||
|
Rename(ctx context.Context, fileID, entityID int, newFileName string) error
|
||||||
|
Search(ctx context.Context, ownerID int, query string, offset int) ([]SearchResult, int64, error)
|
||||||
|
// IndexReady reports whether the search index exists and has the required
|
||||||
|
// configuration (filterable/searchable attributes, etc.).
|
||||||
|
IndexReady(ctx context.Context) (bool, error)
|
||||||
|
EnsureIndex(ctx context.Context) error
|
||||||
|
Close() error
|
||||||
|
}
|
||||||
|
|
||||||
|
type TextExtractor interface {
|
||||||
|
Exts() []string
|
||||||
|
MaxFileSize() int64
|
||||||
|
Extract(ctx context.Context, reader io.Reader) (string, error)
|
||||||
|
}
|
||||||
@ -0,0 +1,101 @@
|
|||||||
|
package indexer
|
||||||
|
|
||||||
|
import "strings"
|
||||||
|
|
||||||
|
const defaultMaxBytes = 2000
|
||||||
|
|
||||||
|
// ChunkText splits text into chunks of approximately maxBytes bytes each.
|
||||||
|
// It splits on paragraph breaks (\n\n), combines small paragraphs until the
|
||||||
|
// byte limit is reached, and splits large paragraphs at word boundaries.
|
||||||
|
func ChunkText(text string, maxBytes int) []string {
|
||||||
|
if maxBytes <= 0 {
|
||||||
|
maxBytes = defaultMaxBytes
|
||||||
|
}
|
||||||
|
|
||||||
|
text = strings.TrimSpace(text)
|
||||||
|
if text == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
paragraphs := strings.Split(text, "\n\n")
|
||||||
|
var chunks []string
|
||||||
|
var current []string
|
||||||
|
currentBytes := 0
|
||||||
|
|
||||||
|
for _, para := range paragraphs {
|
||||||
|
para = strings.TrimSpace(para)
|
||||||
|
if para == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
paraBytes := len(para)
|
||||||
|
|
||||||
|
// If a single paragraph exceeds maxBytes, split it at word boundaries
|
||||||
|
if paraBytes > maxBytes {
|
||||||
|
// Flush accumulated content first
|
||||||
|
if currentBytes > 0 {
|
||||||
|
chunks = append(chunks, strings.Join(current, "\n\n"))
|
||||||
|
current = nil
|
||||||
|
currentBytes = 0
|
||||||
|
}
|
||||||
|
chunks = append(chunks, splitByBytes(para, maxBytes)...)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// If adding this paragraph (plus separator) would exceed the limit, flush
|
||||||
|
joinerLen := 0
|
||||||
|
if currentBytes > 0 {
|
||||||
|
joinerLen = 2 // "\n\n"
|
||||||
|
}
|
||||||
|
if currentBytes+joinerLen+paraBytes > maxBytes && currentBytes > 0 {
|
||||||
|
chunks = append(chunks, strings.Join(current, "\n\n"))
|
||||||
|
current = nil
|
||||||
|
currentBytes = 0
|
||||||
|
}
|
||||||
|
|
||||||
|
if currentBytes > 0 {
|
||||||
|
currentBytes += 2 // account for "\n\n" joiner
|
||||||
|
}
|
||||||
|
current = append(current, para)
|
||||||
|
currentBytes += paraBytes
|
||||||
|
}
|
||||||
|
|
||||||
|
// Flush remaining
|
||||||
|
if currentBytes > 0 {
|
||||||
|
chunks = append(chunks, strings.Join(current, "\n\n"))
|
||||||
|
}
|
||||||
|
|
||||||
|
return chunks
|
||||||
|
}
|
||||||
|
|
||||||
|
// splitByBytes splits text into chunks at word boundaries, each at most maxBytes bytes.
|
||||||
|
func splitByBytes(text string, maxBytes int) []string {
|
||||||
|
words := strings.Fields(text)
|
||||||
|
var chunks []string
|
||||||
|
var current []string
|
||||||
|
currentBytes := 0
|
||||||
|
|
||||||
|
for _, w := range words {
|
||||||
|
wLen := len(w)
|
||||||
|
spaceLen := 0
|
||||||
|
if currentBytes > 0 {
|
||||||
|
spaceLen = 1
|
||||||
|
}
|
||||||
|
|
||||||
|
if currentBytes+spaceLen+wLen > maxBytes && currentBytes > 0 {
|
||||||
|
chunks = append(chunks, strings.Join(current, " "))
|
||||||
|
current = nil
|
||||||
|
currentBytes = 0
|
||||||
|
spaceLen = 0
|
||||||
|
}
|
||||||
|
|
||||||
|
current = append(current, w)
|
||||||
|
currentBytes += spaceLen + wLen
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(current) > 0 {
|
||||||
|
chunks = append(chunks, strings.Join(current, " "))
|
||||||
|
}
|
||||||
|
|
||||||
|
return chunks
|
||||||
|
}
|
||||||
@ -0,0 +1,84 @@
|
|||||||
|
package indexer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestChunkText_Empty(t *testing.T) {
|
||||||
|
assert.Nil(t, ChunkText("", 500))
|
||||||
|
assert.Nil(t, ChunkText(" ", 500))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_SingleSmallParagraph(t *testing.T) {
|
||||||
|
chunks := ChunkText("Hello world", 500)
|
||||||
|
assert.Equal(t, []string{"Hello world"}, chunks)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_MultipleParagraphsCombined(t *testing.T) {
|
||||||
|
text := "First paragraph.\n\nSecond paragraph.\n\nThird paragraph."
|
||||||
|
chunks := ChunkText(text, 500)
|
||||||
|
assert.Len(t, chunks, 1)
|
||||||
|
assert.Contains(t, chunks[0], "First paragraph.")
|
||||||
|
assert.Contains(t, chunks[0], "Third paragraph.")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_SplitOnParagraphBoundary(t *testing.T) {
|
||||||
|
// Create two paragraphs each ~300 bytes
|
||||||
|
para := strings.Repeat("abcde ", 50) // 300 bytes
|
||||||
|
para = strings.TrimSpace(para)
|
||||||
|
text := para + "\n\n" + para
|
||||||
|
|
||||||
|
chunks := ChunkText(text, 500)
|
||||||
|
assert.Len(t, chunks, 2)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_LargeParagraphSplit(t *testing.T) {
|
||||||
|
// Create a single paragraph of 1200 bytes (240 words * 5 bytes each)
|
||||||
|
words := make([]string, 240)
|
||||||
|
for i := range words {
|
||||||
|
words[i] = "word" // 4 bytes + 1 space = 5 per word
|
||||||
|
}
|
||||||
|
text := strings.Join(words, " ") // 240*4 + 239 = 1199 bytes
|
||||||
|
|
||||||
|
chunks := ChunkText(text, 500)
|
||||||
|
assert.Len(t, chunks, 3)
|
||||||
|
assert.LessOrEqual(t, len(chunks[0]), 500)
|
||||||
|
assert.LessOrEqual(t, len(chunks[1]), 500)
|
||||||
|
assert.Greater(t, len(chunks[2]), 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_DefaultMaxBytes(t *testing.T) {
|
||||||
|
chunks := ChunkText("hello", 0)
|
||||||
|
assert.Equal(t, []string{"hello"}, chunks)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_EmptyParagraphsIgnored(t *testing.T) {
|
||||||
|
text := "First.\n\n\n\n\n\nSecond."
|
||||||
|
chunks := ChunkText(text, 500)
|
||||||
|
assert.Len(t, chunks, 1)
|
||||||
|
assert.Equal(t, "First.\n\nSecond.", chunks[0])
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_ParagraphKeptWhole(t *testing.T) {
|
||||||
|
// A paragraph under the limit should not be split
|
||||||
|
para := strings.Repeat("x", 400)
|
||||||
|
chunks := ChunkText(para, 500)
|
||||||
|
assert.Len(t, chunks, 1)
|
||||||
|
assert.Equal(t, para, chunks[0])
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChunkText_JoinerAccountedInLimit(t *testing.T) {
|
||||||
|
// Two paragraphs that fit individually but exceed the limit when joined with "\n\n"
|
||||||
|
p1 := strings.Repeat("a", 250)
|
||||||
|
p2 := strings.Repeat("b", 250)
|
||||||
|
text := p1 + "\n\n" + p2
|
||||||
|
|
||||||
|
chunks := ChunkText(text, 500)
|
||||||
|
// 250 + 2 + 250 = 502 > 500, so they should be split
|
||||||
|
assert.Len(t, chunks, 2)
|
||||||
|
assert.Equal(t, p1, chunks[0])
|
||||||
|
assert.Equal(t, p2, chunks[1])
|
||||||
|
}
|
||||||
@ -0,0 +1,384 @@
|
|||||||
|
package indexer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"slices"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/logging"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/searcher"
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/setting"
|
||||||
|
"github.com/meilisearch/meilisearch-go"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
indexName = "cloudreve_files"
|
||||||
|
embedderName = "cr-text"
|
||||||
|
)
|
||||||
|
|
||||||
|
// MeilisearchIndexer implements SearchIndexer using Meilisearch.
|
||||||
|
type MeilisearchIndexer struct {
|
||||||
|
client meilisearch.ServiceManager
|
||||||
|
l logging.Logger
|
||||||
|
pageSize int
|
||||||
|
chunkSize int
|
||||||
|
cfg *setting.FTSIndexMeilisearchSetting
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewMeilisearchIndexer creates a new MeilisearchIndexer.
|
||||||
|
func NewMeilisearchIndexer(msCfg *setting.FTSIndexMeilisearchSetting, chunkSize int, l logging.Logger) *MeilisearchIndexer {
|
||||||
|
client := meilisearch.New(msCfg.Endpoint, meilisearch.WithAPIKey(msCfg.APIKey))
|
||||||
|
return &MeilisearchIndexer{
|
||||||
|
client: client,
|
||||||
|
l: l,
|
||||||
|
pageSize: msCfg.PageSize,
|
||||||
|
chunkSize: chunkSize,
|
||||||
|
cfg: msCfg,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var (
|
||||||
|
requiredFilterable = []string{"owner_id", "file_id", "entity_id"}
|
||||||
|
requiredSearchable = []string{"text", "file_name"}
|
||||||
|
requiredDistinct = "file_id"
|
||||||
|
)
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) IndexReady(ctx context.Context) (bool, error) {
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
|
||||||
|
settings, err := index.GetSettingsWithContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
// If the index doesn't exist, Meilisearch returns an error.
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check filterable attributes.
|
||||||
|
for _, attr := range requiredFilterable {
|
||||||
|
if !slices.Contains(settings.FilterableAttributes, attr) {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check searchable attributes.
|
||||||
|
for _, attr := range requiredSearchable {
|
||||||
|
if !slices.Contains(settings.SearchableAttributes, attr) {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check distinct attribute.
|
||||||
|
if settings.DistinctAttribute == nil || *settings.DistinctAttribute != requiredDistinct {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check embedder if embedding is enabled.
|
||||||
|
if m.cfg.EmbeddingEnbaled {
|
||||||
|
if settings.Embedders == nil {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
if _, ok := settings.Embedders[embedderName]; !ok {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) EnsureIndex(ctx context.Context) error {
|
||||||
|
_, err := m.client.CreateIndexWithContext(ctx, &meilisearch.IndexConfig{
|
||||||
|
Uid: indexName,
|
||||||
|
PrimaryKey: "id",
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
m.l.Debug("Create index returned (may already exist): %s", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
|
||||||
|
filterableAttrs := []any{"owner_id", "file_id", "entity_id"}
|
||||||
|
if _, err := index.UpdateFilterableAttributesWithContext(ctx, &filterableAttrs); err != nil {
|
||||||
|
return fmt.Errorf("failed to set filterable attributes: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
searchableAttrs := []string{"text", "file_name"}
|
||||||
|
if _, err := index.UpdateSearchableAttributesWithContext(ctx, &searchableAttrs); err != nil {
|
||||||
|
return fmt.Errorf("failed to set searchable attributes: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = index.UpdateDistinctAttributeWithContext(ctx, "file_id")
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to set distinct attribute: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if m.cfg.EmbeddingEnbaled {
|
||||||
|
var embedder meilisearch.Embedder
|
||||||
|
if err := json.Unmarshal([]byte(m.cfg.EmbeddingSetting), &embedder); err != nil {
|
||||||
|
m.cfg.EmbeddingEnbaled = false
|
||||||
|
m.l.Warning("Failed to unmarshal embedding setting: %s, fallback to disable embedding", err)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := index.UpdateEmbeddersWithContext(ctx, map[string]meilisearch.Embedder{
|
||||||
|
embedderName: embedder,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to set embedders: %w", err)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
_, err := index.ResetEmbeddersWithContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
m.l.Warning("Failed to reset embedder: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) IndexFile(ctx context.Context, ownerID, fileID, entityID int, fileName, text string) error {
|
||||||
|
chunks := ChunkText(text, m.chunkSize)
|
||||||
|
if len(chunks) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
docs := make([]searcher.SearchDocument, 0, len(chunks))
|
||||||
|
for i, chunk := range chunks {
|
||||||
|
docs = append(docs, searcher.SearchDocument{
|
||||||
|
ID: fmt.Sprintf("%d_%d", fileID, i),
|
||||||
|
FileID: fileID,
|
||||||
|
OwnerID: ownerID,
|
||||||
|
EntityID: entityID,
|
||||||
|
ChunkIdx: i,
|
||||||
|
FileName: fileName,
|
||||||
|
Text: chunk,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
pk := "id"
|
||||||
|
if _, err := index.AddDocumentsWithContext(ctx, docs, &meilisearch.DocumentOptions{PrimaryKey: &pk}); err != nil {
|
||||||
|
return fmt.Errorf("failed to add documents: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) DeleteByFileIDs(ctx context.Context, fileID ...int) error {
|
||||||
|
if len(fileID) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
strs := make([]string, len(fileID))
|
||||||
|
for i, id := range fileID {
|
||||||
|
strs[i] = fmt.Sprintf("%d", id)
|
||||||
|
}
|
||||||
|
filter := fmt.Sprintf("file_id IN [%s]", strings.Join(strs, ", "))
|
||||||
|
if _, err := index.DeleteDocumentsByFilterWithContext(ctx, filter, nil); err != nil {
|
||||||
|
return fmt.Errorf("failed to delete documents by file_ids: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) ChangeOwner(ctx context.Context, fileID, oldOwnerID, newOwnerID int) error {
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
filter := fmt.Sprintf("file_id = %d AND owner_id = %d", fileID, oldOwnerID)
|
||||||
|
|
||||||
|
// Fetch all existing document chunks in batches.
|
||||||
|
const batchSize int64 = 100
|
||||||
|
var allDocs []searcher.SearchDocument
|
||||||
|
for offset := int64(0); ; offset += batchSize {
|
||||||
|
var result meilisearch.DocumentsResult
|
||||||
|
if err := index.GetDocumentsWithContext(ctx, &meilisearch.DocumentsQuery{
|
||||||
|
Filter: filter,
|
||||||
|
Limit: batchSize,
|
||||||
|
Offset: offset,
|
||||||
|
}, &result); err != nil {
|
||||||
|
return fmt.Errorf("failed to get documents: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, hit := range result.Results {
|
||||||
|
var doc searcher.SearchDocument
|
||||||
|
if err := hit.DecodeInto(&doc); err != nil {
|
||||||
|
m.l.Warning("Failed to decode document during owner change: %s", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
allDocs = append(allDocs, doc)
|
||||||
|
}
|
||||||
|
|
||||||
|
if int64(len(result.Results)) < batchSize {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(allDocs) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Update owner_id in place — primary key is {fileID}_{chunkIdx} so it stays the same.
|
||||||
|
for i := range allDocs {
|
||||||
|
allDocs[i].OwnerID = newOwnerID
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := index.UpdateDocumentsInBatchesWithContext(ctx, allDocs, 100, nil); err != nil {
|
||||||
|
return fmt.Errorf("failed to update documents with new owner: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) CopyByFileID(ctx context.Context, srcFileID, dstFileID, dstOwnerID, dstEntityID int) error {
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
filter := fmt.Sprintf("file_id = %d", srcFileID)
|
||||||
|
|
||||||
|
const batchSize int64 = 100
|
||||||
|
var allDocs []searcher.SearchDocument
|
||||||
|
for offset := int64(0); ; offset += batchSize {
|
||||||
|
var result meilisearch.DocumentsResult
|
||||||
|
if err := index.GetDocumentsWithContext(ctx, &meilisearch.DocumentsQuery{
|
||||||
|
Filter: filter,
|
||||||
|
Limit: batchSize,
|
||||||
|
Offset: offset,
|
||||||
|
}, &result); err != nil {
|
||||||
|
return fmt.Errorf("failed to get source documents: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, hit := range result.Results {
|
||||||
|
var doc searcher.SearchDocument
|
||||||
|
if err := hit.DecodeInto(&doc); err != nil {
|
||||||
|
m.l.Warning("Failed to decode document during copy: %s", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
allDocs = append(allDocs, doc)
|
||||||
|
}
|
||||||
|
|
||||||
|
if int64(len(result.Results)) < batchSize {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(allDocs) == 0 {
|
||||||
|
return fmt.Errorf("no source documents found for file %d", srcFileID)
|
||||||
|
}
|
||||||
|
|
||||||
|
for i := range allDocs {
|
||||||
|
if allDocs[i].EntityID != dstEntityID {
|
||||||
|
m.l.Warning("Entity id mismatch for file %d, original: %d, destination: %d", srcFileID, allDocs[i].EntityID, dstEntityID)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
allDocs[i].ID = fmt.Sprintf("%d_%d", dstFileID, allDocs[i].ChunkIdx)
|
||||||
|
allDocs[i].FileID = dstFileID
|
||||||
|
allDocs[i].OwnerID = dstOwnerID
|
||||||
|
allDocs[i].EntityID = dstEntityID
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(allDocs) == 0 {
|
||||||
|
return fmt.Errorf("no source documents found for file %d", srcFileID)
|
||||||
|
}
|
||||||
|
|
||||||
|
pk := "id"
|
||||||
|
if _, err := index.AddDocumentsWithContext(ctx, allDocs, &meilisearch.DocumentOptions{PrimaryKey: &pk}); err != nil {
|
||||||
|
return fmt.Errorf("failed to add copied documents: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) Rename(ctx context.Context, fileID, entityID int, newFileName string) error {
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
filter := fmt.Sprintf("file_id = %d AND entity_id = %d", fileID, entityID)
|
||||||
|
|
||||||
|
const batchSize int64 = 100
|
||||||
|
var allDocs []searcher.SearchDocument
|
||||||
|
for offset := int64(0); ; offset += batchSize {
|
||||||
|
var result meilisearch.DocumentsResult
|
||||||
|
if err := index.GetDocumentsWithContext(ctx, &meilisearch.DocumentsQuery{
|
||||||
|
Filter: filter,
|
||||||
|
Limit: batchSize,
|
||||||
|
Offset: offset,
|
||||||
|
}, &result); err != nil {
|
||||||
|
return fmt.Errorf("failed to get documents for rename: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, hit := range result.Results {
|
||||||
|
var doc searcher.SearchDocument
|
||||||
|
if err := hit.DecodeInto(&doc); err != nil {
|
||||||
|
m.l.Warning("Failed to decode document during rename: %s", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
doc.FileName = newFileName
|
||||||
|
allDocs = append(allDocs, doc)
|
||||||
|
}
|
||||||
|
|
||||||
|
if int64(len(result.Results)) < batchSize {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(allDocs) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := index.UpdateDocumentsInBatchesWithContext(ctx, allDocs, 100, nil); err != nil {
|
||||||
|
return fmt.Errorf("failed to update documents with new file name: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) Search(ctx context.Context, ownerID int, query string, offset int) ([]searcher.SearchResult, int64, error) {
|
||||||
|
index := m.client.Index(indexName)
|
||||||
|
|
||||||
|
searchReq := &meilisearch.SearchRequest{
|
||||||
|
Filter: fmt.Sprintf("owner_id = %d", ownerID),
|
||||||
|
Limit: int64(m.pageSize),
|
||||||
|
Offset: int64(offset),
|
||||||
|
AttributesToHighlight: []string{"text"},
|
||||||
|
}
|
||||||
|
|
||||||
|
if m.cfg.EmbeddingEnbaled {
|
||||||
|
searchReq.Hybrid = &meilisearch.SearchRequestHybrid{
|
||||||
|
Embedder: embedderName,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := index.SearchWithContext(ctx, query, searchReq)
|
||||||
|
if err != nil {
|
||||||
|
return nil, 0, fmt.Errorf("search failed: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
results := make([]searcher.SearchResult, 0, len(resp.Hits))
|
||||||
|
seen := make(map[int]struct{})
|
||||||
|
for _, hit := range resp.Hits {
|
||||||
|
var doc searcher.SearchDocument
|
||||||
|
if err := hit.DecodeInto(&doc); err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, exists := seen[doc.FileID]; exists {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seen[doc.FileID] = struct{}{}
|
||||||
|
|
||||||
|
// Extract text from raw JSON for display
|
||||||
|
textStr := doc.Text
|
||||||
|
if doc.Formated != nil {
|
||||||
|
textStr = doc.Formated.Text
|
||||||
|
}
|
||||||
|
|
||||||
|
results = append(results, searcher.SearchResult{
|
||||||
|
FileID: doc.FileID,
|
||||||
|
OwnerID: doc.OwnerID,
|
||||||
|
FileName: doc.FileName,
|
||||||
|
Text: textStr,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
return results, resp.EstimatedTotalHits, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *MeilisearchIndexer) Close() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
@ -0,0 +1,46 @@
|
|||||||
|
package indexer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
|
||||||
|
"github.com/cloudreve/Cloudreve/v4/pkg/searcher"
|
||||||
|
)
|
||||||
|
|
||||||
|
// NoopIndexer is a no-op implementation of SearchIndexer, used when FTS is disabled.
|
||||||
|
type NoopIndexer struct{}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) IndexFile(ctx context.Context, ownerID, fileID, entityID int, fileName, text string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) DeleteByFileIDs(ctx context.Context, fileID ...int) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) ChangeOwner(ctx context.Context, fileID, oldOwnerID, newOwnerID int) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) CopyByFileID(ctx context.Context, srcFileID, dstFileID, dstOwnerID, dstEntityID int) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) Rename(ctx context.Context, fileID, entityID int, newFileName string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) Search(ctx context.Context, ownerID int, query string, offset int) ([]searcher.SearchResult, int64, error) {
|
||||||
|
return nil, 0, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) IndexReady(ctx context.Context) (bool, error) {
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) EnsureIndex(ctx context.Context) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *NoopIndexer) Close() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
Loading…
Reference in new issue