forked from iamkroot/erp-gcal
-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathutils.py
138 lines (108 loc) · 3.42 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import json
import re
from datetime import date, datetime, time, timedelta, tzinfo
from difflib import SequenceMatcher
from functools import lru_cache
import requests
import toml
def read_toml(path):
try:
with open(path) as f:
return toml.load(f)
except FileNotFoundError:
print(f"Missing toml file at {path}.")
quit()
except toml.TomlDecodeError as e:
print(f"Invalid TOML in file: {path}")
print(f"Error (probably) in line {e.lineno}.")
quit()
config = read_toml('config.toml')
def read_json(path):
try:
with open(path) as f:
return json.load(f)
except FileNotFoundError:
print(f"Missing json file at {path}.")
quit()
except json.JSONDecodeError:
print(f"Invalid json file at {path}.")
quit()
def write_json(file, data):
with open(file, 'w') as f:
json.dump(data, f, indent=4)
# @lru_cache()
# def get_weekday(isoweekday) -> date:
# today = date.today()
# return today - timedelta(days=today.weekday() + isoweekday - 1)
class IST(tzinfo):
def utcoffset(self, dt):
return timedelta(hours=5, minutes=30)
def dst(self, dt):
return timedelta(0)
@lru_cache()
def combine_dt(date_, time_=time(0, 0), tz=IST):
return datetime.combine(date_, time_, tz())
def take(n, iterable):
return (val for _, val in zip(range(n), iterable))
def retry_on_conn_error(func):
def wrapper(*args, **kwargs):
for _ in range(5):
try:
return func(*args, **kwargs)
except requests.exceptions.ConnectionError:
print("Connection Error. Retrying.")
continue
except KeyboardInterrupt:
print("Error")
quit()
else:
print("Connection Error! Maximum retries exceeded.")
quit()
return wrapper
def pprint_json(data):
print(json.dumps(data, indent=4))
def to_title(s, exceptions=('I', 'II', 'III', 'IV')):
"""Convert string to Title Case"""
word_list = re.split(' ', s)
final = [word_list[0].capitalize()]
for word in word_list[1:]:
if not word:
continue
if word.lower() in ('and', 'of', 'in', 'to', 'its'):
word = word.lower()
elif word not in exceptions:
word = word.capitalize()
final.append(word)
return " ".join(final)
def fuzzymatch_dicts(target, source, fields=None, thresh=0.8):
total = 0
num = 0
for k, v in target.items():
if fields and k not in fields:
continue
if not isinstance(v, str):
continue
orig = source.get(k)
if not orig:
continue
matcher = SequenceMatcher(None, v, orig)
total += matcher.ratio()
num += 1
try:
ratio = total / num
return ratio if ratio >= thresh else 0
except ZeroDivisionError:
return 0
def find_entity(entity, entities, fields, thresh=0.8):
max_ratio = 0
best_match = None
for e in entities:
ratio = fuzzymatch_dicts(entity, e, fields, thresh)
if max_ratio < ratio:
max_ratio = ratio
best_match = e
return best_match
@lru_cache
def get_weekday(weekday=0, date_=date.today(), future_date=False) -> date:
value = date_ + timedelta(weekday - date_.weekday() % 7)
return value + timedelta(7 * (future_date and date_ == value))