code cleanup add typing

main.py

@@ -1,9 +1,9 @@
 """application entry point"""
 
-from src.connect import Jellyfin, TubeArchivist, folder_check
+from src.connect import Jellyfin, TubeArchivist, env_check
 from src.series import Library
 
-folder_check()
+env_check()
 Jellyfin().ping()
 TubeArchivist().ping()
 

src/config.py

@@ -2,10 +2,12 @@
 
 import json
 
+from src.static_types import ConfigType
 
-def get_config():
+
+def get_config() -> ConfigType:
     """get connection config"""
     with open("config.json", "r", encoding="utf-8") as f:
-        config_content = json.loads(f.read())
+        config_content: ConfigType = json.loads(f.read())
 
     return config_content

src/connect.py

@@ -6,39 +6,42 @@ import os
 import requests
 
 from src.config import get_config
+from src.static_types import ConfigType, TAVideo
 
-CONFIG = get_config()
+CONFIG: ConfigType = get_config()
 
 
 class Jellyfin:
     """connect to jellyfin"""
 
-    headers = {"Authorization": "MediaBrowser Token=" + CONFIG["jf_token"]}
+    headers: dict = {
-    base = CONFIG["jf_url"]
+        "Authorization": "MediaBrowser Token=" + CONFIG["jf_token"]
+    }
+    base: str = CONFIG["jf_url"]
 
-    def get(self, path):
+    def get(self, path: str) -> dict:
         """make a get request"""
-        url = f"{self.base}/{path}"
+        url: str = f"{self.base}/{path}"
         response = requests.get(url, headers=self.headers, timeout=10)
         if response.ok:
             return response.json()
 
         print(response.text)
-        return False
+        return {}
 
-    def post(self, path, data):
+    def post(self, path: str, data: dict | bool) -> None:
         """make a post request"""
-        url = f"{self.base}/{path}"
+        url: str = f"{self.base}/{path}"
         response = requests.post(
             url, headers=self.headers, json=data, timeout=10
         )
         if not response.ok:
             print(response.text)
 
-    def post_img(self, path, thumb_base64):
+    def post_img(self, path: str, thumb_base64: bytes) -> None:
         """set image"""
-        url = f"{self.base}/{path}"
+        url: str = f"{self.base}/{path}"
-        new_headers = self.headers.copy()
+        new_headers: dict = self.headers.copy()
         new_headers.update({"Content-Type": "image/jpeg"})
         response = requests.post(
             url, headers=new_headers, data=thumb_base64, timeout=10
@@ -46,10 +49,9 @@ class Jellyfin:
         if not response.ok:
             print(response.text)
 
-    def ping(self):
+    def ping(self) -> None:
         """ping the server"""
-        path = "Users"
+        response = self.get("Users")
-        response = self.get(path)
         if not response:
             raise ConnectionError("failed to connect to jellyfin")
 
@@ -59,12 +61,13 @@ class Jellyfin:
 class TubeArchivist:
     """connect to Tube Archivist"""
 
-    headers = {"Authorization": "Token " + CONFIG.get("ta_token")}
+    ta_token: str = CONFIG["ta_token"]
-    base = CONFIG["ta_url"]
+    headers: dict = {"Authorization": f"Token {ta_token}"}
+    base: str = CONFIG["ta_url"]
 
-    def get(self, path):
+    def get(self, path: str) -> TAVideo:
         """get document from ta"""
-        url = f"{self.base}/api/{path}"
+        url: str = f"{self.base}/api/{path}"
         response = requests.get(url, headers=self.headers, timeout=10)
 
         if response.ok:
@@ -74,19 +77,19 @@ class TubeArchivist:
 
             return response.json()
 
-        print(response.text)
+        raise ValueError(f"video not found in TA: {path}")
-        return False
 
-    def get_thumb(self, path):
+    def get_thumb(self, path: str) -> bytes:
         """get encoded thumbnail from ta"""
-        url = CONFIG.get("ta_url") + path
+        url: str = CONFIG["ta_url"] + path
         response = requests.get(
             url, headers=self.headers, stream=True, timeout=10
         )
+        base64_thumb: bytes = base64.b64encode(response.content)
 
-        return base64.b64encode(response.content)
+        return base64_thumb
 
-    def ping(self):
+    def ping(self) -> None:
         """ping tubearchivist server"""
         response = self.get("ping/")
         if not response:
@@ -95,7 +98,7 @@ class TubeArchivist:
         print("[connection] verified tube archivist connection")
 
 
-def folder_check():
+def env_check() -> None:
     """check if ta_video_path is accessible"""
     if not os.path.exists("config.json"):
         raise FileNotFoundError("config.json file not found")

src/episode.py

@@ -3,33 +3,34 @@
 from datetime import datetime
 
 from src.connect import Jellyfin, TubeArchivist
+from src.static_types import TAVideo
 
 
 class Episode:
     """interact with an single episode"""
 
-    def __init__(self, youtube_id, jf_id):
+    def __init__(self, youtube_id: str, jf_id: str):
-        self.youtube_id = youtube_id
+        self.youtube_id: str = youtube_id
-        self.jf_id = jf_id
+        self.jf_id: str = jf_id
 
-    def sync(self):
+    def sync(self) -> None:
         """sync episode metadata"""
-        ta_video = self.get_ta_video()
+        ta_video: TAVideo = self.get_ta_video()
         self.update_metadata(ta_video)
         self.update_artwork(ta_video)
 
-    def get_ta_video(self):
+    def get_ta_video(self) -> TAVideo:
         """get video metadata from ta"""
-        path = f"/video/{self.youtube_id}"
+        path: str = f"/video/{self.youtube_id}"
-        ta_video = TubeArchivist().get(path)
+        ta_video: TAVideo = TubeArchivist().get(path)
 
         return ta_video
 
-    def update_metadata(self, ta_video):
+    def update_metadata(self, ta_video: TAVideo) -> None:
         """update jellyfin metadata from item_id"""
-        published = ta_video.get("published")
+        published: str = ta_video["published"]
-        published_date = datetime.strptime(published, "%d %b, %Y")
+        published_date: datetime = datetime.strptime(published, "%d %b, %Y")
-        data = {
+        data: dict = {
             "Id": self.jf_id,
             "Name": ta_video.get("title"),
             "Genres": [],
@@ -40,19 +41,24 @@ class Episode:
             "PremiereDate": published_date.isoformat(),
             "Overview": self._get_desc(ta_video),
         }
-        path = f"Items/{self.jf_id}"
+        path: str = f"Items/{self.jf_id}"
         Jellyfin().post(path, data)
 
-    def update_artwork(self, ta_video):
+    def update_artwork(self, ta_video: TAVideo) -> None:
         """update episode artwork in jf"""
-        thumb_base64 = TubeArchivist().get_thumb(ta_video.get("vid_thumb_url"))
+        thumb_path: str = ta_video["vid_thumb_url"]
-        path = f"Items/{self.jf_id}/Images/Primary"
+        thumb_base64: bytes = TubeArchivist().get_thumb(thumb_path)
+        path: str = f"Items/{self.jf_id}/Images/Primary"
         Jellyfin().post_img(path, thumb_base64)
 
-    def _get_desc(self, ta_video):
+    def _get_desc(self, ta_video: TAVideo) -> str | bool:
         """get description"""
-        raw_desc = ta_video.get("description").replace("\n", "<br>")
+        raw_desc: str = ta_video["description"]
-        if len(raw_desc) > 500:
+        if not raw_desc:
-            return raw_desc[:500] + " ..."
+            return False
 
-        return raw_desc
+        desc_clean: str = raw_desc.replace("\n", "<br>")
+        if len(raw_desc) > 500:
+            return desc_clean[:500] + " ..."
+
+        return desc_clean

src/series.py

@@ -7,115 +7,120 @@ from time import sleep
 from src.config import get_config
 from src.connect import Jellyfin, TubeArchivist
 from src.episode import Episode
+from src.static_types import JFEpisode, JFShow, TAChannel, TAVideo
 
 
 class Library:
     """grouped series"""
 
-    def __init__(self):
+    COLLECTION_ART = "assets/collection-art.jpg"
-        self.yt_collection = self.get_yt_collection()
 
-    def get_yt_collection(self):
+    def __init__(self) -> None:
+        self.yt_collection: str = self.get_yt_collection()
+
+    def get_yt_collection(self) -> str:
         """get collection id for youtube folder"""
-        path = "Items?Recursive=true&includeItemTypes=Folder"
+        path: str = "Items?Recursive=true&includeItemTypes=Folder"
-        folders = Jellyfin().get(path)
+        folders: dict = Jellyfin().get(path)
         for folder in folders["Items"]:
             if folder.get("Name").lower() == "youtube":
                 return folder.get("Id")
 
         raise ValueError("youtube folder not found")
 
-    def validate_series(self):
+    def validate_series(self) -> None:
         """validate all series"""
-        all_shows = self._get_all_series()
+        all_shows: list[JFShow] = self._get_all_series()["Items"]
-        for show in all_shows["Items"]:
+        for show in all_shows:
             show_handler = Show(show)
-            folders = show_handler.create_folders()
+            folders: list[str] = show_handler.create_folders()
             show_handler.validate_show()
             show_handler.validate_episodes()
             show_handler.delete_folders(folders)
 
         self.set_collection_art()
 
-    def _get_all_series(self):
+    def _get_all_series(self) -> dict:
         """get all shows indexed in jf"""
-        path = f"Items?Recursive=true&IncludeItemTypes=Series&fields=ParentId,Path&ParentId={self.yt_collection}"  # noqa: E501
+        path: str = f"Items?Recursive=true&IncludeItemTypes=Series&fields=ParentId,Path&ParentId={self.yt_collection}"  # noqa: E501
-        all_shows = Jellyfin().get(path)
+        all_shows: dict = Jellyfin().get(path)
 
         return all_shows
 
-    def set_collection_art(self):
+    def set_collection_art(self) -> None:
         """set collection ta art"""
-        with open("assets/collection-art.jpg", "rb") as f:
+        with open(self.COLLECTION_ART, "rb") as f:
-            asset = f.read()
+            asset: bytes = f.read()
 
-        folders = Jellyfin().get("Library/MediaFolders")
+        folders: dict = Jellyfin().get("Library/MediaFolders")
         for folder in folders["Items"]:
             if folder.get("Name").lower() == "youtube":
-                jf_id = folder.get("Id")
+                jf_id: str = folder.get("Id")
-                path = f"Items/{jf_id}/Images/Primary"
+                path: str = f"Items/{jf_id}/Images/Primary"
                 Jellyfin().post_img(path, base64.b64encode(asset))
 
 
 class Show:
     """interact with a single show"""
 
-    def __init__(self, show):
+    def __init__(self, show: JFShow):
-        self.show = show
+        self.show: JFShow = show
 
-    def _get_all_episodes(self):
+    def _get_all_episodes(self) -> list[JFEpisode]:
         """get all episodes of show"""
-        series_id = self.show.get("Id")
+        series_id: str = self.show["Id"]
-        path = f"Shows/{series_id}/Episodes?fields=Path"
+        path: str = f"Shows/{series_id}/Episodes?fields=Path"
         all_episodes = Jellyfin().get(path)
 
         return all_episodes["Items"]
 
-    def _get_expected_seasons(self):
+    def _get_expected_seasons(self) -> set[str]:
         """get all expected seasons"""
-        episodes = self._get_all_episodes()
+        episodes: list[JFEpisode] = self._get_all_episodes()
-        all_years = {os.path.split(i["Path"])[-1][:4] for i in episodes}
+        all_years: set[str] = {
+            os.path.split(i["Path"])[-1][:4] for i in episodes
+        }
 
         return all_years
 
-    def _get_existing_seasons(self):
+    def _get_existing_seasons(self) -> list[str]:
         """get all seasons indexed of series"""
-        series_id = self.show.get("Id")
+        series_id: str = self.show["Id"]
-        path = f"Shows/{series_id}/Seasons"
+        path: str = f"Shows/{series_id}/Seasons"
-        all_seasons = Jellyfin().get(path)
+        all_seasons: dict = Jellyfin().get(path)
 
         return [str(i.get("IndexNumber")) for i in all_seasons["Items"]]
 
-    def create_folders(self):
+    def create_folders(self) -> list[str]:
         """create season folders if needed"""
-        all_expected = self._get_expected_seasons()
+        all_expected: set[str] = self._get_expected_seasons()
-        all_existing = self._get_existing_seasons()
+        all_existing: list[str] = self._get_existing_seasons()
 
-        base = get_config().get("ta_video_path")
+        base: str = get_config()["ta_video_path"]
-        channel_name = os.path.split(self.show["Path"])[-1]
+        channel_name: str = os.path.split(self.show["Path"])[-1]
-        folders = []
+        folders: list[str] = []
         for year in all_expected:
             if year not in all_existing:
-                path = os.path.join(base, channel_name, year)
+                path: str = os.path.join(base, channel_name, year)
                 os.mkdir(path)
                 folders.append(path)
 
         self._wait_for_seasons()
         return folders
 
-    def delete_folders(self, folders):
+    def delete_folders(self, folders: list[str]) -> None:
         """delete temporary folders created"""
         for folder in folders:
             os.removedirs(folder)
 
-    def _wait_for_seasons(self):
+    def _wait_for_seasons(self) -> None:
         """wait for seasons to be created"""
-        jf_id = self.show["Id"]
+        jf_id: str = self.show["Id"]
-        path = f"Items/{jf_id}/Refresh?Recursive=true&ImageRefreshMode=Default&MetadataRefreshMode=Default"  # noqa: E501
+        path: str = f"Items/{jf_id}/Refresh?Recursive=true&ImageRefreshMode=Default&MetadataRefreshMode=Default"  # noqa: E501
         Jellyfin().post(path, False)
         for _ in range(12):
-            all_existing = set(self._get_existing_seasons())
+            all_existing: set[str] = set(self._get_existing_seasons())
-            all_expected = self._get_expected_seasons()
+            all_expected: set[str] = self._get_expected_seasons()
             if all_expected.issubset(all_existing):
                 return
 
@@ -124,50 +129,54 @@ class Show:
 
         raise TimeoutError("timeout reached for creating season folder")
 
-    def validate_show(self):
+    def validate_show(self) -> None:
         """set show metadata"""
-        ta_channel = self._get_ta_channel()
+        ta_channel: TAChannel = self._get_ta_channel()
         self.update_metadata(ta_channel)
         self.update_artwork(ta_channel)
 
-    def _get_ta_channel(self):
+    def _get_ta_channel(self) -> TAChannel:
         """get ta channel metadata"""
-        episode = self._get_all_episodes()[0]
+        episode: JFEpisode = self._get_all_episodes()[0]
-        youtube_id = os.path.split(episode["Path"])[-1][9:20]
+        youtube_id: str = os.path.split(episode["Path"])[-1][9:20]
         path = f"/video/{youtube_id}"
-        ta_video = TubeArchivist().get(path)
 
-        return ta_video.get("channel")
+        ta_video: TAVideo = TubeArchivist().get(path)
+        ta_channel: TAChannel = ta_video["channel"]
 
-    def update_metadata(self, ta_channel):
+        return ta_channel
+
+    def update_metadata(self, ta_channel: TAChannel) -> None:
         """update channel metadata"""
-        path = "Items/" + self.show["Id"]
+        path: str = "Items/" + self.show["Id"]
         data = {
             "Id": self.show["Id"],
-            "Name": ta_channel.get("channel_name"),
+            "Name": ta_channel["channel_name"],
-            "Overview": ta_channel.get("channel_description"),
+            "Overview": ta_channel["channel_description"],
             "Genres": [],
             "Tags": [],
             "ProviderIds": {},
         }
         Jellyfin().post(path, data)
 
-    def update_artwork(self, ta_channel):
+    def update_artwork(self, ta_channel: TAChannel) -> None:
         """set channel artwork"""
-        jf_id = self.show["Id"]
+        jf_id: str = self.show["Id"]
         jf_handler = Jellyfin()
+
         primary = TubeArchivist().get_thumb(ta_channel["channel_thumb_url"])
         jf_handler.post_img(f"Items/{jf_id}/Images/Primary", primary)
         jf_handler.post_img(f"Items/{jf_id}/Images/Logo", primary)
+
         banner = TubeArchivist().get_thumb(ta_channel["channel_banner_url"])
         jf_handler.post_img(f"Items/{jf_id}/Images/Banner", banner)
+
         tvart = TubeArchivist().get_thumb(ta_channel["channel_tvart_url"])
         jf_handler.post_img(f"Items/{jf_id}/Images/Backdrop", tvart)
 
-    def validate_episodes(self):
+    def validate_episodes(self) -> None:
         """sync all episodes"""
-        all_episodes = self._get_all_episodes()
+        all_episodes: list[JFEpisode] = self._get_all_episodes()
         for video in all_episodes:
-            youtube_id = os.path.split(video["Path"])[-1][9:20]
+            youtube_id: str = os.path.split(video["Path"])[-1][9:20]
-            jf_id = video["Id"]
+            Episode(youtube_id, video["Id"]).sync()
-            Episode(youtube_id, jf_id).sync()

src/static_types.py

@@ -0,0 +1,47 @@
+"""describe types"""
+
+from typing import TypedDict
+
+
+class ConfigType(TypedDict):
+    """describes the confic dict"""
+
+    ta_video_path: str
+    ta_url: str
+    ta_token: str
+    jf_url: str
+    jf_token: str
+
+
+class TAChannel(TypedDict):
+    """describes channel from TA API"""
+
+    channel_name: str
+    channel_description: str
+    channel_thumb_url: str
+    channel_banner_url: str
+    channel_tvart_url: str
+
+
+class TAVideo(TypedDict):
+    """describes video from TA API"""
+
+    published: str
+    title: str
+    vid_thumb_url: str
+    description: str
+    channel: TAChannel
+
+
+class JFShow(TypedDict):
+    """describes a show from JF API"""
+
+    Id: str
+    Path: str
+
+
+class JFEpisode(TypedDict):
+    """describes an episode in JF API"""
+
+    Id: str
+    Path: str