sima/utils/utils.py

   1 # -*- coding: utf-8 -*-
   2 #
   3 # Copyright (c) 2010, 2011, 2013, 2014, 2015, 2020 kaliko <kaliko@azylum.org>
   4 #
   5 #  This file is part of sima
   6 #
   7 #  sima is free software: you can redistribute it and/or modify
   8 #  it under the terms of the GNU General Public License as published by
   9 #  the Free Software Foundation, either version 3 of the License, or
  10 #  (at your option) any later version.
  11 #
  12 #  sima is distributed in the hope that it will be useful,
  13 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 #  GNU General Public License for more details.
  16 #
  17 #  You should have received a copy of the GNU General Public License
  18 #  along with sima.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20 #
  21 """generic tools and utilities for sima
  22 """
  23 # pylint: disable=C0111
  24
  25 import logging
  26 import traceback
  27 import sys
  28
  29 from argparse import ArgumentError, Action
  30 from base64 import b64decode as push
  31 from codecs import getencoder
  32 from datetime import datetime
  33 from os import environ, access, getcwd, W_OK, R_OK
  34 from os.path import dirname, isabs, join, normpath, exists, isdir, isfile
  35 from time import sleep
  36
  37
  38 def getws(dic):
  39     """
  40     Decode Obfuscated api key.
  41     Only preventing API keys harvesting over the network
  42     https://developer.echonest.com/forums/thread/105
  43     """
  44     aka = push(bytes(dic.get('apikey') + '=', 'utf-8'))
  45     aka = getencoder('rot-13')(str((aka), 'utf-8'))[0]
  46     dic.update({'apikey': aka})
  47
  48
  49 def get_mpd_environ():
  50     """
  51     Retrieve MPD env. var.
  52     """
  53     passwd = host = None
  54     mpd_host_env = environ.get('MPD_HOST')
  55     if mpd_host_env:
  56         # If password is set:
  57         # mpd_host_env = ['pass', 'host'] because MPD_HOST=pass@host
  58         mpd_host_env = mpd_host_env.split('@')
  59         mpd_host_env.reverse()
  60         host = mpd_host_env[0]
  61         if len(mpd_host_env) > 1 and mpd_host_env[1]:
  62             passwd = mpd_host_env[1]
  63     return (host, environ.get('MPD_PORT', None), passwd)
  64
  65
  66 def normalize_path(path):
  67     """Get absolute path
  68     """
  69     if not isabs(path):
  70         return normpath(join(getcwd(), path))
  71     return path
  72
  73
  74 def exception_log():
  75     """Log unknown exceptions"""
  76     log = logging.getLogger(__name__)
  77     log.error('Unhandled Exception!!!')
  78     log.error(''.join(traceback.format_exc()))
  79     log.info('Please report the previous message'
  80              ' along with some log entries right before the crash.')
  81     log.info('thanks for your help :)')
  82     log.info('Quiting now!')
  83     sys.exit(1)
  84
  85
  86 class SigHup(Exception):
  87     """SIGHUP raises this Exception"""
  88
  89
  90 # ArgParse Callbacks
  91 class Obsolete(Action):
  92     # pylint: disable=R0903
  93     """Deal with obsolete arguments
  94     """
  95     def __call__(self, parser, namespace, values, option_string=None):
  96         raise ArgumentError(self, 'obsolete argument')
  97
  98
  99 class FileAction(Action):
 100     """Generic class to inherit from for ArgParse action on file/dir
 101     """
 102     # pylint: disable=R0903
 103     def __call__(self, parser, namespace, values, option_string=None):
 104         self._file = normalize_path(values)
 105         self._dir = dirname(self._file)
 106         self.parser = parser
 107         self.checks()
 108         setattr(namespace, self.dest, self._file)
 109
 110     def checks(self):
 111         """control method
 112         """
 113         pass
 114
 115
 116 class Wfile(FileAction):
 117     # pylint: disable=R0903
 118     """Is file writable
 119     """
 120     def checks(self):
 121         if isdir(self._file):
 122             self.parser.error('need a file not a directory: {}'.format(self._file))
 123         if not exists(self._dir):
 124             #raise ArgumentError(self, '"{0}" does not exist'.format(self._dir))
 125             self.parser.error('directory does not exist: {0}'.format(self._dir))
 126         if not exists(self._file):
 127             # Is parent directory writable then
 128             if not access(self._dir, W_OK):
 129                 self.parser.error('no write access to "{0}"'.format(self._dir))
 130         else:
 131             if not access(self._file, W_OK):
 132                 self.parser.error('no write access to "{0}"'.format(self._file))
 133
 134
 135 class Rfile(FileAction):
 136     # pylint: disable=R0903
 137     """Is file readable
 138     """
 139     def checks(self):
 140         if not exists(self._file):
 141             self.parser.error('file does not exist: {0}'.format(self._file))
 142         if not isfile(self._file):
 143             self.parser.error('not a file: {0}'.format(self._file))
 144         if not access(self._file, R_OK):
 145             self.parser.error('no read access to "{0}"'.format(self._file))
 146
 147
 148 class Wdir(FileAction):
 149     # pylint: disable=R0903
 150     """Is directory writable
 151     """
 152     def checks(self):
 153         if not exists(self._file):
 154             self.parser.error('directory does not exist: {0}'.format(self._file))
 155         if not isdir(self._file):
 156             self.parser.error('not a directory: {0}'.format(self._file))
 157         if not access(self._file, W_OK):
 158             self.parser.error('no write access to "{0}"'.format(self._file))
 159
 160
 161 class Throttle:
 162     """throttle decorator"""
 163     def __init__(self, wait):
 164         self.wait = wait
 165         self.last_called = datetime.now()
 166
 167     def __call__(self, func):
 168         def wrapper(*args, **kwargs):
 169             while self.last_called + self.wait > datetime.now():
 170                 sleep(0.1)
 171             result = func(*args, **kwargs)
 172             self.last_called = datetime.now()
 173             return result
 174         return wrapper
 175
 176
 177 class MPDSimaException(Exception):
 178     """Generic MPD_sima Exception"""
 179
 180
 181 # http client exceptions (for webservices)
 182 class WSError(MPDSimaException):
 183     pass
 184
 185
 186 class WSNotFound(WSError):
 187     pass
 188
 189
 190 class WSTimeout(WSError):
 191     pass
 192
 193
 194 class WSHTTPError(WSError):
 195     pass
 196
 197
 198 class PluginException(MPDSimaException):
 199     pass
 200
 201 # VIM MODLINE
 202 # vim: ai ts=4 sw=4 sts=4 expandtab