jellyfin/app/src/series.py

"""set metadata to shows"""

import base64
import os
from time import sleep

from src.config import get_config
from src.connect import Jellyfin, TubeArchivist, clean_overview
from src.episode import Episode
from src.static_types import JFEpisode, JFShow, TAChannel, TAVideo


class Library:
    """grouped series"""

    COLLECTION_ART = "assets/collection-art.jpg"

    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: 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) -> None:
        """validate all series"""
        collection_id: str = self._get_collection()
        self.refresh_collection(collection_id)
        all_shows: list[JFShow] = self._get_all_series()["Items"]
        for show in all_shows:
            show_handler = Show(show)
            show_handler.validate_show()
            folders: list[str] | None = show_handler.validate_episodes()
            if folders:
                show_handler.delete_folders(folders)

        self.set_collection_art(collection_id)
        self.refresh_collection(collection_id)

    def _get_all_series(self) -> dict:
        """get all shows indexed in jf"""
        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 _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:
        """set collection ta art"""
        with open(self.COLLECTION_ART, "rb") as f:
            asset: bytes = f.read()

        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)

        for _ in range(12):
            response = Jellyfin().get("Library/VirtualFolders")
            for folder in response:
                if not folder["ItemId"] == collection_id:
                    continue

                if folder["RefreshStatus"] == "Idle":
                    return

                print("waiting for library refresh")
                sleep(5)


class Show:
    """interact with a single show"""

    def __init__(self, show: JFShow):
        self.show: JFShow = show

    def validate_show(self) -> None:
        """set show metadata"""
        ta_channel: TAChannel | None = self._get_ta_channel()
        if ta_channel is None:
            return
        self.update_metadata(ta_channel)
        self.update_artwork(ta_channel)

    def _get_ta_channel(self) -> TAChannel | None:
        """get ta channel metadata"""
        channel_id: str = self.show["Path"].replace("\\", "/").split("/")[-1]
        ta_channel: TAChannel | None = TubeArchivist().get_channel(channel_id)

        return ta_channel

    def update_metadata(self, ta_channel: TAChannel) -> None:
        """update channel metadata"""
        path: str = "Items/" + self.show["Id"]
        data: dict = {
            "Id": self.show["Id"],
            "Name": ta_channel["channel_name"],
            "Overview": self._get_desc(ta_channel),
            "Genres": [],
            "Tags": [],
            "ProviderIds": {},
        }
        Jellyfin().post(path, data)

    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

    def update_artwork(self, ta_channel: TAChannel) -> None:
        """set channel artwork"""
        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) -> list[str] | None:
        """sync all episodes"""
        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 None

        print(f"[show][{showname}] indexing {len(new_episodes)} videos")
        seasons_created: list[str] = []
        for jf_ep in new_episodes:
            youtube_id: str = os.path.basename(
                jf_ep["Path"].replace("\\", "/")
            ).split(".")[0]
            episode_handler = Episode(youtube_id, jf_ep["Id"])
            ta_video: TAVideo = episode_handler.get_ta_video()
            season_folder: str | None = self.create_season(ta_video, jf_ep)
            episode_handler.sync(ta_video)
            if season_folder:
                seasons_created.append(season_folder)

        return seasons_created

    def _get_all_episodes(self, filter_new: bool = False) -> list[JFEpisode]:
        """get all episodes of show"""
        series_id: str = self.show["Id"]
        path: str = f"Shows/{series_id}/Episodes?fields=Path,Studios"

        all_episodes = Jellyfin().get(path)
        all_items: list[JFEpisode] = all_episodes["Items"]

        if filter_new:
            all_items = [i for i in all_items if not i["Studios"]]

        return all_items

    def create_season(self, ta_video: TAVideo, jf_ep: JFEpisode) -> str | None:
        """create season folders"""
        existing_seasons = self._get_existing_seasons()
        expected_season = ta_video["published"].split("-")[0]
        if expected_season in existing_seasons:
            return None

        base: str = get_config()["ta_video_path"]
        channel_folder = os.path.split(
            os.path.split(jf_ep["Path"].replace("\\", "/"))[0]
        )[-1]
        season_folder = os.path.join(base, channel_folder, expected_season)
        os.makedirs(season_folder)
        self._wait_for_season(expected_season)

        return season_folder

    def _wait_for_season(self, expected_season: str) -> None:
        """wait for season to be created in JF"""
        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[str] = set(self._get_existing_seasons())

            if expected_season in all_existing:
                return

            print(f"[setup][{jf_id}] waiting for seasons to be created")
            sleep(5)

        raise TimeoutError("timeout reached for creating season folder")

    def _get_existing_seasons(self) -> list[str]:
        """get all seasons indexed of series"""
        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 delete_folders(self, folders: list[str]) -> None:
        """delete temporary folders created"""
        for folder in folders:
            os.removedirs(folder)