同步上游 zip:合并任务后台化 + 合并整剧可直接播放;保留 ultrafast+4线程 转码优化与 .gitignore 本地规则

This commit is contained in:
yvan
2026-09-18 10:16:15 +08:00
parent 80485e9dba
commit b96363d787
21 changed files with 1173 additions and 172 deletions
+64 -111
View File
@@ -12,59 +12,6 @@ import (
"time"
)
type mergeMediaInfo struct {
signature string
width int
height int
audio bool
duration time.Duration
}
func probeMergeMedia(ctx context.Context, ffmpeg, path string) (mergeMediaInfo, error) {
probeCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
defer cancel()
command := exec.CommandContext(probeCtx, ffmpeg, "-hide_banner", "-nostdin", "-loglevel", "info",
"-protocol_whitelist", "file,pipe", "-i", path, "-map", "0:v:0", "-map", "0:a:0?",
"-c", "copy", "-t", "0", "-f", "framehash", "-hash", "sha256", "pipe:1")
output := &cappedStringWriter{limit: 64 * 1024}
diagnostics := &cappedStringWriter{limit: 64 * 1024}
command.Stdout, command.Stderr = output, diagnostics
if err := command.Run(); err != nil {
if probeCtx.Err() != nil {
return mergeMediaInfo{}, probeCtx.Err()
}
return mergeMediaInfo{}, fmt.Errorf("检测 %s 失败:%w %s", filepath.Base(path), err, truncate(diagnostics.String(), 1000))
}
var info mergeMediaInfo
var signature []string
for _, line := range strings.Split(output.String(), "\n") {
line = strings.TrimSpace(line)
for _, prefix := range []string{"#extradata ", "#tb ", "#media_type ", "#codec_id ", "#dimensions ", "#sar ", "#sample_rate ", "#channel_layout"} {
if strings.HasPrefix(line, prefix) {
signature = append(signature, line)
break
}
}
if strings.HasPrefix(line, "#dimensions 0:") {
_, _ = fmt.Sscanf(strings.TrimSpace(strings.TrimPrefix(line, "#dimensions 0:")), "%dx%d", &info.width, &info.height)
}
if line == "#media_type 1: audio" {
info.audio = true
}
}
if match := playbackDurationPattern.FindStringSubmatch(diagnostics.String()); len(match) == 4 {
hours, _ := strconv.ParseFloat(match[1], 64)
minutes, _ := strconv.ParseFloat(match[2], 64)
seconds, _ := strconv.ParseFloat(match[3], 64)
info.duration = time.Duration((hours*3600 + minutes*60 + seconds) * float64(time.Second))
}
if info.width <= 0 || info.height <= 0 || info.duration <= 0 {
return mergeMediaInfo{}, fmt.Errorf("%s 缺少有效的视频尺寸或时长,无法安全合并", filepath.Base(path))
}
info.signature = strings.Join(signature, "\n")
return info, nil
}
func runMergeFFmpeg(ctx context.Context, ffmpeg string, args []string, progress func(time.Duration)) error {
command := exec.CommandContext(ctx, ffmpeg, append([]string{
"-hide_banner", "-nostdin", "-nostats", "-loglevel", "error", "-progress", "pipe:1",
@@ -98,29 +45,6 @@ func runMergeFFmpeg(ctx context.Context, ffmpeg string, args []string, progress
return scanner.Err()
}
func normalizedMergeArgs(input, output string, info mergeMediaInfo, width, height int, audio bool) []string {
args := []string{"-xerror", "-threads", "2", "-protocol_whitelist", "file,pipe", "-i", input}
if audio && !info.audio {
args = append(args, "-f", "lavfi", "-i", "anullsrc=channel_layout=stereo:sample_rate=48000")
}
args = append(args, "-map", "0:v:0")
if audio {
audioInput := "0:a:0"
if !info.audio {
audioInput = "1:a:0"
}
args = append(args, "-map", audioInput, "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2",
"-af", "aresample=48000:async=1:first_pts=0,apad", "-shortest")
} else {
args = append(args, "-an")
}
filter := fmt.Sprintf("setpts=PTS-STARTPTS,fps=30,scale=%d:%d:force_original_aspect_ratio=decrease:force_divisible_by=2,pad=%d:%d:(ow-iw)/2:(oh-ih)/2,setsar=1", width, height, width, height)
return append(args, "-sn", "-dn", "-map_metadata", "-1", "-vf", filter,
"-c:v", "libx264", "-preset", "veryfast", "-crf", "18", "-profile:v", "high", "-pix_fmt", "yuv420p",
"-threads:v", "2", "-g", "60", "-sc_threshold", "0", "-video_track_timescale", "90000",
"-max_muxing_queue_size", "4096", "-t", strconv.FormatFloat(info.duration.Seconds(), 'f', 6, 64), "-f", "mp4", "-y", output)
}
func mergeMediaFiles(ctx context.Context, ffmpeg string, paths []string, output string, report func(int, string)) (string, error) {
if len(paths) == 0 {
return "", fmt.Errorf("没有可合并的分集")
@@ -140,77 +64,98 @@ func mergeMediaFiles(ctx context.Context, ffmpeg string, paths []string, output
}
metadata := make([]mergeMediaInfo, 0, len(paths))
var duration time.Duration
var width, height int
var audio, normalize bool
for index, path := range paths {
notify(index*5/len(paths), fmt.Sprintf("检测分集编码 %d/%d", index+1, len(paths)))
info, err := probeMergeMedia(ctx, ffmpeg, path)
if err != nil {
return "", err
}
if index > 0 && info.signature != metadata[0].signature {
normalize = true
}
metadata = append(metadata, info)
duration += info.duration
if info.width > width {
width = info.width
}
if info.height > height {
height = info.height
}
audio = audio || info.audio
}
plan := planMergeMedia(metadata)
method := plan.description()
work, err := os.MkdirTemp(filepath.Dir(output), ".merge-*")
if err != nil {
return "", err
}
defer os.RemoveAll(work)
inputs := append([]string(nil), paths...)
method, start := "原编码快速合并", 5
if normalize {
method, start = "兼容合并(H.264 / AAC)", 95
width += width % 2
height += height % 2
var processed time.Duration
for index, path := range paths {
detail := fmt.Sprintf("编码不一致,统一格式 %d/%d", index+1, len(paths))
notify(5+int(processed*90/duration), detail)
temporary := filepath.Join(work, fmt.Sprintf("%06d.mp4", index+1))
err := runMergeFFmpeg(ctx, ffmpeg, normalizedMergeArgs(path, temporary, metadata[index], width, height, audio), func(elapsed time.Duration) {
var preparedDuration, processed time.Duration
for index, step := range plan.steps {
if step.video || step.audio || step.remux {
preparedDuration += metadata[index].duration
}
}
for index, step := range plan.steps {
if !step.video && !step.audio && !step.remux {
continue
}
action := "无损换封装"
if step.video {
action = "转换此集视频"
} else if step.audio {
action = "仅调整此集音轨"
}
detail := fmt.Sprintf("%s · %s %d/%d", method, action, index+1, len(paths))
notify(5+int(processed*60/preparedDuration), detail)
extension := ".mp4"
if plan.transport {
extension = ".ts"
}
temporary := filepath.Join(work, fmt.Sprintf("%06d%s", index+1, extension))
args, err := plan.prepareArgs(paths[index], temporary, metadata[index], step)
if err == nil {
err = runMergeFFmpeg(ctx, ffmpeg, args, func(elapsed time.Duration) {
if elapsed > metadata[index].duration {
elapsed = metadata[index].duration
}
notify(5+int((processed+elapsed)*90/duration), detail)
notify(5+int((processed+elapsed)*60/preparedDuration), detail)
})
if err != nil {
return "", fmt.Errorf("转换 %s 失败:%w;原分集已保留", filepath.Base(path), err)
}
processed += metadata[index].duration
inputs[index] = temporary
}
if err != nil {
return "", fmt.Errorf("处理 %s 失败:%w;原分集已保留,未将整剧转码", filepath.Base(paths[index]), err)
}
processed += metadata[index].duration
inputs[index] = temporary
}
var manifest strings.Builder
for _, path := range inputs {
manifest.WriteString("ffconcat version 1.0\n")
for index, path := range inputs {
absolute, err := filepath.Abs(path)
if err != nil {
return "", err
}
if strings.ContainsAny(absolute, "\r\n") {
return "", fmt.Errorf("分集路径包含换行符,无法合并")
}
fmt.Fprintf(&manifest, "file '%s'\n", strings.ReplaceAll(filepath.ToSlash(absolute), "'", "'\\''"))
if plan.transport {
fmt.Fprintf(&manifest, "duration %.6f\n", metadata[index].duration.Seconds())
}
}
list := filepath.Join(work, "inputs.ffconcat")
if err := os.WriteFile(list, []byte(manifest.String()), 0600); err != nil {
return "", err
}
notify(start, method)
err = runMergeFFmpeg(ctx, ffmpeg, []string{"-protocol_whitelist", "file,pipe", "-f", "concat", "-safe", "0", "-i", list,
"-map", "0:v:0", "-map", "0:a:0?", "-sn", "-dn", "-c", "copy", "-movflags", "+faststart", "-y", output}, func(elapsed time.Duration) {
notify(start+int(elapsed*time.Duration(99-start)/duration), method)
notify(65, method)
args := []string{"-protocol_whitelist", "file,pipe", "-f", "concat", "-safe", "0", "-i", list,
"-map", "0:v:0", "-map", "0:a:0?", "-sn", "-dn", "-c", "copy", "-video_track_timescale", strconv.Itoa(plan.timescale), "-movflags", "+faststart"}
if plan.parameters {
if plan.video.videoCodec == "h264" {
args = append(args, "-tag:v", "avc3")
} else if plan.video.videoCodec == "hevc" {
args = append(args, "-tag:v", "hev1")
}
}
args = append(args, "-y", output)
err = runMergeFFmpeg(ctx, ffmpeg, args, func(elapsed time.Duration) {
notify(65+int(elapsed*15/duration), method)
})
if err != nil {
return "", err
}
notify(99, "核对合并文件时长")
notify(80, "核对合并文件时长")
merged, err := probeMergeMedia(ctx, ffmpeg, output)
if err != nil {
return "", err
@@ -222,5 +167,13 @@ func mergeMediaFiles(ctx context.Context, ffmpeg string, paths []string, output
if merged.duration < duration-tolerance || merged.duration > duration+tolerance {
return "", fmt.Errorf("合并文件时长不符(预计 %.1f 秒,实际 %.1f 秒),原分集已保留", duration.Seconds(), merged.duration.Seconds())
}
notify(80, "校验完整视频和音轨(只解码,不转码)")
err = runMergeFFmpeg(ctx, ffmpeg, []string{"-xerror", "-err_detect", "explode", "-threads", "2", "-protocol_whitelist", "file,pipe", "-i", output,
"-map", "0:v:0", "-map", "0:a:0?", "-threads:v", "2", "-fps_mode", "passthrough", "-abort_on", "empty_output", "-f", "null", "-"}, func(elapsed time.Duration) {
notify(80+int(elapsed*19/duration), "校验完整视频和音轨(只解码,不转码)")
})
if err != nil {
return "", fmt.Errorf("合并文件解码校验失败:%w;原分集已保留", err)
}
return method, nil
}
+222
View File
@@ -0,0 +1,222 @@
package app
import (
"fmt"
"strconv"
"strings"
"time"
)
type mergeMediaPlan struct {
video, audio mergeMediaInfo
steps []mergeMediaStep
timescale int
transport bool
parameters bool
videoCount int
audioCount int
remuxCount int
}
type mergeMediaStep struct {
video, audio, remux bool
}
func (info mergeMediaInfo) videoKey() string {
key := fmt.Sprintf("%s|%s|%s|%dx%d|%s", info.videoCodec, info.videoProfile, info.pixelFormat, info.width, info.height, info.sar)
if info.videoCodec != "h264" && info.videoCodec != "hevc" {
key += "|" + info.videoExtra
}
return key
}
func (info mergeMediaInfo) audioKey() string {
return fmt.Sprintf("%s|%s|%d|%s|%s", info.audioCodec, info.audioProfile, info.sampleRate, info.layout, info.audioExtra)
}
func majorityMergeMedia(infos []mergeMediaInfo, key func(mergeMediaInfo) string, audio bool) mergeMediaInfo {
counts := make(map[string]int)
durations := make(map[string]time.Duration)
for _, info := range infos {
if audio && !info.audio {
continue
}
value := key(info)
counts[value]++
durations[value] += info.duration
}
var chosen mergeMediaInfo
bestCount, bestDuration := 0, time.Duration(0)
for _, info := range infos {
if audio && !info.audio {
continue
}
value := key(info)
if counts[value] > bestCount || counts[value] == bestCount && durations[value] > bestDuration {
chosen, bestCount, bestDuration = info, counts[value], durations[value]
}
}
return chosen
}
func planMergeMedia(infos []mergeMediaInfo) mergeMediaPlan {
plan := mergeMediaPlan{timescale: 90000}
plan.video = majorityMergeMedia(infos, mergeMediaInfo.videoKey, false)
var compatible []mergeMediaInfo
for _, info := range infos {
if info.videoKey() == plan.video.videoKey() {
compatible = append(compatible, info)
}
}
plan.video = majorityMergeMedia(compatible, func(info mergeMediaInfo) string { return info.videoTB }, false)
plan.audio = majorityMergeMedia(infos, mergeMediaInfo.audioKey, true)
if value, ok := strings.CutPrefix(plan.video.videoTB, "1/"); ok {
if scale, err := strconv.Atoi(value); err == nil && scale > 0 && scale <= 10000000 {
plan.timescale = scale
}
}
for _, info := range infos {
step := mergeMediaStep{video: info.videoKey() != plan.video.videoKey(), audio: plan.audio.audio && (!info.audio || info.audioKey() != plan.audio.audioKey())}
if step.video || info.videoExtra != plan.video.videoExtra {
plan.parameters = true
}
step.remux = info.videoTB != fmt.Sprintf("1/%d", plan.timescale) || plan.audio.audio && info.audioTB != plan.audio.audioTB || info.reorder
plan.steps = append(plan.steps, step)
}
plan.transport = plan.video.videoCodec == "hevc" && plan.parameters
for i := range plan.steps {
step := &plan.steps[i]
step.remux = step.remux || plan.transport
if step.video {
plan.videoCount++
}
if step.audio {
plan.audioCount++
}
if step.remux && !step.video && !step.audio {
plan.remuxCount++
}
}
return plan
}
func (plan mergeMediaPlan) description() string {
parts := []string{}
if plan.videoCount > 0 {
parts = append(parts, fmt.Sprintf("仅转视频 %d/%d 集(%s %dx%d)", plan.videoCount, len(plan.steps), strings.ToUpper(plan.video.videoCodec), plan.video.width, plan.video.height))
}
if plan.audioCount > 0 {
parts = append(parts, fmt.Sprintf("调整音轨 %d/%d 集", plan.audioCount, len(plan.steps)))
}
if len(parts) == 0 {
if plan.remuxCount > 0 {
return "无损换封装合并(不转码)"
}
return "原编码快速合并(不转码)"
}
return strings.Join(parts, ";")
}
func mergeVideoEncoder(info mergeMediaInfo) ([]string, error) {
profile := strings.ToLower(info.videoProfile)
var args []string
switch info.videoCodec {
case "h264":
profiles := map[string]string{"constrained baseline": "baseline", "baseline": "baseline", "main": "main", "high": "high", "high 10": "high10", "high 4:2:2": "high422", "high 4:4:4 predictive": "high444"}
value, ok := profiles[profile]
if !ok {
return nil, fmt.Errorf("暂不能将少数分集转为多数使用的 H.264 %s", info.videoProfile)
}
args = []string{"-c:v", "libx264", "-preset", "veryfast", "-crf", "18", "-profile:v", value, "-threads:v", "2"}
case "hevc":
profiles := map[string]string{"main": "main", "main 10": "main10"}
value, ok := profiles[profile]
if !ok {
return nil, fmt.Errorf("暂不能将少数分集转为多数使用的 HEVC %s", info.videoProfile)
}
args = []string{"-c:v", "libx265", "-preset", "veryfast", "-crf", "18", "-profile:v", value, "-threads:v", "2", "-x265-params", "pools=2:frame-threads=2:log-level=error"}
default:
return nil, fmt.Errorf("暂不支持将少数分集转为多数使用的 %s 编码;未对整剧转码", info.videoCodec)
}
if info.pixelFormat == "" || info.width%2 != 0 || info.height%2 != 0 {
return nil, fmt.Errorf("多数分集的像素格式或尺寸无法安全转换")
}
return append(args, "-pix_fmt", info.pixelFormat), nil
}
func mergeAudioEncoder(info mergeMediaInfo) ([]string, error) {
encoder := info.audioCodec
switch encoder {
case "aac":
if info.audioProfile != "" && info.audioProfile != "LC" {
return nil, fmt.Errorf("暂不能将少数音轨转为多数使用的 AAC %s", info.audioProfile)
}
case "mp3":
encoder = "libmp3lame"
case "ac3", "eac3", "alac":
default:
return nil, fmt.Errorf("暂不支持将少数音轨转为多数使用的 %s", info.audioCodec)
}
if info.sampleRate <= 0 || info.layout == "" {
return nil, fmt.Errorf("多数音轨缺少采样率或声道布局,无法安全转换")
}
args := []string{"-c:a", encoder, "-ar", strconv.Itoa(info.sampleRate), "-channel_layout:a", info.layout}
if encoder == "aac" {
args = append(args, "-profile:a", "aac_low", "-b:a", "192k")
}
return args, nil
}
func (plan mergeMediaPlan) prepareArgs(input, output string, info mergeMediaInfo, step mergeMediaStep) ([]string, error) {
args := []string{"-xerror", "-threads", "2", "-protocol_whitelist", "file,pipe", "-i", input}
if plan.audio.audio && !info.audio {
args = append(args, "-f", "lavfi", "-i", fmt.Sprintf("anullsrc=channel_layout=%s:sample_rate=%d", plan.audio.layout, plan.audio.sampleRate))
}
args = append(args, "-map", "0:v:0", "-sn", "-dn", "-map_metadata", "-1", "-c:v", "copy")
if step.video {
encoder, err := mergeVideoEncoder(plan.video)
if err != nil {
return nil, err
}
sar := plan.video.sar
if sar == "" || sar == "0/1" {
sar = "1/1"
}
filter := "setpts=PTS-STARTPTS,"
if plan.video.frameRate > 0 {
filter += "fps=" + strconv.FormatFloat(plan.video.frameRate, 'f', -1, 64) + ","
}
filter += fmt.Sprintf("scale=w='max(2,trunc(min(%d,%d*dar/(%s))/2)*2)':h='max(2,trunc(min(%d,%d*(%s)/dar)/2)*2)',pad=%d:%d:(ow-iw)/2:(oh-ih)/2,setsar=%s", plan.video.width, plan.video.height, sar, plan.video.height, plan.video.width, sar, plan.video.width, plan.video.height, sar)
args = append(args, "-vf", filter)
args = append(args, encoder...)
}
if plan.audio.audio {
stream := "0:a:0"
if !info.audio {
stream = "1:a:0"
}
args = append(args, "-map", stream)
if step.audio {
encoder, err := mergeAudioEncoder(plan.audio)
if err != nil {
return nil, err
}
args = append(args, encoder...)
args = append(args, "-af", fmt.Sprintf("aresample=%d:async=1:first_pts=0,apad", plan.audio.sampleRate))
} else {
args = append(args, "-c:a", "copy")
}
} else {
args = append(args, "-an")
}
args = append(args, "-max_muxing_queue_size", "4096")
if step.video || step.audio {
args = append(args, "-t", strconv.FormatFloat(info.duration.Seconds(), 'f', 6, 64))
}
if plan.transport {
args = append(args, "-bsf:v", "hevc_mp4toannexb", "-f", "mpegts")
} else {
args = append(args, "-video_track_timescale", strconv.Itoa(plan.timescale), "-f", "mp4")
}
return append(args, "-y", output), nil
}
+115
View File
@@ -0,0 +1,115 @@
package app
import (
"context"
"fmt"
"os/exec"
"path/filepath"
"regexp"
"strconv"
"strings"
"time"
)
type mergeMediaInfo struct {
width, height int
audio bool
duration time.Duration
videoCodec, audioCodec string
videoExtra, audioExtra string
videoTB, audioTB string
videoProfile, audioProfile string
pixelFormat, sar, layout string
sampleRate int
frameRate float64
reorder bool
}
var mergeVideoDescription = regexp.MustCompile(`Video: [a-zA-Z0-9_]+(?: \(([^)]*)\))?`)
var mergeAudioDescription = regexp.MustCompile(`Audio: [a-zA-Z0-9_]+(?: \(([^)]*)\))?`)
var mergePixelFormat = regexp.MustCompile(`, ([a-zA-Z0-9_]+)(?:\([^)]*\))?, \d+x\d+`)
var mergeFrameRate = regexp.MustCompile(`, ([0-9.]+) (?:fps|tbr)`)
var mergeStreamIndex = regexp.MustCompile(`Stream #0:(\d+)`)
func probeMergeMedia(ctx context.Context, ffmpeg, path string) (mergeMediaInfo, error) {
probeCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
defer cancel()
command := exec.CommandContext(probeCtx, ffmpeg, "-hide_banner", "-nostdin", "-loglevel", "info",
"-protocol_whitelist", "file,pipe", "-i", path, "-map", "0:v:0", "-map", "0:a:0?",
"-c", "copy", "-t", "0", "-f", "framehash", "-hash", "sha256", "pipe:1")
output := &cappedStringWriter{limit: 64 * 1024}
diagnostics := &cappedStringWriter{limit: 64 * 1024}
command.Stdout, command.Stderr = output, diagnostics
if err := command.Run(); err != nil {
if probeCtx.Err() != nil {
return mergeMediaInfo{}, probeCtx.Err()
}
return mergeMediaInfo{}, fmt.Errorf("检测 %s 失败:%w %s", filepath.Base(path), err, truncate(diagnostics.String(), 1000))
}
var info mergeMediaInfo
fields := make(map[string]string)
for _, line := range strings.Split(output.String(), "\n") {
if key, value, ok := strings.Cut(strings.TrimSpace(line), ":"); ok {
fields[key] = strings.TrimSpace(value)
}
if strings.HasPrefix(line, "#extradata ") {
parts := strings.Split(line, ",")
if len(parts) == 3 {
if strings.TrimSpace(parts[0]) == "#extradata 0" {
info.videoExtra = strings.TrimSpace(parts[2])
} else if strings.TrimSpace(parts[0]) == "#extradata 1" {
info.audioExtra = strings.TrimSpace(parts[2])
}
}
}
}
_, _ = fmt.Sscanf(fields["#dimensions 0"], "%dx%d", &info.width, &info.height)
info.audio = fields["#media_type 1"] == "audio"
info.videoCodec, info.audioCodec = fields["#codec_id 0"], fields["#codec_id 1"]
info.videoTB, info.audioTB = fields["#tb 0"], fields["#tb 1"]
info.sar = fields["#sar 0"]
if info.sar == "" || info.sar == "0/1" {
info.sar = "1/1"
}
info.sampleRate, _ = strconv.Atoi(fields["#sample_rate 1"])
info.layout = firstNonEmpty(fields["#channel_layout_name 1"], fields["#channel_layout 1"])
body, _, _ := strings.Cut(diagnostics.String(), "Stream mapping:")
if match := playbackDurationPattern.FindStringSubmatch(body); len(match) == 4 {
hours, _ := strconv.ParseFloat(match[1], 64)
minutes, _ := strconv.ParseFloat(match[2], 64)
seconds, _ := strconv.ParseFloat(match[3], 64)
info.duration = time.Duration((hours*3600 + minutes*60 + seconds) * float64(time.Second))
}
streams, video, audio := 0, false, false
for _, line := range strings.Split(body, "\n") {
index := mergeStreamIndex.FindStringSubmatch(line)
if len(index) != 2 {
continue
}
streams++
if strings.Contains(line, "Video:") && !video {
video = true
info.reorder = info.reorder || index[1] != "0"
if match := mergeVideoDescription.FindStringSubmatch(line); len(match) == 2 {
info.videoProfile = match[1]
}
if match := mergePixelFormat.FindStringSubmatch(line); len(match) == 2 {
info.pixelFormat = match[1]
}
if match := mergeFrameRate.FindStringSubmatch(line); len(match) == 2 {
info.frameRate, _ = strconv.ParseFloat(match[1], 64)
}
} else if strings.Contains(line, "Audio:") && !audio {
audio = true
info.reorder = info.reorder || index[1] != "1"
if match := mergeAudioDescription.FindStringSubmatch(line); len(match) == 2 {
info.audioProfile = match[1]
}
}
}
info.reorder = info.reorder || streams > 2
if info.width <= 0 || info.height <= 0 || info.duration <= 0 || info.videoCodec == "" {
return mergeMediaInfo{}, fmt.Errorf("%s 缺少有效的视频编码、尺寸或时长,无法安全合并", filepath.Base(path))
}
return info, nil
}
+149
View File
@@ -0,0 +1,149 @@
package app
import (
"context"
"os"
"os/exec"
"path/filepath"
"reflect"
"strings"
"testing"
"time"
)
func mergeFixtureCommand(t *testing.T, ffmpeg string, args ...string) string {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
command := exec.CommandContext(ctx, ffmpeg, append([]string{"-hide_banner", "-nostdin", "-loglevel", "error", "-y"}, args...)...)
output, err := command.CombinedOutput()
if err != nil {
t.Fatalf("synthetic merge fixture: %v\n%s", err, output)
}
return string(output)
}
func mergeFixtureFrames(t *testing.T, ffmpeg, path string) []string {
t.Helper()
output := mergeFixtureCommand(t, ffmpeg, "-i", path, "-map", "0:v:0", "-fps_mode", "passthrough", "-f", "framemd5", "-")
var frames []string
for _, line := range strings.Split(output, "\n") {
if line == "" || strings.HasPrefix(line, "#") {
continue
}
parts := strings.Split(line, ",")
frames = append(frames, strings.TrimSpace(parts[len(parts)-1]))
}
return frames
}
func TestMergePreservesMajorityAndOnlyConvertsIncompatibleStreams(t *testing.T) {
ffmpeg, err := exec.LookPath("ffmpeg")
if err != nil {
t.Skip("local FFmpeg unavailable")
}
work := t.TempDir()
base := filepath.Join(work, "majority.mp4")
mergeFixtureCommand(t, ffmpeg, "-f", "lavfi", "-i", "testsrc2=size=320x180:rate=25:duration=2", "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=2", "-c:v", "libx264", "-preset", "veryfast", "-threads:v", "2", "-pix_fmt", "yuv420p", "-c:a", "aac", "-ac", "2", base)
variants := map[string][]string{
"timebase": {"-c", "copy", "-video_track_timescale", "90000"},
"audio": {"-c:v", "copy", "-c:a", "aac", "-ar", "44100"},
"small": {"-c:v", "libx264", "-preset", "veryfast", "-threads:v", "2", "-vf", "scale=240:136", "-c:a", "copy"},
"parameters": {"-c:v", "libx264", "-preset", "fast", "-threads:v", "2", "-c:a", "copy"},
"silent": {"-c:v", "copy", "-an"},
"reordered": {"-map", "0:a:0", "-map", "0:v:0", "-c", "copy"},
}
for name, args := range variants {
mergeFixtureCommand(t, ffmpeg, append(append([]string{"-i", base}, args...), filepath.Join(work, name+".mp4"))...)
}
baseFrames := mergeFixtureFrames(t, ffmpeg, base)
for _, test := range []struct {
name string
video, audio int
}{
{"timebase", 0, 0}, {"parameters", 0, 0}, {"audio", 0, 1}, {"small", 1, 0}, {"silent", 0, 1}, {"reordered", 0, 0},
} {
t.Run(test.name, func(t *testing.T) {
minority := filepath.Join(work, test.name+".mp4")
paths := []string{minority, base, base}
var infos []mergeMediaInfo
for _, path := range paths {
info, err := probeMergeMedia(context.Background(), ffmpeg, path)
if err != nil {
t.Fatal(err)
}
infos = append(infos, info)
}
plan := planMergeMedia(infos)
if plan.videoCount != test.video || plan.audioCount != test.audio || plan.video.width != 320 || plan.video.height != 180 {
t.Fatalf("wrong majority/stream selection: %+v", plan)
}
output := filepath.Join(t.TempDir(), "merged.mp4")
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
method, err := mergeMediaFiles(ctx, ffmpeg, paths, output, nil)
if err != nil {
t.Fatal(err)
}
frames := mergeFixtureFrames(t, ffmpeg, output)
if len(frames) != 150 || !reflect.DeepEqual(frames[50:100], baseFrames) || !reflect.DeepEqual(frames[100:], baseFrames) {
t.Fatalf("majority frames changed or disappeared: frames=%d method=%s", len(frames), method)
}
if test.video == 0 && !reflect.DeepEqual(frames[:50], mergeFixtureFrames(t, ffmpeg, minority)) {
t.Fatal("copied minority video changed")
}
merged, err := probeMergeMedia(ctx, ffmpeg, output)
if err != nil || !merged.audio || merged.sampleRate != 48000 || merged.width != 320 || merged.height != 180 {
t.Fatalf("output format: %+v %v", merged, err)
}
})
}
t.Run("retain-real-audio-when-majority-silent", func(t *testing.T) {
silent := filepath.Join(work, "silent.mp4")
output := filepath.Join(t.TempDir(), "merged.mp4")
method, err := mergeMediaFiles(context.Background(), ffmpeg, []string{silent, base, silent}, output, nil)
if err != nil || !strings.Contains(method, "调整音轨 2/3") || strings.Contains(method, "转视频") {
t.Fatal("real audio was lost or silent majority video encoded", method, err)
}
info, err := probeMergeMedia(context.Background(), ffmpeg, output)
if err != nil || !info.audio {
t.Fatal(info, err)
}
})
t.Run("cancel-preserves-inputs", func(t *testing.T) {
before, _ := os.ReadFile(base)
ctx, cancel := context.WithCancel(context.Background())
cancel()
if _, err := mergeMediaFiles(ctx, ffmpeg, []string{base, base}, filepath.Join(t.TempDir(), "out.mp4"), nil); err == nil {
t.Fatal("canceled merge succeeded")
}
after, _ := os.ReadFile(base)
if !reflect.DeepEqual(before, after) {
t.Fatal("merge altered original episode")
}
})
t.Run("hevc-parameter-sets", func(t *testing.T) {
encoders := mergeFixtureCommand(t, ffmpeg, "-encoders")
if !strings.Contains(encoders, "libx265") {
t.Skip("HEVC encoder unavailable")
}
hevc := filepath.Join(work, "hevc.mp4")
mergeFixtureCommand(t, ffmpeg, "-i", base, "-c:v", "libx265", "-preset", "ultrafast", "-x265-params", "pools=1:frame-threads=1:log-level=error", "-c:a", "copy", hevc)
for _, majority := range []string{base, hevc} {
minority := hevc
if majority == hevc {
minority = filepath.Join(work, "small.mp4")
}
output := filepath.Join(t.TempDir(), "merged.mp4")
method, err := mergeMediaFiles(context.Background(), ffmpeg, []string{minority, majority, majority}, output, nil)
if err != nil || !strings.Contains(method, "仅转视频 1/3") {
t.Fatal(method, err)
}
frames := mergeFixtureFrames(t, ffmpeg, output)
want := mergeFixtureFrames(t, ffmpeg, majority)
if len(frames) != 150 || !reflect.DeepEqual(frames[50:100], want) || !reflect.DeepEqual(frames[100:], want) {
t.Fatalf("parameter changes corrupted copied majority frames: %d", len(frames))
}
}
})
}
+141
View File
@@ -0,0 +1,141 @@
package app
import (
"context"
"fmt"
"net/http"
"time"
)
type uiMergeJob struct {
dramaID string
tasks []*UITask
remove bool
ctx context.Context
cancel context.CancelFunc
done chan struct{}
result uiMergeResult
running bool
}
func (a *UIApp) enqueueMergesLocked(ids []string, deleteEpisodes bool) ([]*uiMergeJob, error) {
if a.mergeStopped {
return nil, fmt.Errorf("服务正在停止,无法提交合并")
}
if len(a.mergeJobs)+len(ids) > 256 {
return nil, fmt.Errorf("合并队列已满,请等待当前任务完成后再提交")
}
if a.mergeJobs == nil {
a.mergeJobs = make(map[string]*uiMergeJob)
}
if a.merges == nil {
a.merges = make(map[string]*UIMergeState)
}
byDrama := make(map[string][]*UITask)
for _, taskID := range a.taskOrder {
if task := a.tasks[taskID]; task != nil && task.Status == uiStatusSuccess && !task.RemoveRequested {
byDrama[task.DramaID] = append(byDrama[task.DramaID], cloneUITask(task))
}
}
previous := cloneMergeStates(a.merges)
var jobs, added []*uiMergeJob
for _, id := range ids {
job := a.mergeJobs[id]
if job == nil {
ctx, cancel := context.WithCancel(context.Background())
job = &uiMergeJob{dramaID: id, tasks: byDrama[id], remove: deleteEpisodes, ctx: ctx, cancel: cancel, done: make(chan struct{})}
a.mergeJobs[id] = job
added = append(added, job)
title := ""
if len(job.tasks) > 0 {
title = job.tasks[0].DramaTitle
}
a.merges[id] = &UIMergeState{DramaID: id, DramaTitle: title, Status: "queued", Detail: "已提交后台合并,可关闭或刷新页面", DeleteEpisodes: deleteEpisodes, UpdatedAt: time.Now()}
}
jobs = append(jobs, job)
}
if err := a.saveStateLocked(); err != nil {
for _, job := range added {
delete(a.mergeJobs, job.dramaID)
job.cancel()
}
a.merges = previous
return nil, fmt.Errorf("保存合并任务失败:%w", err)
}
if len(added) > 0 {
go a.runMergeJobs(added)
}
return jobs, nil
}
func (a *UIApp) runMergeJobs(jobs []*uiMergeJob) {
a.mergeMu.Lock()
defer a.mergeMu.Unlock()
for _, job := range jobs {
a.mu.Lock()
if a.mergeJobs[job.dramaID] != job {
a.mu.Unlock()
continue
}
job.running = true
a.mu.Unlock()
if job.ctx.Err() == nil {
job.result = a.mergeDrama(job.ctx, job.dramaID, job.tasks, job.remove)
} else {
job.result = uiMergeResult{DramaID: job.dramaID, Error: "合并已取消,原分集已保留"}
a.setMergeState(job.result, "failed", 0, false, job.remove)
}
job.cancel()
a.mu.Lock()
delete(a.mergeJobs, job.dramaID)
close(job.done)
a.mu.Unlock()
}
}
func (a *UIApp) stopMergeJobs() {
a.mu.Lock()
a.mergeStopped = true
var jobs []*uiMergeJob
for _, job := range a.mergeJobs {
job.cancel()
jobs = append(jobs, job)
}
a.mu.Unlock()
for _, job := range jobs {
<-job.done
}
}
func (a *UIApp) handleMergeCancel(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"})
return
}
if !requireDownload(w, r) {
return
}
ids, ok := readDramaIDsRequest(w, r)
if !ok || !a.requireDramaSources(w, r, ids) {
return
}
a.mu.Lock()
defer a.mu.Unlock()
canceled := 0
for _, id := range ids {
if job := a.mergeJobs[id]; job != nil {
job.cancel()
if !job.running {
job.result = uiMergeResult{DramaID: id, Error: "合并已取消,原分集已保留"}
if state := a.merges[id]; state != nil {
state.Status, state.Detail, state.Error = "failed", "", job.result.Error
}
delete(a.mergeJobs, id)
close(job.done)
}
canceled++
}
}
_ = a.saveStateLocked()
writeJSON(w, http.StatusOK, map[string]int{"canceled": canceled})
}
+110
View File
@@ -0,0 +1,110 @@
package app
import (
"bytes"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"sync"
"testing"
"time"
)
func TestMergeJobsSurviveRequestCancellationAndDeduplicate(t *testing.T) {
for _, async := range []bool{false, true} {
t.Run(map[bool]string{false: "curl-compatible", true: "async-page"}[async], func(t *testing.T) {
app, session := nativePlaybackFixture(t, "2", "320x180")
app.taskOrder = session.downloadIDs
app.statePath = filepath.Join(app.cfg.dataDirectory(), "ui-state.json")
app.downloader.ffmpegInstaller = &ffmpegInstaller{state: ffmpegInstallState{Status: "ready", Path: app.downloader.cfg.FFmpeg}}
app.cfg.SkipBytes = 1
app.mergeMu.Lock()
var once sync.Once
unlock := func() { once.Do(app.mergeMu.Unlock) }
defer func() { unlock(); app.stopMergeJobs() }()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
body, _ := json.Marshal(map[string]any{"dramaIds": []string{historyFixtureDramaID}, "async": async})
request := httptest.NewRequest(http.MethodPost, "/api/ui/merge", bytes.NewReader(body)).WithContext(ctx)
writer := httptest.NewRecorder()
returned := make(chan struct{})
go func() { app.handleMerge(writer, request); close(returned) }()
var job *uiMergeJob
deadline := time.Now().Add(5 * time.Second)
for job == nil && time.Now().Before(deadline) {
app.mu.Lock()
job = app.mergeJobs[historyFixtureDramaID]
app.mu.Unlock()
time.Sleep(time.Millisecond)
}
if job == nil {
t.Fatal("merge was not queued")
}
cancel()
select {
case <-returned:
case <-time.After(time.Second):
t.Fatal("canceled request remained tied to the merge")
}
if async && writer.Code != http.StatusAccepted || job.ctx.Err() != nil {
t.Fatal("request canceled its background merge", writer.Code, job.ctx.Err())
}
app.mu.Lock()
duplicate, err := app.enqueueMergesLocked([]string{historyFixtureDramaID}, true)
app.mu.Unlock()
if err != nil || len(duplicate) != 1 || duplicate[0] != job || job.remove {
t.Fatal("duplicate submission changed or duplicated the original merge", err)
}
unlock()
select {
case <-job.done:
case <-time.After(30 * time.Second):
t.Fatal("detached merge did not finish")
}
if !job.result.OK {
t.Fatal("background merge failed", job.result.Error)
}
if _, err := os.Stat(job.result.OutputPath); err != nil {
t.Fatal("background output missing", err)
}
if _, err := os.Stat(session.tasks[0].OutPath); err != nil {
t.Fatal("duplicate submission deleted original episodes", err)
}
})
}
}
func TestMergeQueueCancelAndRestartAreExplicit(t *testing.T) {
app := viewerTestApp(t)
app.statePath = filepath.Join(app.cfg.dataDirectory(), "ui-state.json")
app.mergeMu.Lock()
var once sync.Once
unlock := func() { once.Do(app.mergeMu.Unlock) }
defer func() { unlock(); app.stopMergeJobs() }()
app.mu.Lock()
jobs, err := app.enqueueMergesLocked([]string{historyFixtureDramaID}, false)
app.mu.Unlock()
if err != nil {
t.Fatal(err)
}
restored := &UIApp{cfg: app.cfg, statePath: app.statePath, tasks: map[string]*UITask{}}
restored.loadState()
if state := restored.merges[historyFixtureDramaID]; state == nil || state.Status != "failed" || state.Error == "" {
t.Fatal("restart left an interrupted merge permanently queued", state)
}
writer := historyJSONRequest(t, app, app.handleMergeCancel, "/api/ui/merge/cancel", map[string]any{"dramaIds": []string{historyFixtureDramaID}})
if writer.Code != http.StatusOK {
t.Fatal(writer.Code, writer.Body.String())
}
select {
case <-jobs[0].done:
case <-time.After(time.Second):
t.Fatal("queued cancellation waited for an unrelated active merge")
}
if jobs[0].result.OK || app.merges[historyFixtureDramaID].Status != "failed" {
t.Fatal("canceled merge reported success")
}
}
+21 -9
View File
@@ -51,6 +51,7 @@ type playbackSession struct {
}
type playbackEpisode struct {
Merged bool `json:"merged,omitempty"`
VIP bool `json:"vip,omitempty"`
Index int `json:"index"`
Episode string `json:"episode"`
@@ -190,15 +191,22 @@ func (app *UIApp) handlePlaybackOpen(writer http.ResponseWriter, request *http.R
var collectionErr error
app.mu.Lock()
if input.FromHistory && previous.Mode == "collection" && downloadAllowed(request.Context()) {
candidate := app.tasks[previous.TaskID]
if isPlayableDownloadTask(candidate) && candidate.DramaID == previous.DramaID {
input.TaskID = candidate.ID
} else {
for _, taskID := range app.taskOrder {
candidate = app.tasks[taskID]
if isPlayableDownloadTask(candidate) && candidate.DramaID == previous.DramaID {
input.TaskID = candidate.ID
break
if strings.HasPrefix(previous.TaskID, mergedPlaybackPrefix) && previous.TaskID == mergedPlaybackPrefix+previous.DramaID {
if _, _, err := app.mergedPlaybackTaskLocked(previous.TaskID); err == nil {
input.TaskID = previous.TaskID
}
}
if input.TaskID == "" {
candidate := app.tasks[previous.TaskID]
if isPlayableDownloadTask(candidate) && candidate.DramaID == previous.DramaID {
input.TaskID = candidate.ID
} else {
for _, taskID := range app.taskOrder {
candidate = app.tasks[taskID]
if isPlayableDownloadTask(candidate) && candidate.DramaID == previous.DramaID {
input.TaskID = candidate.ID
break
}
}
}
}
@@ -310,6 +318,10 @@ func (app *UIApp) handlePlaybackOpen(writer http.ResponseWriter, request *http.R
_, _, episode.Danmaku = hongguoPlaybackIDs(task)
if len(downloadIDs) > 0 {
episode.TaskID = downloadIDs[index]
episode.Merged = strings.HasPrefix(episode.TaskID, mergedPlaybackPrefix)
if episode.Merged {
episode.Danmaku = false
}
}
episodes = append(episodes, episode)
}
+16
View File
@@ -26,6 +26,13 @@ func hasCollectionPlaybackMedia(task Task) bool {
}
func (app *UIApp) collectionPlaybackTasksLocked(selectedID string) (string, []Task, []string, int, error) {
if strings.HasPrefix(selectedID, mergedPlaybackPrefix) {
task, _, err := app.mergedPlaybackTaskLocked(selectedID)
if err != nil {
return "", nil, nil, 0, err
}
return task.DramaTitle, []Task{task}, []string{selectedID}, 1, nil
}
selected := app.tasks[selectedID]
if !isPlayableDownloadTask(selected) {
return "", nil, nil, 0, errors.New("此分集不存在或尚未完成章节解析,请更新合集后再播放")
@@ -89,6 +96,9 @@ func completedPlaybackPath(task *UITask) string {
func (app *UIApp) playbackCollectionTask(id string) (Task, string, error) {
app.mu.Lock()
defer app.mu.Unlock()
if strings.HasPrefix(id, mergedPlaybackPrefix) {
return app.mergedPlaybackTaskLocked(id)
}
task := app.tasks[id]
if !isPlayableDownloadTask(task) {
return Task{}, "", errors.New("分集任务已被清理或正在移除,请重新打开合集")
@@ -105,6 +115,12 @@ type collectionPlaybackPreparation struct {
func (app *UIApp) prepareCollectionDownload(ctx context.Context, id string) (collectionPlaybackPreparation, error) {
app.mu.Lock()
defer app.mu.Unlock()
if strings.HasPrefix(id, mergedPlaybackPrefix) {
if _, _, err := app.mergedPlaybackTaskLocked(id); err != nil {
return collectionPlaybackPreparation{}, err
}
return collectionPlaybackPreparation{Source: "merged", DownloadStatus: uiStatusSuccess}, ctx.Err()
}
task := app.tasks[id]
if !isPlayableDownloadTask(task) {
return collectionPlaybackPreparation{}, errors.New("分集任务已被清理或正在移除,请重新打开合集")
+30
View File
@@ -0,0 +1,30 @@
package app
import (
"encoding/json"
"errors"
"fmt"
"strings"
)
const mergedPlaybackPrefix = "merged:"
func (app *UIApp) mergedPlaybackTaskLocked(id string) (Task, string, error) {
dramaID, ok := strings.CutPrefix(id, mergedPlaybackPrefix)
if !ok || dramaID == "" {
return Task{}, "", errors.New("合并播放记录无效")
}
state := app.merges[dramaID]
if state == nil || state.Status != "success" {
return Task{}, "", errors.New("本剧尚无合并成功的文件")
}
path := completedPlaybackPath(&UITask{Status: uiStatusSuccess, Path: state.OutputPath})
if path == "" {
return Task{}, "", errors.New("本地合并文件不存在或已移动,请重新合并")
}
label := fmt.Sprintf("全集(第%d–%d集)", state.StartEpisode, state.EndEpisode)
episode, _ := json.Marshal(label)
task := Task{DramaID: dramaID, DramaTitle: state.DramaTitle, Index: 1, Total: 1, OutPath: path,
Chapter: Chapter{ID: fmt.Sprintf("%s:%d-%d", id, state.StartEpisode, state.EndEpisode), Source: sourceFromDramaID(dramaID), Title: label, CurrentEpisode: episode}}
return task, path, nil
}
@@ -0,0 +1,67 @@
package app
import (
"encoding/json"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"sync"
"testing"
"time"
)
func TestMergedPlaybackBrowserFixture(t *testing.T) {
root := os.Getenv("JUKU_MERGED_BROWSER_ROOT")
if root == "" {
t.Skip("isolated browser fixture is opt-in")
}
if err := os.MkdirAll(root, 0700); err != nil {
t.Fatal(err)
}
ffmpeg, err := exec.LookPath("ffmpeg")
if err != nil {
t.Fatal(err)
}
video := filepath.Join(root, "synthetic-full.mp4")
mergeFixtureCommand(t, ffmpeg, "-f", "lavfi", "-i", "testsrc2=size=320x180:rate=25:duration=12", "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=12", "-c:v", "libx264", "-preset", "veryfast", "-threads:v", "2", "-c:a", "aac", "-ac", "2", "-movflags", "+faststart", video)
app := viewerTestApp(t)
app.loadedAt, app.libraryAttempted = time.Now(), true
app.statePath = filepath.Join(app.cfg.dataDirectory(), "ui-state.json")
app.tasks = map[string]*UITask{}
app.cond = sync.NewCond(&app.mu)
app.merges = map[string]*UIMergeState{historyFixtureDramaID: {DramaID: historyFixtureDramaID, DramaTitle: "本地合成全集", Status: "success", Progress: 100, StartEpisode: 1, EndEpisode: 3, Merged: 3, OutputPath: video}}
app.downloader.ffmpegInstaller = &ffmpegInstaller{state: ffmpegInstallState{Status: "ready", Path: ffmpeg}}
app.cfg.SkipBytes = 1
for index, id := range []string{"synthetic-one", "synthetic-two"} {
task := Task{DramaID: "hongguo:7000000000000000005", DramaTitle: "待合并的本地分集", Index: index + 1, Total: 2, OutPath: video, Chapter: Chapter{Source: sourceHongguo, CurrentEpisode: rawEpisode(index + 1)}}
app.tasks[id] = &UITask{ID: id, DramaID: task.DramaID, DramaTitle: task.DramaTitle, Path: video, Status: uiStatusSuccess, Source: task}
app.taskOrder = append(app.taskOrder, id)
}
app.cfg.adminUsername, app.cfg.adminPassword, app.cfg.adminPasswordExplicit = "admin", accountFixturePassword, true
if err := app.prepareBrowserViewers(); err != nil {
t.Fatal(err)
}
t.Cleanup(app.embySyncer().stop)
t.Cleanup(app.stopMergeJobs)
server := httptest.NewServer(app.routes())
defer server.Close()
body, _ := json.Marshal(map[string]string{"url": server.URL})
if err := os.WriteFile(filepath.Join(root, "ready.json"), body, 0600); err != nil {
t.Fatal(err)
}
deadline := time.NewTimer(5 * time.Minute)
defer deadline.Stop()
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if _, err := os.Stat(filepath.Join(root, "stop")); err == nil {
return
}
case <-deadline.C:
t.Fatal("merged playback browser fixture timed out")
}
}
}
+123
View File
@@ -0,0 +1,123 @@
package app
import (
"bytes"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
)
func TestMergedPlaybackLocalRangeResumeWithoutEpisodes(t *testing.T) {
app, fixture := nativePlaybackFixture(t, "6", "320x180")
path := fixture.tasks[0].OutPath
dramaID := fixture.tasks[0].DramaID
id := mergedPlaybackPrefix + dramaID
app.merges = map[string]*UIMergeState{dramaID: {DramaID: dramaID, DramaTitle: "合成全集", Status: "success", StartEpisode: 1, EndEpisode: 3, Merged: 3, OutputPath: path}}
app.taskOrder = append([]string(nil), fixture.downloadIDs...)
for _, taskID := range fixture.downloadIDs {
app.removeTaskLocked(taskID)
}
if len(app.tasks) != 0 || app.merges[dramaID] == nil {
t.Fatal("clearing the final episode task removed the merged file record")
}
app.statePath = filepath.Join(app.cfg.dataDirectory(), "ui-state.json")
app.downloader.cfg.FFmpeg = filepath.Join(t.TempDir(), "ffmpeg-must-not-run")
app.mediaResources().settings.Enabled = false
app.mu.Lock()
err := app.saveStateLocked()
app.mu.Unlock()
if err != nil {
t.Fatal(err)
}
restored := &UIApp{cfg: app.cfg, statePath: app.statePath, tasks: map[string]*UITask{}}
restored.loadState()
if _, local, err := restored.mergedPlaybackTaskLocked(id); err != nil || local != path {
t.Fatal("restart lost merged file without episode tasks", local, err)
}
open := func(body any) map[string]json.RawMessage {
t.Helper()
result := historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", body)
if result.Code != http.StatusOK {
t.Fatal(result.Code, result.Body.String())
}
var response map[string]json.RawMessage
if err := json.Unmarshal(result.Body.Bytes(), &response); err != nil {
t.Fatal(err)
}
return response
}
opened := open(map[string]any{"taskId": id, "resume": true})
var sessionID string
_ = json.Unmarshal(opened["session"], &sessionID)
defer app.closePlaybacks()
var episodes []playbackEpisode
_ = json.Unmarshal(opened["episodes"], &episodes)
if len(episodes) != 1 || !episodes[0].Merged || episodes[0].TaskID != id || episodes[0].Danmaku {
t.Fatal("merged file did not appear as a standalone full-length selection", string(opened["episodes"]))
}
preparation := historyJSONRequest(t, app, app.handlePlaybackPrepare, "/api/ui/playback/prepare", map[string]any{"session": sessionID, "episode": 1})
if preparation.Code != http.StatusOK || !strings.Contains(preparation.Body.String(), `"source":"merged"`) || len(app.tasks) != 0 {
t.Fatal("merged playback attempted episode downloads", preparation.Code, preparation.Body.String())
}
result := historyJSONRequest(t, app, app.handlePlaybackPlan, "/api/ui/playback/plan", map[string]any{"session": sessionID, "episode": 1, "version": 1, "start": 2.0,
"client": playbackClientCapabilities{MP4: true, Video: []string{"h264"}, Audio: []string{"aac"}}})
var plan playbackMediaPlan
if result.Code != http.StatusOK || json.Unmarshal(result.Body.Bytes(), &plan) != nil || plan.Delivery != "local" || plan.Player != "mp4" || plan.Processing != "original" {
t.Fatal("local merged playback depended on the online optimization flag or FFmpeg", result.Code, result.Body.String())
}
body, _ := os.ReadFile(path)
for _, test := range []struct {
method, span string
code, length int
}{{"GET", "bytes=0-15", 206, 16}, {"GET", "bytes=-16", 206, 16}, {"HEAD", "", 200, 0}} {
request := httptest.NewRequest(test.method, "http://localhost"+plan.URL, nil)
request.Header.Set("Range", test.span)
writer := httptest.NewRecorder()
app.handlePlaybackMediaAsset(writer, viewerFixtureRequest(app, request))
if writer.Code != test.code || writer.Body.Len() != test.length || test.span == "bytes=-16" && !bytes.Equal(writer.Body.Bytes(), body[len(body)-16:]) {
t.Fatal("merged file did not serve local byte ranges", writer.Code, writer.Body.Len())
}
}
session := app.playbacks[sessionID]
if _, saved, err := app.recordPlaybackProgress(fixtureViewer(app), sessionID, playbackHistoryProgress{Run: session.run, Sequence: 1, Episode: 1, Position: 2.5, Duration: 6}); err != nil || !saved {
t.Fatal("merged progress was not saved", saved, err)
}
app.closePlayback(sessionID)
resumed := open(map[string]any{"dramaId": dramaID, "resume": true, "fromHistory": true})
if string(resumed["initialPosition"]) != "2.5" || !bytes.Contains(resumed["episodes"], []byte(`"merged":true`)) {
t.Fatal("merged history returned to online episodes or lost the full-file timestamp", string(resumed["initialPosition"]), string(resumed["episodes"]))
}
for _, scope := range []accountRecord{{OnlineOnly: true}, {Sources: []string{"no-access"}}} {
body, _ := json.Marshal(map[string]string{"taskId": id})
request := viewerFixtureRequest(app, httptest.NewRequest(http.MethodPost, "/api/ui/playback/open", bytes.NewReader(body)))
request.Header.Set("Content-Type", "application/json")
request = request.WithContext(withSourceScope(request.Context(), scope))
writer := httptest.NewRecorder()
app.handlePlaybackOpen(writer, request)
if writer.Code != http.StatusForbidden {
t.Fatal("merged playback bypassed source/download permissions", scope, writer.Code, writer.Body.String())
}
}
other := app.browserViewers().acquire(viewerID("different-merged-viewer"))
defer other.release()
request := httptest.NewRequest(http.MethodGet, "http://localhost"+plan.URL, nil)
writer := httptest.NewRecorder()
app.handlePlaybackMediaAsset(writer, request.WithContext(context.WithValue(request.Context(), viewerContextKey{}, other)))
if writer.Code != http.StatusGone {
t.Fatal("merged playback leaked to another viewer", writer.Code)
}
if err := os.Remove(path); err != nil {
t.Fatal(err)
}
if _, _, err := app.playbackCollectionTask(id); err == nil {
t.Fatal("missing merged file fell back to upstream video")
}
if _, err := app.prepareCollectionDownload(context.Background(), id); err == nil || len(app.tasks) != 0 {
t.Fatal("missing merged file created download tasks", err)
}
}
+8 -2
View File
@@ -167,8 +167,14 @@ func (app *UIApp) handlePlaybackPlan(writer http.ResponseWriter, request *http.R
return
}
if !app.mediaResources().snapshot().Enabled {
writeJSON(writer, http.StatusOK, playbackMediaPlan{Player: "legacy", Reason: "optimization_disabled"})
return
app.playbackMu.Lock()
session := app.playbacks[input.Session]
merged := session != nil && input.Episode <= len(session.downloadIDs) && strings.HasPrefix(session.downloadIDs[input.Episode-1], mergedPlaybackPrefix)
app.playbackMu.Unlock()
if !merged {
writeJSON(writer, http.StatusOK, playbackMediaPlan{Player: "legacy", Reason: "optimization_disabled"})
return
}
}
run, status, err := app.beginPlayback(context.WithoutCancel(request.Context()), input.Session, input.Episode, input.Start, input.Quality, input.Version, false)
if err != nil {
+44 -26
View File
@@ -67,6 +67,8 @@ type UIApp struct {
address string
runningCancels map[string]context.CancelFunc
mergeMu sync.Mutex
mergeJobs map[string]*uiMergeJob
mergeStopped bool
parsingInFlight map[string]bool
parsingCancels map[string]context.CancelFunc
closed bool
@@ -103,6 +105,7 @@ type uiState struct {
}
type UIMergeState struct {
PlaybackTaskID string `json:"playbackTaskId,omitempty"`
DramaID string `json:"dramaId"`
DramaTitle string `json:"dramaTitle,omitempty"`
Status string `json:"status"`
@@ -240,6 +243,7 @@ func (a *UIApp) ListenAndServe(addr string) error {
a.mu.Unlock()
defer a.closePlaybacks()
defer a.stopMergeJobs()
server := &http.Server{Addr: addr, Handler: a.routes(), ReadHeaderTimeout: 10 * time.Second}
return server.ListenAndServe()
@@ -277,6 +281,7 @@ func (a *UIApp) routes() http.Handler {
mux.HandleFunc("/api/ui/tasks/resume", a.handleResume)
mux.HandleFunc("/api/ui/tasks/clear", a.handleClear)
mux.HandleFunc("/api/ui/merge", a.handleMerge)
mux.HandleFunc("/api/ui/merge/cancel", a.handleMergeCancel)
mux.HandleFunc("/api/ui/ffmpeg", a.handleFFmpeg)
mux.HandleFunc("/api/ui/config", a.handleConfig)
mux.HandleFunc("/api/ui/network/check", a.handleNetworkCheck)
@@ -529,6 +534,12 @@ func (a *UIApp) loadState() {
}
seen := map[string]bool{}
changed := false
for _, state := range a.merges {
if state != nil && (state.Status == "running" || state.Status == "queued") {
state.Status, state.Detail, state.Error = "failed", "", "服务已重启,合并已中断;原分集已保留,可重新合并"
changed = true
}
}
for _, task := range state.Tasks {
if task == nil || task.RemoveRequested {
changed = true
@@ -1372,6 +1383,11 @@ func (a *UIApp) handleTasks(w http.ResponseWriter, r *http.Request) {
for id := range merges {
if !dramaAllowed(r.Context(), id, "") {
delete(merges, id)
continue
}
merges[id].PlaybackTaskID = ""
if _, _, err := a.mergedPlaybackTaskLocked(mergedPlaybackPrefix + id); err == nil {
merges[id].PlaybackTaskID = mergedPlaybackPrefix + id
}
}
a.mu.Unlock()
@@ -1548,37 +1564,33 @@ func (a *UIApp) handleMerge(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"})
return
}
ids, deleteEpisodes, ok := readMergeRequest(w, r)
if !ok {
ids, deleteEpisodes, async, ok := readMergeRequest(w, r)
if !ok || !requireDownload(w, r) || !a.requireDramaSources(w, r, ids) {
return
}
if !a.requireDramaSources(w, r, ids) {
return
}
a.mergeMu.Lock()
defer a.mergeMu.Unlock()
a.mu.Lock()
if !a.requireTaskSourcesLocked(w, r, taskSelection{DramaIDs: ids}) {
a.mu.Unlock()
return
}
byDrama := map[string][]*UITask{}
for _, id := range a.taskOrder {
task := a.tasks[id]
if task == nil || task.Status != uiStatusSuccess {
continue
}
byDrama[task.DramaID] = append(byDrama[task.DramaID], cloneUITask(task))
}
jobs, err := a.enqueueMergesLocked(ids, deleteEpisodes)
a.mu.Unlock()
results := make([]uiMergeResult, 0, len(ids))
for _, dramaID := range ids {
items := byDrama[dramaID]
res := a.mergeDrama(r.Context(), dramaID, items, deleteEpisodes)
results = append(results, res)
if err != nil {
writeJSON(w, http.StatusConflict, map[string]string{"error": a.redactError(err)})
return
}
if async {
writeJSON(w, http.StatusAccepted, map[string]any{"accepted": len(jobs), "async": true})
return
}
results := make([]uiMergeResult, 0, len(jobs))
for _, job := range jobs {
select {
case <-r.Context().Done():
return
case <-job.done:
results = append(results, job.result)
}
}
writeJSON(w, http.StatusOK, map[string]any{"data": results})
}
@@ -1673,6 +1685,11 @@ func (a *UIApp) mergeDrama(ctx context.Context, dramaID string, tasks []*UITask,
return res
}
res.Detail = method
if err := cmdCtx.Err(); err != nil {
res.Error = "合并已取消,原分集已保留"
a.setMergeState(res, "failed", 0, false, deleteEpisodes)
return res
}
if ok, _ := existingGood(partPath, a.cfg.SkipBytes); !ok {
res.Error = "合并输出文件无效"
a.setMergeState(res, "failed", 0, false, deleteEpisodes)
@@ -1731,8 +1748,9 @@ func (a *UIApp) setMergeState(res uiMergeResult, status string, progress int, sk
a.mu.Unlock()
}
func readMergeRequest(w http.ResponseWriter, r *http.Request) ([]string, bool, bool) {
func readMergeRequest(w http.ResponseWriter, r *http.Request) ([]string, bool, bool, bool) {
var req struct {
Async bool `json:"async"`
DramaIDs []string `json:"dramaIds"`
DeleteEpisodes bool `json:"deleteEpisodes"`
}
@@ -1741,10 +1759,10 @@ func readMergeRequest(w http.ResponseWriter, r *http.Request) ([]string, bool, b
dec.DisallowUnknownFields()
if err := dec.Decode(&req); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json: " + err.Error()})
return nil, false, false
return nil, false, false, false
}
ids, ok := cleanIDList(w, req.DramaIDs)
return ids, req.DeleteEpisodes, ok
return ids, req.DeleteEpisodes, req.Async, ok
}
func (a *UIApp) handleConfig(w http.ResponseWriter, r *http.Request) {
+3 -1
View File
@@ -45,7 +45,9 @@ func (a *UIApp) removeTaskLocked(id string) {
}
a.taskOrder = order
if task != nil && !groupRemains {
delete(a.merges, task.DramaID)
if _, _, err := a.mergedPlaybackTaskLocked(mergedPlaybackPrefix + task.DramaID); err != nil && a.mergeJobs[task.DramaID] == nil {
delete(a.merges, task.DramaID)
}
}
}