diff --git a/docs/data-model.md b/docs/data-model.md index 03a69a2..df81858 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -109,14 +109,14 @@ CREATE TABLE media_files ( track_id uuid, -- FK ставится после создания tracks missing_since timestamptz, -- файл пропал при скане scan_error text, + fingerprinted_at timestamptz, -- Chromaprint посчитан (или fpcalc не смог); NULL — в очереди UNIQUE (source_id, rel_path) ); CREATE INDEX ON media_files (content_hash); CREATE INDEX ON media_files (sha256); CREATE TABLE track_fingerprints ( -- Chromaprint, для поиска дублей при загрузке - file_id uuid PRIMARY KEY REFERENCES media_files(id) ON DELETE CASCADE, - track_id uuid, + file_id uuid PRIMARY KEY REFERENCES media_files(id) ON DELETE CASCADE, -- трек — через media_files duration_ms int NOT NULL, fingerprint int[] NOT NULL, -- raw-отпечаток fpcalc fp_prefix bigint NOT NULL -- первые 64 бита — грубый индекс для отбора кандидатов diff --git a/docs/library-and-storage.md b/docs/library-and-storage.md index 69cd4af..d156cb6 100644 --- a/docs/library-and-storage.md +++ b/docs/library-and-storage.md @@ -104,7 +104,7 @@ lower-case, `ё → е`, удаление пунктуации, схлопыва | `scan_source(full=true)` | Перечитать теги / шаблоны у всех файлов | | `scan_path(source_id, rel_path)` | Точечный скан одного файла после импорта из очереди | | `extract_cover(album_id)` | Новый альбом | -| `fingerprint_file(file_id)` | Новый / изменённый файл: полный SHA-256 + Chromaprint (`fpcalc`) — для проверки дублей при загрузке; низкий приоритет | +| `fingerprint_file(file_id)` | Новый / изменённый файл: полный SHA-256 + Chromaprint (`fpcalc -raw`, первые 120 с) — для проверки дублей при загрузке; низкий приоритет. Реализация — `library.Fingerprinter`: один проход по файлам с `sha256 IS NULL` / `fingerprinted_at IS NULL` после каждого скана и при старте, по одному файлу. Без `fpcalc` (`FPCALC_PATH`) считается только SHA-256. Изменение содержимого файла сбрасывает отпечатки. Сравнение — `library.Similarity` (доля совпавших бит лучшего выравнивания, сдвиг ±80 точек; ≥ 0.75 — тот же трек) | | `reindex_search` | Вручную / после крупного скана | Алгоритм инкрементального скана: diff --git a/services/api/Dockerfile b/services/api/Dockerfile index 69e9fe0..c5cc5d7 100644 --- a/services/api/Dockerfile +++ b/services/api/Dockerfile @@ -14,9 +14,9 @@ RUN --mount=type=cache,target=/go/pkg/mod --mount=type=cache,target=/root/.cache go build -trimpath -ldflags "-s -w -X pipu/api/internal/buildinfo.Version=${VERSION}" -o /out/api ./cmd/api && \ go build -trimpath -ldflags "-s -w -X pipu/api/internal/buildinfo.Version=${VERSION}" -o /out/storage-agent ./cmd/storage-agent -# api вызывает ffprobe (сканер) и ffmpeg (транскодинг, M3) — нужен не distroless, а alpine с ffmpeg. +# api вызывает ffprobe (сканер), ffmpeg (транскодинг, M3) и fpcalc (отпечатки, M5.5) — нужен не distroless, а alpine. FROM alpine:3.22 AS api -RUN apk add --no-cache ffmpeg ca-certificates tzdata +RUN apk add --no-cache ffmpeg chromaprint ca-certificates tzdata COPY --from=build /out/api /usr/local/bin/api USER 65532:65532 EXPOSE 8080 diff --git a/services/api/cmd/api/main.go b/services/api/cmd/api/main.go index 4777958..2929c9c 100644 --- a/services/api/cmd/api/main.go +++ b/services/api/cmd/api/main.go @@ -106,11 +106,21 @@ func run() error { log.Warn("STREAM_SIGNING_KEY is not set: cover and stream links will break on restart") } covers := library.NewCovers(libStore, libSvc, cfg.FFmpegPath, filepath.Join(cfg.CacheDir, "covers"), log) + fingerprints := library.NewFingerprinter(libStore, library.FPCalc{Bin: cfg.FPCalcPath, Timeout: time.Minute}, + func(ctx context.Context, sourceID uuid.UUID) (string, error) { + src, err := libStore.GetSource(ctx, sourceID) + if err != nil { + return "", err + } + return libSvc.Root(src) + }, log) scanner.AfterScan = func(ctx context.Context) { covers.Run(ctx) indexer.Trigger() + fingerprints.Trigger(ctx) } go covers.Run(ctx) // альбомы, оставшиеся без обложки с прошлого запуска + fingerprints.Trigger(ctx) concurrency := cfg.TranscodeConcurrency if concurrency <= 0 { diff --git a/services/api/internal/library/fingerprint.go b/services/api/internal/library/fingerprint.go new file mode 100644 index 0000000..f0505b1 --- /dev/null +++ b/services/api/internal/library/fingerprint.go @@ -0,0 +1,251 @@ +package library + +import ( + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "math/bits" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/google/uuid" +) + +// Отпечатки файлов библиотеки (docs/uploads.md#4-проверка-на-дубли): полный SHA-256 ловит точную копию, +// Chromaprint — тот же трек в другом формате / битрейте. Считаются фоном после скана, по одному файлу: +// задача низкого приоритета (~0.3 с на трек), чтобы не мешать стриму. + +// FingerprintSeconds — сколько начала трека берёт fpcalc (его значение по умолчанию). +const FingerprintSeconds = 120 + +// Fingerprint — raw-отпечаток Chromaprint. +type Fingerprint struct { + DurationMs int + Points []uint32 +} + +// Prefix — первые 64 бита отпечатка: грубый индекс для отбора кандидатов. +func (f Fingerprint) Prefix() int64 { + var p uint64 + if len(f.Points) > 0 { + p = uint64(f.Points[0]) << 32 + } + if len(f.Points) > 1 { + p |= uint64(f.Points[1]) + } + return int64(p) +} + +// Similarity — доля совпадающих бит лучшего выравнивания двух отпечатков (0..1). Сдвиг до maxShift точек +// (≈ 0.124 с на точку) прощает тишину в начале и разную длину вступления у разных изданий. +func Similarity(a, b []uint32) float64 { + const maxShift = 80 + best := 0.0 + for shift := -maxShift; shift <= maxShift; shift++ { + var n, diff int + for i := range a { + j := i + shift + if j < 0 || j >= len(b) { + continue + } + n++ + diff += bits.OnesCount32(a[i] ^ b[j]) + } + // Слишком короткое перекрытие (меньше ~10 с) не считается. + if n < 80 { + continue + } + if s := 1 - float64(diff)/float64(32*n); s > best { + best = s + } + } + return best +} + +// SameRecording — порог «тот же трек»: у разных записей совпадает ~50% бит (случайные), у перекодирования — 85%+. +const SameRecording = 0.75 + +// FPCalc — Chromaprint через внешний fpcalc. Путь передаётся отдельным аргументом, без shell. +type FPCalc struct { + Bin string + Timeout time.Duration +} + +// ErrNoFPCalc — fpcalc не установлен: отпечатки Chromaprint не считаются. +var ErrNoFPCalc = errors.New("library: fpcalc not found") + +// Fingerprint считает raw-отпечаток файла. +func (f FPCalc) Fingerprint(ctx context.Context, absPath string) (Fingerprint, error) { + bin := f.Bin + if bin == "" { + bin = "fpcalc" + } + if f.Timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, f.Timeout) + defer cancel() + } + out, err := exec.CommandContext(ctx, bin, "-raw", "-json", "-length", fmt.Sprint(FingerprintSeconds), "--", absPath).Output() + if err != nil { + if errors.Is(err, exec.ErrNotFound) { + return Fingerprint{}, ErrNoFPCalc + } + if ee, ok := err.(*exec.ExitError); ok && len(ee.Stderr) > 0 { + return Fingerprint{}, fmt.Errorf("fpcalc: %s", strings.TrimSpace(string(ee.Stderr))) + } + return Fingerprint{}, fmt.Errorf("fpcalc: %w", err) + } + return parseFPCalc(out) +} + +func parseFPCalc(raw []byte) (Fingerprint, error) { + var o struct { + Duration float64 `json:"duration"` + Fingerprint []uint32 `json:"fingerprint"` + } + if err := json.Unmarshal(raw, &o); err != nil { + return Fingerprint{}, fmt.Errorf("fpcalc: decode: %w", err) + } + if len(o.Fingerprint) == 0 { + return Fingerprint{}, errors.New("fpcalc: empty fingerprint") + } + return Fingerprint{DurationMs: int(o.Duration * 1000), Points: o.Fingerprint}, nil +} + +// FileSHA256 — полный SHA-256 файла. +func FileSHA256(abs string) ([]byte, error) { + f, err := os.Open(abs) + if err != nil { + return nil, err + } + defer func() { _ = f.Close() }() + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return nil, err + } + return h.Sum(nil), nil +} + +// Fingerprinter считает SHA-256 и Chromaprint для файлов библиотеки без отпечатков. +// Без fpcalc работает в режиме «только SHA-256», а файлы остаются в очереди на Chromaprint до его установки. +type Fingerprinter struct { + store *Store + calc FPCalc + rootOf func(ctx context.Context, sourceID uuid.UUID) (string, error) + log *slog.Logger + warned bool + + mu sync.Mutex + running bool + again bool +} + +// NewFingerprinter создаёт обработчик; rootOf — каталог источника в файловой системе api. +func NewFingerprinter(store *Store, calc FPCalc, rootOf func(ctx context.Context, sourceID uuid.UUID) (string, error), + log *slog.Logger) *Fingerprinter { + return &Fingerprinter{store: store, calc: calc, rootOf: rootOf, log: log} +} + +// Trigger запускает проход в фоне; если проход уже идёт — после него будет ещё один (новые файлы скана). +func (fp *Fingerprinter) Trigger(ctx context.Context) { + fp.mu.Lock() + defer fp.mu.Unlock() + if fp.running { + fp.again = true + return + } + fp.running = true + go func() { + for { + fp.Run(ctx) + fp.mu.Lock() + if !fp.again || ctx.Err() != nil { + fp.running = false + fp.mu.Unlock() + return + } + fp.again = false + fp.mu.Unlock() + } + }() +} + +func (fp *Fingerprinter) chromaprint() bool { + bin := fp.calc.Bin + if bin == "" { + bin = "fpcalc" + } + if _, err := exec.LookPath(bin); err != nil { + if !fp.warned { + fp.log.Warn("fpcalc is not installed: only exact (SHA-256) duplicates will be detected") + fp.warned = true + } + return false + } + return true +} + +// Run — один проход по файлам без отпечатков (смонтированные источники, курсор по id). Файл с ошибкой +// чтения в этом проходе пропускается; следующий скан или перезапуск попробует снова. +func (fp *Fingerprinter) Run(ctx context.Context) { + withFP := fp.chromaprint() + roots := map[uuid.UUID]string{} + var after uuid.UUID + for ctx.Err() == nil { + todo, err := fp.store.FilesToFingerprint(ctx, withFP, after, 200) + if err != nil { + fp.log.Error("fingerprints: list files", "err", err) + return + } + if len(todo) == 0 { + return + } + for _, f := range todo { + if ctx.Err() != nil { + return + } + after = f.ID + root, ok := roots[f.SourceID] + if !ok { + if root, err = fp.rootOf(ctx, f.SourceID); err != nil { + fp.log.Warn("fingerprints: source root", "source_id", f.SourceID, "err", err) + continue + } + roots[f.SourceID] = root + } + fp.process(ctx, f, filepath.Join(root, filepath.FromSlash(f.RelPath)), withFP) + } + } +} + +func (fp *Fingerprinter) process(ctx context.Context, f FingerprintTodo, abs string, withFP bool) { + sum := f.SHA256 + if sum == nil { + var err error + if sum, err = FileSHA256(abs); err != nil { + fp.log.Warn("fingerprints: sha256", "file_id", f.ID, "err", err) + return + } + } + var print *Fingerprint + if withFP { + res, err := fp.calc.Fingerprint(ctx, abs) + if err != nil { + // Битый файл: отмечаем обработанным без отпечатка, чтобы не пересчитывать каждый раз. + fp.log.Warn("fingerprints: fpcalc", "file_id", f.ID, "err", err) + } else { + print = &res + } + } + if err := fp.store.SaveFingerprint(ctx, f.ID, sum, print, withFP); err != nil { + fp.log.Error("fingerprints: save", "file_id", f.ID, "err", err) + } +} diff --git a/services/api/internal/library/fingerprint_test.go b/services/api/internal/library/fingerprint_test.go new file mode 100644 index 0000000..8bad357 --- /dev/null +++ b/services/api/internal/library/fingerprint_test.go @@ -0,0 +1,58 @@ +package library + +import ( + "math/rand/v2" + "testing" +) + +func randomPrint(r *rand.Rand, n int) []uint32 { + out := make([]uint32, n) + for i := range out { + out[i] = r.Uint32() + } + return out +} + +func TestSimilarity(t *testing.T) { + r := rand.New(rand.NewPCG(1, 2)) + a := randomPrint(r, 950) // ~2 мин + if s := Similarity(a, a); s != 1 { + t.Fatalf("same print: %v", s) + } + + // Перекодирование: другое вступление (сдвиг на 20 точек) и ~8% перевёрнутых бит. + b := append(randomPrint(r, 20), a...) + for i := range b { + for bit := range 32 { + if r.IntN(100) < 8 { + b[i] ^= 1 << bit + } + } + } + if s := Similarity(a, b); s < SameRecording { + t.Fatalf("re-encoded copy: %v < %v", s, SameRecording) + } + + if s := Similarity(a, randomPrint(r, 950)); s >= SameRecording { + t.Fatalf("different recordings: %v >= %v", s, SameRecording) + } + if s := Similarity(a, a[:40]); s != 0 { + t.Fatalf("too short overlap must not match: %v", s) + } +} + +func TestParseFPCalc(t *testing.T) { + fp, err := parseFPCalc([]byte(`{"duration": 215.4, "fingerprint": [4294967295, 1, 2]}`)) + if err != nil { + t.Fatal(err) + } + if fp.DurationMs != 215400 || len(fp.Points) != 3 || fp.Points[0] != 4294967295 { + t.Fatalf("parsed %+v", fp) + } + if fp.Prefix() != int64(-4294967295) { // 0xFFFFFFFF_00000001 как int64 + t.Fatalf("prefix %d", fp.Prefix()) + } + if _, err := parseFPCalc([]byte(`{"duration": 1, "fingerprint": []}`)); err == nil { + t.Fatal("empty fingerprint must fail") + } +} diff --git a/services/api/internal/library/store.go b/services/api/internal/library/store.go index beacb86..5b7474a 100644 --- a/services/api/internal/library/store.go +++ b/services/api/internal/library/store.go @@ -271,7 +271,12 @@ func (st *Store) SaveFile(ctx context.Context, sourceID, id uuid.UUID, d FileDat mtime = EXCLUDED.mtime, content_hash = EXCLUDED.content_hash, format = EXCLUDED.format, codec = EXCLUDED.codec, bitrate = EXCLUDED.bitrate, sample_rate = EXCLUDED.sample_rate, bit_depth = EXCLUDED.bit_depth, channels = EXCLUDED.channels, duration_ms = EXCLUDED.duration_ms, - track_id = EXCLUDED.track_id, scan_error = EXCLUDED.scan_error, missing_since = NULL`, + track_id = EXCLUDED.track_id, scan_error = EXCLUDED.scan_error, missing_since = NULL, + -- Содержимое изменилось — отпечатки посчитаются заново. + sha256 = CASE WHEN media_files.content_hash IS DISTINCT FROM EXCLUDED.content_hash + OR media_files.size_bytes <> EXCLUDED.size_bytes THEN NULL ELSE media_files.sha256 END, + fingerprinted_at = CASE WHEN media_files.content_hash IS DISTINCT FROM EXCLUDED.content_hash + OR media_files.size_bytes <> EXCLUDED.size_bytes THEN NULL ELSE media_files.fingerprinted_at END`, id, sourceID, d.RelPath, d.Size, d.Mtime, d.ContentHash, p.Format, p.Codec, p.Bitrate, p.SampleRate, p.BitDepth, p.Channels, p.DurationMs, trackID, scanErr) if err != nil { @@ -623,3 +628,58 @@ func (st *Store) WaveMetrics(ctx context.Context) (WaveMetrics, error) { } return m, nil } + +// FingerprintTodo — файл, которому нужны отпечатки. +type FingerprintTodo struct { + ID uuid.UUID + SourceID uuid.UUID + RelPath string + SHA256 []byte // уже посчитан — повторно не читаем +} + +// FilesToFingerprint — файлы смонтированных источников без SHA-256 (и без Chromaprint, если withFP), после after. +func (st *Store) FilesToFingerprint(ctx context.Context, withFP bool, after uuid.UUID, limit int) ([]FingerprintTodo, error) { + rows, err := st.pool.Query(ctx, ` + SELECT mf.id, mf.source_id, mf.rel_path, mf.sha256 + FROM media_files mf + JOIN library_sources s ON s.id = mf.source_id AND s.status = 'mounted' + WHERE mf.missing_since IS NULL AND mf.scan_error IS NULL + AND (mf.sha256 IS NULL OR ($1 AND mf.fingerprinted_at IS NULL)) AND mf.id > $2 + ORDER BY mf.id + LIMIT $3`, withFP, after, limit) + if err != nil { + return nil, fmt.Errorf("library: files to fingerprint: %w", err) + } + return pgx.CollectRows(rows, func(r pgx.CollectableRow) (FingerprintTodo, error) { + var f FingerprintTodo + return f, r.Scan(&f.ID, &f.SourceID, &f.RelPath, &f.SHA256) + }) +} + +// SaveFingerprint сохраняет SHA-256 и (если есть) отпечаток Chromaprint; chromaprintDone — fpcalc отработал +// (в том числе с ошибкой), повторять не нужно. +func (st *Store) SaveFingerprint(ctx context.Context, fileID uuid.UUID, sum []byte, fp *Fingerprint, chromaprintDone bool) error { + tx, err := st.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("library: save fingerprint: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + if _, err := tx.Exec(ctx, ` + UPDATE media_files SET sha256 = $2, fingerprinted_at = CASE WHEN $3 THEN now() ELSE fingerprinted_at END + WHERE id = $1`, fileID, sum, chromaprintDone); err != nil { + return fmt.Errorf("library: save fingerprint: %w", err) + } + if fp != nil { + points := make([]int32, len(fp.Points)) + for i, p := range fp.Points { + points[i] = int32(p) + } + if _, err := tx.Exec(ctx, ` + INSERT INTO track_fingerprints (file_id, duration_ms, fingerprint, fp_prefix) VALUES ($1, $2, $3, $4) + ON CONFLICT (file_id) DO UPDATE SET duration_ms = EXCLUDED.duration_ms, fingerprint = EXCLUDED.fingerprint, + fp_prefix = EXCLUDED.fp_prefix`, fileID, fp.DurationMs, points, fp.Prefix()); err != nil { + return fmt.Errorf("library: save fingerprint: %w", err) + } + } + return tx.Commit(ctx) +} diff --git a/services/api/internal/platform/config/config.go b/services/api/internal/platform/config/config.go index 8e7d717..a79370e 100644 --- a/services/api/internal/platform/config/config.go +++ b/services/api/internal/platform/config/config.go @@ -39,6 +39,8 @@ type API struct { StorageAgentURL string `env:"STORAGE_AGENT_URL" envDefault:"http://storage-agent:8090"` InternalToken string `env:"INTERNAL_TOKEN"` FFprobePath string `env:"FFPROBE_PATH" envDefault:"ffprobe"` + // FPCalcPath — fpcalc (Chromaprint) для акустических отпечатков; нет бинарника — считается только SHA-256. + FPCalcPath string `env:"FPCALC_PATH" envDefault:"fpcalc"` // SigningKey — ключ подписи ссылок на обложки и стрим (base64, ≥ 16 байт). Пусто — случайный до перезапуска. SigningKey string `env:"STREAM_SIGNING_KEY"` diff --git a/services/api/migrations/00007_fingerprints.sql b/services/api/migrations/00007_fingerprints.sql new file mode 100644 index 0000000..78e6199 --- /dev/null +++ b/services/api/migrations/00007_fingerprints.sql @@ -0,0 +1,19 @@ +-- +goose Up +-- Отпечатки файлов библиотеки для проверки дублей при загрузке (docs/uploads.md#4-проверка-на-дубли): +-- полный SHA-256 — в media_files.sha256, акустический Chromaprint — здесь. +ALTER TABLE media_files ADD COLUMN fingerprinted_at timestamptz; +CREATE INDEX media_files_fingerprint_todo_idx ON media_files (id) WHERE fingerprinted_at IS NULL; + +CREATE TABLE track_fingerprints ( + file_id uuid PRIMARY KEY REFERENCES media_files(id) ON DELETE CASCADE, + duration_ms int NOT NULL, + fingerprint int[] NOT NULL, -- raw-отпечаток fpcalc (uint32 как int4) + fp_prefix bigint NOT NULL -- первые 64 бита — грубый индекс для отбора кандидатов +); +CREATE INDEX track_fingerprints_prefix_idx ON track_fingerprints (fp_prefix); +CREATE INDEX track_fingerprints_duration_idx ON track_fingerprints (duration_ms); + +-- +goose Down +DROP TABLE track_fingerprints; +DROP INDEX media_files_fingerprint_todo_idx; +ALTER TABLE media_files DROP COLUMN fingerprinted_at;