super().__init__(_PATH_TEMPLATES, (config.host, config.port),
_TaskHandler, *args, **kwargs)
self.config = config
- self.player = Player(config.whitelist_tags_display,
- config.whitelist_tags_prefilter,
- config.needed_tags_prefilter)
+ self.player = Player()
self.downloads = DownloadsManager()
self.downloads.clean_unfinished()
self.downloads.start_thread()
elif ((not selected)
or (selected.digest != self.server.player.current_digest)):
with DbConn() as conn:
- selected = VideoFile.get_one_with_whitelist_tags_display(
- conn,
- self.server.player.current_digest,
- self.server.config.whitelist_tags_display)
+ selected = VideoFile.get_one(
+ conn, self.server.player.current_digest)
if last_sent < self.server.player.last_update:
last_sent = self.server.player.last_update
title, tags, digest = '', '', ''
def _send_file_data(self) -> None:
digest = Hash.from_b64(self.path_toks[2])
with DbConn() as conn:
- file = VideoFile.get_one_with_whitelist_tags_display(
- conn, digest, self.server.config.whitelist_tags_display)
+ file = VideoFile.get_one(conn, digest)
unused_tags = file.unused_tags(conn)
self._send_rendered_template(
_NAME_TEMPLATE_FILE_DATA,
files = VideoFile.get_filtered(
conn,
filter_path,
- self.server.config.needed_tags_prefilter,
TagSet.from_joined(needed_tags_str),
- self.server.config.whitelist_tags_prefilter,
- self.server.config.whitelist_tags_display,
show_absent)
files.sort(key=lambda t: t.rel_path)
- for f in files:
- f.whitelist_tags_display =\
- self.server.config.whitelist_tags_display
self._send_rendered_template(_NAME_TEMPLATE_FILES,
{'files': files,
'selected': 'files',
set_attrs_from_dict({k[len(ENVIRON_PREFIX):].lower(): v
for k, v in environ.items()
if k.isupper() and k.startswith(ENVIRON_PREFIX)})
+ for attr_name in ('needed_tags_prefilter',
+ 'whitelist_tags_prefilter',
+ 'whitelist_tags_display'):
+ setattr(VideoFile, attr_name, getattr(self, attr_name))
class YoutubeQuery(DbData):
rel_path: Path
digest: Hash
tags: TagSet
+ # class attributes
+ needed_tags_prefilter: TagSet
+ whitelist_tags_prefilter: TagSet
+ whitelist_tags_display: TagSet
def __init__(self,
digest: Optional[Hash],
else:
self.last_update = last_update
self._hash_on_last_update = hash(self)
- self.whitelist_tags_display: Optional[TagSet] = None
def __hash__(self) -> int:
return hash(f'{self.digest.b64}|{self.rel_path}|{self.flags}|'
self._renew_last_update()
return super().save(conn)
- @classmethod
- def get_one_with_whitelist_tags_display(
- cls, conn: DbConn, id_: Hash, whitelist_tags_display: TagSet
- ) -> Self:
- """Same as .get_one except sets .whitelist_tags_display."""
- vf = cls.get_one(conn, id_)
- vf.whitelist_tags_display = whitelist_tags_display
- return vf
-
@classmethod
def get_by_yt_id(cls, conn: DbConn, yt_id: YoutubeId) -> Self:
"""Return VideoFile of .yt_id."""
def get_filtered(cls,
conn: DbConn,
filter_path: FilterStr,
- needed_tags_dark: TagSet,
needed_tags_seen: TagSet,
- whitelist_tags_filter: TagSet,
- whitelist_tags_display: TagSet,
show_absent: bool = False
) -> list[Self]:
"""Return cls.get_all matching provided filter criteria."""
- files = [f for f in cls.get_all(conn)
- if (show_absent or f.present)
- and str(filter_path).lower() in str(f.rel_path).lower()
- and (needed_tags_dark.are_all_in(f.tags)
- or not whitelist_tags_filter.all_also_in(f.tags).empty)
- and needed_tags_seen.whitelisted(whitelist_tags_display
- ).are_all_in(f.tags)]
- for f in files:
- f.whitelist_tags_display = whitelist_tags_display
- return files
+ return [
+ f for f in cls.get_all(conn)
+ if (show_absent or f.present)
+ and str(filter_path).lower() in str(f.rel_path).lower()
+ and (cls.needed_tags_prefilter.are_all_in(f.tags)
+ or not cls.whitelist_tags_prefilter.all_also_in(f.tags).empty)
+ and needed_tags_seen.whitelisted(cls.whitelist_tags_display
+ ).are_all_in(f.tags)]
@property
def tags_showable(self) -> TagSet:
"""Show .tags passing .whitelist_tags_display, if latter set."""
- if self.whitelist_tags_display is None:
- raise HandledException(
- 'canot show display-whitelisted tags on unset whitelist')
return self.tags.whitelisted(self.whitelist_tags_display)
def unused_tags(self, conn: DbConn) -> TagSet:
"""Return tags used among other VideoFiles, not in self."""
- if self.whitelist_tags_display is None:
- raise HandledException(
- 'canot show display-whitelisted tags on unset whitelist')
tags = TagSet()
for file in self.get_all(conn):
tags.add(file.tags.all_not_in(self.tags).whitelisted(
_idx: int
last_update: DatetimeStr
- def __init__(self,
- whitelist_tags_display: TagSet,
- whitelist_tags_prefilter: TagSet,
- needed_tags_prefilter: TagSet,
- ) -> None:
- # filters setup
+ def __init__(self) -> None:
self.filter_path = FilterStr('')
self.needed_tags = TagSet()
- self._whitelist_tags_prefilter = whitelist_tags_prefilter
- self._whitelist_tags_display = whitelist_tags_display
- self._needed_tags_prefilter = needed_tags_prefilter
- # actual playlist and player setup
self._mpv: Optional[MPV] = None
self._monitoring_kill: bool = False
self._kill_queue: Queue = Queue()
def load_files_and_mpv(self) -> None:
"""Collect filtered files into playlist, shuffle, start player."""
with DbConn() as conn:
- known_files = {
- f.full_path: f for f
- in VideoFile.get_filtered(
- conn,
- self.filter_path,
- self._needed_tags_prefilter,
- self.needed_tags,
- self._whitelist_tags_prefilter,
- self._whitelist_tags_display)}
+ known_files = {f.full_path: f for f in VideoFile.get_filtered(
+ conn, self.filter_path, self.needed_tags)}
self.playlist = [known_files[p] for p in PATH_DOWNLOADS.iterdir()
if p in known_files
and p.is_file()
projects / ytplom / commitdiff