chris/nsfwapp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updated

Browse Source
This commit is contained in:
Linrador 2026-05-24 21:13:11 +02:00
parent c9f2086b79
commit dccb6ff1ae
3 changed files with 339 additions and 180 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

176
backend/main.go
View File

@ -823,25 +823,14 @@ func (h *HTTPClient) FetchPage(ctx context.Context, url, cookieStr string) (stri
}
func remuxTSToMP4(tsPath, mp4Path string) error {
// ffmpeg -y -i in.ts -c copy -movflags +faststart out.mp4
cmd := exec.Command(ffmpegPath,
"-y",
"-i", tsPath,
"-c", "copy",
"-movflags", "+faststart",
return remuxTSToMP4WithProgress(
context.Background(),
tsPath,
mp4Path,
0,
0,
nil,
)
cmd.SysProcAttr = &syscall.SysProcAttr{
HideWindow: true,
CreationFlags: 0x08000000, // CREATE_NO_WINDOW
}
var stderr bytes.Buffer
cmd.Stderr = &stderr
if err := cmd.Run(); err != nil {
return appErrorf("ffmpeg remux failed: %v (%s)", err, stderr.String())
}
return nil
}
func parseFFmpegOutTime(v string) float64 {
@ -849,16 +838,19 @@ func parseFFmpegOutTime(v string) float64 {
if v == "" {
return 0
}
parts := strings.Split(v, ":")
if len(parts) != 3 {
return 0
}
h, err1 := strconv.Atoi(parts[0])
m, err2 := strconv.Atoi(parts[1])
s, err3 := strconv.ParseFloat(parts[2], 64) // Sekunden können Dezimalstellen haben
s, err3 := strconv.ParseFloat(parts[2], 64)
if err1 != nil || err2 != nil || err3 != nil {
return 0
}
return float64(h*3600+m*60) + s
}
@ -869,7 +861,83 @@ func remuxTSToMP4WithProgress(
inSize int64,
onRatio func(r float64),
) error {
// ffmpeg progress kommt auf stdout als key=value
tsPath = strings.TrimSpace(tsPath)
mp4Path = strings.TrimSpace(mp4Path)
if tsPath == "" {
return appErrorf("remux input path empty")
}
if mp4Path == "" {
return appErrorf("remux output path empty")
}
_ = removeWithRetry(mp4Path)
// 1) Normaler schneller Remux.
err := runRemuxTSToMP4AttemptWithProgress(
ctx,
tsPath,
mp4Path,
durationSec,
inSize,
onRatio,
false,
)
if err == nil {
return nil
}
if ctx != nil && ctx.Err() != nil {
_ = removeWithRetry(mp4Path)
return err
}
firstErr := err
_ = removeWithRetry(mp4Path)
appLogln(
"⚠️ normal remux failed, retrying tolerant remux:",
filepath.Base(tsPath),
firstErr,
)
// 2) Toleranter Remux für Stream-Ende, kaputte Segmente, Timestamp-Probleme.
err = runRemuxTSToMP4AttemptWithProgress(
ctx,
tsPath,
mp4Path,
durationSec,
inSize,
onRatio,
true,
)
if err == nil {
return nil
}
_ = removeWithRetry(mp4Path)
return appErrorf(
"normal remux failed (%v), tolerant remux failed: %w",
firstErr,
err,
)
}
func runRemuxTSToMP4AttemptWithProgress(
ctx context.Context,
tsPath string,
mp4Path string,
durationSec float64,
inSize int64,
onRatio func(r float64),
tolerant bool,
) error {
if ctx == nil {
ctx = context.Background()
}
args := []string{
"-y",
"-nostats",
@ -877,18 +945,36 @@ func remuxTSToMP4WithProgress(
"-hide_banner",
"-loglevel", "error",
}
if tolerant {
args = append(args,
"-fflags", "+genpts+discardcorrupt",
"-err_detect", "ignore_err",
)
} else {
args = append(args, ffmpegInputTol...)
}
args = append(args,
"-i", tsPath,
// Video muss vorhanden sein.
"-map", "0:v:0",
"-map", "0:a:0?",
// Audio optional. Wenn kein Audio da ist, darf Remux trotzdem durchlaufen.
"-map", "0:a?",
"-dn",
"-ignore_unknown",
"-avoid_negative_ts", "make_zero",
"-max_interleave_delta", "0",
"-muxpreload", "0",
"-muxdelay", "0",
// Wichtig: kein Reencode, nur Containerwechsel TS -> MP4.
"-c", "copy",
"-movflags", "+faststart",
mp4Path,
)
@ -911,6 +997,10 @@ func remuxTSToMP4WithProgress(
return err
}
if onRatio != nil {
onRatio(0)
}
sc := bufio.NewScanner(stdout)
sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
@ -920,7 +1010,10 @@ func remuxTSToMP4WithProgress(
)
send := func(outSec float64, totalSize int64, force bool) {
// bevorzugt: Zeit/Dauer
if onRatio == nil {
return
}
if durationSec > 0 && outSec > 0 {
r := outSec / durationSec
if r < 0 {
@ -929,13 +1022,10 @@ func remuxTSToMP4WithProgress(
if r > 1 {
r = 1
}
if onRatio != nil {
onRatio(r)
}
return
}
// fallback: Bytes (bei remux meist okay-ish)
if inSize > 0 && totalSize > 0 {
r := float64(totalSize) / float64(inSize)
if r < 0 {
@ -944,14 +1034,11 @@ func remuxTSToMP4WithProgress(
if r > 1 {
r = 1
}
if onRatio != nil {
onRatio(r)
}
return
}
// force (z.B. end)
if force && onRatio != nil {
if force {
onRatio(1)
}
}
@ -961,33 +1048,37 @@ func remuxTSToMP4WithProgress(
if line == "" {
continue
}
k, v, ok := strings.Cut(line, "=")
if !ok {
continue
}
switch k {
switch strings.TrimSpace(k) {
case "out_time_us":
if n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64); err == nil && n > 0 {
lastOutSec = float64(n) / 1_000_000.0
send(lastOutSec, lastTotalSz, false)
}
case "out_time_ms":
if n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64); err == nil && n > 0 {
// out_time_ms ist i.d.R. Millisekunden
lastOutSec = float64(n) / 1_000.0
send(lastOutSec, lastTotalSz, false)
}
case "out_time":
if s := parseFFmpegOutTime(v); s > 0 {
lastOutSec = s
send(lastOutSec, lastTotalSz, false)
}
case "total_size":
if n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64); err == nil && n > 0 {
lastTotalSz = n
send(lastOutSec, lastTotalSz, false)
}
case "progress":
if strings.TrimSpace(v) == "end" {
send(lastOutSec, lastTotalSz, true)
@ -995,9 +1086,30 @@ func remuxTSToMP4WithProgress(
}
}
if err := cmd.Wait(); err != nil {
return appErrorf("ffmpeg remux failed: %v (%s)", err, strings.TrimSpace(stderr.String()))
scanErr := sc.Err()
waitErr := cmd.Wait()
if waitErr != nil {
return appErrorf(
"ffmpeg remux failed tolerant=%v: %v (%s)",
tolerant,
waitErr,
strings.TrimSpace(stderr.String()),
)
}
if scanErr != nil {
return appErrorf("ffmpeg progress read failed tolerant=%v: %w", tolerant, scanErr)
}
if err := requireNonEmptyRegularFile(mp4Path, "remux result"); err != nil {
return err
}
if onRatio != nil {
onRatio(1)
}
return nil
}

17
backend/record_stream_hls.go
View File

@ -470,6 +470,18 @@ func handleM3U8Mode(
ck := strings.TrimSpace(httpCookie)
appendInputOptions := func(args []string) []string {
args = append(args,
"-rw_timeout", "15000000",
// HLS-Segmente robuster laden:
// kurze Netzwerkhänger / einzelne kaputte Segment-Requests nicht sofort hart abbrechen.
"-reconnect", "1",
"-reconnect_streamed", "1",
"-reconnect_on_network_error", "1",
"-reconnect_on_http_error", "403,404,408,429,500,502,503,504",
"-reconnect_delay_max", "4",
)
if ua != "" {
args = append(args, "-user_agent", ua)
}
@ -515,6 +527,11 @@ func handleM3U8Mode(
case ".ts":
args = append(args,
"-f", "mpegts",
// Hilft, wenn Retry-Chunks später byteweise an die Haupt-TS angehängt werden.
// PAT/PMT Header werden öfter erneut geschrieben.
"-mpegts_flags", "+resend_headers",
outFile,
)
case ".mp4":

318
backend/recorder.go
View File

@ -3,9 +3,9 @@ package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"math"
"net/http"
"net/url"
@ -370,6 +370,134 @@ func waitForUsableOutput(path string, timeout time.Duration) (os.FileInfo, error
return nil, lastErr
}
func waitForStableUsableOutput(path string, timeout time.Duration, stableFor time.Duration) (os.FileInfo, error) {
path = strings.TrimSpace(path)
if path == "" {
return nil, appErrorf("path empty")
}
deadline := time.Now().Add(timeout)
var lastFi os.FileInfo
var lastSize int64 = -1
var stableSince time.Time
var lastErr error
for {
fi, err := os.Stat(path)
if err == nil && fi != nil && !fi.IsDir() && fi.Size() > 0 {
size := fi.Size()
if size == lastSize {
if stableSince.IsZero() {
stableSince = time.Now()
}
if time.Since(stableSince) >= stableFor {
return fi, nil
}
} else {
lastSize = size
lastFi = fi
stableSince = time.Now()
}
} else {
lastFi = fi
stableSince = time.Time{}
if err != nil {
lastErr = err
} else if fi == nil {
lastErr = appErrorf("stat returned nil")
} else if fi.IsDir() {
lastErr = appErrorf("path is dir")
} else {
lastErr = appErrorf("file size is 0")
}
}
if time.Now().After(deadline) {
if lastFi != nil && !lastFi.IsDir() && lastFi.Size() > 0 {
// Besser trotzdem weiterverarbeiten als eine fast fertige Datei wegzuwerfen.
return lastFi, nil
}
if lastErr == nil {
lastErr = appErrorf("stable output not ready in time")
}
return nil, lastErr
}
time.Sleep(250 * time.Millisecond)
}
}
func isLikelyNaturalStreamEndError(err error) bool {
if err == nil {
return false
}
// Manuelles Stoppen bleibt "stopped", nicht "finished".
if errors.Is(err, context.Canceled) {
return false
}
if errors.Is(err, io.EOF) {
return true
}
s := strings.ToLower(strings.TrimSpace(err.Error()))
return strings.Contains(s, "eof") ||
strings.Contains(s, "unexpected eof") ||
strings.Contains(s, "stream ended") ||
strings.Contains(s, "stream closed") ||
strings.Contains(s, "server closed") ||
strings.Contains(s, "connection reset") ||
strings.Contains(s, "connection refused") ||
strings.Contains(s, "playlist") ||
strings.Contains(s, "refresh-limit") ||
strings.Contains(s, "cb refresh-limit") ||
strings.Contains(s, "error opening input") ||
strings.Contains(s, "invalid data found") ||
// Model-/Room-Zustände
strings.Contains(s, "offline") ||
strings.Contains(s, "model offline") ||
strings.Contains(s, "room offline") ||
strings.Contains(s, "not broadcasting") ||
strings.Contains(s, "no longer available") ||
strings.Contains(s, "private") ||
strings.Contains(s, "hidden") ||
strings.Contains(s, "away") ||
// HTTP-Zustände, die bei Stream-Ende / Token-Wechsel häufig kommen
strings.Contains(s, "http 401") ||
strings.Contains(s, "http 403") ||
strings.Contains(s, "http 404") ||
strings.Contains(s, "http 410") ||
strings.Contains(s, "401") ||
strings.Contains(s, "403") ||
strings.Contains(s, "404") ||
strings.Contains(s, "410") ||
strings.Contains(s, "forbidden")
}
func hasUsableRecordedFileForNaturalEnd(out string) bool {
out = strings.TrimSpace(out)
if out == "" {
return false
}
fi, err := os.Stat(out)
if err != nil || fi == nil || fi.IsDir() {
return false
}
// Kleine Fragmente sollen weiterhin durch deine bestehende Auto-Delete-Logik laufen.
return fi.Size() > 0
}
func shouldKeepStoppedJobForInspection(target JobStatus, out string) bool {
if target != JobStopped {
return false
@ -598,105 +726,6 @@ func buildPostworkTempMP4Path(tsPath string) (string, error) {
return uniqueDestPath(tmpDir, base)
}
func moveRepairSourceTSBestEffort(tsPath string, finalMP4Path string) string {
tsPath = strings.TrimSpace(tsPath)
if tsPath == "" {
return ""
}
if err := requireNonEmptyRegularFile(tsPath, "repair source ts"); err != nil {
appLogln("⚠️ repair source ts not usable:", err)
return tsPath
}
s := getSettings()
doneAbs, err := resolvePathRelativeToApp(s.DoneDir)
if err != nil || strings.TrimSpace(doneAbs) == "" {
doneAbs = filepath.Dir(strings.TrimSpace(finalMP4Path))
}
doneAbs = strings.TrimSpace(doneAbs)
if doneAbs == "" {
return tsPath
}
repairDir := filepath.Join(doneAbs, ".repair_sources")
if err := os.MkdirAll(repairDir, 0o755); err != nil {
appLogln("⚠️ repair source dir create failed:", err)
return tsPath
}
dst, err := uniqueDestPath(repairDir, filepath.Base(tsPath))
if err != nil {
appLogln("⚠️ repair source unique path failed:", err)
return tsPath
}
if err := renameWithRetryAggressive(tsPath, dst); err != nil {
appLogln("⚠️ repair source move failed:", err)
return tsPath
}
purgeDurationCacheForPath(tsPath)
return dst
}
func markVideoNeedsRepairBestEffort(videoPath string, sourceTSPath string, reason string, detail string) {
videoPath = strings.TrimSpace(videoPath)
if videoPath == "" {
return
}
assetID := strings.TrimSpace(assetIDFromVideoPath(videoPath))
if assetID == "" {
assetID = canonicalAssetIDFromName(filepath.Base(videoPath))
}
if assetID == "" {
return
}
marker := map[string]any{
"needsRepair": true,
"reason": strings.TrimSpace(reason),
"detail": strings.TrimSpace(detail),
"videoPath": videoPath,
"sourceTSPath": strings.TrimSpace(sourceTSPath),
"createdAt": time.Now().UTC().Format(time.RFC3339Nano),
}
b, err := json.MarshalIndent(marker, "", " ")
if err != nil {
appLogln("⚠️ repair marker marshal failed:", err)
return
}
repairPath := ""
if _, _, _, _, metaPath, err := assetPathsForID(assetID); err == nil {
metaPath = strings.TrimSpace(metaPath)
if metaPath != "" {
dir := filepath.Dir(metaPath)
if err := os.MkdirAll(dir, 0o755); err == nil {
repairPath = filepath.Join(dir, "repair.json")
}
}
}
if repairPath == "" {
stem := strings.TrimSuffix(filepath.Base(videoPath), filepath.Ext(videoPath))
repairPath = filepath.Join(filepath.Dir(videoPath), stem+".repair.json")
}
if err := os.WriteFile(repairPath, b, 0o644); err != nil {
appLogln("⚠️ repair marker write failed:", err)
return
}
appLogln("⚠️ video marked for deferred repair:", filepath.Base(videoPath), "marker=", repairPath)
}
func runPrimaryPostworkPipeline(
ctx context.Context,
job *RecordJob,
@ -711,11 +740,6 @@ func runPrimaryPostworkPipeline(
cleanupTSAfterMove := ""
removeTempOutOnMoveFailure := false
needsRepair := false
repairReason := ""
repairDetail := ""
repairSourceTS := ""
// 1) TS -> MP4 zuerst
if strings.EqualFold(filepath.Ext(out), ".ts") {
tsPath := out
@ -763,6 +787,7 @@ func runPrimaryPostworkPipeline(
setPhase("remuxing", int(math.Round(r*100)))
},
)
if err != nil {
_ = removeWithRetry(mp4Path)
return tsPath, appErrorf("ts remux failed: %w", err)
@ -774,25 +799,10 @@ func runPrimaryPostworkPipeline(
}
if err := validateVideoDecodesNearEnd(ctx, mp4Path); err != nil {
// Wichtig:
// Nicht mehr inline reencoden. Reencode ist CPU-teuer und blockiert die postWorkQ.
// Das remuxte MP4 bleibt erhalten und wird als "needs repair" markiert.
appLogln("⚠️ remux mp4 decode validation failed; keeping remux result and deferring repair:", err)
needsRepair = true
repairReason = "decode_near_end_failed"
repairDetail = err.Error()
repairSourceTS = tsPath
appLogln("⚠️ remux mp4 decode validation failed; keeping remux result without repair:", err)
}
// TS erst NACH erfolgreichem Move löschen.
// Bei Repair-Verdacht behalten wir die TS als Reparaturquelle.
if !needsRepair {
cleanupTSAfterMove = tsPath
} else {
cleanupTSAfterMove = ""
repairSourceTS = tsPath
}
removeTempOutOnMoveFailure = true
out = mp4Path
@ -832,15 +842,6 @@ func runPrimaryPostworkPipeline(
return out, err
}
if needsRepair {
keptSource := strings.TrimSpace(repairSourceTS)
if keptSource != "" {
keptSource = moveRepairSourceTSBestEffort(keptSource, out)
}
markVideoNeedsRepairBestEffort(out, keptSource, repairReason, repairDetail)
}
if cleanupTSAfterMove != "" {
if err := removeWithRetry(cleanupTSAfterMove); err != nil && !os.IsNotExist(err) {
appLogln("⚠️ ts cleanup after successful move:", err)
@ -1068,13 +1069,37 @@ func runMFCRecording(
func finalizeRecordingAttempt(job *RecordJob, runErr error) (JobStatus, string) {
end := time.Now()
jobsMu.RLock()
out := strings.TrimSpace(job.Output)
jobsMu.RUnlock()
target := JobFinished
var errText string
errText := ""
if runErr != nil {
if errors.Is(runErr, context.Canceled) {
switch {
case errors.Is(runErr, context.Canceled):
// Manuelles Stoppen oder Auto-Stop durch deine Logik:
// Datei darf trotzdem nachbearbeitet werden, aber Status bleibt "stopped".
target = JobStopped
} else {
case isLikelyNaturalStreamEndError(runErr) && hasUsableRecordedFileForNaturalEnd(out):
// Wichtig:
// Stream ist wahrscheinlich normal weggekippt, aber wir haben Daten.
// Nicht als failed markieren, sondern wie fertige Aufnahme behandeln.
target = JobFinished
errText = ""
if verboseLogs() {
appLogln(
"ℹ️ recording ended naturally, keeping output:",
filepath.Base(out),
"reason:",
runErr,
)
}
default:
target = JobFailed
errText = runErr.Error()
}
@ -1086,11 +1111,15 @@ func finalizeRecordingAttempt(job *RecordJob, runErr error) (JobStatus, string)
job.EndedAt = &end
job.EndedAtMs = end.UnixMilli()
job.Status = target
job.Phase = "postwork"
if errText != "" {
job.Error = errText
} else {
job.Error = ""
}
job.Phase = "postwork"
out := strings.TrimSpace(job.Output)
out = strings.TrimSpace(job.Output)
jobsMu.Unlock()
return target, out
@ -1115,17 +1144,14 @@ func prepareOutputForPostwork(job *RecordJob, target JobStatus, out string) bool
}
func ensureUsableRecordedOutput(job *RecordJob, target JobStatus, out string) bool {
var (
fi os.FileInfo
err error
)
if target == JobStopped {
fi, err = waitForUsableOutput(out, 3*time.Second)
} else {
fi, err = os.Stat(out)
out = strings.TrimSpace(out)
if out == "" {
clearJobPostworkState(job)
removeJobAndPublish(job)
return false
}
fi, err := waitForStableUsableOutput(out, 8*time.Second, 750*time.Millisecond)
if err == nil && fi != nil && !fi.IsDir() && fi.Size() > 0 {
return true
}
@ -1135,6 +1161,10 @@ func ensureUsableRecordedOutput(job *RecordJob, target JobStatus, out string) bo
return false
}
if verboseLogs() {
appLogln("⚠️ recorded output unusable, removing:", out, "err:", err, "stat:", statOutputForLog(out))
}
_ = removeWithRetry(out)
purgeDurationCacheForPath(out)
removeJobAndPublish(job)