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
139
140
141
142
143
144
145
146
|
import contextlib, enum, ipaddress, multiprocessing, os.path, queue, signal
import threading, typing
from . import cli, consts, datetime, decorators, io, irc, http, parse, security
from .decorators import export, hook, kwarg, spec
from .settings import (BoolSetting, FunctionSetting, IntRangeSetting,
IntSetting, OptionsSetting, sensitive_format, SensitiveSetting, Setting)
from .errors import (EventError, EventNotEnoughArgsError, EventResultsError,
EventUsageError)
class Direction(enum.Enum):
Send = 0
Recv = 1
def prevent_highlight(nickname: str) -> str:
return nickname[0]+"\u200c"+nickname[1:]
class MultiCheck(object):
def __init__(self,
requests: typing.List[typing.Tuple[str, typing.List[str]]]):
self._requests = requests
def to_multi(self):
return self
def requests(self):
return self._requests[:]
def __or__(self, other: "Check"):
return MultiCheck(self._requests+[(other.request, other.args)])
class Check(object):
def __init__(self, request: str, *args: str):
self.request = request
self.args = list(args)
def to_multi(self):
return MultiCheck([(self.request, self.args)])
def __or__(self, other: "Check"):
return MultiCheck([(self.request, self.args),
(other.request, other.args)])
TOP_10_CALLABLE = typing.Callable[[typing.Any], typing.Any]
def top_10(items: typing.Dict[typing.Any, typing.Any],
convert_key: TOP_10_CALLABLE=lambda x: x,
value_format: TOP_10_CALLABLE=lambda x: x):
top_10 = sorted(items.keys())
top_10 = sorted(top_10, key=items.get, reverse=True)[:10]
top_10_items = []
for key in top_10:
top_10_items.append("%s (%s)" % (convert_key(key),
value_format(items[key])))
return top_10_items
class CaseInsensitiveDict(dict):
def __init__(self, other: typing.Dict[str, typing.Any]):
dict.__init__(self, ((k.lower(), v) for k, v in other.items()))
def __getitem__(self, key: str) -> typing.Any:
return dict.__getitem__(self, key.lower())
def __setitem__(self, key: str, value: typing.Any) -> typing.Any:
return dict.__setitem__(self, key.lower(), value)
def __contains__(self, key: typing.Any) -> bool:
if isinstance(key, str):
return dict.__contains__(self, key.lower())
else:
raise TypeError("Expected string, not %r" % key)
def get(self, key: str, default: typing.Any=None):
return dict.get(self, key.lower(), default)
def is_ip(s: str) -> bool:
try:
ipaddress.ip_address(s)
except ValueError:
return False
return True
def is_main_thread() -> bool:
return threading.current_thread() is threading.main_thread()
class DeadlineExceededException(Exception):
pass
def _raise_deadline():
raise DeadlineExceededException()
@contextlib.contextmanager
def deadline(seconds: int=10):
old_handler = signal.signal(signal.SIGALRM,
lambda _1, _2: _raise_deadline())
old_seconds, _ = signal.setitimer(signal.ITIMER_REAL, seconds, 0)
try:
if not old_seconds == 0.0 and seconds > old_seconds:
raise ValueError(
"Deadline timeout larger than parent deadline (%s > %s)" %
(seconds, old_seconds))
yield
finally:
signal.signal(signal.SIGALRM, old_handler)
signal.setitimer(signal.ITIMER_REAL, old_seconds, 0)
DeadlineProcessReturnType = typing.TypeVar("DeadlineProcessReturnType")
def deadline_process(func: typing.Callable[[], DeadlineProcessReturnType],
seconds: int=10) -> DeadlineProcessReturnType:
q: multiprocessing.Queue[
typing.Tuple[bool, DeadlineProcessReturnType]] = multiprocessing.Queue()
def _wrap(func, q):
try:
q.put([True, func()])
except Exception as e:
q.put([False, e])
p = multiprocessing.Process(target=_wrap, args=(func, q))
p.start()
deadlined = False
try:
success, out = q.get(block=True, timeout=seconds)
except queue.Empty:
p.kill() # type: ignore # to make mypy pass on Python 3.6
deadlined = True
finally:
q.close()
if deadlined:
_raise_deadline()
if success:
return out
else:
raise out # type: ignore
def git_commit(bot_directory: str
) -> typing.Tuple[typing.Optional[str], typing.Optional[str]]:
git_dir = os.path.join(bot_directory, ".git")
head_filepath = os.path.join(git_dir, "HEAD")
if os.path.isfile(head_filepath):
with open(head_filepath, "r") as head_file:
ref_line = head_file.readline().strip()
if not ref_line.startswith("ref: "):
return None, ref_line
else:
ref = ref_line.split(" ", 1)[1]
branch = ref.rsplit("/", 1)[1]
ref_filepath = os.path.join(git_dir, ref)
with open(ref_filepath, "r") as ref_file:
return branch, ref_file.readline().strip()[:8]
return None, None
|
