mirror of https://gitee.com/godoos/godoos.git
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
435 lines
11 KiB
435 lines
11 KiB
package vector
|
|
|
|
import (
|
|
"fmt"
|
|
"godo/libs"
|
|
"log"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/fsnotify/fsnotify"
|
|
)
|
|
|
|
var (
|
|
MapFilePathMonitors = map[string]uint{}
|
|
watcher *fsnotify.Watcher
|
|
fileQueue = make(chan string, 100) // 队列大小可以根据需要调整
|
|
numWorkers = 3 // 工作协程的数量
|
|
wg sync.WaitGroup
|
|
syncingKnowledgeIds = make(map[uint]syncingStats) // 记录正在同步的 knowledgeId 及其同步状态
|
|
syncMutex sync.Mutex // 保护 syncingKnowledgeIds 的互斥锁
|
|
renameMap = make(map[string]string) // 临时映射存储 Remove 事件的路径
|
|
renameMutex sync.Mutex // 保护 renameMap 的互斥锁
|
|
watcherMutex sync.Mutex // 保护 watcher 的互斥锁
|
|
)
|
|
|
|
type syncingStats struct {
|
|
totalFiles int
|
|
processedFiles int
|
|
}
|
|
|
|
func InitMonitor() {
|
|
var err error
|
|
watcherMutex.Lock()
|
|
watcher, err = fsnotify.NewWatcher()
|
|
if err != nil {
|
|
log.Fatalf("Error creating watcher: %s", err.Error())
|
|
}
|
|
watcherMutex.Unlock()
|
|
go FolderMonitor()
|
|
go startWatching()
|
|
|
|
// 启动 worker
|
|
for i := 0; i < numWorkers; i++ {
|
|
wg.Add(1)
|
|
go worker()
|
|
}
|
|
}
|
|
|
|
func startWatching() {
|
|
for {
|
|
select {
|
|
case event, ok := <-watcher.Events:
|
|
if !ok {
|
|
log.Println("error event")
|
|
return
|
|
}
|
|
filePath := filepath.Clean(event.Name)
|
|
result, exists := shouldProcess(filePath)
|
|
if result > 0 {
|
|
if event.Has(fsnotify.Write) || event.Has(fsnotify.Create) {
|
|
log.Printf("Event: %v, File: %s", event.Op, filePath)
|
|
if isFileComplete(filePath) {
|
|
// 将文件路径放入队列
|
|
fileQueue <- filePath
|
|
}
|
|
}
|
|
if event.Has(fsnotify.Create) {
|
|
if info, err := os.Stat(filePath); err == nil && info.IsDir() {
|
|
addRecursive(filePath, watcher)
|
|
}
|
|
// 检查是否是重命名事件
|
|
handleRenameCreateEvent(event)
|
|
}
|
|
if event.Has(fsnotify.Remove) {
|
|
//log.Printf("Event: %v, File: %s,exists:%d", event.Op, filePath, exists)
|
|
isDir := true
|
|
newFileName := fmt.Sprintf(".godoos.%d.%s.json", result, filepath.Base(filePath))
|
|
newFilePath := filepath.Join(filepath.Dir(filePath), newFileName)
|
|
if libs.PathExists(newFilePath) {
|
|
isDir = false
|
|
}
|
|
if isDir {
|
|
watcherMutex.Lock()
|
|
if watcher != nil {
|
|
watcher.Remove(filePath)
|
|
}
|
|
watcherMutex.Unlock()
|
|
}
|
|
if exists == 1 {
|
|
err := DeleteVector(result)
|
|
if err != nil {
|
|
log.Printf("Error deleting vector %d: %v", result, err)
|
|
}
|
|
}
|
|
if exists == 2 && !isDir {
|
|
err := DeleteVectorFile(result, filePath)
|
|
if err != nil {
|
|
log.Printf("Error deleting vector file %d: %v", result, err)
|
|
}
|
|
}
|
|
//存储 Remove 事件的路径
|
|
handleRenameRemoveEvent(event)
|
|
}
|
|
}
|
|
case err, ok := <-watcher.Errors:
|
|
if !ok {
|
|
return
|
|
}
|
|
log.Println("error:", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func handleRenameRemoveEvent(event fsnotify.Event) {
|
|
renameMutex.Lock()
|
|
defer renameMutex.Unlock()
|
|
//log.Printf("handleRenameRemoveEvent: %v, File: %s", event.Op, event.Name)
|
|
renameMap[event.Name] = event.Name
|
|
}
|
|
|
|
func handleRenameCreateEvent(event fsnotify.Event) {
|
|
renameMutex.Lock()
|
|
defer renameMutex.Unlock()
|
|
//log.Printf("handleRenameCreateEvent: %v, File: %s", event.Op, event.Name)
|
|
// 规范化路径
|
|
newPath := filepath.Clean(event.Name)
|
|
|
|
// 检查是否是重命名事件
|
|
for oldPath := range renameMap {
|
|
if oldPath != "" {
|
|
// 找到对应的 Remove 事件
|
|
oldPathClean := filepath.Clean(oldPath)
|
|
if oldPathClean == newPath {
|
|
//log.Printf("File renamed from %s to %s", oldPath, newPath)
|
|
|
|
// 更新 MapFilePathMonitors
|
|
for path, id := range MapFilePathMonitors {
|
|
if path == oldPathClean {
|
|
delete(MapFilePathMonitors, path)
|
|
MapFilePathMonitors[newPath] = id
|
|
log.Printf("Updated MapFilePathMonitors: %s -> %s", oldPathClean, newPath)
|
|
break
|
|
}
|
|
}
|
|
|
|
// 更新 watcher
|
|
watcherMutex.Lock()
|
|
if watcher != nil {
|
|
if err := watcher.Remove(oldPathClean); err != nil {
|
|
log.Printf("Error removing old path %s from watcher: %v", oldPathClean, err)
|
|
}
|
|
if err := watcher.Add(newPath); err != nil {
|
|
log.Printf("Error adding new path %s to watcher: %v", newPath, err)
|
|
}
|
|
}
|
|
watcherMutex.Unlock()
|
|
|
|
// 如果是目录,递归更新子目录
|
|
if info, err := os.Stat(newPath); err == nil && info.IsDir() {
|
|
addRecursive(newPath, watcher)
|
|
}
|
|
|
|
// 清除临时映射中的路径
|
|
delete(renameMap, oldPath)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func worker() {
|
|
defer wg.Done()
|
|
for filePath := range fileQueue {
|
|
knowledgeId, exists := shouldProcess(filePath)
|
|
if exists == 0 {
|
|
log.Printf("File path %s is not being monitored", filePath)
|
|
continue
|
|
}
|
|
|
|
// 更新已处理文件数
|
|
syncMutex.Lock()
|
|
if stats, ok := syncingKnowledgeIds[knowledgeId]; ok {
|
|
stats.processedFiles++
|
|
syncingKnowledgeIds[knowledgeId] = stats
|
|
}
|
|
syncMutex.Unlock()
|
|
|
|
err := handleGodoosFile(filePath, knowledgeId)
|
|
if err != nil {
|
|
log.Printf("Error handling file %s: %v", filePath, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func FolderMonitor() {
|
|
basePath, err := libs.GetOsDir()
|
|
if err != nil {
|
|
log.Printf("Error getting base path: %s", err.Error())
|
|
return
|
|
}
|
|
|
|
// 递归添加所有子目录
|
|
addRecursive(basePath, watcher)
|
|
|
|
// Add a path.
|
|
watcherMutex.Lock()
|
|
if watcher != nil {
|
|
err = watcher.Add(basePath)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
watcherMutex.Unlock()
|
|
|
|
// Block main goroutine forever.
|
|
<-make(chan struct{})
|
|
}
|
|
|
|
func AddWatchFolder(folderPath string, knowledgeId uint, callback func()) error {
|
|
if watcher == nil {
|
|
InitMonitor()
|
|
}
|
|
// 规范化路径
|
|
folderPath = filepath.Clean(folderPath)
|
|
|
|
// 检查文件夹是否存在
|
|
if !libs.PathExists(folderPath) {
|
|
return fmt.Errorf("folder path does not exist: %s", folderPath)
|
|
}
|
|
|
|
// 检查文件夹是否已经存在于监视器中
|
|
if _, exists := MapFilePathMonitors[folderPath]; exists {
|
|
return fmt.Errorf("folder path is already being monitored: %s", folderPath)
|
|
}
|
|
|
|
// 递归添加所有子目录
|
|
addRecursive(folderPath, watcher)
|
|
|
|
// 计算总文件数
|
|
totalFiles, err := countFiles(folderPath)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to count files in folder path: %w", err)
|
|
}
|
|
|
|
// 更新 syncingKnowledgeIds
|
|
syncMutex.Lock()
|
|
syncingKnowledgeIds[knowledgeId] = syncingStats{
|
|
totalFiles: totalFiles,
|
|
processedFiles: 0,
|
|
}
|
|
syncMutex.Unlock()
|
|
|
|
// 更新 MapFilePathMonitors
|
|
MapFilePathMonitors[folderPath] = knowledgeId
|
|
|
|
// 添加文件夹路径到监视器
|
|
err = watcher.Add(folderPath)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to add folder path to watcher: %w", err)
|
|
}
|
|
|
|
// 调用回调函数
|
|
if callback != nil {
|
|
callback()
|
|
}
|
|
|
|
log.Printf("Added folder path %s to watcher with knowledgeId %d", folderPath, knowledgeId)
|
|
return nil
|
|
}
|
|
|
|
// RemoveWatchFolder 根据路径删除观察文件夹
|
|
func RemoveWatchFolder(folderPath string) error {
|
|
// 规范化路径
|
|
folderPath = filepath.Clean(folderPath)
|
|
|
|
// 检查文件夹是否存在于监视器中
|
|
knowledgeId, exists := MapFilePathMonitors[folderPath]
|
|
if !exists {
|
|
return fmt.Errorf("folder path is not being monitored: %s", folderPath)
|
|
}
|
|
|
|
// 从 watcher 中移除路径
|
|
watcherMutex.Lock()
|
|
if watcher != nil {
|
|
err := watcher.Remove(folderPath)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to remove folder path from watcher: %w", err)
|
|
}
|
|
}
|
|
watcherMutex.Unlock()
|
|
|
|
// 递归移除所有子目录
|
|
err := filepath.Walk(folderPath, func(path string, info os.FileInfo, err error) error {
|
|
if err != nil {
|
|
log.Printf("Error walking path %s: %v", path, err)
|
|
return err
|
|
}
|
|
if info.IsDir() {
|
|
result, _ := shouldProcess(path)
|
|
if result > 0 {
|
|
// 从 watcher 中移除路径
|
|
watcherMutex.Lock()
|
|
if watcher != nil {
|
|
err := watcher.Remove(path)
|
|
if err != nil {
|
|
log.Printf("Error removing path %s from watcher: %v", path, err)
|
|
return err
|
|
}
|
|
}
|
|
watcherMutex.Unlock()
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("failed to remove folder path from watcher: %w", err)
|
|
}
|
|
|
|
// 从 MapFilePathMonitors 中删除条目
|
|
delete(MapFilePathMonitors, folderPath)
|
|
|
|
// 从 syncingKnowledgeIds 中删除条目
|
|
syncMutex.Lock()
|
|
delete(syncingKnowledgeIds, knowledgeId)
|
|
syncMutex.Unlock()
|
|
|
|
log.Printf("Removed folder path %s from watcher with knowledgeId %d", folderPath, knowledgeId)
|
|
return nil
|
|
}
|
|
|
|
func shouldProcess(filePath string) (uint, int) {
|
|
// 规范化路径
|
|
filePath = filepath.Clean(filePath)
|
|
|
|
// 检查文件路径是否在 MapFilePathMonitors 中
|
|
for path, id := range MapFilePathMonitors {
|
|
if id < 1 {
|
|
return 0, 0
|
|
}
|
|
path = filepath.Clean(path)
|
|
if filePath == path {
|
|
return id, 1 // 完全相等
|
|
}
|
|
if strings.HasPrefix(filePath, path+string(filepath.Separator)) {
|
|
return id, 2 // 包含
|
|
}
|
|
}
|
|
return 0, 0 // 不存在
|
|
}
|
|
|
|
func addRecursive(path string, watcher *fsnotify.Watcher) {
|
|
err := filepath.Walk(path, func(path string, info os.FileInfo, err error) error {
|
|
if err != nil {
|
|
log.Printf("Error walking path %s: %v", path, err)
|
|
return err
|
|
}
|
|
if info.IsDir() {
|
|
result, _ := shouldProcess(path)
|
|
if result > 0 {
|
|
if err := watcher.Add(path); err != nil {
|
|
log.Printf("Error adding path %s to watcher: %v", path, err)
|
|
return err
|
|
}
|
|
log.Printf("Added path %s to watcher", path)
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
log.Printf("Error adding recursive paths: %v", err)
|
|
}
|
|
}
|
|
|
|
// countFiles 递归计算文件夹中的文件数
|
|
func countFiles(folderPath string) (int, error) {
|
|
var count int
|
|
err := filepath.Walk(folderPath, func(path string, info os.FileInfo, err error) error {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !info.IsDir() {
|
|
count++
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// GetSyncPercentage 计算并返回同步百分比
|
|
func GetSyncPercentage(knowledgeId uint) float64 {
|
|
syncMutex.Lock()
|
|
defer syncMutex.Unlock()
|
|
if stats, ok := syncingKnowledgeIds[knowledgeId]; ok {
|
|
if stats.totalFiles == 0 {
|
|
return 0.0
|
|
}
|
|
return float64(stats.processedFiles) / float64(stats.totalFiles) * 100
|
|
}
|
|
return 0.0
|
|
}
|
|
|
|
// isFileComplete 检查文件是否已经完全创建
|
|
func isFileComplete(filePath string) bool {
|
|
// 等待一段时间确保文件已经完全创建
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
// 检查文件是否存在
|
|
if _, err := os.Stat(filePath); err != nil {
|
|
log.Printf("File %s does not exist: %v", filePath, err)
|
|
return false
|
|
}
|
|
|
|
// 检查文件大小是否达到预期
|
|
fileInfo, err := os.Stat(filePath)
|
|
if err != nil {
|
|
log.Printf("Error stat file %s: %v", filePath, err)
|
|
return false
|
|
}
|
|
// 例如,检查文件大小是否大于某个阈值
|
|
if fileInfo.Size() == 0 {
|
|
log.Printf("File %s is empty", filePath)
|
|
return false
|
|
}
|
|
if fileInfo.IsDir() {
|
|
log.Printf("File %s is a directory", filePath)
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|