]> kaliko git repositories - mpd-sima.git/blob - sima/utils/utils.py
Use os.getenv instead of os.environ.get
[mpd-sima.git] / 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