diff --git a/internal/app/media_merge.go b/internal/app/media_merge.go new file mode 100644 index 0000000..b00e250 --- /dev/null +++ b/internal/app/media_merge.go @@ -0,0 +1,180 @@ +package app + +import ( + "bufio" + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "time" +) + +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", + }, args...)...) + diagnostics := &cappedStringWriter{limit: 64 * 1024} + command.Stderr = diagnostics + output, err := command.StdoutPipe() + if err != nil { + return err + } + defer output.Close() + if err := command.Start(); err != nil { + return err + } + scanner := bufio.NewScanner(output) + for scanner.Scan() { + key, value, ok := strings.Cut(scanner.Text(), "=") + if ok && key == "out_time_us" && progress != nil { + if micros, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64); err == nil && micros >= 0 { + progress(time.Duration(micros) * time.Microsecond) + } + } + } + err = command.Wait() + if ctx.Err() != nil { + return ctx.Err() + } + if err != nil { + return fmt.Errorf("FFmpeg 处理失败:%w %s", err, truncate(diagnostics.String(), 1500)) + } + return scanner.Err() +} + +func mergeMediaFiles(ctx context.Context, ffmpeg string, paths []string, output string, report func(int, string)) (string, error) { + if len(paths) == 0 { + return "", fmt.Errorf("没有可合并的分集") + } + lastProgress, lastDetail := -1, "" + notify := func(progress int, detail string) { + if progress > 99 { + progress = 99 + } + if progress < lastProgress { + progress = lastProgress + } + if report != nil && (progress != lastProgress || detail != lastDetail) { + report(progress, detail) + } + lastProgress, lastDetail = progress, detail + } + metadata := make([]mergeMediaInfo, 0, len(paths)) + var duration time.Duration + 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 + } + metadata = append(metadata, info) + duration += info.duration + } + 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...) + preparedArgs := make([][]string, len(paths)) + var preparedDuration, processed time.Duration + for index, step := range plan.steps { + if step.video || step.audio || step.remux { + preparedDuration += metadata[index].duration + extension := ".mp4" + if plan.transport { + extension = ".ts" + } + inputs[index] = filepath.Join(work, fmt.Sprintf("%06d%s", index+1, extension)) + preparedArgs[index], err = plan.prepareArgs(paths[index], inputs[index], metadata[index], step) + if err != nil { + return "", fmt.Errorf("无法准备 %s:%w;原分集已保留", filepath.Base(paths[index]), err) + } + } + } + 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) + err := runMergeFFmpeg(ctx, ffmpeg, preparedArgs[index], func(elapsed time.Duration) { + if elapsed > metadata[index].duration { + elapsed = metadata[index].duration + } + notify(5+int((processed+elapsed)*60/preparedDuration), detail) + }) + if err != nil { + return "", fmt.Errorf("处理 %s 失败:%w;原分集已保留", filepath.Base(paths[index]), err) + } + processed += metadata[index].duration + } + var manifest strings.Builder + 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(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(80, "核对合并文件时长") + merged, err := probeMergeMedia(ctx, ffmpeg, output) + if err != nil { + return "", err + } + tolerance := duration / 100 + if tolerance < 2*time.Second { + tolerance = 2 * time.Second + } + 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 +} diff --git a/internal/app/media_merge_audio_test.go b/internal/app/media_merge_audio_test.go new file mode 100644 index 0000000..7c2c9bb --- /dev/null +++ b/internal/app/media_merge_audio_test.go @@ -0,0 +1,107 @@ +package app + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "reflect" + "strings" + "testing" + "time" +) + +func TestMergeRealHEAACAudioPreservesVideo(t *testing.T) { + fixture := os.Getenv("JUKU_MERGE_HEAAC_FIXTURE") + if fixture == "" { + t.Skip("set JUKU_MERGE_HEAAC_FIXTURE to a local HE-AAC or HE-AACv2 sample; no network downloads") + } + ffmpeg, err := exec.LookPath("ffmpeg") + if err != nil { + t.Fatal(err) + } + work := t.TempDir() + base, minority := filepath.Join(work, "he-aac.mp4"), filepath.Join(work, "lc.mp4") + mergeFixtureCommand(t, ffmpeg, "-f", "lavfi", "-i", "testsrc2=size=320x180:rate=25:duration=2", "-i", fixture, + "-map", "0:v:0", "-map", "1:a:0", "-t", "2", "-c:v", "libx264", "-preset", "veryfast", "-threads:v", "2", "-pix_fmt", "yuv420p", "-c:a", "copy", base) + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + info, err := probeMergeMedia(ctx, ffmpeg, base) + if err != nil || !strings.HasPrefix(info.audioProfile, "HE-AAC") { + t.Fatalf("fixture is not real HE-AAC audio: %s %v", info.audioProfile, err) + } + mergeFixtureCommand(t, ffmpeg, "-i", base, "-c:v", "copy", "-c:a", "aac", "-profile:a", "aac_low", minority) + baseFrames := mergeFixtureFrames(t, ffmpeg, base) + for _, mixed := range []bool{false, true} { + name := "homogeneous" + paths := []string{base, base, base} + profile := info.audioProfile + if mixed { + name, paths[1], profile = "mixed", minority, "LC" + } + t.Run(name, func(t *testing.T) { + output := filepath.Join(work, name+"-merged.mp4") + method, err := mergeMediaFiles(ctx, ffmpeg, paths, output, nil) + if err != nil || strings.Contains(method, "转视频") || strings.Contains(method, "统一音轨为 AAC-LC") != mixed { + t.Fatal("real HE-AAC merge failed or unnecessarily encoded video", method, err) + } + merged, err := probeMergeMedia(ctx, ffmpeg, output) + if err != nil || merged.audioProfile != profile || merged.sampleRate != info.sampleRate { + t.Fatalf("unexpected merged audio: %+v %v", merged, err) + } + want := append(append(append([]string(nil), baseFrames...), baseFrames...), baseFrames...) + if frames := mergeFixtureFrames(t, ffmpeg, output); !reflect.DeepEqual(frames, want) { + t.Fatalf("copied video changed: got %d frames, want %d", len(frames), len(want)) + } + }) + } +} + +func TestMergeMixedHEAACUsesUniformLCAudioAndCopiesVideo(t *testing.T) { + for _, profile := range []string{"HE-AAC", "HE-AACv2"} { + t.Run(profile, func(t *testing.T) { + base := mergeMediaInfo{width: 1080, height: 1920, videoCodec: "h264", videoProfile: "High", + pixelFormat: "yuv420p", sar: "1/1", videoTB: "1/90000", videoExtra: "unchanged-video", + audio: true, audioCodec: "aac", audioProfile: profile, audioExtra: "he-configuration", + sampleRate: 48000, layout: "stereo", audioTB: "1/48000", duration: time.Minute} + minority := base + minority.audioProfile, minority.audioExtra = "LC", "lc-configuration" + infos := []mergeMediaInfo{base, minority, base} + plan := planMergeMedia(infos) + if plan.videoCount != 0 || plan.audioCount != len(infos) || !plan.audioFallback || plan.audio.audioProfile != "LC" { + t.Fatalf("mixed HE-AAC was not normalized without video encoding: %+v", plan) + } + for index, step := range plan.steps { + args, err := plan.prepareArgs("input.mp4", "output.mp4", infos[index], step) + joined := strings.Join(args, " ") + if err != nil || !strings.Contains(joined, "-c:v copy") || !strings.Contains(joined, "-profile:a aac_low") || + strings.Contains(joined, "libx264") || strings.Contains(joined, "-vf") { + t.Fatal("audio fallback encoded video or kept mixed AAC profiles", joined, err) + } + } + same := planMergeMedia([]mergeMediaInfo{base, base, base}) + if same.audioCount != 0 || same.videoCount != 0 || same.audioFallback { + t.Fatal("homogeneous HE-AAC should stay lossless", same.description()) + } + minority.audio = false + if silent := planMergeMedia([]mergeMediaInfo{base, minority, base}); silent.audioCount != 3 || silent.videoCount != 0 { + t.Fatal("silent episode did not normalize the full audio track") + } + }) + } +} + +func TestMergeFailureRetainsProgress(t *testing.T) { + app := &UIApp{statePath: filepath.Join(t.TempDir(), "ui-state.json")} + result := uiMergeResult{DramaID: historyFixtureDramaID, DramaTitle: "合成进度测试"} + app.setMergeState(result, "running", 41, false, false) + result.Error = "fixture failure" + app.setMergeState(result, "failed", 0, false, false) + if state := app.merges[result.DramaID]; state.Progress != 41 || state.Status != "failed" || state.Error == "" { + t.Fatal("failure reset the completed work", state) + } + app.setMergeState(result, "running", 1, false, false) + if app.merges[result.DramaID].Progress != 1 { + t.Fatal("a new attempt inherited the previous progress") + } +} diff --git a/internal/app/media_merge_plan.go b/internal/app/media_merge_plan.go new file mode 100644 index 0000000..abc0c1e --- /dev/null +++ b/internal/app/media_merge_plan.go @@ -0,0 +1,237 @@ +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 + audioFallback bool +} + +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 plan.audio.audio && plan.audio.audioCodec == "aac" && plan.audio.audioProfile != "" && plan.audio.audioProfile != "LC" { + for _, info := range infos { + if !info.audio || info.audioKey() != plan.audio.audioKey() { + + plan.audioFallback = true + plan.audio.audioProfile, plan.audio.audioExtra = "LC", "" + break + } + } + } + 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 && (plan.audioFallback || !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 { + if plan.audioFallback { + parts = append(parts, fmt.Sprintf("统一音轨为 AAC-LC %d/%d 集", plan.audioCount, len(plan.steps))) + } else { + 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 +} diff --git a/internal/app/media_merge_probe.go b/internal/app/media_merge_probe.go new file mode 100644 index 0000000..a45cc6d --- /dev/null +++ b/internal/app/media_merge_probe.go @@ -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 +} diff --git a/internal/app/media_merge_test.go b/internal/app/media_merge_test.go new file mode 100644 index 0000000..d8fbe63 --- /dev/null +++ b/internal/app/media_merge_test.go @@ -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)) + } + } + }) +} diff --git a/internal/app/media_playlist.go b/internal/app/media_playlist.go new file mode 100644 index 0000000..a3ded86 --- /dev/null +++ b/internal/app/media_playlist.go @@ -0,0 +1,49 @@ +package app + +import ( + "math" + "net/url" + "regexp" + "strconv" + "strings" + "time" +) + +func m3u8Duration(raw string) time.Duration { + var total time.Duration + for _, line := range strings.Split(raw, "\n") { + line = strings.TrimSpace(line) + if !strings.HasPrefix(line, "#EXTINF:") { + continue + } + value := strings.TrimSpace(strings.TrimPrefix(line, "#EXTINF:")) + if comma := strings.IndexByte(value, ','); comma >= 0 { + value = value[:comma] + } + seconds, err := strconv.ParseFloat(strings.TrimSpace(value), 64) + if err != nil || seconds < 0 || math.IsNaN(seconds) || math.IsInf(seconds, 0) { + continue + } + total += time.Duration(seconds * float64(time.Second)) + } + return total +} + +func rewriteM3U8(raw, playlistURL string) string { + base, err := url.Parse(playlistURL) + lines := strings.Split(raw, "\n") + for i, line := range lines { + trimmed := strings.TrimSpace(line) + if strings.Contains(trimmed, "URI=") && strings.Contains(trimmed, "/api/app/vid/sec") { + lines[i] = regexp.MustCompile(`URI="[^"]+"`).ReplaceAllString(line, `URI="key.bin"`) + continue + } + if err == nil && trimmed != "" && !strings.HasPrefix(trimmed, "#") && !strings.HasPrefix(trimmed, "http://") && !strings.HasPrefix(trimmed, "https://") { + ref, refErr := url.Parse(trimmed) + if refErr == nil { + lines[i] = base.ResolveReference(ref).String() + } + } + } + return strings.Join(lines, "\n") +} diff --git a/internal/app/media_resume.go b/internal/app/media_resume.go new file mode 100644 index 0000000..aed40fe --- /dev/null +++ b/internal/app/media_resume.go @@ -0,0 +1,230 @@ +package app + +import ( + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "time" +) + +type mediaByteRange struct { + start int64 + end int64 + total int64 +} + +func parseMediaByteRange(value string) (mediaByteRange, bool) { + var result mediaByteRange + unit, value, ok := strings.Cut(strings.TrimSpace(value), " ") + if !ok || !strings.EqualFold(unit, "bytes") { + return result, false + } + span, total, ok := strings.Cut(value, "/") + if !ok { + return result, false + } + start, end, ok := strings.Cut(span, "-") + if !ok { + return result, false + } + for index, field := range []string{start, end, total} { + if field == "" || strings.IndexFunc(field, func(char rune) bool { return char < '0' || char > '9' }) >= 0 { + return result, false + } + number, err := strconv.ParseInt(field, 10, 64) + if err != nil { + return result, false + } + switch index { + case 0: + result.start = number + case 1: + result.end = number + case 2: + result.total = number + } + } + return result, result.start <= result.end && result.end < result.total +} + +func mediaRequestAttempts(retries int) int { + return min(3, max(1, retries)) +} + +func identityMediaResponse(response *http.Response) bool { + encoding := strings.TrimSpace(response.Header.Get("Content-Encoding")) + return !response.Uncompressed && (encoding == "" || strings.EqualFold(encoding, "identity")) && + !strings.HasPrefix(strings.ToLower(response.Header.Get("Content-Type")), "multipart/") +} + +type resumableMediaReader struct { + proxy *hlsProxy + request *http.Request + body io.ReadCloser + span mediaByteRange + length int64 + read int64 + etag string + modified string + validator string + resumable bool + attempts int + maxAttempts int + pending error + failure error + recovered bool +} + +func (proxy *hlsProxy) mediaReader(request *http.Request, response *http.Response) *resumableMediaReader { + reader := &resumableMediaReader{ + proxy: proxy, request: request.Clone(request.Context()), body: response.Body, + length: -1, maxAttempts: mediaRequestAttempts(proxy.retries) - 1, + etag: strings.TrimSpace(response.Header.Get("ETag")), + modified: strings.TrimSpace(response.Header.Get("Last-Modified")), + } + if response.Request != nil && response.Request.URL != nil { + reader.request.URL = response.Request.URL + } + if response.ContentLength > 0 { + reader.length = response.ContentLength + } + if response.StatusCode == http.StatusOK && reader.length > 0 && response.Header.Get("Content-Range") == "" { + reader.span = mediaByteRange{end: reader.length - 1, total: reader.length} + reader.resumable = true + } else if response.StatusCode == http.StatusPartialContent { + if span, valid := parseMediaByteRange(response.Header.Get("Content-Range")); valid { + length := span.end - span.start + 1 + if response.ContentLength < 0 || response.ContentLength == length { + reader.span, reader.length, reader.resumable = span, length, true + } + } + } + reader.resumable = reader.resumable && request.Method == http.MethodGet && identityMediaResponse(response) + if len(reader.etag) >= 2 && strings.HasPrefix(reader.etag, `"`) && strings.HasSuffix(reader.etag, `"`) { + reader.validator = reader.etag + } else if _, err := http.ParseTime(reader.modified); err == nil { + reader.validator = reader.modified + } + return reader +} + +func (reader *resumableMediaReader) Close() error { + return reader.body.Close() +} + +func (reader *resumableMediaReader) Read(buffer []byte) (int, error) { + if len(buffer) == 0 { + return 0, nil + } + for { + if err := reader.request.Context().Err(); err != nil { + return 0, err + } + if reader.failure != nil { + return 0, reader.failure + } + if reader.length >= 0 && reader.read == reader.length { + if reader.attempts > 0 && !reader.recovered && reader.proxy.diagnostic != nil { + reader.recovered = true + reader.proxy.diagnostic(diagnosticEvent{Level: "info", Event: "media.recovered", Host: reader.request.URL.Hostname(), + OffsetBytes: reader.span.start + reader.read, TotalBytes: reader.span.total, Attempt: reader.attempts, Message: "媒体续读完成"}) + } + return 0, io.EOF + } + if reader.pending != nil { + err := reader.pending + if errors.Is(err, io.EOF) { + if reader.length < 0 { + return 0, io.EOF + } + err = io.ErrUnexpectedEOF + } + if reader.resumable && reader.attempts < reader.maxAttempts { + if err = reader.resume(); err == nil { + reader.pending = nil + continue + } + } + if canceled := reader.request.Context().Err(); canceled != nil { + return 0, canceled + } + reader.failure = fmt.Errorf("媒体资源读取不完整(已读取 %d 字节,续读 %d 次): %w", reader.read, reader.attempts, err) + return 0, reader.failure + } + if reader.length >= 0 && int64(len(buffer)) > reader.length-reader.read { + buffer = buffer[:reader.length-reader.read] + } + count, err := reader.body.Read(buffer) + reader.read += int64(count) + reader.pending = err + if count > 0 || err == nil { + return count, nil + } + } +} + +func (reader *resumableMediaReader) resume() error { + _ = reader.body.Close() + reader.body = http.NoBody + var lastErr error + for reader.attempts < reader.maxAttempts { + ctx := reader.request.Context() + if reader.attempts > 0 { + timer := time.NewTimer(time.Duration(reader.attempts) * 250 * time.Millisecond) + select { + case <-timer.C: + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + } + } + if err := ctx.Err(); err != nil { + return err + } + request := reader.request.Clone(ctx) + request.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", reader.span.start+reader.read, reader.span.end)) + request.Header.Del("If-Range") + if reader.validator != "" { + request.Header.Set("If-Range", reader.validator) + } + reader.attempts++ + if reader.proxy.diagnostic != nil { + reader.proxy.diagnostic(diagnosticEvent{Level: "warn", Event: "media.resume", Host: reader.request.URL.Hostname(), + OffsetBytes: reader.span.start + reader.read, TotalBytes: reader.span.total, Attempt: reader.attempts, Message: "媒体连接中断,正在续读"}) + } + response, err := reader.proxy.client.Do(request) + if err != nil { + lastErr = err + continue + } + if err = reader.validateContinuation(response); err == nil { + reader.body = response.Body + return nil + } + _ = response.Body.Close() + lastErr = err + if response.StatusCode != http.StatusRequestTimeout && response.StatusCode != http.StatusTooManyRequests && response.StatusCode < 500 { + return err + } + } + return lastErr +} + +func (reader *resumableMediaReader) validateContinuation(response *http.Response) error { + if response.StatusCode != http.StatusPartialContent { + return fmt.Errorf("媒体续读未返回所需字节范围(HTTP %d)", response.StatusCode) + } + span, valid := parseMediaByteRange(response.Header.Get("Content-Range")) + if !valid || span.start != reader.span.start+reader.read || span.end != reader.span.end || span.total != reader.span.total || + response.ContentLength >= 0 && response.ContentLength != reader.length-reader.read { + return errors.New("媒体续读字节范围或长度不一致") + } + if !identityMediaResponse(response) || reader.etag != "" && strings.TrimSpace(response.Header.Get("ETag")) != reader.etag || + reader.modified != "" && strings.TrimSpace(response.Header.Get("Last-Modified")) != reader.modified { + return errors.New("媒体续读资源已变化") + } + return nil +} diff --git a/internal/app/media_resume_integration_test.go b/internal/app/media_resume_integration_test.go new file mode 100644 index 0000000..48c7ae9 --- /dev/null +++ b/internal/app/media_resume_integration_test.go @@ -0,0 +1,168 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" +) + +type mediaDisconnectWriter struct { + http.ResponseWriter + remaining int +} + +func (writer *mediaDisconnectWriter) Write(data []byte) (int, error) { + if len(data) > writer.remaining { + count, _ := writer.ResponseWriter.Write(data[:writer.remaining]) + writer.remaining = 0 + return count, io.ErrUnexpectedEOF + } + count, err := writer.ResponseWriter.Write(data) + writer.remaining -= count + return count, err +} + +func verifyResumedTimeline(t *testing.T, ctx context.Context, ffmpeg, input string) { + t.Helper() + command := exec.CommandContext(ctx, ffmpeg, "-hide_banner", "-loglevel", "warning", "-xerror", "-nostdin", "-i", input, + "-map", "0:v:0", "-map", "0:a:0", "-progress", "pipe:1", "-f", "null", "-") + var progress, diagnostic bytes.Buffer + command.Stdout, command.Stderr = &progress, &diagnostic + if err := command.Run(); err != nil { + t.Fatalf("decode resumed timeline: %v\n%s", err, diagnostic.String()) + } + values := make(map[string]string) + for _, line := range strings.Split(progress.String(), "\n") { + key, value, ok := strings.Cut(line, "=") + if ok { + values[key] = strings.TrimSpace(value) + } + } + frames, _ := strconv.Atoi(values["frame"]) + micros, _ := strconv.Atoi(values["out_time_us"]) + if values["progress"] != "end" || frames != 1800 || micros < 59_900_000 || micros > 60_200_000 { + t.Fatalf("interrupted media lost frames, audio or duration: %v\n%s", values, diagnostic.String()) + } +} + +func TestPlaybackAndDownloadResumeInterruptedMedia(t *testing.T) { + fixture, fixtureSession := nativePlaybackFixture(t, "60", "160x90") + video, err := os.ReadFile(fixtureSession.tasks[0].OutPath) + if err != nil { + t.Fatal(err) + } + ffmpeg := fixture.downloader.cfg.FFmpeg + for _, mode := range []string{"MSE", "native HLS", "prefetch", "download", "failed download"} { + t.Run(mode, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + var interrupted atomic.Bool + var resumes atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/fixture.mp4" { + t.Error("fixture requested an unrelated resource") + http.NotFound(writer, request) + return + } + writer.Header().Set("ETag", `"generated-media-v1"`) + if request.Header.Get("If-Range") == `"generated-media-v1"` { + resumes.Add(1) + } + if request.Method == http.MethodGet && (interrupted.CompareAndSwap(false, true) || mode == "failed download") { + writer = &mediaDisconnectWriter{ResponseWriter: writer, remaining: min(64*1024, len(video)/2)} + } + http.ServeContent(writer, request, "fixture.mp4", time.Time{}, bytes.NewReader(video)) + })) + t.Cleanup(server.Close) + app, session := prefetchFixtureApp(t) + d := app.downloader + d.cfg.FFmpeg, d.cfg.Retries, d.cfg.SkipBytes = ffmpeg, 3, 1 + d.client = server.Client() + app.cfg = d.cfg + for index := range session.tasks { + session.tasks[index] = Task{DramaID: "hongguo:7000000000000000001", DramaTitle: "合成断流样本", Index: index + 1, Total: 3, + OutPath: filepath.Join(t.TempDir(), "episode.mp4"), + Chapter: Chapter{ID: fmt.Sprintf("hongguo:7000000000000000001:%d", index+1), Source: sourceHongguo, VideoURL: server.URL + "/fixture.mp4", CurrentEpisode: rawEpisode(index + 1)}} + } + var input string + switch mode { + case "native HLS": + opened := nativeOpenFixture(t, app, 1, 0, 1) + var address string + if err := json.Unmarshal(opened["url"], &address); err != nil { + t.Fatal(err) + } + mux := http.NewServeMux() + app.registerPlaybackRoutes(mux) + local := httptest.NewServer(viewerFixtureHandler(app, mux)) + t.Cleanup(local.Close) + input = local.URL + address + case "download", "failed download": + var completed atomic.Bool + err := d.DownloadEpisodeWithProgress(ctx, session.tasks[0], func(progress DownloadProgress) { + if progress.Phase == "completed" && progress.Percent == 100 { + completed.Store(true) + } + }) + if mode == "failed download" { + _, fileErr := os.Stat(session.tasks[0].OutPath) + if err == nil || completed.Load() || !os.IsNotExist(fileErr) { + t.Fatal("incomplete media was marked as a completed download", err, fileErr) + } + return + } + if err != nil || !completed.Load() { + t.Fatal("interrupted download did not complete", err) + } + input = session.tasks[0].OutPath + default: + episode := 1 + if mode == "prefetch" { + episode = 2 + result := prefetchRequest(app, `{"session":"fixture","episode":2,"run":4,"version":1}`) + if result.Code != http.StatusAccepted { + t.Fatal("prefetch was not accepted", result.Body.String()) + } + select { + case <-session.prefetch.done: + case <-ctx.Done(): + t.Fatal("prefetch did not complete") + } + if session.prefetch.view().State != "ready" { + t.Fatal("recovered media was marked as a failed prefetch", session.prefetch.view()) + } + } + writer := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost/api/ui/playback/stream?session=fixture&episode=%d", episode), nil).WithContext(ctx) + app.handlePlaybackStream(writer, viewerFixtureRequest(app, request)) + state, _ := app.playbackStatus("fixture", false) + if writer.Code != http.StatusOK || state.State != "ended" || state.Error != "" || state.Duration != 60 { + t.Fatalf("recovered playback failed: HTTP=%d state=%+v", writer.Code, state) + } + if mode == "prefetch" && writer.Header().Get("X-Playback-Prefetched") != "1" { + t.Fatal("recovered prefetch was discarded") + } + input = filepath.Join(t.TempDir(), "resumed.mp4") + if err := os.WriteFile(input, writer.Body.Bytes(), 0600); err != nil { + t.Fatal(err) + } + } + verifyResumedTimeline(t, ctx, ffmpeg, input) + if !interrupted.Load() || resumes.Load() == 0 { + t.Fatal("integration did not exercise automatic byte continuation") + } + }) + } +} diff --git a/internal/app/media_resume_test.go b/internal/app/media_resume_test.go new file mode 100644 index 0000000..d61b3e8 --- /dev/null +++ b/internal/app/media_resume_test.go @@ -0,0 +1,343 @@ +package app + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" +) + +func mediaProxyFixture(t *testing.T, handler http.HandlerFunc) (*hlsProxy, *http.Client) { + t.Helper() + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + downloader := &Downloader{cfg: Config{Retries: 3}, client: server.Client()} + proxy, err := downloader.newHLSProxy(context.Background(), providerMedia{URL: server.URL + "/fixture.mp4", Referer: "https://example.invalid/"}, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(proxy.Close) + client := &http.Client{Transport: &http.Transport{Proxy: nil}, Timeout: 10 * time.Second} + t.Cleanup(client.CloseIdleConnections) + return proxy, client +} + +func TestMediaProxyResumesTruncatedResponse(t *testing.T) { + payload := bytes.Repeat([]byte("0123456789abcdef"), 64) + for _, test := range []struct { + name string + rangeHeader string + start int + end int + }{ + {name: "whole", start: 0, end: 1023}, + {name: "bounded seek", rangeHeader: "bytes=100-699", start: 100, end: 699}, + {name: "suffix seek", rangeHeader: "bytes=-129", start: 895, end: 1023}, + } { + t.Run(test.name, func(t *testing.T) { + var calls atomic.Int32 + proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { + attempt := calls.Add(1) + if request.Header.Get("Accept-Encoding") != "identity" || request.Header.Get("Referer") != "https://example.invalid/" { + t.Error("media request lost source headers") + } + start, end, status := test.start, test.end, http.StatusOK + if test.rangeHeader != "" { + status = http.StatusPartialContent + } + if attempt == 1 { + if request.Header.Get("Range") != test.rangeHeader { + t.Error("initial seek range was changed") + } + } else { + start += 13 + status = http.StatusPartialContent + want := fmt.Sprintf("bytes=%d-%d", start, end) + if request.Header.Get("Range") != want || request.Header.Get("If-Range") != `"fixture-v1"` { + t.Errorf("unsafe continuation: Range=%q If-Range=%q, want %s", request.Header.Get("Range"), request.Header.Get("If-Range"), want) + } + } + writer.Header().Set("ETag", `"fixture-v1"`) + writer.Header().Set("Content-Type", "video/mp4") + writer.Header().Set("Content-Length", strconv.Itoa(end-start+1)) + if status == http.StatusPartialContent { + writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(payload))) + } + writer.WriteHeader(status) + if attempt == 1 { + _, _ = writer.Write(payload[start : start+13]) + } else { + _, _ = writer.Write(payload[start : end+1]) + } + }) + request, _ := http.NewRequest(http.MethodGet, proxy.root, nil) + request.Header.Set("Range", test.rangeHeader) + response, err := client.Do(request) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil || !bytes.Equal(body, payload[test.start:test.end+1]) || calls.Load() != 2 || proxy.Err() != nil { + t.Fatalf("interrupted media did not resume exactly: bytes=%d calls=%d read=%v proxy=%v", len(body), calls.Load(), err, proxy.Err()) + } + wantStatus := http.StatusOK + if test.rangeHeader != "" { + wantStatus = http.StatusPartialContent + } + if response.StatusCode != wantStatus || response.ContentLength != int64(test.end-test.start+1) { + t.Fatal("continuation changed the local response headers") + } + }) + } +} + +func TestMediaProxyExhaustsBodyRetries(t *testing.T) { + payload := bytes.Repeat([]byte("0123456789"), 10) + var calls atomic.Int32 + proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { + attempt := int(calls.Add(1)) + start := (attempt - 1) * 13 + writer.Header().Set("Content-Length", strconv.Itoa(len(payload)-start)) + writer.Header().Set("ETag", `"fixture-v1"`) + if attempt > 1 { + if request.Header.Get("Range") != fmt.Sprintf("bytes=%d-99", start) { + t.Error("retry did not preserve its cumulative offset") + } + writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-99/100", start)) + writer.WriteHeader(http.StatusPartialContent) + } + _, _ = writer.Write(payload[start : start+13]) + }) + response, err := client.Get(proxy.root) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err == nil || calls.Load() != 3 || !bytes.Equal(body, payload[:39]) || proxy.Err() == nil { + t.Fatalf("exhausted transfer was not bounded or reported: bytes=%d calls=%d read=%v proxy=%v", len(body), calls.Load(), err, proxy.Err()) + } +} + +func TestMediaProxyRejectsUnsafeContinuation(t *testing.T) { + for _, test := range []struct { + name, contentRange, etag, encoding string + status, length int + }{ + {name: "range ignored", status: 200, length: 100, etag: `"fixture-v1"`}, + {name: "wrong start", status: 206, length: 90, contentRange: "bytes 9-98/100", etag: `"fixture-v1"`}, + {name: "wrong end", status: 206, length: 89, contentRange: "bytes 10-98/100", etag: `"fixture-v1"`}, + {name: "changed length", status: 206, length: 90, contentRange: "bytes 10-99/101", etag: `"fixture-v1"`}, + {name: "changed object", status: 206, length: 90, contentRange: "bytes 10-99/100", etag: `"fixture-v2"`}, + {name: "missing validator", status: 206, length: 90, contentRange: "bytes 10-99/100"}, + {name: "encoded bytes", status: 206, length: 90, contentRange: "bytes 10-99/100", etag: `"fixture-v1"`, encoding: "gzip"}, + {name: "conflicting length", status: 206, length: 80, contentRange: "bytes 10-99/100", etag: `"fixture-v1"`}, + } { + t.Run(test.name, func(t *testing.T) { + var calls atomic.Int32 + proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { + if calls.Add(1) == 1 { + writer.Header().Set("Content-Length", "100") + writer.Header().Set("ETag", `"fixture-v1"`) + _, _ = io.WriteString(writer, "0123456789") + return + } + writer.Header().Set("Content-Length", strconv.Itoa(test.length)) + writer.Header().Set("Content-Range", test.contentRange) + writer.Header().Set("ETag", test.etag) + writer.Header().Set("Content-Encoding", test.encoding) + writer.WriteHeader(test.status) + _, _ = io.WriteString(writer, strings.Repeat("x", test.length)) + }) + response, err := client.Get(proxy.root) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err == nil || string(body) != "0123456789" || calls.Load() != 2 || proxy.Err() == nil { + t.Fatalf("unverified bytes were appended or failure was lost: bytes=%d calls=%d read=%v proxy=%v", len(body), calls.Load(), err, proxy.Err()) + } + }) + } +} + +func TestMediaProxyCanceledRequestDoesNotRecordFailure(t *testing.T) { + proxy, _ := mediaProxyFixture(t, func(http.ResponseWriter, *http.Request) { + t.Error("canceled request reached upstream") + }) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + request := httptest.NewRequest(http.MethodGet, proxy.root, nil).WithContext(ctx) + proxy.ServeHTTP(httptest.NewRecorder(), request) + if proxy.Err() != nil { + t.Fatal("stopping playback poisoned its media proxy", proxy.Err()) + } +} + +func TestMediaProxyCancelsDuringContinuation(t *testing.T) { + started := make(chan struct{}) + var calls atomic.Int32 + proxy, _ := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { + if calls.Add(1) == 1 { + writer.Header().Set("Content-Length", "100") + _, _ = io.WriteString(writer, "0123456789") + return + } + close(started) + <-request.Context().Done() + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + proxy.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, proxy.root, nil).WithContext(ctx)) + }() + select { + case <-started: + case <-time.After(3 * time.Second): + t.Fatal("continuation never started") + } + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("continuation blocked playback cancellation") + } + if proxy.Err() != nil || calls.Load() != 2 { + t.Fatal("cancellation was recorded as an upstream failure", proxy.Err()) + } +} + +func TestMediaProxyRetriesTemporaryContinuationStatus(t *testing.T) { + const modified = "Mon, 14 Sep 2026 12:00:00 GMT" + var calls atomic.Int32 + proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { + attempt := calls.Add(1) + if attempt == 2 { + writer.WriteHeader(http.StatusServiceUnavailable) + return + } + writer.Header().Set("Last-Modified", modified) + if attempt == 1 { + writer.Header().Set("Content-Length", "20") + _, _ = io.WriteString(writer, "0123456789") + return + } + if request.Header.Get("Range") != "bytes=10-19" || request.Header.Get("If-Range") != modified { + t.Error("date validator or retry offset was lost") + } + writer.Header().Set("Content-Length", "10") + writer.Header().Set("Content-Range", "bytes 10-19/20") + writer.WriteHeader(http.StatusPartialContent) + _, _ = io.WriteString(writer, "abcdefghij") + }) + response, err := client.Get(proxy.root) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil || string(body) != "0123456789abcdefghij" || calls.Load() != 3 || proxy.Err() != nil { + t.Fatal("temporary continuation failure was latched", err, proxy.Err()) + } +} + +func TestMediaProxyHeadAndUnknownLength(t *testing.T) { + for _, mode := range []string{"head", "chunked", "broken chunked"} { + t.Run(mode, func(t *testing.T) { + var calls atomic.Int32 + proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { + calls.Add(1) + if mode == "head" { + writer.Header().Set("Content-Length", "100") + return + } + writer.WriteHeader(http.StatusOK) + writer.(http.Flusher).Flush() + _, _ = io.WriteString(writer, "unknown-length-fixture") + writer.(http.Flusher).Flush() + if mode == "broken chunked" { + connection, _, err := writer.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + _ = connection.Close() + } + }) + method := http.MethodGet + if mode == "head" { + method = http.MethodHead + } + request, _ := http.NewRequest(method, proxy.root, nil) + response, err := client.Do(request) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, _ := io.ReadAll(response.Body) + if mode == "head" && len(body) != 0 || mode != "head" && string(body) != "unknown-length-fixture" || calls.Load() != 1 { + t.Fatal("HEAD or unknown-length response changed") + } + if (proxy.Err() != nil) != (mode == "broken chunked") { + t.Fatal("unverifiable truncation was hidden or a complete body was rejected", proxy.Err()) + } + }) + } +} + +type mediaFailureBody struct { + data []byte +} + +func (body *mediaFailureBody) Read(buffer []byte) (int, error) { + count := copy(buffer, body.data) + body.data = body.data[count:] + return count, io.ErrUnexpectedEOF +} + +func (*mediaFailureBody) Close() error { return nil } + +type mediaStoppedWriter struct{ header http.Header } + +func (writer *mediaStoppedWriter) Header() http.Header { return writer.header } +func (*mediaStoppedWriter) WriteHeader(int) {} +func (*mediaStoppedWriter) Write([]byte) (int, error) { return 0, errors.New("consumer stopped") } + +func TestMediaProxyConsumerStopDoesNotLatchPendingReadError(t *testing.T) { + proxy, _ := mediaProxyFixture(t, func(http.ResponseWriter, *http.Request) { + t.Error("fixture transport was bypassed") + }) + var calls int + proxy.client.Transport = rankingTransport(func(request *http.Request) (*http.Response, error) { + calls++ + return &http.Response{StatusCode: 200, Header: http.Header{"Content-Length": {"100"}}, ContentLength: 100, + Body: &mediaFailureBody{data: []byte("0123456789")}, Request: request}, nil + }) + proxy.ServeHTTP(&mediaStoppedWriter{header: make(http.Header)}, httptest.NewRequest(http.MethodGet, proxy.root, nil)) + if calls != 1 || proxy.Err() != nil { + t.Fatal("unused media bytes triggered a retry or false failure", proxy.Err()) + } +} + +func TestMediaReaderAcceptsExactlyCompletedBody(t *testing.T) { + request := httptest.NewRequest(http.MethodGet, "http://fixture.invalid/media", nil) + proxy := &hlsProxy{retries: 3} + reader := proxy.mediaReader(request, &http.Response{StatusCode: 200, ContentLength: 10, Header: make(http.Header), Body: &mediaFailureBody{data: []byte("0123456789")}}) + defer reader.Close() + body, err := io.ReadAll(reader) + if err != nil || string(body) != "0123456789" || reader.attempts != 0 { + t.Fatal("completed response was retried", err) + } +} diff --git a/internal/app/media_transport_parity_test.go b/internal/app/media_transport_parity_test.go new file mode 100644 index 0000000..a18b54b --- /dev/null +++ b/internal/app/media_transport_parity_test.go @@ -0,0 +1,153 @@ +package app + +import ( + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" +) + +func TestParityMediaCredentialsStayOnExactOrigin(t *testing.T) { + var requests int + d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { + requests++ + if request.Header.Get("Sec-Fetch-Mode") != "cors" || request.Header.Get("Sec-Fetch-Dest") != "empty" || request.Header.Get("Referer") != "https://site.example.test/episode/1" { + t.Error("media request lost required headers") + } + response := rankingHTTPResponse(request, 200, "fixture") + if request.URL.Host == "media.example.test" { + if request.Header.Get("Cookie") != "Signed=fixture" { + t.Error("credential missing on its origin") + } + response.StatusCode = 302 + response.Header.Set("Location", "https://other.example.test/segment") + } else if request.Header.Get("Cookie") != "" { + t.Error("signed credential leaked on redirect") + } + return response, nil + }) + credentials := &providerMediaCredentials{origin: "https://media.example.test", cookie: "Signed=fixture", referer: "https://site.example.test/episode/1", expires: time.Now().Add(time.Minute)} + request, _ := http.NewRequestWithContext(providerMediaContext(context.Background(), credentials), http.MethodGet, "https://media.example.test/segment", nil) + response, err := d.doMediaRequest(request) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if requests != 2 || request.Header.Get("Cookie") != "" { + t.Fatal("redirect not followed or caller request mutated") + } + credentials.expires = time.Now().Add(-time.Second) + if _, err := d.doMediaRequest(request); err == nil || requests != 2 { + t.Fatal("expired credentials reached the network") + } + for _, address := range []string{"http://media.example.test/segment", "https://media.example.test:444/segment", "https://child.media.example.test/segment"} { + other, _ := http.NewRequest(http.MethodGet, address, nil) + other.Header.Set("Cookie", "Signed=fixture") + if err := credentials.apply(other); err != nil || other.Header.Get("Cookie") != "" { + t.Fatal("origin scope includes another scheme, port or subdomain") + } + } +} + +func TestParityMediaProxiesKeepCredentialsForListsKeysAndSegments(t *testing.T) { + for _, gateway := range []bool{false, true} { + t.Run(fmt.Sprint(gateway), func(t *testing.T) { + seen := map[string]int{} + d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { + seen[request.URL.Path]++ + if request.Header.Get("Cookie") != "Signed=fixture" || request.Header.Get("Sec-Fetch-Mode") != "cors" { + t.Error("proxy dropped credentials or media headers") + } + switch request.URL.Path { + case "/start.m3u8": + response := rankingHTTPResponse(request, 302, "") + response.Header.Set("Location", "/final/index.m3u8?sig=a%2Fb") + return response, nil + case "/final/index.m3u8": + return rankingHTTPResponse(request, 200, "#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"key\"\n#EXTINF:4,\nsegment.ts\n#EXT-X-ENDLIST\n"), nil + case "/final/key": + return rankingHTTPResponse(request, 200, "0123456789abcdef"), nil + case "/final/segment.ts": + return rankingHTTPResponse(request, 200, "synthetic-segment"), nil + } + return nil, fmt.Errorf("unexpected request: %s", request.URL.Path) + }) + credentials := &providerMediaCredentials{origin: "https://media.example.test", cookie: "Signed=fixture", referer: "https://site.example.test/episode/1"} + ctx := providerMediaContext(context.Background(), credentials) + playlist, final, err := d.fetchMediaPlaylist(ctx, "https://media.example.test/start.m3u8", credentials.referer) + if err != nil || final != "https://media.example.test/final/index.m3u8?sig=a%2Fb" { + t.Fatal("redirect lost its final playlist base", final, err) + } + media := providerMedia{URL: final, Playlist: playlist, Referer: credentials.referer, credentials: credentials} + proxy, err := d.newHLSProxy(context.Background(), media, nil) + if err != nil { + t.Fatal(err) + } + defer proxy.Close() + root := proxy.root + if gateway { + proxy.Close() + session := &playbackMediaSession{media: media, plan: playbackMediaPlan{Player: "hls"}} + app := &UIApp{downloader: d} + app.attachMediaGateway(session, "/media/", "") + proxy = session.proxy + root = session.plan.URL + } + read := func(path string) string { + t.Helper() + writer := httptest.NewRecorder() + proxy.ServeHTTP(writer, httptest.NewRequest(http.MethodGet, path, nil)) + if writer.Code != 200 { + t.Fatalf("proxy response %d: %s", writer.Code, writer.Body.String()) + } + return writer.Body.String() + } + body := read(root) + key := hlsURIAttribute.FindStringSubmatch(body) + if len(key) != 2 || read(key[1]) != "0123456789abcdef" { + t.Fatal("rewritten key not readable") + } + segment := "" + for _, line := range strings.Split(body, "\n") { + if line != "" && !strings.HasPrefix(line, "#") { + segment = line + break + } + } + if segment == "" || read(segment) != "synthetic-segment" || seen["/final/key"] != 1 || seen["/final/segment.ts"] != 1 { + t.Fatal("rewritten segment not readable") + } + }) + } +} + +func TestParityPlaybackQualityInheritsMediaCredentials(t *testing.T) { + d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { + if request.Header.Get("Cookie") != "Signed=fixture" { + t.Error("variant lost credentials") + } + return rankingHTTPResponse(request, 200, "#EXTM3U\n#EXTINF:4,\nsegment.ts\n#EXT-X-ENDLIST\n"), nil + }) + credentials := &providerMediaCredentials{origin: "https://media.example.test", cookie: "Signed=fixture"} + media := providerMedia{URL: "https://media.example.test/high.mp4", Referer: "https://site.example.test/", Quality: 1080, credentials: credentials, + Variants: []providerMedia{{URL: "https://media.example.test/low.m3u8", Quality: 720}}} + selected, err := d.selectPlaybackQuality(context.WithValue(context.Background(), playbackQualityKey{}, 720), media) + if err != nil || selected.credentials != credentials || selected.Referer != media.Referer || selected.Playlist == "" { + t.Fatal("selected variant lost session or playlist", err) + } + request, _ := http.NewRequest(http.MethodGet, "https://media.example.test:443/part", nil) + if providerMediaOrigin(request.URL) != credentials.origin { + t.Fatal("default port changed origin") + } + _, _, err = d.fetchMediaPlaylist(context.Background(), "file:///tmp/fixture", "") + if err == nil { + t.Fatal("non-HTTP playlist accepted") + } + _, _ = io.Discard.Write(nil) + _, _ = url.Parse("https://example.test") +}