-
Notifications
You must be signed in to change notification settings - Fork 0
/
utils.py
117 lines (93 loc) · 3.75 KB
/
utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
import logging
from base64 import b64encode
from gzip import open as gzopen
from io import StringIO
from logging.handlers import TimedRotatingFileHandler
from os import path, makedirs, rename, remove, replace, walk
from urllib.parse import quote as url_quote
import yaml
class Rutils:
def verify_config(self, config):
return all(self.cfg[config].values())
def load_config(self, config="config_lab.yml"):
with open(config, 'r') as stream:
try:
self.cfg = yaml.safe_load(stream)
except yaml.YAMLError as exc:
print(exc)
def __init__(self):
self.cfg = None
@staticmethod
def create_file_path(folder: str, file_name: str):
TOP_DIR = path.dirname(path.abspath(__file__))
allowed_exts = ['csv', 'log', 'yaml','pfx','pem']
input_ext = '.'.join(file_name.split(".")[1:])
if input_ext.lower() not in allowed_exts:
raise ValueError(f'please ensure you using one of the allowed file types you gave {input_ext}')
fName = f'{TOP_DIR}/{folder}/{file_name}'
if not path.exists(f'{TOP_DIR}/{folder}'):
makedirs(f'{TOP_DIR}/{folder}')
# move file to correct dir if needed
if not path.exists(fName):
try:
replace(f'{TOP_DIR}/{file_name}', fName)
except:
# file has yet to be created or not in top path
pass
return fName
@staticmethod
def encode_data(data, base64=True):
if base64:
return b64encode(str.encode(data)).decode('utf-8')
else:
return url_quote(data, safe='')
@staticmethod
def get_yaml_config(config, self_instance):
if isinstance(config, str):
with open(config, 'r') as stream:
try:
return yaml.safe_load(stream)
except yaml.YAMLError as exc:
self_instance.logger.info(f'Error processing config file. Error recevied {exc}')
@staticmethod
def get_files_from_loc(fold_loc:str,files_to_look_for:list):
_, _, filenames = next(walk(fold_loc))
fnames = [names for names in filenames for filames in files_to_look_for if filames in names]
return fnames
@staticmethod
def df_to_string_buffer(df):
with StringIO() as buffer:
df.to_csv(buffer,index=False)
return buffer.getvalue()
def log_collector(log_all=False,file_name='log.txt',func_name='func_name'):
fName = Rutils().create_file_path('Logging', file_name)
if not log_all:
logger = logging.getLogger(func_name)
else:
logger = logging.getLogger()
import http.client as http_client
http_client.HTTPConnection.debuglevel = 1
logger.setLevel(logging.DEBUG)
if logger.hasHandlers():
logger.handlers = []
conHandler = logging.StreamHandler()
conHandler.setLevel(logging.INFO)
logformatCon = logging.Formatter('%(asctime)s %(levelname)s %(message)s', datefmt='%d-%b-%y %H:%M:%S')
conHandler.setFormatter(logformatCon)
logger.addHandler(conHandler)
fileHandler = TimedRotatingFileHandler(filename=fName, when='midnight', backupCount=90, interval=1)
fileHandler.setLevel(logging.DEBUG)
logformatfile = logging.Formatter('%(asctime)s %(name)s %(levelname)s %(message)s', datefmt='%d-%b-%y %H:%M:%S')
fileHandler.setFormatter(logformatfile)
fileHandler.rotator = GZipRotator()
logger.addHandler(fileHandler)
return logger
class GZipRotator:
def __call__(self, source, dest):
rename(source, dest)
f_in = open(dest, 'rb')
f_out = gzopen("{}.gz".format(dest), 'wb')
f_out.writelines(f_in)
f_out.close()
f_in.close()
remove(dest)