jocay/ggjuku
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(app): add internal components batch 2

Browse Source
This commit is contained in:
吴璨
2026-09-30 10:19:54 +08:00
parent a9d79bbe23
commit fce3d91aa0
10 changed files with 2149 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/download_directories_test.go
+183
View File
@@ -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)
}
}
internal/app/download_hls.go
+451
View File
@@ -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)
}
internal/app/download_media.go
+227
View File
@@ -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
}
internal/app/download_progress.go
+177
View File
@@ -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)
}
}
internal/app/download_quality.go
+93
View File
@@ -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
}
internal/app/download_quality_test.go
+228
View File
@@ -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")
}
}
internal/app/downloader.go
+124
View File
@@ -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
}
internal/app/emby_api.go
+190
View File
@@ -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 尚未返回海报状态，将自动重试")
}
internal/app/emby_feedback_test.go
+354
View File
@@ -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("<season>0</season>")) || !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")
}
}
internal/app/emby_files.go
+122
View File
@@ -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
}