feat(app): add internal components batch 3

This commit is contained in:
吴璨
2026-09-30 10:20:01 +08:00
parent fce3d91aa0
commit 713098f5b7
10 changed files with 3110 additions and 0 deletions
+191
View File
@@ -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
}
}
+113
View File
@@ -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
}
+247
View File
@@ -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()
}
}
}
+338
View File
@@ -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)
}
})
}
}
+397
View File
@@ -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)
}
}
+551
View File
@@ -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
}
+285
View File
@@ -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
}
+247
View File
@@ -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)
}
+335
View File
@@ -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):
}
}
+406
View File
@@ -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
}