Clean code and fix #121

.github/workflows/build_docker_images.yml

@@ -72,4 +72,5 @@ jobs:
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}
           cache-from: type=registry,ref=ghcr.io/dmunozv04/isponsorblocktv:buildcache
-          cache-to: type=registry,ref=ghcr.io/dmunozv04/isponsorblocktv:buildcache,mode=max
+          # Only cache if it's not a pull request
+          cache-to: ${{ github.event_name != 'pull_request' && 'type=registry,ref=ghcr.io/dmunozv04/isponsorblocktv:buildcache,mode=max' || '' }}

src/iSponsorBlockTV/__main__.py

@@ -1,8 +1,9 @@
 from . import helpers
 
+
 def main():
     helpers.app_start()
 
 
 if __name__ == "__main__":
-    main()
+    main()

src/iSponsorBlockTV/api_helpers.py

@@ -6,11 +6,11 @@
 import html
 
 
-def listToTuple(function):
+def list_to_tuple(function):
     def wrapper(*args):
-        args = [tuple(x) if type(x) == list else x for x in args]
+        args = [tuple(x) if x is list else x for x in args]
         result = function(*args)
-        result = tuple(result) if type(result) == list else result
+        result = tuple(result) if result is list else result
         return result
 
     return wrapper
@@ -80,26 +80,26 @@ async def search_channels(self, channel):
             return channels
 
         for i in data["items"]:
-            # Get channel subcription number
+            # Get channel subscription number
             params = {"id": i["snippet"]["channelId"], "key": self.apikey, "part": "statistics"}
             url = constants.Youtube_api + "channels"
             async with self.web_session.get(url, params=params) as resp:
-                channelData = await resp.json()
+                channel_data = await resp.json()
 
-            if channelData["items"][0]["statistics"]["hiddenSubscriberCount"]:
-                subCount = "Hidden"
+            if channel_data["items"][0]["statistics"]["hiddenSubscriberCount"]:
+                sub_count = "Hidden"
             else:
-                subCount = int(channelData["items"][0]["statistics"]["subscriberCount"])
-                subCount = format(subCount, "_")
+                sub_count = int(channel_data["items"][0]["statistics"]["subscriberCount"])
+                sub_count = format(sub_count, "_")
 
-            channels.append((i["snippet"]["channelId"], i["snippet"]["channelTitle"], subCount))
+            channels.append((i["snippet"]["channelId"], i["snippet"]["channelTitle"], sub_count))
         return channels
 
-    @listToTuple  # Convert list to tuple so it can be used as a key in the cache
+    @list_to_tuple  # Convert list to tuple so it can be used as a key in the cache
     @AsyncConditionalTTL(time_to_live=300, maxsize=10)  # 5 minutes for non-locked segments
     async def get_segments(self, vid_id):
         if await self.is_whitelisted(vid_id):
-            return ([], True)  # Return empty list and True to indicate that the cache should last forever
+            return [], True  # Return empty list and True to indicate that the cache should last forever
         vid_id_hashed = sha256(vid_id.encode("utf-8")).hexdigest()[
                         :4
                         ]  # Hashes video id and gets the first 4 characters
@@ -117,7 +117,7 @@ async def get_segments(self, vid_id):
             print(
                 f"Error getting segments for video {vid_id}, hashed as {vid_id_hashed}. "
                 f"Code: {response.status} - {response_text}")
-            return ([], True)
+            return [], True
         for i in response_json:
             if str(i["videoID"]) == str(vid_id):
                 response_json = i
@@ -144,20 +144,20 @@ def process_segments(response):
                     segment_before_end = -10
                 if (
                         segment_dict["start"] - segment_before_end < 1
-                ):  # Less than 1 second appart, combine them and skip them together
+                ):  # Less than 1 second apart, combine them and skip them together
                     segment_dict["start"] = segment_before_start
                     segment_dict["UUID"].extend(segment_before_UUID)
                     segments.pop()
                 segments.append(segment_dict)
         except Exception:
             pass
-        return (segments, ignore_ttl)
+        return segments, ignore_ttl
 
-    async def mark_viewed_segments(self, UUID):
+    async def mark_viewed_segments(self, uuids):
         """Marks the segments as viewed in the SponsorBlock API, if skip_count_tracking is enabled.
         Lets the contributor know that someone skipped the segment (thanks)"""
         if self.skip_count_tracking:
-            for i in UUID:
+            for i in uuids:
                 url = constants.SponsorBlock_api + "viewedVideoSponsorTime/"
                 params = {"UUID": i}
                 await self.web_session.post(url, params=params)

src/iSponsorBlockTV/conditional_ttl_cache.py

@@ -1,3 +1,7 @@
+from cache.key import KEY
+from cache.lru import LRU
+import datetime
+
 """MIT License
 
 Copyright (c) 2020 Rajat Singh
@@ -21,10 +25,6 @@
 SOFTWARE."""
 '''Modified code from https://github.com/iamsinghrajat/async-cache'''
 
-from cache.key import KEY
-from cache.lru import LRU
-import datetime
-
 
 class AsyncConditionalTTL:
     class _TTL(LRU):

src/iSponsorBlockTV/config_setup.py

@@ -49,14 +49,12 @@ def main(config, debug: bool) -> None:
     if apikey:
         if input("API key already specified. Change it? (y/n) ") == "y":
             apikey = input("Enter your API key: ")
-            config["apikey"] = apikey
     else:
         if input("API key only needed for the channel whitelist function. Add it? (y/n) ") == "y":
             print(
                 "Get youtube apikey here: https://developers.google.com/youtube/registering_an_application"
             )
             apikey = input("Enter your API key: ")
-            config["apikey"] = apikey
     config.apikey = apikey
 
     skip_categories = config.skip_categories

src/iSponsorBlockTV/dial_client.py

@@ -46,12 +46,12 @@ def get_ip():
     try:
         # doesn't even have to be reachable
         s.connect(('10.254.254.254', 1))
-        IP = s.getsockname()[0]
+        ip = s.getsockname()[0]
     except Exception:
-        IP = '127.0.0.1'
+        ip = '127.0.0.1'
     finally:
         s.close()
-    return IP
+    return ip
 
 
 class Handler(ssdp.aio.SSDP):

src/iSponsorBlockTV/helpers.py

@@ -36,7 +36,7 @@ def __init__(self, data_dir):
 
         self.devices = []
         self.apikey = ""
-        self.skip_categories = []
+        self.skip_categories = []  # These are the categories on the config file
         self.channel_whitelist = []
         self.skip_count_tracking = True
         self.mute_ads = False
@@ -61,7 +61,7 @@ def validate(self):
         if not self.apikey and self.channel_whitelist:
             raise ValueError("No youtube API key found and channel whitelist is not empty")
         if not self.skip_categories:
-            self.categories = ["sponsor"]
+            self.skip_categories = ["sponsor"]
             print("No categories found, using default: sponsor")
 
     def __load(self):
@@ -109,7 +109,7 @@ def __eq__(self, other):
 
 
 def app_start():
-    #If env has a data dir use that, otherwise use the default
+    # If env has a data dir use that, otherwise use the default
     default_data_dir = os.getenv("iSPBTV_data_dir") or user_data_dir("iSponsorBlockTV", "dmunozv04")
     parser = argparse.ArgumentParser(description="iSponsorblockTV")
     parser.add_argument("--data-dir", "-d", default=default_data_dir, help="data directory")

src/iSponsorBlockTV/logging_helpers.py

@@ -1,16 +1,40 @@
 import logging
 
 from rich.logging import RichHandler
-from rich._log_render import LogRender
 from rich.text import Text
 from rich.style import Style
+
+'''
+Copyright (c) 2020 Will McGugan
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+Modified code from rich (https://github.com/textualize/rich)
+'''
+
+
 class LogHandler(RichHandler):
     def __init__(self, device_name, log_name_len, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.filter_strings = []
         self._log_render = LogRender(
-            device_name = device_name,
-            log_name_len = log_name_len,
+            device_name=device_name,
+            log_name_len=log_name_len,
             show_time=True,
             show_level=True,
             show_path=False,
@@ -20,34 +44,47 @@ def __init__(self, device_name, log_name_len, *args, **kwargs):
 
     def add_filter_string(self, s):
         self.filter_strings.append(s)
-    
+
     def _filter(self, s):
         for i in self.filter_strings:
             s = s.replace(i, "REDACTED")
         return s
-    
+
     def format(self, record):
         original = super().format(record)
         return self._filter(original)
 
 
-class LogRender(LogRender):
-    def __init__(self, device_name, log_name_len, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+class LogRender:
+    def __init__(self, device_name,
+                 log_name_len,
+                 show_time=True,
+                 show_level=False,
+                 show_path=True,
+                 time_format="[%x %X]",
+                 omit_repeated_times=True,
+                 level_width=8):
         self.filter_strings = []
         self.log_name_len = log_name_len
         self.device_name = device_name
-
+        self.show_time = show_time
+        self.show_level = show_level
+        self.show_path = show_path
+        self.time_format = time_format
+        self.omit_repeated_times = omit_repeated_times
+        self.level_width = level_width
+        self._last_time = None
+
     def __call__(
-        self,
-        console,
-        renderables,
-        log_time,
-        time_format = None,
-        level = "",
-        path = None,
-        line_no = None,
-        link_path = None,
+            self,
+            console,
+            renderables,
+            log_time,
+            time_format=None,
+            level="",
+            path=None,
+            line_no=None,
+            link_path=None,
     ):
         from rich.containers import Renderables
         from rich.table import Table
@@ -58,7 +95,7 @@ def __call__(
             output.add_column(style="log.time")
         if self.show_level:
             output.add_column(style="log.level", width=self.level_width)
-        output.add_column(width = self.log_name_len, style=Style(color="yellow"), overflow="fold")
+        output.add_column(width=self.log_name_len, style=Style(color="yellow"), overflow="fold")
         output.add_column(ratio=1, style="log.message", overflow="fold")
         if self.show_path and path:
             output.add_column(style="log.path")
@@ -100,14 +137,16 @@ class LogFormatter(logging.Formatter):
     def __init__(self, fmt=None, datefmt=None, style="%", validate=True, filter_strings=None):
         super().__init__(fmt, datefmt, style, validate)
         self.filter_strings = filter_strings or []
+
     def add_filter_string(self, s):
         self.filter_strings.append(s)
-    
+
     def _filter(self, s):
         print(s)
         for i in self.filter_strings:
             s = s.replace(i, "REDACTED")
         return s
+
     def format(self, record):
         original = logging.Formatter.format(self, record)
         return self._filter(original)

src/iSponsorBlockTV/macos_install.py

@@ -40,14 +40,15 @@ def main():
         create_plist(correct_path)
         run_setup(correct_path + "/config.json")
         print(
-            "Launch daemon installed. Please restart the computer to enable it or use:\n launchctl load ~/Library/LaunchAgents/com.dmunozv04.iSponsorBlockTV.plist"
+            'Launch daemon installed. Please restart the computer to enable it or use:\n launchctl load '
+            '~/Library/LaunchAgents/com.dmunozv04.iSponsorBlockTV.plist'
         )
     else:
         if not os.path.exists(correct_path):
             os.makedirs(correct_path)
         print(
             "Please move the program to the correct path: "
             + correct_path
-            + "opeing now on finder..."
+            + "opening now on finder..."
         )
         os.system("open -R " + correct_path)

src/iSponsorBlockTV/main.py

@@ -3,14 +3,15 @@
 import time
 import logging
 import rich
+from typing import Optional
 from signal import signal, SIGINT, SIGTERM
 
 from . import api_helpers, ytlounge, logging_helpers
 
 
 class DeviceListener:
     def __init__(self, api_helper, config, device, log_name_len, debug: bool):
-        self.task: asyncio.Task = None
+        self.task: Optional[asyncio.Task] = None
         self.api_helper = api_helper
         self.offset = device.offset
         self.name = device.name
@@ -52,7 +53,6 @@ async def loop(self):
                 self.logger.debug("Refreshing auth")
                 await lounge_controller.refresh_auth()
             except:
-                # traceback.print_exc()
                 await asyncio.sleep(10)
         while not self.cancelled:
             while not (await self.is_available()) and not self.cancelled:
@@ -68,7 +68,7 @@ async def loop(self):
                     await lounge_controller.connect()
                 except:
                     pass
-            self.logger.info(f"Connected to device {lounge_controller.screen_name} ({self.name})")
+            self.logger.info("Connected to device %s (%s)", lounge_controller.screen_name, self.name)
             try:
                 # print("Subscribing to lounge")
                 sub = await lounge_controller.subscribe_monitored(self)
@@ -115,11 +115,11 @@ async def time_to_segment(self, segments, position, time_start):
             await self.skip(time_to_next, next_segment["end"], next_segment["UUID"])
 
     # Skips to the next segment (waits for the time to pass)
-    async def skip(self, time_to, position, UUID):
+    async def skip(self, time_to, position, uuids):
         await asyncio.sleep(time_to)
-        self.logger.info(f"Skipping segment: seeking to {position}")
-        asyncio.create_task(self.api_helper.mark_viewed_segments(UUID))
-        asyncio.create_task(self.lounge_controller.seek_to(position))
+        self.logger.info("Skipping segment: seeking to %s", position)
+        await asyncio.create_task(self.api_helper.mark_viewed_segments(uuids))
+        await asyncio.create_task(self.lounge_controller.seek_to(position))
 
     # Stops the connection to the device
     async def cancel(self):