From 713098f5b75b57158bd6839bdda054c588fc3139 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E7=92=A8?= <1579649885@qq.com> Date: Wed, 30 Sep 2026 10:20:01 +0800 Subject: [PATCH] feat(app): add internal components batch 3 --- internal/app/emby_local_posters.go | 191 ++++++++++ internal/app/emby_merged.go | 113 ++++++ internal/app/emby_playback.go | 247 +++++++++++++ internal/app/emby_playback_test.go | 338 ++++++++++++++++++ internal/app/emby_posters_test.go | 397 +++++++++++++++++++++ internal/app/emby_sync.go | 551 +++++++++++++++++++++++++++++ internal/app/emby_sync_files.go | 285 +++++++++++++++ internal/app/emby_sync_posters.go | 247 +++++++++++++ internal/app/emby_sync_test.go | 335 ++++++++++++++++++ internal/app/ffmpeg_install.go | 406 +++++++++++++++++++++ 10 files changed, 3110 insertions(+) create mode 100644 internal/app/emby_local_posters.go create mode 100644 internal/app/emby_merged.go create mode 100644 internal/app/emby_playback.go create mode 100644 internal/app/emby_playback_test.go create mode 100644 internal/app/emby_posters_test.go create mode 100644 internal/app/emby_sync.go create mode 100644 internal/app/emby_sync_files.go create mode 100644 internal/app/emby_sync_posters.go create mode 100644 internal/app/emby_sync_test.go create mode 100644 internal/app/ffmpeg_install.go diff --git a/internal/app/emby_local_posters.go b/internal/app/emby_local_posters.go new file mode 100644 index 0000000..ebda41e --- /dev/null +++ b/internal/app/emby_local_posters.go @@ -0,0 +1,191 @@ +package app + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "os" + "path/filepath" + "sort" + "time" +) + +var embyPosterNames = []string{"poster.jpg", "poster.jpeg", "poster.png", "poster.webp", "poster.gif", "folder.jpg", "folder.png"} + +func embyArtworkHash(body []byte) string { + digest := sha256.Sum256(body) + return hex.EncodeToString(digest[:]) +} + +func readEmbyPosterDirectory(settings embySyncSettings, drama Drama, folder string) (string, embyFolderManifest, error) { + var manifest embyFolderManifest + root, err := filepath.EvalSymlinks(settings.OutputDir) + if err != nil { + return "", manifest, err + } + if _, err = embyDirectoryParent(root, folder, false); err != nil { + return "", manifest, err + } + directory := filepath.Join(root, filepath.FromSlash(folder)) + info, err := os.Lstat(directory) + if err != nil { + return "", manifest, err + } + if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { + return "", manifest, errors.New("Emby 海报目录不是普通文件夹") + } + manifestPath := filepath.Join(directory, ".juku-emby.json") + info, err = os.Lstat(manifestPath) + if err != nil { + return "", manifest, err + } + if !info.Mode().IsRegular() || info.Size() > 8<<20 { + return "", manifest, errors.New("Emby 目录标识不是有效的普通文件") + } + if err = readEmbyJSON(manifestPath, &manifest, 8<<20); err != nil { + return "", manifest, err + } + if manifest.Version != 1 || manifest.DramaID != drama.ID { + return "", manifest, errors.New("Emby 海报目录所属剧集不匹配") + } + return directory, manifest, nil +} + +func inspectEmbyArtwork(directory string, manifest embyFolderManifest) (managed, preserved string, err error) { + for _, name := range embyPosterNames { + path := filepath.Join(directory, name) + info, err := os.Lstat(path) + if errors.Is(err, os.ErrNotExist) { + continue + } + if err != nil { + return "", "", err + } + if !info.Mode().IsRegular() { + return "", "", errors.New("Emby 海报不是普通文件，已保留原内容") + } + if name != manifest.PosterFile || manifest.PosterHash == "" || info.Size() > maxCoverBytes { + return "", name, nil + } + body, err := os.ReadFile(path) + if err != nil { + return "", "", err + } + if embyArtworkHash(body) != manifest.PosterHash { + return "", name, nil + } + managed = name + } + return managed, "", nil +} + +func (app *UIApp) syncLocalEmbyPoster(ctx context.Context, settings embySyncSettings, drama Drama, folder, remote string) (string, bool, error) { + directory, manifest, err := readEmbyPosterDirectory(settings, drama, folder) + if err != nil { + return "", false, err + } + managed, preserved, err := inspectEmbyArtwork(directory, manifest) + if err != nil || preserved != "" { + return preserved, false, err + } + sourceHash := embyPosterSourceHash(remote) + if managed != "" && manifest.PosterSource == sourceHash { + return managed, false, nil + } + source := sourceHongguo + data, err := app.loadCoverImage(context.WithValue(ctx, coverSourceKey{}, source), remote, nil) + if err != nil { + return "", false, err + } + extension := map[string]string{"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp", "image/gif": ".gif"}[imageContentType(data)] + if extension == "" { + return "", false, errors.New("海报图片格式不受支持") + } + if err = ctx.Err(); err != nil { + return "", false, err + } + + directory, manifest, err = readEmbyPosterDirectory(settings, drama, folder) + if err != nil { + return "", false, err + } + managed, preserved, err = inspectEmbyArtwork(directory, manifest) + if err != nil || preserved != "" { + return preserved, false, err + } + name := "poster" + extension + changed, err := writeEmbyFile(filepath.Join(directory, name), data, 0644) + if err != nil { + return "", false, err + } + previousHash := manifest.PosterHash + manifest.PosterFile, manifest.PosterHash, manifest.PosterSource = name, embyArtworkHash(data), sourceHash + if err = writeEmbyJSON(filepath.Join(directory, ".juku-emby.json"), manifest, 0600); err != nil { + return name, changed, err + } + if managed != "" && managed != name { + old := filepath.Join(directory, managed) + if info, err := os.Lstat(old); err == nil && info.Mode().IsRegular() && info.Size() <= maxCoverBytes { + if body, err := os.ReadFile(old); err == nil && embyArtworkHash(body) == previousHash { + _ = os.Remove(old) + } + } + } + return name, changed, nil +} + +func (manager *embySyncManager) synchronizeLocalPosters(ctx context.Context, document *embySyncDocument, targets []Drama) { + interval := time.Duration(document.Settings.IntervalMinutes) * time.Minute + var due []Drama + for _, drama := range targets { + entry := document.Entries[drama.ID] + poster := entry.Poster + if entry.Folder == "" { + continue + } + if next := embyLocalPosterDue(poster, interval); !next.IsZero() && !next.After(time.Now()) { + due = append(due, drama) + } + } + sort.Slice(due, func(i, j int) bool { + left, right := document.Entries[due[i].ID].Poster.LocalCheckedAt, document.Entries[due[j].ID].Poster.LocalCheckedAt + if left.Equal(right) { + return due[i].ID < due[j].ID + } + return left.Before(right) + }) + if len(due) > 20 { + due = due[:20] + } + for _, drama := range due { + if ctx.Err() != nil { + return + } + entry := document.Entries[drama.ID] + poster := entry.Poster + requestCtx, cancel := context.WithTimeout(ctx, 45*time.Second) + name, changed, err := manager.app.syncLocalEmbyPoster(requestCtx, document.Settings, drama, entry.Folder, poster.SourceURL) + cancel() + document.PosterChecked++ + poster.LocalCheckedAt = time.Now() + poster.LocalError = manager.safeError(err, document.Settings.APIKey) + if err != nil { + poster.LocalRetryAt = time.Now().Add(5 * time.Minute) + } else { + poster.LocalFile, poster.LocalSource, poster.LocalRetryAt = name, embyPosterSourceHash(poster.SourceURL), time.Time{} + } + if changed { + document.WrittenFiles++ + document.RefreshPending = document.Settings.ServerURL != "" + } + if document.Settings.ServerURL == "" { + poster.Status, poster.Error, poster.CheckedAt, poster.RetryAt = "local", poster.LocalError, poster.LocalCheckedAt, poster.LocalRetryAt + if err != nil { + poster.Status = "failed" + } + } + entry.Poster = poster + document.Entries[drama.ID] = entry + } +} diff --git a/internal/app/emby_merged.go b/internal/app/emby_merged.go new file mode 100644 index 0000000..a1b62ca --- /dev/null +++ b/internal/app/emby_merged.go @@ -0,0 +1,113 @@ +package app + +import ( + "context" + "encoding/xml" + "fmt" + "net/http" + "net/url" + "os" + "time" +) + +const embyMergedChapterID = "juku:merged" + +type embyMergedRecord struct { + StartEpisode int `json:"startEpisode"` + EndEpisode int `json:"endEpisode"` +} + +func (record *embyMergedRecord) valid() bool { + return record != nil && record.StartEpisode >= 1 && record.EndEpisode >= record.StartEpisode && record.EndEpisode <= 2000 +} + +func (app *UIApp) embyMergedRecord(id string) *embyMergedRecord { + app.mu.Lock() + defer app.mu.Unlock() + state := app.merges[id] + if state == nil || state.Status != "success" || completedPlaybackPath(&UITask{Status: uiStatusSuccess, Path: state.OutputPath}) == "" { + return nil + } + record := &embyMergedRecord{state.StartEpisode, state.EndEpisode} + if !record.valid() { + return nil + } + return record +} + +func writeEmbyMerged(drama Drama, record *embyMergedRecord, base string, key []byte, owner string, write func(string, []byte) error) error { + if !record.valid() { + return fmt.Errorf("合并版集数范围无效") + } + episode := struct { + XMLName xml.Name `xml:"episodedetails"` + Title string `xml:"title"` + Show string `xml:"showtitle"` + Season int `xml:"season"` + Episode int `xml:"episode"` + }{Title: fmt.Sprintf("合并版（第%d–%d集）", record.StartEpisode, record.EndEpisode), Show: drama.DisplayTitle(), Season: 0, Episode: 1} + body, err := xml.MarshalIndent(episode, "", " ") + if err != nil { + return err + } + if err = write("Season 00/S00E001.nfo", append([]byte(xml.Header), body...)); err != nil { + return err + } + query := url.Values{"id": {drama.ID}, "chapter": {embyMergedChapterID}, "key": {embyToken(key, drama.ID, embyMergedChapterID, owner)}} + if owner != "" { + query.Set("account", owner) + } + return write("Season 00/S00E001.strm", []byte(base+"/api/emby/merged.mp4?"+query.Encode()+"\n")) +} + +func (app *UIApp) handleEmbyMerged(writer http.ResponseWriter, request *http.Request) { + id, chapter, ok := app.authorizeEmby(writer, request) + if !ok { + return + } + if chapter != embyMergedChapterID { + writeJSON(writer, http.StatusForbidden, map[string]string{"error": "合并版播放链接无效，请重新同步"}) + return + } + app.mu.Lock() + _, path, err := app.mergedPlaybackTaskLocked(mergedPlaybackPrefix + id) + app.mu.Unlock() + if err != nil { + app.writeEmbyPlaybackError(writer, request, http.StatusNotFound, app.redactError(err)) + return + } + release, err := app.mediaResources().acquire(request.Context(), "media", false) + if err != nil { + app.writeEmbyPlaybackError(writer, request, http.StatusServiceUnavailable, app.redactError(err)) + return + } + defer release() + file, err := os.Open(path) + if err != nil { + app.writeEmbyPlaybackError(writer, request, http.StatusNotFound, "本地合并文件无法读取，请检查文件是否移动") + return + } + defer file.Close() + info, err := file.Stat() + if err != nil || !info.Mode().IsRegular() || info.Size() == 0 { + app.writeEmbyPlaybackError(writer, request, http.StatusNotFound, "本地合并文件无效，请重新合并") + return + } + writer.Header().Set("Content-Type", "video/mp4") + writer.Header().Set("X-Accel-Buffering", "no") + http.ServeContent(writer, request, "merged.mp4", info.ModTime(), file) +} + +func (app *UIApp) embyExportChapters(ctx context.Context, id string, merged *embyMergedRecord) (string, []Chapter, error) { + if merged == nil { + return app.downloader.GetDramaChapters(ctx, id) + } + lookup, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + title, chapters, err := app.downloader.GetDramaChapters(lookup, id) + if err != nil && ctx.Err() == nil { + app.downloader.recordDiagnostic(diagnosticEvent{Level: "warn", Event: "emby.metadata_fallback", DramaID: id, Message: app.redactError(err)}) + return "", nil, nil + } + return title, chapters, err +} diff --git a/internal/app/emby_playback.go b/internal/app/emby_playback.go new file mode 100644 index 0000000..8cd75ce --- /dev/null +++ b/internal/app/emby_playback.go @@ -0,0 +1,247 @@ +package app + +import ( + "context" + "errors" + "math" + "net/http" + "net/url" + "time" +) + +func (app *UIApp) activePlaybackSessionsLocked() int { + count := 0 + for _, session := range app.playbacks { + if !session.emby || session.mediaWaiters > 0 { + count++ + } + } + return count +} + +func (app *UIApp) trimEmbyPlaybacksLocked(reserve int) []context.CancelFunc { + limit := app.mediaResources().snapshot().MaxSessions - reserve + var cancellations []context.CancelFunc + for { + count := 0 + var oldest *playbackSession + for _, session := range app.playbacks { + if !session.emby || session.native == nil && (session.media == nil || session.media.generated == nil) { + continue + } + count++ + if session.mediaWaiters == 0 && session.state != "opening" && (oldest == nil || session.expires.Before(oldest.expires)) { + oldest = session + } + } + if count <= limit || oldest == nil { + return cancellations + } + if oldest.cancel != nil { + cancellations = append(cancellations, oldest.cancel) + } + if oldest.media != nil { + oldest.embyProcessing = oldest.media.plan.Processing + } + oldest.cancel, oldest.native, oldest.media = nil, nil, nil + oldest.mediaReady, oldest.mediaError, oldest.state = nil, nil, "ready" + } +} + +func (app *UIApp) releaseEmbyPlayback(session *playbackSession) { + app.playbackMu.Lock() + session.mediaWaiters-- + var cancellations []context.CancelFunc + if app.playbacks[session.id] == session && session.mediaWaiters == 0 { + if session.state == "opening" || session.mediaError != nil { + delete(app.playbacks, session.id) + session.timer.Stop() + if session.cancel != nil { + cancellations = append(cancellations, session.cancel) + } + } else { + app.touchPlaybackLocked(session) + } + } + cancellations = append(cancellations, app.trimEmbyPlaybacksLocked(0)...) + app.playbackMu.Unlock() + for _, cancel := range cancellations { + cancel() + } +} + +func (app *UIApp) writeEmbyPlaybackError(writer http.ResponseWriter, request *http.Request, status int, message string) { + if status == http.StatusTooManyRequests { + writer.Header().Set("Retry-After", "2") + } + id := request.URL.Query().Get("id") + app.downloader.recordDiagnostic(diagnosticEvent{Event: "emby.playback_failed", HTTPStatus: status, + Source: sourceFromDramaID(id), DramaID: id, Message: message}) + writeJSON(writer, status, map[string]string{"error": message}) +} + +func (app *UIApp) acquireEmbyPlayback(writer http.ResponseWriter, request *http.Request, id, chapter, sessionID string, legacy bool) *playbackSession { + owner := request.URL.Query().Get("account") + app.playbackMu.Lock() + var session *playbackSession + if sessionID != "" { + session = app.playbacks[sessionID] + if session == nil || !session.emby || session.viewer != nil || session.accountID != owner || session.dramaID != id || len(session.tasks) != 1 || session.tasks[0].Chapter.ID != chapter { + app.playbackMu.Unlock() + app.writeEmbyPlaybackError(writer, request, http.StatusGone, "播放已过期，请重新播放") + return nil + } + } else { + for _, candidate := range app.playbacks { + if candidate.emby && candidate.embyLegacy == legacy && candidate.accountID == owner && candidate.dramaID == id && len(candidate.tasks) == 1 && candidate.tasks[0].Chapter.ID == chapter && candidate.mediaError == nil { + session = candidate + break + } + } + } + if (session == nil || session.mediaWaiters == 0) && app.activePlaybackSessionsLocked() >= app.mediaResources().snapshot().MaxSessions { + app.playbackMu.Unlock() + app.writeEmbyPlaybackError(writer, request, http.StatusTooManyRequests, "同时播放数量已达上限，请稍后再试") + return nil + } + if session == nil { + sessionID = randomHex(24) + if sessionID == "" { + app.playbackMu.Unlock() + app.writeEmbyPlaybackError(writer, request, http.StatusInternalServerError, "无法创建播放会话") + return nil + } + session = &playbackSession{id: sessionID, accountID: owner, dramaID: id, emby: true, embyLegacy: legacy, run: 1, + tasks: []Task{{DramaID: id, Chapter: Chapter{ID: chapter}}}} + if app.playbacks == nil { + app.playbacks = make(map[string]*playbackSession) + } + app.playbacks[sessionID] = session + session.timer = time.AfterFunc(playbackIdleTimeout, func() { app.expirePlayback(sessionID) }) + } + session.mediaWaiters++ + app.touchPlaybackLocked(session) + var cancellations []context.CancelFunc + if session.mediaWaiters == 1 { + failed := sessionID == "" && session.media != nil && (session.media.failed.Load() || session.media.ctx.Err() != nil || session.media.proxy != nil && session.media.proxy.Err() != nil) + if session.native != nil { + state, err := session.native.state() + failed = failed || err != nil || state == "stopped" + } + if failed { + if session.cancel != nil { + cancellations = append(cancellations, session.cancel) + } + session.cancel, session.native, session.media, session.mediaReady, session.mediaError = nil, nil, nil, nil, nil + } + } + var parent context.Context + if session.mediaReady == nil { + cancellations = append(cancellations, app.trimEmbyPlaybacksLocked(1)...) + parent, session.cancel = context.WithCancel(context.Background()) + session.mediaReady, session.state = make(chan struct{}), "opening" + } + ready := session.mediaReady + app.playbackMu.Unlock() + for _, cancel := range cancellations { + cancel() + } + if parent != nil { + query := url.Values{"id": {id}, "chapter": {chapter}, "key": {request.URL.Query().Get("key")}} + if owner != "" { + query.Set("account", owner) + } + go app.prepareEmbyPlayback(parent, session, ready, query.Encode(), request.Header.Get("Origin")) + } + select { + case <-ready: + case <-request.Context().Done(): + app.releaseEmbyPlayback(session) + return nil + } + app.playbackMu.Lock() + err := session.mediaError + if err == nil && (app.playbacks[session.id] != session || session.media == nil && session.native == nil) { + err = errors.New("播放会话已结束") + } + app.playbackMu.Unlock() + if err != nil { + app.releaseEmbyPlayback(session) + app.writeEmbyPlaybackError(writer, request, http.StatusBadGateway, "准备 Emby 播放失败："+app.redactError(err)) + return nil + } + return session +} + +func (app *UIApp) prepareEmbyPlayback(parent context.Context, session *playbackSession, ready chan struct{}, query, origin string) { + ctx, cancel := context.WithTimeout(parent, 90*time.Second) + defer cancel() + task, downloadID, err := app.embyTask(ctx, session.dramaID, session.tasks[0].Chapter.ID) + var media *playbackMediaSession + var native *playbackNative + var duration float64 + if err == nil && !session.embyLegacy { + media, err = app.resolveMediaSession(parent, ctx, task, downloadID, 0) + if err == nil { + base := "/api/emby/media/" + session.id + "/" + mode := "auto" + if session.embyProcessing == "remux" { + mode = "compatible" + } else if session.embyProcessing == "audio" { + mode = "audio" + } + err = app.planMedia(ctx, media, playbackClientCapabilities{MP4: true, NativeHLS: true}, mode, sourceFromDramaID(task.DramaID), origin, base, query) + if err == nil { + duration = media.plan.Duration + if media.plan.Player == "legacy" { + media.Close() + media = nil + } + } + } + } + if err == nil && media == nil { + native = newPlaybackNative(app, parent, task, downloadID, 0, 0, false) + native.start() + _, err = native.segment(ctx, 0) + if err == nil { + _, duration, _ = native.metadata() + if duration <= 0 || duration > 24*60*60 || math.IsNaN(duration) || math.IsInf(duration, 0) { + err = errors.New("未取得有效播放时长") + } + } + } + app.playbackMu.Lock() + if err == nil && (app.playbacks[session.id] != session || session.mediaReady != ready || ctx.Err() != nil) { + err = context.Canceled + } + if err == nil { + previous := session.cancel + session.cancel = func() { + previous() + if media != nil { + media.Close() + } + if native != nil { + native.Close() + } + } + session.tasks, session.media, session.native, session.duration, session.state = []Task{task}, media, native, duration, "streaming" + if media != nil { + media.plan.Run = session.run + } + app.touchPlaybackLocked(session) + } else { + session.mediaError, session.state = err, "failed" + } + close(ready) + app.playbackMu.Unlock() + if err != nil { + if media != nil { + media.Close() + } + if native != nil { + native.Close() + } + } +} diff --git a/internal/app/emby_playback_test.go b/internal/app/emby_playback_test.go new file mode 100644 index 0000000..8cd6fdf --- /dev/null +++ b/internal/app/emby_playback_test.go @@ -0,0 +1,338 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "os/exec" + "strings" + "sync" + "testing" + "time" +) + +func embyScanFixture(t *testing.T, optimized bool) (*UIApp, func(int) string) { + t.Helper() + app, fixture := nativePlaybackFixture(t, "4", "160x90") + app.closePlayback(fixture.id) + app.mediaResources().settings.Enabled = optimized + app.mediaResources().settings.MaxSessions = 1 + for index := 1; index <= 8; index++ { + task := fixture.tasks[0] + task.Index, task.Total = index, 8 + task.Chapter.ID = fmt.Sprintf("%s:%d", task.DramaID, index) + task.Chapter.CurrentEpisode = rawEpisode(index) + id := fmt.Sprintf("scan-%d", index) + app.tasks[id] = &UITask{ID: id, DramaID: task.DramaID, Status: uiStatusSuccess, Path: task.OutPath, Source: task} + app.taskOrder = append(app.taskOrder, id) + } + key, err := app.embySigningKey(true) + if err != nil { + t.Fatal(err) + } + return app, func(index int) string { + task := app.tasks[fmt.Sprintf("scan-%d", index)].Source + query := url.Values{"id": {task.DramaID}, "chapter": {task.Chapter.ID}, "key": {embyToken(key, task.DramaID, task.Chapter.ID)}} + return "/api/emby/stream.m3u8?" + query.Encode() + } +} + +func embyRequest(app *UIApp, method, address string) *httptest.ResponseRecorder { + writer := httptest.NewRecorder() + app.routes().ServeHTTP(writer, httptest.NewRequest(method, "http://localhost"+address, nil)) + return writer +} + +func embyFirstSegment(t *testing.T, playlist string) string { + t.Helper() + for _, line := range strings.Split(playlist, "\n") { + if strings.HasPrefix(line, "segment.ts?") { + return "/api/emby/" + line + } + } + t.Fatal("missing signed HLS segment") + return "" +} + +func TestEmbySerialProbesReleaseSlotsAndReusePlayback(t *testing.T) { + probe, err := exec.LookPath("ffprobe") + if err != nil { + t.Skip("ffprobe unavailable") + } + for _, optimized := range []bool{false, true} { + t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) { + app, address := embyScanFixture(t, optimized) + server := httptest.NewServer(app.routes()) + defer server.Close() + first := embyRequest(app, http.MethodGet, address(1)) + firstBody, firstLocation := first.Body.String(), first.Header().Get("Location") + for index := 1; index <= 8; index++ { + if head := embyRequest(app, http.MethodHead, address(index)); head.Code != 200 || head.Body.Len() != 0 { + t.Fatal("HEAD failed", head.Code) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + output, err := exec.CommandContext(ctx, probe, "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", server.URL+address(index)).CombinedOutput() + cancel() + if err != nil || !strings.Contains(string(output), "4.000000") { + t.Fatalf("serial probe %d failed: %v %s", index, err, output) + } + app.playbackMu.Lock() + active := app.activePlaybackSessionsLocked() + cached := 0 + for _, session := range app.playbacks { + if session.native != nil { + cached++ + } + } + app.playbackMu.Unlock() + if active != 0 || cached > 1 { + t.Fatal("finished probe retained a slot or unbounded media cache", active, cached) + } + } + if !optimized { + part := embyRequest(app, http.MethodGet, embyFirstSegment(t, firstBody)) + if part.Code != 200 || part.Body.Len() < 188 || part.Body.Bytes()[0] != 0x47 { + t.Fatal("evicting an idle cache broke an existing HLS URL", part.Code, part.Body.String()) + } + } + repeated := embyRequest(app, http.MethodGet, address(1)) + if repeated.Code != first.Code || repeated.Body.String() != firstBody || repeated.Header().Get("Location") != firstLocation { + t.Fatal("repeated episode did not reuse its prepared session", repeated.Code) + } + web := historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"}) + if web.Code != 200 { + t.Fatal("completed Emby scans blocked a new browser playback", web.Code, web.Body.String()) + } + }) + } +} + +type embyBlockingWriter struct { + *httptest.ResponseRecorder + entered chan struct{} + resume chan struct{} + once sync.Once +} + +func (writer *embyBlockingWriter) Write(body []byte) (int, error) { + writer.once.Do(func() { + close(writer.entered) + <-writer.resume + }) + return writer.ResponseRecorder.Write(body) +} + +func TestEmbyActiveRequestsShareLimitAndReleaseOnCompletion(t *testing.T) { + for _, optimized := range []bool{false, true} { + t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) { + app, address := embyScanFixture(t, optimized) + first := embyRequest(app, http.MethodGet, address(1)) + asset := first.Header().Get("Location") + if !optimized { + asset = embyFirstSegment(t, first.Body.String()) + } + writer := &embyBlockingWriter{ResponseRecorder: httptest.NewRecorder(), entered: make(chan struct{}), resume: make(chan struct{})} + done := make(chan struct{}) + go func() { + defer close(done) + app.routes().ServeHTTP(writer, httptest.NewRequest(http.MethodGet, "http://localhost"+asset, nil)) + }() + var resume sync.Once + defer resume.Do(func() { close(writer.resume); <-done }) + select { + case <-writer.entered: + case <-time.After(10 * time.Second): + t.Fatal("media request never reached the response writer") + } + if second := embyRequest(app, http.MethodGet, address(2)); second.Code != 429 || second.Header().Get("Retry-After") == "" { + t.Fatal("active requests bypassed the configured limit", second.Code) + } + if same := embyRequest(app, http.MethodGet, address(1)); same.Code != first.Code { + t.Fatal("concurrent requests for the same episode were counted twice", same.Code) + } + web := historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"}) + if web.Code != 429 { + t.Fatal("browser playback bypassed an active Emby request", web.Code) + } + resume.Do(func() { close(writer.resume); <-done }) + if writer.Code != 200 { + t.Fatal("the original media request was interrupted", writer.Code) + } + if second := embyRequest(app, http.MethodGet, address(2)); second.Code != first.Code { + t.Fatal("completed request did not release its slot", second.Code) + } + web = historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"}) + if web.Code != 200 { + t.Fatal("released Emby slot was not available to the browser", web.Code, web.Body.String()) + } + var opened struct{ Session string } + if err := json.Unmarshal(web.Body.Bytes(), &opened); err != nil { + t.Fatal(err) + } + app.closePlayback(opened.Session) + }) + } +} + +func TestEmbyCanceledProbeReleasesPendingPreparation(t *testing.T) { + started, stopped := make(chan struct{}), make(chan struct{}) + var startOnce, stopOnce sync.Once + d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { + startOnce.Do(func() { close(started) }) + <-request.Context().Done() + stopOnce.Do(func() { close(stopped) }) + return nil, request.Context().Err() + }) + app := &UIApp{downloader: d, cfg: d.cfg} + t.Cleanup(app.closePlaybacks) + key, err := app.embySigningKey(true) + if err != nil { + t.Fatal(err) + } + id, chapter := "hongguo:7000000000000000001", "hongguo:7000000000000000001:1" + query := url.Values{"id": {id}, "chapter": {chapter}, "key": {embyToken(key, id, chapter)}} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + request := httptest.NewRequest(http.MethodGet, "http://localhost/api/emby/stream.m3u8?"+query.Encode(), nil).WithContext(ctx) + app.handleEmbyStream(httptest.NewRecorder(), request) + }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("preparation did not start") + } + other, cancelOther := context.WithCancel(context.Background()) + defer cancelOther() + otherDone := make(chan struct{}) + go func() { + defer close(otherDone) + request := httptest.NewRequest(http.MethodGet, "http://localhost/api/emby/stream.m3u8?"+query.Encode(), nil).WithContext(other) + app.handleEmbyStream(httptest.NewRecorder(), request) + }() + deadline := time.Now().Add(5 * time.Second) + for { + app.playbackMu.Lock() + waiters := 0 + for _, session := range app.playbacks { + waiters += session.mediaWaiters + } + app.playbackMu.Unlock() + if waiters == 2 { + break + } + if time.Now().After(deadline) { + t.Fatal("second request did not join the preparation") + } + time.Sleep(time.Millisecond) + } + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("first canceled request did not finish") + } + select { + case <-stopped: + t.Fatal("one canceled client stopped another client's preparation") + default: + } + cancelOther() + for _, signal := range []chan struct{}{otherDone, stopped} { + select { + case <-signal: + case <-time.After(5 * time.Second): + t.Fatal("canceling the probe left its preparation running") + } + } + app.playbackMu.Lock() + defer app.playbackMu.Unlock() + if len(app.playbacks) != 0 { + t.Fatal("canceled probe retained a playback session") + } +} + +func TestEmbyEvictedRemuxRestoresExistingSegmentURLs(t *testing.T) { + app, address := embyScanFixture(t, true) + first := embyRequest(app, http.MethodGet, address(1)) + if first.Code != http.StatusFound { + t.Fatal(first.Code, first.Body.String()) + } + parsed, _ := url.Parse(first.Header().Get("Location")) + base := strings.TrimSuffix(parsed.Path, "file.mp4") + sessionID := strings.Split(base, "/")[4] + app.playbackMu.Lock() + session := app.playbacks[sessionID] + media := session.media + app.playbackMu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := app.generatePlayback(ctx, media, "remux", base, parsed.RawQuery); err != nil { + t.Fatal(err) + } + segment := media.generated.assetURL(media.generated.segments[0].Name) + before := embyRequest(app, http.MethodGet, segment) + if before.Code != 200 { + t.Fatal("prepared remux was unreadable", before.Code) + } + if next := embyRequest(app, http.MethodGet, address(2)); next.Code != http.StatusFound { + t.Fatal("could not prepare another episode", next.Code) + } + if media.ctx.Err() == nil { + t.Fatal("idle remux cache was not released") + } + after := embyRequest(app, http.MethodGet, segment) + if after.Code != 200 || !bytes.Equal(before.Body.Bytes(), after.Body.Bytes()) { + t.Fatal("reclaiming cached files changed or invalidated an existing segment", after.Code) + } +} + +func TestEmbyPlaybackCacheRemainsAccountScoped(t *testing.T) { + for _, optimized := range []bool{false, true} { + t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) { + app, address := embyScanFixture(t, optimized) + store := app.browserViewers().accountStore() + store.mu.Lock() + for _, owner := range []string{"owner-a", "owner-b"} { + store.state.Accounts[owner] = accountRecord{ID: owner, Sources: []string{sourceHongguo}} + } + store.mu.Unlock() + key, err := app.embySigningKey(false) + if err != nil { + t.Fatal(err) + } + sign := func(path, owner string) string { + parsed, _ := url.Parse(path) + query := parsed.Query() + query.Set("account", owner) + query.Set("key", embyToken(key, query.Get("id"), query.Get("chapter"), owner)) + parsed.RawQuery = query.Encode() + return parsed.String() + } + first := embyRequest(app, http.MethodGet, sign(address(1), "owner-a")) + asset := first.Header().Get("Location") + if !optimized { + asset = embyFirstSegment(t, first.Body.String()) + } + if forged := embyRequest(app, http.MethodGet, sign(asset, "owner-b")); forged.Code != http.StatusGone { + t.Fatal("valid credentials for another account reused a private cache", forged.Code) + } + second := embyRequest(app, http.MethodGet, sign(address(1), "owner-b")) + if second.Code != first.Code { + t.Fatal("second account could not prepare its own session", second.Code) + } + app.playbackMu.Lock() + count := len(app.playbacks) + app.playbackMu.Unlock() + if count != 2 { + t.Fatal("accounts shared a playback session", count) + } + }) + } +} diff --git a/internal/app/emby_posters_test.go b/internal/app/emby_posters_test.go new file mode 100644 index 0000000..8c5cef0 --- /dev/null +++ b/internal/app/emby_posters_test.go @@ -0,0 +1,397 @@ +package app + +import ( + "archive/zip" + "bytes" + "context" + "encoding/json" + "encoding/xml" + "errors" + "image" + "image/png" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +const embyFixtureCover = "https://p3-shortvideo.byteimg.com/synthetic.png?signature=private-cover-token" + +type embyPosterTestTransport func(*http.Request) (*http.Response, error) + +func (transport embyPosterTestTransport) RoundTrip(request *http.Request) (*http.Response, error) { + return transport(request) +} + +func embySyntheticPoster(t *testing.T) []byte { + t.Helper() + var data bytes.Buffer + if err := png.Encode(&data, image.NewRGBA(image.Rect(0, 0, 1, 1))); err != nil { + t.Fatal(err) + } + return data.Bytes() +} + +func TestEmbyArchiveIncludesSignedPosterMetadataOnly(t *testing.T) { + drama := Drama{ID: historyFixtureDramaID, Title: "海报 <&>", CoverURL: embyFixtureCover} + key := bytes.Repeat([]byte{7}, 32) + body, err := buildEmbyArchive(drama, embySyncTestChapters(), "https://library.test/juku", key, "poster-owner") + if err != nil { + t.Fatal(err) + } + archive, err := zip.NewReader(bytes.NewReader(body), int64(len(body))) + if err != nil { + t.Fatal(err) + } + if len(archive.File) != 5 { + t.Fatal("export should contain only STRM and NFO files") + } + found := false + for _, file := range archive.File { + if !strings.HasSuffix(file.Name, ".strm") && !strings.HasSuffix(file.Name, ".nfo") { + t.Fatal("export created an artwork or media file", file.Name) + } + if !strings.HasSuffix(file.Name, "/tvshow.nfo") { + continue + } + reader, _ := file.Open() + var nfo struct { + Title string `xml:"title"` + Thumb struct { + URL string `xml:",chardata"` + Aspect string `xml:"aspect,attr"` + } `xml:"thumb"` + Identity struct { + Value string `xml:",chardata"` + Type string `xml:"type,attr"` + } `xml:"uniqueid"` + } + err = xml.NewDecoder(reader).Decode(&nfo) + reader.Close() + if err != nil || nfo.Title != drama.Title || nfo.Thumb.Aspect != "poster" || nfo.Identity.Type != "juku" || nfo.Identity.Value != embyMetadataID(key, drama.ID) { + t.Fatal("invalid NFO metadata", err) + } + cover, err := url.Parse(nfo.Thumb.URL) + if err != nil || cover.Scheme != "https" || cover.Host != "library.test" || cover.Path != "/juku/api/emby/cover" { + t.Fatal("cover URL does not preserve the configured reverse proxy path") + } + query := cover.Query() + if query.Get("id") != drama.ID || query.Get("account") != "poster-owner" || + query.Get("key") != embyCoverToken(key, drama.ID, "poster-owner", query.Get("v")) || strings.Contains(nfo.Thumb.URL, "private-cover-token") { + t.Fatal("cover URL is not scoped to this account and drama") + } + found = true + } + if !found { + t.Fatal("missing tvshow.nfo") + } + drama.CoverURL = "https://untrusted.example.invalid/image" + if embyCoverURL(drama, "https://library.test", key, "poster-owner") != "" { + t.Fatal("untrusted source was included in the exported poster URL") + } +} + +func TestEmbyCoverWithoutCookiesChecksSignatureAndCurrentPermissions(t *testing.T) { + app, admin := administratorFixture(t) + member := newAccountTestBrowser(t, app) + member.register(t, "poster-viewer") + store := app.browserViewers().accountStore() + owner := store.state.Accounts["poster-viewer"].ID + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/admin/settings", map[string]bool{"requireLogin": true, "allowRegistration": true})) + app.dramas[0].CoverURL = embyFixtureCover + data := embySyntheticPoster(t) + var calls atomic.Int32 + app.downloader.client.Transport = embyPosterTestTransport(func(request *http.Request) (*http.Response, error) { + calls.Add(1) + if request.URL.String() != embyFixtureCover || request.Header.Get("Referer") == "" || request.Header.Get("X-Emby-Token") != "" { + t.Error("cover request lost source headers or leaked Emby credentials") + } + return &http.Response{StatusCode: 200, Header: http.Header{}, Body: io.NopCloser(bytes.NewReader(data)), Request: request}, nil + }) + key, err := app.embySigningKey(true) + if err != nil { + t.Fatal(err) + } + address := embyCoverURL(app.dramas[0], "http://library.test", key, owner) + request := func(handler http.Handler, method, address string) *httptest.ResponseRecorder { + result := httptest.NewRecorder() + handler.ServeHTTP(result, httptest.NewRequest(method, address, nil)) + return result + } + for _, mutation := range []struct{ name, value string }{ + {"id", "hongguo:7000000000000000002"}, {"account", "another-account"}, {"v", strings.Repeat("0", 32)}, + {"key", embyToken(key, historyFixtureDramaID, "chapter-1", owner)}, {"key", "invalid"}, + } { + parsed, _ := url.Parse(address) + query := parsed.Query() + query.Set(mutation.name, mutation.value) + parsed.RawQuery = query.Encode() + if result := request(app.routes(), http.MethodGet, parsed.String()); result.Code != 403 { + t.Fatal("forged cover was accepted", mutation.name, result.Code) + } + } + if calls.Load() != 0 { + t.Fatal("invalid signatures reached the source") + } + result := request(app.routes(), http.MethodGet, address) + if result.Code != 200 || !bytes.Equal(result.Body.Bytes(), data) || result.Header().Get("Content-Type") != "image/png" || + result.Header().Get("Cache-Control") != "private, no-store" { + t.Fatal("signed cover was not usable without browser cookies", result.Code) + } + head := request(app.routes(), http.MethodHead, address) + if head.Code != 200 || head.Body.Len() != 0 || calls.Load() != 1 { + t.Fatal("cover HEAD/cache failed") + } + if result := request(app.routes(), http.MethodPost, address); result.Code != 405 { + t.Fatal("unsupported cover method accepted") + } + restarted := &UIApp{cfg: app.cfg, downloader: app.downloader, dramas: app.dramas} + if result := request(restarted.routes(), http.MethodGet, address); result.Code != 200 { + t.Fatal("poster link did not survive restart", result.Code) + } + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/admin/accounts/permissions", map[string]any{"username": "poster-viewer", "onlineOnly": true})) + before := calls.Load() + if result := request(app.routes(), http.MethodGet, address); result.Code != 403 || calls.Load() != before { + t.Fatal("online-only account retained poster export access") + } +} + +func TestEmbyPosterMatchesIdentityAndPortableFolderPaths(t *testing.T) { + folder, identity := "同名剧 [0123456789abcdef]", "fixture-provider-id" + for _, prefix := range []string{"/mnt/emby/", "D:\\Media\\Emby\\"} { + item, err := matchEmbySeries([]embyMediaItem{ + {ID: "other", Type: "Series", Path: prefix + "同名剧 [fedcba9876543210]"}, + {ID: "wanted", Type: "Series", Path: prefix + folder, ProviderIDs: map[string]string{"Juku": identity}}, + }, folder, identity) + if err != nil || item.ID != "wanted" { + t.Fatal("portable folder/identity match failed", item.ID, err) + } + } + items := []embyMediaItem{ + {ID: "first", Type: "Series", Path: "/one/" + folder}, + {ID: "second", Type: "Series", Path: "/two/" + folder}, + } + if _, err := matchEmbySeries(items, folder, identity); err == nil { + t.Fatal("ambiguous folders were matched by title") + } + items[0].ProviderIDs = map[string]string{"juku": "different-instance"} + items[1].ID = "../escape" + if item, err := matchEmbySeries(items, folder, identity); err != nil || item.ID != "" { + t.Fatal("wrong instance or unsafe item ID accepted") + } +} + +func TestEmbyPostersScanDelayRetryRestartAndCustomArtwork(t *testing.T) { + app, admin := administratorFixture(t) + manager := app.embySyncer() + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/following", map[string]any{"dramaId": historyFixtureDramaID, "saved": true})) + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/admin/settings", map[string]bool{"requireLogin": true, "allowRegistration": true})) + app.dramas[0].CoverURL = embyFixtureCover + data := embySyntheticPoster(t) + var imageRequests atomic.Int32 + app.downloader.client.Transport = embyPosterTestTransport(func(request *http.Request) (*http.Response, error) { + if request.URL.Host != "p3-shortvideo.byteimg.com" { + t.Error("unexpected external request") + return nil, errors.New("external requests forbidden") + } + imageRequests.Add(1) + return &http.Response{StatusCode: 200, Header: http.Header{}, Body: io.NopCloser(bytes.NewReader(data)), Request: request}, nil + }) + library := httptest.NewServer(app.routes()) + defer library.Close() + key, err := app.embySigningKey(true) + if err != nil { + t.Fatal(err) + } + folder := embyFolderName(app.dramas[0]) + var ready, failDownload, delayConfirmation atomic.Bool + var scans, uploads atomic.Int32 + var imageTag atomic.Value + imageTag.Store("") + item := func() embyMediaItem { + return embyMediaItem{ID: "series-1", Type: "Series", Path: "/container/shared/" + folder, + ProviderIDs: map[string]string{"juku": embyMetadataID(key, historyFixtureDramaID)}, ImageTags: map[string]string{"Primary": imageTag.Load().(string)}} + } + emby := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.Header.Get("X-Emby-Token") != "private-api-key" || strings.Contains(request.URL.RawQuery, "private-api-key") { + t.Error("API key missing from header or exposed in URL") + } + switch request.URL.Path { + case "/emby/Library/Refresh": + scans.Add(1) + writer.WriteHeader(204) + case "/emby/Items": + result := embyItemList{} + if ready.Load() && !(request.URL.Query().Get("Ids") != "" && delayConfirmation.Load()) { + result.Items, result.TotalRecordCount = []embyMediaItem{item()}, 1 + } + json.NewEncoder(writer).Encode(result) + case "/emby/Items/series-1/RemoteImages/Download": + uploads.Add(1) + if failDownload.Load() { + http.Error(writer, "sensitive upstream body private-api-key", 503) + return + } + if request.Method != http.MethodPost || request.URL.Query().Get("Type") != "Primary" || request.Header.Get("Content-Type") != "application/json" { + t.Error("invalid poster download request") + } + address := request.URL.Query().Get("ImageUrl") + if !strings.HasPrefix(address, library.URL+"/api/emby/cover?") { + t.Error("Emby was sent a source URL instead of the signed application URL") + writer.WriteHeader(400) + return + } + response, err := http.Get(address) + if err != nil { + t.Error(err) + writer.WriteHeader(502) + return + } + body, err := io.ReadAll(response.Body) + response.Body.Close() + if err != nil || response.StatusCode != 200 || !bytes.Equal(body, data) { + t.Error("Emby could not fetch the synthetic cover without cookies", response.StatusCode) + writer.WriteHeader(502) + return + } + imageTag.Store("managed-" + request.URL.Query().Get("ImageUrl")) + writer.WriteHeader(204) + default: + t.Error("unexpected Emby request", request.URL.Path) + writer.WriteHeader(404) + } + })) + defer emby.Close() + manager.document.OwnerID = app.browserViewers().accountStore().state.Accounts["admin"].ID + manager.document.Settings = embySyncSettings{Enabled: true, Following: true, IntervalMinutes: 60, OutputDir: t.TempDir(), BaseURL: library.URL, ServerURL: emby.URL + "/emby", APIKey: "private-api-key"} + forcePoster := func() { + entry := manager.document.Entries[historyFixtureDramaID] + entry.Poster.CheckedAt, entry.Poster.RetryAt = time.Time{}, time.Time{} + manager.document.Entries[historyFixtureDramaID] = entry + } + run := func() embySyncEntry { + manager.runOnce(context.Background()) + return manager.document.Entries[historyFixtureDramaID] + } + first := run() + if first.Episodes != 3 || first.Error != "" || first.Poster.Status != "waiting" || scans.Load() != 1 || uploads.Load() != 0 || + manager.document.PostersPending != 1 || time.Until(manager.document.NextRunAt) > 2*time.Minute { + t.Fatal("first scan did not queue a durable poster retry", first.Poster.Status, manager.document.Error) + } + file := filepath.Join(manager.document.Settings.OutputDir, first.Folder, "Season 01", "S01E001.strm") + before, _ := os.Stat(file) + ready.Store(true) + forcePoster() + second := run() + after, _ := os.Stat(file) + if second.Poster.Status != "synced" || uploads.Load() != 1 || imageRequests.Load() != 1 || manager.document.WrittenFiles != 0 || + scans.Load() != 1 || !before.ModTime().Equal(after.ModTime()) { + t.Fatal("poster retry rewrote/refetched episodes or did not synchronize", second.Poster.Status, manager.document.Error) + } + forcePoster() + run() + if uploads.Load() != 1 { + t.Fatal("unchanged artwork was downloaded again") + } + restarted := &UIApp{cfg: app.cfg, downloader: app.downloader, dramas: app.dramas} + manager = restarted.embySyncer() + if manager.loadErr != nil || manager.document.Entries[historyFixtureDramaID].Poster.ImageTag == "" { + t.Fatal("restart lost managed artwork state", manager.loadErr) + } + forcePoster() + run() + if uploads.Load() != 1 || scans.Load() != 1 { + t.Fatal("restart duplicated existing artwork or files") + } + setCover := func(address string) { + app.mu.Lock() + app.dramas = append([]Drama(nil), app.dramas...) + app.dramas[0].CoverURL = address + dramas := append([]Drama(nil), app.dramas...) + app.mu.Unlock() + restarted.mu.Lock() + restarted.dramas = dramas + restarted.mu.Unlock() + } + setCover(embyFixtureCover + "&revision=2") + failDownload.Store(true) + failed := run() + if failed.Error != "" || failed.Poster.Status != "failed" || manager.document.PostersFailed != 1 || + !strings.Contains(failed.Poster.Error, "HTTP 503") || time.Until(failed.Poster.RetryAt) < 30*time.Second { + t.Fatal("poster failure affected episode import or lost retry state", failed.Poster.Status, manager.document.Error) + } + serialized, _ := json.Marshal(manager.view()) + if bytes.Contains(serialized, []byte("private-api-key")) || bytes.Contains(serialized, []byte("private-cover-token")) { + t.Fatal("sync status exposed private URLs or credentials") + } + count := uploads.Load() + run() + if uploads.Load() != count { + t.Fatal("failed artwork ignored backoff") + } + failDownload.Store(false) + delayConfirmation.Store(true) + forcePoster() + unconfirmed := run() + if unconfirmed.Poster.SyncedSource == "" || unconfirmed.Poster.Status != "failed" { + t.Fatal("successful submission with delayed confirmation was lost") + } + count = uploads.Load() + delayConfirmation.Store(false) + imageTag.Store("custom-during-confirmation-failure") + forcePoster() + if confirmed := run(); confirmed.Poster.Status != "existing" || confirmed.Poster.ImageTag != "" || uploads.Load() != count { + t.Fatal("unconfirmed image ownership overwrote or adopted custom artwork") + } + imageTag.Store("user-custom-cover") + forcePoster() + if custom := run(); custom.Poster.Status != "existing" || uploads.Load() != count { + t.Fatal("custom artwork was overwritten") + } + setCover(embyFixtureCover + "&revision=3") + if custom := run(); custom.Poster.Status != "existing" || uploads.Load() != count { + t.Fatal("metadata update overwrote custom artwork") + } + imageTag.Store("") + forcePoster() + if repaired := run(); repaired.Poster.Status != "synced" || uploads.Load() != count+1 { + t.Fatal("missing artwork was not repaired") + } + err = filepath.WalkDir(manager.document.Settings.OutputDir, func(path string, entry os.DirEntry, err error) error { + if err == nil && !entry.IsDir() && !strings.HasSuffix(path, ".strm") && !strings.HasSuffix(path, ".nfo") && entry.Name() != ".juku-emby.json" && entry.Name() != "poster.png" { + t.Error("synchronization wrote an unexpected file into the export directory", entry.Name()) + } + return err + }) + if err != nil { + t.Fatal(err) + } + local := filepath.Join(manager.document.Settings.OutputDir, folder, "poster.png") + if body, err := os.ReadFile(local); err != nil || !bytes.Equal(body, data) { + t.Fatal("automatic sync did not preserve the local poster", err) + } +} + +func TestEmbyPosterAPIDoesNotForwardCredentialsOnRedirect(t *testing.T) { + var forwarded atomic.Int32 + target := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + forwarded.Add(1) + writer.WriteHeader(200) + })) + defer target.Close() + redirect := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + http.Redirect(writer, request, target.URL, http.StatusFound) + })) + defer redirect.Close() + client := newEmbyAPIClient(embySyncSettings{ServerURL: redirect.URL, APIKey: "private-api-key"}) + defer client.close() + if _, err := client.series(context.Background()); err == nil || !strings.Contains(err.Error(), "HTTP 302") || forwarded.Load() != 0 { + t.Fatal("Emby request followed a redirect with credentials", err) + } +} diff --git a/internal/app/emby_sync.go b/internal/app/emby_sync.go new file mode 100644 index 0000000..c4f4f1e --- /dev/null +++ b/internal/app/emby_sync.go @@ -0,0 +1,551 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" +) + +type embySyncSettings struct { + GroupBySource bool `json:"-"` + Enabled bool `json:"enabled"` + OutputDir string `json:"outputDir"` + BaseURL string `json:"baseUrl"` + IntervalMinutes int `json:"intervalMinutes"` + Following bool `json:"following"` + Downloads bool `json:"downloads"` + ServerURL string `json:"serverUrl"` + APIKey string `json:"apiKey,omitempty"` +} + +type embySyncEntry struct { + Merged bool `json:"merged,omitempty"` + Folder string `json:"folder,omitempty"` + Title string `json:"title"` + Episodes int `json:"episodes"` + CheckedAt time.Time `json:"checkedAt"` + SyncedAt time.Time `json:"syncedAt"` + Error string `json:"error,omitempty"` + Poster embyPosterState `json:"poster"` +} + +type embySyncDocument struct { + Version int `json:"version"` + Settings embySyncSettings `json:"settings"` + OwnerID string `json:"ownerId"` + OwnerName string `json:"owner"` + Entries map[string]embySyncEntry `json:"entries"` + LastStartedAt time.Time `json:"lastStartedAt"` + LastFinishedAt time.Time `json:"lastFinishedAt"` + NextRunAt time.Time `json:"nextRunAt"` + WrittenFiles int `json:"writtenFiles"` + Succeeded int `json:"succeeded"` + Failed int `json:"failed"` + Pending int `json:"pending"` + Error string `json:"error,omitempty"` + RefreshPending bool `json:"refreshPending"` + PostersSynced int `json:"postersSynced"` + PostersPending int `json:"postersPending"` + PostersFailed int `json:"postersFailed"` + PostersUpdated int `json:"postersUpdated"` + PosterChecked int `json:"-"` +} + +type embySyncManager struct { + app *UIApp + path string + mu sync.Mutex + document embySyncDocument + loadErr error + running bool + generation uint64 + wake chan struct{} + cancel context.CancelFunc + runCancel context.CancelFunc + done chan struct{} + closed bool + dirty map[string]bool +} + +func (app *UIApp) embySyncer() *embySyncManager { + app.embySyncOnce.Do(func() { + manager := &embySyncManager{app: app, path: filepath.Join(app.cfg.dataDirectory(), "emby-sync.json"), wake: make(chan struct{}, 1)} + manager.document = embySyncDocument{Version: 1, Settings: embySyncSettings{ + OutputDir: filepath.Join(app.cfg.OutputDir, "Emby"), IntervalMinutes: 60, Following: true, Downloads: true, + }, Entries: make(map[string]embySyncEntry)} + var stored embySyncDocument + err := readEmbyJSON(manager.path, &stored, 8<<20) + if err == nil { + if stored.Version != 1 || len(stored.Entries) > 10000 { + err = errors.New("Emby 同步记录格式无效") + } + if err == nil { + err = stored.Settings.normalize() + } + if err == nil { + if stored.Entries == nil { + stored.Entries = make(map[string]embySyncEntry) + } + for id, entry := range stored.Entries { + canonical, _, valid := playbackHistoryIdentity(id) + if !valid || canonical != id || !validEmbyFolder(entry.Folder) || entry.Episodes < 0 || entry.Episodes > 2000 { + err = errors.New("Emby 同步记录含无效的剧集或路径") + break + } + } + } + if err == nil { + manager.document = stored + } + } + if err != nil && !errors.Is(err, os.ErrNotExist) { + manager.loadErr = err + } + app.embySync = manager + }) + return app.embySync +} + +func (settings *embySyncSettings) normalize() error { + if settings.IntervalMinutes < 5 || settings.IntervalMinutes > 1440 { + return errors.New("同步间隔范围为 5–1440 分钟") + } + if settings.Enabled && !settings.Following && !settings.Downloads { + return errors.New("请至少选择一种同步范围") + } + if len(settings.OutputDir) > 2048 || strings.TrimSpace(settings.OutputDir) == "" || strings.ContainsAny(settings.OutputDir, "\x00\r\n") { + return errors.New("请输入有效的 Emby 输出目录") + } + directory, err := normalizedOutputDirectory(settings.OutputDir) + if err == nil { + directory, err = filepath.Abs(directory) + } + if err != nil { + return errors.New("Emby 输出目录无效") + } + settings.OutputDir = directory + if settings.Enabled || settings.BaseURL != "" { + settings.BaseURL, err = embyBaseURL(settings.BaseURL) + if err != nil { + return err + } + } + if settings.ServerURL != "" { + settings.ServerURL, err = embyBaseURL(settings.ServerURL) + if err != nil { + return errors.New("请输入有效的 Emby 服务器 HTTP/HTTPS 地址") + } + if settings.Enabled && settings.APIKey == "" { + return errors.New("填写 Emby 服务器后，需要同时填写 API Key 才能通知扫描和同步海报") + } + } + if len(settings.APIKey) > 2048 || strings.ContainsAny(settings.APIKey, "\x00\r\n") { + return errors.New("Emby API Key 格式无效") + } + return nil +} + +func (manager *embySyncManager) start() { + manager.mu.Lock() + defer manager.mu.Unlock() + if manager.cancel != nil || manager.closed { + return + } + ctx, cancel := context.WithCancel(context.Background()) + manager.cancel, manager.done = cancel, make(chan struct{}) + manager.notify() + go manager.loop(ctx) +} + +func (manager *embySyncManager) stop() { + manager.mu.Lock() + manager.closed = true + cancel, done := manager.cancel, manager.done + manager.mu.Unlock() + if cancel != nil { + cancel() + <-done + } +} + +func (manager *embySyncManager) notify() { + select { + case manager.wake <- struct{}{}: + default: + } +} + +func (app *UIApp) notifyEmbySync(ids ...string) { + manager := app.embySyncer() + manager.mu.Lock() + enabled := manager.document.Settings.Enabled + if len(ids) > 0 { + if manager.dirty == nil { + manager.dirty = make(map[string]bool) + } + for _, id := range ids { + manager.dirty[id] = true + } + } + manager.mu.Unlock() + if enabled { + manager.notify() + } +} + +func (manager *embySyncManager) loop(ctx context.Context) { + defer close(manager.done) + timer := time.NewTimer(time.Minute) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-manager.wake: + case <-timer.C: + } + if ctx.Err() != nil { + return + } + manager.runOnce(ctx) + manager.mu.Lock() + delay := time.Until(manager.document.NextRunAt) + if !manager.document.Settings.Enabled || delay < time.Minute { + delay = time.Minute + } + manager.mu.Unlock() + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(delay) + } +} + +func (manager *embySyncManager) owner(id string) (accountRecord, error) { + store := manager.app.browserViewers().accountStore() + store.mu.Lock() + defer store.mu.Unlock() + if store.err != nil { + return accountRecord{}, store.err + } + for _, account := range store.state.Accounts { + if account.ID == id && account.Admin && !account.RequirePasswordChange { + return account, nil + } + } + return accountRecord{}, errors.New("启用同步的管理员账号不可用，请重新登录管理员并保存同步设置") +} + +func (manager *embySyncManager) targets(ctx context.Context, config embySyncSettings, owner accountRecord) ([]Drama, error) { + selected := map[string]Drama{} + if config.Following { + viewer := manager.app.browserViewers().acquire(owner.viewerID()) + defer viewer.release() + entries, err := viewer.followingStore().list() + if err != nil { + return nil, err + } + history, err := viewer.playbackHistory().list() + if err != nil { + return nil, err + } + ids := make([]string, 0, len(entries)+len(history)) + for _, entry := range entries { + ids = append(ids, entry.DramaID) + } + for _, entry := range history { + ids = append(ids, entry.DramaID) + } + metadata := manager.app.followingMetadata(ids) + completed := map[string]bool{} + for _, entry := range entries { + completed[entry.DramaID] = entry.Completed && (entry.KnownEpisodes == 0 || metadata[entry.DramaID].KnownEpisodes <= entry.KnownEpisodes) + if entry.Saved && !completed[entry.DramaID] { + selected[entry.DramaID] = Drama{ID: entry.DramaID, Title: entry.Title, Source: entry.Source} + } + } + for _, entry := range history { + if !completed[entry.DramaID] && !(entry.Completed && entry.Index >= entry.Total && entry.Index >= metadata[entry.DramaID].KnownEpisodes) { + selected[entry.DramaID] = Drama{ID: entry.DramaID, Title: entry.Title, Source: entry.Source} + } + } + } + app := manager.app + app.mu.Lock() + if config.Downloads { + for _, task := range app.tasks { + if task != nil && task.Status == uiStatusSuccess && !isChapterPlaceholderTask(task) && taskSourceAllowed(ctx, task.Source) { + selected[task.DramaID] = Drama{ID: task.DramaID, Title: task.DramaTitle, Source: task.Source.Chapter.Source} + } + } + for id, merged := range app.merges { + if merged != nil && merged.Status == "success" && dramaAllowed(ctx, id, sourceFromDramaID(id)) && completedPlaybackPath(&UITask{Status: uiStatusSuccess, Path: merged.OutputPath}) != "" { + selected[id] = Drama{ID: id, Title: merged.DramaTitle, Source: sourceFromDramaID(id)} + } + } + } + for _, drama := range app.dramas { + if _, exists := selected[drama.ID]; exists { + selected[drama.ID] = drama + } + } + app.mu.Unlock() + if len(selected) > 1000 { + return nil, errors.New("自动同步最多支持 1000 部，请缩小追剧或下载同步范围") + } + result := make([]Drama, 0, len(selected)) + for _, drama := range selected { + id, _, valid := playbackHistoryIdentity(drama.ID) + if valid && id == drama.ID && dramaAllowed(ctx, drama.ID, drama.Source) { + result = append(result, drama) + } + } + return result, nil +} + +func (manager *embySyncManager) runOnce(parent context.Context) { + manager.mu.Lock() + if manager.running || manager.loadErr != nil || !manager.document.Settings.Enabled || manager.closed { + manager.mu.Unlock() + return + } + previous := manager.document + document := manager.document + document.Entries = make(map[string]embySyncEntry, len(manager.document.Entries)) + for id, entry := range manager.document.Entries { + document.Entries[id] = entry + } + dirty := manager.dirty + manager.dirty = nil + for id := range dirty { + if entry, exists := document.Entries[id]; exists { + entry.CheckedAt = time.Time{} + document.Entries[id] = entry + } + } + generation := manager.generation + ctx, cancel := context.WithTimeout(parent, 10*time.Minute) + manager.running, manager.runCancel = true, cancel + manager.document.LastStartedAt = time.Now() + document.LastStartedAt = manager.document.LastStartedAt + manager.mu.Unlock() + defer cancel() + document.WrittenFiles, document.Succeeded, document.Failed, document.Pending, document.Error = 0, 0, 0, 0, "" + document.PostersUpdated, document.PosterChecked = 0, 0 + document.NextRunAt = time.Now().Add(time.Duration(document.Settings.IntervalMinutes) * time.Minute) + err := manager.synchronize(ctx, &document) + if err != nil { + document.Error = manager.safeError(err, document.Settings.APIKey) + } + document.LastFinishedAt = time.Now() + if document.Pending > 0 || document.RefreshPending || time.Until(document.NextRunAt) < time.Minute { + document.NextRunAt = time.Now().Add(time.Minute) + } + if document.Succeeded == 0 && document.Failed == 0 && document.PosterChecked == 0 && err == nil && !previous.RefreshPending { + document.LastStartedAt, document.LastFinishedAt = previous.LastStartedAt, previous.LastFinishedAt + document.Succeeded, document.Failed, document.WrittenFiles, document.Error = previous.Succeeded, previous.Failed, previous.WrittenFiles, previous.Error + } + manager.mu.Lock() + manager.running, manager.runCancel = false, nil + if generation == manager.generation { + if err = writeEmbyJSON(manager.path, document, 0600); err != nil { + document.Error = "同步状态保存失败：" + manager.safeError(err, document.Settings.APIKey) + } + manager.document = document + } else { + if manager.dirty == nil { + manager.dirty = make(map[string]bool) + } + for id := range dirty { + manager.dirty[id] = true + } + } + if len(manager.dirty) > 0 { + manager.notify() + } + manager.mu.Unlock() +} + +func (manager *embySyncManager) safeError(err error, key string) string { + if err == nil { + return "" + } + text := err.Error() + if key != "" { + text = strings.ReplaceAll(text, key, "[redacted]") + } + return truncate(manager.app.redactError(errors.New(text)), 1024) +} + +func (manager *embySyncManager) synchronize(ctx context.Context, document *embySyncDocument) error { + manager.app.mu.Lock() + document.Settings.GroupBySource = manager.app.cfg.GroupBySource + manager.app.mu.Unlock() + owner, err := manager.owner(document.OwnerID) + if err != nil { + return err + } + ctx = context.WithValue(withSourceScope(ctx, owner), backgroundCatalogKey{}, true) + targets, err := manager.targets(ctx, document.Settings, owner) + if err != nil { + return err + } + interval := time.Duration(document.Settings.IntervalMinutes) * time.Minute + defer func() { + next := time.Now().Add(interval) + for _, drama := range targets { + if due := document.Entries[drama.ID].CheckedAt.Add(interval); due.Before(next) { + next = due + } + if due := embyPosterDue(document.Entries[drama.ID].Poster, interval); !due.IsZero() && due.Before(next) { + next = due + } + } + document.NextRunAt = next + document.countPosters(targets) + }() + var key []byte + if len(targets) > 0 { + key, err = manager.app.embySigningKey(true) + if err != nil { + return err + } + } + due := make([]Drama, 0, len(targets)) + for _, drama := range targets { + if time.Since(document.Entries[drama.ID].CheckedAt) >= interval { + due = append(due, drama) + } + } + sort.Slice(due, func(i, j int) bool { + left, right := document.Entries[due[i].ID].CheckedAt, document.Entries[due[j].ID].CheckedAt + if left.Equal(right) { + return due[i].ID < due[j].ID + } + return left.Before(right) + }) + document.Pending = len(due) + if len(due) > 50 { + due = due[:50] + } + if len(due) > 0 { + for _, drama := range due { + if err := ctx.Err(); err != nil { + return err + } + entry := document.Entries[drama.ID] + entry.Title, entry.CheckedAt = drama.DisplayTitle(), time.Now() + dramaCtx, cancel := context.WithTimeout(ctx, 90*time.Second) + merged := manager.app.embyMergedRecord(drama.ID) + title, chapters, fetchErr := manager.app.embyExportChapters(dramaCtx, drama.ID, merged) + if fetchErr == nil { + drama.Title = firstNonEmpty(title, drama.DisplayTitle()) + var written, episodes int + entry.Folder, written, episodes, fetchErr = syncEmbyDramaFiles(dramaCtx, document.Settings, drama, chapters, entry.Folder, key, owner.ID, merged) + document.WrittenFiles += written + if written > 0 { + document.RefreshPending = document.Settings.ServerURL != "" + } + if fetchErr == nil { + entry.Title, entry.Episodes, entry.SyncedAt = drama.DisplayTitle(), episodes, time.Now() + entry.Merged = entry.Merged || merged != nil + } + } + cancel() + entry.Error = manager.safeError(fetchErr, document.Settings.APIKey) + if fetchErr == nil { + document.Succeeded++ + } else { + document.Failed++ + } + document.Entries[drama.ID] = entry + document.Pending-- + } + } + if document.WrittenFiles > 0 { + document.RefreshPending = document.Settings.ServerURL != "" + } + manager.preparePosters(document, targets, key) + manager.synchronizeLocalPosters(ctx, document, targets) + if document.RefreshPending && document.Settings.ServerURL != "" { + if err := notifyEmbyLibrary(ctx, document.Settings); err != nil { + return fmt.Errorf("文件已保留，通知 Emby 扫描失败：%w", err) + } + document.RefreshPending = false + } + manager.synchronizePosters(ctx, document, targets, key) + document.countPosters(targets) + if document.Failed > 0 { + return fmt.Errorf("%d 部同步失败，已导入内容保留，可查看下方详情或稍后重试", document.Failed) + } + return document.posterError() +} + +func notifyEmbyLibrary(ctx context.Context, settings embySyncSettings) error { + request, err := http.NewRequestWithContext(ctx, http.MethodPost, settings.ServerURL+"/Library/Refresh", nil) + if err != nil { + return err + } + request.Header.Set("X-Emby-Token", settings.APIKey) + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.Proxy = nil + defer transport.CloseIdleConnections() + client := &http.Client{Transport: transport, Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + response, err := client.Do(request) + if err != nil { + return publicError(err) + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("Emby HTTP %d", response.StatusCode) + } + _, err = io.Copy(io.Discard, io.LimitReader(response.Body, 4096)) + return err +} + +func readEmbyJSON(path string, target any, limit int64) error { + file, err := os.Open(path) + if err != nil { + return err + } + defer file.Close() + decoder := json.NewDecoder(io.LimitReader(file, limit+1)) + if err := decoder.Decode(target); err != nil { + return err + } + if decoder.Decode(new(any)) != io.EOF { + return errors.New("Emby 配置或清单格式无效") + } + info, err := file.Stat() + if err != nil { + return err + } + if info.Size() > limit { + return errors.New("Emby 配置或清单过大") + } + return nil +} + +func writeEmbyJSON(path string, value any, mode os.FileMode) error { + body, err := json.MarshalIndent(value, "", " ") + if err != nil { + return err + } + if err = os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return err + } + _, err = writeEmbyFile(path, append(body, '\n'), mode) + return err +} diff --git a/internal/app/emby_sync_files.go b/internal/app/emby_sync_files.go new file mode 100644 index 0000000..bf96376 --- /dev/null +++ b/internal/app/emby_sync_files.go @@ -0,0 +1,285 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" +) + +type embyFolderManifest struct { + Version int `json:"version"` + DramaID string `json:"dramaId"` + Chapters []embyChapter `json:"chapters"` + Merged *embyMergedRecord `json:"merged,omitempty"` + PosterFile string `json:"posterFile,omitempty"` + PosterHash string `json:"posterHash,omitempty"` + PosterSource string `json:"posterSource,omitempty"` +} + +func validEmbyFolder(folder string) bool { + if folder == "" { + return true + } + parts := strings.Split(folder, "/") + if len(parts) > 2 || len(parts) == 2 && !isSourceFolder(parts[0]) { + return false + } + for _, part := range parts { + if part == "" || len(part) > 240 || part == "." || part == ".." || strings.ContainsAny(part, "\\:\x00\r\n") || filepath.Base(part) != part { + return false + } + } + return true +} + +func isSourceFolder(folder string) bool { + if folder == "其他来源" { + return true + } + for _, choice := range accountSourceChoices { + if folder == safeFilename(choice.Name) { + return true + } + } + return false +} + +func embyDirectoryParent(root, folder string, create bool) (string, error) { + if !validEmbyFolder(folder) || folder == "" { + return "", errors.New("Emby 剧集目录无效") + } + parts := strings.Split(folder, "/") + if len(parts) == 1 { + return root, nil + } + parent := filepath.Join(root, parts[0]) + if create { + return parent, ensureEmbyDirectory(parent) + } + info, err := os.Lstat(parent) + if err != nil { + return "", err + } + if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { + return "", errors.New("Emby 站源目录不是普通文件夹") + } + return parent, nil +} + +func findManagedEmbyFolder(root string, drama Drama) (string, error) { + name := embyFolderName(drama) + suffix := name[strings.LastIndex(name, " ["):] + var found string + var scan func(string, string) error + scan = func(directory, prefix string) error { + entries, err := os.ReadDir(directory) + if err != nil { + return err + } + for _, entry := range entries { + if !entry.IsDir() || entry.Type()&os.ModeSymlink != 0 { + continue + } + if prefix == "" && isSourceFolder(entry.Name()) { + if err := scan(filepath.Join(directory, entry.Name()), entry.Name()+"/"); err != nil { + return err + } + } + folder := prefix + entry.Name() + if !strings.HasSuffix(entry.Name(), suffix) || !validEmbyFolder(folder) { + continue + } + var manifest embyFolderManifest + if err := readEmbyJSON(filepath.Join(directory, entry.Name(), ".juku-emby.json"), &manifest, 8<<20); err != nil || manifest.Version != 1 || manifest.DramaID != drama.ID { + continue + } + if found != "" && found != folder { + return errors.New("发现本剧的多个 Emby 目录，已停止生成重复媒体") + } + found = folder + } + return nil + } + err := scan(root, "") + return found, err +} + +func ensureEmbyDirectory(path string) error { + info, err := os.Lstat(path) + if errors.Is(err, os.ErrNotExist) { + return os.Mkdir(path, 0755) + } + if err != nil { + return err + } + if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { + return errors.New("Emby 生成目录不是普通文件夹，已停止写入") + } + return nil +} + +func syncEmbyDramaFiles(ctx context.Context, settings embySyncSettings, drama Drama, chapters []Chapter, folder string, key []byte, owner string, merged ...*embyMergedRecord) (string, int, int, error) { + if !validEmbyFolder(folder) { + return folder, 0, 0, errors.New("Emby 剧集目录无效") + } + if err := os.MkdirAll(settings.OutputDir, 0755); err != nil { + return folder, 0, 0, err + } + root, err := filepath.EvalSymlinks(settings.OutputDir) + if err != nil { + return folder, 0, 0, err + } + if folder == "" { + folder, err = findManagedEmbyFolder(root, drama) + if err != nil { + return folder, 0, 0, err + } + if folder == "" { + folder = embyFolderName(drama) + if settings.GroupBySource { + folder = dramaSourceFolder(drama) + "/" + folder + } + } + } + parent, err := embyDirectoryParent(root, folder, true) + if err != nil { + return folder, 0, 0, err + } + target := filepath.Join(root, folder) + work := target + var previous []embyChapter + manifest := embyFolderManifest{Version: 1, DramaID: drama.ID} + staged := false + info, err := os.Lstat(target) + if errors.Is(err, os.ErrNotExist) { + work, err = os.MkdirTemp(parent, ".juku-emby-") + if err != nil { + return folder, 0, 0, err + } + staged = true + defer os.RemoveAll(work) + if err = os.Chmod(work, 0755); err != nil { + return folder, 0, 0, err + } + } else { + if err != nil { + return folder, 0, 0, err + } + if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { + return folder, 0, 0, errors.New("Emby 剧集目录不是普通文件夹") + } + manifestPath := filepath.Join(work, ".juku-emby.json") + info, err := os.Lstat(manifestPath) + if err != nil || !info.Mode().IsRegular() { + return folder, 0, 0, errors.New("同名目录不属于自动同步，原内容已保留；请使用单独的 Emby 输出目录") + } + if err = readEmbyJSON(manifestPath, &manifest, 8<<20); err != nil { + return folder, 0, 0, err + } + if manifest.Version != 1 || manifest.DramaID != drama.ID { + return folder, 0, 0, errors.New("Emby 目录所属剧集不匹配，原内容已保留") + } + previous = manifest.Chapters + } + if len(merged) > 0 && merged[0] != nil { + manifest.Merged = merged[0] + } + if manifest.Merged != nil && !manifest.Merged.valid() { + return folder, 0, 0, errors.New("Emby 合并版记录无效，原文件已保留") + } + if len(chapters) == 0 && manifest.Merged == nil { + return folder, 0, 0, errors.New("站源未返回分集，已有 Emby 文件已保留，稍后将重试") + } + var records []embyChapter + if len(chapters) > 0 || len(previous) > 0 || manifest.Merged == nil { + records, err = embyChapterRecords(drama.ID, chapters, previous) + } + if err != nil { + return folder, 0, 0, err + } + for _, season := range []struct { + name string + enabled bool + }{{"Season 01", len(records) > 0}, {"Season 00", manifest.Merged != nil}} { + if season.enabled { + if err = ensureEmbyDirectory(filepath.Join(work, season.name)); err != nil { + return folder, 0, 0, err + } + } + } + written := 0 + err = writeEmbyFiles(drama, records, settings.BaseURL, key, owner, func(name string, body []byte) error { + if err := ctx.Err(); err != nil { + return err + } + changed, err := writeEmbyFile(filepath.Join(work, filepath.FromSlash(name)), body, 0644) + if changed { + written++ + } + return err + }, manifest.Merged) + if err == nil { + manifest.Chapters = records + body, marshalErr := json.MarshalIndent(manifest, "", " ") + err = marshalErr + if err == nil { + _, err = writeEmbyFile(filepath.Join(work, ".juku-emby.json"), append(body, '\n'), 0600) + } + } + if err == nil { + err = ctx.Err() + } + if staged { + if err == nil { + err = os.Rename(work, target) + } + if err != nil { + written = 0 + } + } + return folder, written, len(records), err +} + +func writeEmbyFile(path string, body []byte, mode os.FileMode) (bool, error) { + info, err := os.Lstat(path) + if err == nil { + if !info.Mode().IsRegular() || info.Size() > 8<<20 { + return false, errors.New("Emby 目标文件不是普通文件或文件过大，已停止写入") + } + old, err := os.ReadFile(path) + if err != nil { + return false, err + } + if bytes.Equal(old, body) { + return false, nil + } + } else if !errors.Is(err, os.ErrNotExist) { + return false, err + } + file, err := os.CreateTemp(filepath.Dir(path), ".juku-emby-write-") + if err != nil { + return false, err + } + defer os.Remove(file.Name()) + if err = file.Chmod(mode); err == nil { + _, err = file.Write(body) + } + if err == nil { + err = file.Sync() + } + if closeErr := file.Close(); err == nil { + err = closeErr + } + if err != nil { + return false, err + } + if err = os.Rename(file.Name(), path); err != nil { + return false, fmt.Errorf("Emby 文件替换失败：%w", err) + } + return true, nil +} diff --git a/internal/app/emby_sync_posters.go b/internal/app/emby_sync_posters.go new file mode 100644 index 0000000..89fc30d --- /dev/null +++ b/internal/app/emby_sync_posters.go @@ -0,0 +1,247 @@ +package app + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "sort" + "time" +) + +type embyPosterState struct { + LocalFile string `json:"localFile,omitempty"` + LocalSource string `json:"localSource,omitempty"` + LocalError string `json:"localError,omitempty"` + LocalCheckedAt time.Time `json:"localCheckedAt"` + LocalRetryAt time.Time `json:"localRetryAt"` + SourceURL string `json:"sourceUrl,omitempty"` + Destination string `json:"destination,omitempty"` + ItemID string `json:"itemId,omitempty"` + ImageTag string `json:"imageTag,omitempty"` + SyncedSource string `json:"syncedSource,omitempty"` + Status string `json:"status,omitempty"` + Error string `json:"error,omitempty"` + CheckedAt time.Time `json:"checkedAt"` + SyncedAt time.Time `json:"syncedAt"` + RetryAt time.Time `json:"retryAt"` + Failures int `json:"failures,omitempty"` +} + +func embyPosterDestination(settings embySyncSettings, key []byte) string { + digest := sha256.Sum256([]byte(settings.ServerURL + "\x00" + settings.OutputDir + "\x00" + embyMetadataID(key, "instance"))) + return hex.EncodeToString(digest[:]) +} + +func embyPosterSourceHash(remote string) string { + digest := sha256.Sum256([]byte(remote)) + return hex.EncodeToString(digest[:]) +} + +func embyRemotePosterDue(poster embyPosterState, interval time.Duration) time.Time { + if poster.Status == "missing" || poster.Status == "unconfigured" || poster.Status == "local" || poster.SourceURL == "" { + return time.Time{} + } + due := poster.CheckedAt.Add(interval) + if !poster.RetryAt.IsZero() { + due = poster.RetryAt + } + if poster.CheckedAt.IsZero() { + due = time.Now() + } + return due +} + +func embyLocalPosterDue(poster embyPosterState, interval time.Duration) time.Time { + if poster.SourceURL == "" { + return time.Time{} + } + if !poster.LocalRetryAt.IsZero() { + return poster.LocalRetryAt + } + if poster.LocalCheckedAt.IsZero() || poster.LocalFile == "" || poster.LocalSource != embyPosterSourceHash(poster.SourceURL) { + return time.Now() + } + return poster.LocalCheckedAt.Add(interval) +} + +func embyPosterDue(poster embyPosterState, interval time.Duration) time.Time { + due := embyRemotePosterDue(poster, interval) + if local := embyLocalPosterDue(poster, interval); !local.IsZero() && (due.IsZero() || local.Before(due)) { + due = local + } + return due +} + +func (manager *embySyncManager) preparePosters(document *embySyncDocument, targets []Drama, key []byte) { + destination := embyPosterDestination(document.Settings, key) + for _, drama := range targets { + entry := document.Entries[drama.ID] + if entry.Episodes == 0 && !entry.Merged { + continue + } + poster := entry.Poster + if poster.Destination != destination { + poster = embyPosterState{SourceURL: poster.SourceURL, Destination: destination} + } + remote := embyDramaCover(drama) + if remote == "" { + remote = embyDramaCover(Drama{CoverURL: poster.SourceURL}) + } + if poster.SourceURL != remote { + poster.SourceURL = remote + poster.CheckedAt, poster.RetryAt, poster.Failures = time.Time{}, time.Time{}, 0 + poster.Status, poster.Error = "pending", "" + poster.LocalCheckedAt, poster.LocalRetryAt = time.Time{}, time.Time{} + } + if remote == "" { + poster.Status, poster.Error = "missing", "" + } else if document.Settings.ServerURL == "" { + if poster.LocalFile != "" && poster.LocalSource == embyPosterSourceHash(remote) && poster.LocalError == "" { + poster.Status, poster.Error = "local", "" + } else if poster.LocalError != "" { + poster.Status, poster.Error = "failed", poster.LocalError + } else { + poster.Status, poster.Error = "pending", "" + } + } else if poster.Status == "" || poster.Status == "missing" || poster.Status == "unconfigured" { + poster.Status = "pending" + poster.CheckedAt, poster.RetryAt = time.Time{}, time.Time{} + } + entry.Poster = poster + document.Entries[drama.ID] = entry + } +} + +func (manager *embySyncManager) postponePoster(poster *embyPosterState, waiting bool, err error, key string) { + poster.Status = "failed" + if waiting { + poster.Status = "waiting" + } + poster.Error = manager.safeError(err, key) + poster.CheckedAt = time.Now() + if poster.Failures < 5 { + poster.Failures++ + } + delays := [...]time.Duration{time.Minute, 2 * time.Minute, 5 * time.Minute, 10 * time.Minute, 15 * time.Minute} + poster.RetryAt = time.Now().Add(delays[poster.Failures-1]) +} + +func (manager *embySyncManager) synchronizePosters(ctx context.Context, document *embySyncDocument, targets []Drama, key []byte) { + if document.Settings.ServerURL == "" { + return + } + interval := time.Duration(document.Settings.IntervalMinutes) * time.Minute + var due []string + for _, drama := range targets { + poster := document.Entries[drama.ID].Poster + if next := embyRemotePosterDue(poster, interval); !next.IsZero() && !next.After(time.Now()) { + due = append(due, drama.ID) + } + } + sort.Slice(due, func(i, j int) bool { + left, right := document.Entries[due[i]].Poster.CheckedAt, document.Entries[due[j]].Poster.CheckedAt + if left.Equal(right) { + return due[i] < due[j] + } + return left.Before(right) + }) + if len(due) > 20 { + due = due[:20] + } + if len(due) == 0 { + return + } + client := newEmbyAPIClient(document.Settings) + defer client.close() + listCtx, cancel := context.WithTimeout(ctx, time.Minute) + items, listErr := client.series(listCtx) + cancel() + for _, id := range due { + if ctx.Err() != nil { + break + } + entry := document.Entries[id] + poster := entry.Poster + document.PosterChecked++ + if listErr != nil { + manager.postponePoster(&poster, false, listErr, document.Settings.APIKey) + } else { + item, err := matchEmbySeries(items, entry.Folder, embyMetadataID(key, id)) + if err != nil { + manager.postponePoster(&poster, false, err, document.Settings.APIKey) + } else if item.ID == "" { + manager.postponePoster(&poster, true, errors.New("等待 Emby 扫描此剧；请确认输出目录已加入电视剧媒体库，并启用 NFO 元数据读取"), document.Settings.APIKey) + } else { + manager.synchronizePoster(ctx, client, document, id, &poster, item, key) + } + } + entry.Poster = poster + document.Entries[id] = entry + } +} + +func (manager *embySyncManager) synchronizePoster(ctx context.Context, client *embyAPIClient, document *embySyncDocument, id string, poster *embyPosterState, item embyMediaItem, key []byte) { + currentTag := item.ImageTags["Primary"] + sourceHash := embyPosterSourceHash(poster.SourceURL) + sameItem := poster.ItemID == item.ID + managed := sameItem && poster.ImageTag != "" && currentTag == poster.ImageTag + if currentTag != "" && !managed { + poster.Status, poster.Error = "existing", "" + poster.ItemID, poster.ImageTag, poster.SyncedSource = item.ID, "", "" + poster.SyncedAt = time.Time{} + poster.CheckedAt, poster.RetryAt, poster.Failures = time.Now(), time.Time{}, 0 + return + } + if currentTag != "" && managed && poster.SyncedSource == sourceHash { + poster.Status, poster.Error, poster.ImageTag = "synced", "", currentTag + poster.CheckedAt, poster.RetryAt, poster.Failures = time.Now(), time.Time{}, 0 + return + } + address := embyCoverURL(Drama{ID: id, CoverURL: poster.SourceURL}, document.Settings.BaseURL, key, document.OwnerID) + if address == "" { + manager.postponePoster(poster, false, errors.New("海报地址无效，请更新剧集资料后重试"), document.Settings.APIKey) + return + } + requestCtx, cancel := context.WithTimeout(ctx, 95*time.Second) + defer cancel() + tag, uploaded, err := client.downloadPoster(requestCtx, item.ID, address) + if uploaded { + poster.ItemID, poster.SyncedSource, poster.ImageTag = item.ID, sourceHash, "" + poster.SyncedAt = time.Now() + document.PostersUpdated++ + } + if err != nil { + manager.postponePoster(poster, false, err, document.Settings.APIKey) + return + } + poster.ImageTag, poster.Status, poster.Error = tag, "synced", "" + poster.CheckedAt, poster.RetryAt, poster.Failures = time.Now(), time.Time{}, 0 +} + +func (document *embySyncDocument) countPosters(targets []Drama) { + document.PostersPending, document.PostersFailed, document.PostersSynced = 0, 0, 0 + for _, drama := range targets { + poster := document.Entries[drama.ID].Poster + if poster.LocalError != "" { + document.PostersFailed++ + continue + } + switch poster.Status { + case "pending", "waiting": + document.PostersPending++ + case "failed": + document.PostersFailed++ + case "synced", "local": + document.PostersSynced++ + } + } +} + +func (document *embySyncDocument) posterError() error { + if document.PostersFailed == 0 { + return nil + } + return fmt.Errorf("%d 部海报同步失败，分集文件保留，海报会自动重试；请查看下方详情", document.PostersFailed) +} diff --git a/internal/app/emby_sync_test.go b/internal/app/emby_sync_test.go new file mode 100644 index 0000000..f37f10e --- /dev/null +++ b/internal/app/emby_sync_test.go @@ -0,0 +1,335 @@ +package app + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +func embySyncTestChapters() []Chapter { + return []Chapter{{ID: "fixture-1", Title: "第一集"}, {ID: "fixture-2", Title: "第二集"}} +} + +func TestEmbySyncFilesIdempotentStableAndNonDestructive(t *testing.T) { + settings := embySyncSettings{OutputDir: t.TempDir(), BaseURL: "http://library.test:8998"} + drama := Drama{ID: historyFixtureDramaID, Title: "合成测试 <&>"} + key := make([]byte, 32) + chapters := embySyncTestChapters() + folder, written, total, err := syncEmbyDramaFiles(context.Background(), settings, drama, chapters, "", key, "fixture-owner") + if err != nil || written != 5 || total != 2 { + t.Fatal(folder, written, total, err) + } + first := filepath.Join(settings.OutputDir, folder, "Season 01", "S01E001.strm") + before, _ := os.Stat(first) + original, _ := os.ReadFile(first) + if !strings.Contains(string(original), "account=fixture-owner") { + t.Fatal("automated link not bound to owner") + } + _, written, _, err = syncEmbyDramaFiles(context.Background(), settings, drama, chapters, folder, key, "fixture-owner") + after, _ := os.Stat(first) + if err != nil || written != 0 || !before.ModTime().Equal(after.ModTime()) { + t.Fatal("unchanged export rewrote files", written, err) + } + note := filepath.Join(settings.OutputDir, folder, "my-notes.txt") + if err = os.WriteFile(note, []byte("user text"), 0600); err != nil { + t.Fatal(err) + } + chapters = []Chapter{chapters[1], chapters[0], {ID: "fixture-3", Title: "第三集"}} + drama.Title = "资料更新后的合成剧名" + same, written, total, err := syncEmbyDramaFiles(context.Background(), settings, drama, chapters, folder, key, "fixture-owner") + if err != nil || same != folder || total != 3 { + t.Fatal("renamed/reordered series duplicated or renumbered", same, total, err) + } + current, _ := os.ReadFile(first) + if string(current) != string(original) { + t.Fatal("existing episode link changed after source order changed") + } + if body, _ := os.ReadFile(note); string(body) != "user text" { + t.Fatal("unrelated file was changed") + } + _, _, total, err = syncEmbyDramaFiles(context.Background(), settings, drama, chapters[:1], folder, key, "fixture-owner") + if err != nil || total != 3 { + t.Fatal("temporarily shorter upstream deleted previous episodes", total, err) + } + marker := filepath.Join(settings.OutputDir, folder, ".juku-emby.json") + var manifest embyFolderManifest + if err = readEmbyJSON(marker, &manifest, 8<<20); err != nil || len(manifest.Chapters) != 3 { + t.Fatal(err) + } +} + +func TestEmbySyncRefusesUnmanagedFoldersAndSymlinks(t *testing.T) { + root := t.TempDir() + settings := embySyncSettings{OutputDir: root, BaseURL: "http://library.test"} + drama := Drama{ID: historyFixtureDramaID, Title: "独立目录"} + folder := embyFolderName(drama) + if err := os.Mkdir(filepath.Join(root, folder), 0755); err != nil { + t.Fatal(err) + } + sentinel := filepath.Join(root, folder, "sentinel.txt") + os.WriteFile(sentinel, []byte("preserved"), 0600) + if _, _, _, err := syncEmbyDramaFiles(context.Background(), settings, drama, embySyncTestChapters(), "", make([]byte, 32), "owner"); err == nil { + t.Fatal("unmanaged directory adopted without permission") + } + if data, _ := os.ReadFile(sentinel); string(data) != "preserved" { + t.Fatal("unmanaged directory modified") + } + settings.OutputDir = t.TempDir() + folder, _, _, err := syncEmbyDramaFiles(context.Background(), settings, drama, embySyncTestChapters(), "", make([]byte, 32), "owner") + if err != nil { + t.Fatal(err) + } + target := filepath.Join(settings.OutputDir, folder, "Season 01", "S01E001.strm") + if err = os.Remove(target); err != nil { + t.Fatal(err) + } + if err = os.Symlink(sentinel, target); err != nil { + t.Skip("symlinks unavailable") + } + if _, _, _, err = syncEmbyDramaFiles(context.Background(), settings, drama, embySyncTestChapters(), folder, make([]byte, 32), "owner"); err == nil { + t.Fatal("managed file symlink followed") + } + if data, _ := os.ReadFile(sentinel); string(data) != "preserved" { + t.Fatal("symlink target modified") + } + for _, folder := range []string{"../escape", "..", "/absolute", "nested/path", "nested\\path"} { + if _, _, _, err := syncEmbyDramaFiles(context.Background(), settings, drama, embySyncTestChapters(), folder, make([]byte, 32), "owner"); err == nil { + t.Fatal("unsafe folder accepted", folder) + } + } +} + +func waitEmbySync(t *testing.T, manager *embySyncManager, after time.Time) embySyncDocument { + t.Helper() + deadline := time.Now().Add(8 * time.Second) + for time.Now().Before(deadline) { + manager.mu.Lock() + done := !manager.running && manager.document.LastFinishedAt.After(after) + snapshot := manager.document + manager.mu.Unlock() + if done { + return snapshot + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("automatic Emby synchronization did not finish") + return embySyncDocument{} +} + +func TestEmbySyncAutomaticFollowedImportRefreshAndRestart(t *testing.T) { + app, admin := administratorFixture(t) + manager := app.embySyncer() + t.Cleanup(manager.stop) + ordinary := newAccountTestBrowser(t, app) + ordinary.register(t, "other-viewer") + otherID := "hongguo:7000000000000000002" + app.dramas = append(app.dramas, Drama{ID: otherID, Source: sourceHongguo, Title: "另一个用户的剧"}) + viewerResultOK(t, ordinary.request(t, http.MethodPost, "/api/ui/following", map[string]any{"dramaId": otherID, "saved": true})) + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/following", map[string]any{"dramaId": historyFixtureDramaID, "saved": true})) + var scans atomic.Int32 + emby := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/emby/Library/Refresh" || r.Header.Get("X-Emby-Token") != "private-test-key" || r.URL.RawQuery != "" { + t.Error("invalid Emby refresh request") + } + scans.Add(1) + w.WriteHeader(http.StatusNoContent) + })) + defer emby.Close() + settings := map[string]any{"enabled": true, "outputDir": t.TempDir(), "baseUrl": "http://library.test:8998", "intervalMinutes": 60, "following": true, "downloads": false, "serverUrl": emby.URL + "/emby", "apiKey": "private-test-key"} + started := time.Now() + response := admin.request(t, http.MethodPost, "/api/ui/admin/emby", settings) + viewerResultOK(t, response) + if strings.Contains(response.Body.String(), "private-test-key") { + t.Fatal("API response leaked the Emby key") + } + result := waitEmbySync(t, manager, started) + if result.Error != "" || result.Succeeded != 1 || len(result.Entries) != 1 || result.Entries[historyFixtureDramaID].Episodes != 3 || scans.Load() != 1 { + t.Fatal("automatic import failed or crossed accounts", result.Error, result.Succeeded, len(result.Entries), scans.Load()) + } + output := settings["outputDir"].(string) + entry := result.Entries[historyFixtureDramaID] + strm, err := os.ReadFile(filepath.Join(output, entry.Folder, "Season 01", "S01E001.strm")) + if err != nil { + t.Fatal(err) + } + link := strings.TrimSpace(string(strm)) + parsed, _ := url.Parse(link) + if parsed.Query().Get("account") == "" { + t.Fatal("automatic export created a guest link") + } + writer := httptest.NewRecorder() + app.handleEmbyStream(writer, httptest.NewRequest(http.MethodHead, link, nil)) + if writer.Code != 200 { + t.Fatal("automatic signed link unusable", writer.Code) + } + info, err := os.Stat(manager.path) + if err != nil || info.Mode().Perm() != 0600 { + t.Fatal("sync settings not saved privately") + } + manager.stop() + restarted := &UIApp{cfg: app.cfg, downloader: app.downloader, dramas: app.dramas} + next := restarted.embySyncer() + if next.loadErr != nil || !next.document.Settings.Enabled || next.document.Settings.APIKey != "private-test-key" { + t.Fatal("restart lost configuration", next.loadErr) + } + for id, entry := range next.document.Entries { + entry.CheckedAt = time.Time{} + next.document.Entries[id] = entry + } + next.runOnce(context.Background()) + if next.document.WrittenFiles != 0 || scans.Load() != 1 || next.document.Error != "" { + t.Fatal("restart duplicated exports or scans", next.document.WrittenFiles, scans.Load(), next.document.Error) + } + if data, err := os.ReadFile(filepath.Join(output, entry.Folder, "Season 01", "S01E001.strm")); err != nil || string(data) != string(strm) { + t.Fatal("restart changed stable link") + } +} + +func TestEmbySyncAdminGateDisabledDefaultsAndValidation(t *testing.T) { + app, admin := administratorFixture(t) + manager := app.embySyncer() + t.Cleanup(manager.stop) + guest := newAccountTestBrowser(t, app) + member := newAccountTestBrowser(t, app) + member.register(t, "online-only-fixture") + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/admin/accounts/permissions", map[string]any{"username": "online-only-fixture", "onlineOnly": true})) + for _, browser := range []*accountTestBrowser{guest, member} { + for _, method := range []string{http.MethodGet, http.MethodPost} { + response := browser.request(t, method, "/api/ui/admin/emby", map[string]any{}) + if response.Code != http.StatusForbidden { + t.Fatal("non-admin accessed server export settings", response.Code) + } + } + } + response := admin.request(t, http.MethodGet, "/api/ui/admin/emby", nil) + viewerResultOK(t, response) + if manager.document.Settings.Enabled { + t.Fatal("automatic export enabled by default") + } + if _, err := os.Stat(manager.path); !os.IsNotExist(err) { + t.Fatal("reading settings created an export job") + } + input := map[string]any{"enabled": true, "outputDir": t.TempDir(), "baseUrl": "file:///private/fixture", "intervalMinutes": 60, "following": true, "downloads": false} + if result := admin.request(t, http.MethodPost, "/api/ui/admin/emby", input); result.Code != http.StatusBadRequest { + t.Fatal("invalid playback address accepted", result.Code) + } + input["baseUrl"] = "http://library.test" + input["intervalMinutes"] = 1 + if result := admin.request(t, http.MethodPost, "/api/ui/admin/emby", input); result.Code != http.StatusBadRequest { + t.Fatal("unbounded refresh frequency accepted", result.Code) + } + input["intervalMinutes"] = 60 + input["enabled"] = false + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/admin/emby", input)) + if result := admin.request(t, http.MethodPost, "/api/ui/admin/emby/sync", map[string]any{}); result.Code != http.StatusConflict { + t.Fatal("disabled sync executed", result.Code) + } +} + +func TestEmbySyncPartialFailurePreservesFilesAndRetriesScan(t *testing.T) { + app, admin := administratorFixture(t) + manager := app.embySyncer() + t.Cleanup(manager.stop) + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/following", map[string]any{"dramaId": historyFixtureDramaID, "saved": true})) + owner := app.browserViewers().accountStore().state.Accounts["admin"] + var scans atomic.Int32 + emby := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + count := scans.Add(1) + if count == 1 { + w.WriteHeader(503) + return + } + w.WriteHeader(204) + })) + defer emby.Close() + manager.document.Settings = embySyncSettings{Enabled: true, OutputDir: t.TempDir(), BaseURL: "http://library.test", IntervalMinutes: 60, Following: true, ServerURL: emby.URL, APIKey: "scan-secret"} + manager.document.OwnerID = owner.ID + manager.runOnce(context.Background()) + if !manager.document.RefreshPending || !strings.Contains(manager.document.Error, "HTTP 503") { + t.Fatal("failed scan was not retained for retry", manager.document.Error) + } + entry := manager.document.Entries[historyFixtureDramaID] + before, _ := os.ReadFile(filepath.Join(manager.document.Settings.OutputDir, entry.Folder, "Season 01", "S01E001.strm")) + manager.runOnce(context.Background()) + if manager.document.RefreshPending || scans.Load() != 2 || manager.document.WrittenFiles != 0 { + t.Fatal("scan retry rewrote files or was skipped") + } + client := app.downloader.hongguoClient() + broken := client.details["7000000000000000001"] + broken.Chapters = nil + client.details["7000000000000000001"] = broken + entry.CheckedAt = time.Time{} + manager.document.Entries[historyFixtureDramaID] = entry + manager.runOnce(context.Background()) + if manager.document.Failed != 1 || manager.document.Entries[historyFixtureDramaID].Episodes != 3 { + t.Fatal("upstream failure lost previous imported episodes") + } + after, _ := os.ReadFile(filepath.Join(manager.document.Settings.OutputDir, entry.Folder, "Season 01", "S01E001.strm")) + if string(before) != string(after) { + t.Fatal("failed source response overwrote completed export") + } + state, _ := json.Marshal(manager.view()) + if strings.Contains(string(state), "scan-secret") { + t.Fatal("status leaked API key") + } +} + +func TestEmbySyncDownloadedCollectionAndLocalPlaybackPriority(t *testing.T) { + app, _ := administratorFixture(t) + owner := app.browserViewers().accountStore().state.Accounts["admin"] + cached := app.downloader.hongguoClient().details["7000000000000000001"] + path := filepath.Join(t.TempDir(), "local-fixture.mp4") + if err := os.WriteFile(path, []byte("synthetic file identity only"), 0600); err != nil { + t.Fatal(err) + } + task := Task{DramaID: historyFixtureDramaID, DramaTitle: "下载完成的合成剧", Chapter: cached.Chapters[0], OutPath: path, Index: 1, Total: 3} + app.tasks = map[string]*UITask{"download-fixture": {ID: "download-fixture", DramaID: historyFixtureDramaID, DramaTitle: task.DramaTitle, Path: path, Status: uiStatusSuccess, Source: task}} + manager := app.embySyncer() + manager.document.OwnerID = owner.ID + manager.document.Settings = embySyncSettings{Enabled: true, OutputDir: t.TempDir(), BaseURL: "http://library.test", IntervalMinutes: 60, Downloads: true} + manager.runOnce(context.Background()) + if manager.document.Error != "" || manager.document.Entries[historyFixtureDramaID].Episodes != 3 { + t.Fatal("download collection not imported", manager.document.Error) + } + local, downloadID, err := app.embyTask(context.Background(), historyFixtureDramaID, task.Chapter.ID) + if err != nil || downloadID != "download-fixture" || local.OutPath != path { + t.Fatal("Emby did not prefer completed local file", downloadID, err) + } + firstRun := manager.document.LastFinishedAt + nextRun := manager.document.NextRunAt + manager.runOnce(context.Background()) + if manager.document.LastFinishedAt != firstRun || manager.document.NextRunAt.After(nextRun.Add(time.Second)) { + t.Fatal("unrelated notification postponed scheduled updates") + } +} + +func TestEmbySyncStartupDoesNotImmediatelyRepeatFailedScan(t *testing.T) { + app, admin := administratorFixture(t) + manager := app.embySyncer() + t.Cleanup(manager.stop) + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/following", map[string]any{"dramaId": historyFixtureDramaID, "saved": true})) + scans := make(chan struct{}, 4) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { scans <- struct{}{}; w.WriteHeader(503) })) + defer server.Close() + started := time.Now() + input := map[string]any{"enabled": true, "outputDir": t.TempDir(), "baseUrl": "http://library.test", "intervalMinutes": 60, "following": true, "downloads": false, "serverUrl": server.URL, "apiKey": "fixture-scan-key"} + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/admin/emby", input)) + result := waitEmbySync(t, manager, started) + if result.Succeeded != 1 || !result.RefreshPending || !strings.Contains(result.Error, "HTTP 503") { + t.Fatal("failed scan status not preserved", result.Error) + } + <-scans + select { + case <-scans: + t.Fatal("startup and settings wake triggered duplicate immediate scan") + case <-time.After(500 * time.Millisecond): + } +} diff --git a/internal/app/ffmpeg_install.go b/internal/app/ffmpeg_install.go new file mode 100644 index 0000000..ffeeb9a --- /dev/null +++ b/internal/app/ffmpeg_install.go @@ -0,0 +1,406 @@ +package app + +import ( + "compress/gzip" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "time" +) + +const ffmpegReleaseURL = "https://github.com/eugeneware/ffmpeg-static/releases/download/b6.1.1/" + +type ffmpegPackage struct { + platform string + archiveSize int64 + binarySize int64 + archiveSHA256 string + binarySHA256 string + licenseSHA256 string + readmeSHA256 string +} + +var ffmpegPackages = map[string]ffmpegPackage{ + "darwin/amd64": {"darwin-x64", 25296431, 78862176, + "929b375c1182d956c51f7ac25e0b2b0411fb01f6f407aa15c9758efeb4242106", "ebdddc936f61e14049a2d4b549a412b8a40deeff6540e58a9f2a2da9e6b18894", + "2e1d16c72fd74e12063776371da757322f8b77589386532f4fd8634bde7de1af", "e88a0325f8e5b75210355e37341824f074d3cd82def2125be54c914b62848a36"}, + "darwin/arm64": {"darwin-arm64", 19246198, 45568216, + "8923876afa8db5585022d7860ec7e589af192f441c56793971276d450ed3bbfa", "a90e3db6a3fd35f6074b013f948b1aa45b31c6375489d39e572bea3f18336584", + "cb48bf09a11f5fb576cddb0431c8f5ed0a60157a9ec942adffc13907cbe083f2", "05ba4b92c96605434b1aaae3eedf5a2c280c9607bf78ffca9a5b536d9af2dc6a"}, + "linux/amd64": {"linux-x64", 29354986, 79826272, + "bfe8a8fc511530457b528c48d77b5737527b504a3797a9bc4866aeca69c2dffa", "e7e7fb30477f717e6f55f9180a70386c62677ef8a4d4d1a5d948f4098aa3eb99", + "8ceb4b9ee5adedde47b31e975c1d90c73ad27b6b165a1dcd80c7c545eb65b903", "72f4b1b06d419d22ace6e7cc75f06826f90737345aa0b1736158929f4aacc537"}, + "windows/amd64": {"win32-x64", 29581307, 82797568, + "8883a3dffbd0a16cf4ef95206ea05283f78908dbfb118f73c83f4951dcc06d77", "04e1307997530f9cf2fe35cba2ca7e8875ca91da02f89d6c7243df819c94ad00", + "8ceb4b9ee5adedde47b31e975c1d90c73ad27b6b165a1dcd80c7c545eb65b903", "a636a7183c58006351acbaf35303c0ed85c6e1320fd4e80de453ba6157de6311"}, +} + +type ffmpegInstallState struct { + Status string `json:"status"` + Detail string `json:"detail,omitempty"` + Path string `json:"path,omitempty"` + DownloadedBytes int64 `json:"downloadedBytes"` + TotalBytes int64 `json:"totalBytes"` + Error string `json:"error,omitempty"` +} + +type ffmpegInstaller struct { + mu sync.Mutex + configured string + directory string + name string + client *http.Client + pack ffmpegPackage + state ffmpegInstallState + record func(diagnosticEvent) + done chan struct{} + cancel context.CancelFunc + lastError error + lastTry time.Time +} + +func (downloader *Downloader) ffmpegInstallation() *ffmpegInstaller { + downloader.ffmpegMu.Lock() + defer downloader.ffmpegMu.Unlock() + if downloader.ffmpegInstaller == nil { + directory, name := ffmpegManagedDirectory(), ffmpegExecutableName() + client := *downloader.client + client.Timeout = 0 + client.CheckRedirect = func(request *http.Request, via []*http.Request) error { + if request.URL.Scheme != "https" || len(via) >= 10 { + return errors.New("FFmpeg 下载重定向不安全或次数过多") + } + return nil + } + installer := &ffmpegInstaller{ + configured: downloader.cfg.FFmpeg, directory: directory, name: name, client: &client, + pack: ffmpegPackages[runtime.GOOS+"/"+runtime.GOARCH], state: ffmpegInstallState{Status: "idle"}, record: downloader.recordDiagnostic, + } + downloader.ffmpegInstaller = installer + } + return downloader.ffmpegInstaller +} + +func (downloader *Downloader) ensureFFmpeg(ctx context.Context) (string, error) { + return downloader.ffmpegInstallation().ensure(ctx) +} + +func (installer *ffmpegInstaller) snapshot() ffmpegInstallState { + installer.mu.Lock() + defer installer.mu.Unlock() + return installer.state +} + +func (installer *ffmpegInstaller) update(status, detail string, downloaded, total int64) { + installer.mu.Lock() + installer.state.Status, installer.state.Detail = status, detail + installer.state.DownloadedBytes, installer.state.TotalBytes = downloaded, total + installer.mu.Unlock() +} + +func (installer *ffmpegInstaller) ensure(ctx context.Context) (string, error) { + if err := ctx.Err(); err != nil { + return "", err + } + installer.mu.Lock() + if installer.state.Status == "ready" { + if path, err := exec.LookPath(installer.state.Path); err == nil { + installer.mu.Unlock() + return path, nil + } + installer.state.Status = "idle" + } + if installer.done == nil { + if installer.lastError != nil && time.Since(installer.lastTry) < 30*time.Second { + err := installer.lastError + installer.mu.Unlock() + return "", err + } + installer.done = make(chan struct{}) + installer.lastTry = time.Now() + installer.state = ffmpegInstallState{Status: "verifying", Detail: "正在检查 FFmpeg 的播放能力"} + installCtx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) + installer.cancel = cancel + go installer.run(installCtx) + } + done := installer.done + installer.mu.Unlock() + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-done: + installer.mu.Lock() + defer installer.mu.Unlock() + return installer.state.Path, installer.lastError + } +} + +func (installer *ffmpegInstaller) run(ctx context.Context) { + path, err := installer.prepare(ctx) + err = publicError(err) + event := diagnosticEvent{Level: "info", Event: "ffmpeg.ready", Message: "FFmpeg 已就绪：" + path} + if err != nil { + event = diagnosticEvent{Event: "ffmpeg.failed", Message: err.Error()} + } + if installer.record != nil { + installer.record(event) + } + installer.mu.Lock() + defer installer.mu.Unlock() + installer.cancel() + installer.cancel = nil + installer.lastError = err + if err != nil { + installer.state.Status = "failed" + installer.state.Error = err.Error() + installer.state.Detail = "FFmpeg 准备失败，请检查路径、完整版本或代理" + } else { + installer.state = ffmpegInstallState{Status: "ready", Path: path, Detail: event.Message} + } + close(installer.done) + installer.done = nil +} + +func (installer *ffmpegInstaller) retry() { + installer.mu.Lock() + if installer.done == nil { + installer.lastError = nil + installer.state = ffmpegInstallState{Status: "idle"} + } + installer.mu.Unlock() +} + +func (installer *ffmpegInstaller) stop() { + installer.mu.Lock() + defer installer.mu.Unlock() + if installer.cancel != nil { + installer.cancel() + } +} + +func (installer *ffmpegInstaller) fetch(ctx context.Context, name, target, checksum string, limit int64, progress bool) error { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, ffmpegReleaseURL+name, nil) + if err != nil { + return err + } + request.Header.Set("User-Agent", "juku") + request.Header.Set("Accept-Encoding", "identity") + response, err := installer.client.Do(request) + if err != nil { + return publicError(err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return fmt.Errorf("FFmpeg 下载失败：HTTP %d，请检查代理设置后重试", response.StatusCode) + } + if response.ContentLength > limit { + return errors.New("FFmpeg 下载文件超过预期大小") + } + file, err := os.OpenFile(target, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return err + } + defer file.Close() + digest := sha256.New() + reader := io.TeeReader(io.LimitReader(response.Body, limit+1), digest) + var downloaded int64 + buffer := make([]byte, 128*1024) + lastReport := time.Time{} + for { + count, readErr := reader.Read(buffer) + if count > 0 { + downloaded += int64(count) + if downloaded > limit { + return errors.New("FFmpeg 下载文件超过预期大小") + } + if _, err := file.Write(buffer[:count]); err != nil { + return err + } + if progress && time.Since(lastReport) >= 200*time.Millisecond { + installer.update("downloading", "正在自动下载 FFmpeg", downloaded, installer.pack.archiveSize) + lastReport = time.Now() + } + } + if readErr != nil { + if readErr != io.EOF { + return readErr + } + break + } + } + if hex.EncodeToString(digest.Sum(nil)) != checksum { + return errors.New("FFmpeg 下载文件 SHA-256 校验失败，未安装或执行该文件") + } + return file.Close() +} + +func unpackFFmpeg(archive, output string, pack ffmpegPackage) error { + input, err := os.Open(archive) + if err != nil { + return err + } + defer input.Close() + reader, err := gzip.NewReader(input) + if err != nil { + return err + } + defer reader.Close() + file, err := os.OpenFile(output, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return err + } + defer file.Close() + digest := sha256.New() + size, err := io.Copy(io.MultiWriter(file, digest), io.LimitReader(reader, pack.binarySize+1)) + if err != nil { + return err + } + if size != pack.binarySize || hex.EncodeToString(digest.Sum(nil)) != pack.binarySHA256 { + return errors.New("FFmpeg 可执行文件校验失败，未安装或执行该文件") + } + if err := file.Close(); err != nil { + return err + } + return os.Chmod(output, 0755) +} + +var configureFFmpegProbeProcess = func(command *exec.Cmd) {} + +func validateFFmpeg(ctx context.Context, path string) error { + checkCtx, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + command := exec.CommandContext(checkCtx, path, "-hide_banner", "-encoders") + output := &cappedStringWriter{limit: 256 * 1024} + command.Stdout, command.Stderr = output, output + command.WaitDelay = time.Second + configureFFmpegProbeProcess(command) + if err := command.Run(); err != nil { + if checkCtx.Err() != nil { + return fmt.Errorf("FFmpeg 能力检查未完成：%w", checkCtx.Err()) + } + return fmt.Errorf("FFmpeg 无法启动：%w", err) + } + var video, audio bool + for _, line := range strings.Split(output.String(), "\n") { + fields := strings.Fields(line) + if len(fields) >= 2 { + video = video || fields[1] == "libx264" + audio = audio || fields[1] == "aac" + } + } + if !video || !audio { + return errors.New("FFmpeg 缺少 libx264 或 AAC 编码器，需要完整版本") + } + command = exec.CommandContext(checkCtx, path, + "-hide_banner", "-loglevel", "error", "-nostdin", "-filter_threads", "1", "-filter_complex_threads", "1", + "-f", "lavfi", "-i", "color=c=black:s=16x16:r=30:d=0.1", + "-f", "lavfi", "-i", "anullsrc=channel_layout=stereo:sample_rate=48000", + "-t", "0.1", "-map", "0:v:0", "-map", "1:a:0", + "-vf", "fps=30,scale=trunc(iw/2)*2:trunc(ih/2)*2,setsar=1", + "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-profile:v", "baseline", + "-pix_fmt", "yuv420p", "-crf", "23", "-threads", "1", + "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", + "-movflags", "+frag_keyframe+empty_moov+default_base_moof", "-f", "mp4", "pipe:1") + output = &cappedStringWriter{limit: 16 * 1024} + command.Stdout, command.Stderr = io.Discard, output + command.WaitDelay = time.Second + configureFFmpegProbeProcess(command) + if err := command.Run(); err != nil { + if checkCtx.Err() != nil { + return fmt.Errorf("FFmpeg 播放能力检查未完成：%w", checkCtx.Err()) + } + return fmt.Errorf("FFmpeg 无法完成 H.264/AAC 播放转码（需要 preset 等编码选项）：%w %s", err, truncate(strings.TrimSpace(output.String()), 1500)) + } + return nil +} + +func (installer *ffmpegInstaller) prepare(ctx context.Context) (string, error) { + candidates := ffmpegCandidates(installer.configured, installer.directory) + for _, path := range candidates { + installer.update("verifying", "正在检查 FFmpeg："+path, 0, 0) + err := validateFFmpeg(ctx, path) + if err == nil { + return path, nil + } + if ctx.Err() != nil { + return "", ctx.Err() + } + if !automaticFFmpeg(installer.configured) { + return "", fmt.Errorf("指定的 FFmpeg 不可用（%s）：%w；请更正 download.ffmpeg / -ffmpeg，或设为 ffmpeg 使用自动查找", path, err) + } + if installer.record != nil { + installer.record(diagnosticEvent{Level: "warn", Event: "ffmpeg.rejected", Message: "跳过不可用 FFmpeg：" + path + "：" + err.Error()}) + } + } + if !automaticFFmpeg(installer.configured) { + return "", fmt.Errorf("找不到指定的 FFmpeg（%s）；请更正 download.ffmpeg / -ffmpeg，或设为 ffmpeg 使用自动查找", normalizedFFmpegSetting(installer.configured)) + } + detail := "未找到可用 FFmpeg，正在自动准备" + if len(candidates) > 0 { + detail = "现有 FFmpeg 未通过播放检查，正在自动准备兼容版本" + } + installer.update("downloading", detail, 0, installer.pack.archiveSize) + return installer.install(ctx) +} + +func (installer *ffmpegInstaller) install(ctx context.Context) (string, error) { + if !automaticFFmpeg(installer.configured) { + return "", errors.New("指定的 FFmpeg 不可用，请更正 -ffmpeg 路径，或恢复默认 ffmpeg 以启用自动下载") + } + if installer.pack.platform == "" { + return "", fmt.Errorf("暂不支持自动下载 %s/%s 的 FFmpeg，请将对应可执行文件放入 bin 目录", runtime.GOOS, runtime.GOARCH) + } + if err := os.MkdirAll(filepath.Dir(installer.directory), 0755); err != nil { + return "", fmt.Errorf("无法准备 FFmpeg 目录，请将程序移到可写目录：%w", err) + } + work, err := os.MkdirTemp(filepath.Dir(installer.directory), ".ffmpeg-install-*") + if err != nil { + return "", err + } + defer os.RemoveAll(work) + archive := filepath.Join(work, "download.gz") + pack := installer.pack + if err := installer.fetch(ctx, "ffmpeg-"+pack.platform+".gz", archive, pack.archiveSHA256, pack.archiveSize, true); err != nil { + return "", err + } + installer.update("verifying", "正在校验、解压 FFmpeg", pack.archiveSize, pack.archiveSize) + binary := filepath.Join(work, installer.name) + if err := unpackFFmpeg(archive, binary, pack); err != nil { + return "", err + } + if err := os.Remove(archive); err != nil { + return "", err + } + for _, document := range []struct{ remote, local, checksum string }{ + {pack.platform + ".LICENSE", "LICENSE.txt", pack.licenseSHA256}, + {pack.platform + ".README", "README.txt", pack.readmeSHA256}, + } { + if err := installer.fetch(ctx, document.remote, filepath.Join(work, document.local), document.checksum, 256*1024, false); err != nil { + return "", fmt.Errorf("下载 FFmpeg 许可与构建说明失败：%w", err) + } + } + if err := validateFFmpeg(ctx, binary); err != nil { + return "", err + } + if err := ctx.Err(); err != nil { + return "", err + } + if err := os.Rename(work, installer.directory); err != nil { + installed := filepath.Join(installer.directory, installer.name) + if validationErr := validateFFmpeg(ctx, installed); validationErr == nil { + return installed, nil + } + return "", fmt.Errorf("保存 FFmpeg 失败，请检查 bin 目录权限或已有文件：%w", err) + } + return filepath.Join(installer.directory, installer.name), nil +}