#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ twitch-irl-docker Copyright (C) 2022 gpkvt This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see . """ import importlib import time import random import logging import memcache import irc.bot from pprint import pprint from lib import obs as obsremote try: from vol import irc_commands logging.info('Successfully imported external commands file') ext_commands = True except Exception as e: logging.warning('No valid external commands file found') logging.warning(e) ext_commands = False ignore = [ "StreamElements", "Nightbot", "Streamlabs", "Moobot" ] greeter = [] last_online_message = 10 last_offline_message = 10 last_bitrate_msg = 10 class TwitchBot(irc.bot.SingleServerIRCBot): def __init__(self, conf): self.channel = '#' + conf['IRC_CHANNEL'] self.cooldown = conf['IRC_COOLDOWN'] self.cooldownTime = time.time() - int(conf['IRC_COOLDOWN']) self.conf = conf self._active = True server = 'irc.chat.twitch.tv' port = 6667 logging.info('Connecting to ' + str(server) + ' on port ' + str(port) + '...') irc.bot.SingleServerIRCBot.__init__(self, [(server, port, 'oauth:' + conf['IRC_OAUTH_TOKEN'])], conf['IRC_USERNAME'], conf['IRC_USERNAME']) def start(self): logging.debug('Starting IRC-Bot...') self._connect() while self._active: self.reactor.process_once(timeout=0.2) def die(self): logging.debug('Stopping IRC-Bot...') self.connection.disconnect("Disconnect") self._active = False def on_welcome(self, c, e): logging.info('Joining ' + str(self.channel)) c.cap('REQ', ':twitch.tv/membership') c.cap('REQ', ':twitch.tv/tags') c.cap('REQ', ':twitch.tv/commands') c.join(self.channel) logging.info('Bot is ready') c.privmsg(self.channel, self.conf['IRC_BOT_EMOTE'] + ' A wild IRL bot appears!') def on_pubmsg(self, c, e): global greeter if e.arguments[0][:1] == '!': cmd = e.arguments[0].split(' ')[0][1:] self.do_command(e, cmd) elif self.conf['IRC_GREETER_WELCOME']: logging.debug('Check for greet') for tag in e.tags: if tag['key'] == 'display-name': nickname = tag['value'] logging.debug('Found display-name') if nickname not in greeter and nickname not in ignore: c.privmsg(self.channel, str(self.conf['IRC_GREETER_WELCOME']) + " " + str(nickname) + " " + str(self.conf['IRC_GREETER_EMOTE'])) logging.debug('Greetings send') greeter.append(nickname) else: logging.debug('Already greeted') else: logging.debug('Could not found display-name, skipping greet') return def display_size(self, bytes_in): units = [ 'bytes', 'KB', 'MB', 'GB', 'TB' ] step_unit = 1024 for unit in units: if bytes_in < step_unit: return "%3.1f %s" % (bytes_in, unit) bytes_in /= step_unit def display_time(self, seconds, granularity=2): intervals = ( ('weeks', 604800), # 60 * 60 * 24 * 7 ('days', 86400), # 60 * 60 * 24 ('hours', 3600), # 60 * 60 ('minutes', 60), ('seconds', 1), ) result = [] for name, count in intervals: value = int(seconds) // count if value: seconds -= int(value) * count if int(value) == 1: name = name.rstrip('s') result.append("{} {}".format(value, name)) return ', '.join(result[:granularity]) def do_command(self, e, cmd): mc = memcache.Client(['localhost:11211'], debug=0) obs_remote = obsremote.ObsRemote(self.conf) obs_remote.connect() c = self.connection currentTime = time.time() logging.info('Got command') if float(currentTime) - float(self.cooldownTime) >= float(self.cooldown) or self.get_role(e.tags) == "moderator" or self.get_role(e.tags) == "broadcaster": logging.debug(e.tags) if cmd == "bitrate": try: current_bitrate = mc.get('RTMP_BITRATE') c.privmsg(self.channel, str(self.conf['IRC_BITRATE_MSG_PRE']) + ": " + str(current_bitrate) + " " + str(self.conf['IRC_BITRATE_MSG_SUF'])) except: c.privmsg(self.channel, "There was an error fetching the bitrate, sorry. Please try again, later.") if cmd == "reload": logging.info('Got RELOAD command') role = self.get_role(e.tags) if role == "broadcaster" or role == "moderator": if mc.get('OBS_CURRENT_SCENE') == self.conf['OBS_LIVE_SCENE']: logging.debug('Reloading LIVE scene') obs_remote.set_scene(self.conf['OBS_BRB_SCENE']) obs_remote.set_scene(self.conf['OBS_LIVE_SCENE']) else: logging.info('Not in LIVE scene, will not reload') else: logging.debug('User was not authorized for RELOAD command') if cmd == "brb": logging.info('Got BRB command') role = self.get_role(e.tags) if role == "broadcaster" or role == "moderator": logging.debug('Switching to FORCED_BRB_SCENE') obs_remote.set_scene(self.conf['OBS_FORCED_BRB_SCENE']) else: logging.debug('User was not authorized for BRB command') if cmd == "live": logging.info('Got LIVE command') role = self.get_role(e.tags) if role == "broadcaster": logging.debug('Switching to LIVE_SCENE') obs_remote.set_scene(self.conf['OBS_LIVE_SCENE']) else: logging.debug('User was not authorized for LIVE command') if cmd == "irlstatus": logging.info('Got IRLSTATUS command') role = self.get_role(e.tags) if role == "broadcaster" or role == "moderator": try: uptime = self.display_time(float(mc.get('RTMP_UPTIME')), 5) connections = float(mc.get('RTMP_NACCEPTED')) / 2 if mc.get('RTMP_NACCEPTED') else "an unknown number of" bytes_in = self.display_size(float(mc.get('RTMP_BYTES_IN_SUM'))) last_reconnect = self.display_time(float(mc.get('RTMP_LAST_RECONNECT'))) c.privmsg(self.channel, "The current uptime is: " + str(uptime) + " there were " + str(connections) + " reconnects. A total of " + str(bytes_in) + " were transferred. The last disconnect was "+ str(last_reconnect) +" ago.") except: c.privmsg(self.channel, "There was an error fetching the status, sorry. Please try again, later.") if cmd == "irlclear": logging.info('Got IRLclear command') role = self.get_role(e.tags) if role == "broadcaster" or role == "moderator": self.clear_greeter() c.privmsg(self.channel, "Greeter was cleared.") if ext_commands: importlib.reload(irc_commands) if cmd in irc_commands.commands: logging.info('Got Custom command') try: roles = irc_commands.commands[cmd]['roles'] text = irc_commands.commands[cmd]['text'] logging.debug(roles) logging.debug(text) if self.get_role(e.tags) in roles or "any" in roles: c.privmsg(self.channel, str(random.choice(text))) except Exception as e: logging.warning('Error during processing of custom command') logging.warning(e) else: logging.info('Still in cooldown period') return def get_role(self, tags): for tag in tags: if tag['key'] == "badges": if tag['value']: badges = tag['value'].split(',') for badge in badges: role = badge.split('/') logging.debug('Role: ' + str(role[0])) return role[0] logging.debug('Could not get role') return False def clear_greeter(self): global greeter greeter = [] return def main(): from lib import config conf = config.get_config() mc = memcache.Client(['localhost:11211'], debug=0) while not mc.get('OBS_STREAMING_STATUS'): logging.info('OBS is not streaming, delaying bot init, will check again in 10 seconds...') time.sleep(10) logging.basicConfig(format='%(asctime)s %(module)s %(levelname)s: %(message)s') logging.getLogger().setLevel(conf['LOG_LEVEL']) bot = TwitchBot(conf) bot.start() if __name__ == "__main__": main()