From dccb6ff1ae043ade63e0409c4c85086054d1817f Mon Sep 17 00:00:00 2001 From: Linrador <68631622+Linrador@users.noreply.github.com> Date: Sun, 24 May 2026 21:13:11 +0200 Subject: [PATCH] updated --- backend/main.go | 182 ++++++++++++++++---- backend/record_stream_hls.go | 17 ++ backend/recorder.go | 320 +++++++++++++++++++---------------- 3 files changed, 339 insertions(+), 180 deletions(-) diff --git a/backend/main.go b/backend/main.go index f822340..12768c3 100644 --- a/backend/main.go +++ b/backend/main.go @@ -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", } - args = append(args, ffmpegInputTol...) + + 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) - } + 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) - } + 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 } diff --git a/backend/record_stream_hls.go b/backend/record_stream_hls.go index dd2acdf..a6ec58c 100644 --- a/backend/record_stream_hls.go +++ b/backend/record_stream_hls.go @@ -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": diff --git a/backend/recorder.go b/backend/recorder.go index 103d67f..40718c5 100644 --- a/backend/recorder.go +++ b/backend/recorder.go @@ -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 - } + cleanupTSAfterMove = 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)