Title string

Channel string

Series string

Description string

FileSize int64

Extension string `json:"ext"`

SponsorBlockChapters []interface{} `json:"sponsorblock_chapters"`

Title string

Artist string

Description string

FileSize int64

Extension string

DownloadURL string

SponsorBlock bool

Pct string

FileSize string

ETA string

Timeout time.Duration

WebRoot string

SponsorBlock bool

SponsorBlockCats string

OutPath string

RemoteAddr string

if ws.Timeout == 0 {

ws.Timeout = time.Duration(DefaultProcessTimeout)

}

ctx, cancel := context.WithCancel(context.Background())

defer cancel()

gconn, err := upgrader.Upgrade(w, r, nil)

if err != nil {

log.Println(err)

return

}

ws.RemoteAddr = gconn.RemoteAddr().String()

log.Printf("WS %s: Client connected\n", ws.RemoteAddr)

// wrap Gorilla conn with our conn so we can extend functionality

conn := Conn{sync.Mutex{}, gconn}

wg := sync.WaitGroup{}

wg.Add(1)

// setup ping/pong to keep connection open

go func(remoteAddr string) {

ticker := time.NewTicker(WSPingInterval)

defer ticker.Stop()

defer wg.Done()

for {

select {

case <-ctx.Done():

log.Printf("WS %s: ping, context done\n", remoteAddr)

return

case <-ticker.C:

// WriteControl can be called concurrently

if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(WSWriteWait)); err != nil {

log.Printf("WS %s: ping client, err %s\n", remoteAddr, err)

cancel()

return

}

}

}

}(ws.RemoteAddr)

outCh := make(chan Msg)

defer close(outCh)

errCh := make(chan error)

defer close(errCh)

// wait for goroutines to return before closing channels (defers are last in first out)

defer wg.Wait()

wg.Add(1)

go func() {

defer wg.Done()

for {

select {

case <-ctx.Done():

return

case m, open := <-outCh:

if !open {

return

}

err := conn.writeMsg(m)

if err != nil {

errCh <- err

}

case err := <-errCh:

if err != nil {

m := Msg{Key: "error", Value: err.Error()}

conn.writeMsg(m)

}

return

}

}

}()

// Send recently retrieved URLs

recentURLs, err := GetRecentURLs(ctx, ws.WebRoot, ws.OutPath, ws.Timeout)

if err != nil {

log.Printf("WS %s: GetRecentURLS err %s\n", ws.RemoteAddr, err)

return

}

m := Msg{Key: "recent", Value: recentURLs}

conn.writeMsg(m)

for {

msgType, raw, err := conn.ReadMessage()

if err != nil {

log.Printf("WS %s: ReadMessage err %s\n", ws.RemoteAddr, err)

return

}

log.Printf("WS %s: read message %s\n", ws.RemoteAddr, string(raw))

if msgType == websocket.TextMessage {

wg.Add(1)

go func() {

defer wg.Done()

var req Request

err = json.Unmarshal(raw, &req)

if err != nil {

log.Printf("json unmarshal error: %s\n", err)

return

}

err := ws.msgHandler(ctx, outCh, req)

if err != nil {

log.Printf("Error '%s'\n", err)

errCh <- err

}

}()

} else {

log.Printf("unknown message type - close websocket\n")

conn.Close()

return

}

log.Printf("WS %s: end main loop\n", ws.RemoteAddr)

}

c.Lock()

defer c.Unlock()

j, err := json.Marshal(val)

if err != nil {

return err

}

log.Printf("WS %s: write message %s\n", c.RemoteAddr(), string(j))

if err = c.WriteMessage(websocket.TextMessage, j); err != nil {

return err

}

return nil

if req.URL == "" && len(req.DeleteURLs) == 0 {

return fmt.Errorf("unknown parameters")

}

if len(req.DeleteURLs) > 0 {

err := DeleteFiles(req.DeleteURLs, ws.WebRoot)

if err != nil {

return err

}

} else if req.URL != "" {

url, err := url.Parse(req.URL)

if err != nil {

return err

}

if url.String() == "" {

return fmt.Errorf("url was empty")

}

restartCh := make(chan bool)

ctx1, cancel1 := context.WithCancel(ctx)

defer cancel1()

err = ws.ytDownload(ctx1, outCh, restartCh, url)

if err != nil {

return err

}

select {

case <-restartCh:

ctx2, cancel2 := context.WithCancel(ctx)

defer cancel2()

cancel1()

var restartCh2 chan bool

err = ws.ytDownload(ctx2, outCh, restartCh2, url)

if err != nil {

return err

}

default:

}

}

return nil

forceOpus := true

for _, h := range noReencodeSites {

if strings.Contains(url.Host, h) {

forceOpus = false

break

}

}

tCtx, cancel := context.WithTimeout(ctx, ws.Timeout)

defer cancel()

errCh := make(chan error)

defer close(errCh)

// filename is md5 sum of URL

urlSum := md5.Sum([]byte(url.String()))

diskFileNameTmp := filepath.Join(ws.WebRoot, ws.OutPath, "t", "ytdl-"+fmt.Sprintf("%x", urlSum))

log.Printf("Fetching url %s\n", url.String())

args := []string{

"--write-info-json",

"--max-filesize", fmt.Sprintf("%d", int(MaxFileSize)),

// output progress bar as newlines

"--newline",

"--progress-template", "%(progress.downloaded_bytes)s of %(progress.total_bytes)s / %(progress.total_bytes_estimate)s eta %(progress.eta)s",

// Do not use the Last-modified header to set the file modification time

"--no-mtime",

"--socket-timeout", fmt.Sprintf("%d", YtdlpSocketTimeoutSec),

"--no-playlist",

"-o", diskFileNameTmp + ".%(ext)s",

"--embed-metadata",

// extract audio

"-x",

// print final output filename (after postprocessing etc)

"--print-to-file", "after_move:filepath", diskFileNameTmp + ".ext",

// proto:dash is needed for fast Youtube downloads

// sort by size, bitrate in ascending order

"-S", "proto:dash,+size,+br",

// Faster youtube downloads: this combined with -S proto:dash ensures that we get dash https://github.com/yt-dlp/yt-dlp/issues/7417

"--extractor-args", "youtube:formats=duplicate",

}

if ws.SponsorBlock {

args = append(args, []string{

"--sponsorblock-remove", ws.SponsorBlockCats,

}...)

}

if forceOpus {

args = append(args, []string{

"--audio-format", "opus",

"--audio-quality", "32K",

// "--postprocessor-args", `ExtractAudio:-compression_level 0`, // fastest, lowest quality compression

}...)

}

args = append(args, url.String())

log.Printf("Running command %v\n", append([]string{YTCmd}, args...))

cmdOutCh, cmdErrCh, err := RunCommandCh(tCtx, YTCmd, args...)

if err != nil {

return err

}

var info Info

count := 0

for {

count++

if count > 20 {

return fmt.Errorf("waited too long for info file")

}

infoFileName := diskFileNameTmp + ".info.json"

time.Sleep(500 * time.Millisecond)

if _, err := os.Stat(infoFileName); os.IsNotExist(err) {

continue

}

raw, err := os.ReadFile(infoFileName)

if err != nil {

return fmt.Errorf("info file read error: %s", err)

}

var ytInfo YTInfo

err = json.Unmarshal(raw, &ytInfo)

if err != nil {

return fmt.Errorf("info file json unmarshal error: %s", err)

}

info.Title = ytInfo.Title

info.Artist = ytInfo.Channel

if info.Artist == "" {

info.Artist = ytInfo.Series

}

info.Description = ytInfo.Description

info.FileSize = ytInfo.FileSize

info.Extension = ytInfo.Extension

info.SponsorBlock = len(ytInfo.SponsorBlockChapters) > 0

if info.FileSize > MaxFileSize {

return fmt.Errorf("filesize %d too large", info.FileSize)

}

m := Msg{Key: "info", Value: info}

outCh <- m

break

}

// output size of opus file as it gets written

if forceOpus {

go getOpusFileSize(tCtx, info, outCh, errCh, diskFileNameTmp+".opus", ws.OutPath)

}

var startDownload time.Time

lastOut := time.Now()

var line string

var open bool

loop:

for {

select {

case err := <-errCh:

return err

case err, open := <-cmdErrCh:

if !open {

break loop

}

return err

case line, open = <-cmdOutCh:

if !open {

break loop

}

//fmt.Println(line)

p := getYTProgress(line)

if p != nil {

if startDownload.IsZero() {

startDownload = time.Now()

}

pct, err := strconv.ParseFloat(p.Pct, 64)

if err != nil {

return err

}

if restartCh != nil {

if time.Since(startDownload) > time.Second*5 && pct < 10 {

close(restartCh)

msg := "Restarting download...\n"

log.Print(msg)

m := Msg{Key: "unknown", Value: msg}

outCh <- m

return nil

}

}

if time.Since(lastOut) > 500*time.Millisecond {

m := Msg{Key: "progress", Value: *p}

outCh <- m

lastOut = time.Now()

}

} else {

m := Msg{Key: "unknown", Value: line}

outCh <- m

}

}

}

// if we got here, then command completed successfully

// yt-dlp writes it's final output filename to a temporary file. Read that back

diskFileNameTmp2b, err := os.ReadFile(diskFileNameTmp + ".ext")

if err != nil {

return err

}

diskFileNameTmp2 := strings.TrimSpace(string(diskFileNameTmp2b))

// read the first line

idx := bytes.Index(diskFileNameTmp2b, []byte{'\n'})

if idx > 0 {

diskFileNameTmp2 = string(diskFileNameTmp2b[:idx])

}

/*

log.Println("Fetching title from media file metadata")

ff, err := runFFprobe(ctx, FFprobeCmd, diskFileNameTmp2, ws.Timeout)

if err != nil {

return err

}

info.Title, info.Artist, info.Description = titleArtistDescription(ff)

if info.Title == "" {

return fmt.Errorf("unknown error (title was empty), last line: '%s'", line)

}

if info.Artist == "" {

info.Artist = "unknown"

}

*/

sanitizedTitle := filenameReplacer.Replace(info.Artist + "-" + info.Title)

sanitizedTitle = filenameRegexp.ReplaceAllString(sanitizedTitle, "")

sanitizedTitle = strings.Join(strings.Fields(sanitizedTitle), " ") // remove double spaces

finalFileNameNoExt := filepath.Join(ws.WebRoot, ws.OutPath, "ytdl-"+sanitizedTitle)

ext := path.Ext(diskFileNameTmp2)

// rename .opus to .oga. It's already an OGG container and most clients prefer .oga extension.

if ext == ".opus" {

ext = ".oga"

}

finalFileName := finalFileNameNoExt + ext

if forceOpus {

fi, err := os.Stat(diskFileNameTmp2)

if err != nil {

return err

}

m := Msg{

Key: "unknown",

Value: fmt.Sprintf("opus file size %.2f MB\n", float32(fi.Size())*1e-6),

}

outCh <- m

}

log.Printf("rename %s to %s", diskFileNameTmp2, finalFileName)

err = os.Rename(diskFileNameTmp2, finalFileName)

if err != nil {

return err

}

info.DownloadURL = filepath.Join(ws.OutPath, filepath.Base(finalFileName))

// don't send link for forceOpus as that's handled in getOpusFileSize goroutine

if !forceOpus {

m := Msg{Key: "link_stream", Value: info}

outCh <- m

}

m := Msg{Key: "completed", Value: "true"}

outCh <- m

// Send recently retrieved URLs

recentURLs, err := GetRecentURLs(ctx, ws.WebRoot, ws.OutPath, ws.Timeout)

if err != nil {

return fmt.Errorf("WS %s: GetRecentURLS err %w", ws.RemoteAddr, err)

}

m = Msg{Key: "recent", Value: recentURLs}

outCh <- m

return nil

matches := ytProgressRe.FindStringSubmatch(v)

var p *Progress

if len(matches) == 5 {

p = new(Progress)

downloaded, _ := strconv.Atoi(matches[1])

total := 0

// if total_bytes is missing, try total_bytes_estimate

if matches[2] != "NA" {

total, _ = strconv.Atoi(matches[2])

} else {

// for some reason we get decimal for the estimated bytes

totalf, _ := strconv.ParseFloat(matches[3], 64)

// don't care about loss of precision

total = int(totalf)

}

eta, _ := strconv.Atoi(matches[4])

pct := float64(downloaded) / float64(total) * 100.0

p.Pct = fmt.Sprintf("%.2f", pct)

p.FileSize = fmt.Sprintf("%.2f", float64(total)/(1024.0*1024.0))

p.ETA = fmt.Sprintf("%v", time.Duration(eta)*time.Second)

}

return p

var startTime time.Time

streamURLSent := false

for {

select {

case <-ctx.Done():

return

default:

}

opusFI, err := os.Stat(filename)

// abort on errors except for ErrNotExist

if err != nil {

if !errors.Is(err, fs.ErrNotExist) {

errCh <- fmt.Errorf("error getting stat on opus file '%s': %w", filename, err)

return

}

time.Sleep(time.Second)

continue

}

// wait until we have some data before sending stream URL

if !streamURLSent && opusFI.Size() > 10000 {

info.DownloadURL = filepath.Join(webPath, "stream", "t", filepath.Base(filename))

m := Msg{Key: "link_stream", Value: info}

outCh <- m

streamURLSent = true

}

if startTime.IsZero() {

startTime = time.Now()

}

if info.Extension == "mp3" {

mp3FI, err := os.Stat(strings.TrimSuffix(filename, filepath.Ext(filename)) + ".mp3")

if err == nil {

// Opus compression ratio from MP3 approximately 1:4

pctf := (float64(opusFI.Size()) / (float64(mp3FI.Size()) / 4)) * 100

pct := fmt.Sprintf("%.1f", pctf)

diff := time.Since(startTime)

etaStr := ""

if pctf > 0 {

eta := time.Duration((float64(diff) / pctf) * (100 - pctf)).Round(time.Second)

etaStr = eta.String()

}

m := Msg{

Key: "progress",

Value: Progress{

Pct: pct,

ETA: etaStr,

},

}

outCh <- m

}

}

m := Msg{

Key: "unknown",

Value: fmt.Sprintf("opus file size %.2f MB\n", float32(opusFI.Size())*1e-6),

}

outCh <- m

time.Sleep(3 * time.Second)

}

return func(w http.ResponseWriter, r *http.Request) {

dir := http.Dir(webRoot)

filename := strings.Replace(path.Clean(r.URL.Path), "stream/", "", 1)

f, err := dir.Open(filename)

if err != nil {

msg, code := toHTTPError(err)

http.Error(w, msg, code)

return

}

defer f.Close()

lastData := time.Now()

for {

// io.Copy doesn't return error on EOF

i, err := io.Copy(w, f)

if err != nil {

log.Printf("servestream copy err %s\n", err)

return

}

if i == 0 {

if time.Since(lastData) > time.Duration(StreamSourceTimeout) {

log.Printf("servestream timeout\n")

return

}

time.Sleep(time.Duration(1 * time.Second))

} else {

lastData = time.Now()

}

}

}

if errors.Is(err, fs.ErrNotExist) {

return "404 page not found", http.StatusNotFound

}

if errors.Is(err, fs.ErrPermission) {

return "403 Forbidden", http.StatusForbidden

}

// Default:

return "500 Internal Server Error", http.StatusInternalServerError