Repository: yt-dlp/yt-dlp
File: yt_dlp/downloader/niconico.py (Python)

Code before:
import json
import threading
import time

from .common import FileDownloader
from .external import FFmpegFD
from ..networking import Request
from ..utils import DownloadError, str_or_none, truncate_string
from ..utils.traversal import traverse_obj


class NiconicoLiveFD(FileDownloader):
    """ Downloads niconico live without being stopped """

    def real_download(self, filename, info_dict):
        video_id = info_dict['id']
        quality, ws_extractor, ws_url = map(
            info_dict['downloader_options'].get, ('max_quality', 'ws', 'ws_url'))
        dl = FFmpegFD(self.ydl, self.params or {})

        new_info_dict = info_dict.copy()
        new_info_dict['protocol'] = 'm3u8'

        def communicate_ws(reconnect):
            if reconnect:
                ws = self.ydl.urlopen(Request(
                    ws_url, headers={'Origin': 'https://live.nicovideo.jp'}))
                if self.ydl.params.get('verbose', False):
                    self.write_debug('Sending startWatching request')
                ws.send(json.dumps({
                    'data': {
                        'reconnect': True,
                        'room': {
                            'commentable': True,
                            'protocol': 'webSocket',
                        },
                        'stream': {
                            'accessRightMethod': 'single_cookie',
                            'chasePlay': False,
                            'latency': 'high',
                            'protocol': 'hls',
                            'quality': quality,
                        },

Review comment (refactor):
```suggestion
        opts = info_dict['downloader_options']
        quality, ws_extractor, ws_url = opts['max_quality'], opts['ws'], opts['ws_url']
```

Code after:
import json
import threading
import time

from .common import FileDownloader
from .external import FFmpegFD
from ..networking import Request
from ..networking.websocket import WebSocketResponse
from ..utils import DownloadError, str_or_none, truncate_string
from ..utils.traversal import traverse_obj


class NiconicoLiveFD(FileDownloader):
    """ Downloads niconico live without being stopped """

    def real_download(self, filename, info_dict):
        video_id = info_dict['id']
        opts = info_dict['downloader_options']
        quality, ws_extractor, ws_url = opts['max_quality'], opts['ws'], opts['ws_url']
        dl = FFmpegFD(self.ydl, self.params or {})

        new_info_dict = info_dict.copy()
        new_info_dict['protocol'] = 'm3u8'

        def communicate_ws(reconnect):
            # Support --load-info-json as if it is a reconnect attempt
            if reconnect or not isinstance(ws_extractor, WebSocketResponse):
                ws = self.ydl.urlopen(Request(
                    ws_url, headers={'Origin': 'https://live.nicovideo.jp'}))
                if self.ydl.params.get('verbose', False):
                    self.write_debug('Sending startWatching request')
                ws.send(json.dumps({
                    'data': {
                        'reconnect': True,
                        'room': {
                            'commentable': True,
                            'protocol': 'webSocket',
                        },
                        'stream': {
                            'accessRightMethod': 'single_cookie',
                            'chasePlay': False,
                            'latency': 'high',
                            'protocol': 'hls',