parent
4dc6f6507d
commit
661d689c9e
Binary file not shown.
@ -1,7 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
SCRIPT_PATH=$(dirname "$0")
|
||||
TOKEN_PATH="$SCRIPT_PATH/data/twitch-oauth.gpg"
|
||||
|
||||
TWITCH_OAUTH=$(gpg -q --decrypt $TOKEN_PATH ) \
|
||||
exec python3 "$SCRIPT_PATH/twitch-notify"
|
@ -1,111 +0,0 @@
|
||||
#!/usr/bin/python3
|
||||
|
||||
from subprocess import run
|
||||
from textwrap import dedent
|
||||
import os
|
||||
import sys
|
||||
import socket
|
||||
import re
|
||||
import signal
|
||||
|
||||
config = {
|
||||
"token": None,
|
||||
"nickname": 'ianmethyst',
|
||||
"channel": '#ianmethyst',
|
||||
"supress_own": True,
|
||||
"short_notification": True
|
||||
}
|
||||
|
||||
HOST = 'irc.chat.twitch.tv'
|
||||
PORT = 6667
|
||||
MESSAGE_REGEX = r':(.*)!.*@.* PRIVMSG #(.*) :(.*)'
|
||||
|
||||
sock = socket.socket()
|
||||
connected = False
|
||||
|
||||
|
||||
def get_token_from_env():
|
||||
"""Get OAUTH token from environment variables or exit"""
|
||||
|
||||
try:
|
||||
config['token'] = os.environ['TWITCH_OAUTH']
|
||||
except KeyError:
|
||||
helpText = """
|
||||
Get your OAUTH token here: https://twitchapps.com/tmi/
|
||||
Then pass it as an environment variable:
|
||||
TWITCH_OAUTH=<token> {__file__}"
|
||||
"""
|
||||
print(dedent(helpText))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def connect():
|
||||
"""Connect to IRC and authenticate"""
|
||||
|
||||
sock.connect((HOST, PORT))
|
||||
sock.send(f"PASS {config['token']}\n".encode('utf-8'))
|
||||
sock.send(f"NICK {config['nickname']}\n".encode('utf-8'))
|
||||
sock.send(f"JOIN {config['channel']}\n".encode('utf-8'))
|
||||
|
||||
|
||||
def pong(resp):
|
||||
"""Answer to server PING to keep socket connection alive"""
|
||||
|
||||
if resp.startswith('PING'):
|
||||
sock.send("PONG\n".encode('utf-8'))
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def wait_for_chat(resp):
|
||||
"""Wait until all welcome and user list messages are received"""
|
||||
|
||||
if "End of /NAMES list" in resp:
|
||||
global connected
|
||||
connected = True
|
||||
run(["notify-send",
|
||||
f"connected to {config['channel']} as {config['nickname']}"])
|
||||
|
||||
|
||||
def handle_message(resp):
|
||||
"""Parse the message and send it as a notification"""
|
||||
|
||||
print(resp)
|
||||
user, channel, message = re.search(MESSAGE_REGEX, resp).groups()
|
||||
|
||||
if user and channel and message:
|
||||
if config['short_notification']:
|
||||
notification_text = f"@{user}: {message}"
|
||||
else:
|
||||
notification_text = f"@{user} in #{channel}\n{message}"
|
||||
|
||||
if not (config['supress_own'] and config['nickname'] == user):
|
||||
run(["notify-send", notification_text])
|
||||
|
||||
|
||||
def handle_exit(sig, frame):
|
||||
"""Close socket connection before exit"""
|
||||
|
||||
print(f"Got {sig.name}, closing socket")
|
||||
sock.shutdown(socket.SHUT_RDWR)
|
||||
sock.close()
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Register signal handlers
|
||||
signal.signal(signal.SIGINT, handle_exit)
|
||||
signal.signal(signal.SIGTERM, handle_exit)
|
||||
|
||||
get_token_from_env()
|
||||
connect()
|
||||
|
||||
while True:
|
||||
resp = sock.recv(2048).decode('utf-8')
|
||||
|
||||
if len(resp) > 0:
|
||||
if not (pong(resp)):
|
||||
if connected:
|
||||
handle_message(resp)
|
||||
else:
|
||||
wait_for_chat(resp)
|
Loading…
Reference in new issue