From fce3d91aa0ac745d1fd08be3f1e7c3b8cdb7f718 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E7=92=A8?= <1579649885@qq.com> Date: Wed, 30 Sep 2026 10:19:54 +0800 Subject: [PATCH] feat(app): add internal components batch 2 --- internal/app/download_directories_test.go | 183 +++++++++ internal/app/download_hls.go | 451 ++++++++++++++++++++++ internal/app/download_media.go | 227 +++++++++++ internal/app/download_progress.go | 177 +++++++++ internal/app/download_quality.go | 93 +++++ internal/app/download_quality_test.go | 228 +++++++++++ internal/app/downloader.go | 124 ++++++ internal/app/emby_api.go | 190 +++++++++ internal/app/emby_feedback_test.go | 354 +++++++++++++++++ internal/app/emby_files.go | 122 ++++++ 10 files changed, 2149 insertions(+) create mode 100644 internal/app/download_directories_test.go create mode 100644 internal/app/download_hls.go create mode 100644 internal/app/download_media.go create mode 100644 internal/app/download_progress.go create mode 100644 internal/app/download_quality.go create mode 100644 internal/app/download_quality_test.go create mode 100644 internal/app/downloader.go create mode 100644 internal/app/emby_api.go create mode 100644 internal/app/emby_feedback_test.go create mode 100644 internal/app/emby_files.go diff --git a/internal/app/download_directories_test.go b/internal/app/download_directories_test.go new file mode 100644 index 0000000..b6d540f --- /dev/null +++ b/internal/app/download_directories_test.go @@ -0,0 +1,183 @@ +package app + +import ( + "os" + "path/filepath" + "strings" + "sync" + "testing" +) + +func TestDownloadGroupingKeepsExistingDirectoriesAcrossChanges(t *testing.T) { + cfg := defaultConfig() + cfg.OutputDir = t.TempDir() + d := &Downloader{cfg: cfg} + old := Drama{ID: historyFixtureDramaID, Source: sourceHongguo, Title: "同名剧"} + flat, err := d.downloadDramaDirectory(old, old.Title, "") + if err != nil || flat != filepath.Join(cfg.OutputDir, old.Title) { + t.Fatal("default layout changed", flat, err) + } + sentinel := filepath.Join(flat, "001.mp4") + if err := os.WriteFile(sentinel, []byte("original episode"), 0600); err != nil { + t.Fatal(err) + } + grouped := true + d.downloadGrouping = &grouped + old.Title = "更新后的剧名" + if same, err := d.downloadDramaDirectory(old, old.Title, ""); err != nil || same != flat { + t.Fatal("enabling grouping or renaming moved an existing drama", same, err) + } + var groupedPaths []string + for index, id := range []string{"hongguo:7000000000000000002", "hongguo:7000000000000000003"} { + drama := Drama{ID: id, Source: sourceHongguo, Title: "同名剧"} + directory, err := d.downloadDramaDirectory(drama, drama.Title, "") + want := filepath.Join(cfg.OutputDir, "红果", drama.Title) + if index > 0 { + want += "_" + hashShort(drama.ID) + } + if err != nil || directory != want { + t.Fatal("new drama did not use its source directory", directory, err) + } + groupedPaths = append(groupedPaths, directory) + } + if groupedPaths[0] == groupedPaths[1] { + t.Fatal("same title from different drama IDs collided") + } + + d = &Downloader{cfg: cfg} + if same, err := d.downloadDramaDirectory(old, old.Title, ""); err != nil || same != flat { + t.Fatal("restart lost the original flat directory", same, err) + } + for index, id := range []string{"hongguo:7000000000000000002", "hongguo:7000000000000000003"} { + drama := Drama{ID: id, Source: sourceHongguo} + if same, err := d.downloadDramaDirectory(drama, "再次改名", ""); err != nil || same != groupedPaths[index] { + t.Fatal("disabling grouping moved an existing grouped drama", same, err) + } + } + if body, _ := os.ReadFile(sentinel); string(body) != "original episode" { + t.Fatal("directory changes touched existing media") + } +} + +func TestDownloadGroupingRejectsSourceSymlinksAndDramaCollision(t *testing.T) { + for _, kind := range []string{"symlink", "existing-drama"} { + t.Run(kind, func(t *testing.T) { + cfg := defaultConfig() + cfg.OutputDir, cfg.GroupBySource = t.TempDir(), true + drama := Drama{ID: historyFixtureDramaID, Source: sourceHongguo} + group := filepath.Join(cfg.OutputDir, dramaSourceFolder(drama)) + outside := t.TempDir() + if kind == "symlink" { + if err := os.Symlink(outside, group); err != nil { + t.Skip("symlinks unavailable") + } + } else { + if ok, err := claimDownloadDirectory(group, "older-drama", false); err != nil || !ok { + t.Fatal(err) + } + } + if _, err := (&Downloader{cfg: cfg}).downloadDramaDirectory(drama, "新剧", ""); err == nil { + t.Fatal("unsafe source parent accepted") + } + if entries, _ := os.ReadDir(outside); len(entries) != 0 { + t.Fatal("wrote through source symlink") + } + if _, err := os.Stat(filepath.Join(group, "新剧")); !os.IsNotExist(err) { + t.Fatal("created new drama inside an existing drama") + } + }) + } +} + +func TestDownloadDirectoryOwnershipAndConcurrentAllocation(t *testing.T) { + cfg := defaultConfig() + cfg.OutputDir, cfg.GroupBySource = t.TempDir(), true + d := &Downloader{cfg: cfg} + drama := Drama{ID: historyFixtureDramaID, Source: sourceHongguo} + results := make(chan string, 12) + var workers sync.WaitGroup + for i := 0; i < cap(results); i++ { + workers.Add(1) + go func() { + defer workers.Done() + directory, err := d.downloadDramaDirectory(drama, "并发同一剧", "") + if err != nil { + t.Error(err) + } + results <- directory + }() + } + workers.Wait() + close(results) + want := filepath.Join(cfg.OutputDir, "红果", "并发同一剧") + for directory := range results { + if directory != want { + t.Fatal("concurrent enqueue allocated different directories", directory) + } + } + duplicate := filepath.Join(cfg.OutputDir, "重复副本") + if ok, err := claimDownloadDirectory(duplicate, drama.ID, false); err != nil || !ok { + t.Fatal(err) + } + if _, err := (&Downloader{cfg: cfg}).downloadDramaDirectory(drama, "第三份", ""); err == nil || !strings.Contains(err.Error(), "多个") { + t.Fatal("ambiguous ownership created a third copy", err) + } + if same, err := (&Downloader{cfg: cfg}).downloadDramaDirectory(drama, "第三份", want); err != nil || same != want { + t.Fatal("persisted task location did not resolve ambiguity", same, err) + } +} + +func TestSourceGroupingConfigAndTasklessDirectorySurviveRestart(t *testing.T) { + cfg := defaultConfig() + if cfg.GroupBySource || documentFromConfig(cfg).Download.GroupBySource { + t.Fatal("grouping must be opt-in") + } + directory := t.TempDir() + enabled := true + settings := runtimeSettings{Concurrency: 2, RequestConcurrency: 2, RequestIntervalMS: 500, GroupBySource: &enabled} + if err := saveRuntimeSettings(directory, settings); err != nil { + t.Fatal(err) + } + settings.GroupBySource = nil + settings.Concurrency = 3 + if err := saveRuntimeSettings(directory, settings); err != nil { + t.Fatal(err) + } + document, err := readConfigDocument(runtimeSettingsPath(directory)) + if err != nil { + t.Fatal(err) + } + loaded, err := document.config() + if err != nil || !loaded.GroupBySource || loaded.Concurrency != 3 { + t.Fatal("older settings request reset grouping", err) + } + savedPath := filepath.Join(t.TempDir(), "原目录") + app := &UIApp{cfg: cfg, statePath: filepath.Join(directory, "ui-state.json"), + dramaDirectories: map[string]string{historyFixtureDramaID: savedPath}} + if err := app.saveStateLocked(); err != nil { + t.Fatal(err) + } + restarted := &UIApp{cfg: cfg, statePath: app.statePath, tasks: map[string]*UITask{}} + restarted.loadState() + if got := restarted.dramaDirectoryLocked(historyFixtureDramaID); got != savedPath || len(restarted.tasks) != 0 { + t.Fatal("cleaning task records lost the stable download directory", got) + } +} + +func TestDownloadDirectoryRejectsUnsupportedSources(t *testing.T) { + cfg := defaultConfig() + cfg.OutputDir, cfg.GroupBySource = t.TempDir(), true + d := &Downloader{cfg: cfg} + for _, drama := range []Drama{ + {ID: "huangdou:7000000000000000002", Source: "huangdou"}, + {ID: historyFixtureDramaID, Source: "huangdou"}, + {ID: "huangguo_video:123", Source: sourceHongguo}, + } { + if _, err := d.downloadDramaDirectory(drama, "不支持的来源", ""); err == nil { + t.Fatal("accepted a non-Hongguo drama", drama.ID) + } + } + if entries, err := os.ReadDir(cfg.OutputDir); err != nil || len(entries) != 0 { + t.Fatal("unsupported source created download directories", err) + } +} diff --git a/internal/app/download_hls.go b/internal/app/download_hls.go new file mode 100644 index 0000000..0d6534f --- /dev/null +++ b/internal/app/download_hls.go @@ -0,0 +1,451 @@ +package app + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "path" + "regexp" + "strconv" + "strings" + "sync" + "time" +) + +type hlsAsset struct { + extension string + remote string + body []byte + playlist bool + fallbacks []hlsAsset +} + +type hlsProxy struct { + downloader *Downloader + credentials *providerMediaCredentials + client *http.Client + server *http.Server + base string + referer string + key []byte + mediaKey []byte + mu sync.Mutex + assets map[string]hlsAsset + assetIDs map[string]string + failure error + root string + retries int + host string + diagnostic func(diagnosticEvent) + query string + acquire func(context.Context) (func(), error) + cache *downloadMediaCache +} + +var hlsURIAttribute = regexp.MustCompile(`URI="([^"]+)"`) + +func (d *Downloader) newHLSProxy(ctx context.Context, media providerMedia, key []byte) (*hlsProxy, error) { + nonce := randomHex(16) + if nonce == "" { + return nil, errors.New("无法创建本地下载会话") + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return nil, fmt.Errorf("创建本地 HLS 转发失败: %w", err) + } + proxy := &hlsProxy{ + downloader: d, + credentials: media.credentials, + client: &http.Client{Transport: d.client.Transport, Jar: d.client.Jar, CheckRedirect: d.client.CheckRedirect}, + base: "http://" + listener.Addr().String() + "/" + nonce + "/", + referer: media.Referer, + key: key, + mediaKey: media.HLSKey, + assets: map[string]hlsAsset{}, + assetIDs: map[string]string{}, + retries: d.cfg.Retries, + diagnostic: d.recordDiagnostic, + } + if remote, err := url.Parse(media.URL); err == nil { + proxy.host = remote.Hostname() + } + proxy.root = proxy.addAsset(playbackRootAsset(media)) + proxy.server = &http.Server{ + Handler: proxy, + ReadHeaderTimeout: 5 * time.Second, + BaseContext: func(net.Listener) context.Context { return ctx }, + } + go func() { + if err := proxy.server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) { + proxy.recordError(err) + } + }() + return proxy, nil +} + +func playbackRootAsset(media providerMedia) hlsAsset { + asset := hlsAsset{remote: media.URL, body: []byte(media.Playlist), playlist: media.Playlist != ""} + for _, fallback := range mediaFallbackVariants(media) { + asset.fallbacks = append(asset.fallbacks, hlsAsset{remote: fallback.URL, body: []byte(fallback.Playlist), playlist: fallback.Playlist != ""}) + } + if asset.playlist && !strings.Contains(media.Playlist, "#EXT-X-ENDLIST") && !strings.Contains(media.Playlist, "#EXT-X-STREAM-INF:") { + asset.body = nil + } + return asset +} + +func (proxy *hlsProxy) Close() { + if proxy.server != nil { + _ = proxy.server.Close() + } +} + +func (proxy *hlsProxy) Err() error { + proxy.mu.Lock() + defer proxy.mu.Unlock() + return proxy.failure +} + +func (proxy *hlsProxy) recordError(err error) { + if err == nil { + return + } + proxy.mu.Lock() + first := proxy.failure == nil + if proxy.failure == nil { + proxy.failure = publicError(err) + } + proxy.mu.Unlock() + if first && proxy.diagnostic != nil { + proxy.diagnostic(diagnosticEvent{Event: "media.failed", Host: proxy.host, Message: err.Error()}) + } +} + +func (proxy *hlsProxy) addAsset(asset hlsAsset) string { + proxy.mu.Lock() + defer proxy.mu.Unlock() + if existing := proxy.assetIDs[asset.remote]; existing != "" { + return existing + } + if len(proxy.assets) >= 16384 { + return "" + } + extension := asset.extension + if extension == "" { + remote, _ := url.Parse(asset.remote) + if remote != nil { + extension = strings.ToLower(path.Ext(remote.Path)) + } + switch extension { + case ".m3u8": + asset.playlist = true + case ".mp4", ".m4s", ".m4a", ".aac", ".mp3", ".vtt", ".key", ".ts": + default: + extension = ".ts" + } + } + if asset.playlist { + extension = ".m3u8" + } else if len(asset.body) > 0 { + extension = ".key" + } + asset.extension = extension + local := proxy.base + strconv.Itoa(len(proxy.assets)+1) + extension + if proxy.query != "" { + local += "?" + proxy.query + } + parsed, _ := url.Parse(local) + proxy.assets[parsed.Path] = asset + proxy.assetIDs[asset.remote] = local + return local +} + +func (proxy *hlsProxy) rewritePlaylist(raw, baseURL string) (string, error) { + base, err := url.Parse(baseURL) + if err != nil { + return "", err + } + rewrite := func(reference string, playlist, key, initialization bool) (string, error) { + parsed, err := url.Parse(reference) + if err != nil { + return "", err + } + remote := base.ResolveReference(parsed).String() + if !isProviderHTTPMediaURL(remote) { + return "", errors.New("HLS 包含非 HTTP/HTTPS 资源地址") + } + asset := hlsAsset{remote: remote, playlist: playlist || strings.HasSuffix(strings.ToLower(parsed.Path), ".m3u8")} + if key { + asset.extension = ".key" + } else if initialization { + asset.extension = ".mp4" + } + resolved, _ := url.Parse(remote) + if key && len(proxy.key) > 0 && strings.Contains(parsed.Path, "/api/app/vid/sec") { + asset.body = proxy.key + } + if key && len(proxy.mediaKey) == 16 && strings.HasSuffix(resolved.Path, "/enc.key") { + asset.body = proxy.mediaKey + } + local := proxy.addAsset(asset) + if local == "" { + return "", errors.New("HLS 资源数量超过限制") + } + return local, nil + } + lines := strings.Split(strings.TrimPrefix(raw, "\ufeff"), "\n") + nextPlaylist := false + for index, line := range lines { + trimmed := strings.TrimSpace(line) + if strings.HasPrefix(trimmed, "#EXT-X-STREAM-INF:") { + nextPlaylist = true + } + if strings.HasPrefix(trimmed, "#") { + var rewriteErr error + lines[index] = hlsURIAttribute.ReplaceAllStringFunc(line, func(attribute string) string { + reference := hlsURIAttribute.FindStringSubmatch(attribute)[1] + local, err := rewrite(reference, strings.HasPrefix(trimmed, "#EXT-X-MEDIA:") || strings.HasPrefix(trimmed, "#EXT-X-I-FRAME-STREAM-INF:"), strings.HasPrefix(trimmed, "#EXT-X-KEY:") || strings.HasPrefix(trimmed, "#EXT-X-SESSION-KEY:"), strings.HasPrefix(trimmed, "#EXT-X-MAP:")) + if err != nil { + rewriteErr = err + return attribute + } + return `URI="` + local + `"` + }) + if rewriteErr != nil { + return "", rewriteErr + } + } else if trimmed != "" { + lines[index], err = rewrite(trimmed, nextPlaylist, false, false) + if err != nil { + return "", err + } + nextPlaylist = false + } + } + return strings.Join(lines, "\n"), nil +} + +func hlsAssetContentType(extension string) string { + switch extension { + case ".mp4", ".m4s": + return "video/mp4" + case ".ts": + return "video/mp2t" + case ".m4a": + return "audio/mp4" + case ".aac": + return "audio/aac" + case ".mp3": + return "audio/mpeg" + case ".vtt": + return "text/vtt" + case ".key": + return "application/octet-stream" + } + return "" +} + +func (proxy *hlsProxy) ServeHTTP(writer http.ResponseWriter, request *http.Request) { + if request.Method != http.MethodGet && request.Method != http.MethodHead { + writer.WriteHeader(http.StatusMethodNotAllowed) + return + } + proxy.mu.Lock() + asset, found := proxy.assets[request.URL.Path] + proxy.mu.Unlock() + if !found { + http.NotFound(writer, request) + return + } + body := asset.body + if len(body) == 0 { + if proxy.acquire != nil { + release, err := proxy.acquire(request.Context()) + if err != nil { + http.Error(writer, "媒体连接正在等待资源,请稍后重试", http.StatusServiceUnavailable) + return + } + defer release() + } + method := request.Method + if asset.playlist { + method = http.MethodGet + } + buildRequest := func(candidate hlsAsset) (*http.Request, error) { + upstream, err := http.NewRequestWithContext(request.Context(), method, candidate.remote, nil) + if err != nil { + return nil, err + } + upstream.Header.Set("User-Agent", userAgent) + upstream.Header.Set("Referer", proxy.referer) + upstream.Header.Set("Accept-Encoding", "identity") + if origin, err := url.Parse(proxy.referer); err == nil && origin.Host != "" { + upstream.Header.Set("Origin", origin.Scheme+"://"+origin.Host) + } + for _, header := range []string{"Range", "If-Range"} { + if value := request.Header.Get(header); value != "" && !candidate.playlist { + upstream.Header.Set(header, value) + } + } + return upstream, nil + } + candidates := append([]hlsAsset{asset}, asset.fallbacks...) + if proxy.cache != nil && !asset.playlist && request.Method == http.MethodGet && request.Header.Get("Range") == "" { + for _, candidate := range candidates { + file, size, cacheErr := proxy.cache.open(request.Context(), proxy, candidate, buildRequest) + if cacheErr == nil { + defer file.Close() + if contentType := hlsAssetContentType(candidate.extension); contentType != "" { + writer.Header().Set("Content-Type", contentType) + } + writer.Header().Set("Content-Length", strconv.FormatInt(size, 10)) + http.ServeContent(writer, request, "media", time.Time{}, file) + return + } + } + } + var upstream *http.Request + var response *http.Response + var err error + for index, candidate := range candidates { + upstream, err = buildRequest(candidate) + if err != nil { + break + } + response, err = proxy.fetch(upstream) + if err == nil && response.StatusCode >= 200 && response.StatusCode < 300 { + asset = candidate + break + } + if response != nil { + if response.StatusCode == http.StatusRequestedRangeNotSatisfiable || response.StatusCode >= 400 && response.StatusCode < 500 && response.StatusCode != http.StatusUnauthorized && response.StatusCode != http.StatusForbidden && response.StatusCode != http.StatusNotFound && response.StatusCode != http.StatusGone { + break + } + _ = response.Body.Close() + } + if index == len(candidates)-1 { + break + } + } + if err != nil { + proxy.fail(writer, request, err) + return + } + if response.StatusCode < 200 || response.StatusCode >= 300 { + _ = response.Body.Close() + if response.StatusCode == http.StatusRequestedRangeNotSatisfiable { + writer.Header().Set("Content-Range", response.Header.Get("Content-Range")) + writer.WriteHeader(response.StatusCode) + return + } + if response.StatusCode == http.StatusUnauthorized || response.StatusCode == http.StatusForbidden || response.StatusCode == http.StatusNotFound || response.StatusCode == http.StatusGone { + proxy.recordError(fmt.Errorf("媒体源 HTTP %d", response.StatusCode)) + writer.WriteHeader(response.StatusCode) + return + } + proxy.fail(writer, request, fmt.Errorf("HLS 资源请求失败: %s HTTP %d", upstream.URL.Hostname(), response.StatusCode)) + return + } + if asset.playlist || strings.Contains(response.Header.Get("Content-Type"), "mpegurl") { + defer response.Body.Close() + body, err = io.ReadAll(io.LimitReader(response.Body, providerMaxBodyBytes+1)) + if err != nil || len(body) > providerMaxBodyBytes { + proxy.fail(writer, request, errors.New("读取 HLS 播放列表失败或内容过大")) + return + } + asset.playlist = true + if response.Request != nil && response.Request.URL != nil { + asset.remote = response.Request.URL.String() + } + } else { + reader := proxy.mediaReader(upstream, response) + defer reader.Close() + for _, header := range []string{"Content-Type", "Content-Length", "Content-Range", "Accept-Ranges", "Content-Encoding", "ETag", "Last-Modified"} { + if value := response.Header.Get(header); value != "" { + writer.Header().Set(header, value) + } + } + if contentType := hlsAssetContentType(asset.extension); contentType != "" { + writer.Header().Set("Content-Type", contentType) + } + writer.WriteHeader(response.StatusCode) + if request.Method == http.MethodHead { + return + } + if _, err := io.Copy(writer, reader); err != nil && reader.failure != nil && request.Context().Err() == nil { + proxy.recordError(reader.failure) + } + return + } + } + if asset.playlist { + if !bytes.HasPrefix(bytes.TrimSpace(bytes.TrimPrefix(body, []byte("\ufeff"))), []byte("#EXTM3U")) { + proxy.fail(writer, request, errors.New("上游没有返回有效的 HLS 播放列表")) + return + } + rewritten, err := proxy.rewritePlaylist(string(body), asset.remote) + if err != nil { + proxy.fail(writer, request, err) + return + } + body = []byte(rewritten) + writer.Header().Set("Content-Type", "application/vnd.apple.mpegurl") + } else { + writer.Header().Set("Content-Type", "application/octet-stream") + } + http.ServeContent(writer, request, "media", time.Time{}, bytes.NewReader(body)) +} + +func (proxy *hlsProxy) fetch(request *http.Request) (*http.Response, error) { + request = request.Clone(providerMediaContext(request.Context(), proxy.credentials)) + attempts := mediaRequestAttempts(proxy.retries) + var lastErr error + for attempt := 0; attempt < attempts; attempt++ { + if attempt > 0 { + timer := time.NewTimer(time.Duration(attempt) * time.Second) + select { + case <-timer.C: + case <-request.Context().Done(): + timer.Stop() + return nil, request.Context().Err() + } + } + var response *http.Response + var err error + if proxy.downloader != nil { + response, err = proxy.downloader.doMediaRequestWithClient(request.Clone(request.Context()), proxy.client) + } else { + upstream := request.Clone(request.Context()) + mediaRequestHeaders(upstream, proxy.referer) + client := proxy.client + if proxy.credentials != nil { + if err := proxy.credentials.apply(upstream); err != nil { + return nil, err + } + client = proxy.credentials.client(client) + } + response, err = client.Do(upstream) + } + if err == nil { + return response, nil + } + if request.Context().Err() != nil { + return nil, request.Context().Err() + } + lastErr = err + } + return nil, fmt.Errorf("媒体资源连接失败(%s,已尝试 %d 次;可在代理设置中检测连接): %w", request.URL.Hostname(), attempts, publicError(lastErr)) +} + +func (proxy *hlsProxy) fail(writer http.ResponseWriter, request *http.Request, err error) { + if request.Context().Err() != nil { + return + } + proxy.recordError(err) + http.Error(writer, "上游媒体请求失败", http.StatusBadGateway) +} diff --git a/internal/app/download_media.go b/internal/app/download_media.go new file mode 100644 index 0000000..d89d14e --- /dev/null +++ b/internal/app/download_media.go @@ -0,0 +1,227 @@ +package app + +import ( + "bufio" + "context" + "encoding/hex" + "errors" + "fmt" + "io" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "time" +) + +func (d *Downloader) downloadMediaWithProgress(ctx context.Context, task Task, callback func(DownloadProgress)) error { + if ok, size := existingGood(task.OutPath, d.cfg.SkipBytes); ok { + if callback != nil { + callback(DownloadProgress{Percent: 100, DownloadedBytes: size, TotalBytes: size, Phase: "completed"}) + } + return nil + } + if callback != nil { + callback(DownloadProgress{Phase: "preparing"}) + } + partPath := strings.TrimSuffix(task.OutPath, filepath.Ext(task.OutPath)) + ".part.mp4" + if err := os.MkdirAll(filepath.Dir(task.OutPath), 0o755); err != nil { + return err + } + progress := newDownloadProgressState(partPath, task.OutPath, task.Chapter.MediaSize, callback) + progress.report("resolving", true) + retries := d.cfg.Retries + if retries <= 0 { + retries = 3 + } + var lastErr error + for attempt := 1; attempt <= retries; attempt++ { + if err := ctx.Err(); err != nil { + return err + } + if attempt > 1 { + _ = os.Remove(partPath) + progress = newDownloadProgressState(partPath, task.OutPath, task.Chapter.MediaSize, callback) + progress.report("retrying", true) + select { + case <-time.After(time.Duration(attempt*2) * time.Second): + case <-ctx.Done(): + return ctx.Err() + } + } + progress.report("resolving", true) + media, err := d.resolveProviderMedia(ctx, task) + if err != nil { + lastErr = publicError(err) + continue + } + media, err = d.selectDownloadQuality(ctx, task, media) + if err != nil { + lastErr = publicError(err) + continue + } + if len(media.CENCKey) != 0 && (len(media.CENCKey) != 16 || media.Playlist != "") { + return errors.New("CENC 媒体的密钥或格式无效") + } + if handled, directErr := d.downloadDirectProviderMedia(ctx, task, media, nil, progress); handled { + if directErr == nil { + return nil + } + lastErr = directErr + continue + } + ffmpeg, err := d.ensureFFmpeg(ctx) + if err != nil { + return err + } + _ = os.Remove(partPath) + progress.setMediaTotal(media.Duration) + proxy, err := d.newHLSProxy(ctx, media, nil) + if err != nil { + return err + } + if media.Playlist != "" { + if cache, cacheErr := prepareDownloadMediaCache(partPath+".segments", media, nil); cacheErr == nil { + proxy.cache = cache + } else { + proxy.Close() + lastErr = cacheErr + continue + } + } + inputURL := proxy.root + origin, _ := url.Parse(media.Referer) + headers := fmt.Sprintf("User-Agent: %s\r\nReferer: %s\r\n", userAgent, media.Referer) + if origin != nil && origin.Host != "" { + headers += "Origin: " + origin.Scheme + "://" + origin.Host + "\r\n" + } + cmdCtx, cancel := context.WithTimeout(ctx, 30*time.Minute) + args := []string{ + "-hide_banner", "-loglevel", "error", "-nostats", + "-rw_timeout", "20000000", + "-protocol_whitelist", "http,https,tcp,tls,crypto,httpproxy", + "-headers", headers, + } + if media.Playlist != "" { + args = append(args, "-allowed_extensions", "ALL") + } + if len(media.CENCKey) > 0 { + args = append(args, "-decryption_key", hex.EncodeToString(media.CENCKey)) + } + args = append(args, + "-i", inputURL, + "-c", "copy", + "-progress", "pipe:1", + "-f", "mp4", + "-y", partPath, + ) + cmd := ffmpegMediaCommand(cmdCtx, ffmpeg, args...) + stdout, err := cmd.StdoutPipe() + if err != nil { + if proxy != nil { + proxy.Close() + } + cancel() + return err + } + stderr, err := cmd.StderrPipe() + if err != nil { + if proxy != nil { + proxy.Close() + } + cancel() + return err + } + if err := cmd.Start(); err != nil { + if proxy != nil { + proxy.Close() + } + cancel() + lastErr = err + continue + } + progress.report("downloading", true) + stderrDone := make(chan string, 1) + go func() { + var b cappedStringWriter + b.limit = 64 * 1024 + _, _ = io.Copy(&b, stderr) + stderrDone <- b.String() + }() + progressDone := make(chan struct{}) + go func() { + defer close(progressDone) + scanner := bufio.NewScanner(stdout) + for scanner.Scan() { + key, val, ok := strings.Cut(scanner.Text(), "=") + if ok && (key == "out_time_us" || key == "out_time_ms") { + if micros, err := strconv.ParseInt(strings.TrimSpace(val), 10, 64); err == nil && micros >= 0 { + progress.setMediaElapsed(time.Duration(micros) * time.Microsecond) + progress.report("downloading", false) + } + } + } + }() + tickerDone := make(chan struct{}) + go func() { + defer close(tickerDone) + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + select { + case <-ticker.C: + progress.report("downloading", false) + case <-progressDone: + return + } + } + }() + waitErr := cmd.Wait() + <-progressDone + <-tickerDone + stderrText := <-stderrDone + commandErr := cmdCtx.Err() + cancel() + var proxyErr error + if proxy != nil { + proxyErr = proxy.Err() + proxy.Close() + } + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr + } + if commandErr != nil { + lastErr = commandErr + _ = os.Remove(partPath) + continue + } + if proxyErr != nil { + lastErr = proxyErr + continue + } + if waitErr != nil { + lastErr = fmt.Errorf("ffmpeg failed: %v %s", waitErr, truncate(stderrText, 1000)) + _ = os.Remove(partPath) + continue + } + st, err := os.Stat(partPath) + if err != nil { + lastErr = err + continue + } + if st.Size() < 100*1024 { + lastErr = fmt.Errorf("downloaded file too small: %d bytes", st.Size()) + _ = os.Remove(partPath) + continue + } + if err := os.Rename(partPath, task.OutPath); err != nil { + lastErr = err + continue + } + _ = os.RemoveAll(partPath + ".segments") + progress.report("completed", true) + return nil + } + return lastErr +} diff --git a/internal/app/download_progress.go b/internal/app/download_progress.go new file mode 100644 index 0000000..8366bbf --- /dev/null +++ b/internal/app/download_progress.go @@ -0,0 +1,177 @@ +package app + +import ( + "math" + "os" + "strings" + "sync" + "time" +) + +type cappedStringWriter struct { + b strings.Builder + limit int +} + +func (w *cappedStringWriter) Write(p []byte) (int, error) { + if w.limit <= 0 { + return len(p), nil + } + remaining := w.limit - w.b.Len() + if remaining > 0 { + if len(p) > remaining { + _, _ = w.b.Write(p[:remaining]) + } else { + _, _ = w.b.Write(p) + } + } + return len(p), nil +} + +func (w *cappedStringWriter) String() string { return w.b.String() } + +type downloadProgressState struct { + mu sync.Mutex + callbackMu sync.Mutex + callback func(DownloadProgress) + partPath string + outPath string + totalBytes int64 + mediaTotal time.Duration + mediaDone time.Duration + started time.Time + lastSample time.Time + lastBytes int64 + byteProgress int64 + hasByteProgress bool + speed float64 + lastReport time.Time + lastPercent int + lastPhase string +} + +func newDownloadProgressState(partPath, outPath string, totalBytes int64, callback func(DownloadProgress)) *downloadProgressState { + return &downloadProgressState{partPath: partPath, outPath: outPath, totalBytes: totalBytes, callback: callback, started: time.Now()} +} + +func (p *downloadProgressState) setMediaTotal(total time.Duration) { + p.mu.Lock() + p.mediaTotal = total + p.mu.Unlock() +} + +func (p *downloadProgressState) setTotalBytes(total int64) { + if total <= 0 { + return + } + p.mu.Lock() + p.totalBytes = total + p.mu.Unlock() +} + +func (p *downloadProgressState) setByteProgress(downloaded, total int64) { + if downloaded < 0 { + downloaded = 0 + } + p.mu.Lock() + p.byteProgress = downloaded + p.hasByteProgress = true + if total > 0 { + p.totalBytes = total + } + p.mu.Unlock() +} + +func (p *downloadProgressState) setMediaElapsed(elapsed time.Duration) { + if elapsed < 0 { + elapsed = 0 + } + p.mu.Lock() + if elapsed > p.mediaDone { + p.mediaDone = elapsed + } + p.mu.Unlock() +} + +func (p *downloadProgressState) snapshot(now time.Time, phase string) DownloadProgress { + p.mu.Lock() + defer p.mu.Unlock() + bytesDownloaded := int64(0) + if p.hasByteProgress { + bytesDownloaded = p.byteProgress + } else if st, err := os.Stat(p.partPath); err == nil && !st.IsDir() { + bytesDownloaded = st.Size() + } else if phase == "completed" && p.outPath != "" { + if st, err := os.Stat(p.outPath); err == nil && !st.IsDir() { + bytesDownloaded = st.Size() + } + } + if !p.lastSample.IsZero() { + seconds := now.Sub(p.lastSample).Seconds() + if seconds > 0 && bytesDownloaded >= p.lastBytes { + p.speed = float64(bytesDownloaded-p.lastBytes) / seconds + } + } + p.lastSample = now + p.lastBytes = bytesDownloaded + if p.speed < 0 || math.IsNaN(p.speed) || math.IsInf(p.speed, 0) { + p.speed = 0 + } + elapsed := now.Sub(p.started) + if elapsed < 0 { + elapsed = 0 + } + mediaElapsed := p.mediaDone + mediaTotal := p.mediaTotal + percent := 0 + if mediaTotal > 0 { + percent = int(float64(mediaElapsed) / float64(mediaTotal) * 100) + } else if p.totalBytes > 0 { + percent = int(float64(bytesDownloaded) / float64(p.totalBytes) * 100) + } + if phase == "completed" { + percent = 100 + if mediaTotal > 0 { + mediaElapsed = mediaTotal + } + } + if percent < 0 { + percent = 0 + } + if phase != "completed" && percent >= 100 { + percent = 99 + } else if percent > 100 { + percent = 100 + } + remaining := time.Duration(0) + if mediaTotal > 0 && mediaElapsed > 0 { + remaining = time.Duration(float64(mediaTotal-mediaElapsed) * float64(elapsed) / float64(mediaElapsed)) + } else if p.totalBytes > 0 && p.speed > 0 && bytesDownloaded < p.totalBytes { + remaining = time.Duration(float64(p.totalBytes-bytesDownloaded) / p.speed * float64(time.Second)) + } + if remaining < 0 || math.IsNaN(float64(remaining)) || math.IsInf(float64(remaining), 0) { + remaining = 0 + } + return DownloadProgress{Percent: percent, DownloadedBytes: bytesDownloaded, TotalBytes: p.totalBytes, SpeedBytesPerSecond: p.speed, Elapsed: elapsed, Remaining: remaining, MediaElapsed: mediaElapsed, MediaTotal: mediaTotal, Phase: phase} +} + +func (p *downloadProgressState) report(phase string, force bool) { + if p.callback == nil { + return + } + now := time.Now() + p.callbackMu.Lock() + defer p.callbackMu.Unlock() + progress := p.snapshot(now, phase) + p.mu.Lock() + changed := force || progress.Percent != p.lastPercent || progress.Phase != p.lastPhase || now.Sub(p.lastReport) >= time.Second + if changed { + p.lastPercent = progress.Percent + p.lastPhase = progress.Phase + p.lastReport = now + } + p.mu.Unlock() + if changed { + p.callback(progress) + } +} diff --git a/internal/app/download_quality.go b/internal/app/download_quality.go new file mode 100644 index 0000000..b8e24b3 --- /dev/null +++ b/internal/app/download_quality.go @@ -0,0 +1,93 @@ +package app + +import ( + "context" + "encoding/json" + "io" + "net/http" +) + +func readDownloadRequest(writer http.ResponseWriter, request *http.Request) ([]string, *int, bool) { + var input struct { + IDs []string `json:"ids"` + Quality *int `json:"quality"` + } + decoder := json.NewDecoder(http.MaxBytesReader(writer, request.Body, 1<<20)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&input); err != nil { + writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "下载参数无效"}) + return nil, nil, false + } + if decoder.Decode(new(any)) != io.EOF || input.Quality != nil && (*input.Quality < 0 || *input.Quality > 4320) { + writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "下载清晰度选项无效"}) + return nil, nil, false + } + ids, ok := cleanIDList(writer, input.IDs) + if !ok { + return nil, nil, false + } + for index, id := range ids { + _, sourceID, valid := splitProviderDramaID(id) + if !valid { + writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "此版本仅支持红果剧集 ID"}) + return nil, nil, false + } + ids[index] = providerDramaID(sourceHongguo, sourceID) + } + return ids, input.Quality, true +} + +func preferredDownloadQuality(requested int, available []int) int { + best, lowest := 0, 0 + for _, quality := range available { + if quality <= 0 || quality > 4320 { + continue + } + if lowest == 0 || quality < lowest { + lowest = quality + } + if (requested <= 0 || quality <= requested) && quality > best { + best = quality + } + } + if best == 0 { + return lowest + } + return best +} + +func (downloader *Downloader) selectDownloadQuality(ctx context.Context, task Task, media providerMedia) (providerMedia, error) { + qualities := []int{media.Quality} + validVariants := make([]providerMedia, 0, len(media.Variants)+1) + for _, variant := range media.Variants { + if isProviderHTTPMediaURL(variant.URL) { + qualities = append(qualities, variant.Quality) + validVariants = append(validVariants, variant) + } + } + if isProviderHTTPMediaURL(media.URL) { + current := media + current.Variants = nil + validVariants = append(validVariants, current) + } + media.Variants = validVariants + if variants := playbackHLSVariants(media.Playlist); len(variants) > 0 { + qualities = nil + for _, variant := range variants { + qualities = append(qualities, variant.quality) + } + } + quality := preferredDownloadQuality(task.DownloadQuality, qualities) + selected, err := downloader.selectPlaybackQuality(context.WithValue(ctx, playbackQualityKey{}, quality), media) + if err != nil { + return providerMedia{}, err + } + var available []int + for _, option := range playbackQualityOptions(selected) { + available = append(available, option.Value) + } + downloader.recordDiagnostic(diagnosticEvent{Level: "info", Event: "download.media", Source: sourceFromDramaID(task.DramaID), + DramaID: task.DramaID, DramaTitle: task.DramaTitle, Episode: task.Index, Quality: selected.Quality, RequestedQuality: task.DownloadQuality, AvailableQualities: available, + Message: "下载保留源媒体的音视频编码和分辨率"}) + return selected, nil +} diff --git a/internal/app/download_quality_test.go b/internal/app/download_quality_test.go new file mode 100644 index 0000000..6768705 --- /dev/null +++ b/internal/app/download_quality_test.go @@ -0,0 +1,228 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "testing" + "time" +) + +func TestDownloadQualityExplicitChoiceAndFallback(t *testing.T) { + d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { + return rankingHTTPResponse(request, http.StatusOK, "#EXTM3U\n#EXTINF:2,\nsynthetic.ts\n#EXT-X-ENDLIST\n"), nil + }) + low := providerMedia{URL: "https://media.example.test/low.mp4", Quality: 720} + high := providerMedia{URL: "https://media.example.test/high.mp4", Quality: 1080} + media := high + media.Variants = []providerMedia{high, low, {URL: "file:///invalid", Quality: 2160}} + master := "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1000000,RESOLUTION=1080x1920\nhigh.m3u8\n#EXT-X-STREAM-INF:BANDWIDTH=2000000,RESOLUTION=720x1280\nlow.m3u8\n" + for _, input := range []struct{ requested, expected int }{{0, 1080}, {2160, 1080}, {1080, 1080}, {900, 720}, {720, 720}, {480, 720}} { + task := Task{DownloadQuality: input.requested} + for _, source := range []providerMedia{media, {URL: "https://media.example.test/master.m3u8", Playlist: master}} { + selected, err := d.selectDownloadQuality(context.Background(), task, source) + if err != nil || selected.Quality != input.expected { + t.Fatalf("requested %d: selected %d, expected %d: %v", input.requested, selected.Quality, input.expected, err) + } + } + } + events := readDiagnosticEvents(t, d.diagnostics.path) + last := events[len(events)-1] + if last.RequestedQuality != 480 || last.Quality != 720 { + t.Fatal("fallback log lost requested and actual quality", last) + } +} + +func TestDownloadQualityRequestValidation(t *testing.T) { + for _, input := range []struct { + body string + valid bool + quality int + }{ + {`{"ids":["hongguo:7000000000000000001"]}`, true, 0}, + {`{"ids":["hongguo:7000000000000000001"],"quality":720}`, true, 720}, + {`{"ids":["hongguo:7000000000000000001"],"quality":-1}`, false, 0}, + {`{"ids":["hongguo:7000000000000000001"],"quality":9999}`, false, 0}, + {`{"ids":["hongguo:7000000000000000001"],"quality":"720"}`, false, 0}, + {`{"ids":["hongguo:7000000000000000001"],"quality":720.5}`, false, 0}, + {`{"ids":[]}`, false, 0}, + {`{"ids":["hongguo:7000000000000000001"]} {}`, false, 0}, + } { + writer := httptest.NewRecorder() + _, quality, valid := readDownloadRequest(writer, httptest.NewRequest(http.MethodPost, "/api/ui/download", strings.NewReader(input.body))) + if valid != input.valid || valid && quality != nil && *quality != input.quality { + t.Fatal(input.body, valid, quality, writer.Body.String()) + } + } +} + +func downloadQualityApp(t *testing.T) *UIApp { + t.Helper() + d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { + t.Error("download task fixture attempted a network request", request.URL) + return nil, errors.New("fixture only") + }) + a := &UIApp{cfg: d.cfg, downloader: d, tasks: map[string]*UITask{}, statePath: filepath.Join(d.cfg.dataDirectory(), "ui-state.json")} + a.cond = sync.NewCond(&a.mu) + drama := Drama{ID: "hongguo:7000000000000000001", Source: sourceHongguo, SourceID: "7000000000000000001", Title: "无图画质任务"} + a.dramas = []Drama{drama} + client := d.hongguoClient() + client.details[drama.SourceID] = hongguoDetailEntry{Drama: drama, Chapters: []Chapter{ + {ID: "hongguo:7000000000000000001:7000000000000000002", Source: sourceHongguo, Title: "第 1 集", CurrentEpisode: rawEpisode(1)}, + }, ExpiresAt: time.Now().Add(time.Hour)} + return a +} + +func waitDownloadQualityParse(t *testing.T, a *UIApp) []uiTaskView { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + a.mu.Lock() + pending := len(a.parsingInFlight) > 0 + views := a.taskViewsLocked() + a.mu.Unlock() + if !pending { + return views + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("chapter parsing did not finish") + return nil +} + +func TestDownloadQualitySurvivesQueueRestartRetryAndUpdate(t *testing.T) { + a := downloadQualityApp(t) + body := `{"ids":["hongguo:7000000000000000001"],"quality":720}` + writer := httptest.NewRecorder() + a.handleDownload(writer, httptest.NewRequest(http.MethodPost, "/api/ui/download", strings.NewReader(body))) + if writer.Code != http.StatusAccepted { + t.Fatal(writer.Code, writer.Body.String()) + } + views := waitDownloadQualityParse(t, a) + if len(views) != 1 || views[0].DownloadQuality != 720 { + t.Fatal("queued task lost selected quality", views) + } + restarted := &UIApp{cfg: a.cfg, downloader: a.downloader, tasks: map[string]*UITask{}, statePath: a.statePath} + restarted.cond = sync.NewCond(&restarted.mu) + restarted.loadState() + task := restarted.tasks[views[0].ID] + if task == nil || task.Source.DownloadQuality != 720 { + t.Fatal("restart lost quality", task) + } + task.Status = uiStatusFailed + selection, _ := json.Marshal(map[string]any{"ids": []string{task.ID}}) + restarted.handleRetry(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/api/ui/tasks/retry", bytes.NewReader(selection))) + if task.Status != uiStatusQueued || task.Source.DownloadQuality != 720 { + t.Fatal("retry changed quality", task) + } + restarted.handlePause(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/api/ui/tasks/pause", bytes.NewReader(selection))) + restarted.handleResume(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/api/ui/tasks/resume", bytes.NewReader(selection))) + if task.Status != uiStatusQueued || task.Source.DownloadQuality != 720 { + t.Fatal("pause/resume changed quality", task) + } + client := a.downloader.hongguoClient() + entry := client.details["7000000000000000001"] + entry.Chapters = append(entry.Chapters, Chapter{ID: "hongguo:7000000000000000001:7000000000000000003", Source: sourceHongguo, Title: "第 2 集", CurrentEpisode: rawEpisode(2)}) + client.details["7000000000000000001"] = entry + restarted.handleUpdate(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/api/ui/update", strings.NewReader(`{"ids":["hongguo:7000000000000000001"]}`))) + views = waitDownloadQualityParse(t, restarted) + if len(views) != 2 || views[0].DownloadQuality != 720 || views[1].DownloadQuality != 720 { + t.Fatal("new episode update did not retain collection quality", views) + } +} + +func TestDownloadQualityFailedChapterParseKeepsPreference(t *testing.T) { + a := downloadQualityApp(t) + client := a.downloader.hongguoClient() + entry := client.details["7000000000000000001"] + empty := entry + empty.Chapters = nil + client.details["7000000000000000001"] = empty + a.handleDownload(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/api/ui/download", strings.NewReader(`{"ids":["hongguo:7000000000000000001"],"quality":540}`))) + views := waitDownloadQualityParse(t, a) + if len(views) != 1 || views[0].Status != uiStatusFailed || views[0].DownloadQuality != 540 { + t.Fatal("chapter failure lost quality", views) + } + client.details["7000000000000000001"] = entry + selection, _ := json.Marshal(map[string]any{"ids": []string{views[0].ID}}) + a.handleRetry(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/api/ui/tasks/retry", bytes.NewReader(selection))) + views = waitDownloadQualityParse(t, a) + if len(views) != 1 || views[0].Status != uiStatusQueued || views[0].DownloadQuality != 540 { + t.Fatal("reparsed chapter lost quality", views) + } +} + +func TestDownloadQualityKeepsHighestAvailableSource(t *testing.T) { + d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { + if request.URL.String() != "https://media.example.test/high.m3u8" { + t.Fatal("download requested an unselected quality", request.URL) + } + return rankingHTTPResponse(request, http.StatusOK, "#EXTM3U\n#EXTINF:2.0,\nsynthetic.ts\n#EXT-X-ENDLIST\n"), nil + }) + task := Task{DramaID: "hongguo:7000000000000000001", Index: 1} + ctx := context.WithValue(context.Background(), playbackQualityKey{}, 720) + low := providerMedia{URL: "https://media.example.test/low.mp4", Quality: 720} + high := providerMedia{URL: "https://media.example.test/high.mp4", Quality: 1080} + media := low + media.Variants = []providerMedia{low, high} + selected, err := d.selectDownloadQuality(ctx, task, media) + if err != nil || selected.Quality != 1080 || selected.URL != high.URL { + t.Fatal("playback quality reduced the download quality", selected.Quality, err) + } + selected, err = d.selectDownloadQuality(ctx, task, low) + if err != nil || selected.Quality != 720 || selected.URL != low.URL { + t.Fatal("download invented a higher source quality", selected, err) + } + master := "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=9000000,RESOLUTION=720x1280\nlow.m3u8\n#EXT-X-STREAM-INF:BANDWIDTH=2000000,RESOLUTION=1080x1920\nhigh.m3u8\n" + selected, err = d.selectDownloadQuality(ctx, task, providerMedia{URL: "https://media.example.test/master.m3u8", Playlist: master, Quality: 720}) + if err != nil || selected.Quality != 1080 || strings.Contains(selected.Playlist, "low.m3u8") || !strings.Contains(selected.Playlist, "high.m3u8") { + t.Fatal("download did not explicitly select the highest HLS resolution", selected, err) + } + events := readDiagnosticEvents(t, d.diagnostics.path) + last := events[len(events)-1] + if last.Event != "download.media" || last.Quality != 1080 || !reflect.DeepEqual(last.AvailableQualities, []int{1080, 720}) { + t.Fatal("download log did not retain the source quality evidence", last) + } +} + +func TestDownloadPreserves1080pPixelsAndPackets(t *testing.T) { + app, session := nativePlaybackFixture(t, "2", "1080x1920") + source := session.tasks[0].OutPath + body, err := os.ReadFile(source) + if err != nil { + t.Fatal(err) + } + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/fixture.mp4" { + t.Error("download requested an unrelated resource") + http.NotFound(writer, request) + return + } + http.ServeContent(writer, request, "fixture.mp4", time.Time{}, bytes.NewReader(body)) + })) + defer server.Close() + d := app.downloader + d.client, d.cfg.Retries, d.cfg.SkipBytes = server.Client(), 1, 1 + task := Task{DramaID: "hongguo:7000000000000000001", DramaTitle: "合成原画下载样本", Index: 1, + OutPath: filepath.Join(t.TempDir(), "download.mp4"), Chapter: Chapter{ID: "hongguo:7000000000000000001:1", Source: sourceHongguo, VideoURL: server.URL + "/fixture.mp4"}} + ctx, cancel := context.WithTimeout(context.WithValue(context.Background(), playbackQualityKey{}, 720), 30*time.Second) + defer cancel() + if err := d.DownloadEpisode(ctx, task); err != nil { + t.Fatal(err) + } + info, err := probeMergeMedia(ctx, d.cfg.FFmpeg, task.OutPath) + if err != nil || info.width != 1080 || info.height != 1920 { + t.Fatal("download reduced the source resolution", info, err) + } + if !reflect.DeepEqual(playbackFixturePackets(t, d.cfg.FFmpeg, source), playbackFixturePackets(t, d.cfg.FFmpeg, task.OutPath)) { + t.Fatal("download re-encoded the original video or audio") + } +} diff --git a/internal/app/downloader.go b/internal/app/downloader.go new file mode 100644 index 0000000..29ffa50 --- /dev/null +++ b/internal/app/downloader.go @@ -0,0 +1,124 @@ +package app + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net/http" + "path/filepath" + "sync" + "sync/atomic" + "time" +) + +type Downloader struct { + cfg Config + client *http.Client + providerMu sync.Mutex + providerHosts map[string]string + hongguoOnce sync.Once + hongguo *hongguoAppClient + rankings rankingCache + limiter *requestLimiter + proxyRouter *proxyRouter + ffmpegMu sync.Mutex + ffmpegInstaller *ffmpegInstaller + diagnostics *diagnosticLog + directoryMu sync.Mutex + downloadDirectories map[string]string + downloadGrouping *bool +} + +func NewDownloader(cfg Config) *Downloader { + if cfg.OutputDir == "" { + cfg.OutputDir = defaultOutputDir() + } + if cfg.FFmpeg == "" { + cfg.FFmpeg = "ffmpeg" + } + if cfg.Concurrency <= 0 { + cfg.Concurrency = 2 + } + if cfg.RequestConcurrency <= 0 { + cfg.RequestConcurrency = 2 + } + if cfg.RequestIntervalMS <= 0 { + cfg.RequestIntervalMS = 500 + } + loadRuntimeSettings(&cfg) + if absolute, err := filepath.Abs(cfg.dataDirectory()); err == nil { + cfg.dataDir = absolute + } + cfg.outputDirSetting = cfg.OutputDir + if absolute, err := filepath.Abs(cfg.OutputDir); err == nil { + cfg.OutputDir = absolute + } + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: cfg.InsecureTLS} + transport.MaxIdleConnsPerHost = 8 + transport.ResponseHeaderTimeout = 20 * time.Second + router := &proxyRouter{} + router.configure(cfg.ProxyURL) + transport.Proxy = router.proxy + return &Downloader{cfg: cfg, client: &http.Client{Transport: newImageTransport(transport, transport), Timeout: 45 * time.Second}, providerHosts: map[string]string{}, limiter: newRequestLimiter(cfg.RequestConcurrency, time.Duration(cfg.RequestIntervalMS)*time.Millisecond), proxyRouter: router, diagnostics: newDiagnosticLog(cfg.dataDirectory())} +} + +func (d *Downloader) DownloadEpisode(ctx context.Context, task Task) error { + return d.DownloadEpisodeWithProgress(ctx, task, nil) +} + +func (d *Downloader) DownloadEpisodeWithProgress(ctx context.Context, task Task, callback func(DownloadProgress)) (resultErr error) { + defer func() { + if ctx.Err() == nil { + d.recordTaskFailure("download.failed", task, 0, resultErr) + } + }() + if !isHongguoTask(task) { + return errors.New("此版本仅支持红果剧集") + } + return d.downloadMediaWithProgress(ctx, task, callback) +} + +func DownloadTasks(ctx context.Context, d *Downloader, tasks []Task) []Result { + if len(tasks) == 0 { + return nil + } + jobs := make(chan Task) + results := make(chan Result) + var done int64 + workers := d.cfg.Concurrency + if workers > len(tasks) { + workers = len(tasks) + } + var wg sync.WaitGroup + for i := 1; i <= workers; i++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for task := range jobs { + cur := atomic.AddInt64(&done, 1) + fmt.Printf(" [线程%d] [%d/%d] 《%s》 第%d/%d 集\n", id, cur, len(tasks), task.DramaTitle, task.Index, task.Total) + err := d.DownloadEpisode(ctx, task) + if err != nil { + results <- Result{Task: task, OK: false, Err: err.Error()} + } else { + results <- Result{Task: task, OK: true} + } + } + }(i) + } + go func() { + for _, task := range tasks { + jobs <- task + } + close(jobs) + wg.Wait() + close(results) + }() + var out []Result + for r := range results { + out = append(out, r) + } + return out +} diff --git a/internal/app/emby_api.go b/internal/app/emby_api.go new file mode 100644 index 0000000..e663e1e --- /dev/null +++ b/internal/app/emby_api.go @@ -0,0 +1,190 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +type embyAPIClient struct { + settings embySyncSettings + client *http.Client + transport *http.Transport +} + +type embyMediaItem struct { + ID string `json:"Id"` + Type string `json:"Type"` + Path string `json:"Path"` + ProviderIDs map[string]string `json:"ProviderIds"` + ImageTags map[string]string `json:"ImageTags"` +} + +type embyItemList struct { + Items []embyMediaItem `json:"Items"` + TotalRecordCount int `json:"TotalRecordCount"` +} + +func newEmbyAPIClient(settings embySyncSettings) *embyAPIClient { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.Proxy = nil + return &embyAPIClient{settings: settings, transport: transport, + client: &http.Client{Transport: transport, Timeout: 95 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}} +} + +func (client *embyAPIClient) close() { + client.transport.CloseIdleConnections() +} + +func (client *embyAPIClient) request(ctx context.Context, method, endpoint string, query url.Values, result any) error { + address := client.settings.ServerURL + endpoint + if len(query) > 0 { + address += "?" + query.Encode() + } + request, err := http.NewRequestWithContext(ctx, method, address, nil) + if err != nil { + return errors.New("Emby 请求地址无效") + } + request.Header.Set("X-Emby-Token", client.settings.APIKey) + request.Header.Set("Accept", "application/json") + if method == http.MethodPost && strings.HasSuffix(endpoint, "/RemoteImages/Download") { + request.Body = io.NopCloser(strings.NewReader("{}")) + request.ContentLength = 2 + request.Header.Set("Content-Type", "application/json") + } + response, err := client.client.Do(request) + if err != nil { + var failure *url.Error + if errors.As(err, &failure) { + err = failure.Err + } + return fmt.Errorf("Emby 连接失败:%w", publicError(err)) + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("Emby HTTP %d", response.StatusCode) + } + if result == nil { + _, err = io.Copy(io.Discard, io.LimitReader(response.Body, 4096)) + return err + } + const limit = 4 << 20 + body, err := io.ReadAll(io.LimitReader(response.Body, limit+1)) + if err != nil { + return errors.New("Emby 响应读取失败") + } + if len(body) > limit || json.Unmarshal(body, result) != nil { + return errors.New("Emby 返回的剧集信息无效或过大") + } + return nil +} + +func (client *embyAPIClient) series(ctx context.Context) ([]embyMediaItem, error) { + var items []embyMediaItem + seen := make(map[string]bool) + for start := 0; start < 100000; { + query := url.Values{"Recursive": {"true"}, "IncludeItemTypes": {"Series"}, "Fields": {"Path,ProviderIds"}, + "EnableImages": {"true"}, "EnableImageTypes": {"Primary"}, "ImageTypeLimit": {"1"}, + "SortBy": {"SortName"}, "SortOrder": {"Ascending"}, "StartIndex": {strconv.Itoa(start)}, "Limit": {"200"}} + var page embyItemList + if err := client.request(ctx, http.MethodGet, "/Items", query, &page); err != nil { + return nil, err + } + added := 0 + for _, item := range page.Items { + if validEmbyItemID(item.ID) && item.Type == "Series" && !seen[item.ID] { + seen[item.ID] = true + items = append(items, item) + added++ + } + } + start += len(page.Items) + if len(page.Items) == 0 || page.TotalRecordCount > 0 && start >= page.TotalRecordCount || + page.TotalRecordCount == 0 && len(page.Items) < 200 { + return items, nil + } + if added == 0 { + return nil, errors.New("Emby 剧集分页未前进,请稍后重试") + } + } + return nil, errors.New("Emby 剧集列表超过 100000 部,暂时无法完成海报匹配") +} + +func validEmbyItemID(id string) bool { + if len(id) == 0 || len(id) > 128 { + return false + } + for _, char := range id { + if char != '-' && char != '_' && (char < '0' || char > '9') && (char < 'a' || char > 'z') && (char < 'A' || char > 'Z') { + return false + } + } + return true +} + +func embyProviderID(item embyMediaItem, provider string) string { + for name, value := range item.ProviderIDs { + if strings.EqualFold(name, provider) { + return value + } + } + return "" +} + +func matchEmbySeries(items []embyMediaItem, folder, identity string) (embyMediaItem, error) { + var matches []embyMediaItem + for _, item := range items { + if item.Type != "Series" || !validEmbyItemID(item.ID) { + continue + } + providerID := embyProviderID(item, "juku") + if providerID != "" && providerID != identity { + continue + } + itemPath := strings.TrimRight(strings.ReplaceAll(item.Path, "\\", "/"), "/") + if !validEmbyFolder(folder) || folder == "" || !strings.HasSuffix("/"+itemPath, "/"+folder) { + continue + } + matches = append(matches, item) + } + if len(matches) > 1 { + return embyMediaItem{}, errors.New("Emby 中存在多个相同导出目录,请移除重复的媒体库挂载后重试") + } + if len(matches) == 0 { + return embyMediaItem{}, nil + } + return matches[0], nil +} + +func (client *embyAPIClient) downloadPoster(ctx context.Context, id, address string) (string, bool, error) { + if !validEmbyItemID(id) { + return "", false, errors.New("Emby 剧集 ID 无效") + } + query := url.Values{"Type": {"Primary"}, "ImageUrl": {address}} + if err := client.request(ctx, http.MethodPost, "/Items/"+id+"/RemoteImages/Download", query, nil); err != nil { + return "", false, err + } + tag, err := client.posterTag(ctx, id) + return tag, true, err +} + +func (client *embyAPIClient) posterTag(ctx context.Context, id string) (string, error) { + var result embyItemList + if err := client.request(ctx, http.MethodGet, "/Items", url.Values{"Ids": {id}, "EnableImages": {"true"}}, &result); err != nil { + return "", err + } + for _, item := range result.Items { + if item.ID == id && item.ImageTags["Primary"] != "" { + return item.ImageTags["Primary"], nil + } + } + return "", errors.New("Emby 尚未返回海报状态,将自动重试") +} diff --git a/internal/app/emby_feedback_test.go b/internal/app/emby_feedback_test.go new file mode 100644 index 0000000..4abad16 --- /dev/null +++ b/internal/app/emby_feedback_test.go @@ -0,0 +1,354 @@ +package app + +import ( + "archive/zip" + "bytes" + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestEmbySourceGroupingKeepsOldFoldersAndStableLinks(t *testing.T) { + settings := embySyncSettings{OutputDir: t.TempDir(), BaseURL: "http://library.test"} + drama := Drama{ID: historyFixtureDramaID, Source: sourceHongguo, Title: "同名剧"} + key := bytes.Repeat([]byte{9}, 32) + ctx := context.Background() + flat, _, _, err := syncEmbyDramaFiles(ctx, settings, drama, embySyncTestChapters(), "", key, "owner") + if err != nil { + t.Fatal(err) + } + first := filepath.Join(settings.OutputDir, flat, "Season 01", "S01E001.strm") + before, _ := os.ReadFile(first) + settings.GroupBySource = true + drama.Title = "修改后的标题" + + same, _, _, err := syncEmbyDramaFiles(ctx, settings, drama, embySyncTestChapters(), "", key, "owner") + if err != nil || same != flat || strings.Contains(same, "/") { + t.Fatal("grouping or title change duplicated an old export", same, err) + } + after, _ := os.ReadFile(first) + if !bytes.Equal(before, after) { + t.Fatal("grouping changed a previous episode link") + } + for _, id := range []string{"hongguo:7000000000000000002", "hongguo:7000000000000000003"} { + newDrama := Drama{ID: id, Source: sourceHongguo, Title: "同名剧"} + folder, _, _, err := syncEmbyDramaFiles(ctx, settings, newDrama, embySyncTestChapters(), "", key, "owner") + if err != nil || !strings.HasPrefix(folder, dramaSourceFolder(newDrama)+"/") { + t.Fatal("new export did not use its source folder", folder, err) + } + settings.GroupBySource = false + newDrama.Title = "新的资料标题" + recovered, _, _, err := syncEmbyDramaFiles(ctx, settings, newDrama, embySyncTestChapters(), "", key, "owner") + if err != nil || recovered != folder { + t.Fatal("switching grouping off moved an existing export", recovered, err) + } + item, err := matchEmbySeries([]embyMediaItem{{ID: "matched", Type: "Series", Path: "D:\\Emby\\" + strings.ReplaceAll(folder, "/", "\\")}}, folder, "") + if err != nil || item.ID != "matched" { + t.Fatal("nested Windows Emby path did not match", err) + } + settings.GroupBySource = true + } + settings.OutputDir = t.TempDir() + outside := t.TempDir() + if err := os.Symlink(outside, filepath.Join(settings.OutputDir, "红果")); err != nil { + t.Skip("symlinks unavailable") + } + if _, _, _, err := syncEmbyDramaFiles(ctx, settings, drama, embySyncTestChapters(), "", key, "owner"); err == nil { + t.Fatal("followed a source-folder symlink") + } + if entries, _ := os.ReadDir(outside); len(entries) != 0 { + t.Fatal("wrote outside the export root") + } +} + +func TestEmbyMergedArchiveSignedRangesAndPermissions(t *testing.T) { + app, admin := administratorFixture(t) + member := newAccountTestBrowser(t, app) + member.register(t, "merged-viewer") + owner := app.browserViewers().accountStore().state.Accounts["merged-viewer"].ID + drama := app.dramas[0] + body := bytes.Repeat([]byte("synthetic-merged-media"), 30) + path := filepath.Join(t.TempDir(), "merged.mp4") + if err := os.WriteFile(path, body, 0600); err != nil { + t.Fatal(err) + } + app.merges = map[string]*UIMergeState{drama.ID: {DramaID: drama.ID, DramaTitle: drama.Title, Status: "success", + StartEpisode: 1, EndEpisode: 88, Merged: 88, OutputPath: path}} + key, err := app.embySigningKey(true) + if err != nil { + t.Fatal(err) + } + record := app.embyMergedRecord(drama.ID) + folder := "红果/" + embyFolderName(drama) + archiveBody, err := buildEmbyArchiveInFolder(drama, embySyncTestChapters(), "http://library.test", key, owner, folder, record) + if err != nil { + t.Fatal(err) + } + archive, err := zip.NewReader(bytes.NewReader(archiveBody), int64(len(archiveBody))) + if err != nil || len(archive.File) != 7 { + t.Fatal("archive lost ordinary episodes or the merged special", err) + } + var address string + for _, file := range archive.File { + if !strings.HasPrefix(file.Name, folder+"/") { + t.Fatal("archive escaped the source folder", file.Name) + } + reader, err := file.Open() + if err != nil { + t.Fatal(err) + } + data, _ := io.ReadAll(reader) + reader.Close() + if strings.HasSuffix(file.Name, "Season 00/S00E001.strm") { + address = strings.TrimSpace(string(data)) + } + if strings.HasSuffix(file.Name, "Season 00/S00E001.nfo") && + (!bytes.Contains(data, []byte("0")) || !bytes.Contains(data, []byte("第1–88集"))) { + t.Fatal("merged special has incorrect NFO") + } + } + if address == "" { + t.Fatal("merged STRM missing") + } + request := func(target *UIApp, method, link, span string) *httptest.ResponseRecorder { + writer := httptest.NewRecorder() + req := httptest.NewRequest(method, link, nil) + req.Header.Set("Range", span) + target.routes().ServeHTTP(writer, req) + return writer + } + for _, test := range []struct { + method, span string + code, length int + }{{"GET", "bytes=0-15", 206, 16}, {"GET", "bytes=-16", 206, 16}, {"HEAD", "", 200, 0}, {"GET", "bytes=99999-", 416, -1}, {"POST", "", 405, -1}} { + result := request(app, test.method, address, test.span) + if result.Code != test.code || test.length >= 0 && result.Body.Len() != test.length { + t.Fatal("merged media failed an Emby range/probe request", test, result.Code, result.Body.String()) + } + } + app.statePath = filepath.Join(t.TempDir(), "ui-state.json") + if err := app.saveStateLocked(); err != nil { + t.Fatal(err) + } + restarted := &UIApp{cfg: app.cfg, downloader: app.downloader, statePath: app.statePath, tasks: map[string]*UITask{}} + restarted.loadState() + if result := request(restarted, "GET", address, "bytes=0-15"); result.Code != 206 || !bytes.Equal(result.Body.Bytes(), body[:16]) { + t.Fatal("merged link did not survive restart without episode tasks", result.Code) + } + for _, field := range []string{"key", "account", "chapter", "id"} { + parsed, _ := url.Parse(address) + query := parsed.Query() + query.Set(field, "forged") + parsed.RawQuery = query.Encode() + if result := request(app, "GET", parsed.String(), ""); result.Code != 403 { + t.Fatal("forged merged link accepted", field, result.Code) + } + } + viewerResultOK(t, admin.request(t, http.MethodPost, "/api/ui/admin/accounts/permissions", map[string]any{"username": "merged-viewer", "onlineOnly": true})) + if result := request(app, "GET", address, ""); result.Code != 403 { + t.Fatal("revoked export permission retained merged access", result.Code) + } + if err := os.Rename(path, path+".moved"); err != nil { + t.Fatal(err) + } + if app.embyMergedRecord(drama.ID) != nil { + t.Fatal("missing merged file was offered for export") + } +} + +func TestEmbyMergeCompletionWakesSyncAndMergedOnlyLibrary(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})) + manager.document.OwnerID = app.browserViewers().accountStore().state.Accounts["admin"].ID + manager.document.Settings = embySyncSettings{Enabled: true, Following: true, IntervalMinutes: 60, OutputDir: t.TempDir(), BaseURL: "http://library.test"} + manager.runOnce(context.Background()) + entry := manager.document.Entries[historyFixtureDramaID] + first := filepath.Join(manager.document.Settings.OutputDir, entry.Folder, "Season 01", "S01E001.strm") + before, err := os.ReadFile(first) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "merged.mp4") + if err := os.WriteFile(path, []byte("completed media fixture"), 0600); err != nil { + t.Fatal(err) + } + app.statePath = filepath.Join(t.TempDir(), "ui-state.json") + app.setMergeState(uiMergeResult{DramaID: historyFixtureDramaID, DramaTitle: "已合并短剧", + StartEpisode: 1, EndEpisode: 3, Merged: 3, OutputPath: path}, "success", 100, false, false) + manager.runOnce(context.Background()) + entry = manager.document.Entries[historyFixtureDramaID] + if !entry.Merged || entry.Episodes != 3 || manager.document.WrittenFiles != 2 { + t.Fatal("recently synced drama did not receive the completed merge", entry, manager.document.WrittenFiles) + } + if after, _ := os.ReadFile(first); !bytes.Equal(before, after) { + t.Fatal("adding a merged version altered the existing episode link") + } + broken := app.downloader.hongguoClient().details["7000000000000000001"] + broken.Chapters = nil + app.downloader.hongguoClient().details["7000000000000000001"] = broken + app.tasks, app.taskOrder, app.dramas = nil, nil, nil + manager.document.Settings.Following, manager.document.Settings.Downloads = false, true + manager.document.Settings.OutputDir = t.TempDir() + manager.document.Entries = map[string]embySyncEntry{} + app.cfg.GroupBySource = true + manager.runOnce(context.Background()) + entry = manager.document.Entries[historyFixtureDramaID] + if entry.Error != "" || !entry.Merged || entry.Episodes != 0 || !strings.HasPrefix(entry.Folder, "红果/") { + t.Fatal("merged-only drama was omitted from downloads sync", entry, manager.document.Error) + } + if _, err := os.Stat(filepath.Join(manager.document.Settings.OutputDir, entry.Folder, "Season 00", "S00E001.strm")); err != nil { + t.Fatal("merged-only STRM missing", err) + } +} + +func TestEmbyMergedExportSurvivesUnavailableSource(t *testing.T) { + app, admin := administratorFixture(t) + drama := app.dramas[0] + path := filepath.Join(t.TempDir(), "merged.mp4") + if err := os.WriteFile(path, []byte("completed local merge"), 0600); err != nil { + t.Fatal(err) + } + app.merges = map[string]*UIMergeState{drama.ID: {DramaID: drama.ID, DramaTitle: drama.Title, + Status: "success", StartEpisode: 1, EndEpisode: 3, OutputPath: path}} + client := app.downloader.hongguoClient() + client.mu.Lock() + delete(client.details, "7000000000000000001") + client.mu.Unlock() + var requests atomic.Int32 + app.downloader.client = &http.Client{Transport: rankingTransport(func(request *http.Request) (*http.Response, error) { + requests.Add(1) + return rankingHTTPResponse(request, http.StatusServiceUnavailable, "synthetic unavailable source"), nil + })} + response := admin.request(t, http.MethodPost, "/api/emby/export", map[string]any{"dramaId": drama.ID, "baseUrl": "http://library.test"}) + if response.Code != http.StatusOK || requests.Load() == 0 { + t.Fatal("local merged export depended on source availability", response.Code, response.Body.String()) + } + archive, err := zip.NewReader(bytes.NewReader(response.Body.Bytes()), int64(response.Body.Len())) + if err != nil || len(archive.File) != 3 { + t.Fatal("merged-only ZIP must contain the show NFO and merged STRM/NFO", err) + } + manager := app.embySyncer() + manager.document.OwnerID = app.browserViewers().accountStore().state.Accounts["admin"].ID + manager.document.Settings = embySyncSettings{Enabled: true, Downloads: true, IntervalMinutes: 60, OutputDir: t.TempDir(), BaseURL: "http://library.test"} + manager.runOnce(context.Background()) + entry := manager.document.Entries[drama.ID] + if entry.Error != "" || !entry.Merged || entry.Episodes != 0 { + t.Fatal("automatic sync lost an offline local merge", entry) + } + if _, err := os.Stat(filepath.Join(manager.document.Settings.OutputDir, entry.Folder, "Season 00", "S00E001.strm")); err != nil { + t.Fatal(err) + } + if _, _, err := app.embyExportChapters(context.Background(), drama.ID, nil); err == nil { + t.Fatal("source failure was hidden without a local merge") + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, _, err := app.embyExportChapters(ctx, drama.ID, app.embyMergedRecord(drama.ID)); !errors.Is(err, context.Canceled) { + t.Fatal("merged export ignored cancellation", err) + } +} + +func TestEmbyLocalPosterIsStableAndPreservesCustomArtwork(t *testing.T) { + app := viewerTestApp(t) + drama := app.dramas[0] + drama.CoverURL = embyFixtureCover + settings := embySyncSettings{GroupBySource: true, OutputDir: t.TempDir(), BaseURL: "http://library.test"} + ctx := context.Background() + key := bytes.Repeat([]byte{8}, 32) + folder, _, _, err := syncEmbyDramaFiles(ctx, settings, drama, embySyncTestChapters(), "", key, "owner") + if err != nil { + t.Fatal(err) + } + data := embySyntheticPoster(t) + var requests atomic.Int32 + app.downloader.client.Transport = embyPosterTestTransport(func(request *http.Request) (*http.Response, error) { + requests.Add(1) + return &http.Response{StatusCode: 200, Header: http.Header{}, Body: io.NopCloser(bytes.NewReader(data)), Request: request}, nil + }) + name, changed, err := app.syncLocalEmbyPoster(ctx, settings, drama, folder, embyFixtureCover) + if err != nil || !changed || name != "poster.png" || !strings.HasPrefix(folder, "红果/") { + t.Fatal("local artwork did not join the grouped media directory", name, changed, err) + } + posterPath := filepath.Join(settings.OutputDir, filepath.FromSlash(folder), name) + if _, _, _, err := syncEmbyDramaFiles(ctx, settings, drama, embySyncTestChapters(), folder, key, "owner", &embyMergedRecord{1, 2}); err != nil { + t.Fatal(err) + } + if _, changed, err := app.syncLocalEmbyPoster(ctx, settings, drama, folder, embyFixtureCover); err != nil || changed || requests.Load() != 1 { + t.Fatal("core sync lost artwork ownership or rewrote the poster", err) + } + custom := []byte("user-edited poster") + if err := os.WriteFile(posterPath, custom, 0600); err != nil { + t.Fatal(err) + } + if _, changed, err := app.syncLocalEmbyPoster(ctx, settings, drama, folder, embyFixtureCover+"&revision=2"); err != nil || changed || requests.Load() != 1 { + t.Fatal("user-modified poster was overwritten", err) + } + if body, _ := os.ReadFile(posterPath); !bytes.Equal(body, custom) { + t.Fatal("custom image bytes changed") + } + + settings.OutputDir = t.TempDir() + folder, _, _, err = syncEmbyDramaFiles(ctx, settings, drama, embySyncTestChapters(), "", key, "owner") + if err != nil { + t.Fatal(err) + } + posterPath = filepath.Join(settings.OutputDir, filepath.FromSlash(folder), "poster.png") + app.downloader.client.Transport = embyPosterTestTransport(func(request *http.Request) (*http.Response, error) { + if err := os.WriteFile(posterPath, custom, 0600); err != nil { + t.Error(err) + } + return &http.Response{StatusCode: 200, Header: http.Header{}, Body: io.NopCloser(bytes.NewReader(data)), Request: request}, nil + }) + if _, changed, err := app.syncLocalEmbyPoster(ctx, settings, drama, folder, embyFixtureCover+"&revision=during-fetch"); err != nil || changed { + t.Fatal("artwork added during fetch was replaced", err) + } + if body, _ := os.ReadFile(posterPath); !bytes.Equal(body, custom) { + t.Fatal("fetch raced with user artwork") + } +} + +func TestEmbyLocalPosterFailureRetriesWithoutLosingSTRM(t *testing.T) { + app := viewerTestApp(t) + manager := app.embySyncer() + drama := app.dramas[0] + drama.CoverURL = embyFixtureCover + settings := embySyncSettings{OutputDir: t.TempDir(), BaseURL: "http://library.test", IntervalMinutes: 60} + folder, _, _, err := syncEmbyDramaFiles(context.Background(), settings, drama, embySyncTestChapters(), "", make([]byte, 32), "owner") + if err != nil { + t.Fatal(err) + } + var calls atomic.Int32 + app.downloader.client.Transport = embyPosterTestTransport(func(*http.Request) (*http.Response, error) { + calls.Add(1) + return nil, errors.New("fixture offline") + }) + document := embySyncDocument{Settings: settings, Entries: map[string]embySyncEntry{drama.ID: {Folder: folder, Episodes: 2}}} + manager.preparePosters(&document, []Drama{drama}, make([]byte, 32)) + manager.synchronizeLocalPosters(context.Background(), &document, []Drama{drama}) + poster := document.Entries[drama.ID].Poster + if poster.LocalError == "" || poster.Status != "failed" || time.Until(poster.LocalRetryAt) < 4*time.Minute { + t.Fatal("offline poster lost bounded retry state", poster) + } + manager.synchronizeLocalPosters(context.Background(), &document, []Drama{drama}) + if calls.Load() != 1 { + t.Fatal("failed image was fetched before retry time") + } + if _, err := os.Stat(filepath.Join(settings.OutputDir, folder, "Season 01", "S01E001.strm")); err != nil { + t.Fatal("poster failure damaged STRM", err) + } + poster.LocalFile, poster.LocalSource = "poster.png", embyPosterSourceHash(poster.SourceURL) + poster.LocalRetryAt = time.Now().Add(-time.Second) + if next := embyLocalPosterDue(poster, time.Hour); next.After(time.Now()) { + t.Fatal("an existing poster suppressed the scheduled repair") + } +} diff --git a/internal/app/emby_files.go b/internal/app/emby_files.go new file mode 100644 index 0000000..cad537f --- /dev/null +++ b/internal/app/emby_files.go @@ -0,0 +1,122 @@ +package app + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/xml" + "errors" + "fmt" + "net/url" + "sort" +) + +type embyChapter struct { + ID string `json:"id"` + Title string `json:"title"` + Number int `json:"number"` +} + +func embyFolderName(drama Drama) string { + folder := []rune(safeFilename(drama.DisplayTitle())) + for len(string(folder)) > 180 { + folder = folder[:len(folder)-1] + } + identity := sha256.Sum256([]byte(drama.ID)) + return string(folder) + " [" + hex.EncodeToString(identity[:8]) + "]" +} + +func embyChapterRecords(id string, chapters []Chapter, previous []embyChapter) ([]embyChapter, error) { + if len(chapters) == 0 && len(previous) == 0 || len(chapters) > 2000 || len(previous) > 2000 { + return nil, errors.New("可导出分集数无效") + } + byID := make(map[string]embyChapter, len(previous)+len(chapters)) + used := make(map[int]bool, len(previous)+len(chapters)) + for _, chapter := range previous { + if !validEmbyIdentity(id, chapter.ID) || chapter.Number < 1 || chapter.Number > 2000 || used[chapter.Number] || byID[chapter.ID].ID != "" { + return nil, errors.New("已有 Emby 分集清单无效,原文件已保留") + } + byID[chapter.ID], used[chapter.Number] = chapter, true + } + seen := make(map[string]bool, len(chapters)) + for index, chapter := range chapters { + if !validEmbyIdentity(id, chapter.ID) || seen[chapter.ID] { + return nil, errors.New("站源未提供唯一、稳定的分集 ID,暂不能导出") + } + seen[chapter.ID] = true + record := byID[chapter.ID] + if record.ID == "" { + number := index + 1 + for used[number] && number <= 2000 { + number++ + } + if number > 2000 { + return nil, errors.New("Emby 分集数超过 2000 集") + } + record = embyChapter{ID: chapter.ID, Number: number} + used[number] = true + } + record.Title = firstNonEmpty(chapter.Title, "第 "+chapter.EpisodeString(index+1)+" 集") + byID[chapter.ID] = record + } + result := make([]embyChapter, 0, len(byID)) + for _, chapter := range byID { + result = append(result, chapter) + } + sort.Slice(result, func(i, j int) bool { return result[i].Number < result[j].Number }) + return result, nil +} + +func writeEmbyFiles(drama Drama, chapters []embyChapter, base string, key []byte, owner string, write func(string, []byte) error, merged ...*embyMergedRecord) error { + title := drama.DisplayTitle() + type nfoValue struct { + Value string `xml:",chardata"` + Type string `xml:"type,attr,omitempty"` + Aspect string `xml:"aspect,attr,omitempty"` + } + var thumb *nfoValue + if cover := embyCoverURL(drama, base, key, owner); cover != "" { + thumb = &nfoValue{Value: cover, Aspect: "poster"} + } + show := struct { + XMLName xml.Name `xml:"tvshow"` + Title string `xml:"title"` + Plot string `xml:"plot,omitempty"` + Identity nfoValue `xml:"uniqueid"` + Thumb *nfoValue `xml:"thumb,omitempty"` + }{Title: title, Plot: firstNonEmpty(drama.Desc, drama.Intro), Identity: nfoValue{Value: embyMetadataID(key, drama.ID), Type: "juku"}, Thumb: thumb} + body, err := xml.MarshalIndent(show, "", " ") + if err != nil { + return err + } + if err = write("tvshow.nfo", append([]byte(xml.Header), body...)); err != nil { + return err + } + for _, chapter := range chapters { + query := url.Values{"id": {drama.ID}, "chapter": {chapter.ID}, "key": {embyToken(key, drama.ID, chapter.ID, owner)}} + if owner != "" { + query.Set("account", owner) + } + path := fmt.Sprintf("Season 01/S01E%03d", chapter.Number) + episode := struct { + XMLName xml.Name `xml:"episodedetails"` + Title string `xml:"title"` + Show string `xml:"showtitle"` + Season int `xml:"season"` + Episode int `xml:"episode"` + }{Title: chapter.Title, Show: title, Season: 1, Episode: chapter.Number} + body, err = xml.MarshalIndent(episode, "", " ") + if err != nil { + return err + } + if err = write(path+".nfo", append([]byte(xml.Header), body...)); err != nil { + return err + } + if err = write(path+".strm", []byte(base+"/api/emby/stream.m3u8?"+query.Encode()+"\n")); err != nil { + return err + } + } + if len(merged) > 0 && merged[0] != nil { + return writeEmbyMerged(drama, merged[0], base, key, owner, write) + } + return nil +}