XMPP_webhooks/hooks.py

#!/var/www/html/webhooks/.venv/bin/python
# File name: Hooks.py
# Date created: 09/16/2022
# Date last modified: 10/15/2022
# Python Version: 3.9.2
# Copyright © 2022 DeadTOm
# Description: Webhooks for Owncast to send notifications to Discord, Mastodon and Twitter, and to interact with
# my Minecraft server
# TODO: Make routes for various chat and video links

try:
    from names import *
    from config import *
    from auth import *
    from flask import Flask, jsonify, current_app, request
    import requests
    import logging
    import time
    from twython import Twython, TwythonError
    import socket
    from mcstatus import JavaServer
except Exception as import_error:  # Log any errors loading modules, and try to keep running
    fail_log = open(log_location, 'a')
    fail_log.write(f'------{import_error}------\n')
    fail_log.close()

logging.basicConfig(filename='/var/www/html/webhooks.log', level=logging.INFO)
#logging.basicConfig(filename='/var/www/html/webhooks.log', level=logging.DEBUG)
#logging.basicConfig(level=logging.DEBUG)

current_names = []  # Initialize empty list to hold current Twitter and Mastodon names
testing = 0  # Are we testing? 1 for testing. 0 for live.

twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
get_t_info = twitter.verify_credentials()  # Get current Twitter name

get_m_info = requests.get('https://fosstodon.org/api/v1/accounts/verify_credentials', headers={'Authorization': m_api_key, 'Accept': 'application/json'}) # Get Mastodon account information
get_m_info = get_m_info.json()  # Jsonify the response

app = Flask(__name__)

def get_now():  # This creates and returns a time stamp
    now = str(time.strftime("%Y/%m/%d %H:%M:%S"))
    return now

logging.info(f'\n\n\n\n{get_now()} - Webhook called.\n')

def streaming_t_name():  # Change Twitter name to show we're streaming'
    get_t_info = twitter.verify_credentials()  # Get current Twitter name
    current_t_name = f't_name = \'{get_t_info["name"]}\''  # Create Twitter list entry
    current_names.append(current_t_name)  # Append to list of current names
    logging.info(f'{get_now()} - Current Twitter name is \"{get_t_info["name"]}\".')

    set_t_name = twitter.update_profile(name=t_streaming_name)

    logging.info(f'{get_now()} - Twitter name is now {set_t_name["name"]}.')

def streaming_m_name():  # Change Mastodon name to show we're streaming
    get_m_info = requests.get('https://fosstodon.org/api/v1/accounts/verify_credentials', headers={'Authorization':     m_api_key, 'Accept': 'application/json'}) # Get Mastodon account information
    get_m_info = get_m_info.json()  # Jsonify the response
    current_m_name = f'm_name = \'{get_m_info["display_name"]}\''  # Create Mastodon list entry
    current_names.append(current_m_name)  # Append to list of current names
    logging.info(f'{get_now()} - Current Mastodon name is \"{get_m_info["display_name"]}\".')

    set_m_name = requests.patch('https://fosstodon.org/api/v1/accounts/update_credentials', params={'display_name': m_streaming_name}, headers={'Authorization': m_api_key, 'Accept': 'application/json'}) # Set Mastodon display name
    set_m_name = set_m_name.json()  # Jsonify the response

    logging.info(f'{get_now()} - Mastodon name is now \"{set_m_name["display_name"]}\".')

def reg_t_name():  # Change Twitter name to regular name
    set_t_name = twitter.update_profile(name=t_name)

    logging.info(f'{get_now()} - Twitter name is now {set_t_name["name"]}.')

def reg_m_name():  # Change Mastodon name to regular name
    set_m_name = requests.patch('https://fosstodon.org/api/v1/accounts/update_credentials', params={'display_name': m_name}, headers={'Authorization': m_api_key, 'Accept': 'application/json'}) # Set Mastodon display name
    set_m_name = set_m_name.json()  # Jsonify the response

    logging.info(f'{get_now()} - Mastodon name is now \"{set_m_name["display_name"]}\".')

def write_current_names():  # Write current names to names.py
    logging.info(f'{get_now()} - Storing {current_names}.')
    file = open("names.py", "w")
    for name in current_names:
        file.write(f'{name}\n')
    file.close()

def mc_chat(mc_msg):  # Send chat message to Minecraft chat
    logging.info(f'{get_now()} - Checking Minecraft server for players.')
    mc_server = JavaServer.lookup('mc.deadtom.me:25565')
    query = mc_server.query()  # Query Minecraft server for players
    cur_players = query.players.names
    if '_DeadTOm_' in cur_players:  # If I'm on the server, send message
        logging.info(f'{get_now()} - DeadTOm is on the server, sending message.')
        logging.info(f'{get_now()} - Connecting...')
        sock_host = 'mc.deadtom.me'
        sock_port = 6791
        mySocket = socket.socket()
        mySocket.connect((sock_host, sock_port))
        time.sleep(1)
        logging.info(f'{get_now()} - Connected. Sending {mc_msg}...')
        mySocket.send(mc_msg.encode())
        logging.info(f'{get_now()} - sent.')
        mySocket.close()
    else:  # If I'm not on the server, don't send the message
        logging.info(f'{get_now()} - DeadTOm is not on the server, so not sending message.')

def set_hashtags(title):  # Sets up hash tags to be appended to social media
    logging.info(f'{get_now()} - Examining stream title \"{title}\" to apply hashtags.')
    check_title = title.lower()
    default_tags = '#Owncast '
    tags = ''
    # tag_dict located in config.py
    for tag in tag_dict.keys():  # Iterate through each dict entry, and check for hashtag triggers
        if tag in check_title:
            print(f'Found {tag}, adding {tag_dict[tag]} to hashtags.')
            tags = f'{tags}{tag_dict[tag]} '
    tags = f'{tags}#Owncast #NSFW'  # Adding NSFW tag, just for good measure
    logging.info(f'{get_now()} - Adding {tags} to title.')
    return tags

def social_post(msg, discmsg):  # Post to Mastodon
    # Send to Mastodon
    logging.info(f'{get_now()} - Posting to Mastodon.')
    response = requests.post(m_api_url,
                             params={'status': msg},
                             headers={'Authorization': m_api_key,
                                      'visibility': 'public',
                                      'Accept': 'application/json'})
    json_response = response.json()
    for line in json_response:
        logging.debug(f'{get_now()} - API returned: {line}: {json_response[line]}')

    # Send to Twitter
    twitter.update_status(status=msg)  # Tweet the message
    logging.info(f'{get_now()} - Posted to Twitter.')

    # Send to Discord
    # for all params, see https://discordapp.com/developers/docs/resources/webhook#execute-webhook
    data = {
        'content': discmsg
    }
    talkback = requests.post(discordwebhookurl, json=data)
    if 200 <= talkback.status_code < 300:
        logging.info(f'{get_now()} - Discord message sent {talkback.status_code}')
    else:
        logging.info(
            f'{get_now()} - Discord message not sent with {talkback.status_code}, response:\n{talkback.json()}')

@app.route('/stream_started/', methods=["POST"])
def start():
    logging.info(f'{get_now()} - stream_started request')
    raw_data = request.get_json(force=True)  # Get the raw data
    event_data = raw_data['eventData']
    stream_title = event_data['streamTitle']
    hashtags = set_hashtags(stream_title.lower())
    msg = f'I\'m streaming on Owncast, at https://owncast.deadtom.me. {stream_title} {hashtags}'
    logging.debug(f'{get_now()} - Constructed Mastodon/Twitter message: {msg}')
    discmsg = f'DeadTOm is streaming on Owncast, at https://owncast.deadtom.me. {stream_title}'
    if testing != 1:  # If we're testing, don't actually send out notification
        social_post(msg, discmsg)
        streaming_m_name()
        streaming_t_name()
        write_current_names()
    else:
        logging.info(f'-----------------------\n\n'
                     f'{get_now()} - Currently running in test mode. We are streaming, but no notifications are being sent to social media.'
                     f'\n\n---------------------------------')

    return jsonify({"Data": raw_data,})

@app.route('/stream_stopped/', methods=["POST"])
def stop():
    logging.info(f'{get_now()} - stream_stopped request')
    raw_data = request.get_json(force=True)  # Get the raw data
    event_data = raw_data['eventData']
    stream_title = event_data['streamTitle']
    hashtags = set_hashtags(stream_title.lower())  # Make title lower case, for comparison with list of tags
    msg = f'All done streaming, for now. Maybe catch me next time on Owncast, at https://owncast.deadtom.me.\n\n{hashtags}'
    discmsg = f'DeadTOm is done streaming.'
    if testing != 1:  # If we're testing, don't actually send out notification
        social_post(msg, discmsg)

    return jsonify({"Data": raw_data,})

@app.route('/user_joined/', methods=["POST"])
def joined():
    logging.info(f'{get_now()} - user_joined request')
    raw_data = request.get_json(force=True)  # Get the raw data
    event_data = raw_data['eventData']
    chatter_name = event_data['user']['displayName']
    ownchat_msg = f'{chatter_name} joined the chat.'
    logging.info(f'{get_now()} - {ownchat_msg}')
    mc_chat(ownchat_msg)
    logging.info(f'{get_now()} - Sent to Minecraft server.')

    return jsonify({"Data": raw_data,})

@app.route('/name_changed/', methods=["POST"])
def changed():
    logging.info(f'{get_now()} - name_changed request')
    raw_data = request.get_json(force=True)  # Get the raw data
    event_data = raw_data['eventData']
    chatter_old_name = event_data['user']['previousNames']
    chatter_new_name = event_data['user']['displayName']
    last_name = len(chatter_old_name) - 1  # Get last name in previousNames list
    chatter_old_name = event_data['user']['previousNames'][last_name]
    ownchat_msg = f'{chatter_old_name} changed their name to {chatter_new_name}'
    logging.debug(f'{get_now()} - {type}\n{raw_data}')
    logging.info(f'{get_now()} - {ownchat_msg}')
    mc_chat(ownchat_msg)
    logging.info(f'{get_now()} - Sent to Minecraft server.')

    return jsonify({"Data": raw_data,})

@app.route('/message_sent/', methods=["POST"])
def sent():
    logging.info(f'{get_now()} - message_sent request')
    raw_data = request.get_json(force=True)  # Get the raw data
    event_data = raw_data['eventData']
    chatter_name = event_data['user']['displayName']
    chat_msg = event_data['rawBody']
    ownchat_msg = f'{chatter_name} on Owncast says: {chat_msg}'
    logging.info(f'{get_now()} - Chat message: \"{ownchat_msg}\".')
    mc_chat(ownchat_msg)
    logging.info(f'{get_now()} - Sent to Minecraft server.')

    return jsonify({"Data": raw_data,})


if __name__ == '__main__':
    try:
        app.run()
    except Exception as main_error:
        logging.info(f'{get_now()} - {main_error}')