diff --git a/bare.py b/bare.py index 9d11e23..3dfbae9 100644 --- a/bare.py +++ b/bare.py @@ -14,8 +14,7 @@ r = ... handlers = ... -def mfind(message: str, find: list, usePrefix: bool = True) -> bool: - ... +def mfind(message: str, find: list, usePrefix: bool = True) -> bool: ... class bot: @@ -38,41 +37,28 @@ class bot: threads: list[str] lastfmLink: LastFMNetwork - def __init__(self, server: str): - ... + def __init__(self, server: str): ... - def connect(self) -> None: - ... + def connect(self) -> None: ... - def join(self, chan: str, origin: str, lock: bool = True) -> None: - ... + def join(self, chan: str, origin: str, lock: bool = True) -> None: ... - def ping(self, ircmsg: str) -> int: - ... + def ping(self, ircmsg: str) -> int: ... - def send(self, command: str) -> int: - ... + def send(self, command: str) -> int: ... - def recv(self) -> bytes: - ... + def recv(self) -> bytes: ... - def log(self, message: str, level: str = "LOG") -> None: - ... + def log(self, message: str, level: str = "LOG") -> None: ... - def exit(self, message: str) -> NoReturn: - ... + def exit(self, message: str) -> NoReturn: ... - def msg(self, msg: str, target: str) -> None: - ... + def msg(self, msg: str, target: str) -> None: ... - def op(self, name: str, chan: str) -> Union[int, None]: - ... + def op(self, name: str, chan: str) -> Union[int, None]: ... - def notice(self, msg: str, target: str, silent: bool = False) -> int: - ... + def notice(self, msg: str, target: str, silent: bool = False) -> int: ... - def sendraw(self, command: str) -> int: - ... + def sendraw(self, command: str) -> int: ... - def mainloop(self) -> NoReturn: - ... + def mainloop(self) -> NoReturn: ... diff --git a/bot.py b/bot.py index 23ef648..2c1e7a2 100644 --- a/bot.py +++ b/bot.py @@ -36,7 +36,9 @@ class bot(bare.bot): conf.servers[server]["admins"] if "admins" in conf.servers[server] else [] ) self.ignores = ( - conf.servers[server]["ignores"] if "ignores" in conf.servers[server] else 6667 + conf.servers[server]["ignores"] + if "ignores" in conf.servers[server] + else 6667 ) self.__version__ = conf.__version__ self.npallowed = conf.npallowed @@ -144,7 +146,9 @@ class bot(bare.bot): elif code == 480: self.log(f"Joining {chan} failed (+S)", "WARN") if origin != "null": - self.msg(f"{chan} is +S, and I'm not connected over SSL.", origin) + self.msg( + f"{chan} is +S, and I'm not connected over SSL.", origin + ) break elif code == 519: self.log(f"Joining {chan} failed (+A)", "WARN") @@ -258,11 +262,15 @@ class bot(bare.bot): if res == "reload" and type(chan) == str: reload(conf) self.adminnames = ( - conf.servers[self.server]["admins"] if "admins" in conf.servers[self.server] else [] - ) + conf.servers[self.server]["admins"] + if "admins" in conf.servers[self.server] + else [] + ) self.ignores = ( - conf.servers[self.server]["ignores"] if "ignores" in conf.servers[self.server] else [] - ) + conf.servers[self.server]["ignores"] + if "ignores" in conf.servers[self.server] + else [] + ) self.__version__ = conf.__version__ self.npallowed = conf.npallowed self.interval = ( diff --git a/commands.py b/commands.py index 3fd9192..ce1596a 100644 --- a/commands.py +++ b/commands.py @@ -10,9 +10,11 @@ def fpmp(bot: bare.bot, chan: str, name: str, message: str) -> None: bot.msg("Firepup's master playlist", chan) bot.msg("https://open.spotify.com/playlist/4ctNy3O0rOwhhXIKyLvUZM", chan) + def version(bot: bare.bot, chan: str, name: str, message: str) -> None: bot.msg("Version: " + bot.__version__, chan) + def goat(bot: bare.bot, chan: str, name: str, message: str) -> None: bot.log("GOAT DETECTED") bot.msg("Hello Goat", chan) @@ -105,7 +107,9 @@ def quote(bot: bare.bot, chan: str, name: str, message: str) -> None: query = "" if " " in message: query = message.split(" ", 1)[1] - qfilter = query.replace(" ", "\s") # pyright: ignore [reportInvalidStringEscapeSequence] + qfilter = query.replace( + " ", "\s" + ) # pyright: ignore [reportInvalidStringEscapeSequence] r.seed() with open("mastermessages.txt", "r") as mm: q = list(filter(lambda x: re.search(qfilter, x), mm.readlines())) @@ -138,8 +142,7 @@ def debug(bot: bare.bot, chan: str, name: str, message: str) -> None: "VERSION": bot.__version__, "NICKLEN": bot.nicklen, "NICK": bot.nick, - "ADMINS": str(bot.adminnames) - + " (Does not include hostname checks)", + "ADMINS": str(bot.adminnames) + " (Does not include hostname checks)", "CHANNELS": bot.channels, } bot.msg(f"[DEBUG] {dbg_out}", chan) @@ -174,12 +177,22 @@ def nowplaying(bot: bare.bot, chan: str, name: str, message: str) -> None: def fmpull(bot: bare.bot, chan: str, name: str, message: str) -> None: try: - bot.msg("Firepup is currently listening to: " + bot.lastfmLink.get_user("Firepup650").get_now_playing().__str__(), chan) + bot.msg( + "Firepup is currently listening to: " + + bot.lastfmLink.get_user("Firepup650").get_now_playing().__str__(), + chan, + ) except: - bot.msg("Sorry, the music api isn't cooperating, please try again in a minute", chan) + bot.msg( + "Sorry, the music api isn't cooperating, please try again in a minute", chan + ) + def whoami(bot: bare.bot, chan: str, name: str, message: str) -> None: - bot.msg(f"I think you are {name}{' (bridge)' if bot.current == 'bridge' else '@{bot.tmpHost}'}", chan) + bot.msg( + f"I think you are {name}{' (bridge)' if bot.current == 'bridge' else '@{bot.tmpHost}'}", + chan, + ) data: dict[str, dict[str, Any]] = { @@ -190,19 +203,23 @@ data: dict[str, dict[str, Any]] = { "restart": {"prefix": True, "aliases": ["reboot", "stop"], "check": checks.admin}, "uptime": {"prefix": True, "aliases": []}, "raw ": {"prefix": True, "aliases": ["cmd "], "check": checks.admin}, - "debug": {"prefix": True, "aliases": ["dbg","d"], "check": checks.admin}, + "debug": {"prefix": True, "aliases": ["dbg", "d"], "check": checks.admin}, "8ball": {"prefix": True, "aliases": ["eightball", "8b"]}, "join ": {"prefix": True, "aliases": [], "check": checks.admin}, "quote": {"prefix": True, "aliases": ["q"]}, "goat.mode.activate": {"prefix": True, "aliases": ["g.m.a"], "check": checks.admin}, - "goat.mode.deactivate": {"prefix": True, "aliases": ["g.m.d"], "check": checks.admin}, + "goat.mode.deactivate": { + "prefix": True, + "aliases": ["g.m.d"], + "check": checks.admin, + }, "help": {"prefix": True, "aliases": ["?"]}, "amiadmin": {"prefix": True, "aliases": []}, "ping": {"prefix": True, "aliases": []}, "op me": {"prefix": False, "aliases": [], "check": checks.admin}, "whoami": {"prefix": True, "aliases": []}, "fpmp": {"prefix": True, "aliases": []}, - "version": {"prefix": True, "aliases": ["ver","v"]}, + "version": {"prefix": True, "aliases": ["ver", "v"]}, "np": {"prefix": True, "aliases": []}, } regexes: list[str] = [conf.npbase, conf.su] diff --git a/config.py b/config.py index 9a0cffb..2f77757 100644 --- a/config.py +++ b/config.py @@ -7,7 +7,9 @@ import bare, pylast load_dotenv() __version__ = "v3.0.2-dev" -npbase: str = "\[\x0303last\.fm\x03\] [A-Za-z0-9_[\]{}\\|\-^]{1,$MAX} (is listening|last listened) to: \x02.+ - .*\x02( \([0-9]+ plays\)( \[.*\])?)?" # pyright: ignore [reportInvalidStringEscapeSequence] +npbase: str = ( + "\[\x0303last\.fm\x03\] [A-Za-z0-9_[\]{}\\|\-^]{1,$MAX} (is listening|last listened) to: \x02.+ - .*\x02( \([0-9]+ plays\)( \[.*\])?)?" # pyright: ignore [reportInvalidStringEscapeSequence] +) su = "^(su|sudo|(su .*|sudo .*))$" servers: dict[str, dict[str, Any]] = { "ircnow": { @@ -27,7 +29,16 @@ servers: dict[str, dict[str, Any]] = { "replirc": { "address": "127.0.0.1", "pass": env["replirc_pass"], - "channels": {"#random": 0, "#dice": 0, "#main": 0, "#bots": 0, "#firebot": 0, "#sshchat": 0, "#firemc": 0, "#fp-radio": 0}, + "channels": { + "#random": 0, + "#dice": 0, + "#main": 0, + "#bots": 0, + "#firebot": 0, + "#sshchat": 0, + "#firemc": 0, + "#fp-radio": 0, + }, "ignores": ["#fp-radio"], "admins": ["h-tl"], "hosts": ["owner.firepi"], @@ -38,7 +49,11 @@ servers: dict[str, dict[str, Any]] = { "port": 6607, "channels": {"#default": 0, "#botrebellion": 0, "#main/replirc": 0}, "ignores": ["#main/replirc"], - "hosts": ["172.20.171.225", "169.254.253.107", "2600-6c5a-637f-1a85-0000-0000-0000-6667.inf6.spectrum.com"], + "hosts": [ + "172.20.171.225", + "169.254.253.107", + "2600-6c5a-637f-1a85-0000-0000-0000-6667.inf6.spectrum.com", + ], }, } admin_hosts: list[str] = ["firepup.firepi", "47.221.227.180"] @@ -57,6 +72,7 @@ prefix = "." lastfmLink = pylast.LastFMNetwork(env["FM_KEY"], env["FM_SECRET"]) npallowed: list[str] = ["FireBitBot"] + def decode_escapes(s: str) -> str: def decode_match(match): return codecs.decode(match.group(0), "unicode-escape") diff --git a/core.py b/core.py index 062abec..994bf93 100644 --- a/core.py +++ b/core.py @@ -5,6 +5,7 @@ from threading import Thread from logs import log from timers import threadManager + def launch(server: str) -> None: system(f"python3 -u ircbot.py {server}") diff --git a/handlers.py b/handlers.py index 1f39a79..16f6ddc 100644 --- a/handlers.py +++ b/handlers.py @@ -131,7 +131,7 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str] if not handled and len(message.split("\x01")) == 3: if not CTCP(bot, message): kind = message.split("\x01")[1] - if kind.startswith("ACTION ducks") and len(kind.split(' ', 2)) == 3: + if kind.startswith("ACTION ducks") and len(kind.split(" ", 2)) == 3: bot.msg( f"\x01ACTION gets hit by {kind.split(' ', 2)[2]}\x01", chan, @@ -143,9 +143,7 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str] bot.channels[chan] = 0 with open("mastermessages.txt", "r") as mm: sel = conf.decode_escapes( - r.sample(mm.readlines(), 1)[0] - .replace("\\n", "") - .replace("\n", "") + r.sample(mm.readlines(), 1)[0].replace("\\n", "").replace("\n", "") ) bot.msg(f"[QUOTE] {sel}", chan) return None, None diff --git a/timers.py b/timers.py index 0cdb408..ce13120 100644 --- a/timers.py +++ b/timers.py @@ -8,6 +8,7 @@ from threading import Thread from time import sleep from traceback import format_exc + def is_dead(thr: Thread) -> bool: thr.join(timeout=0) return not thr.is_alive() @@ -45,7 +46,12 @@ def startThread(data: dict) -> Thread: return t -def threadManager(threads: dict[str, dict[str, Any]], output: bool = False, mgr: str = "TManager", interval: int = 60) -> NoReturn: +def threadManager( + threads: dict[str, dict[str, Any]], + output: bool = False, + mgr: str = "TManager", + interval: int = 60, +) -> NoReturn: if output: log("Begin init of thread manager", mgr) running = {} @@ -79,7 +85,9 @@ def radio(instance: bare.bot) -> NoReturn: if thisTrack != lastTrack: lastTrack = thisTrack instance.msg("f.sp " + thisTrack, "#fp-radio") - instance.sendraw(f"TOPIC #fp-radio :Firepup radio ({thisTrack}) - https://open.spotify.com/playlist/4ctNy3O0rOwhhXIKyLvUZM") + instance.sendraw( + f"TOPIC #fp-radio :Firepup radio ({thisTrack}) - https://open.spotify.com/playlist/4ctNy3O0rOwhhXIKyLvUZM" + ) except Exception: Err = format_exc() for line in Err.split("\n"): @@ -88,6 +96,7 @@ def radio(instance: bare.bot) -> NoReturn: instance.log("Thread while loop broken", "FATAL") exit(1) + data: dict[str, dict[str, Any]] = { "radio": {"noWrap": True, "func": radio, "args": []}, }