From 9cecea41f3e064b2dffb3f09e2c8521c98a23ce5 Mon Sep 17 00:00:00 2001 From: Tanq16 <37408906+Tanq16@users.noreply.github.com> Date: Fri, 14 Nov 2025 23:52:17 +0000 Subject: [PATCH] add dailymotion extractor; fix hls segment store --- internal/downloaders/live-stream/download.go | 131 ++++++++++++++++-- .../downloaders/live-stream/extractors.go | 88 +++++++++++- internal/downloaders/live-stream/helpers.go | 36 ++++- 3 files changed, 234 insertions(+), 21 deletions(-) diff --git a/internal/downloaders/live-stream/download.go b/internal/downloaders/live-stream/download.go index 95c3d21..5a4a160 100644 --- a/internal/downloaders/live-stream/download.go +++ b/internal/downloaders/live-stream/download.go @@ -5,6 +5,7 @@ import ( "os" "os/exec" "path/filepath" + "strings" "sync" "sync/atomic" @@ -17,19 +18,31 @@ func (d *M3U8Downloader) Download(job *utils.DanzoJob) error { if err := os.MkdirAll(tempDir, 0755); err != nil { return fmt.Errorf("error creating temp directory: %v", err) } - defer os.RemoveAll(tempDir) + var downloadErr error + defer func() { + if downloadErr == nil { + os.RemoveAll(tempDir) + } else { + log.Warn().Str("op", "live-stream/download").Msgf("Preserving segments in %s due to error", tempDir) + } + }() + client := utils.NewDanzoHTTPClient(job.HTTPClientConfig) log.Debug().Str("op", "live-stream/download").Msgf("Fetching manifest from %s", job.URL) manifestContent, err := getM3U8Contents(job.URL, client) if err != nil { - return fmt.Errorf("error fetching manifest: %v", err) + downloadErr = fmt.Errorf("error fetching manifest: %v", err) + return downloadErr } - segmentURLs, err := processM3U8Content(manifestContent, job.URL, client) + m3u8Info, err := parseM3U8Content(manifestContent, job.URL, client) if err != nil { - return fmt.Errorf("error processing manifest: %v", err) + downloadErr = fmt.Errorf("error processing manifest: %v", err) + return downloadErr } + segmentURLs := m3u8Info.SegmentURLs if len(segmentURLs) == 0 { - return fmt.Errorf("no segments found in manifest") + downloadErr = fmt.Errorf("no segments found in manifest") + return downloadErr } log.Info().Str("op", "live-stream/download").Msgf("Found %d segments to download", len(segmentURLs)) @@ -47,20 +60,43 @@ func (d *M3U8Downloader) Download(job *utils.DanzoJob) error { job.Metadata["segmentSizes"] = segmentSizes log.Debug().Str("op", "live-stream/download").Msgf("Total estimated size: %s", utils.FormatBytes(uint64(totalSize))) + // Detect fMP4 format + isFMP4 := detectFMP4Format(job.URL, segmentURLs) + if isFMP4 { + log.Debug().Str("op", "live-stream/download").Msg("Detected fMP4 format segments") + } + log.Info().Str("op", "live-stream/download").Msg("Starting parallel download of segments") - segmentFiles, err := downloadSegmentsParallel(segmentURLs, tempDir, job.Connections, client, job.ProgressFunc, totalSize) + segmentFiles, err := downloadSegmentsParallel(segmentURLs, tempDir, job.Connections, client, job.ProgressFunc, totalSize, isFMP4) if err != nil { - return fmt.Errorf("error downloading segments: %v", err) + downloadErr = fmt.Errorf("error downloading segments: %v", err) + return downloadErr } log.Info().Str("op", "live-stream/download").Msg("All segments downloaded, merging with ffmpeg") - if err := mergeSegments(segmentFiles, job.OutputPath); err != nil { - return fmt.Errorf("error merging segments: %v", err) + if err := mergeSegments(segmentFiles, job.OutputPath, isFMP4, m3u8Info.InitSegment, tempDir, client); err != nil { + downloadErr = fmt.Errorf("error merging segments: %v", err) + return downloadErr } log.Info().Str("op", "live-stream/download").Msg("Segments merged successfully") return nil } -func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers int, client *utils.DanzoHTTPClient, progressFunc func(int64, int64), totalSize int64) ([]string, error) { +func detectFMP4Format(manifestURL string, segmentURLs []string) bool { + if strings.Contains(manifestURL, "/fmp4/") || strings.Contains(manifestURL, "frag") { + return true + } + if len(segmentURLs) > 0 { + firstSegment := segmentURLs[0] + if strings.Contains(firstSegment, "/fmp4/") || + strings.Contains(firstSegment, ".m4s") || + strings.Contains(firstSegment, "frag") { + return true + } + } + return false +} + +func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers int, client *utils.DanzoHTTPClient, progressFunc func(int64, int64), totalSize int64, isFMP4 bool) ([]string, error) { var downloadedFiles []string var mu sync.Mutex var totalDownloaded int64 @@ -75,6 +111,10 @@ func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers } close(jobCh) downloadedFiles = make([]string, len(segmentURLs)) + ext := ".ts" + if isFMP4 { + ext = ".m4s" + } var wg sync.WaitGroup for range numWorkers { @@ -82,7 +122,7 @@ func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers go func() { defer wg.Done() for job := range jobCh { - outputPath := filepath.Join(outputDir, fmt.Sprintf("segment_%04d.ts", job.index)) + outputPath := filepath.Join(outputDir, fmt.Sprintf("segment_%04d%s", job.index, ext)) size, err := downloadSegment(job.url, outputPath, client) if err != nil { mu.Lock() @@ -110,7 +150,14 @@ func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers return downloadedFiles, nil } -func mergeSegments(segmentFiles []string, outputPath string) error { +func mergeSegments(segmentFiles []string, outputPath string, isFMP4 bool, initSegment string, tempDir string, client *utils.DanzoHTTPClient) error { + if isFMP4 { + return mergeFMP4Segments(segmentFiles, outputPath, initSegment, tempDir, client) + } + return mergeTSSegments(segmentFiles, outputPath) +} + +func mergeTSSegments(segmentFiles []string, outputPath string) error { tempListFile := filepath.Join(filepath.Dir(outputPath), ".segment_list.txt") f, err := os.Create(tempListFile) if err != nil { @@ -133,6 +180,66 @@ func mergeSegments(segmentFiles []string, outputPath string) error { log.Debug().Str("op", "live-stream/download").Msgf("Executing ffmpeg command: %s", cmd.String()) output, err := cmd.CombinedOutput() if err != nil { + log.Error().Str("op", "live-stream/download").Msgf("FFmpeg output:\n%s", string(output)) + return fmt.Errorf("ffmpeg error: %v\nOutput: %s", err, string(output)) + } + return nil +} + +func mergeFMP4Segments(segmentFiles []string, outputPath string, initSegment string, tempDir string, client *utils.DanzoHTTPClient) error { + tempConcatFile := filepath.Join(filepath.Dir(outputPath), ".concat_temp.m4s") + defer os.Remove(tempConcatFile) + log.Debug().Str("op", "live-stream/download").Msgf("Concatenating %d fMP4 segments", len(segmentFiles)) + outFile, err := os.Create(tempConcatFile) + if err != nil { + return fmt.Errorf("error creating temp concat file: %v", err) + } + if initSegment != "" { + log.Debug().Str("op", "live-stream/download").Msg("Downloading init segment") + initPath := filepath.Join(tempDir, "init.mp4") + _, err := downloadSegment(initSegment, initPath, client) + if err != nil { + outFile.Close() + return fmt.Errorf("error downloading init segment: %v", err) + } + initData, err := os.ReadFile(initPath) + if err != nil { + outFile.Close() + return fmt.Errorf("error reading init segment: %v", err) + } + if _, err := outFile.Write(initData); err != nil { + outFile.Close() + return fmt.Errorf("error writing init segment: %v", err) + } + log.Debug().Str("op", "live-stream/download").Msgf("Init segment written (%d bytes)", len(initData)) + } + for i, segmentFile := range segmentFiles { + data, err := os.ReadFile(segmentFile) + if err != nil { + outFile.Close() + return fmt.Errorf("error reading segment %d: %v", i, err) + } + if _, err := outFile.Write(data); err != nil { + outFile.Close() + return fmt.Errorf("error writing segment %d: %v", i, err) + } + } + outFile.Close() + + log.Debug().Str("op", "live-stream/download").Msg("Remuxing concatenated fMP4 segments") + cmd := exec.Command( + "ffmpeg", + "-i", tempConcatFile, + "-c", "copy", + "-movflags", "+faststart", + "-y", + outputPath, + ) + log.Debug().Str("op", "live-stream/download").Msgf("Executing ffmpeg command: %s", cmd.String()) + output, err := cmd.CombinedOutput() + if err != nil { + log.Error().Str("op", "live-stream/download").Msgf("FFmpeg failed with error: %v", err) + log.Error().Str("op", "live-stream/download").Msgf("FFmpeg output:\n%s", string(output)) return fmt.Errorf("ffmpeg error: %v\nOutput: %s", err, string(output)) } return nil diff --git a/internal/downloaders/live-stream/extractors.go b/internal/downloaders/live-stream/extractors.go index 6fa6b27..2184f6b 100644 --- a/internal/downloaders/live-stream/extractors.go +++ b/internal/downloaders/live-stream/extractors.go @@ -28,11 +28,21 @@ type RumbleJSResponse struct { } `json:"ua"` } +// JSON response from Dailymotion metadata endpoint +type DailymotionMetadata struct { + Qualities map[string][]struct { + Type string `json:"type"` + URL string `json:"url"` + } `json:"qualities"` +} + func runExtractor(job *utils.DanzoJob) error { extractor, _ := job.Metadata["extract"].(string) switch strings.ToLower(extractor) { case "rumble": return extractRumbleURL(job) + case "dailymotion": + return extractDailymotionURL(job) default: return fmt.Errorf("unsupported extractor: %s", extractor) } @@ -63,7 +73,6 @@ func getRumbleVideoID(pageURL string) (string, error) { req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36") req.Header.Set("Connection", "keep-alive") req.Header.Set("Upgrade-Insecure-Requests", "1") - resp, err := client.Do(req) if err != nil { return "", fmt.Errorf("failed to fetch rumble page: %w", err) @@ -73,7 +82,6 @@ func getRumbleVideoID(pageURL string) (string, error) { if err != nil { return "", fmt.Errorf("failed to read rumble page body: %w", err) } - re := regexp.MustCompile(`"embedUrl":\s*"https://rumble\.com/embed/([^/"]+)/"`) // re := regexp.MustCompile(`https://rumble\.com/embed/([^&",/]*)`) matches := re.FindStringSubmatch(string(body)) @@ -90,7 +98,6 @@ func getRumbleM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfi maps.Copy(newClientConfig.Headers, clientConfig.Headers) newClientConfig.Headers["Referer"] = "https://rumble.com/" client := utils.NewDanzoHTTPClient(newClientConfig) - req, err := http.NewRequest("GET", jsonURL, nil) if err != nil { return "", fmt.Errorf("failed to create request for rumble json: %w", err) @@ -100,7 +107,6 @@ func getRumbleM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfi return "", fmt.Errorf("failed to fetch rumble json: %w", err) } defer resp.Body.Close() - var data RumbleJSResponse if err := json.NewDecoder(resp.Body).Decode(&data); err != nil { return "", fmt.Errorf("failed to decode rumble json: %w", err) @@ -113,3 +119,77 @@ func getRumbleM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfi } return "", fmt.Errorf("could not find m3u8 url in rumble json response") } + +func extractDailymotionURL(job *utils.DanzoJob) error { + log.Debug().Str("op", "live-stream/extractor").Msgf("Extracting Dailymotion URL from %s", job.URL) + videoID, err := getDailymotionVideoID(job.URL) + if err != nil { + return err + } + log.Debug().Str("op", "live-stream/extractor").Msgf("Found Dailymotion video ID: %s", videoID) + m3u8URL, err := getDailymotionM3U8FromVideoID(videoID, job.HTTPClientConfig) + if err != nil { + return err + } + job.URL = m3u8URL + return nil +} + +func getDailymotionVideoID(pageURL string) (string, error) { + re := regexp.MustCompile(`dai\.ly/([^/?&#]+)`) // dai.ly/{id} + if matches := re.FindStringSubmatch(pageURL); len(matches) >= 2 { + return matches[1], nil + } + re = regexp.MustCompile(`dailymotion\.[a-z]{2,3}/video/([^/?&#]+)`) // dailymotion.com/video/{id} + if matches := re.FindStringSubmatch(pageURL); len(matches) >= 2 { + return matches[1], nil + } + re = regexp.MustCompile(`[?&]video=([^&#]+)`) // player.html?video={id} + if matches := re.FindStringSubmatch(pageURL); len(matches) >= 2 { + return matches[1], nil + } + return "", fmt.Errorf("could not extract Dailymotion video ID from URL: %s", pageURL) +} + +func getDailymotionM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfig) (string, error) { + metadataURL := fmt.Sprintf("https://www.dailymotion.com/player/metadata/video/%s", videoID) + newClientConfig := clientConfig + newClientConfig.Headers = make(map[string]string) + maps.Copy(newClientConfig.Headers, clientConfig.Headers) + newClientConfig.Headers["Referer"] = "https://www.dailymotion.com/" + newClientConfig.Headers["Origin"] = "https://www.dailymotion.com" + client := utils.NewDanzoHTTPClient(newClientConfig) + req, err := http.NewRequest("GET", metadataURL+"?app=com.dailymotion.neon", nil) + if err != nil { + return "", fmt.Errorf("failed to create request for dailymotion metadata: %w", err) + } + resp, err := client.Do(req) + if err != nil { + return "", fmt.Errorf("failed to fetch dailymotion metadata: %w", err) + } + defer resp.Body.Close() + var metadata DailymotionMetadata + if err := json.NewDecoder(resp.Body).Decode(&metadata); err != nil { + return "", fmt.Errorf("failed to decode dailymotion metadata: %w", err) + } + qualityPriority := []string{"auto", "1080", "720", "480", "380", "240"} + for _, quality := range qualityPriority { + if mediaList, ok := metadata.Qualities[quality]; ok { + for _, media := range mediaList { + if media.Type == "application/x-mpegURL" && media.URL != "" { + log.Debug().Str("op", "live-stream/extractor").Msgf("Found m3u8 URL at quality %s", quality) + return media.URL, nil + } + } + } + } + for quality, mediaList := range metadata.Qualities { + for _, media := range mediaList { + if media.Type == "application/x-mpegURL" && media.URL != "" { + log.Debug().Str("op", "live-stream/extractor").Msgf("Found m3u8 URL at quality %s", quality) + return media.URL, nil + } + } + } + return "", fmt.Errorf("could not find m3u8 URL in dailymotion metadata response") +} diff --git a/internal/downloaders/live-stream/helpers.go b/internal/downloaders/live-stream/helpers.go index 93a507d..99a5026 100644 --- a/internal/downloaders/live-stream/helpers.go +++ b/internal/downloaders/live-stream/helpers.go @@ -14,6 +14,11 @@ import ( "github.com/tanq16/danzo/internal/utils" ) +type M3U8Info struct { + SegmentURLs []string + InitSegment string +} + func getM3U8Contents(manifestURL string, client *utils.DanzoHTTPClient) (string, error) { req, err := http.NewRequest("GET", manifestURL, nil) if err != nil { @@ -35,7 +40,7 @@ func getM3U8Contents(manifestURL string, client *utils.DanzoHTTPClient) (string, return string(content), nil } -func processM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClient) ([]string, error) { +func parseM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClient) (*M3U8Info, error) { baseURL, err := url.Parse(manifestURL) if err != nil { return nil, fmt.Errorf("error parsing manifest URL: %v", err) @@ -44,10 +49,28 @@ func processM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClie var segmentURLs []string var masterPlaylistURLs []string var isMasterPlaylist bool - + var initSegment string for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) - if line == "" || (strings.HasPrefix(line, "#") && !strings.Contains(line, "#EXT-X-STREAM-INF")) { + if line == "" { + continue + } + // Check for init segment (fMP4) + if strings.HasPrefix(line, "#EXT-X-MAP:") { + if idx := strings.Index(line, `URI="`); idx != -1 { + uriStart := idx + 5 + if uriEnd := strings.Index(line[uriStart:], `"`); uriEnd != -1 { + uri := line[uriStart : uriStart+uriEnd] + initSegment, err = resolveURL(baseURL, uri) + if err != nil { + return nil, fmt.Errorf("error resolving init segment URL: %v", err) + } + log.Debug().Str("op", "live-stream/helpers").Msgf("Found init segment: %s", initSegment) + } + } + continue + } + if strings.HasPrefix(line, "#") && !strings.Contains(line, "#EXT-X-STREAM-INF") { continue } if strings.Contains(line, "#EXT-X-STREAM-INF") { @@ -77,9 +100,12 @@ func processM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClie if err != nil { return nil, fmt.Errorf("error fetching sub-playlist: %v", err) } - return processM3U8Content(subContent, masterPlaylistURLs[0], client) + return parseM3U8Content(subContent, masterPlaylistURLs[0], client) } - return segmentURLs, nil + return &M3U8Info{ + SegmentURLs: segmentURLs, + InitSegment: initSegment, + }, nil } func resolveURL(baseURL *url.URL, urlStr string) (string, error) {