aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/Control.py12
-rw-r--r--src/EventManager.py4
-rw-r--r--src/IRCBuffer.py6
-rw-r--r--src/IRCLine.py4
-rw-r--r--src/IRCServer.py6
-rw-r--r--src/IRCUser.py12
-rw-r--r--src/Logging.py2
-rw-r--r--src/ModuleManager.py11
-rw-r--r--src/utils/__init__.py28
-rw-r--r--src/utils/http.py4
10 files changed, 51 insertions, 38 deletions
diff --git a/src/Control.py b/src/Control.py
index f9dc3637..494d8ff1 100644
--- a/src/Control.py
+++ b/src/Control.py
@@ -12,7 +12,7 @@ class ControlClient(object):
def fileno(self) -> int:
return self._socket.fileno()
- def read_lines(self) -> typing.List[str]:
+ def read_lines(self) -> typing.Optional[typing.List[str]]:
try:
data = self._socket.recv(2048)
except:
@@ -45,11 +45,11 @@ class Control(PollSource.PollSource):
self._filename = filename
self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
- self._clients = {}
+ self._clients: typing.Dict[int, ControlClient] = {}
def _on_log(self, levelno: int, line: str):
for client in self._clients.values():
- if not client.log_level == None and client.log_level <= levelno:
+ if client.log_level is not None and client.log_level <= levelno:
self._send_action(client, "log", line)
def bind(self):
@@ -69,7 +69,7 @@ class Control(PollSource.PollSource):
elif fileno in self._clients:
client = self._clients[fileno]
lines = client.read_lines()
- if lines == None:
+ if lines is None:
client.disconnect()
del self._clients[fileno]
else:
@@ -109,7 +109,7 @@ class Control(PollSource.PollSource):
if not keepalive:
client.disconnect()
- def _send_action(self, client: ControlClient, action: str, data: str,
- id: int=None):
+ def _send_action(self, client: ControlClient, action: str,
+ data: typing.Optional[str], id: typing.Optional[str]=None):
client.write_line(
json.dumps({"action": action, "data": data, "id": id}))
diff --git a/src/EventManager.py b/src/EventManager.py
index 46ea3ed7..02a0e08f 100644
--- a/src/EventManager.py
+++ b/src/EventManager.py
@@ -37,8 +37,8 @@ class EventHook(object):
self.priority = priority
self.docstring = utils.parse.docstring(func.__doc__ or "")
- self._kwargs = {}
- self._multi_kwargs = {}
+ self._kwargs: typing.Dict[str, typing.Any] = {}
+ self._multi_kwargs: typing.Dict[str, typing.List[typing.Any]] = {}
for key, value in kwargs:
if key in self._multi_kwargs:
self._multi_kwargs[key].append(value)
diff --git a/src/IRCBuffer.py b/src/IRCBuffer.py
index a809fdc4..f30b13ca 100644
--- a/src/IRCBuffer.py
+++ b/src/IRCBuffer.py
@@ -67,7 +67,11 @@ class Buffer(object):
return None
def find_from(self, nickname: str) -> typing.Optional[BufferLine]:
- return (self.find_many_from(nickname, 1) or [None])[0]
+ lines = self.find_many_from(nickname, 1)
+ if lines:
+ return lines[0]
+ else:
+ return None
def find_many_from(self, nickname: str, max: int
) -> typing.List[BufferLine]:
nickname_lower = self.server.irc_lower(nickname)
diff --git a/src/IRCLine.py b/src/IRCLine.py
index 75c88215..269e1dc0 100644
--- a/src/IRCLine.py
+++ b/src/IRCLine.py
@@ -1,5 +1,5 @@
import datetime, typing, uuid
-from src import IRCObject, utils
+from src import EventManager, IRCObject, utils
# this should be 510 (RFC1459, 512 with \r\n) but a server BitBot uses is broken
LINE_MAX = 470
@@ -130,7 +130,7 @@ class ParsedLine(object):
def _newline_truncate(self, line: str) -> typing.Tuple[str, str]:
line, sep, overflow = line.partition("\n")
- return [line, overflow]
+ return (line, overflow)
def _line_max(self, hostmask: str, margin: int) -> int:
return LINE_MAX-len((":%s " % hostmask).encode("utf8"))-margin
def truncate(self, hostmask: str, margin: int=0) -> typing.Tuple[str, str]:
diff --git a/src/IRCServer.py b/src/IRCServer.py
index 7adde4eb..053a8bfa 100644
--- a/src/IRCServer.py
+++ b/src/IRCServer.py
@@ -166,7 +166,7 @@ class Server(IRCObject.Object):
def has_user(self, nickname: str) -> bool:
return self.irc_lower(nickname) in self.users
- def get_user(self, nickname: str, username: str=None, hostname: str=None,
+ def get_user(self, nickname: str, username: typing.Optional[str]=None, hostname: str=None,
create: bool=True) -> typing.Optional[IRCUser.User]:
new = False
if not self.has_user(nickname) and create:
@@ -178,9 +178,9 @@ class Server(IRCObject.Object):
user = self.users.get(self.irc_lower(nickname), None)
if user:
- if not username == None:
+ if username is not None:
user.username = username
- if not hostname == None:
+ if hostname is not None:
user.hostname = hostname
if new:
self.events.on("new.user").call(user=new_user, server=self)
diff --git a/src/IRCUser.py b/src/IRCUser.py
index 7ba334c5..7e6d60ac 100644
--- a/src/IRCUser.py
+++ b/src/IRCUser.py
@@ -11,11 +11,11 @@ class User(IRCObject.Object):
self.server = server
self.set_nickname(nickname)
self._id = id
- self.username = None
- self.hostname = None
- self.realname = None
+ self.username: typing.Optional[str] = None
+ self.hostname: typing.Optional[str] = None
+ self.realname: typing.Optional[str] = None
self.bot = bot
- self.channels = set([]) # type: typing.Set[IRCChannel.Channel]
+ self.channels: typing.Set[IRCChannel.Channel] = set([])
self.identified_account = None
self.identified_account_override = None
@@ -23,7 +23,7 @@ class User(IRCObject.Object):
self.identified_account_id = None
self.identified_account_id_override = None
self.away = False
- self.away_message = None # type: typing.Optional[str]
+ self.away_message: typing.Optional[str] = None
self.buffer = IRCBuffer.Buffer(bot, server)
@@ -32,7 +32,7 @@ class User(IRCObject.Object):
def __str__(self) -> str:
return self.nickname
- def hostmask(self) -> str:
+ def hostmask(self) -> typing.Optional[str]:
if self.nickname and self.username and self.hostname:
return "%s!%s@%s" % (self.nickname, self.username, self.hostname)
return None
diff --git a/src/Logging.py b/src/Logging.py
index a1cdb123..e883540a 100644
--- a/src/Logging.py
+++ b/src/Logging.py
@@ -24,6 +24,8 @@ class HookedHandler(logging.StreamHandler):
self._func(record.levelno, self.format(record))
class Log(object):
+ _hooks: typing.List[typing.Callable[[int, str], None]]
+
def __init__(self, to_file: bool, level: str, location: str):
self._hooks = []
diff --git a/src/ModuleManager.py b/src/ModuleManager.py
index 36221798..6497129b 100644
--- a/src/ModuleManager.py
+++ b/src/ModuleManager.py
@@ -1,4 +1,5 @@
-import enum, gc, glob, importlib, io, inspect, os, sys, typing, uuid
+import enum, gc, glob, importlib, importlib.util, io, inspect, os, sys
+import typing, uuid
from src import Config, EventManager, Exports, IRCBot, Logging, Timers, utils
class ModuleException(Exception):
@@ -155,7 +156,7 @@ class ModuleManager(object):
def _module_name(self, path: str) -> str:
return os.path.basename(path).rsplit(".py", 1)[0].lower()
- def _module_paths(self, name: str) -> str:
+ def _module_paths(self, name: str) -> typing.List[str]:
paths = []
for directory in self.directories:
paths.append(os.path.join(directory, name))
@@ -180,7 +181,7 @@ class ModuleManager(object):
return getattr(obj, magic) if hasattr(obj, magic) else default
def _check_hashflags(self, bot: "IRCBot.Bot", definition: ModuleDefinition
- ) -> bool:
+ ) -> None:
for hashflag, value in definition.hashflags:
if hashflag == "ignore":
# nope, ignore this module.
@@ -321,7 +322,7 @@ class ModuleManager(object):
def load_modules(self, bot: "IRCBot.Bot", whitelist: typing.List[str]=[],
blacklist: typing.List[str]=[]
- ) -> typing.Tuple[typing.List[str], typing.List[str]]:
+ ) -> None:
loadable, nonloadable = self._list_valid_modules(bot, whitelist, blacklist)
for definition in nonloadable:
@@ -388,7 +389,7 @@ class ModuleManager(object):
failed = (definition, e)
break
- if not failed == None:
+ if failed is not None:
for module in self.modules.values():
self._unload_module(module)
self.modules = old_modules
diff --git a/src/utils/__init__.py b/src/utils/__init__.py
index bbf11e73..834c6139 100644
--- a/src/utils/__init__.py
+++ b/src/utils/__init__.py
@@ -177,7 +177,7 @@ class BitBotMagic(object):
self._kwargs.append((key, value))
def get_hooks(self):
- hooks: typing.List[typing.Tuple[str, typing.List[str, typing.Any]]] = []
+ hooks: typing.List[typing.Tuple[str, typing.List[Tuple[str, typing.Any]]]] = []
for hook, kwargs in self._hooks:
hooks.append((hook, self._kwargs.copy()+list(kwargs.items())))
return hooks
@@ -254,8 +254,11 @@ class CaseInsensitiveDict(dict):
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: str):
- return dict.__contains__(self, key.lower())
+ 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)
@@ -296,7 +299,7 @@ class Setting(object):
SETTING_TRUE = ["true", "yes", "on", "y"]
SETTING_FALSE = ["false", "no", "off", "n"]
class BoolSetting(Setting):
- example = "on"
+ example: typing.Optional[str] = "on"
def parse(self, value: str) -> typing.Any:
value_lower = value.lower()
if value_lower in SETTING_TRUE:
@@ -306,7 +309,7 @@ class BoolSetting(Setting):
return None
class IntSetting(Setting):
- example = "10"
+ example: typing.Optional[str] = "10"
def parse(self, value: str) -> typing.Any:
if value == "0":
return 0
@@ -317,7 +320,7 @@ class IntSetting(Setting):
return None
class IntRangeSetting(IntSetting):
- example = None
+ example: typing.Optional[str] = None
def __init__(self, n_min: int, n_max: int, name: str, help: str=None,
example: str=None):
self._n_min = n_min
@@ -365,7 +368,7 @@ class FunctionSetting(Setting):
self._func = func
Setting.__init__(self, name, help, example)
if not format == None:
- self.format = format
+ self.format = format # type: ignore
def parse(self, value: str) -> typing.Any:
return self._func(value)
@@ -399,8 +402,11 @@ def deadline(seconds: int=10):
signal.signal(signal.SIGALRM, old_handler)
signal.setitimer(signal.ITIMER_REAL, old_seconds, 0)
-def deadline_process(func: typing.Callable[[], None], seconds: int=10):
- q = multiprocessing.Queue()
+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()])
@@ -414,7 +420,7 @@ def deadline_process(func: typing.Callable[[], None], seconds: int=10):
try:
success, out = q.get(block=True, timeout=seconds)
except queue.Empty:
- p.kill()
+ p.kill() # type: ignore # to make mypy pass on Python 3.6
deadlined = True
finally:
q.close()
@@ -425,4 +431,4 @@ def deadline_process(func: typing.Callable[[], None], seconds: int=10):
if success:
return out
else:
- raise out
+ raise out # type: ignore
diff --git a/src/utils/http.py b/src/utils/http.py
index 936d3118..f31da62c 100644
--- a/src/utils/http.py
+++ b/src/utils/http.py
@@ -162,13 +162,13 @@ def _find_encoding(soup: bs4.BeautifulSoup) -> typing.Optional[str]:
return None
def request(request_obj: typing.Union[str, Request], **kwargs) -> Response:
- if type(request_obj) == str:
+ if isinstance(request_obj, str):
request_obj = Request(request_obj, **kwargs)
return _request(request_obj)
def _request(request_obj: Request) -> Response:
- def _wrap():
+ def _wrap() -> Response:
headers = request_obj.get_headers()
response = requests.request(
request_obj.method,
generated by cgit v1.3.1-10-gc9f91 (git 2.54.0) at 2026-06-11 15:28:42 +0000