Skip to content

add dailymotion extractor; fix hls segment store #28

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Tanq16 merged 1 commit into from
Nov 14, 2025
Merged

add dailymotion extractor; fix hls segment store #28

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 119 additions & 12 deletions internal/downloaders/live-stream/download.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"os"
"os/exec"
"path/filepath"
"strings"
"sync"
"sync/atomic"

Expand All @@ -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))

Expand All @@ -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
Expand All @@ -75,14 +111,18 @@ 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 {
wg.Add(1)
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()
Expand Down Expand Up @@ -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 {
Expand All @@ -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
Expand Down
88 changes: 84 additions & 4 deletions internal/downloaders/live-stream/extractors.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
Expand All @@ -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))
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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")
}
36 changes: 31 additions & 5 deletions internal/downloaders/live-stream/helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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)
Expand All @@ -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") {
Expand Down Expand Up @@ -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) {
Expand Down