Sourcery Starbot ⭐ refactored SoulMelody/hls-get

hls_get/cli.py

@@ -13,7 +13,7 @@ async def download(links, path, names, coros, headers, timeout, clean_up, verify
     headers_parser = HeadersParser()
     header_lines = [b'', *(line.encode('latin-1') for line in headers), b'']
     parsed_headers, raw_headers = headers_parser.parse_headers(header_lines)
-    kwargs = dict()
+    kwargs = {}
     if not verify:
         kwargs['connector'] = TCPConnector(verify_ssl=False)
     for link, name in zip_longest(links, names):

hls_get/downloader.py

@@ -32,7 +32,7 @@ def __init__(self, link, path, name, coros, timeout, clean_up=True, **kwargs):
         self.sem = asyncio.Semaphore(coros)
         self.path = path
         self.timeout = timeout or None
-        self.key_cache = dict()
+        self.key_cache = {}
         self.clean_up = clean_up
 
     async def __aenter__(self):
@@ -45,11 +45,10 @@ async def __aexit__(self, *args):
     @cached_property
     def retry_wrapper(cls):
         ctx = click.get_current_context()
-        wrapper = tenacity.retry(
+        return tenacity.retry(
             wait=tenacity.wait_fixed(ctx.params['delay']),
-            stop=tenacity.stop_after_attempt(ctx.params['retry_times'])
+            stop=tenacity.stop_after_attempt(ctx.params['retry_times']),
         )
-        return wrapper
 
     @cached_property
     def cache_dir(self):
@@ -66,12 +65,9 @@ def on_success(self):
     @retry_with_options
     async def fetch_with_retry(self, link, text=False):
         with async_timeout.timeout(self.timeout):
-            async with self.sem, self.session.get(link) as resp:
+            async with (self.sem, self.session.get(link) as resp):
                 resp.raise_for_status()
-                if text:
-                    return await resp.text()
-                else:
-                    return await resp.read()
+                return await resp.text() if text else await resp.read()
 
     async def download_segment(self, seq_num, segment, bar):
         filename = f'{self.cache_dir}/{seq_num}.ts'
@@ -90,45 +86,45 @@ async def download_segment(self, seq_num, segment, bar):
 
     async def download(self, link):
         m3u8_obj = m3u8.loads(await self.fetch_with_retry(link, text=True), uri=link)
-        if not m3u8_obj.media_sequence:
-            if m3u8_obj.is_variant:
-                for i, playlist in enumerate(m3u8_obj.playlists):
-                    click.echo(
-                        f'{i}: bandwidth={playlist.stream_info.bandwidth} '
-                        f'resolution={playlist.stream_info.resolution} '
-                        f'codecs={playlist.stream_info.codecs} '
-                    )
-                index = click.prompt(
-                    'Which playlist to download?',
-                    type=click.Choice(list(range(len(m3u8_obj.playlists)))),
-                    value_proc=int,
-                    default=0
+        if m3u8_obj.media_sequence:
+            click.echo('Live streaming media is not suppported!')
+
+        elif m3u8_obj.is_variant:
+            for i, playlist in enumerate(m3u8_obj.playlists):
+                click.echo(
+                    f'{i}: bandwidth={playlist.stream_info.bandwidth} '
+                    f'resolution={playlist.stream_info.resolution} '
+                    f'codecs={playlist.stream_info.codecs} '
                 )
-                return await self.download(m3u8_obj.playlists[index].absolute_uri)
-            else:
-                tmp_list = m3u8.M3U8()
-                tmp_list.version = '3'
-                tmp_list.media_sequence = '0'
-                tmp_list.target_duration = m3u8_obj.target_duration
-                tmp_list.is_endlist = True
-                tasks = []
-                os.makedirs(self.cache_dir, exist_ok=True)
-                bar = ShadyBar(self.name, max=len(m3u8_obj.segments), suffix='%(percent).1f%% - %(eta_td)s')
-                for i, segment in enumerate(m3u8_obj.segments):
-                    tmp_list.add_segment(
-                        m3u8.Segment(
-                            f'{os.path.realpath(self.cache_dir)}/{i}.ts',
-                            duration=segment.duration,
-                            base_uri='file://'
-                        )
+            index = click.prompt(
+                'Which playlist to download?',
+                type=click.Choice(list(range(len(m3u8_obj.playlists)))),
+                value_proc=int,
+                default=0
+            )
+            return await self.download(m3u8_obj.playlists[index].absolute_uri)
+        else:
+            tmp_list = m3u8.M3U8()
+            tmp_list.version = '3'
+            tmp_list.media_sequence = '0'
+            tmp_list.target_duration = m3u8_obj.target_duration
+            tmp_list.is_endlist = True
+            tasks = []
+            os.makedirs(self.cache_dir, exist_ok=True)
+            bar = ShadyBar(self.name, max=len(m3u8_obj.segments), suffix='%(percent).1f%% - %(eta_td)s')
+            for i, segment in enumerate(m3u8_obj.segments):
+                tmp_list.add_segment(
+                    m3u8.Segment(
+                        f'{os.path.realpath(self.cache_dir)}/{i}.ts',
+                        duration=segment.duration,
+                        base_uri='file://'
                     )
-                    tasks.append(
-                        asyncio.ensure_future(
-                            self.download_segment(i, segment, bar)
-                        )
+                )
+                tasks.append(
+                    asyncio.ensure_future(
+                        self.download_segment(i, segment, bar)
                     )
-                tmp_list.dump(f'{self.cache_dir}/filelist.m3u8')
-                await asyncio.gather(*tasks)
-                bar.finish()
-        else:
-            click.echo('Live streaming media is not suppported!')
+                )
+            tmp_list.dump(f'{self.cache_dir}/filelist.m3u8')
+            await asyncio.gather(*tasks)
+            bar.finish()