612 lines
18 KiB
Go
612 lines
18 KiB
Go
package library
|
|
|
|
import (
|
|
"context"
|
|
"encoding/binary"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"io/fs"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/cespare/xxhash/v2"
|
|
"github.com/google/uuid"
|
|
|
|
"pipu/api/internal/library/meta"
|
|
)
|
|
|
|
// UnknownAlbum — альбом файла без тегов в корне источника.
|
|
const UnknownAlbum = "Unknown album"
|
|
|
|
// MissingRetention — сколько пропавший файл хранится, прежде чем трек удалится.
|
|
const MissingRetention = 30 * 24 * time.Hour
|
|
|
|
// Progress — состояние скана одного источника (для админки).
|
|
type Progress struct {
|
|
Phase string
|
|
WalkedDirs int
|
|
CurrentDir string
|
|
SourceID uuid.UUID
|
|
Running bool
|
|
Full bool
|
|
StartedAt time.Time
|
|
FinishedAt *time.Time
|
|
Found int // аудиофайлов найдено обходом
|
|
ToProcess int // требуют чтения
|
|
Processed int
|
|
Added int
|
|
Updated int
|
|
Moved int
|
|
Missing int
|
|
Errors int
|
|
Error string // фатальная ошибка скана
|
|
}
|
|
|
|
// ProgressBoard хранит прогресс сканов в памяти процесса.
|
|
type ProgressBoard struct {
|
|
mu sync.Mutex
|
|
m map[uuid.UUID]*Progress
|
|
}
|
|
|
|
// NewProgressBoard создаёт пустую доску.
|
|
func NewProgressBoard() *ProgressBoard { return &ProgressBoard{m: map[uuid.UUID]*Progress{}} }
|
|
|
|
// Get — копия прогресса источника.
|
|
func (b *ProgressBoard) Get(id uuid.UUID) (Progress, bool) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
p, ok := b.m[id]
|
|
if !ok {
|
|
return Progress{}, false
|
|
}
|
|
return *p, true
|
|
}
|
|
|
|
// All — копии прогресса всех источников.
|
|
func (b *ProgressBoard) All() []Progress {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
out := make([]Progress, 0, len(b.m))
|
|
for _, p := range b.m {
|
|
out = append(out, *p)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// start регистрирует новый скан; false — скан этого источника уже идёт.
|
|
func (b *ProgressBoard) start(id uuid.UUID, full bool, now time.Time) bool {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if p, ok := b.m[id]; ok && p.Running {
|
|
return false
|
|
}
|
|
b.m[id] = &Progress{SourceID: id, Running: true, Full: full, StartedAt: now, Phase: "walking"}
|
|
return true
|
|
}
|
|
|
|
func (b *ProgressBoard) update(id uuid.UUID, fn func(*Progress)) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if p, ok := b.m[id]; ok {
|
|
fn(p)
|
|
}
|
|
}
|
|
|
|
// Scanner сканирует источники и обновляет каталог.
|
|
type Scanner struct {
|
|
store *Store
|
|
prober Prober
|
|
log *slog.Logger
|
|
board *ProgressBoard
|
|
workers int
|
|
now func() time.Time
|
|
// AfterScan вызывается после каждого успешного скана (переиндексация поиска).
|
|
AfterScan func(ctx context.Context)
|
|
}
|
|
|
|
// NewScanner создаёт сканер с 8 воркерами чтения.
|
|
func NewScanner(store *Store, prober Prober, board *ProgressBoard, log *slog.Logger) *Scanner {
|
|
return &Scanner{store: store, prober: prober, log: log, board: board, workers: 8, now: func() time.Time { return time.Now().UTC() }}
|
|
}
|
|
|
|
// walked — аудиофайл, найденный обходом.
|
|
type walked struct {
|
|
rel string
|
|
size int64
|
|
mtime time.Time
|
|
hash []byte
|
|
}
|
|
|
|
func hasPresent(files map[string]FileRow) bool {
|
|
for _, f := range files {
|
|
if !f.Missing {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// job — файл, который нужно прочитать.
|
|
type job struct {
|
|
walked
|
|
fileID uuid.UUID // Nil — новый
|
|
oldTrack *uuid.UUID // трек, к которому файл был привязан
|
|
}
|
|
|
|
type probed struct {
|
|
job
|
|
probe Probe
|
|
err error
|
|
}
|
|
|
|
// ErrScanRunning — скан источника уже идёт.
|
|
var ErrScanRunning = errors.New("library: scan already running")
|
|
|
|
// ErrSourceEmpty — в каталоге источника нет ни одного файла, а в библиотеке они есть: скорее всего, диск
|
|
// не смонтирован. Скан прерывается, чтобы не пометить всю библиотеку отсутствующей (и через MissingRetention —
|
|
// не удалить треки вместе с историей).
|
|
var ErrSourceEmpty = errors.New("library: source folder is empty but the library has its files — is the disk mounted?")
|
|
|
|
// Start запускает скан в фоне. root — каталог источника в файловой системе api.
|
|
func (sc *Scanner) Start(src Source, root string, full bool) error {
|
|
if !sc.board.start(src.ID, full, sc.now()) {
|
|
return ErrScanRunning
|
|
}
|
|
go func() {
|
|
ctx := context.Background()
|
|
err := sc.run(ctx, src, root, full)
|
|
finished := sc.now()
|
|
sc.board.update(src.ID, func(p *Progress) {
|
|
p.Running, p.FinishedAt = false, &finished
|
|
if err != nil {
|
|
p.Error = err.Error()
|
|
}
|
|
})
|
|
if err != nil {
|
|
sc.log.Error("scan failed", "source_id", src.ID, "err", err)
|
|
return
|
|
}
|
|
if sc.AfterScan != nil {
|
|
sc.AfterScan(ctx)
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
func (sc *Scanner) run(ctx context.Context, src Source, root string, full bool) error {
|
|
start := sc.now()
|
|
rows, err := sc.store.Patterns(ctx, src.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var exprs []string
|
|
for _, r := range rows {
|
|
if r.IsEnabled {
|
|
exprs = append(exprs, r.Regex)
|
|
}
|
|
}
|
|
patterns, err := meta.CompilePatterns(exprs)
|
|
if err != nil {
|
|
return fmt.Errorf("library: patterns: %w", err)
|
|
}
|
|
|
|
existing, err := sc.store.Files(ctx, src.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Excluded files keep their catalog rows and are not treated as missing or rename candidates.
|
|
for rel := range existing {
|
|
if pathExcluded(rel, src.ExcludedDirs) {
|
|
delete(existing, rel)
|
|
}
|
|
}
|
|
found, err := walk(root, src.ExcludedDirs, func(rel string) bool { return meta.MatchPath(patterns, rel).Skip }, func(found, dirs int, currentDir string) {
|
|
sc.board.update(src.ID, func(p *Progress) {
|
|
p.Found, p.WalkedDirs, p.CurrentDir = found, dirs, currentDir
|
|
})
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(found) == 0 && hasPresent(existing) {
|
|
return ErrSourceEmpty
|
|
}
|
|
sc.board.update(src.ID, func(p *Progress) { p.Found, p.Phase = len(found), "preparing" })
|
|
|
|
var (
|
|
jobs []job
|
|
newFiles []walked
|
|
seen = make(map[string]bool, len(found))
|
|
)
|
|
for _, w := range found {
|
|
seen[w.rel] = true
|
|
f, ok := existing[w.rel]
|
|
switch {
|
|
case !ok:
|
|
newFiles = append(newFiles, w)
|
|
case !full && f.Size == w.size && f.Mtime.Equal(w.mtime):
|
|
if f.Missing {
|
|
if err := sc.store.SetPresent(ctx, f.ID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
default:
|
|
jobs = append(jobs, job{walked: w, fileID: f.ID, oldTrack: f.TrackID})
|
|
}
|
|
}
|
|
var gone []FileRow
|
|
for rel, f := range existing {
|
|
if !seen[rel] {
|
|
gone = append(gone, f)
|
|
}
|
|
}
|
|
|
|
// Переименования: новый путь с тем же содержимым, что у пропавшего файла.
|
|
if len(gone) > 0 {
|
|
for i := range newFiles {
|
|
if h, err := contentHash(filepath.Join(root, filepath.FromSlash(newFiles[i].rel)), newFiles[i].size); err == nil {
|
|
newFiles[i].hash = h
|
|
}
|
|
}
|
|
}
|
|
moves, rest, stillGone := matchRenames(newFiles, gone)
|
|
for _, mv := range moves {
|
|
if err := sc.store.MoveFile(ctx, mv.from.ID, mv.to.rel, mv.to.size, mv.to.mtime); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for _, w := range rest {
|
|
jobs = append(jobs, job{walked: w})
|
|
}
|
|
sc.board.update(src.ID, func(p *Progress) { p.ToProcess, p.Moved, p.Phase = len(jobs), len(moves), "processing" })
|
|
|
|
if err := sc.process(ctx, src, root, patterns, jobs); err != nil {
|
|
return err
|
|
}
|
|
sc.board.update(src.ID, func(p *Progress) { p.Phase = "finalizing" })
|
|
|
|
var goneIDs []uuid.UUID
|
|
affected := map[uuid.UUID]bool{}
|
|
for _, f := range stillGone {
|
|
goneIDs = append(goneIDs, f.ID)
|
|
if f.TrackID != nil {
|
|
affected[*f.TrackID] = true
|
|
}
|
|
}
|
|
if err := sc.store.MarkMissing(ctx, goneIDs, start); err != nil {
|
|
return err
|
|
}
|
|
if _, err := sc.store.PurgeMissing(ctx, src.ID, start.Add(-MissingRetention)); err != nil {
|
|
return err
|
|
}
|
|
for id := range affected {
|
|
if err := sc.store.RefreshPrimaryFile(ctx, id); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
sc.board.update(src.ID, func(p *Progress) { p.Missing = len(goneIDs) })
|
|
if err := sc.store.CleanupOrphans(ctx); err != nil {
|
|
return err
|
|
}
|
|
// Folder images can appear without changes to the audio files. A regular scan
|
|
// retries missing covers without forcing tags to be read again.
|
|
if err := sc.store.ResetCoverChecks(ctx, src.ID); err != nil {
|
|
return err
|
|
}
|
|
return sc.store.SetLastScan(ctx, src.ID, sc.now())
|
|
}
|
|
|
|
// ScanFile — точечный скан одного файла (scan_path): после импорта из очереди трек сразу появляется в каталоге.
|
|
// Возвращает id записи media_files. Хуки AfterScan (обложки, поиск, отпечатки) вызывает сам.
|
|
func (sc *Scanner) ScanFile(ctx context.Context, src Source, root, rel string) (uuid.UUID, error) {
|
|
if pathExcluded(filepath.ToSlash(rel), src.ExcludedDirs) {
|
|
return uuid.Nil, fmt.Errorf("library: file is in an excluded scan folder: %s", rel)
|
|
}
|
|
rows, err := sc.store.Patterns(ctx, src.ID)
|
|
if err != nil {
|
|
return uuid.Nil, err
|
|
}
|
|
var exprs []string
|
|
for _, r := range rows {
|
|
if r.IsEnabled {
|
|
exprs = append(exprs, r.Regex)
|
|
}
|
|
}
|
|
patterns, err := meta.CompilePatterns(exprs)
|
|
if err != nil {
|
|
return uuid.Nil, fmt.Errorf("library: patterns: %w", err)
|
|
}
|
|
info, err := os.Stat(filepath.Join(root, filepath.FromSlash(rel)))
|
|
if err != nil {
|
|
return uuid.Nil, fmt.Errorf("library: scan file: %w", err)
|
|
}
|
|
j := job{walked: walked{rel: rel, size: info.Size(), mtime: info.ModTime().UTC().Truncate(time.Microsecond)}}
|
|
if f, ok, err := sc.store.FileByPath(ctx, src.ID, rel); err != nil {
|
|
return uuid.Nil, err
|
|
} else if ok {
|
|
j.fileID, j.oldTrack = f.ID, f.TrackID
|
|
}
|
|
if err := sc.process(ctx, src, root, patterns, []job{j}); err != nil {
|
|
return uuid.Nil, err
|
|
}
|
|
f, ok, err := sc.store.FileByPath(ctx, src.ID, rel)
|
|
if err != nil {
|
|
return uuid.Nil, err
|
|
}
|
|
if !ok {
|
|
return uuid.Nil, fmt.Errorf("library: scanned file %s not saved", rel)
|
|
}
|
|
if sc.AfterScan != nil {
|
|
go sc.AfterScan(context.WithoutCancel(ctx))
|
|
}
|
|
return f.ID, nil
|
|
}
|
|
|
|
// process читает файлы параллельно, а каталог обновляет один писатель —
|
|
// так дубли и новые исполнители не создаются дважды в гонке.
|
|
func (sc *Scanner) process(ctx context.Context, src Source, root string, patterns []meta.Pattern, jobs []job) error {
|
|
in := make(chan job)
|
|
out := make(chan probed)
|
|
var wg sync.WaitGroup
|
|
for range max(1, sc.workers) {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for j := range in {
|
|
abs := filepath.Join(root, filepath.FromSlash(j.rel))
|
|
p, err := sc.prober.Probe(ctx, abs)
|
|
if j.hash == nil {
|
|
j.hash, _ = contentHash(abs, j.size)
|
|
}
|
|
out <- probed{job: j, probe: p, err: err}
|
|
}
|
|
}()
|
|
}
|
|
go func() {
|
|
for _, j := range jobs {
|
|
in <- j
|
|
}
|
|
close(in)
|
|
wg.Wait()
|
|
close(out)
|
|
}()
|
|
|
|
var firstErr error
|
|
for r := range out {
|
|
if firstErr != nil {
|
|
continue // дочитываем канал, чтобы воркеры завершились
|
|
}
|
|
added, err := sc.apply(ctx, src, patterns, r)
|
|
if err != nil {
|
|
firstErr = err
|
|
continue
|
|
}
|
|
sc.board.update(src.ID, func(p *Progress) {
|
|
p.Processed++
|
|
switch {
|
|
case r.err != nil:
|
|
p.Errors++
|
|
case added:
|
|
p.Added++
|
|
default:
|
|
p.Updated++
|
|
}
|
|
})
|
|
}
|
|
return firstErr
|
|
}
|
|
|
|
// apply записывает прочитанный файл в каталог. added — файл новый.
|
|
func (sc *Scanner) apply(ctx context.Context, src Source, patterns []meta.Pattern, r probed) (bool, error) {
|
|
data := FileData{RelPath: r.rel, Size: r.size, Mtime: r.mtime, ContentHash: r.hash, Probe: r.probe}
|
|
added := r.fileID == uuid.Nil
|
|
if r.err != nil {
|
|
data.Probe = Probe{Format: strings.TrimPrefix(strings.ToLower(filepath.Ext(r.rel)), ".")}
|
|
data.ScanError = r.err.Error()
|
|
if _, err := sc.store.SaveFile(ctx, src.ID, r.fileID, data, nil); err != nil {
|
|
return false, err
|
|
}
|
|
if r.oldTrack != nil {
|
|
return added, sc.store.RefreshPrimaryFile(ctx, *r.oldTrack)
|
|
}
|
|
return added, nil
|
|
}
|
|
|
|
pm := meta.MatchPath(patterns, r.rel)
|
|
m := meta.Merge(r.probe.Tags, meta.FromPath(pm.Fields), src.TagsFirst, r.rel)
|
|
if m.Album == "" {
|
|
m.Album = UnknownAlbum
|
|
}
|
|
known := func(k string) bool { return sc.store.ArtistKeyExists(ctx, k) }
|
|
// Альбом принадлежит одному исполнителю: у составного «A, B» это первый основной.
|
|
albumArtistName := m.AlbumArtist
|
|
if credits := meta.SplitArtists(m.AlbumArtist, known); len(credits) > 0 {
|
|
albumArtistName = credits[0].Name
|
|
}
|
|
albumArtist, err := sc.store.EnsureArtist(ctx, albumArtistName)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
albumID, err := sc.store.EnsureAlbum(ctx, albumArtist, m.Album, m.Year)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
in := TrackInput{
|
|
AlbumID: albumID, Title: m.Title, Disc: m.Disc, Track: m.Track, Year: m.Year, DurationMs: r.probe.DurationMs,
|
|
}
|
|
trackID, dup, err := sc.store.FindTrack(ctx, in)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
// Тот же файл, перечитанный в тот же трек, заново разбирает исполнителей: правила разбора меняются.
|
|
sameTrack := dup && r.oldTrack != nil && *r.oldTrack == trackID
|
|
if !dup || sameTrack {
|
|
// Перечитанный файл сохраняет свой трек (лайки, история); новый — получает новый.
|
|
if r.oldTrack != nil {
|
|
trackID = *r.oldTrack
|
|
}
|
|
if trackID, err = sc.store.SaveTrack(ctx, trackID, in); err != nil {
|
|
return false, err
|
|
}
|
|
credits := meta.SplitArtists(m.Artist, known)
|
|
ids := make([]uuid.UUID, 0, len(credits))
|
|
roles := make([]string, 0, len(credits))
|
|
for _, c := range credits {
|
|
id, err := sc.store.EnsureArtist(ctx, c.Name)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
ids, roles = append(ids, id), append(roles, c.Role)
|
|
}
|
|
if err := sc.store.SetTrackArtists(ctx, trackID, ids, roles); err != nil {
|
|
return false, err
|
|
}
|
|
if err := sc.store.SetTrackGenres(ctx, trackID, m.Genres); err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
if _, err := sc.store.SaveFile(ctx, src.ID, r.fileID, data, &trackID); err != nil {
|
|
return false, err
|
|
}
|
|
if err := sc.store.RefreshPrimaryFile(ctx, trackID); err != nil {
|
|
return false, err
|
|
}
|
|
if r.oldTrack != nil && *r.oldTrack != trackID {
|
|
if err := sc.store.RefreshPrimaryFile(ctx, *r.oldTrack); err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
return added, nil
|
|
}
|
|
|
|
// walk обходит дерево источника и возвращает аудиофайлы. Скрытые файлы и каталоги пропускаются;
|
|
// skip — пути, которые шаблоны велят игнорировать. Ошибки отдельных каталогов не прерывают обход.
|
|
func walk(root string, excludedDirs []string, skip func(rel string) bool, report func(found, dirs int, currentDir string)) ([]walked, error) {
|
|
if _, err := os.Stat(root); err != nil {
|
|
return nil, fmt.Errorf("library: source root: %w", err)
|
|
}
|
|
var out []walked
|
|
dirs := 0
|
|
err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
|
|
if err != nil {
|
|
if p == root {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
name := d.Name()
|
|
if p != root && strings.HasPrefix(name, ".") {
|
|
if d.IsDir() {
|
|
return filepath.SkipDir
|
|
}
|
|
return nil
|
|
}
|
|
rel, err := filepath.Rel(root, p)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
rel = filepath.ToSlash(rel)
|
|
if pathExcluded(rel, excludedDirs) {
|
|
if d.IsDir() {
|
|
return filepath.SkipDir
|
|
}
|
|
return nil
|
|
}
|
|
if d.IsDir() {
|
|
dirs++
|
|
if report != nil {
|
|
report(len(out), dirs, rel)
|
|
}
|
|
return nil
|
|
}
|
|
if !IsAudio(name) {
|
|
return nil
|
|
}
|
|
if skip(rel) {
|
|
return nil
|
|
}
|
|
info, err := d.Info()
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
// PostgreSQL хранит время с точностью до микросекунд.
|
|
out = append(out, walked{rel: rel, size: info.Size(), mtime: info.ModTime().UTC().Truncate(time.Microsecond)})
|
|
if report != nil {
|
|
report(len(out), dirs, filepath.ToSlash(filepath.Dir(rel)))
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("library: walk: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
const hashChunk = 1 << 20
|
|
|
|
// contentHash — xxhash64 от размера, первого и последнего мегабайта: быстро даже по NFS
|
|
// и не меняется при переименовании.
|
|
func contentHash(abs string, size int64) ([]byte, error) {
|
|
f, err := os.Open(abs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = f.Close() }()
|
|
h := xxhash.New()
|
|
var sz [8]byte
|
|
binary.LittleEndian.PutUint64(sz[:], uint64(size))
|
|
_, _ = h.Write(sz[:])
|
|
if _, err := io.CopyN(h, f, hashChunk); err != nil && !errors.Is(err, io.EOF) {
|
|
return nil, err
|
|
}
|
|
if size > 2*hashChunk {
|
|
if _, err := f.Seek(-hashChunk, io.SeekEnd); err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err := io.CopyN(h, f, hashChunk); err != nil && !errors.Is(err, io.EOF) {
|
|
return nil, err
|
|
}
|
|
}
|
|
return h.Sum(nil), nil
|
|
}
|
|
|
|
type move struct {
|
|
from FileRow
|
|
to walked
|
|
}
|
|
|
|
// matchRenames сопоставляет новые пути с пропавшими файлами по содержимому (хэш и размер).
|
|
// Каждый пропавший файл используется не больше одного раза.
|
|
func matchRenames(newFiles []walked, gone []FileRow) (moves []move, rest []walked, stillGone []FileRow) {
|
|
byHash := map[string][]int{}
|
|
for i, f := range gone {
|
|
if len(f.ContentHash) > 0 {
|
|
k := fmt.Sprintf("%x:%d", f.ContentHash, f.Size)
|
|
byHash[k] = append(byHash[k], i)
|
|
}
|
|
}
|
|
used := make([]bool, len(gone))
|
|
for _, w := range newFiles {
|
|
k := fmt.Sprintf("%x:%d", w.hash, w.size)
|
|
if len(w.hash) > 0 && len(byHash[k]) > 0 {
|
|
i := byHash[k][0]
|
|
byHash[k] = byHash[k][1:]
|
|
used[i] = true
|
|
moves = append(moves, move{from: gone[i], to: w})
|
|
continue
|
|
}
|
|
rest = append(rest, w)
|
|
}
|
|
for i, f := range gone {
|
|
if !used[i] {
|
|
stillGone = append(stillGone, f)
|
|
}
|
|
}
|
|
return moves, rest, stillGone
|
|
}
|