projects-jenz/stoat_map_notifications/python/app.py

#!/home/nonroot/stoat_map_notifications/venv/bin/python3
from settings import get_connection_unloze_playtime, stoat_token, stoat_url_map_notifications, stoat_url_50_last_messages, ips
import requests
import re
import traceback
import json
from flask import Flask
from flask import request
from flask_cors import CORS
from werkzeug.middleware.proxy_fix import ProxyFix
import threading
import time
import difflib

app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1) 
CORS(app)

map_images = []

#runs every 3 minutes to inform about a new map

def insert_maps(fixed_param):
    with get_connection_unloze_playtime() as conn:
        with conn.cursor() as cur:
            placeholders = ','.join(['(%s)'] * len(fixed_param))
            flattened_values = [row for row in fixed_param]
            sql_statement = f"""
                INSERT IGNORE INTO unloze_playtimestats.map_notifications 
                    (mapname) 
                    VALUES {placeholders}
            """
            cur.execute(sql_statement, flattened_values)
            conn.commit()

def get_author_map_notifcations(author):
    with get_connection_unloze_playtime() as conn:
        with conn.cursor() as cur:
            sql_statement = f"""
                SELECT mapname 
                    FROM unloze_playtimestats.map_notifications
                WHERE JSON_CONTAINS(users_to_notify, JSON_QUOTE(%s));
            """
            cur.execute(sql_statement, [author])
            return cur.fetchall()

def handle_map_notification(map_notify, author, thing):
    with get_connection_unloze_playtime() as conn:
        with conn.cursor() as cur:
            sql_statement = f"""
                select users_to_notify  
                    from unloze_playtimestats.map_notifications
                where mapname = %s
            """
            cur.execute(sql_statement, [map_notify])
            res = cur.fetchone()[0]
            if res:
                users = json.loads(res)
            else:
                users = []

            if thing == "added":
                if author not in users:
                    users.append(author) #add user to map notifications for map
                else:
                    return #user is already in map notification for the map
            else:
                if author not in users:
                    return #user is not in map notification for the map
                else:
                    users.remove(author) #remove the user from map notifications for the map

            sql_statement = f"""
                UPDATE unloze_playtimestats.map_notifications 
                    SET users_to_notify = %s 
                WHERE mapname = %s
            """
            cur.execute(sql_statement, [json.dumps(users), map_notify])
            conn.commit()

def stoat_send_map_notification_list(show_map_notifications):
    message = f""
    for author in show_map_notifications:
        author_notifications = get_author_map_notifcations(author)
        message += f"<@{author}> your map notifications are:\n"
        for author_notification in author_notifications:
            message += f"{author_notification[0]} "
        message += "\n"

    headers = {
        "x-bot-token": f"{stoat_token}",
        "Content-Type": "application/json"
    }
    data = {
        "content": message
    }
    response = requests.post(stoat_url_map_notifications, headers=headers, json=data)

def stoat_send_invalid_map_notifications(invalid_map_notification_cmds):
    message = f""
    for cmd in invalid_map_notification_cmds:
        map_notify = cmd[0]
        author = cmd[1]

        message += f"<@{author}> you need to write the full map name correct you dumbass. {map_notify} is not valid input. example: \nadd ze_azathoth_v1_css\nremove ze_azathoth_v1_css\nshowlist -> lists all your map notifications.\n"
    headers = {
        "x-bot-token": f"{stoat_token}",
        "Content-Type": "application/json"
    }
    data = {
        "content": message
    }
    response = requests.post(stoat_url_map_notifications, headers=headers, json=data)

def stoat_send_valid_map_notifications(thing, valid_map_notification_cmds):
    message = f""
    for cmd in valid_map_notification_cmds:
        map_notify = cmd[0]
        author = cmd[1]

        message += f"<@{author}> {thing} map notification for {map_notify}\n"
        handle_map_notification(map_notify, author, thing)
    headers = {
        "x-bot-token": f"{stoat_token}",
        "Content-Type": "application/json"
    }
    data = {
        "content": message
    }
    response = requests.post(stoat_url_map_notifications, headers=headers, json=data)

def check_new_map_notification_status(fixed_param, last_msg_id):
    headers = {
        "x-bot-token": stoat_token 
    }

    valid_map_notification_cmds_add= []
    valid_map_notification_cmds_remove= []
    invalid_map_notification_cmds = []
    show_map_notifications = []
    #if first time we take the most recent message as starting point
    if not last_msg_id:
        url = stoat_url_50_last_messages.replace('/messages?limit=5&include_users=true&sort=Oldest&after=', '/messages?limit=5&include_users=true')
        response = requests.get(url, headers=headers)
    else:
        response = requests.get(stoat_url_50_last_messages + last_msg_id, headers=headers)

    try:
        data = response.json()
    except:
        #well that should not happen
        err = traceback.format_exc()
        print("err: ", err)
        return None, None, None, None, None
        
    for index, msg in enumerate(data['messages']):
        #print(msg)
        msg_id = msg['_id']
        author = msg['author']
        last_msg_id = msg_id
        dont_process_bot_msg = False
        for user in data['users']:
            if user['_id'] == author and 'bot' in user:
                dont_process_bot_msg = True
                break
        if dont_process_bot_msg: continue

        try:
            content = msg['content']
            if content.startswith("add "):
                content = content.split("add ")[1]
                if content in fixed_param:
                    valid_map_notification_cmds_add.append([content, author])
                else:
                    invalid_map_notification_cmds.append([content, author])

            elif content.startswith("remove "):
                content = content.split("remove ")[1]
                if content in fixed_param:
                    valid_map_notification_cmds_remove.append([content, author])
                else:
                    invalid_map_notification_cmds.append([content, author])
            elif content == "showlist":
                show_map_notifications.append(author)
            else:
                invalid_map_notification_cmds.append([content, author])
        except:
            err = traceback.format_exc()
            print("err: ", err)
            print("msg in error: ", msg)

    #print('valid_map_notification_cmds_add: ', valid_map_notification_cmds_add)
    #print('valid_map_notification_cmds_remove: ', valid_map_notification_cmds_remove)
    #print('invalid_map_notification_cmds: ', invalid_map_notification_cmds)
    #print('last_msg_id')

    return valid_map_notification_cmds_add, valid_map_notification_cmds_remove, invalid_map_notification_cmds, last_msg_id, show_map_notifications

def send_post_notify_message_to_stoat(list_of_people_to_notify, content, infomap):
    list_of_people_to_notify = list_of_people_to_notify[0]
    message = f"{content}\n\n"
    try:
        list_of_people_to_notify = json.loads(list_of_people_to_notify)
        message += "People to notify: "
    except:
        #in case it fails there should be nobody to notify
        list_of_people_to_notify = []
        pass
    #print('list_of_people_to_notify: ', list_of_people_to_notify)
    for people in list_of_people_to_notify:
        if not people or people == "[]": continue
        people = people.replace('["', '').replace('"]', '')
        #print('people: ', people)
        message += f"<@{people}> "
    if not message.endswith("\n"):
        message += "\n"
    message += f"commands: showlist, add {infomap}, remove {infomap}"
    headers = {
        "x-bot-token": f"{stoat_token}",
        "Content-Type": "application/json"
    }

    match = difflib.get_close_matches(infomap, map_images, n=1, cutoff=0.6)
    thumbnail_name = match[0] if match else "ze_atix_panic"

    image_url = f"https://vauff.com/mapimgs/240/{thumbnail_name}.jpg"
    data = {
        "content": image_url,
        "embeds": [{
            "description": message,
        }]
    }
    response = requests.post(stoat_url_map_notifications, headers=headers, json=data)

def update_last_msg_id(msg_id):
    with open("last_msg_id.txt", 'w') as f:
        f.write(msg_id)

def get_last_msg_id_read():
    with open("last_msg_id.txt", 'r') as f:
        return f.readline()

def notify_of_new_map(infomap):
    with get_connection_unloze_playtime() as conn:
        with conn.cursor() as cur:
            sql_statement = f"""
                select users_to_notify 
                    from unloze_playtimestats.map_notifications
                    where mapname = %s
            """
            cur.execute(sql_statement, [infomap])
            res = cur.fetchone()
            return res

#nginx used for reserve proxy
@app.route('/', methods = ['POST'])
def got_new_map_to_announce():
    real_ip = request.headers.get('X-Real-IP', request.remote_addr)
    #ipv4 and ipv6 checks
    if real_ip != ips[0] and not real_ip.startswith(ips[1]):
        return "invalid"
    try:
        content = request.get_json()["content"]
        #print('content: ', content)
        infomap = content.split("Map changed to: ")[1].split("\nPlayerCount:")[0]
        list_of_people_to_notify = notify_of_new_map(infomap)
        send_post_notify_message_to_stoat(list_of_people_to_notify, content, infomap)
    except:
        err = traceback.format_exc()
    return ""

def main_loop():
    #get all maps
    r = requests.get("https://uk-fastdl.unloze.com/css_ze/maps/")
    bz2_files = re.findall(r'([^\s]+\.bz2)', r.text)

    fixed = []
    for bz2 in bz2_files:
        bz2 = bz2.replace('href="', '')
        bz2 = bz2.split('.bsp')[0]
        fixed.append(bz2)

    r_map_images = requests.get("https://vauff.com/mapimgs/list.php")
    d = r_map_images.json().get("240", [])
    for map_img in d:
        map_images.append(map_img)

    #insert maps if they are not already.
    insert_maps(fixed)

    #check last 5 messages for new map notification updates
    while True:
        last_msg_id = get_last_msg_id_read()
        valid_map_notification_cmds_add, valid_map_notification_cmds_remove, invalid_map_notification_cmds, last_msg_id, show_map_notifications = check_new_map_notification_status(fixed, last_msg_id)

        if valid_map_notification_cmds_add:
            stoat_send_valid_map_notifications("added", valid_map_notification_cmds_add)
        if valid_map_notification_cmds_remove:
            stoat_send_valid_map_notifications("removed", valid_map_notification_cmds_remove)
        if invalid_map_notification_cmds:
            stoat_send_invalid_map_notifications(invalid_map_notification_cmds)
        if show_map_notifications:
            stoat_send_map_notification_list(show_map_notifications)

        if last_msg_id:
            update_last_msg_id(last_msg_id)
        time.sleep(5)
    
if __name__ == '__main__':
    t = threading.Thread(target=main_loop, daemon=True)
    t.start()

    from waitress import serve
    serve(app, host="127.0.0.1", port=5086, threads = 1)