Skip to content

Commit

Permalink
#49 Create TagManager module for enhanced RSPL tag handling
Browse files Browse the repository at this point in the history
Introduce the TagManager class to manage RSPL tags systematically, fetching different tag categories from RSPlaylist. Refactor RSPL tag structures in the configuration to improve consistency and simplify tag initialization. Prepare integration points for future tag validation and user-based customizations.
  • Loading branch information
kozaka-tv committed Dec 31, 2024
1 parent 5cebad6 commit cb15815
Show file tree
Hide file tree
Showing 6 changed files with 98 additions and 17 deletions.
35 changes: 21 additions & 14 deletions config/config_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import sys

from config.config_data import ConfigData, ConfRockSniffer, ConfSetlistLogger, ConfSceneSwitcher, ConfSongLoader, \
ConfFileManager, RSPLTags
ConfFileManager, RSPLTagNames
from config.config_reader import log, ConfigReader
from utils import string_utils
from utils.exceptions import ConfigError
Expand Down Expand Up @@ -90,20 +90,27 @@ def __create_conf_file_manager(conf):
return ConfFileManager(enabled, destination_dir, using_cfsm, download_dirs)


# Define constants for tag names
TAG_NAMES = [
TAG_TO_DOWNLOAD,
TAG_DOWNLOADED,
TAG_LOADED,
TAG_NEW_VIEWER_REQ,
TAG_RAIDER_REQ,
TAG_VIP_VIEWER_REQ
]


def __create_rspl_tags(conf):
tag_to_download = __get_tag(conf, TAG_TO_DOWNLOAD) # TODO temporary get it without validation. Validate in tag manager #49
tag_downloaded = __get_tag(conf, TAG_DOWNLOADED)
tag_loaded = __get_tag(conf, TAG_LOADED) # TODO temporary get it without validation. Validate in tag manager #49

tag_new_viewer_request = __get_tag(conf, TAG_NEW_VIEWER_REQ)
tag_raider_request = __get_tag(conf, TAG_RAIDER_REQ)
tag_vip_viewer_request = __get_tag(conf, TAG_VIP_VIEWER_REQ)
return RSPLTags(tag_to_download,
tag_downloaded,
tag_loaded,
tag_new_viewer_request,
tag_raider_request,
tag_vip_viewer_request)
tags = __fetch_tags(conf, TAG_NAMES)
return RSPLTagNames(*tags)


def __fetch_tags(conf, tag_names):
"""
Fetch multiple tags based on the provided tag names.
"""
return [__get_tag(conf, tag_name) for tag_name in tag_names]


def __create_conf_song_loader(conf):
Expand Down
4 changes: 2 additions & 2 deletions config/config_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ class ConfFileManager:


@dataclass
class RSPLTags:
class RSPLTagNames:
tag_to_download: str
tag_downloaded: str
tag_loaded: str
Expand All @@ -37,7 +37,7 @@ class ConfSongLoader:
enabled: bool
twitch_channel: str
phpsessid: str
rspl_tags: RSPLTags
rspl_tags: RSPLTagNames
cdlc_archive_dir: str
destination_dir: str
rocksmith_cdlc_dir: str
Expand Down
2 changes: 2 additions & 0 deletions modules/servant/servant.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from modules.servant.setlist.setlist_logger import SetlistLogger
from modules.servant.song_loader.song_loader import SongLoader
from modules.servant.song_loader.songs import Songs
from modules.servant.tag_manager.tag_manager import TagManager
from utils.cmd_line_parser import parse_args
from utils.exceptions import RocksnifferConnectionError, RSPLNotLoggedInError, \
RSPLPlaylistIsNotEnabledError, ConfigError
Expand Down Expand Up @@ -52,6 +53,7 @@ def __init__(self):
self.file_manager = FileManager(config_data)
self.songs = Songs()
self.song_loader = SongLoader(config_data, self.songs)
self.tag_manager = TagManager(config_data, self.song_loader)
self.scene_switcher = SceneSwitcher(config_data)

try:
Expand Down
2 changes: 1 addition & 1 deletion modules/servant/song_loader/song_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ def __init__(self, config_data: ConfigData, songs):
self.rsplaylist = None
self.rsplaylist_json = None
self.rsplaylist_updated = True
self.rspl_tags = config_data.song_loader.rspl_tags
self.rspl_tags = config_data.song_loader.rspl_tags # TODO itt nem kéne, hogy a tag manager-t izzítsuk és tölsük be? Vagy csak none?

self.cdlc_archive_dir = check_cdlc_archive_dir(config_data.song_loader.cdlc_archive_dir)
self.destination_dir = config_data.song_loader.destination_dir
Expand Down
21 changes: 21 additions & 0 deletions modules/servant/tag_manager/rspl_tag_data.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
from dataclasses import dataclass


@dataclass
class RSPLTag:
name: str
rspl_id: str
icon: str
color: str
user: bool


@dataclass
class RSPLTags:
tag_to_download: RSPLTag
tag_downloaded: RSPLTag
tag_loaded: RSPLTag
tag_new_viewer_request: RSPLTag
tag_raider_request: RSPLTag
tag_vip_viewer_request: RSPLTag

51 changes: 51 additions & 0 deletions modules/servant/tag_manager/tag_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
import logging

from dacite import from_dict

from config.config_data import ConfigData
from modules.servant.song_loader.rs_playlist_data import RsPlaylist
from modules.servant.song_loader.song_loader import SongLoader
from utils.rs_playlist_util import get_playlist

log = logging.getLogger()


class TagManager:
def __init__(self, config_data: ConfigData, song_loader: SongLoader):
# TODO
# TODO
# TODO
rsplaylist = self.get_rsplaylist(config_data)
self.all_tags = self.__fetch_all_tags(rsplaylist)
self.user_tags = self.__fetch_user_tags(rsplaylist)
self.server_tags = self.__fetch_server_tags(rsplaylist)

pass

@staticmethod
def get_rsplaylist(config_data):
loader = config_data.song_loader
new_playlist = get_playlist(loader.twitch_channel, loader.phpsessid)
rsplaylist = from_dict(data_class=RsPlaylist, data=new_playlist)
return rsplaylist

@staticmethod
def __fetch_all_tags(rsplaylist):
return {value.name: key for key, value in rsplaylist.channel_tags.items()}

@staticmethod
def __fetch_server_tags(rsplaylist):
return {value.name: key for key, value in rsplaylist.channel_tags.items() if not value.user}

@staticmethod
def __fetch_user_tags(rsplaylist):
return {value.name: key for key, value in rsplaylist.channel_tags.items() if value.user}

# def update_tags(self, new_playlist):
# self.rsplaylist_json = new_playlist
# self.rsplaylist = from_dict(data_class=RsPlaylist, data=new_playlist)
#
# def __log_available_rspl_tags(self):
# tags = self.rsplaylist.channel_tags
# user_tags = {value.name: key for key, value in tags.items() if value.user}
# log.warning("Tags set in RSPlaylist:\n%s", pprint.pformat(user_tags))

0 comments on commit cb15815

Please sign in to comment.