nsfwapp/backend/transcode.go
2026-02-09 14:42:56 +01:00

458 lines
10 KiB
Go

// backend\transcode.go
package main
import (
"bytes"
"context"
"errors"
"fmt"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"golang.org/x/sync/singleflight"
)
// -------------------------
// Transcode config / globals
// -------------------------
// max parallel ffmpeg jobs
var transcodeSem = make(chan struct{}, 2)
// de-dupe concurrent requests for same output
var transcodeSF singleflight.Group
type heightCacheEntry struct {
mtime time.Time
size int64
height int
}
var heightCacheMu sync.Mutex
var heightCache = map[string]heightCacheEntry{}
func probeVideoHeight(ctx context.Context, inPath string) (int, error) {
// ffprobe -v error -select_streams v:0 -show_entries stream=height -of csv=p=0 <file>
cmd := exec.CommandContext(ctx, "ffprobe",
"-v", "error",
"-select_streams", "v:0",
"-show_entries", "stream=height",
"-of", "csv=p=0",
inPath,
)
out, err := cmd.Output()
if err != nil {
return 0, err
}
s := strings.TrimSpace(string(out))
if s == "" {
return 0, fmt.Errorf("ffprobe returned empty height")
}
h, err := strconv.Atoi(s)
if err != nil || h <= 0 {
return 0, fmt.Errorf("bad height %q", s)
}
return h, nil
}
func getVideoHeightCached(ctx context.Context, inPath string) (int, error) {
fi, err := os.Stat(inPath)
if err != nil || fi.IsDir() || fi.Size() <= 0 {
return 0, fmt.Errorf("input not usable")
}
heightCacheMu.Lock()
if e, ok := heightCache[inPath]; ok {
if e.size == fi.Size() && e.mtime.Equal(fi.ModTime()) && e.height > 0 {
h := e.height
heightCacheMu.Unlock()
return h, nil
}
}
heightCacheMu.Unlock()
h, err := probeVideoHeight(ctx, inPath)
if err != nil {
return 0, err
}
heightCacheMu.Lock()
heightCache[inPath] = heightCacheEntry{mtime: fi.ModTime(), size: fi.Size(), height: h}
heightCacheMu.Unlock()
return h, nil
}
type TranscodeProfile struct {
Name string // "1080p" | "720p" | "480p"
Height int
}
func profileFromQuality(q string) (TranscodeProfile, bool) {
switch strings.ToLower(strings.TrimSpace(q)) {
case "", "auto":
return TranscodeProfile{Name: "auto", Height: 0}, true
case "2160p":
return TranscodeProfile{Name: "2160p", Height: 2160}, true
case "1080p":
return TranscodeProfile{Name: "1080p", Height: 1080}, true
case "720p":
return TranscodeProfile{Name: "720p", Height: 720}, true
case "480p":
return TranscodeProfile{Name: "480p", Height: 480}, true
default:
return TranscodeProfile{}, false
}
}
// Cache layout: <doneAbs>/.transcodes/<canonicalID>/<quality>.mp4
func transcodeCachePath(doneAbs, canonicalID, quality string) string {
const v = "v1"
return filepath.Join(doneAbs, ".transcodes", canonicalID, v, quality+".mp4")
}
func ensureFFmpegAvailable() error {
_, err := exec.LookPath("ffmpeg")
if err != nil {
return fmt.Errorf("ffmpeg not found in PATH")
}
return nil
}
func ensureFFprobeAvailable() error {
_, err := exec.LookPath("ffprobe")
if err != nil {
return fmt.Errorf("ffprobe not found in PATH")
}
return nil
}
func fileUsable(p string) (os.FileInfo, bool) {
fi, err := os.Stat(p)
if err != nil {
return nil, false
}
if fi.IsDir() || fi.Size() <= 0 {
return nil, false
}
return fi, true
}
func isCacheFresh(inPath, outPath string) bool {
inFi, err := os.Stat(inPath)
if err != nil || inFi.IsDir() || inFi.Size() <= 0 {
return false
}
outFi, ok := fileUsable(outPath)
if !ok {
return false
}
// if out is not older than input -> ok
return !outFi.ModTime().Before(inFi.ModTime())
}
func acquireTranscodeSlot(ctx context.Context) error {
select {
case transcodeSem <- struct{}{}:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func releaseTranscodeSlot() {
select {
case <-transcodeSem:
default:
}
}
func tailString(s string, max int) string {
s = strings.TrimSpace(s)
if len(s) <= max {
return s
}
return s[len(s)-max:]
}
func runFFmpeg(ctx context.Context, args []string) error {
cmd := exec.CommandContext(ctx, "ffmpeg", args...)
var buf bytes.Buffer
cmd.Stdout = &buf
cmd.Stderr = &buf
err := cmd.Run()
if err == nil {
return nil
}
// Wenn ctx abgebrochen wurde (Timeout oder Cancel), gib Output trotzdem mit aus.
if ctx.Err() != nil {
return fmt.Errorf("ffmpeg aborted: %v (output=%s)", ctx.Err(), tailString(buf.String(), 4000))
}
return fmt.Errorf("ffmpeg failed: %w (output=%s)", err, tailString(buf.String(), 4000))
}
// -------------------------
// Public entry used by recordVideo
// -------------------------
// maybeTranscodeForRequest inspects "quality" query param.
// If quality is "auto" (or empty), it returns original outPath unchanged.
// Otherwise it ensures cached transcode exists & is fresh, and returns the cached path.
func maybeTranscodeForRequest(rctx context.Context, originalPath string, quality string) (string, error) {
prof, ok := profileFromQuality(quality)
if !ok {
return "", fmt.Errorf("bad quality %q", quality)
}
if prof.Name == "auto" {
return originalPath, nil
}
// ensure ffmpeg is present
if err := ensureFFmpegAvailable(); err != nil {
return "", err
}
// optional: skip transcode if source is already <= requested height (prevents upscaling)
if prof.Height > 0 {
// ffprobe is needed only for this optimization
if err := ensureFFprobeAvailable(); err == nil {
// short timeout for probing
pctx, cancel := context.WithTimeout(rctx, 5*time.Second)
defer cancel()
if srcH, err := getVideoHeightCached(pctx, originalPath); err == nil && srcH > 0 {
// if source is already at/below requested (with tiny tolerance), don't transcode
if srcH <= prof.Height+8 {
return originalPath, nil
}
}
}
}
// Need doneAbs for cache root
s := getSettings()
doneAbs, err := resolvePathRelativeToApp(s.DoneDir)
if err != nil || strings.TrimSpace(doneAbs) == "" {
return "", fmt.Errorf("doneDir missing or invalid")
}
// canonicalID = basename stem without ext and without "HOT "
base := filepath.Base(originalPath)
stem := strings.TrimSuffix(base, filepath.Ext(base))
canonicalID := stripHotPrefix(stem)
canonicalID = strings.TrimSpace(canonicalID)
if canonicalID == "" {
return "", fmt.Errorf("canonical id empty")
}
cacheOut := transcodeCachePath(doneAbs, canonicalID, prof.Name)
// fast path: already exists & fresh
if isCacheFresh(originalPath, cacheOut) {
return cacheOut, nil
}
// singleflight key: input + cacheOut
key := originalPath + "|" + cacheOut
_, err, _ = transcodeSF.Do(key, func() (any, error) {
// check again inside singleflight
if isCacheFresh(originalPath, cacheOut) {
return nil, nil
}
// If stale exists, remove (best-effort)
_ = os.Remove(cacheOut)
// ensure dir
if err := os.MkdirAll(filepath.Dir(cacheOut), 0o755); err != nil {
return nil, err
}
// timeout for transcode
// ✅ NICHT an rctx hängen, sonst killt Client-Abbruch ffmpeg beim Quality-Wechsel
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
defer cancel()
if err := acquireTranscodeSlot(ctx); err != nil {
return nil, err
}
defer releaseTranscodeSlot()
// ✅ Temp muss eine "echte" Video-Endung haben, sonst kann ffmpeg das Format nicht wählen
tmp := cacheOut + ".part.mp4"
_ = os.Remove(tmp)
// ffmpeg args
args := buildFFmpegArgs(originalPath, tmp, prof)
if err := runFFmpeg(ctx, args); err != nil {
_ = os.Remove(tmp)
if errors.Is(ctx.Err(), context.DeadlineExceeded) {
return nil, fmt.Errorf("transcode timeout: %w", err)
}
return nil, err
}
// validate tmp
if _, ok := fileUsable(tmp); !ok {
_ = os.Remove(tmp)
return nil, fmt.Errorf("transcode output invalid")
}
// atomic replace
_ = os.Remove(cacheOut)
if err := os.Rename(tmp, cacheOut); err != nil {
_ = os.Remove(tmp)
return nil, err
}
return nil, nil
})
if err != nil {
return "", err
}
// final validate
if _, ok := fileUsable(cacheOut); !ok {
return "", fmt.Errorf("transcode cache missing after build")
}
return cacheOut, nil
}
// -------------------------
// ffmpeg profiles
// -------------------------
func buildFFmpegArgs(inPath, outPath string, prof TranscodeProfile) []string {
// You can tune these defaults:
// - CRF: lower => better quality, bigger file (1080p ~22, 720p ~23, 480p ~24/25)
// - preset: veryfast is good for on-demand
crf := "23"
switch prof.Name {
case "1080p":
crf = "22"
case "720p":
crf = "23"
case "480p":
crf = "25"
}
// Keyframes: choose a stable value; if you want dynamic based on fps you can extend later.
gop := "60"
// ✅ Für fertige MP4-Dateien: NICHT fragmentieren.
// faststart reicht, damit "moov" vorne liegt.
movflags := "+faststart"
// scale keeps aspect ratio, ensures even width
vf := fmt.Sprintf("scale=-2:%d", prof.Height)
return []string{
"-hide_banner",
"-loglevel", "error",
"-nostdin",
"-y",
"-i", inPath,
// ✅ robust: falls Audio fehlt, trotzdem kein Fehler
"-map", "0:v:0?",
"-map", "0:a:0?",
"-sn",
"-vf", vf,
"-c:v", "libx264",
"-preset", "veryfast",
"-crf", crf,
"-pix_fmt", "yuv420p",
"-max_muxing_queue_size", "1024",
"-g", gop,
"-keyint_min", gop,
"-sc_threshold", "0",
// Audio nur wenn vorhanden (wegen "-map 0:a:0?")
"-c:a", "aac",
"-b:a", "128k",
"-ac", "2",
"-movflags", movflags,
outPath,
}
}
func buildFFmpegStreamArgs(inPath string, prof TranscodeProfile) []string {
crf := "23"
switch prof.Name {
case "1080p":
crf = "22"
case "720p":
crf = "23"
case "480p":
crf = "25"
}
gop := "60"
vf := fmt.Sprintf("scale=-2:%d", prof.Height)
movflags := "frag_keyframe+empty_moov+default_base_moof"
return []string{
"-hide_banner",
"-loglevel", "error",
"-nostdin",
"-y",
"-i", inPath,
// ✅ robust (wie im File-Transcode)
"-map", "0:v:0?",
"-map", "0:a:0?",
"-sn",
"-vf", vf,
"-c:v", "libx264",
"-preset", "veryfast",
"-crf", crf,
"-pix_fmt", "yuv420p",
"-max_muxing_queue_size", "1024",
"-g", gop,
"-keyint_min", gop,
"-sc_threshold", "0",
"-c:a", "aac",
"-b:a", "128k",
"-ac", "2",
"-movflags", movflags,
"-f", "mp4",
"pipe:1",
}
}
// -------------------------
// Cleanup helper
// -------------------------
func removeTranscodesForID(doneAbs, canonicalID string) {
_ = os.RemoveAll(filepath.Join(doneAbs, ".transcodes", canonicalID))
}