djk/internal/app/playback_generated.go

package app

import (
	"context"
	"errors"
	"fmt"
	"math"
	"net/http"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const playbackProgressiveSegments = 4

type playbackSegmentIndex struct {
	Name     string
	Start    float64
	Duration float64
}

type playbackGenerated struct {
	ctx       context.Context
	cancel    context.CancelFunc
	directory string
	base      string
	query     string
	playlist  string
	segments  []playbackSegmentIndex
	lastEnd   float64
	files     map[string]bool
	bytes     int64
	resources *playbackResources
	done      chan struct{}
	doneOnce  sync.Once
	closeOnce sync.Once
	complete  atomic.Bool
	failed    atomic.Bool
	mu        sync.RWMutex
}

func (generated *playbackGenerated) finish() {
	generated.doneOnce.Do(func() { close(generated.done) })
}

func playbackGeneratedArgs(media providerMedia, input, directory, processing string) []string {
	args := playbackInputArgs(media, input, 0)
	if processing == "audio" {
		args = append(args, "-c:v", "copy", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2")
	} else {
		args = append(args, "-c", "copy")
	}
	return append(args, "-avoid_negative_ts", "make_zero", "-f", "hls", "-hls_time", "2", "-hls_list_size", "0",
		"-hls_playlist_type", "event", "-hls_segment_type", "fmp4", "-hls_fmp4_init_filename", "init.mp4",
		"-hls_flags", "temp_file", "-hls_segment_filename", filepath.ToSlash(filepath.Join(directory, "%06d.m4s")), filepath.ToSlash(filepath.Join(directory, "index.m3u8")))
}

func (app *UIApp) generatePlayback(ctx context.Context, media *playbackMediaSession, processing, base, query string, startSeconds float64) error {
	ffmpeg, err := app.downloader.ensureFFmpeg(ctx)
	if err != nil {
		return err
	}
	release, err := app.mediaResources().acquire(ctx, processing, false)
	if err != nil {
		return err
	}
	directory, err := os.MkdirTemp("", "juku-playback-media-")
	if err != nil {
		release()
		return err
	}
	jobCtx, cancel := context.WithCancel(media.ctx)
	generated := &playbackGenerated{ctx: jobCtx, cancel: cancel, directory: directory, base: base, query: query,
		resources: app.mediaResources(), done: make(chan struct{}), files: make(map[string]bool)}
	media.generated = generated
	input := media.local
	var proxy *hlsProxy
	if input == "" {
		proxy, err = app.downloader.newHLSProxy(jobCtx, media.media, media.key)
		if err != nil {
			generated.finish()
			cancel()
			release()
			return err
		}
		proxy.acquire = func(ctx context.Context) (func(), error) { return app.mediaResources().acquire(ctx, "media", false) }
		input = proxy.root
	}
	command := ffmpegMediaCommand(jobCtx, ffmpeg, playbackGeneratedArgs(media.media, input, directory, processing)...)
	log := &playbackLog{text: cappedStringWriter{limit: 64 * 1024}, ready: make(chan struct{})}
	command.Stderr = log
	if err := command.Start(); err != nil {
		generated.finish()
		cancel()
		release()
		return fmt.Errorf("无法启动媒体封装：%w", err)
	}
	done := make(chan error, 1)
	go func() { done <- command.Wait() }()
	go func() { // 后台：持续刷新渐进式播放列表，等 ffmpeg 收尾后补 ENDLIST 并释放转封装资源
		defer generated.finish()
		defer release()
		ticker := time.NewTicker(300 * time.Millisecond)
		defer ticker.Stop()
		for {
			select {
			case waitErr := <-done:
				if waitErr != nil {
					if jobCtx.Err() == nil {
						generated.failed.Store(true)
						app.downloader.recordDiagnostic(diagnosticEvent{Level: "warn", Event: "playback.generated.failed",
							Message: "媒体封装中断：" + fmt.Sprint(waitErr)})
					}
					return
				}
				_ = generated.accountCache()
				if data, err := os.ReadFile(filepath.Join(generated.directory, "index.m3u8")); err == nil {
					_ = generated.index(string(data))
				}
				return
			case <-jobCtx.Done():
				<-done
				return
			case <-ticker.C:
				if err := generated.accountCache(); err != nil {
					generated.failed.Store(true)
					cancel()
					<-done
					return
				}
				if data, err := os.ReadFile(filepath.Join(generated.directory, "index.m3u8")); err == nil {
					_ = generated.index(string(data))
				}
			}
		}
	}()
	// 前台：前几个分片就绪（并覆盖起播位置）即返回计划，剩余部分边播边转
	ticker := time.NewTicker(80 * time.Millisecond)
	defer ticker.Stop()
	for {
		select {
		case <-generated.done:
			if ctx.Err() != nil {
				return ctx.Err()
			}
			generated.mu.RLock()
			complete, failed := generated.complete.Load(), generated.failed.Load()
			segmentCount := len(generated.segments)
			lastEnd := generated.lastEnd
			generated.mu.RUnlock()
			if failed || !complete || segmentCount == 0 {
				return errors.New("媒体封装未能完成")
			}
			generated.applyPlan(media, processing, lastEnd)
			return nil
		case <-ctx.Done():
			cancel()
			<-generated.done
			return ctx.Err()
		case <-ticker.C:
			if err := generated.accountCache(); err != nil {
				cancel()
				<-generated.done
				return err
			}
			data, err := os.ReadFile(filepath.Join(generated.directory, "index.m3u8"))
			if err != nil {
				continue
			}
			body := string(data)
			if strings.Count(body, "#EXTINF") < playbackProgressiveSegments {
				continue
			}
			if err := generated.index(body); err != nil {
				continue
			}
			generated.mu.RLock()
			lastEnd := generated.lastEnd
			generated.mu.RUnlock()
			if lastEnd < startSeconds+6 {
				continue
			}
			generated.applyPlan(media, processing, lastEnd)
			return nil
		}
	}
}

func (generated *playbackGenerated) applyPlan(media *playbackMediaSession, processing string, lastEnd float64) {
	media.plan.URL = generated.assetURL("index.m3u8")
	media.plan.Player, media.plan.MIME, media.plan.Processing = "hls", "application/vnd.apple.mpegurl", processing
	if media.plan.Duration <= 0 {
		media.plan.Duration, media.plan.SeekEnd = lastEnd, lastEnd
	}
}

func (generated *playbackGenerated) accountCache() error {
	entries, err := os.ReadDir(generated.directory)
	if err != nil {
		return err
	}
	var size int64
	for _, entry := range entries {
		info, err := entry.Info()
		if err != nil {
			if os.IsNotExist(err) {
				continue
			}
			return err
		}
		if !info.Mode().IsRegular() || info.Size() > 64<<20 {
			return errors.New("媒体分片超出缓存限制")
		}
		size += info.Size()
	}
	resources := generated.resources
	resources.mu.Lock()
	defer resources.mu.Unlock()
	if resources.cacheBytes+size-generated.bytes > int64(resources.settings.CacheMB)<<20 {
		return errors.New("媒体临时缓存已达上限，请关闭其他播放后重试")
	}
	resources.cacheBytes += size - generated.bytes
	generated.bytes = size
	return nil
}

func (generated *playbackGenerated) assetURL(name string) string {
	address := generated.base + name
	if generated.query != "" {
		address += "?" + generated.query
	}
	return address
}

func (generated *playbackGenerated) index(playlist string) error {
	complete := strings.Contains(playlist, "#EXT-X-ENDLIST")
	if len(playlist) > 2<<20 || !strings.HasPrefix(playlist, "#EXTM3U") {
		return errors.New("媒体封装尚未就绪")
	}
	lines := strings.Split(playlist, "\n")
	segments := make([]playbackSegmentIndex, 0, 64)
	files := make(map[string]bool, 64)
	var start, duration float64
	validName := func(name string) bool {
		return name != "" && filepath.Base(name) == name && !strings.ContainsAny(name, "\\?\x00") && (strings.HasSuffix(name, ".mp4") || strings.HasSuffix(name, ".m4s"))
	}
	for i, line := range lines {
		if strings.HasPrefix(line, "#EXTINF:") {
			value, _, _ := strings.Cut(strings.TrimPrefix(line, "#EXTINF:"), ",")
			parsed, err := strconv.ParseFloat(value, 64)
			if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) || parsed <= 0 || parsed > 3600 {
				return errors.New("媒体分片时长无效")
			}
			duration = parsed
		} else if strings.HasPrefix(line, "#EXT-X-MAP:") {
			matches := hlsURIAttribute.FindStringSubmatch(line)
			if len(matches) != 2 || !validName(matches[1]) {
				return errors.New("媒体初始化片段无效")
			}
			files[matches[1]] = true
			lines[i] = strings.Replace(line, matches[0], `URI="`+generated.assetURL(matches[1])+`"`, 1)
		} else if line != "" && !strings.HasPrefix(line, "#") {
			if duration <= 0 || !validName(line) || len(segments) >= 16384 || start+duration > 24*60*60 {
				return errors.New("媒体分片索引无效")
			}
			files[line] = true
			segments = append(segments, playbackSegmentIndex{Name: line, Start: start, Duration: duration})
			start += duration
			duration = 0
			lines[i] = generated.assetURL(line)
		}
	}
	if !complete && duration > 0 {
		duration = 0
	}
	if len(segments) == 0 || (complete && duration != 0) {
		return errors.New("媒体播放索引没有完整分片")
	}
	if !complete && len(segments) < len(generated.segments) {
		return nil
	}
	for i, line := range lines {
		if strings.HasPrefix(line, "#EXT-X-TARGETDURATION:") {
			var longest float64
			for _, segment := range segments {
				longest = math.Max(longest, segment.Duration)
			}
			lines[i] = "#EXT-X-TARGETDURATION:" + strconv.Itoa(int(math.Ceil(longest)))
			break
		}
	}
	for name := range files {
		info, err := os.Stat(filepath.Join(generated.directory, name))
		if err != nil {
			return fmt.Errorf("媒体分片未完整生成（%s）：%w", name, err)
		}
		if !info.Mode().IsRegular() {
			return fmt.Errorf("媒体分片未完整生成（%s 不是普通文件）", name)
		}
		if info.Size() <= 0 {
			return fmt.Errorf("媒体分片未完整生成（%s 为空文件）", name)
		}
	}
	generated.mu.Lock()
	generated.segments = segments
	generated.files = files
	generated.lastEnd = start
	generated.playlist = strings.Join(lines, "\n")
	generated.mu.Unlock()
	generated.complete.Store(complete)
	return nil
}

func (generated *playbackGenerated) Close() {
	generated.closeOnce.Do(func() {
		generated.cancel()
		go func() {
			<-generated.done
			removeMediaDirectory(generated.directory)
			generated.resources.mu.Lock()
			defer generated.resources.mu.Unlock()
			generated.resources.cacheBytes -= generated.bytes
		}()
	})
}

func removeMediaDirectory(directory string) {
	for attempt := 0; attempt < 30; attempt++ {
		err := os.RemoveAll(directory)
		if err == nil {
			if _, statErr := os.Stat(directory); os.IsNotExist(statErr) {
				return
			}
		}
		time.Sleep(200 * time.Millisecond)
	}
}

func (generated *playbackGenerated) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
	generated.mu.RLock()
	defer generated.mu.RUnlock()
	if generated.ctx.Err() != nil {
		writer.WriteHeader(http.StatusGone)
		return
	}
	name := strings.TrimPrefix(request.URL.Path, generated.base)
	if name == "index.m3u8" {
		if generated.failed.Load() && !generated.complete.Load() {
			writeJSON(writer, http.StatusBadGateway, map[string]string{"error": "媒体封装中断，请重新播放"})
			return
		}
		writer.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
		http.ServeContent(writer, request, name, time.Time{}, strings.NewReader(generated.playlist))
		return
	}
	if !generated.files[name] {
		http.NotFound(writer, request)
		return
	}
	file, err := os.Open(filepath.Join(generated.directory, name))
	if err != nil {
		http.NotFound(writer, request)
		return
	}
	defer file.Close()
	writer.Header().Set("Content-Type", "video/mp4")
	http.ServeContent(writer, request, name, time.Time{}, file)
}