[media] Support extended format/metadata for media dirs.

This commit is contained in:
Fabio Manganiello 2024-08-19 23:56:05 +02:00
parent 077e12e9a8
commit 213498318f
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
2 changed files with 141 additions and 20 deletions

View file

@ -13,6 +13,7 @@ from typing import (
Type, Type,
Union, Union,
) )
from urllib.parse import urlparse
import requests import requests
@ -27,12 +28,14 @@ from platypush.utils import (
) )
from ._constants import audio_extensions, video_extensions from ._constants import audio_extensions, video_extensions
from ._model import DownloadState, PlayerState from ._model import DownloadState, MediaDirectory, PlayerState
from ._resource import MediaResource from ._resource import MediaResource
from ._resource.downloaders import DownloadThread, MediaResourceDownloader, downloaders from ._resource.downloaders import DownloadThread, MediaResourceDownloader, downloaders
from ._resource.parsers import MediaResourceParser, parsers from ._resource.parsers import MediaResourceParser, parsers
from ._search import MediaSearcher, searchers from ._search import MediaSearcher, searchers
_MediaDirs = Union[str, Iterable[Union[str, dict]], Dict[str, Union[str, dict]]]
class MediaPlugin(RunnablePlugin, ABC): class MediaPlugin(RunnablePlugin, ABC):
""" """
@ -65,7 +68,7 @@ class MediaPlugin(RunnablePlugin, ABC):
def __init__( def __init__(
self, self,
media_dirs: Optional[Union[str, Iterable[str], Dict[str, str]]] = None, media_dirs: Optional[_MediaDirs] = None,
download_dir: Optional[str] = None, download_dir: Optional[str] = None,
env: Optional[Dict[str, str]] = None, env: Optional[Dict[str, str]] = None,
volume: Optional[Union[float, int]] = None, volume: Optional[Union[float, int]] = None,
@ -82,8 +85,51 @@ class MediaPlugin(RunnablePlugin, ABC):
""" """
:param media_dirs: Directories that will be scanned for media files when :param media_dirs: Directories that will be scanned for media files when
a search is performed (default: only ``download_dir``). You can a search is performed (default: only ``download_dir``). You can
specify it either as a list of string or a map in the format specify it either:
``{<name>: <path>}``.
- As a list of strings:
.. code-block:: yaml
media_dirs:
- /mnt/hd/media/movies
- /mnt/hd/media/music
- /mnt/hd/media/series
- As a dictionary where the key is the name of the media display
name and the value is the path:
.. code-block:: yaml
media_dirs:
Movies: /mnt/hd/media/movies
Music: /mnt/hd/media/music
Series: /mnt/hd/media/series
- As a dictionary where the key is the name of the media display
name and the value is a dictionary with the path and additional
display information:
media_dirs:
Movies:
path: /mnt/hd/media/movies
icon:
url: https://example.com/icon.png
# FontAwesome icon classes are supported
class: fa fa-film
Music:
path: /mnt/hd/media/music
icon:
url: https://example.com/icon.png
class: fa fa-music
Series:
path: /mnt/hd/media/series
icon:
url: https://example.com/icon.png
class: fa fa-tv
:param download_dir: Directory where external resources/torrents will be :param download_dir: Directory where external resources/torrents will be
downloaded (default: ~/Downloads) downloaded (default: ~/Downloads)
:param env: Environment variables key-values to pass to the :param env: Environment variables key-values to pass to the
@ -150,10 +196,6 @@ class MediaPlugin(RunnablePlugin, ABC):
if not player: if not player:
raise AttributeError('No media plugin configured') raise AttributeError('No media plugin configured')
self.media_dirs = self._parse_media_dirs(
media_dirs or player_config.get('media_dirs', [])
)
if self.__class__.__name__ == 'MediaPlugin': if self.__class__.__name__ == 'MediaPlugin':
# Populate this plugin with the actions of the configured player # Populate this plugin with the actions of the configured player
for act in player.registered_actions: for act in player.registered_actions:
@ -193,6 +235,10 @@ class MediaPlugin(RunnablePlugin, ABC):
self.ytdl_args = ytdl_args or [] self.ytdl_args = ytdl_args or []
self._latest_resource: Optional[MediaResource] = None self._latest_resource: Optional[MediaResource] = None
self.media_dirs = self._parse_media_dirs(
media_dirs or player_config.get('media_dirs', [])
)
self._parsers: Dict[Type[MediaResourceParser], MediaResourceParser] = { self._parsers: Dict[Type[MediaResourceParser], MediaResourceParser] = {
parser: parser(self) for parser in parsers parser: parser(self) for parser in parsers
} }
@ -202,14 +248,16 @@ class MediaPlugin(RunnablePlugin, ABC):
] = {downloader: downloader(self) for downloader in downloaders} ] = {downloader: downloader(self) for downloader in downloaders}
self._searchers: Dict[Type[MediaSearcher], MediaSearcher] = { self._searchers: Dict[Type[MediaSearcher], MediaSearcher] = {
searcher: searcher(dirs=self.media_dirs.values(), media_plugin=self) searcher: searcher(
dirs=[d.path for d in self.media_dirs.values()], media_plugin=self
)
for searcher in searchers for searcher in searchers
} }
@staticmethod @staticmethod
def _parse_media_dirs( def _parse_media_dirs(
media_dirs: Optional[Union[str, Iterable[str], Dict[str, str]]] media_dirs: Optional[_MediaDirs],
) -> Dict[str, str]: ) -> Dict[str, MediaDirectory]:
dirs = {} dirs = {}
if media_dirs: if media_dirs:
@ -224,11 +272,38 @@ class MediaPlugin(RunnablePlugin, ABC):
ret = {} ret = {}
for k, v in dirs.items(): for k, v in dirs.items():
v = os.path.abspath(os.path.expanduser(v)) assert isinstance(k, str), f'Invalid media_dirs key format: {k}'
if os.path.isdir(v): if isinstance(v, str):
ret[k] = v v = {'path': v}
return ret assert isinstance(v, dict), f'Invalid media_dirs format: {v}'
path = v.get('path')
assert path, f'Missing path in media_dirs entry {k}'
path = os.path.abspath(os.path.expanduser(path))
assert os.path.isdir(path), f'Invalid path in media_dirs entry {k}'
icon = v.get('icon', {})
if isinstance(icon, str):
# Fill up the URL field if it's a URL, otherwise assume that
# it's a FontAwesome icon class
icon = {'url': icon} if urlparse(icon).scheme else {'class': icon}
ret[k] = MediaDirectory.build(
name=k,
path=path,
icon_class=icon.get('class'),
icon_url=icon.get('url'),
)
# Add the downloads directory if it's missing
if not any(d.path == get_default_downloads_dir() for d in ret.values()):
ret['Downloads'] = MediaDirectory.build(
name='Downloads',
path=get_default_downloads_dir(),
icon_class='fas fa-download',
)
return {k: ret[k] for k in sorted(ret.keys())}
def _get_resource( def _get_resource(
self, self,
@ -713,14 +788,11 @@ class MediaPlugin(RunnablePlugin, ABC):
) )
@action @action
def get_media_dirs(self) -> Dict[str, str]: def get_media_dirs(self) -> Dict[str, dict]:
""" """
:return: List of configured media directories. :return: List of configured media directories.
""" """
return { return {dir.name: dir.to_dict() for dir in self.media_dirs.values()}
'Downloads': self.download_dir,
**self.media_dirs,
}
def _get_downloads(self, url: Optional[str] = None, path: Optional[str] = None): def _get_downloads(self, url: Optional[str] = None, path: Optional[str] = None):
assert url or path, 'URL or path must be specified' assert url or path, 'URL or path must be specified'

View file

@ -1,4 +1,6 @@
import enum import enum
from dataclasses import dataclass, field
from typing import Optional
class PlayerState(enum.Enum): class PlayerState(enum.Enum):
@ -26,4 +28,51 @@ class DownloadState(enum.Enum):
ERROR = 'error' ERROR = 'error'
@dataclass
class MediaDirectoryIcon:
"""
Dataclass that represents a media directory icon.
"""
class_: str = 'fas fa-folder'
url: Optional[str] = None
def to_dict(self) -> dict:
"""
Convert the MediaDirectoryIcon instance to a dictionary.
"""
return {'class': self.class_, 'url': self.url}
@dataclass
class MediaDirectory:
"""
Dataclass that represents a media directory.
"""
name: str
path: str
icon: MediaDirectoryIcon = field(default_factory=MediaDirectoryIcon)
@classmethod
def build(
cls,
name: str,
path: str,
icon_class: Optional[str] = None,
icon_url: Optional[str] = None,
) -> 'MediaDirectory':
"""
Create a MediaDirectory instance from a dictionary.
"""
icon_class = icon_class or 'fas fa-folder'
return cls(name, path, MediaDirectoryIcon(icon_class, icon_url))
def to_dict(self) -> dict:
"""
Convert the MediaDirectory instance to a dictionary.
"""
return {'name': self.name, 'path': self.path, 'icon': self.icon.to_dict()}
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et: