initial commit of super simple map notifications for stoat

stoat_map_notifications/python/README.md

@@ -0,0 +1,3 @@
+requests
+mysql-connector-python
+python-a2s

stoat_map_notifications/python/main.py

@@ -0,0 +1,236 @@
+#!/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
+import requests
+import re
+import a2s
+import traceback
+import json
+
+#runs every 3 minutes to inform about a new map
+
+def insert_maps(fixed):
+    with get_connection_unloze_playtime() as conn:
+        with conn.cursor() as cur:
+            placeholders = ','.join(['(%s)'] * len(fixed))
+            flattened_values = [row for row in fixed]
+            sql_statement = f"""
+                INSERT IGNORE INTO unloze_playtimestats.map_notifications 
+                    (mapname) 
+                    VALUES {placeholders}
+            """
+            cur.execute(sql_statement, flattened_values)
+            conn.commit()
+
+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_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\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, last_msg_id):
+    headers = {
+        "x-bot-token": stoat_token 
+    }
+
+    valid_map_notification_cmds_add= []
+    valid_map_notification_cmds_remove= []
+    invalid_map_notification_cmds = []
+    #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=50&include_users=true&sort=Oldest&after=', '/messages?limit=50&include_users=true')
+        response = requests.get(url, headers=headers)
+    else:
+        response = requests.get(stoat_url_50_last_messages + last_msg_id, headers=headers)
+
+    data = response.json()
+    for index, msg in enumerate(data['messages']):
+        #print(msg)
+        msg_id = msg['_id']
+        author = msg['author']
+        content = msg['content']
+        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:
+            if content.startswith("add "):
+                content = content.split("add ")[1]
+                if content in fixed:
+                    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:
+                    valid_map_notification_cmds_remove.append([content, author])
+                else:
+                    invalid_map_notification_cmds.append([content, author])
+
+            else:
+                invalid_map_notification_cmds.append([content, author])
+        except:
+            err = traceback.format_exc()
+            print("err: ", err)
+
+    #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
+
+def send_post_notify_message_to_stoat(list_of_people_to_notify, infomap):
+    #print('list_of_people_to_notify: ', list_of_people_to_notify)
+    message = f"Map changed to: {infomap}\n\n"
+    for people in list_of_people_to_notify:
+        if not people: continue
+        people = people.replace('["', '').replace('"]', '')
+        message += f"<@{people}> "
+    if not message.endswith("\n"):
+        message += "\n"
+    message += "Check <#01KGH9S0R3A1Y7BDVWWD8NMNGS> for information or https://unloze.com/pages/servers\n(updated every 5 minutes)"
+    headers = {
+        "x-bot-token": f"{stoat_token}",
+        "Content-Type": "application/json"
+    }
+    data = {
+        "content": message
+    }
+    #print('messages: ', 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 update_file_current_map(infomap):
+    with open("currentmap_file.txt", 'w') as f:
+        f.write(infomap)
+
+def get_last_msg_id_read():
+    with open("last_msg_id.txt", 'r') as f:
+        return f.readline()
+
+def get_last_current_map():
+    with open("currentmap_file.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
+
+def main():
+    #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)
+
+    #insert maps if they are not already.
+    insert_maps(fixed)
+
+    #check last 50 messages for new map notification updates
+    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  = 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)
+
+    update_last_msg_id(last_msg_id)
+    
+    address = ("51.195.188.106", int("27015"))
+    info = a2s.info(address)
+    infomap = info.map_name
+    file_current_map = get_last_current_map()
+    print('file_current_map: ', file_current_map, ' infomap: ', infomap)
+
+    if infomap != file_current_map:
+        list_of_people_to_notify = notify_of_new_map(infomap)
+        send_post_notify_message_to_stoat(list_of_people_to_notify, infomap)
+
+        update_file_current_map(infomap)
+    
+if __name__ == '__main__':
+    main()

stoat_map_notifications/python/settings.py

@@ -0,0 +1,13 @@
+import mysql.connector
+
+stoat_token = ""
+
+stoat_url_map_notifications = ""
+stoat_url_50_last_messages = ""
+
+def get_connection_unloze_playtime():
+    return mysql.connector.connect(
+        host="127.0.0.1",
+        port=3306,
+        user="",
+        password="")

stoat_map_notifications/sql/create_table.sql

@@ -0,0 +1,5 @@
+CREATE TABLE  unloze_playtimestats.map_notifications (
+  `mapname` varchar(255) NOT NULL,
+  `users_to_notify` JSON,
+  PRIMARY KEY (`mapname`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

stoat_map_notifications/systemctl/stoat_map_notifications.service

@@ -0,0 +1,10 @@
+[Unit]
+Description=updates map notifications ever 5 minutes
+
+[Service]
+Type=simple
+User=nonroot
+Environment=PYTHONUNBUFFERED=1
+Environment=PATH=/home/nonroot/stoat_map_notifications/venv/bin:/usr/local/bin:/usr/bin:/bin:/usr/local/games:/usr/games:/snap/bin
+WorkingDirectory=/home/nonroot/stoat_map_notifications
+ExecStart=/home/nonroot/stoat_map_notifications/main.py

stoat_map_notifications/systemctl/stoat_map_notifications.timer

@@ -0,0 +1,8 @@
+[Unit]
+Description=checks every 3 minutes on stoat about map notifcations
+
+[Timer]
+OnCalendar=*:0/3
+
+[Install]
+WantedBy=multi-user.target