278 lines
10 KiB
Python
278 lines
10 KiB
Python
#!/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 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, 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=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])
|
|
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('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, infomap):
|
|
list_of_people_to_notify = list_of_people_to_notify[0]
|
|
message = f"Map changed to: {infomap}\n\n"
|
|
try:
|
|
list_of_people_to_notify = json.loads(list_of_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 += "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, 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)
|
|
|
|
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()
|