yt_dlp/downloader/niconico.py

   1 import json
   2 import threading
   3 import time
   4
   5 from . import get_suitable_downloader
   6 from .common import FileDownloader
   7 from .external import FFmpegFD
   8 from ..networking import Request
   9 from ..utils import DownloadError, str_or_none, try_get
  10
  11
  12 class NiconicoDmcFD(FileDownloader):
  13     """ Downloading niconico douga from DMC with heartbeat """
  14
  15     def real_download(self, filename, info_dict):
  16         from ..extractor.niconico import NiconicoIE
  17
  18         self.to_screen(f'[{self.FD_NAME}] Downloading from DMC')
  19         ie = NiconicoIE(self.ydl)
  20         info_dict, heartbeat_info_dict = ie._get_heartbeat_info(info_dict)
  21
  22         fd = get_suitable_downloader(info_dict, params=self.params)(self.ydl, self.params)
  23
  24         success = download_complete = False
  25         timer = [None]
  26         heartbeat_lock = threading.Lock()
  27         heartbeat_url = heartbeat_info_dict['url']
  28         heartbeat_data = heartbeat_info_dict['data'].encode()
  29         heartbeat_interval = heartbeat_info_dict.get('interval', 30)
  30
  31         request = Request(heartbeat_url, heartbeat_data)
  32
  33         def heartbeat():
  34             try:
  35                 self.ydl.urlopen(request).read()
  36             except Exception:
  37                 self.to_screen(f'[{self.FD_NAME}] Heartbeat failed')
  38
  39             with heartbeat_lock:
  40                 if not download_complete:
  41                     timer[0] = threading.Timer(heartbeat_interval, heartbeat)
  42                     timer[0].start()
  43
  44         heartbeat_info_dict['ping']()
  45         self.to_screen('[%s] Heartbeat with %d second interval ...' % (self.FD_NAME, heartbeat_interval))
  46         try:
  47             heartbeat()
  48             if type(fd).__name__ == 'HlsFD':
  49                 info_dict.update(ie._extract_m3u8_formats(info_dict['url'], info_dict['id'])[0])
  50             success = fd.real_download(filename, info_dict)
  51         finally:
  52             if heartbeat_lock:
  53                 with heartbeat_lock:
  54                     timer[0].cancel()
  55                     download_complete = True
  56         return success
  57
  58
  59 class NiconicoLiveFD(FileDownloader):
  60     """ Downloads niconico live without being stopped """
  61
  62     def real_download(self, filename, info_dict):
  63         video_id = info_dict['video_id']
  64         ws_url = info_dict['url']
  65         ws_extractor = info_dict['ws']
  66         ws_origin_host = info_dict['origin']
  67         live_quality = info_dict.get('live_quality', 'high')
  68         live_latency = info_dict.get('live_latency', 'high')
  69         dl = FFmpegFD(self.ydl, self.params or {})
  70
  71         new_info_dict = info_dict.copy()
  72         new_info_dict.update({
  73             'protocol': 'm3u8',
  74         })
  75
  76         def communicate_ws(reconnect):
  77             if reconnect:
  78                 ws = self.ydl.urlopen(Request(ws_url, headers={'Origin': f'https://{ws_origin_host}'}))
  79                 if self.ydl.params.get('verbose', False):
  80                     self.to_screen('[debug] Sending startWatching request')
  81                 ws.send(json.dumps({
  82                     'type': 'startWatching',
  83                     'data': {
  84                         'stream': {
  85                             'quality': live_quality,
  86                             'protocol': 'hls+fmp4',
  87                             'latency': live_latency,
  88                             'chasePlay': False,
  89                         },
  90                         'room': {
  91                             'protocol': 'webSocket',
  92                             'commentable': True,
  93                         },
  94                         'reconnect': True,
  95                     },
  96                 }))
  97             else:
  98                 ws = ws_extractor
  99             with ws:
 100                 while True:
 101                     recv = ws.recv()
 102                     if not recv:
 103                         continue
 104                     data = json.loads(recv)
 105                     if not data or not isinstance(data, dict):
 106                         continue
 107                     if data.get('type') == 'ping':
 108                         # pong back
 109                         ws.send(r'{"type":"pong"}')
 110                         ws.send(r'{"type":"keepSeat"}')
 111                     elif data.get('type') == 'disconnect':
 112                         self.write_debug(data)
 113                         return True
 114                     elif data.get('type') == 'error':
 115                         self.write_debug(data)
 116                         message = try_get(data, lambda x: x['body']['code'], str) or recv
 117                         return DownloadError(message)
 118                     elif self.ydl.params.get('verbose', False):
 119                         if len(recv) > 100:
 120                             recv = recv[:100] + '...'
 121                         self.to_screen(f'[debug] Server said: {recv}')
 122
 123         def ws_main():
 124             reconnect = False
 125             while True:
 126                 try:
 127                     ret = communicate_ws(reconnect)
 128                     if ret is True:
 129                         return
 130                 except BaseException as e:
 131                     self.to_screen('[{}] {}: Connection error occured, reconnecting after 10 seconds: {}'.format('niconico:live', video_id, str_or_none(e)))
 132                     time.sleep(10)
 133                     continue
 134                 finally:
 135                     reconnect = True
 136
 137         thread = threading.Thread(target=ws_main, daemon=True)
 138         thread.start()
 139
 140         return dl.download(filename, new_info_dict)