Compare commits

..

No commits in common. "master" and "master" have entirely different histories.

13 changed files with 89 additions and 11061 deletions

View file

@ -3,7 +3,6 @@ from socket import socket
from overrides import bytes, bbytes from overrides import bytes, bbytes
from typing import NoReturn, Union from typing import NoReturn, Union
from pylast import LastFMNetwork from pylast import LastFMNetwork
from markov import MarkovBot
logs = ... logs = ...
re = ... re = ...
@ -40,12 +39,6 @@ class bot:
onIdntCmds: list[str] onIdntCmds: list[str]
onJoinCmds: list[str] onJoinCmds: list[str]
onStrtCmds: list[str] onStrtCmds: list[str]
markov: MarkovBot
autoMethod: str
dnsblMode: str
statuses: dict[str, dict[str, str]]
ops: dict[str, bool]
dns: dict[str, dict[str, Union[str, list[str]]]]
def __init__(self, server: str): ... def __init__(self, server: str): ...

57
bot.py
View file

@ -8,13 +8,11 @@ import commands as cmds
import config as conf import config as conf
from time import sleep from time import sleep
from importlib import reload from importlib import reload
import threads import timers
import random as r import random as r
import handlers import handlers
import bare import bare
from threading import Thread from threading import Thread
from markov import MarkovBot
from traceback import format_exc
def mfind(message: str, find: list, usePrefix: bool = True) -> bool: def mfind(message: str, find: list, usePrefix: bool = True) -> bool:
@ -53,8 +51,6 @@ class bot(bare.bot):
else "FireBot" else "FireBot"
) )
self.queue: list[bbytes] = [] # pyright: ignore [reportInvalidTypeForm] self.queue: list[bbytes] = [] # pyright: ignore [reportInvalidTypeForm]
self.statuses = {"firepup": {}}
self.ops = {}
self.sock = socket(AF_INET, SOCK_STREAM) self.sock = socket(AF_INET, SOCK_STREAM)
self.current = "user" self.current = "user"
self.threads = ( self.threads = (
@ -75,26 +71,7 @@ class bot(bare.bot):
if "onStrtCmds" in conf.servers[server] if "onStrtCmds" in conf.servers[server]
else [] else []
) )
self.autoMethod = (
conf.servers[server]["autoMethod"]
if "autoMethod" in conf.servers[server]
else "QUOTE"
)
self.dnsblMode = (
conf.servers[server]["dnsblMode"]
if "dnsblMode" in conf.servers[server]
else "none"
)
self.dns = {}
self.lastfmLink = conf.lastfmLink self.lastfmLink = conf.lastfmLink
with open("mastermessages.txt") as f:
TMFeed = []
for line in f.readlines():
TMFeed.extend([line.strip().split()])
self.markov = MarkovBot(TMFeed)
conf.prefix = (
conf.servers[server]["prefix"] if "prefix" in conf.servers[server] else "."
)
self.log(f"Start init for {self.server}") self.log(f"Start init for {self.server}")
def connect(self) -> None: def connect(self) -> None:
@ -191,11 +168,6 @@ class bot(bare.bot):
if origin != "null": if origin != "null":
self.msg(f"I'm banned from {chan}.", origin) self.msg(f"I'm banned from {chan}.", origin)
break break
elif code == 475:
self.log(f"Joining {chan} failed (+k without/with bad key)")
if origin != "null":
self.msg(f"{chan} is +k, and either you didn't give me a key, or you gave me the wrong one.", origin)
break
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":
@ -241,12 +213,7 @@ class bot(bare.bot):
def recv(self) -> bytes: def recv(self) -> bytes:
if self.queue: if self.queue:
return bytes(self.queue.pop(0)) return bytes(self.queue.pop(0))
data = bytes(self.sock.recv(2048)) data = bytes(self.sock.recv(2048).strip(b"\r\n"))
if data.lazy_decode() == "":
return data
while not data.endswith(b"\r\n"):
data += bytes(self.sock.recv(2048))
data = bytes(data.strip(b"\r\n"))
if b"\r\n" in data: if b"\r\n" in data:
self.queue.extend(data.split(b"\r\n")) self.queue.extend(data.split(b"\r\n"))
return bytes(self.queue.pop(0)) return bytes(self.queue.pop(0))
@ -302,10 +269,10 @@ class bot(bare.bot):
if self.threads: if self.threads:
tdict = {} tdict = {}
for thread in self.threads: for thread in self.threads:
tdict[thread] = threads.data[thread] tdict[thread] = timers.data[thread]
if tdict[thread]["passInstance"]: if thread in ["radio"]:
tdict[thread]["args"] = [self] tdict[thread]["args"] = [self]
tMgr = Thread(target=threads.threadManager, args=(tdict,)) tMgr = Thread(target=timers.threadManager, args=(tdict,))
tMgr.daemon = True tMgr.daemon = True
tMgr.start() tMgr.start()
while 1: while 1:
@ -324,7 +291,6 @@ class bot(bare.bot):
if action in handlers.handles: if action in handlers.handles:
res, chan = handlers.handles[action](self, ircmsg) res, chan = handlers.handles[action](self, ircmsg)
if res == "reload" and type(chan) == str: if res == "reload" and type(chan) == str:
try:
reload(conf) reload(conf)
self.adminnames = ( self.adminnames = (
conf.servers[self.server]["admins"] conf.servers[self.server]["admins"]
@ -343,22 +309,9 @@ class bot(bare.bot):
if "interval" in conf.servers[self.server] if "interval" in conf.servers[self.server]
else 50 else 50
) )
conf.prefix = (
conf.servers[self.server]["prefix"]
if "prefix" in conf.servers[self.server]
else "."
)
reload(cmds) reload(cmds)
reload(handlers) reload(handlers)
self.msg("Reloaded successfully", chan) self.msg("Reloaded successfully", chan)
except Exception:
Err = format_exc()
for line in Err.split("\n"):
self.log(line, "ERROR")
self.msg(
"Reload failed, likely partially reloaded. Please check error logs.",
chan,
)
else: else:
if ircmsg.startswith("PING "): if ircmsg.startswith("PING "):
self.ping(ircmsg) self.ping(ircmsg)

View file

@ -23,7 +23,7 @@ def goat(bot: bare.bot, chan: str, name: str, message: str) -> None:
def botlist(bot: bare.bot, chan: str, name: str, message: str) -> None: def botlist(bot: bare.bot, chan: str, name: str, message: str) -> None:
bot.msg( bot.msg(
f"Hi! I'm FireBot (https://git.h.hackclub.app/Firepup650/FireBot)! {'My admins on this server are' + str(bot.adminnames) + '.' if bot.adminnames else ''}", # pyright: ignore [reportOperatorIssue] f"Hi! I'm FireBot (https://git.amcforum.wiki/Firepup650/fire-ircbot)! {'My admins on this server are' + str(bot.adminnames) + '.' if bot.adminnames else ''}", # pyright: ignore [reportOperatorIssue]
chan, chan,
) )
@ -67,29 +67,27 @@ def isAdmin(bot: bare.bot, chan: str, name: str, message: str) -> None:
def help(bot: bare.bot, chan: str, name: str, message: str) -> None: def help(bot: bare.bot, chan: str, name: str, message: str) -> None:
helpErr = False helpErr = False
category = None
if len(message.split(" ")) > 1:
category = message.split(" ")[1]
if bot.current == "bridge": if bot.current == "bridge":
helpErr = True helpErr = True
if not helpErr: if not helpErr:
bot.msg("Command list needs rework", chan) bot.msg("Command list needs rework", name)
return return
match category: bot.msg("List of commands:", name)
case None: bot.msg(f'Current bot.prefix is "{bot.prefix}"', name)
bot.msg("Categories of commands: gen, dbg, adm, fun, msc", chan) bot.msg(f"{bot.prefix}help - Sends this help list", name)
case "gen": bot.msg(f"{bot.prefix}quote - Sends a random firepup quote", name)
bot.msg("Commands in the General category: ", chan) bot.msg(
case "dbg": f"{bot.prefix}(eightball,8ball,8b) [question]? - Asks the magic eightball a question",
bot.msg("Commands in the [DEBUG] category: ", chan) name,
case "adm": )
bot.msg("Commands in the Admin category: ", chan) bot.msg(f"(hi,hello) {bot.nick} - The bot says hi to you", name)
case "fun": if checks.admin(bot, name):
bot.msg("Commands in the Fun category: ", chan) bot.msg(f"reboot {bot.rebt} - Restarts the bot", name)
case "msc": bot.msg("op me - Makes the bot try to op you", name)
bot.msg("Commands in the Misc. category: ", chan) bot.msg(
case _: f"{bot.prefix}join [channel(s)] - Joins the bot to the specified channel(s)",
bot.msg("Unknown commands category.", chan) name,
)
else: else:
bot.msg("Sorry, I can't send help to bridged users.", chan) bot.msg("Sorry, I can't send help to bridged users.", chan)
@ -114,11 +112,7 @@ def quote(bot: bare.bot, chan: str, name: str, message: str) -> None:
) # pyright: ignore [reportInvalidStringEscapeSequence] ) # pyright: ignore [reportInvalidStringEscapeSequence]
r.seed() r.seed()
with open("mastermessages.txt", "r") as mm: with open("mastermessages.txt", "r") as mm:
q = []
try:
q = list(filter(lambda x: re.search(qfilter, x), mm.readlines())) q = list(filter(lambda x: re.search(qfilter, x), mm.readlines()))
except re.error:
q = ["Sorry, your query is invalid regex. Please try again."]
if q == []: if q == []:
q = [f'No results for "{query}" '] q = [f'No results for "{query}" ']
sel = conf.decode_escapes( sel = conf.decode_escapes(
@ -154,26 +148,6 @@ def debug(bot: bare.bot, chan: str, name: str, message: str) -> None:
bot.msg(f"[DEBUG] {dbg_out}", chan) bot.msg(f"[DEBUG] {dbg_out}", chan)
def debugInternal(bot: bare.bot, chan: str, name: str, message: str) -> None:
things = dir(bot)
try:
thing = message.split(" ", 1)[1]
except IndexError:
bot.msg("You can't just ask me to lookup nothing.", chan)
return
if thing in things:
bot.msg(f"self.{thing} = {getattr(bot, thing)}", chan)
else:
bot.msg(f'I have nothing called "{thing}"', chan)
def debugEval(bot: bare.bot, chan: str, name: str, message: str) -> None:
try:
bot.msg(str(eval(message.split(" ", 1)[1])), chan)
except Exception as E:
bot.msg(f"Exception: {E}", chan)
def raw(bot: bare.bot, chan: str, name: str, message: str) -> None: def raw(bot: bare.bot, chan: str, name: str, message: str) -> None:
bot.sendraw(message.split(" ", 1)[1]) bot.sendraw(message.split(" ", 1)[1])
@ -185,11 +159,11 @@ def reboot(bot: bare.bot, chan: str, name: str, message: str) -> None:
def sudo(bot: bare.bot, chan: str, name: str, message: str) -> None: def sudo(bot: bare.bot, chan: str, name: str, message: str) -> None:
if checks.admin(bot, name): if checks.admin(bot, name):
bot.msg("Operation not permitted", chan) bot.msg("Error - system failure, contact system operator", chan)
elif "bot" in name.lower(): elif "bot" in name.lower():
bot.log("lol, no.") bot.log("lol, no.")
else: else:
bot.msg('sudo: a password is required', chan) bot.msg("Access Denied", chan)
def nowplaying(bot: bare.bot, chan: str, name: str, message: str) -> None: def nowplaying(bot: bare.bot, chan: str, name: str, message: str) -> None:
@ -225,157 +199,6 @@ def whoami(bot: bare.bot, chan: str, name: str, message: str) -> None:
) )
def markov(bot: bare.bot, chan: str, name: str, message: str) -> None:
word = None
if " " in message:
word = message.split()[1]
proposed = bot.markov.generate_text(word)
if proposed == word:
proposed = f'Chain failed. (Firepup has never been recorded saying "{word}")'
bot.msg(proposed, chan)
def setStatus(bot: bare.bot, chan: str, name: str, message: str) -> None:
user, stat, reas = ("", 0, "")
try:
if message.split(" ")[1] == "help":
bot.msg(
"Assuming you want help with status codes. 1 is Available, 2 is Busy, 3 is Unavailable, anything else is Unknown.",
chan,
)
return
message = message.split(" ", 1)[1]
user = message.split(" ")[0].lower()
stat = int(message.split(" ")[1])
reas = message.split(" ", 2)[2]
except IndexError:
bot.msg(
f"Insufficent information to set a status. Only got {len(message.split(' ')) - (1 if '.sS' in message else 0)}/3 expected args.",
chan,
)
return
except ValueError:
bot.msg("Status parameter must be an int.", chan)
return
match stat:
case 1:
stat = "Available"
case 2:
stat = "Busy"
case 3:
stat = "Unavailable"
case _:
stat = "Unknown"
if user in ["me", "my", "I"]:
user = "firepup"
bot.statuses[user] = {"status": stat, "reason": reas}
bot.msg(f"Status set for '{user}'. Raw data: {bot.statuses[user]}", chan)
def getStatus(bot: bare.bot, chan: str, name: str, message: str) -> None:
user = ""
try:
user = message.split(" ")[1]
except IndexError:
user = "firepup"
if bot.statuses.get(user) is None:
bot.msg("You've gotta provide a nick I actually recognize.", chan)
return
bot.msg(
f"{user}'s status: {'Unknown' if not bot.statuses[user].get('status') else bot.statuses[user]['status']} - {'Reason unset' if not bot.statuses[user].get('reason') else bot.statuses[user]['reason']}",
chan,
)
def check(bot: bare.bot, chan: str, name: str, message: str) -> None:
try:
msg = message.split(" ", 1)[1]
nick = msg.split("!")[0]
host = msg.split("@", 1)[1]
cache = host in bot.dns
dnsbl, raws = conf.dnsblHandler(bot, nick, host, chan)
bot.msg(
f"Blacklist check: {'(Cached) ' if cache else ''}{dnsbl if dnsbl else 'Safe.'} ({raws})",
chan,
)
except IndexError:
try:
host = message.split(" ", 1)[1]
cache = host in bot.dns
dnsbl, raws = conf.dnsblHandler(
bot, "thisusernameshouldbetoolongtoeveractuallybeinuse", host, chan
)
bot.msg(
f"Blacklist check: {'(Cached) ' if cache else ''}{dnsbl if dnsbl else 'Safe.'} ({raws})",
chan,
)
except Exception as E:
bot.msg("Blacklist Lookup Failed. Error recorded to bot logs.", chan)
bot.log(str(E), "FATAL")
except Exception as E:
bot.msg("Blacklist lookup failed. Error recorded to bot logs.", chan)
bot.log(str(E), "FATAL")
def slap(bot: bare.bot, chan: str, name: str, message: str) -> None:
msg = message.split(" ")
if len(msg) > 1:
msg = " ".join(msg[1:]).strip()
if msg == bot.nick or not msg:
msg = name
else:
msg = name
bot.msg(
f"\x01ACTION slaps {msg} around a bit with {r.choice(['a firewall', 'a fireball', 'a large trout', 'a computer', 'an rpi4', 'an rpi5', 'firepi', name])}\x01",
chan,
)
def morning(bot: bare.bot, chan: str, name: str, message: str) -> None:
msg = message.split(" ")
addresse = " ".join(msg[2:]).strip().lower()
postfix = ""
if addresse and addresse[-1] in ["!", ".", "?"]:
postfix = addresse[-1]
addresse = addresse[:-1]
elif message[-1] in ["!", ".", "?"]:
postfix = addresse[-1]
addresse = addresse[:-1]
if addresse not in ["everyone", "people", bot.nick.lower(), ""]:
return
bot.msg(f"Good morning {name}{postfix}", chan)
def night(bot: bare.bot, chan: str, name: str, message: str) -> None:
msg = message.split(" ")
addresse = " ".join(msg[2:]).strip().lower()
postfix = ""
if addresse and addresse[-1] in ["!", ".", "?"]:
postfix = addresse[-1]
addresse = addresse[:-1]
elif message[-1] in ["!", ".", "?"]:
postfix = addresse[-1]
addresse = addresse[:-1]
if addresse not in ["everyone", "people", bot.nick.lower(), ""]:
return
bot.msg(f"Good night {name}{postfix}", chan)
def afternoon(bot: bare.bot, chan: str, name: str, message: str) -> None:
msg = message.split(" ")
addresse = " ".join(msg[2:]).strip().lower()
postfix = ""
if addresse and addresse[-1] in ["!", ".", "?"]:
postfix = addresse[-1]
addresse = addresse[:-1]
elif message[-1] in ["!", ".", "?"]:
postfix = addresse[-1]
addresse = addresse[:-1]
if addresse not in ["everyone", "people", bot.nick.lower(), ""]:
return
bot.msg(f"Good afternoon {name}{postfix}", chan)
data: dict[str, dict[str, Any]] = { data: dict[str, dict[str, Any]] = {
"!botlist": {"prefix": False, "aliases": []}, "!botlist": {"prefix": False, "aliases": []},
"bugs bugs bugs": {"prefix": False, "aliases": []}, "bugs bugs bugs": {"prefix": False, "aliases": []},
@ -386,17 +209,11 @@ data: dict[str, dict[str, Any]] = {
"aliases": ["reboot", "stop", "hardreload", "hr"], "aliases": ["reboot", "stop", "hardreload", "hr"],
"check": checks.admin, "check": checks.admin,
}, },
"uptime": {"prefix": True, "aliases": ["u"]}, "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},
"debugInternal": {
"prefix": True,
"aliases": ["dbgInt", "dI"],
"check": checks.admin,
},
"debugEval": {"prefix": True, "aliases": ["dbgEval", "dE"], "check": checks.admin},
"8ball": {"prefix": True, "aliases": ["eightball", "8b"]}, "8ball": {"prefix": True, "aliases": ["eightball", "8b"]},
"join": {"prefix": True, "aliases": ["j"], "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": { "goat.mode.deactivate": {
@ -404,7 +221,7 @@ data: dict[str, dict[str, Any]] = {
"aliases": ["g.m.d"], "aliases": ["g.m.d"],
"check": checks.admin, "check": checks.admin,
}, },
"help": {"prefix": True, "aliases": ["?", "list", "h"]}, "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},
@ -412,17 +229,6 @@ data: dict[str, dict[str, Any]] = {
"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": []},
"markov": {"prefix": True, "aliases": ["m"]},
"setStatus": {"prefix": True, "aliases": ["sS"], "check": checks.admin},
"getStatus": {"prefix": True, "aliases": ["gS"]},
"check": {"prefix": True, "aliases": [], "check": checks.admin},
"slap": {"prefix": True, "aliases": ["s"]},
"good morning": {
"prefix": False,
"aliases": ["g'morning", "morning", "mornin'", "good mornin'"],
},
"good night": {"prefix": False, "aliases": ["g'night", "night"]},
"good afternoon": {"prefix": False, "aliases": ["g'afternoon", "afternoon"]},
} }
regexes: list[str] = [conf.npbase, conf.su] regexes: list[str] = [conf.npbase, conf.su]
call: dict[str, Callable[[bare.bot, str, str, str], None]] = { call: dict[str, Callable[[bare.bot, str, str, str], None]] = {
@ -433,12 +239,10 @@ call: dict[str, Callable[[bare.bot, str, str, str], None]] = {
conf.su: sudo, conf.su: sudo,
"restart": reboot, "restart": reboot,
"uptime": uptime, "uptime": uptime,
"raw": raw, "raw ": raw,
"debug": debug, "debug": debug,
"debugInternal": debugInternal,
"debugEval": debugEval,
"8ball": eball, "8ball": eball,
"join": join, "join ": join,
"quote": quote, "quote": quote,
"goat.mode.activate": goatOn, "goat.mode.activate": goatOn,
"goat.mode.decativate": goatOff, "goat.mode.decativate": goatOff,
@ -450,12 +254,4 @@ call: dict[str, Callable[[bare.bot, str, str, str], None]] = {
"fpmp": fpmp, "fpmp": fpmp,
"version": version, "version": version,
"np": fmpull, "np": fmpull,
"markov": markov,
"setStatus": setStatus,
"getStatus": getStatus,
"check": check,
"slap": slap,
"good morning": morning,
"good night": night,
"good afternoon": afternoon,
} }

142
config.py
View file

@ -2,39 +2,11 @@
from os import environ as env from os import environ as env
from dotenv import load_dotenv # type: ignore from dotenv import load_dotenv # type: ignore
import re, codecs import re, codecs
from typing import Optional, Any, Union from typing import Optional, Any
import bare, pylast import bare, pylast
from pydnsbl import DNSBLIpChecker, DNSBLDomainChecker, providers as BL
class droneBL(BL.Provider):
def process_response(self, response):
reasons = set()
for result in response:
reason = result.host
if reason in ["127.0.0.3"]:
reasons.add("IRC Spambot")
elif reason in ["127.0.0.19"]:
reasons.add("Abused VPN")
elif reason in ["127.0.0.9", "127.0.0.8"]:
reasons.add("Open Proxy")
elif reason in ["127.0.0.13"]:
reasons.add("Automated Attacks")
else:
print("Unknown dnsbl reason: " + reason, flush=True)
reasons.add("unknown")
return reasons
providers = BL.BASE_PROVIDERS + [droneBL("dnsbl.dronebl.org")]
ipbl = DNSBLIpChecker(providers=providers)
hsbl = DNSBLDomainChecker(providers=providers)
hardbl = ["146.70.59.36"]
load_dotenv() load_dotenv()
__version__ = "v3.0.22" __version__ = "v3.0.6"
npbase: str = ( npbase: str = (
"\[\x0303last\.fm\x03\] [A-Za-z0-9_[\]{}\\|\-^]{1,$MAX} (is listening|last listened) to: \x02.+ - .*\x02( \([0-9]+ plays\)( \[.*\])?)?" # pyright: ignore [reportInvalidStringEscapeSequence] "\[\x0303last\.fm\x03\] [A-Za-z0-9_[\]{}\\|\-^]{1,$MAX} (is listening|last listened) to: \x02.+ - .*\x02( \([0-9]+ plays\)( \[.*\])?)?" # pyright: ignore [reportInvalidStringEscapeSequence]
) )
@ -48,14 +20,11 @@ servers: dict[str, dict[str, Any]] = {
"channels": {"#random": 0, "#dice": 0, "#offtopic": 0, "#main/replirc": 0}, "channels": {"#random": 0, "#dice": 0, "#offtopic": 0, "#main/replirc": 0},
"ignores": ["#main/replirc"], "ignores": ["#main/replirc"],
"hosts": ["9pfs.repl.co"], "hosts": ["9pfs.repl.co"],
"dnsblMode": "kickban",
}, },
"efnet": { "efnet": {
"address": "irc.underworld.no", "address": "irc.underworld.no",
"channels": {"#random": 0, "#dice": 0}, "channels": {"#random": 0, "#dice": 0},
"hosts": ["154.sub-174-251-241.myvzw.com"], "hosts": ["154.sub-174-251-241.myvzw.com"],
"threads": ["pingMon"],
"dnsblMode": "kickban",
}, },
"replirc": { "replirc": {
"address": "127.0.0.1", "address": "127.0.0.1",
@ -69,16 +38,11 @@ servers: dict[str, dict[str, Any]] = {
"#sshchat": 0, "#sshchat": 0,
"#firemc": 0, "#firemc": 0,
"#fp-radio": 0, "#fp-radio": 0,
"#fp-radio-debug": 0,
"#hardfork": 0,
"#opers": 0,
}, },
"ignores": ["#fp-radio"], "ignores": ["#fp-radio"],
"admins": ["h-tl"], "admins": ["h-tl"],
"hosts": ["owner.firepi"], "hosts": ["owner.firepi"],
"threads": ["radio"], "threads": ["radio", "mc-down"],
"autoMethod": "MARKOV",
"dnsblMode": "akill",
}, },
"backupbox": { "backupbox": {
"address": "127.0.0.1", "address": "127.0.0.1",
@ -91,7 +55,6 @@ servers: dict[str, dict[str, Any]] = {
"2600-6c5a-637f-1a85-0000-0000-0000-6667.inf6.spectrum.com", "2600-6c5a-637f-1a85-0000-0000-0000-6667.inf6.spectrum.com",
], ],
"onIdntCmds": ["OPER e e"], "onIdntCmds": ["OPER e e"],
"dnsbl-mode": "gline",
}, },
"twitch": { "twitch": {
"nick": "fireschatbot", "nick": "fireschatbot",
@ -101,10 +64,9 @@ servers: dict[str, dict[str, Any]] = {
"#firepup650": 0, "#firepup650": 0,
}, },
"admins": ["firepup650"], "admins": ["firepup650"],
"prefix": "!",
}, },
} }
admin_hosts: list[str] = ["firepup.firepi", "47.221.98.52"] admin_hosts: list[str] = ["firepup.firepi", "47.221.227.180"]
ESCAPE_SEQUENCE_RE = re.compile( ESCAPE_SEQUENCE_RE = re.compile(
r""" r"""
( \\U........ # 8-digit hex escapes ( \\U........ # 8-digit hex escapes
@ -128,29 +90,6 @@ def decode_escapes(s: str) -> str:
return ESCAPE_SEQUENCE_RE.sub(decode_match, s) return ESCAPE_SEQUENCE_RE.sub(decode_match, s)
def cmdFind(message: str, find: list, usePrefix: bool = True) -> bool:
cmd = message.split(" ")
if not cmd:
return False
if usePrefix:
for match in find:
sMatch = (prefix + match).split(" ")
try:
if all(cmd[i] == sMatch[i] for i in range(len(sMatch))):
return True
except IndexError:
...
else:
for match in find:
sMatch = match.split(" ")
try:
if all(cmd[i] == sMatch[i] for i in range(len(sMatch))):
return True
except IndexError:
...
return False
def mfind(message: str, find: list, usePrefix: bool = True) -> bool: def mfind(message: str, find: list, usePrefix: bool = True) -> bool:
if usePrefix: if usePrefix:
return any(message[: len(match) + 1] == prefix + match for match in find) return any(message[: len(match) + 1] == prefix + match for match in find)
@ -170,76 +109,3 @@ def sub(
if name: if name:
result = result.replace("$SENDER", name).replace("$NAME", name) result = result.replace("$SENDER", name).replace("$NAME", name)
return result return result
def dnsbl(hostname: str) -> tuple[str, dict[str, list[str]]]:
hosts = []
hstDT = {}
try:
hstDT = ipbl.check(hostname).detected_by
except ValueError: # It's not an IP
try:
hstDT = hsbl.check(hostname).detected_by
except ValueError: # It's also not a hostname
hstDT = {}
if hostname in hardbl:
hstDT["hardcoded"] = ["Known bad host"]
for host in hstDT:
if hstDT[host] != ["unknown"]:
hosts.append(host)
if not hosts:
return "", hstDT
hostStr = None
if len(hosts) >= 3:
hostStr = ", and ".join((", ".join(hosts)).rsplit(", ", 1))
else:
hostStr = " and ".join(hosts)
return hostStr, hstDT
def dnsblHandler(
bot: bare.bot, nick: str, hostname: str, chan: str
) -> tuple[str, dict[str, list[str]]]:
dnsblStatus = "Not enabled"
dnsblResps = {}
if bot.dnsblMode != "none":
dnsblStatus, dnsblResps = (
dnsbl(hostname)
if not hostname in bot.dns
else (bot.dns[hostname]["status"], bot.dns[hostname]["resps"])
)
bot.dns[hostname] = {"status": dnsblStatus, "resps": dnsblResps}
if dnsblStatus:
match bot.dnsblMode:
case "kickban":
bot.sendraw(
f"KICK {chan} {nick} :Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
bot.sendraw(f"MODE {chan} +b *!*@{hostname}")
case "akill":
bot.sendraw(
f"OS AKILL ADD *@{hostname} !P Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
case "kline":
bot.sendraw(
f"KILL {nick} :Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
bot.sendraw(
f"KLINE 524160 *@{hostname} :Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
bot.sendraw(
f"KLINE *@{hostname} :Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
case "gline":
bot.sendraw(
f"KILL {nick} :Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
bot.sendraw(
f"GLINE *@{hostname} 524160 :Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
bot.sendraw(
f"GLINE *@{hostname} :Sorry, but you're on the {dnsblStatus} blacklist(s)."
)
case _:
bot.log(f'Unknown dnsbl Mode "{bot.dnsblMode}"!', "WARN")
return dnsblStatus, dnsblResps

View file

@ -3,7 +3,7 @@ from os import system
from time import sleep from time import sleep
from threading import Thread from threading import Thread
from logs import log from logs import log
from threads import threadManager from timers import threadManager
def launch(server: str) -> None: def launch(server: str) -> None:

View file

@ -6,7 +6,6 @@ from typing import Union, Callable
from overrides import bytes, bbytes from overrides import bytes, bbytes
from importlib import reload from importlib import reload
import bare, re, checks import bare, re, checks
from traceback import format_exc
def CTCP(bot: bare.bot, msg: str) -> bool: def CTCP(bot: bare.bot, msg: str) -> bool:
@ -15,7 +14,7 @@ def CTCP(bot: bare.bot, msg: str) -> bool:
bot.log(f'Responding to CTCP "{kind}" from {sender}') bot.log(f'Responding to CTCP "{kind}" from {sender}')
if kind == "VERSION": if kind == "VERSION":
bot.notice( bot.notice(
f"\x01VERSION FireBot {conf.__version__} (https://git.h.hackclub.app/Firepup650/FireBot)\x01", f"\x01VERSION FireBot {conf.__version__} (https://git.amcforum.wiki/Firepup650/fire-ircbot)\x01",
sender, sender,
True, True,
) )
@ -25,7 +24,7 @@ def CTCP(bot: bare.bot, msg: str) -> bool:
return True return True
elif kind == "SOURCE": elif kind == "SOURCE":
bot.notice( bot.notice(
"\x01SOURCE https://git.h.hackclub.app/Firepup650/FireBot\x01", "\x01SOURCE https://git.amcforum.wiki/Firepup650/fire-ircbot\x01",
sender, sender,
True, True,
) )
@ -104,34 +103,16 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str]
triggers = [cmd] triggers = [cmd]
triggers.extend(cmds.data[cmd]["aliases"]) triggers.extend(cmds.data[cmd]["aliases"])
triggers = list(conf.sub(call, bot, chan, name).lower() for call in triggers) triggers = list(conf.sub(call, bot, chan, name).lower() for call in triggers)
if conf.cmdFind( if conf.mfind(
conf.sub(message, bot, chan, name).lower(), conf.sub(message, bot, chan, name).lower(),
triggers, triggers,
cmds.data[cmd]["prefix"], cmds.data[cmd]["prefix"],
): ):
if "check" in cmds.data[cmd] and cmds.data[cmd]["check"]: if "check" in cmds.data[cmd] and cmds.data[cmd]["check"]:
if cmds.data[cmd]["check"](bot, name, host, chan, cmd): if cmds.data[cmd]["check"](bot, name, host, chan, cmd):
try:
cmds.call[cmd](bot, chan, name, message) cmds.call[cmd](bot, chan, name, message)
except Exception:
Err = format_exc()
for line in Err.split("\n"):
bot.log(line, "ERROR")
bot.msg(
"Sorry, I had an error trying to execute that command. Please check error logs.",
chan,
)
else: else:
try:
cmds.call[cmd](bot, chan, name, message) cmds.call[cmd](bot, chan, name, message)
except Exception:
Err = format_exc()
for line in Err.split("\n"):
bot.log(line, "ERROR")
bot.msg(
"Sorry, I had an error trying to execute that command. Please check error logs.",
chan,
)
handled = True handled = True
break break
if not handled: if not handled:
@ -143,7 +124,7 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str]
cmds.call[check](bot, chan, name, message) cmds.call[check](bot, chan, name, message)
handled = True handled = True
break break
if not handled and conf.cmdFind(message, ["reload", "r"]): if not handled and conf.mfind(message, ["reload", "r"]):
if checks.admin(bot, name, host, chan, "reload"): if checks.admin(bot, name, host, chan, "reload"):
return "reload", chan return "reload", chan
handled = True handled = True
@ -158,17 +139,13 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str]
elif kind == "ACTION ducks": elif kind == "ACTION ducks":
bot.msg("\x01ACTION gets hit by a duck\x01", chan) bot.msg("\x01ACTION gets hit by a duck\x01", chan)
if chan in bot.channels and bot.channels[chan] >= bot.interval: if chan in bot.channels and bot.channels[chan] >= bot.interval:
sel = ""
bot.channels[chan] = 0
if bot.autoMethod == "QUOTE":
r.seed() r.seed()
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].replace("\\n", "").replace("\n", "") r.sample(mm.readlines(), 1)[0].replace("\\n", "").replace("\n", "")
) )
else: bot.msg(f"[QUOTE] {sel}", chan)
sel = bot.markov.generate_from_sentence(message)
bot.msg(f"[{bot.autoMethod}] {sel}", chan)
return None, None return None, None
@ -196,45 +173,6 @@ def PART(bot: bare.bot, msg: str) -> tuple[None, None]:
return None, None return None, None
def QUIT(bot: bare.bot, msg: str) -> tuple[None, None]:
if bot.server == "replirc":
quitter = msg.split("!", 1)[0][1:]
if quitter == "FireMCbot":
bot.send("TOPIC #firemc :FireMC Relay channel (offline)\n")
return None, None
def JOIN(bot: bare.bot, msg: str) -> tuple[None, None]:
nick = msg.split("!", 1)[0][1:]
hostname = msg.split("@", 1)[1].split(" ", 1)[0].strip()
chan = msg.split("#")[-1].strip()
conf.dnsblHandler(bot, nick, hostname, chan)
return None, None
def MODE(bot: bare.bot, msg: str) -> tuple[None, None]:
try:
chan = msg.split("#", 1)[1].split(" ", 1)[0]
add = True if msg.split("#", 1)[1].split(" ", 2)[1][0] == "+" else False
modes = msg.split("#", 1)[1].split(" ", 2)[1][1:]
users = ""
try:
users = msg.split("#", 1)[1].split(" ", 2)[2].split()
except IndexError:
...
if len(modes) != len(users):
bot.log("Refusing to handle modes that do not have corresponding users.")
return None, None
for i in range(len(modes)):
if users[i] == bot.nick:
if modes[i] == "o":
bot.ops[chan] = add
bot.log(f"{'Got' if add else 'Lost'} ops in {chan}")
except IndexError: # *our* modes are changing, not a channel
bot.log("Not handling changing of my modes")
return None, None
def NULL(bot: bare.bot, msg: str) -> tuple[None, None]: def NULL(bot: bare.bot, msg: str) -> tuple[None, None]:
return None, None return None, None
@ -246,10 +184,6 @@ handles: dict[
"NICK": NICK, "NICK": NICK,
"KICK": KICK, "KICK": KICK,
"PART": PART, "PART": PART,
"MODE": MODE, "MODE": NULL,
"TOPIC": NULL, "TOPIC": NULL,
"QUIT": QUIT,
"JOIN": JOIN,
"NOTICE": NULL,
"INVITE": NULL,
} }

View file

@ -112,7 +112,7 @@ def CTCPHandler(msg: str, sender: str = "", isRaw: bool = False):
log(f"Responding to CTCP {CTCP} from {sender}", server) log(f"Responding to CTCP {CTCP} from {sender}", server)
if CTCP == "VERSION": if CTCP == "VERSION":
notice( notice(
f"\x01VERSION FireBot {__version__} (https://git.h.hackclub.app/Firepup650/FireBot)\x01", f"\x01VERSION FireBot {__version__} (https://git.amcforum.wiki/Firepup650/fire-ircbot)\x01",
sender, sender,
True, True,
) )
@ -122,7 +122,7 @@ def CTCPHandler(msg: str, sender: str = "", isRaw: bool = False):
return True return True
elif CTCP == "SOURCE": elif CTCP == "SOURCE":
notice( notice(
"\x01SOURCE https://git.h.hackclub.app/Firepup650/FireBot\x01", "\x01SOURCE https://git.amcforum.wiki/Firepup650/fire-ircbot\x01",
sender, sender,
True, True,
) )
@ -304,7 +304,7 @@ def main():
False, False,
): ):
sendmsg( sendmsg(
f"Hi! I'm FireBot (https://git.h.hackclub.app/Firepup650/FireBot)! My admins on this server are {adminnames}.", f"Hi! I'm FireBot (https://git.amcforum.wiki/Firepup650/fire-ircbot)! My admins on this server are {adminnames}.",
chan, chan,
) )
if mfind( if mfind(

View file

@ -10,7 +10,7 @@ def log(
level: str = "LOG", level: str = "LOG",
time: Union[dt, str] = "now", time: Union[dt, str] = "now",
) -> None: ) -> None:
if level in ["EXIT", "CRASH", "FATAL", "ERROR"]: if level in ["EXIT", "CRASH", "FATAL"]:
stream = stderr stream = stderr
else: else:
stream = stdout stream = stdout

View file

@ -1,51 +0,0 @@
import random
from typing import Union
class MarkovBot:
def __init__(self, text: list[list[str]]) -> None:
self.text = text
self.chains = {}
self.__build_chains()
def __build_chains(self) -> None:
for i in range(len(self.text)):
text = self.text[i]
for j in range(len(text) - 1):
current_word = text[j]
next_word = text[j + 1]
if current_word not in self.chains:
self.chains[current_word] = {}
if next_word not in self.chains[current_word]:
self.chains[current_word][next_word] = 0
self.chains[current_word][next_word] += 1
def generate_text(self, word: Union[str, None] = None) -> str:
if not word:
current_word = random.choice(list(self.chains.keys()))
else:
current_word = word
generated_text = current_word
while current_word in self.chains:
next_word = random.choices(
list(self.chains[current_word].keys()),
weights=list(self.chains[current_word].values()),
)[0]
generated_text += " " + next_word
current_word = next_word
return generated_text
def generate_from_sentence(self, msg: Union[str, None] = None) -> str:
if not msg:
word = random.choice(list(self.chains.keys()))
else:
word = random.choice(msg.split())
if (for_word := self.generate_text(word)) != word:
return for_word
else:
return self.generate_text()

File diff suppressed because it is too large Load diff

147
poetry.lock generated
View file

@ -38,53 +38,6 @@ files = [
certifi = "*" certifi = "*"
urllib3 = "*" urllib3 = "*"
[[package]]
name = "black"
version = "24.4.2"
description = "The uncompromising code formatter."
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "black-24.4.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:dd1b5a14e417189db4c7b64a6540f31730713d173f0b63e55fabd52d61d8fdce"},
{file = "black-24.4.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:8e537d281831ad0e71007dcdcbe50a71470b978c453fa41ce77186bbe0ed6021"},
{file = "black-24.4.2-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:eaea3008c281f1038edb473c1aa8ed8143a5535ff18f978a318f10302b254063"},
{file = "black-24.4.2-cp310-cp310-win_amd64.whl", hash = "sha256:7768a0dbf16a39aa5e9a3ded568bb545c8c2727396d063bbaf847df05b08cd96"},
{file = "black-24.4.2-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:257d724c2c9b1660f353b36c802ccece186a30accc7742c176d29c146df6e474"},
{file = "black-24.4.2-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:bdde6f877a18f24844e381d45e9947a49e97933573ac9d4345399be37621e26c"},
{file = "black-24.4.2-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e151054aa00bad1f4e1f04919542885f89f5f7d086b8a59e5000e6c616896ffb"},
{file = "black-24.4.2-cp311-cp311-win_amd64.whl", hash = "sha256:7e122b1c4fb252fd85df3ca93578732b4749d9be076593076ef4d07a0233c3e1"},
{file = "black-24.4.2-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:accf49e151c8ed2c0cdc528691838afd217c50412534e876a19270fea1e28e2d"},
{file = "black-24.4.2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:88c57dc656038f1ab9f92b3eb5335ee9b021412feaa46330d5eba4e51fe49b04"},
{file = "black-24.4.2-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:be8bef99eb46d5021bf053114442914baeb3649a89dc5f3a555c88737e5e98fc"},
{file = "black-24.4.2-cp312-cp312-win_amd64.whl", hash = "sha256:415e686e87dbbe6f4cd5ef0fbf764af7b89f9057b97c908742b6008cc554b9c0"},
{file = "black-24.4.2-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:bf10f7310db693bb62692609b397e8d67257c55f949abde4c67f9cc574492cc7"},
{file = "black-24.4.2-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:98e123f1d5cfd42f886624d84464f7756f60ff6eab89ae845210631714f6db94"},
{file = "black-24.4.2-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:48a85f2cb5e6799a9ef05347b476cce6c182d6c71ee36925a6c194d074336ef8"},
{file = "black-24.4.2-cp38-cp38-win_amd64.whl", hash = "sha256:b1530ae42e9d6d5b670a34db49a94115a64596bc77710b1d05e9801e62ca0a7c"},
{file = "black-24.4.2-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:37aae07b029fa0174d39daf02748b379399b909652a806e5708199bd93899da1"},
{file = "black-24.4.2-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:da33a1a5e49c4122ccdfd56cd021ff1ebc4a1ec4e2d01594fef9b6f267a9e741"},
{file = "black-24.4.2-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ef703f83fc32e131e9bcc0a5094cfe85599e7109f896fe8bc96cc402f3eb4b6e"},
{file = "black-24.4.2-cp39-cp39-win_amd64.whl", hash = "sha256:b9176b9832e84308818a99a561e90aa479e73c523b3f77afd07913380ae2eab7"},
{file = "black-24.4.2-py3-none-any.whl", hash = "sha256:d36ed1124bb81b32f8614555b34cc4259c3fbc7eec17870e8ff8ded335b58d8c"},
{file = "black-24.4.2.tar.gz", hash = "sha256:c872b53057f000085da66a19c55d68f6f8ddcac2642392ad3a355878406fbd4d"},
]
[package.dependencies]
click = ">=8.0.0"
mypy-extensions = ">=0.4.3"
packaging = ">=22.0"
pathspec = ">=0.9.0"
platformdirs = ">=2"
tomli = {version = ">=1.1.0", markers = "python_version < \"3.11\""}
typing-extensions = {version = ">=4.0.1", markers = "python_version < \"3.11\""}
[package.extras]
colorama = ["colorama (>=0.4.3)"]
d = ["aiohttp (>=3.7.4)", "aiohttp (>=3.7.4,!=3.9.0)"]
jupyter = ["ipython (>=7.8.0)", "tokenize-rt (>=3.2.0)"]
uvloop = ["uvloop (>=0.15.2)"]
[[package]] [[package]]
name = "certifi" name = "certifi"
version = "2024.2.2" version = "2024.2.2"
@ -97,43 +50,16 @@ files = [
{file = "certifi-2024.2.2.tar.gz", hash = "sha256:0569859f95fc761b18b45ef421b1290a0f65f147e92a1e5eb3e635f9a5e4e66f"}, {file = "certifi-2024.2.2.tar.gz", hash = "sha256:0569859f95fc761b18b45ef421b1290a0f65f147e92a1e5eb3e635f9a5e4e66f"},
] ]
[[package]]
name = "click"
version = "8.1.7"
description = "Composable command line interface toolkit"
category = "main"
optional = false
python-versions = ">=3.7"
files = [
{file = "click-8.1.7-py3-none-any.whl", hash = "sha256:ae74fb96c20a0277a1d615f1e4d73c8414f5a98db8b799a7931d1582f3390c28"},
{file = "click-8.1.7.tar.gz", hash = "sha256:ca9853ad459e787e2192211578cc907e7594e294c7ccc834310722b41b9ca6de"},
]
[package.dependencies]
colorama = {version = "*", markers = "platform_system == \"Windows\""}
[[package]]
name = "colorama"
version = "0.4.6"
description = "Cross-platform colored terminal text."
category = "main"
optional = false
python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7"
files = [
{file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"},
{file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"},
]
[[package]] [[package]]
name = "exceptiongroup" name = "exceptiongroup"
version = "1.2.1" version = "1.2.0"
description = "Backport of PEP 654 (exception groups)" description = "Backport of PEP 654 (exception groups)"
category = "main" category = "main"
optional = false optional = false
python-versions = ">=3.7" python-versions = ">=3.7"
files = [ files = [
{file = "exceptiongroup-1.2.1-py3-none-any.whl", hash = "sha256:5258b9ed329c5bbdd31a309f53cbfb0b155341807f6ff7606a1e801a891b29ad"}, {file = "exceptiongroup-1.2.0-py3-none-any.whl", hash = "sha256:4bfd3996ac73b41e9b9628b04e079f193850720ea5945fc96a08633c66912f14"},
{file = "exceptiongroup-1.2.1.tar.gz", hash = "sha256:a4785e48b045528f5bfe627b6ad554ff32def154f42372786903b7abcfe1aa16"}, {file = "exceptiongroup-1.2.0.tar.gz", hash = "sha256:91f5c769735f051a4290d52edd0858999b57e5876e9f85937691bd4c9fa3ed68"},
] ]
[package.extras] [package.extras]
@ -210,59 +136,6 @@ files = [
{file = "idna-3.7.tar.gz", hash = "sha256:028ff3aadf0609c1fd278d8ea3089299412a7a8b9bd005dd08b9f8285bcb5cfc"}, {file = "idna-3.7.tar.gz", hash = "sha256:028ff3aadf0609c1fd278d8ea3089299412a7a8b9bd005dd08b9f8285bcb5cfc"},
] ]
[[package]]
name = "mypy-extensions"
version = "1.0.0"
description = "Type system extensions for programs checked with the mypy type checker."
category = "main"
optional = false
python-versions = ">=3.5"
files = [
{file = "mypy_extensions-1.0.0-py3-none-any.whl", hash = "sha256:4392f6c0eb8a5668a69e23d168ffa70f0be9ccfd32b5cc2d26a34ae5b844552d"},
{file = "mypy_extensions-1.0.0.tar.gz", hash = "sha256:75dbf8955dc00442a438fc4d0666508a9a97b6bd41aa2f0ffe9d2f2725af0782"},
]
[[package]]
name = "packaging"
version = "24.0"
description = "Core utilities for Python packages"
category = "main"
optional = false
python-versions = ">=3.7"
files = [
{file = "packaging-24.0-py3-none-any.whl", hash = "sha256:2ddfb553fdf02fb784c234c7ba6ccc288296ceabec964ad2eae3777778130bc5"},
{file = "packaging-24.0.tar.gz", hash = "sha256:eb82c5e3e56209074766e6885bb04b8c38a0c015d0a30036ebe7ece34c9989e9"},
]
[[package]]
name = "pathspec"
version = "0.12.1"
description = "Utility library for gitignore style pattern matching of file paths."
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "pathspec-0.12.1-py3-none-any.whl", hash = "sha256:a0d503e138a4c123b27490a4f7beda6a01c6f288df0e4a8b79c7eb0dc7b4cc08"},
{file = "pathspec-0.12.1.tar.gz", hash = "sha256:a482d51503a1ab33b1c67a6c3813a26953dbdc71c31dacaef9a838c4e29f5712"},
]
[[package]]
name = "platformdirs"
version = "4.2.1"
description = "A small Python package for determining appropriate platform-specific dirs, e.g. a `user data dir`."
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "platformdirs-4.2.1-py3-none-any.whl", hash = "sha256:17d5a1161b3fd67b390023cb2d3b026bbd40abde6fdb052dfbd3a29c3ba22ee1"},
{file = "platformdirs-4.2.1.tar.gz", hash = "sha256:031cd18d4ec63ec53e82dceaac0417d218a6863f7745dfcc9efe7793b7039bdf"},
]
[package.extras]
docs = ["furo (>=2023.9.10)", "proselint (>=0.13)", "sphinx (>=7.2.6)", "sphinx-autodoc-typehints (>=1.25.2)"]
test = ["appdirs (==1.4.4)", "covdefaults (>=2.3)", "pytest (>=7.4.3)", "pytest-cov (>=4.1)", "pytest-mock (>=3.12)"]
type = ["mypy (>=1.8)"]
[[package]] [[package]]
name = "pylast" name = "pylast"
version = "5.2.0" version = "5.2.0"
@ -308,18 +181,6 @@ files = [
{file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"}, {file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"},
] ]
[[package]]
name = "tomli"
version = "2.0.1"
description = "A lil' TOML parser"
category = "main"
optional = false
python-versions = ">=3.7"
files = [
{file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"},
{file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"},
]
[[package]] [[package]]
name = "typing-extensions" name = "typing-extensions"
version = "4.11.0" version = "4.11.0"
@ -353,4 +214,4 @@ zstd = ["zstandard (>=0.18.0)"]
[metadata] [metadata]
lock-version = "2.0" lock-version = "2.0"
python-versions = "^3.9" python-versions = "^3.9"
content-hash = "206264f4fe64babd8c2fecfc2051adc4e1c2971fe7e67eb5ff648ecae1095099" content-hash = "5f0106196ba3a316e887fe2748bb5ca19159f2905d7678393b8ee51430f9ca45"

View file

@ -9,7 +9,6 @@ python = "^3.9"
apiclient = "^1.0.4" apiclient = "^1.0.4"
python-dotenv = "^1.0.0" python-dotenv = "^1.0.0"
pylast = "^5.2.0" pylast = "^5.2.0"
black = "^24.4.2"
[tool.poetry.dev-dependencies] [tool.poetry.dev-dependencies]

View file

@ -17,7 +17,7 @@ def is_dead(thr: Thread) -> bool:
def threadWrapper(data: dict) -> NoReturn: def threadWrapper(data: dict) -> NoReturn:
if not data["noWrap"]: if not data["noWrap"]:
while 1: while 1:
if data["ignoreErrors"]: if ignoreErrors:
try: try:
data["func"](*data["args"]) data["func"](*data["args"])
except Exception: except Exception:
@ -79,24 +79,12 @@ def radio(instance: bare.bot) -> NoReturn:
lastTrack = "" lastTrack = ""
complained = False complained = False
firstMiss = False firstMiss = False
misses = 0
missChunk = 0
missCap = -5
perChunk = 10
debug = False # instance.server == "replirc"
while 1: while 1:
try: try:
newTrack = instance.lastfmLink.get_user("Firepup650").get_now_playing() newTrack = instance.lastfmLink.get_user("Firepup650").get_now_playing()
if newTrack: if newTrack:
if complained:
complained = False complained = False
misses = 0 firstMiss = False
missChunk = 0
elif misses > missCap:
missChunk += 1
if missChunk >= perChunk:
misses -= 1
missChunk = 0
thisTrack = newTrack.__str__() thisTrack = newTrack.__str__()
if thisTrack != lastTrack: if thisTrack != lastTrack:
lastTrack = thisTrack lastTrack = thisTrack
@ -105,21 +93,8 @@ def radio(instance: bare.bot) -> NoReturn:
f"TOPIC #fp-radio :Firepup radio ({thisTrack}) - https://open.spotify.com/playlist/4ctNy3O0rOwhhXIKyLvUZM" f"TOPIC #fp-radio :Firepup radio ({thisTrack}) - https://open.spotify.com/playlist/4ctNy3O0rOwhhXIKyLvUZM"
) )
elif not complained: elif not complained:
missChunk = 0 if not firstMiss:
if misses < 0: firstMiss = True
misses += 1
if debug:
instance.msg(
str(
{
"misses": misses,
"missChunk": missChunk,
"misses exceed or meet limit": misses <= missCap,
}
),
"#fp-radio-debug",
)
sleep(2)
continue continue
instance.msg( instance.msg(
"Firepup seems to have stopped the music by mistake :/", "#fp-radio" "Firepup seems to have stopped the music by mistake :/", "#fp-radio"
@ -133,27 +108,15 @@ def radio(instance: bare.bot) -> NoReturn:
Err = format_exc() Err = format_exc()
for line in Err.split("\n"): for line in Err.split("\n"):
instance.log(line, "WARN") instance.log(line, "WARN")
if debug:
instance.msg(
str(
{
"misses": misses,
"missChunk": missChunk,
"misses exceed or meet limit": misses <= missCap,
}
),
"#fp-radio-debug",
)
sleep(2) sleep(2)
instance.log("Thread while loop broken", "FATAL") instance.log("Thread while loop broken", "FATAL")
exit(1) exit(1)
def ping(instance: bare.bot) -> NoReturn: def mcDown(instance: bare.bot) -> None:
while 1: instance.sendraw("TOPIC #firemc :FireMC Relay channel (offline)")
instance.sendraw("PING :keepalive")
sleep(30)
data: dict[str, dict[str, Any]] = { data: dict[str, dict[str, Any]] = {
"radio": {"noWrap": True, "func": radio, "passInstance": True}, "radio": {"noWrap": True, "func": radio, "args": []},
"pingMon": {"noWrap": True, "func": ping, "passInstance": True} "mc-down": {"noWrap": False, "func": mcDown, "args": [], "interval": 60}
} }