source/folder.go - main - amne/dca-agent

source/folder.go
227 строк5 KB
СеребристыйМак
feat: scoring fixes, filters, period analytics, weekly-report endpoint, monitor
13 июл 2026, 22:59
13 июл 2026, 22:59bbb6a6f
package source

import (
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/fsnotify/fsnotify"
	"dca-agent/config"
	"dca-agent/storage"
)

// FolderWatcher monitors a directory for new audio files
type FolderWatcher struct {
	watcher   *fsnotify.Watcher
	cfg       config.FolderConfig
	sinceTime time.Time // files modified before this are ignored
	store     *storage.Store
	onFile    func(filename string) // callback when new file detected
	done      chan struct{}
}

func NewFolderWatcher(cfg config.FolderConfig, store *storage.Store, sinceTime time.Time) (*FolderWatcher, error) {
	w, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, fmt.Errorf("create watcher: %w", err)
	}

	// Watch the folder
	if err := w.Add(cfg.Path); err != nil {
		w.Close()
		return nil, fmt.Errorf("watch folder %s: %w", cfg.Path, err)
	}

	fw := &FolderWatcher{
		watcher:   w,
		cfg:       cfg,
		sinceTime: sinceTime,
		store:     store,
		done:      make(chan struct{}),
	}

	return fw, nil
}

// OnFile sets the callback for when a new file is detected
func (fw *FolderWatcher) OnFile(cb func(filename string)) {
	fw.onFile = cb
}

// Start begins watching for new files
func (fw *FolderWatcher) Start() {
	go fw.loop()
	log.Printf("[watcher] watching %s", fw.cfg.Path)
}

// Stop stops the watcher
func (fw *FolderWatcher) Stop() {
	close(fw.done)
	fw.watcher.Close()
}

func (fw *FolderWatcher) loop() {
	// Debounce map: filename → last event time
	debounce := map[string]time.Time{}
	// Periodic cleanup ticker — purges stale debounce entries every 5 minutes
	cleanup := time.NewTicker(5 * time.Minute)
	defer cleanup.Stop()

	for {
		select {
		case <-fw.done:
			return
		case <-cleanup.C:
			// Remove entries older than 10 seconds (debounce window is 2s)
			now := time.Now()
			for k, v := range debounce {
				if now.Sub(v) > 10*time.Second {
					delete(debounce, k)
				}
			}
		case event, ok := <-fw.watcher.Events:
			if !ok {
				return
			}
			if event.Op&fsnotify.Create != fsnotify.Create && event.Op&fsnotify.Write != fsnotify.Write {
				continue
			}

			filename := filepath.Base(event.Name)

			// Check extension
			if !fw.isAudioFile(filename) {
				continue
			}

			// Debounce: ignore events within 2 seconds for same file
			now := time.Now()
			if last, ok := debounce[filename]; ok && now.Sub(last) < 2*time.Second {
				continue
			}
			debounce[filename] = now

			// Wait for file to be fully written (size stable)
			if !fw.waitForStableFile(event.Name) {
				log.Printf("[watcher] file not stable, skipping: %s", filename)
				continue
			}

			// Check if already processed
			exists, err := fw.store.FilenameExists(filename)
			if err != nil {
				log.Printf("[watcher] db error: %v", err)
				continue
			}
			if exists {
				continue
			}

			log.Printf("[watcher] new file: %s", filename)

			// Since filter: skip files modified before cutoff
			if !fw.sinceTime.IsZero() {
				info, err := os.Stat(event.Name)
				if err == nil && info.ModTime().Before(fw.sinceTime) {
					log.Printf("[watcher] skipping old file: %s (mod=%s, since=%s)", filename, info.ModTime().Format("2006-01-02"), fw.sinceTime.Format("2006-01-02"))
					continue
				}
			}

			if fw.onFile != nil {
				fw.onFile(event.Name)
			}

		case err, ok := <-fw.watcher.Errors:
			if !ok {
				return
			}
			log.Printf("[watcher] error: %v", err)
		}
	}
}

func (fw *FolderWatcher) isAudioFile(filename string) bool {
	ext := strings.ToLower(filepath.Ext(filename))
	for _, allowed := range fw.cfg.Extensions {
		if strings.EqualFold(ext, allowed) {
			return true
		}
	}
	return false
}

// waitForStableFile waits until file size stops changing (file fully written)
func (fw *FolderWatcher) waitForStableFile(path string) bool {
	var prevSize int64 = -1
	for i := 0; i < 10; i++ {
		info, err := os.Stat(path)
		if err != nil {
			time.Sleep(500 * time.Millisecond)
			continue
		}
		if info.Size() == prevSize && info.Size() > 0 {
			return true
		}
		prevSize = info.Size()
		time.Sleep(500 * time.Millisecond)
	}
	return prevSize > 0
}

// ScanExisting scans the folder for existing files and enqueues them
func (fw *FolderWatcher) ScanExisting() error {
	entries, err := os.ReadDir(fw.cfg.Path)
	if err != nil {
		return fmt.Errorf("read dir: %w", err)
	}

	count := 0
	skipped := 0
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}
		if !fw.isAudioFile(entry.Name()) {
			continue
		}

		// Since filter: skip files modified before cutoff
		if !fw.sinceTime.IsZero() {
			info, err := entry.Info()
			if err != nil {
				continue
			}
			if info.ModTime().Before(fw.sinceTime) {
				skipped++
				continue
			}
		}

		exists, err := fw.store.FilenameExists(entry.Name())
		if err != nil {
			log.Printf("[watcher] db error for %s: %v", entry.Name(), err)
			continue
		}
		if exists {
			continue
		}

		fullPath := filepath.Join(fw.cfg.Path, entry.Name())
		if fw.onFile != nil {
			fw.onFile(fullPath)
			count++
		}
	}

	if skipped > 0 {
		log.Printf("[watcher] skipped %d files older than %s", skipped, fw.sinceTime.Format("2006-01-02"))
	}
	if count > 0 {
		log.Printf("[watcher] enqueued %d existing files", count)
	}
	return nil
}