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