A BBS for the ax26 protocol.

bbs.py 4.2KB

    import kiss2 import ax26 import sqlite3 from tabulate import tabulate import datetime, time k = kiss2.TCPKISS(host='localhost', port=8001) k.start() ax = ax26.Server(k, "VA1QLE-BBS") def get_board(con, page = 0): cur = con.cursor() cur.execute('SELECT id, author, subject, timestamp FROM board ORDER BY timestamp LIMIT ?, ?', [page * 10, page * 10 + 10]) rows = list(map(list, cur.fetchall())) if len(rows) == 0: return "There are no bulletins on page " + str(page) + "." for row in rows: row[3] = datetime.datetime.fromtimestamp(row[3]).strftime('%Y-%m-%d %H:%M:%S') return tabulate(rows, ["ID", "Author", "Subject", "Date"]) + '\r\nRead with G <id>' def get_bulletin(con, id): cur = con.cursor() cur.execute('SELECT author, subject, body, timestamp FROM board WHERE id = ?',[id]) row = cur.fetchone() if row is None: return "Bulletin not found." row = list(row) row[3] = datetime.datetime.fromtimestamp(row[3]).strftime('%Y-%m-%d %H:%M:%S') body = row[2] row.pop(2) return tabulate([row], ["Author", "Subject", "Date"]) + '\r\n' + body def commit_bulletin(con, author, subject, body): cur = con.cursor() cur.execute('INSERT INTO board(author, subject, body, timestamp) VALUES (?, ?, ?, ?)', [author, subject, body, time.time()]) con.commit() chat_list = [] class handler: def send_msg(self, msg): ax.send_data(self.client, msg.encode()) def send_chat(self, msg): for cli in chat_list: if cli != self.client: ax.send_data(cli, self.client + b": " + msg.encode()) def __init__(self, client_callsign): print("New connection from " + client_callsign.decode()) self.client = client_callsign self.in_chat = False self.writing_bulletin = False self.bulletin_msg = '' self.bulletin_subject = '' self.con = sqlite3.connect('main.db') self.send_msg("AX26 BBS. Written by VA1QLE.\r\n" "Welcome to the VA1QLE-BBS. For available commands, type 'help'.") def bbs_handler(self, text, cmd, args): if cmd == "help": self.send_msg("Available commands: \r\n" "chat - Join chat room\r\n" "L - List latest board messages\r\n" "G <id> - Read board message by id\r\n" "W <subject> - Write board message with specified subject") elif cmd == "chat": self.in_chat = True chat_list.append(self.client) self.send_msg("You're in chat. Type /ex to exit.\r\n" + str(len(chat_list)) + " people currently online: " + ', '.join([chat.decode() for chat in chat_list])) self.send_chat(">" + self.client.decode() + " connected to chat.") elif cmd == "l": page = 0 try: page = int(args.split(' ', 1)[0]) except ValueError: pass self.send_msg(get_board(self.con, page)) elif cmd == 'g': try: id = int(args.split(' ', 1)[0]) self.send_msg(get_bulletin(self.con, id)) except ValueError: self.send_msg("ERROR: Please specify an ID.") elif cmd == "w": if args.lower() == "subject" or args == "": self.send_msg("Please specify a subject.") else: self.send_msg("Writing to bulletin board. Type /ex on a new line when you're done.") # Put the user into write mode self.writing_bulletin = True self.bulletin_subject = args else: self.send_msg("Error: Unknown command.") def chat_handler(self, text, cmd, args): # if cmd == "/help": if cmd == "/ex": self.send_msg("Leaving the chat.") self.send_chat(">" + self.client.decode() + " disconnected from chat.") self.in_chat = False chat_list.remove(self.client) elif not text.isspace(): # Send text to all self.send_chat(text) def write_bulletin(self, cmd, text): if cmd != '/ex': self.bulletin_msg += text + "\r\n" else: self.bulletin_msg = self.bulletin_msg[:-2] # Remove last newline commit_bulletin(self.con, self.client.decode().split('-', 1)[0], self.bulletin_subject, self.bulletin_msg) self.writing_bulletin = False self.send_msg("Saved. Welcome back to the BBS.") def on_recv(self, data): text = data.decode().strip() cmd, args = text.lower(), "" if " " in text: cmd, args = text.split(" ", 1) cmd = cmd.lower() if self.in_chat: self.chat_handler(text, cmd, args) elif self.writing_bulletin: self.write_bulletin(cmd, text) else: self.bbs_handler(text, cmd, args) ax.serve_forever(handler)