diff --git a/main.py b/main.py
index cd08242..eaddd28 100644
--- a/main.py
+++ b/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()
diff --git a/src/config.py b/src/config.py
index fdfbdc4..5b3bb7e 100644
--- a/src/config.py
+++ b/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
diff --git a/src/connect.py b/src/connect.py
index 60da19d..7530177 100644
--- a/src/connect.py
+++ b/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"]}
- base = CONFIG["jf_url"]
+ headers: dict = {
+ "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}"
- new_headers = self.headers.copy()
+ url: str = f"{self.base}/{path}"
+ 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(path)
+ response = self.get("Users")
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")}
- base = CONFIG["ta_url"]
+ ta_token: str = CONFIG["ta_token"]
+ 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)
- return False
+ raise ValueError(f"video not found in TA: {path}")
- 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")
diff --git a/src/episode.py b/src/episode.py
index 4407d7f..3045d87 100644
--- a/src/episode.py
+++ b/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):
- self.youtube_id = youtube_id
- self.jf_id = jf_id
+ def __init__(self, youtube_id: str, jf_id: str):
+ self.youtube_id: str = youtube_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}"
- ta_video = TubeArchivist().get(path)
+ path: str = f"/video/{self.youtube_id}"
+ 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_date = datetime.strptime(published, "%d %b, %Y")
- data = {
+ published: str = ta_video["published"]
+ published_date: datetime = datetime.strptime(published, "%d %b, %Y")
+ 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"))
- path = f"Items/{self.jf_id}/Images/Primary"
+ thumb_path: str = ta_video["vid_thumb_url"]
+ 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", "
")
- if len(raw_desc) > 500:
- return raw_desc[:500] + " ..."
+ raw_desc: str = ta_video["description"]
+ if not raw_desc:
+ return False
- return raw_desc
+ desc_clean: str = raw_desc.replace("\n", "
")
+ if len(raw_desc) > 500:
+ return desc_clean[:500] + " ..."
+
+ return desc_clean
diff --git a/src/series.py b/src/series.py
index f29bc0a..fb27ef7 100644
--- a/src/series.py
+++ b/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):
- self.yt_collection = self.get_yt_collection()
+ COLLECTION_ART = "assets/collection-art.jpg"
- 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"
- folders = Jellyfin().get(path)
+ path: str = "Items?Recursive=true&includeItemTypes=Folder"
+ 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()
- for show in all_shows["Items"]:
+ all_shows: list[JFShow] = self._get_all_series()["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
- all_shows = Jellyfin().get(path)
+ path: str = f"Items?Recursive=true&IncludeItemTypes=Series&fields=ParentId,Path&ParentId={self.yt_collection}" # noqa: E501
+ 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:
- asset = f.read()
+ with open(self.COLLECTION_ART, "rb") as f:
+ 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")
- path = f"Items/{jf_id}/Images/Primary"
+ jf_id: str = folder.get("Id")
+ 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):
- self.show = show
+ def __init__(self, show: JFShow):
+ 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")
- path = f"Shows/{series_id}/Episodes?fields=Path"
+ series_id: str = self.show["Id"]
+ 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()
- all_years = {os.path.split(i["Path"])[-1][:4] for i in episodes}
+ episodes: list[JFEpisode] = self._get_all_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")
- path = f"Shows/{series_id}/Seasons"
- all_seasons = Jellyfin().get(path)
+ series_id: str = self.show["Id"]
+ path: str = f"Shows/{series_id}/Seasons"
+ 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_existing = self._get_existing_seasons()
+ all_expected: set[str] = self._get_expected_seasons()
+ all_existing: list[str] = self._get_existing_seasons()
- base = get_config().get("ta_video_path")
- channel_name = os.path.split(self.show["Path"])[-1]
- folders = []
+ base: str = get_config()["ta_video_path"]
+ channel_name: str = os.path.split(self.show["Path"])[-1]
+ 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"]
- path = f"Items/{jf_id}/Refresh?Recursive=true&ImageRefreshMode=Default&MetadataRefreshMode=Default" # noqa: E501
+ jf_id: str = self.show["Id"]
+ 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_expected = self._get_expected_seasons()
+ all_existing: set[str] = set(self._get_existing_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]
- youtube_id = os.path.split(episode["Path"])[-1][9:20]
+ episode: JFEpisode = self._get_all_episodes()[0]
+ 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"),
- "Overview": ta_channel.get("channel_description"),
+ "Name": ta_channel["channel_name"],
+ "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]
- jf_id = video["Id"]
- Episode(youtube_id, jf_id).sync()
+ youtube_id: str = os.path.split(video["Path"])[-1][9:20]
+ Episode(youtube_id, video["Id"]).sync()
diff --git a/src/static_types.py b/src/static_types.py
new file mode 100644
index 0000000..1b011dd
--- /dev/null
+++ b/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