Compare commits

...

10 commits

11 changed files with 366 additions and 64 deletions

View file

@ -2,6 +2,7 @@
from socket import socket 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
logs = ... logs = ...
re = ... re = ...
@ -33,6 +34,9 @@ class bot:
npallowed: list[str] npallowed: list[str]
current: str current: str
tmpHost: str tmpHost: str
ignores: list[str]
threads: list[str]
lastfmLink: LastFMNetwork
def __init__(self, server: str): def __init__(self, server: str):
... ...

56
bot.py
View file

@ -8,9 +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 timers
import random as r import random as r
import handlers import handlers
import bare import bare
from threading import Thread
def mfind(message: str, find: list, usePrefix: bool = True) -> bool: def mfind(message: str, find: list, usePrefix: bool = True) -> bool:
@ -30,18 +32,27 @@ class bot(bare.bot):
conf.servers[server]["port"] if "port" in conf.servers[server] else 6667 conf.servers[server]["port"] if "port" in conf.servers[server] else 6667
) )
self.channels = conf.servers[server]["channels"] self.channels = conf.servers[server]["channels"]
self.adminnames = (
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
)
self.__version__ = conf.__version__
self.npallowed = conf.npallowed
self.interval = ( self.interval = (
conf.servers[server]["interval"] conf.servers[server]["interval"]
if "interval" in conf.servers[server] if "interval" in conf.servers[server]
else 50 else 50
) )
self.__version__ = conf.__version__
self.nick = "FireBot" self.nick = "FireBot"
self.adminnames = conf.servers[server]["admins"]
self.queue: list[bbytes] = [] # pyright: ignore [reportInvalidTypeForm] self.queue: list[bbytes] = [] # pyright: ignore [reportInvalidTypeForm]
self.sock = socket(AF_INET, SOCK_STREAM) self.sock = socket(AF_INET, SOCK_STREAM)
self.npallowed = ["FireBitBot"]
self.current = "user" self.current = "user"
self.threads = (
conf.servers[server]["threads"] if "threads" in conf.servers[server] else []
)
self.lastfmLink = conf.lastfmLink
self.log(f"Start init for {self.server}") self.log(f"Start init for {self.server}")
def connect(self) -> None: def connect(self) -> None:
@ -125,6 +136,21 @@ class bot(bare.bot):
if origin != "null": if origin != "null":
self.msg(f"{chan} is +i, and I'm not invited.", origin) self.msg(f"{chan} is +i, and I'm not invited.", origin)
break break
elif code == 474:
self.log(f"Joining {chan} failed (+b)", "WARN")
if origin != "null":
self.msg(f"I'm banned from {chan}.", origin)
break
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)
break
elif code == 519:
self.log(f"Joining {chan} failed (+A)", "WARN")
if origin != "null":
self.msg(f"{chan} is +A, and I'm not an admin.", origin)
break
elif code == 520: elif code == 520:
self.log(f"Joining {chan} failed (+O)", "WARN") self.log(f"Joining {chan} failed (+O)", "WARN")
if origin != "null": if origin != "null":
@ -204,6 +230,16 @@ class bot(bare.bot):
sleep(0.5) sleep(0.5)
for chan in self.channels: for chan in self.channels:
self.join(chan, "null", False) self.join(chan, "null", False)
tMgr = None
if self.threads:
tdict = {}
for thread in self.threads:
tdict[thread] = timers.data[thread]
if thread in ["radio"]:
tdict[thread]["args"] = [self]
tMgr = Thread(target=timers.threadManager, args=(tdict,))
tMgr.daemon = True
tMgr.start()
while 1: while 1:
raw = self.recv() raw = self.recv()
ircmsg = raw.safe_decode() ircmsg = raw.safe_decode()
@ -221,7 +257,19 @@ class bot(bare.bot):
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:
reload(conf) reload(conf)
self.adminnames = (
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 []
)
self.__version__ = conf.__version__ self.__version__ = conf.__version__
self.npallowed = conf.npallowed
self.interval = (
conf.servers[self.server]["interval"]
if "interval" in conf.servers[self.server]
else 50
)
reload(cmds) reload(cmds)
reload(handlers) reload(handlers)
self.msg("Reloaded successfully", chan) self.msg("Reloaded successfully", chan)
@ -232,4 +280,6 @@ class bot(bare.bot):
self.exit("I got killed :'(") self.exit("I got killed :'(")
elif ircmsg.startswith("ERROR :Ping "): elif ircmsg.startswith("ERROR :Ping "):
self.exit("Ping timeout") self.exit("Ping timeout")
else:
self.log("Unrecognized server request!", "WARN")
self.exit("While loop broken") self.exit("While loop broken")

View file

@ -13,7 +13,7 @@ def admin(
cmd: Optional[str] = "", cmd: Optional[str] = "",
) -> bool: ) -> bool:
if ( if (
name.lower() in conf.servers[bot.server]["admins"] name.lower() in bot.adminnames
or (host or bot.tmpHost) in conf.admin_hosts or (host or bot.tmpHost) in conf.admin_hosts
or (host or bot.tmpHost) in conf.servers[bot.server]["hosts"] or (host or bot.tmpHost) in conf.servers[bot.server]["hosts"]
): ):

View file

@ -6,6 +6,13 @@ from typing import Any, Callable
import bare, re, checks import bare, re, checks
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: 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)
@ -14,7 +21,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.amcforum.wiki/Firepup650/fire-ircbot)! {'My admins on this server are' + 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,
) )
@ -105,7 +112,7 @@ def quote(bot: bare.bot, chan: str, name: str, message: str) -> None:
if q == []: if q == []:
q = [f'No results for "{query}" '] q = [f'No results for "{query}" ']
sel = conf.decode_escapes( sel = conf.decode_escapes(
str(r.sample(q, 1)).strip("[]'").replace("\\n", "").strip('"') r.sample(q, 1)[0].replace("\\n", "").replace("\n", "")
) )
bot.msg(sel, chan) bot.msg(sel, chan)
@ -131,7 +138,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(conf.servers[bot.server]["admins"]) "ADMINS": str(bot.adminnames)
+ " (Does not include hostname checks)", + " (Does not include hostname checks)",
"CHANNELS": bot.channels, "CHANNELS": bot.channels,
} }
@ -165,8 +172,14 @@ 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)
except:
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 ''}", 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]] = {
@ -174,20 +187,23 @@ data: dict[str, dict[str, Any]] = {
"bugs bugs bugs": {"prefix": False, "aliases": []}, "bugs bugs bugs": {"prefix": False, "aliases": []},
"hi $BOTNICK": {"prefix": False, "aliases": ["hello $BOTNICK"]}, "hi $BOTNICK": {"prefix": False, "aliases": ["hello $BOTNICK"]},
# [npbase, su] # [npbase, su]
"restart": {"prefix": True, "aliases": ["reboot"], "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"], "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": [], "check": checks.admin}, "goat.mode.activate": {"prefix": True, "aliases": ["g.m.a"], "check": checks.admin},
"goat.mode.deactivate": {"prefix": True, "aliases": [], "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": []},
"version": {"prefix": True, "aliases": ["ver","v"]},
"np": {"prefix": True, "aliases": []},
} }
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]] = {
@ -210,4 +226,7 @@ call: dict[str, Callable[[bare.bot, str, str, str], None]] = {
"ping": ping, "ping": ping,
"op me": op, "op me": op,
"whoami": whoami, "whoami": whoami,
"fpmp": fpmp,
"version": version,
"np": fmpull,
} }

View file

@ -3,40 +3,42 @@ 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 from typing import Optional, Any
import bare import bare, pylast
load_dotenv() load_dotenv()
__version__ = "v2.0.6" __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": {
"address": "localhost", "address": "127.0.0.1",
"port": 6601, "port": 6601,
"interval": 200, "interval": 200,
"pass": env["ircnow_pass"], "pass": env["ircnow_pass"],
"channels": {"#random": 0, "#dice": 0, "#offtopic": 0, "#main/replirc": 0}, "channels": {"#random": 0, "#dice": 0, "#offtopic": 0, "#main/replirc": 0},
"admins": [], "ignores": ["#main/replirc"],
"hosts": ["9pfs.repl.co"], "hosts": ["9pfs.repl.co"],
}, },
"efnet": { "efnet": {
"address": "irc.mzima.net", "address": "irc.mzima.net",
"channels": {"#random": 0, "#dice": 0}, "channels": {"#random": 0, "#dice": 0},
"admins": [],
"hosts": ["154.sub-174-251-241.myvzw.com"], "hosts": ["154.sub-174-251-241.myvzw.com"],
}, },
"replirc": { "replirc": {
"address": "localhost", "address": "127.0.0.1",
"pass": env["replirc_pass"], "pass": env["replirc_pass"],
"channels": {"#random": 0, "#dice": 0, "#main": 0, "#bots": 0, "#firebot": 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"], "admins": ["h-tl"],
"hosts": ["owner.firepi"], "hosts": ["owner.firepi"],
"threads": ["radio"],
}, },
"backupbox": { "backupbox": {
"address": "172.23.11.5", "address": "127.0.0.1",
"port": 6607,
"channels": {"#default": 0, "#botrebellion": 0, "#main/replirc": 0}, "channels": {"#default": 0, "#botrebellion": 0, "#main/replirc": 0},
"admins": [], "ignores": ["#main/replirc"],
"hosts": ["172.20.171.225", "169.254.253.107"], "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"]
@ -52,7 +54,8 @@ ESCAPE_SEQUENCE_RE = re.compile(
re.UNICODE | re.VERBOSE, re.UNICODE | re.VERBOSE,
) )
prefix = "." prefix = "."
lastfmLink = pylast.LastFMNetwork(env["FM_KEY"], env["FM_SECRET"])
npallowed: list[str] = ["FireBitBot"]
def decode_escapes(s: str) -> str: def decode_escapes(s: str) -> str:
def decode_match(match): def decode_match(match):

40
core.py
View file

@ -3,43 +3,19 @@ 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 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}")
threads = {} servers = {
servers = [ "ircnow": {"noWrap": True, "func": launch, "args": ["ircnow"]},
"ircnow", "replirc": {"noWrap": True, "func": launch, "args": ["replirc"]},
"replirc", # "efnet": {"noWrap": True, "func": launch, "args": ["efnet"]},
# "efnet", "backupbox": {"noWrap": True, "func": launch, "args": ["backupbox"]},
"backupbox", }
]
def is_dead(thr: Thread) -> bool:
thr.join(timeout=0)
return not thr.is_alive()
def start(server: str) -> Thread:
t = Thread(target=launch, args=(server,))
t.daemon = True
t.start()
return t
if __name__ == "__main__": if __name__ == "__main__":
log("Begin initialization", "CORE") threadManager(servers, True, "CORE")
for server in servers:
threads[server] = start(server)
log("Started all instances. Idling...", "CORE")
while 1:
sleep(60)
log("Running a checkup on all running instances", "CORE")
for server in threads:
t = threads[server]
if is_dead(t):
log(f"The thread for {server} died, restarting it...", "CORE", "WARN")
threads[server] = start(server)

View file

@ -50,7 +50,7 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str]
bot.current = "user" bot.current = "user"
if ( if (
(name.startswith("saxjax") and bot.server == "efnet") (name.startswith("saxjax") and bot.server == "efnet")
or (name == "ReplIRC" and bot.server == "replirc") or (name in ["ReplIRC", "sshchat"] and bot.server == "replirc")
or ( or (
name in ["FirePyLink_", "FirePyLink"] name in ["FirePyLink_", "FirePyLink"]
and bot.server in ["ircnow", "backupbox"] and bot.server in ["ircnow", "backupbox"]
@ -75,6 +75,8 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str]
message = msg.split("PRIVMSG", 1)[1].split(":", 1)[1].strip() message = msg.split("PRIVMSG", 1)[1].split(":", 1)[1].strip()
chan = msg.split("PRIVMSG", 1)[1].split(":", 1)[0].strip() chan = msg.split("PRIVMSG", 1)[1].split(":", 1)[0].strip()
message = conf.sub(message, bot, chan, name) message = conf.sub(message, bot, chan, name)
if chan in bot.ignores:
return None, None
bot.log( bot.log(
f'Got "{bytes(message).lazy_decode()}" from "{name}" in "{chan}" ({bot.current})', f'Got "{bytes(message).lazy_decode()}" from "{name}" in "{chan}" ({bot.current})',
) )
@ -122,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.mfind(message, ["reload"]): 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
@ -141,10 +143,9 @@ 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(
str(r.sample(mm.readlines(), 1)) r.sample(mm.readlines(), 1)[0]
.strip("[]'")
.replace("\\n", "") .replace("\\n", "")
.strip('"') .replace("\n", "")
) )
bot.msg(f"[QUOTE] {sel}", chan) bot.msg(f"[QUOTE] {sel}", chan)
return None, None return None, None

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"]: if level in ["EXIT", "CRASH", "FATAL"]:
stream = stderr stream = stderr
else: else:
stream = stdout stream = stdout

159
poetry.lock generated
View file

@ -1,9 +1,33 @@
# This file is automatically @generated by Poetry 1.6.1 and should not be changed by hand. # This file is automatically @generated by Poetry and should not be changed by hand.
[[package]]
name = "anyio"
version = "4.3.0"
description = "High level compatibility layer for multiple asynchronous event loop implementations"
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "anyio-4.3.0-py3-none-any.whl", hash = "sha256:048e05d0f6caeed70d731f3db756d35dcc1f35747c8c403364a8332c630441b8"},
{file = "anyio-4.3.0.tar.gz", hash = "sha256:f75253795a87df48568485fd18cdd2a3fa5c4f7c5be8e5e36637733fce06fed6"},
]
[package.dependencies]
exceptiongroup = {version = ">=1.0.2", markers = "python_version < \"3.11\""}
idna = ">=2.8"
sniffio = ">=1.1"
typing-extensions = {version = ">=4.1", markers = "python_version < \"3.11\""}
[package.extras]
doc = ["Sphinx (>=7)", "packaging", "sphinx-autodoc-typehints (>=1.2.0)", "sphinx-rtd-theme"]
test = ["anyio[trio]", "coverage[toml] (>=7)", "exceptiongroup (>=1.2.0)", "hypothesis (>=4.0)", "psutil (>=5.9)", "pytest (>=7.0)", "pytest-mock (>=3.6.1)", "trustme", "uvloop (>=0.17)"]
trio = ["trio (>=0.23)"]
[[package]] [[package]]
name = "apiclient" name = "apiclient"
version = "1.0.4" version = "1.0.4"
description = "Framework for making good API client libraries using urllib3." description = "Framework for making good API client libraries using urllib3."
category = "main"
optional = false optional = false
python-versions = "*" python-versions = "*"
files = [ files = [
@ -18,6 +42,7 @@ urllib3 = "*"
name = "certifi" name = "certifi"
version = "2023.11.17" version = "2023.11.17"
description = "Python package for providing Mozilla's CA Bundle." description = "Python package for providing Mozilla's CA Bundle."
category = "main"
optional = false optional = false
python-versions = ">=3.6" python-versions = ">=3.6"
files = [ files = [
@ -25,10 +50,115 @@ files = [
{file = "certifi-2023.11.17.tar.gz", hash = "sha256:9b469f3a900bf28dc19b8cfbf8019bf47f7fdd1a65a1d4ffb98fc14166beb4d1"}, {file = "certifi-2023.11.17.tar.gz", hash = "sha256:9b469f3a900bf28dc19b8cfbf8019bf47f7fdd1a65a1d4ffb98fc14166beb4d1"},
] ]
[[package]]
name = "exceptiongroup"
version = "1.2.0"
description = "Backport of PEP 654 (exception groups)"
category = "main"
optional = false
python-versions = ">=3.7"
files = [
{file = "exceptiongroup-1.2.0-py3-none-any.whl", hash = "sha256:4bfd3996ac73b41e9b9628b04e079f193850720ea5945fc96a08633c66912f14"},
{file = "exceptiongroup-1.2.0.tar.gz", hash = "sha256:91f5c769735f051a4290d52edd0858999b57e5876e9f85937691bd4c9fa3ed68"},
]
[package.extras]
test = ["pytest (>=6)"]
[[package]]
name = "h11"
version = "0.14.0"
description = "A pure-Python, bring-your-own-I/O implementation of HTTP/1.1"
category = "main"
optional = false
python-versions = ">=3.7"
files = [
{file = "h11-0.14.0-py3-none-any.whl", hash = "sha256:e3fe4ac4b851c468cc8363d500db52c2ead036020723024a109d37346efaa761"},
{file = "h11-0.14.0.tar.gz", hash = "sha256:8f19fbbe99e72420ff35c00b27a34cb9937e902a8b810e2c88300c6f0a3b699d"},
]
[[package]]
name = "httpcore"
version = "1.0.4"
description = "A minimal low-level HTTP client."
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "httpcore-1.0.4-py3-none-any.whl", hash = "sha256:ac418c1db41bade2ad53ae2f3834a3a0f5ae76b56cf5aa497d2d033384fc7d73"},
{file = "httpcore-1.0.4.tar.gz", hash = "sha256:cb2839ccfcba0d2d3c1131d3c3e26dfc327326fbe7a5dc0dbfe9f6c9151bb022"},
]
[package.dependencies]
certifi = "*"
h11 = ">=0.13,<0.15"
[package.extras]
asyncio = ["anyio (>=4.0,<5.0)"]
http2 = ["h2 (>=3,<5)"]
socks = ["socksio (>=1.0.0,<2.0.0)"]
trio = ["trio (>=0.22.0,<0.25.0)"]
[[package]]
name = "httpx"
version = "0.27.0"
description = "The next generation HTTP client."
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "httpx-0.27.0-py3-none-any.whl", hash = "sha256:71d5465162c13681bff01ad59b2cc68dd838ea1f10e51574bac27103f00c91a5"},
{file = "httpx-0.27.0.tar.gz", hash = "sha256:a0cb88a46f32dc874e04ee956e4c2764aba2aa228f650b06788ba6bda2962ab5"},
]
[package.dependencies]
anyio = "*"
certifi = "*"
httpcore = ">=1.0.0,<2.0.0"
idna = "*"
sniffio = "*"
[package.extras]
brotli = ["brotli", "brotlicffi"]
cli = ["click (>=8.0.0,<9.0.0)", "pygments (>=2.0.0,<3.0.0)", "rich (>=10,<14)"]
http2 = ["h2 (>=3,<5)"]
socks = ["socksio (>=1.0.0,<2.0.0)"]
[[package]]
name = "idna"
version = "3.6"
description = "Internationalized Domain Names in Applications (IDNA)"
category = "main"
optional = false
python-versions = ">=3.5"
files = [
{file = "idna-3.6-py3-none-any.whl", hash = "sha256:c05567e9c24a6b9faaa835c4821bad0590fbb9d5779e7caa6e1cc4978e7eb24f"},
{file = "idna-3.6.tar.gz", hash = "sha256:9ecdbbd083b06798ae1e86adcbfe8ab1479cf864e4ee30fe4e46a003d12491ca"},
]
[[package]]
name = "pylast"
version = "5.2.0"
description = "A Python interface to Last.fm and Libre.fm"
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "pylast-5.2.0-py3-none-any.whl", hash = "sha256:89c7c01ea9f08c83865999d8907835157a8096e77dd9dc23420246eb66cfcff5"},
{file = "pylast-5.2.0.tar.gz", hash = "sha256:bb046804ef56a0c18072c750d61a282d47ac102a3b0b9c44a023eaf5b0934b0a"},
]
[package.dependencies]
httpx = "*"
[package.extras]
tests = ["flaky", "pytest", "pytest-cov", "pytest-random-order", "pyyaml"]
[[package]] [[package]]
name = "python-dotenv" name = "python-dotenv"
version = "1.0.0" version = "1.0.0"
description = "Read key-value pairs from a .env file and set them as environment variables" description = "Read key-value pairs from a .env file and set them as environment variables"
category = "main"
optional = false optional = false
python-versions = ">=3.8" python-versions = ">=3.8"
files = [ files = [
@ -39,10 +169,35 @@ files = [
[package.extras] [package.extras]
cli = ["click (>=5.0)"] cli = ["click (>=5.0)"]
[[package]]
name = "sniffio"
version = "1.3.1"
description = "Sniff out which async library your code is running under"
category = "main"
optional = false
python-versions = ">=3.7"
files = [
{file = "sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2"},
{file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"},
]
[[package]]
name = "typing-extensions"
version = "4.10.0"
description = "Backported and Experimental Type Hints for Python 3.8+"
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "typing_extensions-4.10.0-py3-none-any.whl", hash = "sha256:69b1a937c3a517342112fb4c6df7e72fc39a38e7891a5730ed4985b5214b5475"},
{file = "typing_extensions-4.10.0.tar.gz", hash = "sha256:b0abd7c89e8fb96f98db18d86106ff1d90ab692004eb746cf6eda2682f91b3cb"},
]
[[package]] [[package]]
name = "urllib3" name = "urllib3"
version = "2.1.0" version = "2.1.0"
description = "HTTP library with thread-safe connection pooling, file post, and more." description = "HTTP library with thread-safe connection pooling, file post, and more."
category = "main"
optional = false optional = false
python-versions = ">=3.8" python-versions = ">=3.8"
files = [ files = [
@ -58,4 +213,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 = "48808e6f2c2ef0f18b46ffba3d13d473eacccad61c6bc369c655a127a7df48fd" content-hash = "5f0106196ba3a316e887fe2748bb5ca19159f2905d7678393b8ee51430f9ca45"

View file

@ -8,6 +8,7 @@ authors = ["Firepup Sixfifty <firepup650@gmail.com>"]
python = "^3.9" 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"
[tool.poetry.dev-dependencies] [tool.poetry.dev-dependencies]

93
timers.py Normal file
View file

@ -0,0 +1,93 @@
#!/usr/bin/python3
import bare, pylast
import config as conf
import random as r
from logs import log
from typing import Any, Callable, NoReturn
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()
def threadWrapper(data: dict) -> NoReturn:
if not data["noWrap"]:
while 1:
if ignoreErrors:
try:
data["func"](*data["args"])
except Exception:
Err = format_exc()
for line in Err.split("\n"):
log(line, "Thread", "WARN")
else:
try:
data["func"](*data["args"])
except Exception:
Err = format_exc()
for line in Err.split("\n"):
log(line, "Thread", "CRASH")
exit(1)
sleep(data["interval"])
log("Threaded loop broken", "Thread", "FATAL")
else:
data["func"](*data["args"])
exit(1)
def startThread(data: dict) -> Thread:
t = Thread(target=threadWrapper, args=(data,))
t.daemon = True
t.start()
return t
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 = {}
for name in threads:
data = threads[name]
running[name] = startThread(data)
if output:
log("All threads running, starting checkup loop", mgr)
while 1:
sleep(interval)
if output:
log("Checking threads", mgr)
for name in running:
t = running[name]
if is_dead(t):
if output:
log(f"Thread {name} has died, restarting", mgr, "WARN")
data = threads[name]
running[name] = startThread(data)
log("Thread manager loop broken", mgr, "FATAL")
exit(1)
def radio(instance: bare.bot) -> NoReturn:
lastTrack = ""
while 1:
try:
newTrack = instance.lastfmLink.get_user("Firepup650").get_now_playing()
if newTrack:
thisTrack = newTrack.__str__()
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")
except Exception:
Err = format_exc()
for line in Err.split("\n"):
instance.log(line, "WARN")
sleep(2)
instance.log("Thread while loop broken", "FATAL")
exit(1)
data: dict[str, dict[str, Any]] = {
"radio": {"noWrap": True, "func": radio, "args": []},
}