mudhorn/TC2-BBS-mesh
1
0
Fork 1
mirror of https://github.com/TheCommsChannel/TC2-BBS-mesh.git synced 2025-03-05 20:51:53 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #7 from Blergo/Yorkshire-BBS

Browse source 
latest updates
This commit is contained in:
Blergo 2024-07-02 19:29:18 +01:00 committed by GitHub
parent a8f2e85280 fc2a8870a2
commit 9b599c8846
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 7 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
command_handlers.py
View file

@ -3,7 +3,6 @@ import random
import time
import psutil
from config_init import initialize_config
from db_operations import (
add_bulletin, add_mail, delete_mail,
get_bulletin_content, get_bulletins,
@ -16,31 +15,26 @@ from utils import (
update_user_state
)
def get_node_name(node_id, interface):
node_info = interface.nodes.get(node_id)
if node_info:
return node_info['user']['longName']
return f"Node {node_id}"
def handle_mail_command(sender_id, interface):
response = "✉️ MAIL MENU ✉️\n\nWhat would you like to do with mail?\n\n[0]Read\n[1]Send\n[2]Exit"
send_message(response, sender_id, interface)
update_user_state(sender_id, {'command': 'MAIL', 'step': 1})
def handle_bulletin_command(sender_id, interface):
response = "📰 BULLETIN MENU 📰\n\nWhich board would you like to enter?\n\n[0]General\n[1]Info\n[2]News\n[3]Urgent\n[4]Exit"
send_message(response, sender_id, interface)
update_user_state(sender_id, {'command': 'BULLETIN', 'step': 1})
def handle_exit_command(sender_id, interface):
send_message("Type 'HELP' for a list of commands.", sender_id, interface)
update_user_state(sender_id, None)
def handle_help_command(sender_id, interface, state=None):
title = "█▓▒░ Yorkshire BBS ░▒▓█\n\n"
commands = [
@ -78,13 +72,11 @@ def handle_help_command(sender_id, interface, state=None):
response = title + "Available commands:\n" + "\n".join(commands)
send_message(response, sender_id, interface)
def handle_stats_command(sender_id, interface):
response = "What stats would you like to view?\n\n[0]Mesh Stats\n[1]Server Stats\n[2]Exit Stats Menu"
send_message(response, sender_id, interface)
update_user_state(sender_id, {'command': 'STATS', 'step': 1})
def handle_fortune_command(sender_id, interface):
try:
with open('fortunes.txt', 'r') as file:
@ -92,13 +84,13 @@ def handle_fortune_command(sender_id, interface):
if not fortunes:
send_message("No fortunes available.", sender_id, interface)
return
# trunk-ignore(bandit/B311)
fortune = random.choice(fortunes).strip()
decorated_fortune = f"🔮 {fortune} 🔮"
send_message(decorated_fortune, sender_id, interface)
except Exception as e:
send_message(f"Error generating fortune: {e}", sender_id, interface)
def handle_stats_steps(sender_id, message, step, interface, bbs_nodes):
if step == 1:
choice = int(message)
@ -113,7 +105,8 @@ def handle_stats_steps(sender_id, message, step, interface, bbs_nodes):
psutil.cpu_percent()
time.sleep(0.1)
cpu = str(psutil.cpu_percent()/psutil.cpu_count())
response = "CPU: " + cpu + "%"
ramu = str(psutil.virtual_memory().percent)
response = "CPU: " + cpu + "% Used\nRAM: " + ramu + "% Used"
send_message(response, sender_id, interface)
handle_stats_command(sender_id, interface)
return
@ -163,7 +156,6 @@ def handle_stats_steps(sender_id, message, step, interface, bbs_nodes):
send_message(f"Total nodes seen in the last {timeframes[choice - 1]}: {total_nodes}", sender_id, interface)
handle_stats_steps(sender_id, '0', 1, interface, bbs_nodes)
def handle_bb_steps(sender_id, message, step, state, interface, bbs_nodes):
boards = {0: "General", 1: "Info", 2: "News", 3: "Urgent"}
@ -243,7 +235,7 @@ def handle_bb_steps(sender_id, message, step, state, interface, bbs_nodes):
update_user_state(sender_id, None)
return
sender_short_name = node_info['user'].get('shortName', f"Node {sender_id}")
unique_id = add_bulletin(board, sender_short_name, subject, content, bbs_nodes, interface)
add_bulletin(board, sender_short_name, subject, content, bbs_nodes, interface)
send_message(f"Your bulletin '{subject}' has been posted to {board}.\n(╯°□°)╯📄📌[{board}]", sender_id, interface)
response = f"What would you like to do in the {board} board?\n\n[0]View Bulletins\n[1]Post Bulletin\n[2]Exit"
send_message(response, sender_id, interface)
@ -252,7 +244,6 @@ def handle_bb_steps(sender_id, message, step, state, interface, bbs_nodes):
state['content'] += message + "\n"
update_user_state(sender_id, state)
def handle_mail_steps(sender_id, message, step, state, interface, bbs_nodes):
if step == 1:
choice = message
@ -275,8 +266,7 @@ def handle_mail_steps(sender_id, message, step, state, interface, bbs_nodes):
elif step == 2:
mail_id = int(message)
try:
try:
# ERROR: sender_id is not what is stored in the DB
sender_node_id = get_node_id_from_num(sender_id, interface)
sender, date, subject, content, unique_id = get_mail_content(mail_id, sender_node_id)
@ -286,7 +276,7 @@ def handle_mail_steps(sender_id, message, step, state, interface, bbs_nodes):
except TypeError:
# get_main_content returned None. Node tried to access somebody's else mail message
logging.info(f"Node {sender_id} tried to access non-existent message")
send_message(f"Mail not found", sender_id, interface)
send_message("Mail not found", sender_id, interface)
update_user_state(sender_id, None)
elif step == 3:
@ -356,7 +346,6 @@ def handle_mail_steps(sender_id, message, step, state, interface, bbs_nodes):
send_message("Okay, feel free to send another command.", sender_id, interface)
update_user_state(sender_id, None)
def handle_wall_of_shame_command(sender_id, interface):
response = "Devices with battery levels below 20%:\n"
for node_id, node in interface.nodes.items():
@ -369,13 +358,11 @@ def handle_wall_of_shame_command(sender_id, interface):
response = "No devices with battery levels below 20% found."
send_message(response, sender_id, interface)
def handle_channel_directory_command(sender_id, interface):
response = "📚 CHANNEL DIRECTORY 📚\n\nWhat would you like to do in the Channel Directory?\n\n[0]View\n[1]Post\n[2]Exit"
send_message(response, sender_id, interface)
update_user_state(sender_id, {'command': 'CHANNEL_DIRECTORY', 'step': 1})
def handle_channel_directory_steps(sender_id, message, step, state, interface):
if step == 1:
choice = message
@ -414,4 +401,4 @@ def handle_channel_directory_steps(sender_id, message, step, state, interface):
channel_name = state['channel_name']
add_channel(channel_name, channel_url)
send_message(f"Your channel '{channel_name}' has been added to the directory.", sender_id, interface)
handle_channel_directory_command(sender_id, interface)
handle_channel_directory_command(sender_id, interface)