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 def get_mpd_environ():
  49     """
  50     Retrieve MPD env. var.
  51     """
  52     passwd = host = None
  53     mpd_host_env = environ.get('MPD_HOST')
  54     if mpd_host_env:
  55         # If password is set:
  56         # mpd_host_env = ['pass', 'host'] because MPD_HOST=pass@host
  57         mpd_host_env = mpd_host_env.split('@')
  58         mpd_host_env.reverse()
  59         host = mpd_host_env[0]
  60         if len(mpd_host_env) > 1 and mpd_host_env[1]:
  61             passwd = mpd_host_env[1]
  62     return (host, environ.get('MPD_PORT', None), passwd)
  63
  64 def normalize_path(path):
  65     """Get absolute path
  66     """
  67     if not isabs(path):
  68         return normpath(join(getcwd(), path))
  69     return path
  70
  71 def exception_log():
  72     """Log unknown exceptions"""
  73     log = logging.getLogger(__name__)
  74     log.error('Unhandled Exception!!!')
  75     log.error(''.join(traceback.format_exc()))
  76     log.info('Please report the previous message'
  77              ' along with some log entries right before the crash.')
  78     log.info('thanks for your help :)')
  79     log.info('Quiting now!')
  80     sys.exit(1)
  81
  82
  83 class SigHup(Exception):
  84     """SIGHUP raises this Exception"""
  85     pass
  86
  87 # ArgParse Callbacks
  88 class Obsolete(Action):
  89     # pylint: disable=R0903
  90     """Deal with obsolete arguments
  91     """
  92     def __call__(self, parser, namespace, values, option_string=None):
  93         raise ArgumentError(self, 'obsolete argument')
  94
  95 class FileAction(Action):
  96     """Generic class to inherit from for ArgParse action on file/dir
  97     """
  98     # pylint: disable=R0903
  99     def __call__(self, parser, namespace, values, option_string=None):
 100         self._file = normalize_path(values)
 101         self._dir = dirname(self._file)
 102         self.parser = parser
 103         self.checks()
 104         setattr(namespace, self.dest, self._file)
 105
 106     def checks(self):
 107         """control method
 108         """
 109         pass
 110
 111 class Wfile(FileAction):
 112     # pylint: disable=R0903
 113     """Is file writable
 114     """
 115     def checks(self):
 116         if isdir(self._file):
 117             self.parser.error('need a file not a directory: {}'.format(self._file))
 118         if not exists(self._dir):
 119             #raise ArgumentError(self, '"{0}" does not exist'.format(self._dir))
 120             self.parser.error('directory does not exist: {0}'.format(self._dir))
 121         if not exists(self._file):
 122             # Is parent directory writable then
 123             if not access(self._dir, W_OK):
 124                 self.parser.error('no write access to "{0}"'.format(self._dir))
 125         else:
 126             if not access(self._file, W_OK):
 127                 self.parser.error('no write access to "{0}"'.format(self._file))
 128
 129 class Rfile(FileAction):
 130     # pylint: disable=R0903
 131     """Is file readable
 132     """
 133     def checks(self):
 134         if not exists(self._file):
 135             self.parser.error('file does not exist: {0}'.format(self._file))
 136         if not isfile(self._file):
 137             self.parser.error('not a file: {0}'.format(self._file))
 138         if not access(self._file, R_OK):
 139             self.parser.error('no read access to "{0}"'.format(self._file))
 140
 141 class Wdir(FileAction):
 142     # pylint: disable=R0903
 143     """Is directory writable
 144     """
 145     def checks(self):
 146         if not exists(self._file):
 147             self.parser.error('directory does not exist: {0}'.format(self._file))
 148         if not isdir(self._file):
 149             self.parser.error('not a directory: {0}'.format(self._file))
 150         if not access(self._file, W_OK):
 151             self.parser.error('no write access to "{0}"'.format(self._file))
 152
 153 class Throttle:
 154     """throttle decorator"""
 155     def __init__(self, wait):
 156         self.wait = wait
 157         self.last_called = datetime.now()
 158
 159     def __call__(self, func):
 160         def wrapper(*args, **kwargs):
 161             while self.last_called + self.wait > datetime.now():
 162                 sleep(0.1)
 163             result = func(*args, **kwargs)
 164             self.last_called = datetime.now()
 165             return result
 166         return wrapper
 167
 168 class MPDSimaException(Exception):
 169     pass
 170
 171 # http client exceptions (for webservices)
 172 class WSError(MPDSimaException):
 173     pass
 174
 175 class WSNotFound(WSError):
 176     pass
 177
 178 class WSTimeout(WSError):
 179     pass
 180
 181 class WSHTTPError(WSError):
 182     pass
 183
 184 class PluginException(MPDSimaException):
 185     pass
 186
 187 # VIM MODLINE
 188 # vim: ai ts=4 sw=4 sts=4 expandtab