Skip to content

Commit 8a6726b

Browse files
Tanq16Tanq16
committed
add dailymotion extractor; fix hls segment store
1 parent 4705ce7 commit 8a6726b

3 files changed

Lines changed: 234 additions & 21 deletions

File tree

‎internal/downloaders/live-stream/download.go‎

Lines changed: 119 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"os"
66
"os/exec"
77
"path/filepath"
8+
"strings"
89
"sync"
910
"sync/atomic"
1011

@@ -17,19 +18,31 @@ func (d *M3U8Downloader) Download(job *utils.DanzoJob) error {
1718
if err := os.MkdirAll(tempDir, 0755); err != nil {
1819
return fmt.Errorf("error creating temp directory: %v", err)
1920
}
20-
defer os.RemoveAll(tempDir)
21+
var downloadErr error
22+
defer func() {
23+
if downloadErr == nil {
24+
os.RemoveAll(tempDir)
25+
} else {
26+
log.Warn().Str("op", "live-stream/download").Msgf("Preserving segments in %s due to error", tempDir)
27+
}
28+
}()
29+
2130
client := utils.NewDanzoHTTPClient(job.HTTPClientConfig)
2231
log.Debug().Str("op", "live-stream/download").Msgf("Fetching manifest from %s", job.URL)
2332
manifestContent, err := getM3U8Contents(job.URL, client)
2433
if err != nil {
25-
return fmt.Errorf("error fetching manifest: %v", err)
34+
downloadErr = fmt.Errorf("error fetching manifest: %v", err)
35+
return downloadErr
2636
}
27-
segmentURLs, err := processM3U8Content(manifestContent, job.URL, client)
37+
m3u8Info, err := parseM3U8Content(manifestContent, job.URL, client)
2838
if err != nil {
29-
return fmt.Errorf("error processing manifest: %v", err)
39+
downloadErr = fmt.Errorf("error processing manifest: %v", err)
40+
return downloadErr
3041
}
42+
segmentURLs := m3u8Info.SegmentURLs
3143
if len(segmentURLs) == 0 {
32-
return fmt.Errorf("no segments found in manifest")
44+
downloadErr = fmt.Errorf("no segments found in manifest")
45+
return downloadErr
3346
}
3447
log.Info().Str("op", "live-stream/download").Msgf("Found %d segments to download", len(segmentURLs))
3548

@@ -47,20 +60,43 @@ func (d *M3U8Downloader) Download(job *utils.DanzoJob) error {
4760
job.Metadata["segmentSizes"] = segmentSizes
4861
log.Debug().Str("op", "live-stream/download").Msgf("Total estimated size: %s", utils.FormatBytes(uint64(totalSize)))
4962

63+
// Detect fMP4 format
64+
isFMP4 := detectFMP4Format(job.URL, segmentURLs)
65+
if isFMP4 {
66+
log.Debug().Str("op", "live-stream/download").Msg("Detected fMP4 format segments")
67+
}
68+
5069
log.Info().Str("op", "live-stream/download").Msg("Starting parallel download of segments")
51-
segmentFiles, err := downloadSegmentsParallel(segmentURLs, tempDir, job.Connections, client, job.ProgressFunc, totalSize)
70+
segmentFiles, err := downloadSegmentsParallel(segmentURLs, tempDir, job.Connections, client, job.ProgressFunc, totalSize, isFMP4)
5271
if err != nil {
53-
return fmt.Errorf("error downloading segments: %v", err)
72+
downloadErr = fmt.Errorf("error downloading segments: %v", err)
73+
return downloadErr
5474
}
5575
log.Info().Str("op", "live-stream/download").Msg("All segments downloaded, merging with ffmpeg")
56-
if err := mergeSegments(segmentFiles, job.OutputPath); err != nil {
57-
return fmt.Errorf("error merging segments: %v", err)
76+
if err := mergeSegments(segmentFiles, job.OutputPath, isFMP4, m3u8Info.InitSegment, tempDir, client); err != nil {
77+
downloadErr = fmt.Errorf("error merging segments: %v", err)
78+
return downloadErr
5879
}
5980
log.Info().Str("op", "live-stream/download").Msg("Segments merged successfully")
6081
return nil
6182
}
6283

63-
func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers int, client *utils.DanzoHTTPClient, progressFunc func(int64, int64), totalSize int64) ([]string, error) {
84+
func detectFMP4Format(manifestURL string, segmentURLs []string) bool {
85+
if strings.Contains(manifestURL, "/fmp4/") || strings.Contains(manifestURL, "frag") {
86+
return true
87+
}
88+
if len(segmentURLs) > 0 {
89+
firstSegment := segmentURLs[0]
90+
if strings.Contains(firstSegment, "/fmp4/") ||
91+
strings.Contains(firstSegment, ".m4s") ||
92+
strings.Contains(firstSegment, "frag") {
93+
return true
94+
}
95+
}
96+
return false
97+
}
98+
99+
func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers int, client *utils.DanzoHTTPClient, progressFunc func(int64, int64), totalSize int64, isFMP4 bool) ([]string, error) {
64100
var downloadedFiles []string
65101
var mu sync.Mutex
66102
var totalDownloaded int64
@@ -75,14 +111,18 @@ func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers
75111
}
76112
close(jobCh)
77113
downloadedFiles = make([]string, len(segmentURLs))
114+
ext := ".ts"
115+
if isFMP4 {
116+
ext = ".m4s"
117+
}
78118

79119
var wg sync.WaitGroup
80120
for range numWorkers {
81121
wg.Add(1)
82122
go func() {
83123
defer wg.Done()
84124
for job := range jobCh {
85-
outputPath := filepath.Join(outputDir, fmt.Sprintf("segment_%04d.ts", job.index))
125+
outputPath := filepath.Join(outputDir, fmt.Sprintf("segment_%04d%s", job.index, ext))
86126
size, err := downloadSegment(job.url, outputPath, client)
87127
if err != nil {
88128
mu.Lock()
@@ -110,7 +150,14 @@ func downloadSegmentsParallel(segmentURLs []string, outputDir string, numWorkers
110150
return downloadedFiles, nil
111151
}
112152

113-
func mergeSegments(segmentFiles []string, outputPath string) error {
153+
func mergeSegments(segmentFiles []string, outputPath string, isFMP4 bool, initSegment string, tempDir string, client *utils.DanzoHTTPClient) error {
154+
if isFMP4 {
155+
return mergeFMP4Segments(segmentFiles, outputPath, initSegment, tempDir, client)
156+
}
157+
return mergeTSSegments(segmentFiles, outputPath)
158+
}
159+
160+
func mergeTSSegments(segmentFiles []string, outputPath string) error {
114161
tempListFile := filepath.Join(filepath.Dir(outputPath), ".segment_list.txt")
115162
f, err := os.Create(tempListFile)
116163
if err != nil {
@@ -133,6 +180,66 @@ func mergeSegments(segmentFiles []string, outputPath string) error {
133180
log.Debug().Str("op", "live-stream/download").Msgf("Executing ffmpeg command: %s", cmd.String())
134181
output, err := cmd.CombinedOutput()
135182
if err != nil {
183+
log.Error().Str("op", "live-stream/download").Msgf("FFmpeg output:\n%s", string(output))
184+
return fmt.Errorf("ffmpeg error: %v\nOutput: %s", err, string(output))
185+
}
186+
return nil
187+
}
188+
189+
func mergeFMP4Segments(segmentFiles []string, outputPath string, initSegment string, tempDir string, client *utils.DanzoHTTPClient) error {
190+
tempConcatFile := filepath.Join(filepath.Dir(outputPath), ".concat_temp.m4s")
191+
defer os.Remove(tempConcatFile)
192+
log.Debug().Str("op", "live-stream/download").Msgf("Concatenating %d fMP4 segments", len(segmentFiles))
193+
outFile, err := os.Create(tempConcatFile)
194+
if err != nil {
195+
return fmt.Errorf("error creating temp concat file: %v", err)
196+
}
197+
if initSegment != "" {
198+
log.Debug().Str("op", "live-stream/download").Msg("Downloading init segment")
199+
initPath := filepath.Join(tempDir, "init.mp4")
200+
_, err := downloadSegment(initSegment, initPath, client)
201+
if err != nil {
202+
outFile.Close()
203+
return fmt.Errorf("error downloading init segment: %v", err)
204+
}
205+
initData, err := os.ReadFile(initPath)
206+
if err != nil {
207+
outFile.Close()
208+
return fmt.Errorf("error reading init segment: %v", err)
209+
}
210+
if _, err := outFile.Write(initData); err != nil {
211+
outFile.Close()
212+
return fmt.Errorf("error writing init segment: %v", err)
213+
}
214+
log.Debug().Str("op", "live-stream/download").Msgf("Init segment written (%d bytes)", len(initData))
215+
}
216+
for i, segmentFile := range segmentFiles {
217+
data, err := os.ReadFile(segmentFile)
218+
if err != nil {
219+
outFile.Close()
220+
return fmt.Errorf("error reading segment %d: %v", i, err)
221+
}
222+
if _, err := outFile.Write(data); err != nil {
223+
outFile.Close()
224+
return fmt.Errorf("error writing segment %d: %v", i, err)
225+
}
226+
}
227+
outFile.Close()
228+
229+
log.Debug().Str("op", "live-stream/download").Msg("Remuxing concatenated fMP4 segments")
230+
cmd := exec.Command(
231+
"ffmpeg",
232+
"-i", tempConcatFile,
233+
"-c", "copy",
234+
"-movflags", "+faststart",
235+
"-y",
236+
outputPath,
237+
)
238+
log.Debug().Str("op", "live-stream/download").Msgf("Executing ffmpeg command: %s", cmd.String())
239+
output, err := cmd.CombinedOutput()
240+
if err != nil {
241+
log.Error().Str("op", "live-stream/download").Msgf("FFmpeg failed with error: %v", err)
242+
log.Error().Str("op", "live-stream/download").Msgf("FFmpeg output:\n%s", string(output))
136243
return fmt.Errorf("ffmpeg error: %v\nOutput: %s", err, string(output))
137244
}
138245
return nil

‎internal/downloaders/live-stream/extractors.go‎

Lines changed: 84 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -28,11 +28,21 @@ type RumbleJSResponse struct {
2828
} `json:"ua"`
2929
}
3030

31+
// JSON response from Dailymotion metadata endpoint
32+
type DailymotionMetadata struct {
33+
Qualities map[string][]struct {
34+
Type string `json:"type"`
35+
URL string `json:"url"`
36+
} `json:"qualities"`
37+
}
38+
3139
func runExtractor(job *utils.DanzoJob) error {
3240
extractor, _ := job.Metadata["extract"].(string)
3341
switch strings.ToLower(extractor) {
3442
case "rumble":
3543
return extractRumbleURL(job)
44+
case "dailymotion":
45+
return extractDailymotionURL(job)
3646
default:
3747
return fmt.Errorf("unsupported extractor: %s", extractor)
3848
}
@@ -63,7 +73,6 @@ func getRumbleVideoID(pageURL string) (string, error) {
6373
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")
6474
req.Header.Set("Connection", "keep-alive")
6575
req.Header.Set("Upgrade-Insecure-Requests", "1")
66-
6776
resp, err := client.Do(req)
6877
if err != nil {
6978
return "", fmt.Errorf("failed to fetch rumble page: %w", err)
@@ -73,7 +82,6 @@ func getRumbleVideoID(pageURL string) (string, error) {
7382
if err != nil {
7483
return "", fmt.Errorf("failed to read rumble page body: %w", err)
7584
}
76-
7785
re := regexp.MustCompile(`"embedUrl":\s*"https://rumble\.com/embed/([^/"]+)/"`)
7886
// re := regexp.MustCompile(`https://rumble\.com/embed/([^&",/]*)`)
7987
matches := re.FindStringSubmatch(string(body))
@@ -90,7 +98,6 @@ func getRumbleM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfi
9098
maps.Copy(newClientConfig.Headers, clientConfig.Headers)
9199
newClientConfig.Headers["Referer"] = "https://rumble.com/"
92100
client := utils.NewDanzoHTTPClient(newClientConfig)
93-
94101
req, err := http.NewRequest("GET", jsonURL, nil)
95102
if err != nil {
96103
return "", fmt.Errorf("failed to create request for rumble json: %w", err)
@@ -100,7 +107,6 @@ func getRumbleM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfi
100107
return "", fmt.Errorf("failed to fetch rumble json: %w", err)
101108
}
102109
defer resp.Body.Close()
103-
104110
var data RumbleJSResponse
105111
if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
106112
return "", fmt.Errorf("failed to decode rumble json: %w", err)
@@ -113,3 +119,77 @@ func getRumbleM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfi
113119
}
114120
return "", fmt.Errorf("could not find m3u8 url in rumble json response")
115121
}
122+
123+
func extractDailymotionURL(job *utils.DanzoJob) error {
124+
log.Debug().Str("op", "live-stream/extractor").Msgf("Extracting Dailymotion URL from %s", job.URL)
125+
videoID, err := getDailymotionVideoID(job.URL)
126+
if err != nil {
127+
return err
128+
}
129+
log.Debug().Str("op", "live-stream/extractor").Msgf("Found Dailymotion video ID: %s", videoID)
130+
m3u8URL, err := getDailymotionM3U8FromVideoID(videoID, job.HTTPClientConfig)
131+
if err != nil {
132+
return err
133+
}
134+
job.URL = m3u8URL
135+
return nil
136+
}
137+
138+
func getDailymotionVideoID(pageURL string) (string, error) {
139+
re := regexp.MustCompile(`dai\.ly/([^/?&#]+)`) // dai.ly/{id}
140+
if matches := re.FindStringSubmatch(pageURL); len(matches) >= 2 {
141+
return matches[1], nil
142+
}
143+
re = regexp.MustCompile(`dailymotion\.[a-z]{2,3}/video/([^/?&#]+)`) // dailymotion.com/video/{id}
144+
if matches := re.FindStringSubmatch(pageURL); len(matches) >= 2 {
145+
return matches[1], nil
146+
}
147+
re = regexp.MustCompile(`[?&]video=([^&#]+)`) // player.html?video={id}
148+
if matches := re.FindStringSubmatch(pageURL); len(matches) >= 2 {
149+
return matches[1], nil
150+
}
151+
return "", fmt.Errorf("could not extract Dailymotion video ID from URL: %s", pageURL)
152+
}
153+
154+
func getDailymotionM3U8FromVideoID(videoID string, clientConfig utils.HTTPClientConfig) (string, error) {
155+
metadataURL := fmt.Sprintf("https://www.dailymotion.com/player/metadata/video/%s", videoID)
156+
newClientConfig := clientConfig
157+
newClientConfig.Headers = make(map[string]string)
158+
maps.Copy(newClientConfig.Headers, clientConfig.Headers)
159+
newClientConfig.Headers["Referer"] = "https://www.dailymotion.com/"
160+
newClientConfig.Headers["Origin"] = "https://www.dailymotion.com"
161+
client := utils.NewDanzoHTTPClient(newClientConfig)
162+
req, err := http.NewRequest("GET", metadataURL+"?app=com.dailymotion.neon", nil)
163+
if err != nil {
164+
return "", fmt.Errorf("failed to create request for dailymotion metadata: %w", err)
165+
}
166+
resp, err := client.Do(req)
167+
if err != nil {
168+
return "", fmt.Errorf("failed to fetch dailymotion metadata: %w", err)
169+
}
170+
defer resp.Body.Close()
171+
var metadata DailymotionMetadata
172+
if err := json.NewDecoder(resp.Body).Decode(&metadata); err != nil {
173+
return "", fmt.Errorf("failed to decode dailymotion metadata: %w", err)
174+
}
175+
qualityPriority := []string{"auto", "1080", "720", "480", "380", "240"}
176+
for _, quality := range qualityPriority {
177+
if mediaList, ok := metadata.Qualities[quality]; ok {
178+
for _, media := range mediaList {
179+
if media.Type == "application/x-mpegURL" && media.URL != "" {
180+
log.Debug().Str("op", "live-stream/extractor").Msgf("Found m3u8 URL at quality %s", quality)
181+
return media.URL, nil
182+
}
183+
}
184+
}
185+
}
186+
for quality, mediaList := range metadata.Qualities {
187+
for _, media := range mediaList {
188+
if media.Type == "application/x-mpegURL" && media.URL != "" {
189+
log.Debug().Str("op", "live-stream/extractor").Msgf("Found m3u8 URL at quality %s", quality)
190+
return media.URL, nil
191+
}
192+
}
193+
}
194+
return "", fmt.Errorf("could not find m3u8 URL in dailymotion metadata response")
195+
}

‎internal/downloaders/live-stream/helpers.go‎

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,11 @@ import (
1414
"github.com/tanq16/danzo/internal/utils"
1515
)
1616

17+
type M3U8Info struct {
18+
SegmentURLs []string
19+
InitSegment string
20+
}
21+
1722
func getM3U8Contents(manifestURL string, client *utils.DanzoHTTPClient) (string, error) {
1823
req, err := http.NewRequest("GET", manifestURL, nil)
1924
if err != nil {
@@ -35,7 +40,7 @@ func getM3U8Contents(manifestURL string, client *utils.DanzoHTTPClient) (string,
3540
return string(content), nil
3641
}
3742

38-
func processM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClient) ([]string, error) {
43+
func parseM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClient) (*M3U8Info, error) {
3944
baseURL, err := url.Parse(manifestURL)
4045
if err != nil {
4146
return nil, fmt.Errorf("error parsing manifest URL: %v", err)
@@ -44,10 +49,28 @@ func processM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClie
4449
var segmentURLs []string
4550
var masterPlaylistURLs []string
4651
var isMasterPlaylist bool
47-
52+
var initSegment string
4853
for scanner.Scan() {
4954
line := strings.TrimSpace(scanner.Text())
50-
if line == "" || (strings.HasPrefix(line, "#") && !strings.Contains(line, "#EXT-X-STREAM-INF")) {
55+
if line == "" {
56+
continue
57+
}
58+
// Check for init segment (fMP4)
59+
if strings.HasPrefix(line, "#EXT-X-MAP:") {
60+
if idx := strings.Index(line, `URI="`); idx != -1 {
61+
uriStart := idx + 5
62+
if uriEnd := strings.Index(line[uriStart:], `"`); uriEnd != -1 {
63+
uri := line[uriStart : uriStart+uriEnd]
64+
initSegment, err = resolveURL(baseURL, uri)
65+
if err != nil {
66+
return nil, fmt.Errorf("error resolving init segment URL: %v", err)
67+
}
68+
log.Debug().Str("op", "live-stream/helpers").Msgf("Found init segment: %s", initSegment)
69+
}
70+
}
71+
continue
72+
}
73+
if strings.HasPrefix(line, "#") && !strings.Contains(line, "#EXT-X-STREAM-INF") {
5174
continue
5275
}
5376
if strings.Contains(line, "#EXT-X-STREAM-INF") {
@@ -77,9 +100,12 @@ func processM3U8Content(content, manifestURL string, client *utils.DanzoHTTPClie
77100
if err != nil {
78101
return nil, fmt.Errorf("error fetching sub-playlist: %v", err)
79102
}
80-
return processM3U8Content(subContent, masterPlaylistURLs[0], client)
103+
return parseM3U8Content(subContent, masterPlaylistURLs[0], client)
81104
}
82-
return segmentURLs, nil
105+
return &M3U8Info{
106+
SegmentURLs: segmentURLs,
107+
InitSegment: initSegment,
108+
}, nil
83109
}
84110

85111
func resolveURL(baseURL *url.URL, urlStr string) (string, error) {

0 commit comments

Comments
 (0)