anonkun/events.py

257 lines
5.7 KiB
Python

#!/usr/bin/env python3
"""
SocketIO events.
"""
import re
import time
import random
import bleach
from flask import session
from flask_socketio import SocketIO, emit, join_room
import tools
import database as db
socketio = SocketIO()
@socketio.on('joined')
def joined(data):
"""
Sent by clients when they enter a room.
"""
room = data["room"]
join_room(room)
@socketio.on('message')
def message(data):
"""
Sent by a client when the user entered a new message.
"""
room = int(data["room"])
message = data["message"]
name = data["name"]
user_id = data.get("user_id")
data = {}
date = int(time.time())
data["date"] = date
data["name"] = name
data["user_id"] = user_id
data["room"] = room
message = message.strip()
if not message:
return
tags = ["b", "code", "i", "s"]
message = bleach.clean(message, tags=tags)
lines = []
for line in message.splitlines():
if line.startswith(">"):
line = '<span class="greenText">' + line + '</span>'
lines.append(line)
message = "<br />".join(lines)
message = tools.handle_img(message)
data["message"] = message
roll_msg = ""
if message.startswith("/dice") or message.startswith("/roll"):
roll_msg = handle_dice(data)
if roll_msg:
data["message"] += '<hr class="msgSrvHr" /><b>' + roll_msg + "</b>"
message_id = db.log_chat_message(data)
emit("message", data, room=room)
if roll_msg:
dice_call_id = db.get_quest_meta(data["room"])[4]
if dice_call_id:
dice_call = db.get_dice_call(dice_call_id)
dice_roll = re.search(r"(\d+d\d+(?:[+-]\d+)?)", message).group(1)
roll_results = re.search(r"Rolled (.+)$", roll_msg).group(1)
db.insert_quest_roll(
message_id, room, dice_call_id, dice_roll, roll_results)
room = data["room"]
data = {}
data["post"] = roll_msg + " (" + dice_roll + ")"
data["post_type"] = "dice"
data["post_id"] = dice_call_id
emit("update_post", data, room=room)
def handle_dice(data):
"""
Handle /dice or /roll messages.
"""
reg = re.search(r"(\d+)d(\d+)([+-]\d+)?", data["message"])
if not reg:
return
try:
groups = [0 if d is None else int(d) for d in reg.groups()]
diceNum, diceSides, diceMod = groups
assert 0 < diceNum < 100
assert 0 < diceSides < 100
assert -1000 < diceMod < 1000
except (ValueError, AssertionError):
return
dice = [random.randint(1, diceSides) for _ in range(diceNum)]
total = sum(dice)
if diceMod:
total += diceMod
msg = f"Rolled {', '.join(map(str, dice))}"
if diceMod:
if diceMod > 0:
msg += " +" + str(diceMod)
else:
msg += " - " + str(diceMod)[1:]
msg += " = " + str(total)
return msg
@socketio.on("new_post")
def new_post(data, internal=False):
"""
Called when the QM makes a new post.
"""
room = data["room"]
res = db.get_quest_meta(quest_id=room)
if not res:
return
if session.get("user_id") != res[3]:
return
post = data["post"]
post = bleach.clean(post.strip())
post = post.replace("\n", "<br />")
post = tools.handle_img(post)
data = {}
data["post"] = [post]
data["post_type"] = "text"
data["date"] = int(time.time())
post_id = db.insert_quest_post(room, "text", post, data["date"])
data["post_id"] = post_id
emit("new_post", data, room=room)
@socketio.on("update_post")
def update_post(data):
"""
Called when the QM edits and saves a post.
"""
room = data["room"]
res = db.get_quest_meta(quest_id=room)
if not res:
return
if session.get("user_id") != res[3]:
return
# TODO: enforce only update text posts
post = data["post"]
post = post.strip().replace("<br>", "<br />")
post = tools.handle_img(post)
post_id = data["post_id"]
db.update_quest_post(post_id, post)
data = {}
data["post"] = post
data["post_id"] = post_id
data["post_type"] = "text"
emit("update_post", data, room=room)
@socketio.on("dice_post")
def dice_post(data):
"""
Called when the QM posts a new dice call.
"""
room = data["room"]
res = db.get_quest_meta(quest_id=room)
if not res:
return
if session.get("user_id") != res[3]:
return
data = {k: v for k, v in data.items() if v}
try:
diceNum = int(data.get("diceNum", 0))
diceSides = int(data.get("diceSides", 0))
diceMod = int(data.get("diceMod", 0))
diceChal = int(data.get("diceChal", 0))
diceRolls = int(data.get("diceRolls", 0))
assert 0 < diceNum < 100
assert 0 < diceSides < 100
assert -1000 < diceMod < 1000
assert 0 <= diceChal < 100
assert 0 <= diceRolls < 100
except (ValueError, AssertionError):
return
diceStrict = bool(data.get("diceStrict"))
dice_roll = f"{data['diceNum']}d{data['diceSides']}"
if diceMod:
if diceMod > 0:
dice_roll += "+"
dice_roll += str(diceMod)
post = "Roll " + dice_roll
if diceChal:
post += " vs DC" + str(diceChal)
data = {}
data["post"] = post
data["post_type"] = "dice"
data["date"] = int(time.time())
post_id = db.insert_quest_post(room, "dice", post, data["date"])
data["post_id"] = post_id
new_call = (dice_roll, diceStrict, diceChal, diceRolls)
db.set_dice_call_open(post_id, room, new_call)
emit("new_post", data, room=room)
@socketio.on('close_dice_call')
def close_dice_call(data):
"""
Closes an active dice call.
"""
room = data["room"]
res = db.get_quest_meta(quest_id=room)
if not res:
return
if session.get("user_id") != res[3]:
return
post_id = data.get("post_id")
db.set_dice_call_closed(room)
data = {}
data["post_id"] = post_id
emit("close_dice_call", data, room=room)
@socketio.on("open_dice_call")
def open_dice_call(data):
"""
Opens a closed dice call. This is only permitted if the dice call is
the last post in the quest.
"""
room = data["room"]
res = db.get_quest_meta(quest_id=room)
if not res:
return
if session.get("user_id") != res[3]:
return
post_id = data.get("post_id")
db.set_dice_call_open(post_id, room)
data = {}
data["post_id"] = post_id
emit("open_dice_call", data, room=room)