nsfwapp/backend/record_stream_cb.go

// backend\record_stream_cb.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"
)

// RecordStream für Chaturbate:
// bewusst WIEDER mit ffmpeg-Input auf die HLS-Playlist.
// Wichtig:
// - Seite laden
// - HLS URL parsen
// - beste Playlist bestimmen
// - wenn separate Audio-Group vorhanden ist, Master-Playlist an ffmpeg geben
// - ffmpeg übernimmt Video + beste verfügbare Audio-Spur
func RecordStream(
	ctx context.Context,
	hc *HTTPClient,
	domain string,
	username string,
	outputPath string,
	httpCookie string,
	job *RecordJob,
) error {
	username = strings.Trim(strings.TrimSpace(username), "/")
	if username == "" {
		return errors.New("leerer username")
	}
	base := strings.TrimRight(domain, "/")
	pageURL := base + "/" + username + "/"

	loadFreshHLS := func() (*selectedHLSStream, error) {
		body, err := hc.FetchPage(ctx, pageURL, httpCookie)
		if err != nil {
			return nil, appErrorf("[%s] seite laden: %w", username, err)
		}

		hlsURL, err := ParseStream(body)
		if err != nil {
			return nil, appErrorf("[%s] stream-parsing: %w", username, err)
		}

		hlsURL = strings.TrimSpace(hlsURL)
		if hlsURL == "" {
			return nil, appErrorf("[%s] leere hls url", username)
		}

		stream, err := getWantedResolutionPlaylistWithHeaders(
			ctx,
			hlsURL,
			httpCookie,
			hc.userAgent,
			pageURL,
		)
		if err != nil {
			return nil, appErrorf("[%s] variant-playlist: %w", username, err)
		}
		if stream == nil || strings.TrimSpace(stream.VideoURL) == "" {
			return nil, appErrorf("[%s] leere final video url", username)
		}

		return stream, nil
	}

	isRetryableFreshHLSError := func(err error) bool {
		if err == nil {
			return false
		}

		msg := strings.ToLower(strings.TrimSpace(err.Error()))

		return strings.Contains(msg, "timeout") ||
			strings.Contains(msg, "deadline exceeded") ||
			strings.Contains(msg, "awaiting headers") ||
			strings.Contains(msg, "connection reset") ||
			strings.Contains(msg, "connection refused") ||
			strings.Contains(msg, "eof") ||
			strings.Contains(msg, "tls handshake timeout") ||
			strings.Contains(msg, "temporary failure") ||
			strings.Contains(msg, "stream-parsing") ||
			strings.Contains(msg, "room dossier nicht gefunden") ||
			strings.Contains(msg, "kein hls-quell-url") ||
			strings.Contains(msg, "leere hls url") ||
			strings.Contains(msg, "variant-playlist") ||
			strings.Contains(msg, "http 401") ||
			strings.Contains(msg, "http 403") ||
			strings.Contains(msg, "http 429") ||
			strings.Contains(msg, "http 5")
	}

	loadFreshHLSWithRetry := func(maxAttempts int, reason string) (*selectedHLSStream, error) {
		if maxAttempts < 1 {
			maxAttempts = 1
		}

		var lastErr error

		for attempt := 1; attempt <= maxAttempts; attempt++ {
			stream, err := loadFreshHLS()
			if err == nil {
				return stream, nil
			}

			lastErr = err

			if ctx.Err() != nil {
				return nil, ctx.Err()
			}

			if !isRetryableFreshHLSError(err) {
				return nil, err
			}

			if attempt >= maxAttempts {
				break
			}

			wait := time.Duration(attempt) * 1500 * time.Millisecond
			if wait > 8*time.Second {
				wait = 8 * time.Second
			}

			if verboseLogs() {
				appLogln(
					"⚠️ chaturbate hls refresh retry",
					reason,
					"attempt",
					attempt,
					"of",
					maxAttempts,
					"error:",
					err,
				)
			}

			select {
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-time.After(wait):
			}
		}

		if lastErr == nil {
			lastErr = errors.New("hls refresh failed")
		}

		return nil, lastErr
	}

	updatePreviewFromStream := func(stream *selectedHLSStream) {
		if job == nil || stream == nil {
			return
		}

		jobsMu.Lock()
		previewURL := strings.TrimSpace(stream.VideoURL)
		if stream.HasSeparateAudio && strings.TrimSpace(stream.MasterURL) != "" {
			previewURL = strings.TrimSpace(stream.MasterURL)
		}
		job.PreviewM3U8 = previewURL
		job.PreviewCookie = httpCookie
		job.PreviewUA = hc.userAgent
		jobsMu.Unlock()
	}

	stream, err := loadFreshHLSWithRetry(5, "initial")
	if err != nil {
		return appErrorf(
			"chaturbate seite/playlist konnte nach mehreren versuchen nicht geladen werden: %w",
			err,
		)
	}

	if job != nil {
		assetID := assetIDForJob(job)
		if assetID != "" {
			_, _ = ensureGeneratedDir(assetID)
		}
	}

	updatePreviewFromStream(stream)

	const maxPlaylistRefreshes = 10
	refreshes := 0
	attempt := 0
	// Der Segment-Recorder schreibt selbst fortlaufend in die Hauptdatei.
	// Retry-Chunks würden nur .retry-Dateien im records-Ordner erzeugen.
	useChunkMerge := false

	for {
		attempt++

		attemptOut := outputPath
		mergeAfterRun := false

		// Nur ab dem ZWEITEN Versuch in Retry-Chunk schreiben.
		// Der erste Versuch muss direkt in outputPath schreiben,
		// damit der laufende Download eine echte Hauptdatei hat.
		if useChunkMerge && attempt > 1 {
			attemptOut = hlsRetryChunkPath(outputPath, attempt)
			_ = os.Remove(attemptOut)
			mergeAfterRun = true
		}

		err = recordHLSPlaylistSegments(ctx, stream, attemptOut, job, httpCookie, hc.userAgent, pageURL)

		if mergeAfterRun && fileExistsNonEmpty(attemptOut) {
			if aerr := appendFileAndRemove(outputPath, attemptOut); aerr != nil {
				return appErrorf("retry-chunk merge failed: %w", aerr)
			}

			if job != nil {
				if fi, statErr := os.Stat(outputPath); statErr == nil && fi != nil && !fi.IsDir() {
					jobsMu.Lock()
					job.SizeBytes = fi.Size()
					jobsMu.Unlock()
					_ = publishJobSnapshot(job.ID)
				}
			}
		}

		if err == nil {
			return nil
		}

		if ctx.Err() != nil {
			return ctx.Err()
		}

		msg := strings.ToLower(strings.TrimSpace(err.Error()))
		shouldRefresh :=
			strings.Contains(msg, "403") ||
				strings.Contains(msg, "401") ||
				strings.Contains(msg, "429") ||
				strings.Contains(msg, "invalid data found") ||
				strings.Contains(msg, "error opening input") ||
				strings.Contains(msg, "stream-parsing") ||
				strings.Contains(msg, "kein hls-quell-url") ||
				strings.Contains(msg, "http ") ||
				strings.Contains(msg, "timeout") ||
				strings.Contains(msg, "deadline exceeded") ||
				strings.Contains(msg, "awaiting headers") ||
				strings.Contains(msg, "connection reset") ||
				strings.Contains(msg, "eof")

		if !shouldRefresh {
			return err
		}

		refreshes++
		if refreshes > maxPlaylistRefreshes {
			return appErrorf("cb refresh-limit erreicht nach %d versuchen: %w", refreshes-1, err)
		}

		wait := 1500 * time.Millisecond

		// Wenn die Seite gerade keinen hls_source enthält, bringt schneller Spam nichts.
		// Das passiert bei CB oft bei public/private-Flackern oder wenn der Room noch nicht
		// vollständig initialisiert ist.
		if strings.Contains(msg, "kein hls-quell-url") ||
			strings.Contains(msg, "room dossier nicht gefunden") ||
			strings.Contains(msg, "stream-parsing") {
			wait = 10 * time.Second
		}

		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(wait):
		}

		newStream, rerr := loadFreshHLSWithRetry(3, "after-ffmpeg-error")
		if rerr != nil {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(2 * time.Second):
			}
			continue
		}

		stream = newStream
		updatePreviewFromStream(stream)
	}
}

// ParseStream (roomDossier → hls_source)
func ParseStream(html string) (string, error) {
	matches := roomDossierRegexp.FindStringSubmatch(html)
	if len(matches) == 0 {
		return "", errors.New("room dossier nicht gefunden")
	}

	decoded, err := strconv.Unquote(
		strings.Replace(strconv.Quote(matches[1]), `\\u`, `\u`, -1),
	)
	if err != nil {
		return "", appErrorf("Unicode-decode failed: %w", err)
	}

	var rd struct {
		HLSSource string `json:"hls_source"`
	}
	if err := json.Unmarshal([]byte(decoded), &rd); err != nil {
		return "", appErrorf("JSON-parse failed: %w", err)
	}
	if rd.HLSSource == "" {
		return "", errors.New("kein HLS-Quell-URL im JSON")
	}
	return rd.HLSSource, nil
}

// Cookie-Hilfsfunktion
func addCookiesFromString(req *http.Request, cookieStr string) {
	if cookieStr == "" {
		return
	}
	pairs := strings.Split(cookieStr, ";")
	for _, pair := range pairs {
		parts := strings.SplitN(strings.TrimSpace(pair), "=", 2)
		if len(parts) != 2 {
			continue
		}
		name := strings.TrimSpace(parts[0])
		value := strings.TrimSpace(parts[1])
		if name == "" {
			continue
		}
		req.AddCookie(&http.Cookie{
			Name:  name,
			Value: value,
		})
	}
}

// helper
func extractUsername(input string) string {
	s := strings.TrimSpace(input)
	if s == "" {
		return ""
	}

	if u, err := url.Parse(s); err == nil && u != nil {
		host := strings.ToLower(strings.TrimSpace(u.Hostname()))
		if strings.Contains(host, "chaturbate.com") {
			p := strings.Trim(u.Path, "/")
			if p == "" {
				return ""
			}
			parts := strings.Split(p, "/")
			return strings.TrimSpace(parts[0])
		}
	}

	s = strings.TrimPrefix(s, "https://")
	s = strings.TrimPrefix(s, "http://")
	s = strings.TrimPrefix(s, "www.")
	if strings.HasPrefix(strings.ToLower(s), "chaturbate.com/") {
		s = s[len("chaturbate.com/"):]
	}
	if idx := strings.IndexAny(s, "/?#"); idx != -1 {
		s = s[:idx]
	}
	return strings.Trim(s, "/\\")
}

func hasChaturbateCookies(cookieStr string) bool {
	m := parseCookieString(cookieStr)
	_, hasCF := m["cf_clearance"]
	_, hasSessID := m["session_id"]
	_, hasSessIdAlt := m["sessionid"]
	return hasCF && (hasSessID || hasSessIdAlt)
}

func parseCookieString(cookieStr string) map[string]string {
	out := map[string]string{}
	for _, pair := range strings.Split(cookieStr, ";") {
		parts := strings.SplitN(strings.TrimSpace(pair), "=", 2)
		if len(parts) != 2 {
			continue
		}
		name := strings.TrimSpace(parts[0])
		value := strings.TrimSpace(parts[1])
		if name == "" {
			continue
		}
		out[strings.ToLower(name)] = value
	}
	return out
}

func detectProvider(raw string) string {
	s := strings.ToLower(raw)

	if strings.Contains(s, "chaturbate.com") {
		return "chaturbate"
	}
	if strings.Contains(s, "myfreecams.com") {
		return "mfc"
	}
	return "unknown"
}