Formatting

This commit is contained in:
Firepup Sixfifty 2024-04-07 00:38:49 -05:00
parent ee5bee742f
commit 016e5d2595
Signed by: Firepup650
GPG key ID: 7C92E2ABBBFAB9BA
7 changed files with 87 additions and 52 deletions

42
bare.py
View file

@ -14,8 +14,7 @@ r = ...
handlers = ... handlers = ...
def mfind(message: str, find: list, usePrefix: bool = True) -> bool: def mfind(message: str, find: list, usePrefix: bool = True) -> bool: ...
...
class bot: class bot:
@ -38,41 +37,28 @@ class bot:
threads: list[str] threads: list[str]
lastfmLink: LastFMNetwork 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: ...
...

16
bot.py
View file

@ -36,7 +36,9 @@ class bot(bare.bot):
conf.servers[server]["admins"] if "admins" in conf.servers[server] else [] conf.servers[server]["admins"] if "admins" in conf.servers[server] else []
) )
self.ignores = ( 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.__version__ = conf.__version__
self.npallowed = conf.npallowed self.npallowed = conf.npallowed
@ -144,7 +146,9 @@ class bot(bare.bot):
elif code == 480: elif code == 480:
self.log(f"Joining {chan} failed (+S)", "WARN") self.log(f"Joining {chan} failed (+S)", "WARN")
if origin != "null": 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 break
elif code == 519: elif code == 519:
self.log(f"Joining {chan} failed (+A)", "WARN") self.log(f"Joining {chan} failed (+A)", "WARN")
@ -258,10 +262,14 @@ class bot(bare.bot):
if res == "reload" and type(chan) == str: if res == "reload" and type(chan) == str:
reload(conf) reload(conf)
self.adminnames = ( 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 = ( 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.__version__ = conf.__version__
self.npallowed = conf.npallowed self.npallowed = conf.npallowed

View file

@ -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("Firepup's master playlist", chan)
bot.msg("https://open.spotify.com/playlist/4ctNy3O0rOwhhXIKyLvUZM", chan) bot.msg("https://open.spotify.com/playlist/4ctNy3O0rOwhhXIKyLvUZM", chan)
def version(bot: bare.bot, chan: str, name: str, message: str) -> None: def version(bot: bare.bot, chan: str, name: str, message: str) -> None:
bot.msg("Version: " + bot.__version__, chan) bot.msg("Version: " + bot.__version__, chan)
def goat(bot: bare.bot, chan: str, name: str, message: str) -> None: def goat(bot: bare.bot, chan: str, name: str, message: str) -> None:
bot.log("GOAT DETECTED") bot.log("GOAT DETECTED")
bot.msg("Hello Goat", chan) bot.msg("Hello Goat", chan)
@ -105,7 +107,9 @@ def quote(bot: bare.bot, chan: str, name: str, message: str) -> None:
query = "" query = ""
if " " in message: if " " in message:
query = message.split(" ", 1)[1] query = message.split(" ", 1)[1]
qfilter = query.replace(" ", "\s") # pyright: ignore [reportInvalidStringEscapeSequence] qfilter = query.replace(
" ", "\s"
) # pyright: ignore [reportInvalidStringEscapeSequence]
r.seed() r.seed()
with open("mastermessages.txt", "r") as mm: with open("mastermessages.txt", "r") as mm:
q = list(filter(lambda x: re.search(qfilter, x), mm.readlines())) 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__, "VERSION": bot.__version__,
"NICKLEN": bot.nicklen, "NICKLEN": bot.nicklen,
"NICK": bot.nick, "NICK": bot.nick,
"ADMINS": str(bot.adminnames) "ADMINS": str(bot.adminnames) + " (Does not include hostname checks)",
+ " (Does not include hostname checks)",
"CHANNELS": bot.channels, "CHANNELS": bot.channels,
} }
bot.msg(f"[DEBUG] {dbg_out}", chan) 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: def fmpull(bot: bare.bot, chan: str, name: str, message: str) -> None:
try: 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: 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: 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]] = { 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}, "restart": {"prefix": True, "aliases": ["reboot", "stop"], "check": checks.admin},
"uptime": {"prefix": True, "aliases": []}, "uptime": {"prefix": True, "aliases": []},
"raw ": {"prefix": True, "aliases": ["cmd "], "check": checks.admin}, "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"]}, "8ball": {"prefix": True, "aliases": ["eightball", "8b"]},
"join ": {"prefix": True, "aliases": [], "check": checks.admin}, "join ": {"prefix": True, "aliases": [], "check": checks.admin},
"quote": {"prefix": True, "aliases": ["q"]}, "quote": {"prefix": True, "aliases": ["q"]},
"goat.mode.activate": {"prefix": True, "aliases": ["g.m.a"], "check": checks.admin}, "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": ["?"]}, "help": {"prefix": True, "aliases": ["?"]},
"amiadmin": {"prefix": True, "aliases": []}, "amiadmin": {"prefix": True, "aliases": []},
"ping": {"prefix": True, "aliases": []}, "ping": {"prefix": True, "aliases": []},
"op me": {"prefix": False, "aliases": [], "check": checks.admin}, "op me": {"prefix": False, "aliases": [], "check": checks.admin},
"whoami": {"prefix": True, "aliases": []}, "whoami": {"prefix": True, "aliases": []},
"fpmp": {"prefix": True, "aliases": []}, "fpmp": {"prefix": True, "aliases": []},
"version": {"prefix": True, "aliases": ["ver","v"]}, "version": {"prefix": True, "aliases": ["ver", "v"]},
"np": {"prefix": True, "aliases": []}, "np": {"prefix": True, "aliases": []},
} }
regexes: list[str] = [conf.npbase, conf.su] regexes: list[str] = [conf.npbase, conf.su]

View file

@ -7,7 +7,9 @@ import bare, pylast
load_dotenv() load_dotenv()
__version__ = "v3.0.2-dev" __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 .*))$" su = "^(su|sudo|(su .*|sudo .*))$"
servers: dict[str, dict[str, Any]] = { servers: dict[str, dict[str, Any]] = {
"ircnow": { "ircnow": {
@ -27,7 +29,16 @@ servers: dict[str, dict[str, Any]] = {
"replirc": { "replirc": {
"address": "127.0.0.1", "address": "127.0.0.1",
"pass": env["replirc_pass"], "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"], "ignores": ["#fp-radio"],
"admins": ["h-tl"], "admins": ["h-tl"],
"hosts": ["owner.firepi"], "hosts": ["owner.firepi"],
@ -38,7 +49,11 @@ servers: dict[str, dict[str, Any]] = {
"port": 6607, "port": 6607,
"channels": {"#default": 0, "#botrebellion": 0, "#main/replirc": 0}, "channels": {"#default": 0, "#botrebellion": 0, "#main/replirc": 0},
"ignores": ["#main/replirc"], "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"] admin_hosts: list[str] = ["firepup.firepi", "47.221.227.180"]
@ -57,6 +72,7 @@ prefix = "."
lastfmLink = pylast.LastFMNetwork(env["FM_KEY"], env["FM_SECRET"]) lastfmLink = pylast.LastFMNetwork(env["FM_KEY"], env["FM_SECRET"])
npallowed: list[str] = ["FireBitBot"] npallowed: list[str] = ["FireBitBot"]
def decode_escapes(s: str) -> str: def decode_escapes(s: str) -> str:
def decode_match(match): def decode_match(match):
return codecs.decode(match.group(0), "unicode-escape") return codecs.decode(match.group(0), "unicode-escape")

View file

@ -5,6 +5,7 @@ from threading import Thread
from logs import log from logs import log
from timers import threadManager from timers import threadManager
def launch(server: str) -> None: def launch(server: str) -> None:
system(f"python3 -u ircbot.py {server}") system(f"python3 -u ircbot.py {server}")

View file

@ -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 handled and len(message.split("\x01")) == 3:
if not CTCP(bot, message): if not CTCP(bot, message):
kind = message.split("\x01")[1] 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( bot.msg(
f"\x01ACTION gets hit by {kind.split(' ', 2)[2]}\x01", f"\x01ACTION gets hit by {kind.split(' ', 2)[2]}\x01",
chan, chan,
@ -143,9 +143,7 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str]
bot.channels[chan] = 0 bot.channels[chan] = 0
with open("mastermessages.txt", "r") as mm: with open("mastermessages.txt", "r") as mm:
sel = conf.decode_escapes( sel = conf.decode_escapes(
r.sample(mm.readlines(), 1)[0] r.sample(mm.readlines(), 1)[0].replace("\\n", "").replace("\n", "")
.replace("\\n", "")
.replace("\n", "")
) )
bot.msg(f"[QUOTE] {sel}", chan) bot.msg(f"[QUOTE] {sel}", chan)
return None, None return None, None

View file

@ -8,6 +8,7 @@ from threading import Thread
from time import sleep from time import sleep
from traceback import format_exc from traceback import format_exc
def is_dead(thr: Thread) -> bool: def is_dead(thr: Thread) -> bool:
thr.join(timeout=0) thr.join(timeout=0)
return not thr.is_alive() return not thr.is_alive()
@ -45,7 +46,12 @@ def startThread(data: dict) -> Thread:
return t 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: if output:
log("Begin init of thread manager", mgr) log("Begin init of thread manager", mgr)
running = {} running = {}
@ -79,7 +85,9 @@ def radio(instance: bare.bot) -> NoReturn:
if thisTrack != lastTrack: if thisTrack != lastTrack:
lastTrack = thisTrack lastTrack = thisTrack
instance.msg("f.sp " + thisTrack, "#fp-radio") 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: except Exception:
Err = format_exc() Err = format_exc()
for line in Err.split("\n"): for line in Err.split("\n"):
@ -88,6 +96,7 @@ def radio(instance: bare.bot) -> NoReturn:
instance.log("Thread while loop broken", "FATAL") instance.log("Thread while loop broken", "FATAL")
exit(1) exit(1)
data: dict[str, dict[str, Any]] = { data: dict[str, dict[str, Any]] = {
"radio": {"noWrap": True, "func": radio, "args": []}, "radio": {"noWrap": True, "func": radio, "args": []},
} }