Merge pull request #283 from kozaka-tv/49-tag-manager

49-tag-manager

.pylintrc

@@ -73,7 +73,7 @@ ignored-modules=
 # Python code to execute, usually for sys.path manipulation such as
 # pygtk.require().
 [MASTER]
-init-hook='import sys; sys.path.append("/path/to/virtualenv_or_pipenv/.../Lib/site-packages/")'
+init-hook='import sys; sys.path.append("/venv/Lib/site-packages/")'
 
 # Use multiple processes to speed up Pylint. Specifying 0 will auto-detect the
 # number of processors available to use, and will cap the count on Windows to

config/config_controller.py

@@ -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
@@ -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):

config/config_data.py

@@ -23,7 +23,7 @@ class ConfFileManager:
 
 
 @dataclass
-class RSPLTags:
+class RSPLTagNames:
     tag_to_download: str
     tag_downloaded: str
     tag_loaded: str
@@ -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

modules/servant/file_manager/cdlc_file_manager.py

@@ -4,7 +4,7 @@
 
 from common.definitions import TMP_DIR
 from utils import file_utils, collection_utils
-from utils.collection_utils import repr_in_multi_line, is_not_empty
+from utils.collection_utils import repr_in_multi_line, is_collection_not_empty
 
 HEARTBEAT = 1
 HEARTBEAT_NOT_PARSED = 5
@@ -63,7 +63,7 @@ def __beat_last_run_not_parsed(self):
     def __move_not_parsed_files_to_tmp(self):
         non_parsed_files = self.__scan_cdlc_files_in_destination_dir()
 
-        if is_not_empty(non_parsed_files):
+        if is_collection_not_empty(non_parsed_files):
             log.warning(
                 "Found %s file(s) in %s dir which one(s) were not yet parsed so I moving them to %s now! Files: %s"
                 , len(non_parsed_files), self.destination_dir, TMP_DIR, repr_in_multi_line(non_parsed_files))
@@ -86,7 +86,7 @@ def __scan_cdlc_files_in_download_dirs(self) -> set:
 
         cdlc_files, bad_dirs = file_utils.get_files_from_directories(self.download_dirs)
 
-        if collection_utils.is_not_empty(bad_dirs):
+        if collection_utils.is_collection_not_empty(bad_dirs):
             log.error("---------------------------------------")
             log.error('Bad definition or could not reach some directories defined in the config '
                       'under the section FileManager with the key download_dirs')

modules/servant/servant.py

@@ -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
@@ -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:

modules/servant/song_loader/song_data.py

@@ -44,28 +44,19 @@ def is_official(self):
         return self.rspl_official in KEY_VALUES_OF_AN_OFFICIAL_CDLC
 
     def __repr__(self):
-        rep = '<SongData: '
-
-        if self.song_filename:
-            rep += f"song_filename={self.song_filename}, "
-
-        if self.artist_title:
-            rep += f"artist_title={self.artist_title}, "
-
-        if self.rspl_request_id:
-            rep += f"rspl_request_id={self.rspl_request_id}, "
-        if self.cdlc_id:
-            rep += f"cdlc_id={self.cdlc_id}, "
-        if self.rspl_song_id:
-            rep += f"rspl_song_id={self.rspl_song_id}, "
-        if self.rspl_official:
-            rep += f"rspl_official={self.rspl_official}, "
-        if self.rspl_position:
-            rep += f"rspl_position={self.rspl_position}, "
-
-        if self.tags:
-            rep += f"tags={self.tags}"
-
-        rep += ">"
-
-        return rep
+        def format_attribute(name, value):
+            return f"{name}={value}" if value else None
+
+        attributes = [
+            format_attribute("song_filename", self.song_filename),
+            format_attribute("artist_title", self.artist_title),
+            format_attribute("rspl_request_id", self.rspl_request_id),
+            format_attribute("cdlc_id", self.cdlc_id),
+            format_attribute("rspl_song_id", self.rspl_song_id),
+            format_attribute("rspl_official", self.rspl_official),
+            format_attribute("rspl_position", self.rspl_position),
+            format_attribute("tags", self.tags)
+        ]
+
+        attributes_str = ", ".join(attr for attr in attributes if attr)
+        return f"<SongData: {attributes_str}>"

modules/servant/song_loader/song_loader.py

@@ -1,6 +1,5 @@
 import logging
 import os
-import pprint
 from time import time
 
 from dacite import from_dict
@@ -12,11 +11,13 @@
 from modules.servant.song_loader.song_data import SongData, ArtistTitle
 from modules.servant.song_loader.song_loader_helper import playlist_does_not_changed, check_cdlc_archive_dir, \
     check_rocksmith_cdlc_dir, update_tags_in_song_data, is_official, log_new_songs_found
+from modules.servant.tag_manager import tag_utils
 from utils import file_utils, psarc_reader
-from utils.collection_utils import is_not_empty, is_empty, repr_in_multi_line
+from utils.collection_utils import is_collection_not_empty, is_collection_empty, repr_in_multi_line
 from utils.exceptions import BadDirectoryError
 from utils.exceptions import RSPLNotLoggedInError
-from utils.rs_playlist_util import get_playlist, user_is_not_logged_in, set_tag_to_download, set_tag_loaded
+from utils.rs_playlist_util import get_playlist, user_is_not_logged_in, set_tag_to_download, set_tag_loaded, \
+    unset_user_tags
 from utils.string_utils import time_float_to_string
 
 HEARTBEAT = 5
@@ -34,6 +35,7 @@ def __init__(self, config_data: ConfigData, songs):
             self.rsplaylist = None
             self.rsplaylist_json = None
             self.rsplaylist_updated = True
+            # TODO itt nem kéne, hogy a tag manager-t izzítsuk és töltsük be? Vagy csak none?
             self.rspl_tags = config_data.song_loader.rspl_tags
 
             self.cdlc_archive_dir = check_cdlc_archive_dir(config_data.song_loader.cdlc_archive_dir)
@@ -123,7 +125,7 @@ def __playlist_has_been_changed(self):
         if self.rsplaylist_json is None:
             self.__update_playlist_with(new_playlist)
             log.info("Initial load of the playlist is done...")
-            self.__log_available_rspl_tags()
+            tag_utils.log_available_rspl_tags(self.rsplaylist.channel_tags)
             return True
 
         if playlist_does_not_changed(self.rsplaylist_json, new_playlist):
@@ -135,11 +137,6 @@ def __playlist_has_been_changed(self):
 
         return True
 
-    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))
-
     def __update_playlist_with(self, new_playlist):
         self.rsplaylist_json = new_playlist
         self.rsplaylist = from_dict(data_class=RsPlaylist, data=new_playlist)
@@ -189,13 +186,13 @@ def __get_cdlc_filenames_from_rs_dir(self):
     def __clean_up_archive_dir_for_duplicates(self, filenames_from_rs_dir, filenames_from_archive_dir):
         log.info('Cleaning up archive dir for duplicates')
         duplicates = set(filenames_from_rs_dir).intersection(filenames_from_archive_dir)
-        if is_not_empty(duplicates):
+        if is_collection_not_empty(duplicates):
             self.__remove_duplicates(duplicates, filenames_from_archive_dir)
 
     def __clean_up_songs_in_db(self, filenames_in_rs_and_archive_dir, filenames_in_db):
         log.info('Cleaning up songs in DB')
         to_delete = filenames_in_db.difference(filenames_in_rs_and_archive_dir)
-        if is_not_empty(to_delete):
+        if is_collection_not_empty(to_delete):
             log.info("Count of songs to be deleted: %s", len(to_delete))
             self.db_manager.delete_song_by_filename(to_delete)
 
@@ -268,7 +265,7 @@ def __extract_song_information(directory, cdlc_file_name: str):
         return song_data
 
     def __move_requested_cdlc_files_from_archive_to_rs(self):
-        if is_empty(self.songs.songs_from_archive_has_to_be_moved):
+        if is_collection_empty(self.songs.songs_from_archive_has_to_be_moved):
             log.debug("---- No file found to move from archive to RS directory.")
             return
 
@@ -306,9 +303,9 @@ def __move_requested_cdlc_files_from_archive_to_rs(self):
                                             song_data.rspl_request_id,
                                             self.rspl_tags)
 
-        if is_not_empty(self.songs.moved_from_archive):
+        if is_collection_not_empty(self.songs.moved_from_archive):
             log.warning("---- Files newly moved and will be parsed: %s", str(actually_loaded_songs))
-        if is_not_empty(self.songs.missing_from_archive):
+        if is_collection_not_empty(self.songs.missing_from_archive):
             log.error("---- Missing files but found in Database: %s", str(self.songs.missing_from_archive))
 
     def __song_already_moved_from_archive(self, filename):
@@ -334,13 +331,18 @@ def __find_existing_song_filenames_from_db_according_to_the_requests(self):
                 official = dlc_set_item.official
                 if is_official(official):
                     log.info("Skipping ODLC request with cdlc_id=%s - %s - %s", cdlc_id, artist, title)
+                    unset_user_tags(self.twitch_channel,
+                                    self.phpsessid,
+                                    playlist_item.id,
+                                    self.rspl_tags,
+                                    playlist_item.tags)
                     continue
 
                 songs_in_the_db = self.db_manager.search_song_by_artist_and_title(artist, title)
 
                 rspl_position = str(playlist_item.position)
 
-                if is_empty(songs_in_the_db):
+                if is_collection_empty(songs_in_the_db):
                     if self.__do_not_has_the_tag_to_download(playlist_item):
                         log.warning("User must download the song: rspl_position=%s cdlc_id=%s - %s - %s",
                                     rspl_position, cdlc_id, artist, title)
@@ -368,7 +370,7 @@ def __find_existing_song_filenames_from_db_according_to_the_requests(self):
 
                     log.info("Request found in DB: %s", song_data)
 
-        if is_not_empty(self.songs.requested_songs_found_in_db):
+        if is_collection_not_empty(self.songs.requested_songs_found_in_db):
             log.info("Existing songs found: %s", repr_in_multi_line(self.songs.requested_songs_found_in_db))
 
     def __calculate_songs_need_to_be_moved_from_archive_to_under_rs(self):
@@ -390,7 +392,7 @@ def __calculate_songs_need_to_be_moved_from_archive_to_under_rs(self):
             elif filename in difference:
                 self.songs.songs_from_archive_has_to_be_moved.update({filename: song_data})
 
-        if is_not_empty(self.songs.songs_from_archive_has_to_be_moved):
+        if is_collection_not_empty(self.songs.songs_from_archive_has_to_be_moved):
             log.info("Songs from archive has to be moved according to the requests: %s",
                      repr_in_multi_line(self.songs.songs_from_archive_has_to_be_moved))
 

modules/servant/tag_manager/rspl_tag_data.py

@@ -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
+

modules/servant/tag_manager/tag_manager.py

@@ -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))

modules/servant/tag_manager/tag_utils.py

@@ -0,0 +1,24 @@
+import logging
+import pprint
+from typing import Dict
+
+from modules.servant.song_loader.rs_playlist_data import ChannelTag
+
+log = logging.getLogger()
+
+
+def log_available_rspl_tags(rsplaylist_channel_tags: Dict[str, ChannelTag]) -> None:
+    user_tags = __filter_tags(rsplaylist_channel_tags, is_user=True)
+    system_tags = __filter_tags(rsplaylist_channel_tags, is_user=False)
+
+    __log_tags(user_tags, log.warning, "Tags set by the user in RSPlaylist")
+    __log_tags(system_tags, log.info, "System tags set in RSPlaylist")
+
+
+def __filter_tags(tags_to_filter, is_user):
+    return {value.name: key for key, value in tags_to_filter.items() if value.user == is_user}
+
+
+def __log_tags(tags_to_log, log_function, prefix_message):
+    formatted_tags = pprint.pformat(tags_to_log)
+    log_function(f"{prefix_message}:\n{formatted_tags}")