Youtube live proxy for 24/7 radio channels (audio only)

yt.py

# python3 yt.py <URL> [proxy player address]
# URL can be something like "https://www.youtube.com/watch?v=8XJ6r4U171I"
#   or "8XJ6r4U171I"
#   or even "https://www.youtube.com/freecodecamp/live"
# Proxy player can be started with `ffplay -rtsp_flags listen rtsp://0.0.0.0:12000/live.sdp`
#   omit proxy player address to play it locally by starting localhost proxy player.

from youtube_dl import YoutubeDL as YDL
from time import time as now, sleep
from urllib.request import urlopen

BUF_SEC = 30  # seconds to buffer

class Youtube():
  def __init__(self, url):
    self._url = url
    self._last_m3u8 = ''
  def _select_stream(self):
    with YDL({'quiet':True}) as ydl: #, 'format':'bestaudio/best'}) as ydl:
      info = ydl.extract_info(self._url, False)
      #return info['formats'][0]['url'] # worst
      max_abr = None
      abr = 0
      # find best abr & worst vbr
      for f in reversed(info['formats']):
        if f['abr'] >= abr:
          abr = f['abr']
          max_abr = f
        else: break
      return max_abr['url']
  def _skip(self, stream):
    if not self._last_m3u8:
      return self._gen(stream)
    from itertools import tee
    urls, _urls = tee(self._gen(stream))
    j=-1
    for i, (t, u) in enumerate(_urls):
      if u == self._last_m3u8:
        j=i+1
        break
    if j<0:
      print('couldn\'t find m3u8:', self._last_m3u8)
      pass
    else:
      print('previous stream found at ', j, 'th')
      for _ in range(j):
        next(urls)
      print(j, 'urls skipped')
    return urls
  def _gen(self, stream):
    for i in range(10):
      try:
        with urlopen(stream) as f:
          tlen = 0
          for l in f:
            if l.startswith(b'#EXTINF:'):
              tlen = float(l[8:-2])
            elif l.startswith(b'https://'):
              turl = l[:-1].decode()
              yield tlen, turl
        break
      except KeyboardInterrupt:
        break
      except:
        sleep(.1*(i+1))
        print('error opening stream:', stream, 'retrying..')
        continue

  def gen(self):
    while True:
      for t, u in self._skip(self._select_stream()):
        yield t, u

class Streamer():
  def __init__(self, addr, url):
    from sys import platform
    if platform == 'win32':
      _codec = '-c:a opus -strict -2'
    else:
      _codec = '-c:a libopus'
    self.is_local = addr[0] in ('localhost', '127.0.0.1')
    from subprocess import Popen, PIPE
    # ffmpeg -re option (realtime emulation) is avoided. Google server closes such slow reading connections.
    # aac re-encoding to avoid 'AAC with no global headers' error
    self._prc = Popen(('ffmpeg -hide_banner -i - -vn %s -b:a 52k -f rtsp -rtsp_transport tcp rtsp://%s:%d/live.sdp' % ('-c:a aac' if self.is_local else _codec, *addr)).split(), stdin=PIPE)
    self.stream_started, self.stream_loaded = 0, 0
    self.yt = Youtube(url)#'Vls4h1GAP-c'
  def __del__(self):
    self._prc.kill()
  def stream(self):
    try:
      #import os, sys
      #with os.fdopen(sys.stdout.fileno(), 'wb') as w:
      if not self.stream_started: self.stream_started = now()
      for t, u in self.yt.gen():
        #print(t, u)
        while True:
          try:
            b = urlopen(u).read()
            break
          except:
            print('error opening m3u8:', u, 'retrying...')
            sleep(.1)
            continue
        #run(('ffprobe -hide_banner -i -').split(), input=b)
        self._prc.stdin.write(b)
        self.stream_loaded += t # usually 5.005 seconds each.
        t = self.stream_started + self.stream_loaded 
        target = now() + BUF_SEC
        if t > target:
          #print('sleep', t-target)
          sleep(t - target)
      self.yt.last_url = u # last url saved for skipping already streamed parts
    except KeyboardInterrupt:
      pass
#from time import time
#from sched import scheduler
#def record(url, speed=1):
#  with open('test.ts', 'wb') as fw:
#    def play(url):
#      print(url)
#      with urlopen(url) as fr:
#        fw.write(fr.read())
#      #run(['ffmpeg', '-hide_banner', '-i', url, '-', str(name)+'.aac'])
#      
#    s = scheduler(time)
#    tlen = 0
#    for t, u in read_m3u8(url):
#      #play(u)
#      s.enter(tlen, 1, play, (l[:-1].decode(),))
#      tlen += t/speed
#    s.run()

  
#  is_serv = addr[0] == '0.0.0.0'
#  with open_socket() as w:
#    if is_serv:
#      w.bind(addr)
#      w.listen(1)
#      c, a = w.accept()
#    
#    from subprocess import Popen, PIPE
#    converter = Popen(('ffmpeg -hide_banner -i - -vn %s -f opus -b:a 52k -'%(_codec,)).split(), stdin=PIPE, stdout=PIPE)
#    
#    def out_stream(u):
#      converter.stdin.write(urlopen(u).read())
#      b = converter.stdout.read()
#      #b = strip_audio(urlopen(u).read())
#      for i in range(len(b)//1024):
#        if is_serv:
#          c.sendall(b[1024*i:1024*i+1024])
#        else:
#          w.sendto(b[1024*i:1024*i+1024], addr)
#      if len(b)%1024:
#        if is_serv:
#          c.sendall(b[1024*i:1024*i+1024])
#        else:
#          w.sendto(b[1024*(i+1):], addr)
#      #w.write(urlopen(u).read())
#      #w.flush()
#    s = scheduler(time)
#    tt = 0
#    cnt = 0
#    for t, u in read_m3u8(url):
#      s.enter(tt, 1, out_stream, (u,))
#      if cnt>1:
#        tt+=t
#      cnt+=1
#    s.run()

#def open_socket():
#  from socket import socket, AF_INET, SOCK_DGRAM
#  sock = socket(AF_INET, SOCK_DGRAM)
#  sock.settimeout(1)
#  return sock
#def strip_audio(b):
#  global _codec
#  from subprocess import run, PIPE
#  return run(('ffmpeg -hide_banner -i - -vn '+_codec+' -f opus -b:a 52k -').split(), input=b, stdout=PIPE).stdout

if __name__ == '__main__':
  from sys import argv
  addr = (argv[2] if len(argv)>2 else 'localhost', 12000)
  streamer = Streamer(addr, argv[1])
  if streamer.is_local:
    from subprocess import Popen
    listener=Popen('ffplay -rtsp_flags listen rtsp://localhost:12000/live.sdp'.split())
    
  try:
    streamer.stream()
  except KeyboardInterrupt:
    if streamer.is_local:
      listener.kill()