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