jellyfin/src/series.py

222 lines
7.7 KiB
Python
Raw Normal View History

2023-04-05 16:31:14 +00:00
"""set metadata to shows"""
import base64
import os
from time import sleep
from src.config import get_config
2023-04-07 05:25:14 +00:00
from src.connect import Jellyfin, TubeArchivist, clean_overview
2023-04-05 16:31:14 +00:00
from src.episode import Episode
2023-04-06 04:58:04 +00:00
from src.static_types import JFEpisode, JFShow, TAChannel, TAVideo
2023-04-05 16:31:14 +00:00
class Library:
"""grouped series"""
2023-04-06 04:58:04 +00:00
COLLECTION_ART = "assets/collection-art.jpg"
2023-04-05 16:31:14 +00:00
2023-04-06 04:58:04 +00:00
def __init__(self) -> None:
self.yt_collection: str = self.get_yt_collection()
def get_yt_collection(self) -> str:
2023-04-05 16:31:14 +00:00
"""get collection id for youtube folder"""
2023-04-06 04:58:04 +00:00
path: str = "Items?Recursive=true&includeItemTypes=Folder"
folders: dict = Jellyfin().get(path)
2023-04-05 16:31:14 +00:00
for folder in folders["Items"]:
if folder.get("Name").lower() == "youtube":
return folder.get("Id")
raise ValueError("youtube folder not found")
2023-04-06 04:58:04 +00:00
def validate_series(self) -> None:
2023-04-05 16:31:14 +00:00
"""validate all series"""
2023-04-06 04:58:04 +00:00
all_shows: list[JFShow] = self._get_all_series()["Items"]
for show in all_shows:
2023-04-05 16:31:14 +00:00
show_handler = Show(show)
2023-04-06 04:58:04 +00:00
folders: list[str] = show_handler.create_folders()
2023-04-05 16:31:14 +00:00
show_handler.validate_show()
show_handler.validate_episodes()
show_handler.delete_folders(folders)
2023-04-07 02:28:16 +00:00
collection_id: str = self._get_collection()
self.set_collection_art(collection_id)
self.refresh_collection(collection_id)
2023-04-05 16:31:14 +00:00
2023-04-06 04:58:04 +00:00
def _get_all_series(self) -> dict:
2023-04-05 16:31:14 +00:00
"""get all shows indexed in jf"""
2023-04-06 04:58:04 +00:00
path: str = f"Items?Recursive=true&IncludeItemTypes=Series&fields=ParentId,Path&ParentId={self.yt_collection}" # noqa: E501
all_shows: dict = Jellyfin().get(path)
2023-04-05 16:31:14 +00:00
return all_shows
2023-04-07 02:28:16 +00:00
def _get_collection(self) -> str:
"""get youtube collection id"""
folders: dict = Jellyfin().get("Library/MediaFolders")
for folder in folders["Items"]:
if folder.get("Name").lower() == "youtube":
return folder["Id"]
raise ValueError("youtube collection folder not found")
def set_collection_art(self, collection_id: str) -> None:
2023-04-05 16:31:14 +00:00
"""set collection ta art"""
2023-04-06 04:58:04 +00:00
with open(self.COLLECTION_ART, "rb") as f:
asset: bytes = f.read()
2023-04-05 16:31:14 +00:00
2023-04-07 02:28:16 +00:00
path: str = f"Items/{collection_id}/Images/Primary"
Jellyfin().post_img(path, base64.b64encode(asset))
def refresh_collection(self, collection_id: str) -> None:
"""trigger collection refresh"""
path: str = f"Items/{collection_id}/Refresh?Recursive=true&ImageRefreshMode=Default&MetadataRefreshMode=Default" # noqa: E501
Jellyfin().post(path, False)
2023-04-05 16:31:14 +00:00
class Show:
"""interact with a single show"""
2023-04-06 04:58:04 +00:00
def __init__(self, show: JFShow):
self.show: JFShow = show
2023-04-05 16:31:14 +00:00
2023-04-07 05:08:52 +00:00
def _get_all_episodes(
self,
filter_new: bool = False,
limit: int | bool = False,
) -> list[JFEpisode]:
2023-04-05 16:31:14 +00:00
"""get all episodes of show"""
2023-04-06 04:58:04 +00:00
series_id: str = self.show["Id"]
2023-04-07 05:08:52 +00:00
path: str = f"Shows/{series_id}/Episodes?fields=Path,Studios"
if limit:
path = f"{path}&limit={limit}"
2023-04-05 16:31:14 +00:00
all_episodes = Jellyfin().get(path)
2023-04-07 05:08:52 +00:00
all_items: list[JFEpisode] = all_episodes["Items"]
if filter_new:
all_items = [i for i in all_items if not i["Studios"]]
2023-04-05 16:31:14 +00:00
2023-04-07 05:08:52 +00:00
return all_items
2023-04-05 16:31:14 +00:00
2023-04-06 04:58:04 +00:00
def _get_expected_seasons(self) -> set[str]:
2023-04-05 16:31:14 +00:00
"""get all expected seasons"""
2023-04-06 04:58:04 +00:00
episodes: list[JFEpisode] = self._get_all_episodes()
all_years: set[str] = {
os.path.split(i["Path"])[-1][:4] for i in episodes
}
2023-04-05 16:31:14 +00:00
return all_years
2023-04-06 04:58:04 +00:00
def _get_existing_seasons(self) -> list[str]:
2023-04-05 16:31:14 +00:00
"""get all seasons indexed of series"""
2023-04-06 04:58:04 +00:00
series_id: str = self.show["Id"]
path: str = f"Shows/{series_id}/Seasons"
all_seasons: dict = Jellyfin().get(path)
2023-04-05 16:31:14 +00:00
return [str(i.get("IndexNumber")) for i in all_seasons["Items"]]
2023-04-06 04:58:04 +00:00
def create_folders(self) -> list[str]:
2023-04-05 16:31:14 +00:00
"""create season folders if needed"""
2023-04-06 04:58:04 +00:00
all_expected: set[str] = self._get_expected_seasons()
all_existing: list[str] = self._get_existing_seasons()
2023-04-05 16:31:14 +00:00
2023-04-06 04:58:04 +00:00
base: str = get_config()["ta_video_path"]
channel_name: str = os.path.split(self.show["Path"])[-1]
folders: list[str] = []
2023-04-05 16:31:14 +00:00
for year in all_expected:
if year not in all_existing:
2023-04-06 04:58:04 +00:00
path: str = os.path.join(base, channel_name, year)
2023-04-05 16:31:14 +00:00
os.mkdir(path)
folders.append(path)
self._wait_for_seasons()
return folders
2023-04-06 04:58:04 +00:00
def delete_folders(self, folders: list[str]) -> None:
2023-04-05 16:31:14 +00:00
"""delete temporary folders created"""
for folder in folders:
os.removedirs(folder)
2023-04-06 04:58:04 +00:00
def _wait_for_seasons(self) -> None:
2023-04-05 16:31:14 +00:00
"""wait for seasons to be created"""
2023-04-06 04:58:04 +00:00
jf_id: str = self.show["Id"]
path: str = f"Items/{jf_id}/Refresh?Recursive=true&ImageRefreshMode=Default&MetadataRefreshMode=Default" # noqa: E501
2023-04-05 16:31:14 +00:00
Jellyfin().post(path, False)
for _ in range(12):
2023-04-06 04:58:04 +00:00
all_existing: set[str] = set(self._get_existing_seasons())
all_expected: set[str] = self._get_expected_seasons()
2023-04-05 16:31:14 +00:00
if all_expected.issubset(all_existing):
return
print(f"[setup][{jf_id}] waiting for seasons to be created")
sleep(5)
raise TimeoutError("timeout reached for creating season folder")
2023-04-06 04:58:04 +00:00
def validate_show(self) -> None:
2023-04-05 16:31:14 +00:00
"""set show metadata"""
2023-04-06 04:58:04 +00:00
ta_channel: TAChannel = self._get_ta_channel()
2023-04-05 16:31:14 +00:00
self.update_metadata(ta_channel)
self.update_artwork(ta_channel)
2023-04-06 04:58:04 +00:00
def _get_ta_channel(self) -> TAChannel:
2023-04-05 16:31:14 +00:00
"""get ta channel metadata"""
2023-04-07 05:08:52 +00:00
episode: JFEpisode = self._get_all_episodes(limit=1)[0]
2023-04-06 04:58:04 +00:00
youtube_id: str = os.path.split(episode["Path"])[-1][9:20]
2023-04-05 16:31:14 +00:00
path = f"/video/{youtube_id}"
2023-04-06 04:58:04 +00:00
ta_video: TAVideo = TubeArchivist().get(path)
ta_channel: TAChannel = ta_video["channel"]
return ta_channel
2023-04-05 16:31:14 +00:00
2023-04-06 04:58:04 +00:00
def update_metadata(self, ta_channel: TAChannel) -> None:
2023-04-05 16:31:14 +00:00
"""update channel metadata"""
2023-04-06 04:58:04 +00:00
path: str = "Items/" + self.show["Id"]
2023-04-07 05:25:14 +00:00
data: dict = {
2023-04-05 16:31:14 +00:00
"Id": self.show["Id"],
2023-04-06 04:58:04 +00:00
"Name": ta_channel["channel_name"],
2023-04-07 05:25:14 +00:00
"Overview": self._get_desc(ta_channel),
2023-04-05 16:31:14 +00:00
"Genres": [],
"Tags": [],
"ProviderIds": {},
}
Jellyfin().post(path, data)
2023-04-07 05:25:14 +00:00
def _get_desc(self, ta_channel: TAChannel) -> str | bool:
"""get parsed description"""
raw_desc: str = ta_channel["channel_description"]
if not raw_desc:
return False
desc_clean: str = clean_overview(raw_desc)
return desc_clean
2023-04-06 04:58:04 +00:00
def update_artwork(self, ta_channel: TAChannel) -> None:
2023-04-05 16:31:14 +00:00
"""set channel artwork"""
2023-04-06 04:58:04 +00:00
jf_id: str = self.show["Id"]
2023-04-05 16:31:14 +00:00
jf_handler = Jellyfin()
2023-04-06 04:58:04 +00:00
2023-04-05 16:31:14 +00:00
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)
2023-04-06 04:58:04 +00:00
2023-04-05 16:31:14 +00:00
banner = TubeArchivist().get_thumb(ta_channel["channel_banner_url"])
jf_handler.post_img(f"Items/{jf_id}/Images/Banner", banner)
2023-04-06 04:58:04 +00:00
2023-04-05 16:31:14 +00:00
tvart = TubeArchivist().get_thumb(ta_channel["channel_tvart_url"])
jf_handler.post_img(f"Items/{jf_id}/Images/Backdrop", tvart)
2023-04-06 04:58:04 +00:00
def validate_episodes(self) -> None:
2023-04-05 16:31:14 +00:00
"""sync all episodes"""
2023-04-07 05:08:52 +00:00
showname: str = self.show["Name"]
new_episodes: list[JFEpisode] = self._get_all_episodes(filter_new=True)
if not new_episodes:
print(f"[show][{showname}] no new videos found")
return
2023-04-07 05:25:14 +00:00
print(f"[show][{showname}] indexing {len(new_episodes)} videos")
2023-04-07 05:08:52 +00:00
for video in new_episodes:
2023-04-06 04:58:04 +00:00
youtube_id: str = os.path.split(video["Path"])[-1][9:20]
Episode(youtube_id, video["Id"]).sync()