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
 114
 115 class Wfile(FileAction):
 116     # pylint: disable=R0903
 117     """Is file writable
 118     """
 119     def checks(self):
 120         if isdir(self._file):
 121             self.parser.error('need a file not a directory: {}'.format(self._file))
 122         if not exists(self._dir):
 123             #raise ArgumentError(self, '"{0}" does not exist'.format(self._dir))
 124             self.parser.error('directory does not exist: {0}'.format(self._dir))
 125         if not exists(self._file):
 126             # Is parent directory writable then
 127             if not access(self._dir, W_OK):
 128                 self.parser.error('no write access to "{0}"'.format(self._dir))
 129         else:
 130             if not access(self._file, W_OK):
 131                 self.parser.error('no write access to "{0}"'.format(self._file))
 132
 133
 134 class Rfile(FileAction):
 135     # pylint: disable=R0903
 136     """Is file readable
 137     """
 138     def checks(self):
 139         if not exists(self._file):
 140             self.parser.error('file does not exist: {0}'.format(self._file))
 141         if not isfile(self._file):
 142             self.parser.error('not a file: {0}'.format(self._file))
 143         if not access(self._file, R_OK):
 144             self.parser.error('no read access to "{0}"'.format(self._file))
 145
 146
 147 class Wdir(FileAction):
 148     # pylint: disable=R0903
 149     """Is directory writable
 150     """
 151     def checks(self):
 152         if not exists(self._file):
 153             self.parser.error('directory does not exist: {0}'.format(self._file))
 154         if not isdir(self._file):
 155             self.parser.error('not a directory: {0}'.format(self._file))
 156         if not access(self._file, W_OK):
 157             self.parser.error('no write access to "{0}"'.format(self._file))
 158
 159
 160 class Throttle:
 161     """throttle decorator"""
 162     def __init__(self, wait):
 163         self.wait = wait
 164         self.last_called = datetime.now()
 165
 166     def __call__(self, func):
 167         def wrapper(*args, **kwargs):
 168             while self.last_called + self.wait > datetime.now():
 169                 sleep(0.1)
 170             result = func(*args, **kwargs)
 171             self.last_called = datetime.now()
 172             return result
 173         return wrapper
 174
 175
 176 class MPDSimaException(Exception):
 177     """Generic MPD_sima Exception"""
 178
 179
 180 # http client exceptions (for webservices)
 181 class WSError(MPDSimaException):
 182     pass
 183
 184
 185 class WSNotFound(WSError):
 186     pass
 187
 188
 189 class WSTimeout(WSError):
 190     pass
 191
 192
 193 class WSHTTPError(WSError):
 194     pass
 195
 196
 197 class PluginException(MPDSimaException):
 198     pass
 199
 200 # VIM MODLINE
 201 # vim: ai ts=4 sw=4 sts=4 expandtab