302 lines
11 KiB
Python
302 lines
11 KiB
Python
import emoji
|
|
import json
|
|
import random
|
|
from dns.resolver import query
|
|
from email.utils import parseaddr
|
|
from mastodon import Mastodon
|
|
from smtplib import SMTP
|
|
from src import ModuleManager, utils
|
|
|
|
CHANNEL = "#tilderadio"
|
|
NOTIFY_CHANNELS = [CHANNEL, "#meta", "#team", "#club"]
|
|
LISTEN_URL = "https://tilderadio.org/listen"
|
|
SCHEDULE_URL = "https://tilderadio.org/schedule/"
|
|
SOURCE_URL = "https://tildegit.org/ben/bitbot-modules"
|
|
RADIO_API_BASE = "https://radio.tildeverse.org/api"
|
|
TOOT_CREDS_FILE = "/home/ben/.bitbot/tilderadio-toot.json"
|
|
SLOGANS = [
|
|
"The soundtrack in your cat's head",
|
|
"Your cat says she likes us better",
|
|
"radio for the people",
|
|
"radio by bots, for bots",
|
|
"On the Internet, no one knows you're a weakly superhuman AGI with a really good speech synth module.",
|
|
"We found your cat",
|
|
"cats.",
|
|
"a hacker's excuse for radio",
|
|
"0::1 is where the heart is",
|
|
"now with the longest slogan in human history",
|
|
"reticulating splines",
|
|
"Hold the shift",
|
|
"a very serious place for very serious happenings",
|
|
"sorry, this extension is invalid",
|
|
"This is the radio you were looking for",
|
|
"Not The Illuminati, We Promise",
|
|
"the radio is coming from INSIDE the house",
|
|
"Dont Sue Us!",
|
|
"it's radio minus radio",
|
|
"where your passwords are stored in plain text",
|
|
"Like radio. Only better.",
|
|
"tomasino 0wnz y0u!",
|
|
"that escalated quickly",
|
|
"music for your brain meats",
|
|
"003 Days Since Last Workplace Accident",
|
|
"*heavy breathing*",
|
|
"Dizzy? We stop the world from spinning",
|
|
"Where Disney Princesses Go Slumming",
|
|
'any slogan but "eat the poop or die"',
|
|
"that's not what she said!",
|
|
"not product placement, we promise! (tildestore.com)",
|
|
]
|
|
EMAIL_TEMPLATE = """
|
|
From: radiobot <radiobot@tilde.chat>
|
|
To: {nickname} <{email_addr}>
|
|
Subject: tilderadio: dj {dj} is now streaming!
|
|
|
|
{dj} just came on the air on tilderadio.org!
|
|
|
|
now playing: {now_playing}
|
|
|
|
have a listen here: {link}
|
|
|
|
~radiobot
|
|
"""
|
|
|
|
|
|
class Module(ModuleManager.BaseModule):
|
|
now_playing = ""
|
|
dj = ""
|
|
song = ""
|
|
listeners = 0
|
|
mastodon = None
|
|
is_online = False
|
|
|
|
def save_nowplaying(self, jsontxt):
|
|
if jsontxt == "":
|
|
data = utils.http.request(RADIO_API_BASE + "/nowplaying/1").json()
|
|
else:
|
|
data = json.loads(jsontxt)
|
|
|
|
self.is_online = data["live"]["is_live"]
|
|
self.dj = data["live"]["streamer_name"]
|
|
self.song = data["now_playing"]["song"]["text"]
|
|
self.listeners = data["listeners"]["current"]
|
|
|
|
def format_nowplaying(self):
|
|
ret = ""
|
|
if self.is_online:
|
|
ret = f"({self.dj}) "
|
|
ret += f"now playing: {self.song} ~~ {self.listeners} listeners"
|
|
return ret
|
|
|
|
def on_load(self):
|
|
self.save_nowplaying("")
|
|
|
|
with open(TOOT_CREDS_FILE, "r") as f:
|
|
creds = json.load(f)
|
|
self.mastodon = Mastodon(
|
|
client_id=creds["client_id"],
|
|
client_secret=creds["client_secret"],
|
|
access_token=creds["access_token"],
|
|
api_base_url=creds["base_url"],
|
|
)
|
|
|
|
@utils.hook("api.get.slogan", authenticated=False)
|
|
def httpslogan(self, event):
|
|
return random.choice(SLOGANS)
|
|
|
|
@utils.hook("api.post.radio")
|
|
def hook(self, event):
|
|
previous_dj = self.dj
|
|
previous_song = self.song
|
|
self.save_nowplaying(event["data"])
|
|
|
|
# new dj online!
|
|
if self.is_online:
|
|
|
|
server = self.bot.get_server_by_alias("tilde")
|
|
if server is not None:
|
|
|
|
# a different dj is online
|
|
if self.dj != "" and self.dj != previous_dj:
|
|
newdj_message = f"{self.dj} is now online playing {self.song}! tune in here now: {LISTEN_URL}"
|
|
|
|
# post toot to @tilderadio
|
|
self.mastodon.toot(newdj_message)
|
|
|
|
# send dm notifications
|
|
for n in self.bot.get_setting("tilderadio-subscriptions", []):
|
|
if n in server.users:
|
|
user = server.users[n]
|
|
self.events.on("send.stdout").call(
|
|
target=user,
|
|
module_name="Tilderadio",
|
|
server=server,
|
|
message=newdj_message,
|
|
)
|
|
|
|
# send email notifications
|
|
with SMTP("localhost") as smtp:
|
|
for nick, email in self.bot.get_setting(
|
|
"tilderadio-email-subscriptions", {}
|
|
).items():
|
|
smtp.sendmail(
|
|
"radiobot@tilde.chat",
|
|
email,
|
|
EMAIL_TEMPLATE.format(
|
|
nickname=nick,
|
|
email_addr=email,
|
|
dj=self.dj,
|
|
now_playing=self.format_nowplaying(),
|
|
link=LISTEN_URL,
|
|
),
|
|
)
|
|
|
|
# post to all registered channels
|
|
for channel_name in NOTIFY_CHANNELS:
|
|
if channel_name in server.channels:
|
|
channel = server.channels[channel_name]
|
|
|
|
self.events.on("send.stdout").call(
|
|
target=channel,
|
|
module_name="Tilderadio",
|
|
server=server,
|
|
message=newdj_message,
|
|
)
|
|
|
|
# post new songs while streaming
|
|
if self.song != "" and self.song != previous_song:
|
|
if CHANNEL in server.channels:
|
|
channel = server.channels[CHANNEL]
|
|
|
|
self.events.on("send.stdout").call(
|
|
target=channel,
|
|
module_name="Tilderadio",
|
|
server=server,
|
|
message=utils.irc.color(
|
|
self.format_nowplaying(), utils.consts.GREEN
|
|
),
|
|
)
|
|
|
|
@utils.hook("received.command.np", alias_of="nowplaying")
|
|
@utils.hook("received.command.nowplaying")
|
|
@utils.kwarg("help", "show the current song on tilderadio")
|
|
def nowplaying(self, event):
|
|
event["stdout"].write(self.format_nowplaying())
|
|
|
|
@utils.hook("received.command.schedule")
|
|
@utils.kwarg("help", "show a link to the tilderadio schedule")
|
|
def schedule(self, event):
|
|
event["stdout"].write(f"you can find the schedule here: {SCHEDULE_URL}")
|
|
|
|
@utils.hook("received.command.un", alias_of="upnext")
|
|
@utils.hook("received.command.upnext")
|
|
@utils.kwarg("help", "show who's up next to stream")
|
|
def upnext(self, event):
|
|
js = utils.http.request(RADIO_API_BASE + "/station/1/schedule?rows=1").json()
|
|
if len(js) < 1:
|
|
event["stdout"].write("nothing scheduled")
|
|
else:
|
|
data = js[0]
|
|
start = utils.datetime.parse.iso8601(data["start"])
|
|
now = utils.datetime.utcnow()
|
|
total_secs = (start - now).total_seconds()
|
|
event["stdout"].write(
|
|
"dj {} is up next at {} UTC in {}!".format(
|
|
data["name"],
|
|
utils.datetime.format.datetime_human(start),
|
|
utils.datetime.format.to_pretty_time(total_secs),
|
|
)
|
|
)
|
|
|
|
@utils.hook("received.command.subscribe")
|
|
@utils.kwarg("help", "sign up to get a direct message when a new dj goes online")
|
|
def subscribe(self, event):
|
|
subs = self.bot.get_setting("tilderadio-subscriptions", [])
|
|
nick = event["user"].nickname
|
|
if not nick in subs:
|
|
subs.append(nick)
|
|
self.bot.set_setting("tilderadio-subscriptions", subs)
|
|
event["stdout"].write("i'll send you a message when a dj goes online")
|
|
else:
|
|
event["stdout"].write("you're already subscribed!")
|
|
|
|
@utils.hook("received.command.unsubscribe")
|
|
@utils.kwarg("help", "unsubscribe from dj announcements")
|
|
def unsubscribe(self, event):
|
|
subs = self.bot.get_setting("tilderadio-subscriptions", [])
|
|
if nick in subs:
|
|
subs.remove(nick)
|
|
self.bot.set_setting("tilderadio-subscriptions", subs)
|
|
event["stdout"].write("ok, i'll no longer send you dj announcements")
|
|
else:
|
|
event["stdout"].write("you weren't subscribed. unable to remove")
|
|
|
|
@utils.hook("received.command.emailsubscribe")
|
|
@utils.kwarg("help", "sign up to get an email when a dj goes online")
|
|
@utils.spec("!<email>word")
|
|
def email_subscribe(self, event):
|
|
email = event["spec"][0]
|
|
_, addr = parseaddr(email)
|
|
if "@" in addr:
|
|
domain = addr.rsplit("@", 1)[-1]
|
|
try:
|
|
mxfound = bool(query(domain, "MX"))
|
|
subs = self.bot.get_setting("tilderadio-email-subscriptions", {})
|
|
nick = event["user"].nickname
|
|
subs.update({nick: addr})
|
|
self.bot.set_setting("tilderadio-email-subscriptions", subs)
|
|
|
|
event["stdout"].write(
|
|
f"ok, i'll send an email to you at {addr} when a dj goes online"
|
|
)
|
|
except:
|
|
event["stdout"].write(
|
|
f"your specified mail server {domain} is not configured to receive mail"
|
|
)
|
|
else:
|
|
event["stdout"].write(f"invalid email address")
|
|
|
|
@utils.hook("received.command.emailunsubscribe")
|
|
@utils.kwarg("help", "stop sending email notifications")
|
|
def email_unsubscribe(self, event):
|
|
subs = self.bot.get_setting("tilderadio-email-subscriptions", {})
|
|
nick = event["user"].nickname
|
|
if nick in subs:
|
|
subs.pop(nick)
|
|
self.bot.set_setting("tilderadio-email-subscriptions", subs)
|
|
event["stdout"].write("ok i'll stop sending you emails")
|
|
else:
|
|
event["stdout"].write("you weren't subscribed. unable to remove")
|
|
|
|
@utils.hook("received.command.dj")
|
|
@utils.kwarg("help", "check if someone is currently streaming")
|
|
def showdj(self, event):
|
|
if self.dj == "":
|
|
message = "no one is currently on the air"
|
|
else:
|
|
message = f"dj {self.dj} is now streaming!"
|
|
event["stdout"].write(message)
|
|
|
|
@utils.hook("received.command.slogan")
|
|
@utils.kwarg("help", "get a random tilderadio slogan")
|
|
def slogan(self, event):
|
|
event["stdout"].write(random.choice(SLOGANS))
|
|
|
|
@utils.hook("received.command.toot")
|
|
@utils.kwarg("require_mode", "v")
|
|
@utils.kwarg("help", "send a toot from the tilderadio mastodon account")
|
|
@utils.spec("!<status>string")
|
|
def toot(self, event):
|
|
status = event["spec"][0]
|
|
if len(status) > 8:
|
|
status = emoji.emojize(status, use_aliases=True)
|
|
uri = self.mastodon.toot(status)["uri"]
|
|
event["stdout"].write(f"tooted!: {uri}")
|
|
|
|
@utils.hook("received.command.radiochannels")
|
|
@utils.kwarg("help", "show tilderadio notification channels")
|
|
def radiochannel(self, event):
|
|
event["stdout"].write(
|
|
f"my primary channel is {CHANNEL} and i will send dj announcements to {NOTIFY_CHANNELS}"
|
|
)
|