sima/plugins/internal/tags.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright (c) 2020, 2021 kaliko <kaliko@azylum.org>
   3 #
   4 #  This file is part of sima
   5 #
   6 #  sima is free software: you can redistribute it and/or modify
   7 #  it under the terms of the GNU General Public License as published by
   8 #  the Free Software Foundation, either version 3 of the License, or
   9 #  (at your option) any later version.
  10 #
  11 #  sima is distributed in the hope that it will be useful,
  12 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 #  GNU General Public License for more details.
  15 #
  16 #  You should have received a copy of the GNU General Public License
  17 #  along with sima.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19 #
  20 """
  21 Add titles based on tags
  22 """
  23
  24 # standard library import
  25 import logging
  26 import random
  27
  28 # third parties components
  29 from musicpd import CommandError
  30
  31 # local import
  32 from ...lib.plugin import AdvancedPlugin
  33 from ...lib.meta import Artist, MetaContainer
  34 from ...utils.utils import PluginException
  35
  36
  37 def control_config(tags_config):
  38     log = logging.getLogger('sima')
  39     sup_tags = Tags.supported_tags
  40     config_tags = {k for k, v in tags_config.items()
  41                    if (v and k in Tags.supported_tags)}
  42     if not tags_config.get('filter', None) and \
  43             config_tags.isdisjoint(sup_tags):
  44         log.warning('Found no config for Tags plugin! '
  45                     'Need at least "filter" or a supported tag')
  46         log.info('Supported Tags are : %s', ', '.join(sup_tags))
  47         return False
  48     if config_tags.difference(sup_tags):
  49         log.error('Found unsupported tag in config: %s',
  50                   config_tags.difference(sup_tags))
  51         return False
  52     return True
  53
  54
  55 def forge_filter(cfg):
  56     tags = set(cfg.keys()) & Tags.supported_tags
  57     cfg_filter = cfg.get('filter', None)
  58     mpd_filter = []
  59     if cfg_filter:
  60         mpd_filter.append(cfg_filter)
  61     for tag in tags:
  62         if not cfg[tag]:  # avoid empty tags entries in config
  63             continue
  64         if ',' in cfg[tag]:
  65             patt = '|'.join(map(str.strip, cfg[tag].split(',')))
  66             mpd_filter.append(f"({tag} =~ '({patt})')")
  67         else:
  68             mpd_filter.append(f"({tag} == '{cfg[tag].strip()}')")
  69     mpd_filter = ' AND '.join(mpd_filter)
  70     # Ensure there is at least an artist name
  71     mpd_filter = f"({mpd_filter} AND (artist != ''))"
  72     return mpd_filter
  73
  74
  75 class Tags(AdvancedPlugin):
  76     """Add track based on tags content
  77     """
  78     supported_tags = {'comment', 'date', 'genre', 'label', 'originaldate'}
  79     # options = {'queue_mode', 'priority', 'filter', 'track_to_add',
  80     #            'album_to_add'}
  81
  82     def __init__(self, daemon):
  83         super().__init__(daemon)
  84         self._control_conf()
  85         self.mpd_filter = forge_filter(self.plugin_conf)
  86         self._setup_tagsneeded()
  87         self.log.debug('mpd filter: %s', self.mpd_filter)
  88
  89     def _control_conf(self):
  90         if not control_config(self.plugin_conf):
  91             raise PluginException('plugin misconfiguration')
  92
  93     def _setup_tagsneeded(self):
  94         """Ensure needed tags are exposed by MPD"""
  95         # At this point mpd_filter concatenetes {tags}+filter
  96         config_tags = set()
  97         for mpd_supp_tags in self.player.MPD_supported_tags:
  98             if mpd_supp_tags.lower() in self.mpd_filter.lower():
  99                 config_tags.add(mpd_supp_tags.lower())
 100         self.log.debug('%s plugin needs the following metadata: %s',
 101                        self, config_tags)
 102         tags = config_tags & Tags.supported_tags
 103         self.player.needed_tags |= tags
 104
 105     def start(self):
 106         if (0, 21, 0) > tuple(map(int, self.player.mpd_version.split('.'))):
 107             self.log.warning('MPD protocol version: %s < 0.21.0',
 108                              self.player.mpd_version)
 109             self.log.error(
 110                 'Need at least MPD 0.21 to use Tags plugin (filters required)')
 111             self.player.disconnect()
 112             raise PluginException('MPD >= 0.21 required')
 113         # Check filter is valid
 114         try:
 115             if self.plugin_conf['filter']:
 116                 # Use window to limit response size
 117                 self.player.find(self.plugin_conf['filter'], "window", (0, 1))
 118         except CommandError as err:
 119             raise PluginException('Badly formated filter in tags plugin configuration: "%s"'
 120                                   % self.plugin_conf['filter']) from err
 121
 122     def callback_need_track(self):
 123         candidates = []
 124         queue_mode = self.plugin_conf.get('queue_mode', 'track')
 125         target = self.plugin_conf.getint(f'{queue_mode}_to_add')
 126         # look for artists acording to filter
 127         artists = [Artist(name=a) for a in self.player.list('artist', self.mpd_filter)]
 128         random.shuffle(artists)
 129         artists = MetaContainer(artists)
 130         if not artists:
 131             self.log.info('Tags plugin found nothing to queue')
 132             return candidates
 133         artists = self.get_reorg_artists_list(artists)
 134         self.log.debug('Tags plugin found: %s', ' / '.join(map(str, artists)))
 135         for artist in artists:
 136             self.log.debug('looking for %s', artist)
 137             tracks = self.player.find_tracks(artist)
 138             if not tracks:
 139                 continue
 140             trk = self.filter_track(tracks, candidates)
 141             if not trk:
 142                 continue
 143             if queue_mode == 'track':
 144                 self.log.info('Tags plugin chose: %s', trk)
 145                 candidates.append(trk)
 146                 if len(candidates) == target:
 147                     break
 148             else:
 149                 album = self.album_candidate(trk.Artist, unplayed=True)
 150                 if not album:
 151                     continue
 152                 candidates.extend(self.player.find_tracks(album))
 153                 if len({t.album for t in candidates}) == target:
 154                     break
 155         return candidates
 156
 157 # VIM MODLINE
 158 # vim: ai ts=4 sw=4 sts=4 expandtab