From a02956cbc75ab215332b8a50328fe3c724853217 Mon Sep 17 00:00:00 2001 From: John Burwell Date: Tue, 25 Apr 2023 19:46:33 -0500 Subject: [PATCH] first crack at separating concerns. broken --- rsbbs/__init__.py | 2 +- rsbbs/bbs.py | 296 -------------------------------------------- rsbbs/commands.py | 112 +++++++++++++++++ rsbbs/config.py | 80 ++++++++++++ rsbbs/console.py | 235 +++++++++++++++++++++++++++++++++++ rsbbs/controller.py | 170 +++++++++++++++++++++++++ rsbbs/message.py | 38 ------ rsbbs/parser.py | 16 +-- rsbbs/rsbbs.py | 26 +++- rsbbs/ui.py | 213 ------------------------------- 10 files changed, 628 insertions(+), 560 deletions(-) delete mode 100644 rsbbs/bbs.py create mode 100644 rsbbs/commands.py create mode 100644 rsbbs/config.py create mode 100644 rsbbs/console.py create mode 100644 rsbbs/controller.py delete mode 100644 rsbbs/message.py delete mode 100644 rsbbs/ui.py diff --git a/rsbbs/__init__.py b/rsbbs/__init__.py index 0ad462c..a7afe4c 100644 --- a/rsbbs/__init__.py +++ b/rsbbs/__init__.py @@ -17,4 +17,4 @@ # along with this program. If not, see . __all__ = ["rsbbs", "bbs", "message", "parser"] -__version__ = "0.1.0" +__version__ = "0.2.0" diff --git a/rsbbs/bbs.py b/rsbbs/bbs.py deleted file mode 100644 index 1b12768..0000000 --- a/rsbbs/bbs.py +++ /dev/null @@ -1,296 +0,0 @@ -#!/usr/bin/env python -# -# Really Simple BBS - a really simple BBS for ax.25 packet radio. -# Copyright (C) 2023 John Burwell -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -import logging.config -import os -import subprocess -import sys -import yaml -import pkg_resources -import platformdirs - -from sqlalchemy import create_engine, delete, select, or_ -from sqlalchemy.orm import Session - -from rsbbs import __version__ -from rsbbs.message import Message, Base -from rsbbs.parser import Parser -from rsbbs.ui import UI - - -# Main BBS class - -class BBS(): - - def __init__(self, sysv_args): - self._sysv_args = sysv_args - - self.config = self._load_config(sysv_args.config_file) - - self.calling_station = sysv_args.calling_station.upper() - - self.ui = UI(self) - self.engine = self._init_engine() - - logging.config.dictConfig(self.config['logging']) - - @property - def config(self): - return self._config - - @config.setter - def config(self, value): - self._config = value - - # - # Config file - # - - def _load_config(self, config_file): - """Load configuration file. - - If a config file is specified, attempt to use that. Otherwise, use a - file in the location appropriate to the host system. Create the file - if it does not exist, using the config_default.yaml as a default. - """ - # Use either the specified file or a file in a system config location - config_path = config_file or os.path.join( - platformdirs.user_config_dir(appname='rsbbs', ensure_exists=True), - 'config.yaml' - ) - if self._sysv_args.debug: - print(config_path) - # If the file doesn't exist there, create it - if not os.path.exists(config_path): - config_template_path = pkg_resources.resource_filename( - __name__, 'config_default.yaml') - try: - with open(config_template_path, 'r') as f: - config_template = yaml.load(f, Loader=yaml.FullLoader) - with open(config_path, 'w') as f: - yaml.dump(config_template, f) - except Exception as e: - print(f"Error creating configuration file: {e}") - exit(1) - # Load it - try: - with open(config_path, 'r') as f: - config = yaml.load(f, Loader=yaml.FullLoader) - except Exception as e: - print(f"Error loading configuration file: {e}") - exit(1) - - logging.info(f"Configuration file was successfully loaded." - f"File name: {config_path}") - - return config - - # - # Database - # - - def _init_engine(self): - """Create a connection to the sqlite3 database. - - The default location is the system-specific user-level data directory. - """ - db_path = os.path.join( - platformdirs.user_data_dir(appname='rsbbs', ensure_exists=True), - 'messages.db') - engine = create_engine( - 'sqlite:///' + db_path, - echo=self._sysv_args.debug) # Echo SQL output if -d is turned on - # Create the database schema if none exists - Base.metadata.create_all(engine) - return engine - - # - # BBS command functions - # - - def bye(self, args): - """Close the connection and exit.""" - self.ui._write_output("Bye!") - exit(0) - - def delete(self, args): - """Delete a message. - - Arguments: - number -- the message number to delete - """ - with Session(self.engine) as session: - try: - message = session.get(Message, args.number) - session.delete(message) - session.commit() - self.ui._write_output(f"Deleted message #{args.number}") - except Exception as e: - self.ui._write_output( - f"Unable to delete message #{args.number}") - - def delete_mine(self, args): - """Delete all messages addressed to the calling station's callsign.""" - self.ui._write_output("Delete all messages addressed to you? Y/N:") - response = sys.stdin.readline().strip() - if response.lower() != "y": - return - else: - with Session(self.engine) as session: - try: - statement = delete(Message).where( - Message.recipient == self.calling_station - ).returning(Message) - results = session.execute(statement) - count = len(results.all()) - if count > 0: - self.ui._write_output(f"Deleted {count} messages") - session.commit() - else: - self.ui._write_output(f"No messages to delete.") - except Exception as e: - self.ui._write_output(f"Unable to delete messages: {e}") - - def heard(self, args): - """Show a log of stations that have been heard by this station, - also known as the 'mheard' (linux) or 'jheard' (KPC, etc.) log. - """ - self.ui._write_output(f"Heard stations:") - result = subprocess.run(['mheard'], capture_output=True, text=True) - self.ui._write_output(result.stdout) - - def help(self, args): - """Show help.""" - self.ui.parser.print_help() - - def list(self, args): - """List all messages.""" - with Session(self.engine) as session: - statement = select(Message).where(or_( - (Message.is_private.is_not(False)), - (Message.recipient.__eq__(self.calling_station))) - ) - results = session.execute(statement) - self.ui.print_message_list(results) - - def list_mine(self, args): - """List only messages addressed to the calling station's callsign, - including public and private messages. - """ - with Session(self.engine) as session: - statement = select(Message).where( - Message.recipient == self.calling_station) - results = session.execute(statement) - self.ui.print_message_list(results) - - def read(self, args): - """Read a message. - - Arguments: - number -- the message number to read - """ - with Session(self.engine) as session: - statement = select(Message).where(Message.id == args.number) - result = session.execute(statement).one() - self.ui.print_message(result) - - def read_mine(self, args): - """Read all messages addressed to the calling station's callsign, - in sequence.""" - with Session(self.engine) as session: - statement = select(Message).where( - Message.recipient == self.calling_station) - result = session.execute(statement) - messages = result.all() - count = len(messages) - if count > 0: - self.ui._write_output(f"Reading {count} messages:") - for message in messages: - self.ui.print_message(message) - self.ui._write_output("Enter to continue...") - sys.stdin.readline() - else: - self.ui._write_output(f"No messages to read.") - - def send(self, args, is_private=False): - """Create a new message addressed to another callsign. - - Required arguments: - callsign -- the recipient's callsign - - Optional arguments: - subject -- message subject - message -- the message itself - """ - if not args.callsign: - args.callsign = self.ui._read_line("Callsign:") - if not args.subject: - args.subject = self.ui._read_line("Subject:") - if not args.message: - args.message = self.ui._read_multiline( - "Message - end with /ex on a single line:") - with Session(self.engine) as session: - session.add(Message( - sender=self.calling_station.upper(), - recipient=args.callsign.upper(), - subject=args.subject, - message=args.message, - is_private=is_private - )) - try: - session.commit() - self.ui._write_output("Message saved!") - except Exception as e: - session.rollback() - self.ui._write_output("Error saving message." - "Contact the sysop for assistance.") - - def send_private(self, args): - self.send(args, is_private=True) - """Send a message visible only to the recipient callsign. - - Required arguments: - callsign -- the recipient's callsign - - Optional arguments: - subject -- message subject - message -- the message itself - """ - - # - # Main loop - # - - def run(self): - - # Show greeting - self.ui.print_greeting() - - # Show initial prompt to the calling user - self.ui._write_output(self.config['command_prompt']) - - # Parse the BBS interactive commands for the rest of time - for line in sys.stdin: - try: - args = self.ui.parser.parse_args(line.split()) - args.func(args) - except Exception as e: - pass - - # Show our prompt to the calling user again - self.ui._write_output(self.config['command_prompt']) diff --git a/rsbbs/commands.py b/rsbbs/commands.py new file mode 100644 index 0000000..23f8209 --- /dev/null +++ b/rsbbs/commands.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python +# +# Really Simple BBS - a really simple BBS for ax.25 packet radio. +# Copyright (C) 2023 John Burwell +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + + +class Commands(): + + def __init__(self, responder): + self.responder = responder + + @property + def commands(self): + commands = [ + # (name, + # aliases, + # helpmsg, + # function, + # callback, + # {arg: + # {arg attributes}, + # ...}) + ('bye', + ['b', 'q'], + 'Sign off and disconnect', + self.responder.bye, + {}), + + ('delete', + ['d', 'k'], + 'Delete a message', + self.responder.delete, + {'number': + {'help': + 'The numeric index of the message to delete'}},), + + ('deletem', + ['dm', 'km'], + 'Delete all your messages', + self.responder.delete_mine, + {}), + + ('help', + ['h', '?'], + 'Show help', + self.responder.help, + {}), + + ('heard', + ['j'], + 'Show heard stations log', + self.responder.heard, + {}), + + ('list', + ['l'], + 'List all messages', + self.responder.list, + {}), + + ('listm', + ['lm'], + 'List only messages addressed to you', + self.responder.list_mine, + {}), + + ('read', + ['r'], + 'Read messages', + self.responder.read, + {'number': {'help': 'Message number to read'}}), + + ('readm', + ['rm'], + 'Read only messages addressed to you', + self.responder.read_mine, + {}), + + ('send', + ['s'], + 'Send a new message to a user', + self.responder.send, + { + 'callsign': {'help': 'Message recipient callsign'}, + '--subject': {'help': 'Message subject'}, + '--message': {'help': 'Message'}, + },), + + ('sendp', + ['sp'], + 'Send a private message to a user', + self.responder.send_private, + { + 'callsign': {'help': 'Message recipient callsign'}, + '--subject': {'help': 'Message subject'}, + '--message': {'help': 'Message'}, + },),] + + return commands diff --git a/rsbbs/config.py b/rsbbs/config.py new file mode 100644 index 0000000..598e4e6 --- /dev/null +++ b/rsbbs/config.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python +# +# Really Simple BBS - a really simple BBS for ax.25 packet radio. +# Copyright (C) 2023 John Burwell +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import os +import pkg_resources +import platformdirs +import yaml + + +class Config(): + + def __init__(self, app_name, args): + self._app_name = app_name + self._config_file_path = args.config_file + + self._load_config() + + self.config['db_path'] = os.path.join( + platformdirs.user_data_dir(appname=app_name, ensure_exists=True), + 'messages.db') + + self.config['debug'] = args.debug + self.config['args'] = args + + @property + def config_path(self): + # Use either the specified file or a file in a system config location + config_path = self._config_file_path or os.path.join( + platformdirs.user_config_dir( + appname=self._app_name, + ensure_exists=True), + 'config.yaml' + ) + return config_path + + def _init_config_file(self): + # If the file doesn't exist there, create it from the default file + # included in the package + if not os.path.exists(self.config_path): + config_default_file_path = pkg_resources.resource_filename( + __name__, + 'config_default.yaml') + try: + with open(config_default_file_path, 'r') as f: + config_template = yaml.load(f, Loader=yaml.FullLoader) + with open(self.config_path, 'w') as f: + yaml.dump(config_template, f) + except Exception as e: + print(f"Error creating configuration file: {e}") + exit(1) + + def _load_config(self): + """Load configuration file. + + If a config file is specified, attempt to use that. Otherwise, use a + file in the location appropriate to the host system. Create the file + if it does not exist, using the config_default.yaml as a default. + """ + # Load it + try: + with open(self.config_path, 'r') as f: + self.config = yaml.load(f, Loader=yaml.FullLoader) + except Exception as e: + print(f"Error loading configuration file: {e}") + exit(1) diff --git a/rsbbs/console.py b/rsbbs/console.py new file mode 100644 index 0000000..8d764f0 --- /dev/null +++ b/rsbbs/console.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python +# +# Really Simple BBS - a really simple BBS for ax.25 packet radio. +# Copyright (C) 2023 John Burwell +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import sys + +import rsbbs +from rsbbs.commands import Commands +from rsbbs.config import Config +from rsbbs.controller import Base, Controller, Message +from rsbbs.parser import Parser + + +# Main UI console class + +class Console(): + + def __init__(self, config: Config, controller: Controller): + self.config = config + self.controller = controller + + self.commands = Commands(self) + self.parser = Parser(self.commands) + + # + # Input and output + # + + def _write_output(self, output): + """Write something to stdout.""" + sys.stdout.write(output + '\r\n') + + def print_greeting(self): + # Show greeting + greeting = [] + greeting.append(f"[RSBBS-{rsbbs.__version__}] listening on " + f"{self.config.config['callsign']} ") + + greeting.append(f"Welcome to {self.config.config['bbs_name']}, " + f"{self.config.config['args'].calling_station}") + + greeting.append(self.config.config['banner_message']) + + greeting.append("For help, enter 'h'") + + self._write_output('\r\n'.join(greeting)) + + def print_message(self, message): + """Print an individual message.""" + # Format the big ol' date and time string + datetime = message.Message.datetime.strftime( + '%A, %B %-d, %Y at %-H:%M %p UTC') + # Print the message + self._write_output(f"") + self._write_output(f"Message: {message.Message.id}") + self._write_output(f"Date: {datetime}") + self._write_output(f"From: {message.Message.sender}") + self._write_output(f"To: {message.Message.recipient}") + self._write_output(f"Subject: {message.Message.subject}") + self._write_output(f"") + self._write_output(f"{message.Message.message}") + + def print_message_list(self, messages): + """Print a list of messages.""" + # Print the column headers + self._write_output(f"{'MSG#': <{5}} " + f"{'TO': <{9}} " + f"{'FROM': <{9}} " + f"{'DATE': <{11}} " + f"SUBJECT") + # Print the messages + for message in messages: + datetime_ = message.Message.datetime.strftime('%Y-%m-%d') + self._write_output(f"{message.Message.id: <{5}} " + f"{message.Message.recipient: <{9}} " + f"{message.Message.sender: <{9}} " + f"{datetime_: <{11}} " + f"{message.Message.subject}") + + # + # Command functions + # + + def bye(self, args): + self._write_output("Bye!") + exit(0) + + def delete(self, args): + self.controller.delete(args) + + def delete_mine(self, args): + """Delete all messages addressed to the calling station's callsign.""" + self._write_output("Delete all messages addressed to you? Y/N:") + response = sys.stdin.readline().strip() + if response.lower() != "y": + return + else: + try: + result = self.controller.delete_mine(args) + if result['count'] > 0: + self._write_output(f"Deleted {result['count']} messages") + else: + self._write_output(f"No messages to delete.") + except Exception as e: + self._write_output(f"Unable to delete messages: {e}") + + def heard(self, args): + """Show a log of stations that have been heard by this station, + also known as the 'mheard' (linux) or 'jheard' (KPC, etc.) log. + """ + self._write_output(f"Heard stations:") + result = self.controller.heard(args) + self._write_output(result.stdout) + + def help(self, args): + self.parser.parser.print_help() + + def list(self, args): + """List all messages.""" + result = self.controller.list(args) + self.print_message_list(result['result']) + + def list_mine(self, args): + """List only messages addressed to the calling station's callsign, + including public and private messages. + """ + result = self.controller.list_mine(args) + self.print_message_list(result) + + def read(self, args): + """Read a message. + + Arguments: + number -- the message number to read + """ + result = self.controller.read(args) + self.print_message(result) + + def read_mine(self, args): + """Read all messages addressed to the calling station's callsign, + in sequence.""" + result = self.controller.list_mine(args) + if result['count'] > 0: + self._write_output(f"Reading {result['count']} messages:") + for message in result['result'].all(): + self.print_message(message) + self._write_output("Enter to continue...") + sys.stdin.readline() + else: + self._write_output(f"No messages to read.") + + def send(self, args, is_private=False): + """Create a new message addressed to another callsign. + + Required arguments: + callsign -- the recipient's callsign + + Optional arguments: + subject -- message subject + message -- the message itself + """ + if not args.callsign: + args.callsign = self._read_line("Callsign:") + if not args.subject: + args.subject = self._read_line("Subject:") + if not args.message: + args.message = self._read_multiline( + "Message - end with /ex on a single line:") + # with Session(self.engine) as session: + # session.add(Message( + # sender=self.calling_station.upper(), + # recipient=args.callsign.upper(), + # subject=args.subject, + # message=args.message, + # is_private=is_private + # )) + # try: + # session.commit() + # self._write_output("Message saved!") + # except Exception as e: + # session.rollback() + # self._write_output("Error saving message." + # "Contact the sysop for assistance.") + + def send_private(self, args): + self.send(args, is_private=True) + """Send a message visible only to the recipient callsign. + + Required arguments: + callsign -- the recipient's callsign + + Optional arguments: + subject -- message subject + message -- the message itself + """ + + # + # Main run method + # + + def run(self): + + # Show greeting + self.print_greeting() + + # Show initial prompt to the calling user + self._write_output(self.config.config['command_prompt']) + + # Parse the BBS interactive commands for the rest of time + for line in sys.stdin: + try: + args = self.parser.parser.parse_args(line.split()) + args.func(args) + except Exception: + if self.config.config['debug']: + raise + else: + pass + + # Show our prompt to the calling user again + self._write_output(self.config.config['command_prompt']) diff --git a/rsbbs/controller.py b/rsbbs/controller.py new file mode 100644 index 0000000..d1ba9f1 --- /dev/null +++ b/rsbbs/controller.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python +# +# Really Simple BBS - a really simple BBS for ax.25 packet radio. +# Copyright (C) 2023 John Burwell +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import subprocess + +from datetime import datetime, timezone + +from sqlalchemy import Boolean, DateTime, String +from sqlalchemy import create_engine, delete, select, or_ + +from sqlalchemy.orm import DeclarativeBase, Mapped, Session +from sqlalchemy.orm import mapped_column + +from rsbbs.config import Config + + +class Base(DeclarativeBase): + pass + + +class Message(Base): + __tablename__ = 'message' + id: Mapped[int] = mapped_column(primary_key=True) + sender: Mapped[str] = mapped_column(String) + recipient: Mapped[str] = mapped_column(String) + subject: Mapped[str] = mapped_column(String) + message: Mapped[str] = mapped_column(String) + datetime: Mapped[DateTime] = mapped_column( + DateTime, default=datetime.now(timezone.utc)) + is_private: Mapped[bool] = mapped_column(Boolean) + + +class Controller(): + + def __init__(self, config: Config): + self.config = config + self._init_datastore() + + def _init_datastore(self): + """Create a connection to the sqlite3 database. + + The default location is the system-specific user-level data directory. + """ + db_path = self.config.config['db_path'] + self.engine = create_engine( + 'sqlite:///' + db_path, + echo=self.config.config['debug']) + + # Create the database schema if none exists + Base.metadata.create_all(self.engine) + + def delete(self, args): + """Delete a message. + + Arguments: + number -- the message number to delete + """ + with Session(self.engine) as session: + try: + message = session.get(Message, args.number) + session.delete(message) + session.commit() + return {} + except Exception: + raise + + def delete_mine(self, args): + """Delete all messages addressed to the calling station's callsign.""" + with Session(self.engine) as session: + try: + statement = delete(Message).where( + Message.recipient == self.calling_station + ).returning(Message) + result = session.execute(statement) + count = len(result.all()) + session.commit() + return {'count': count, 'result': result} + except Exception: + raise + + def heard(self, args): + """Show a log of stations that have been heard by this station, + also known as the 'mheard' (linux) or 'jheard' (KPC, etc.) log. + """ + try: + result = subprocess.run(['mheard'], capture_output=True, text=True) + return {'result': result} + except Exception: + raise + + def list(self, args): + """List all messages.""" + with Session(self.engine) as session: + try: + statement = select(Message).where(or_( + (Message.is_private.is_not(False)), + (Message.recipient.__eq__( + self.config.config['args'].calling_station))) + ) + result = session.execute(statement) + except Exception: + raise + return {'result': result} + + def list_mine(self, args): + """List only messages addressed to the calling station's callsign, + including public and private messages. + """ + with Session(self.engine) as session: + try: + statement = select(Message).where( + Message.recipient == self.calling_station) + result = session.execute(statement) + return {'result': result} + except Exception: + raise + + def read(self, args): + """Read a message. + + Arguments: + number -- the message number to read + """ + with Session(self.engine) as session: + try: + statement = select(Message).where(Message.id == args.number) + result = session.execute(statement).one() + return {'result': result} + except Exception: + raise + + def send(self, args, is_private=False): + """Create a new message addressed to another callsign. + + Required arguments: + callsign -- the recipient's callsign + + Optional arguments: + subject -- message subject + message -- the message itself + """ + with Session(self.engine) as session: + try: + session.add(Message( + sender=self.calling_station.upper(), + recipient=args.callsign.upper(), + subject=args.subject, + message=args.message, + is_private=is_private + )) + session.commit() + return {} + except Exception: + session.rollback() + raise diff --git a/rsbbs/message.py b/rsbbs/message.py deleted file mode 100644 index 43281c5..0000000 --- a/rsbbs/message.py +++ /dev/null @@ -1,38 +0,0 @@ -#!/usr/bin/env python -# -# Really Simple BBS - a really simple BBS for ax.25 packet radio. -# Copyright (C) 2023 John Burwell -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -from datetime import datetime, timezone - -from sqlalchemy import Boolean, DateTime, String -from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column - - -class Base(DeclarativeBase): - pass - - -class Message(Base): - __tablename__ = 'message' - id: Mapped[int] = mapped_column(primary_key=True) - sender: Mapped[str] = mapped_column(String) - recipient: Mapped[str] = mapped_column(String) - subject: Mapped[str] = mapped_column(String) - message: Mapped[str] = mapped_column(String) - datetime: Mapped[DateTime] = mapped_column( - DateTime, default=datetime.now(timezone.utc)) - is_private: Mapped[bool] = mapped_column(Boolean) diff --git a/rsbbs/parser.py b/rsbbs/parser.py index 1bdc78e..c799e84 100644 --- a/rsbbs/parser.py +++ b/rsbbs/parser.py @@ -18,6 +18,8 @@ import argparse +from rsbbs.commands import Commands + # We want to override the error and exit methods of ArgumentParser # to prevent it exiting unexpectedly or spewing error data over the air @@ -35,12 +37,12 @@ class BBSArgumentParser(argparse.ArgumentParser): class Parser(BBSArgumentParser): - def __init__(self, commands): - self.parser = self.init_parser(commands) + def __init__(self, commands: Commands): + self.init_parser(commands) def init_parser(self, commands): # Root parser for BBS commands - parser = BBSArgumentParser( + self.parser = BBSArgumentParser( description='BBS Main Menu', prog='', add_help=False, @@ -48,10 +50,12 @@ class Parser(BBSArgumentParser): ) # We will create a subparser for each individual command - subparsers = parser.add_subparsers(title='Commands', dest='command') + subparsers = self.parser.add_subparsers( + title='Commands', + dest='command') # Loop through the commands and add each as a subparser - for name, aliases, help_msg, func, arguments in commands: + for name, aliases, help_msg, func, arguments in commands.commands: subparser = subparsers.add_parser( name, aliases=aliases, @@ -60,5 +64,3 @@ class Parser(BBSArgumentParser): for arg_name, options in arguments.items(): subparser.add_argument(arg_name, **options) subparser.set_defaults(func=func) - - return parser diff --git a/rsbbs/rsbbs.py b/rsbbs/rsbbs.py index 09ad097..7fab314 100755 --- a/rsbbs/rsbbs.py +++ b/rsbbs/rsbbs.py @@ -20,7 +20,11 @@ import argparse import sys from rsbbs import __version__ -from rsbbs.bbs import BBS +from rsbbs.commands import Commands +from rsbbs.config import Config +from rsbbs.console import Console +from rsbbs.controller import Controller +from rsbbs.parser import Parser def main(): @@ -54,11 +58,23 @@ def main(): # Parse the args from the system sysv_args = sysv_parser.parse_args(sys.argv[1:]) - # Instantiate the BBS object - bbs = BBS(sysv_args) + # Load configuration + config = Config( + app_name='rsbbs', + args=sysv_args) - # Start the main BBS loop - bbs.run() + # Init the contoller + controller = Controller(config) + + # # Set up commands and parser + # commands = Commands(controller) + # parser = Parser(commands) + + # Init the UI console + console = Console(config, controller) + + # Start the app + console.run() if __name__ == "__main__": diff --git a/rsbbs/ui.py b/rsbbs/ui.py deleted file mode 100644 index 03cf257..0000000 --- a/rsbbs/ui.py +++ /dev/null @@ -1,213 +0,0 @@ -#!/usr/bin/env python -# -# Really Simple BBS - a really simple BBS for ax.25 packet radio. -# Copyright (C) 2023 John Burwell -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -import sys - -from rsbbs import __version__ -from rsbbs.parser import Parser - - -# UI - -class UI(): - - def __init__(self, bbs): - # Get the BBS - self.bbs = bbs - self.parser = self._init_parser() - - # - # BBS command parser - # - - def _init_parser(self): - """Define BBS command names, aliases, help messages, action (function), - and any arguments. - - This configures argparser's internal help and maps user commands to BBS - functions. - """ - commands = [ - # (name, - # aliases, - # helpmsg, - # function, - # {arg: - # {arg attributes}, - # ...}) - ('bye', - ['b', 'q'], - 'Sign off and disconnect', - self.bbs.bye, - {}), - - ('delete', - ['d', 'k'], - 'Delete a message', - self.bbs.delete, - {'number': - {'help': - 'The numeric index of the message to delete'}},), - - ('deletem', - ['dm', 'km'], - 'Delete all your messages', - self.bbs.delete_mine, - {}), - - ('help', - ['h', '?'], - 'Show help', - self.bbs.help, - {}), - - ('heard', - ['j'], - 'Show heard stations log', - self.bbs.heard, - {}), - - ('list', - ['l'], - 'List all messages', - self.bbs.list, - {}), - - ('listm', - ['lm'], - 'List only messages addressed to you', - self.bbs.list_mine, - {}), - - ('read', - ['r'], - 'Read messages', - self.bbs.read, - {'number': {'help': 'Message number to read'}}), - - ('readm', - ['rm'], - 'Read only messages addressed to you', - self.bbs.read_mine, - {}), - - ('send', - ['s'], - 'Send a new message to a user', - self.bbs.send, - { - 'callsign': {'help': 'Message recipient callsign'}, - '--subject': {'help': 'Message subject'}, - '--message': {'help': 'Message'}, - },), - - ('sendp', - ['sp'], - 'Send a private message to a user', - self.bbs.send_private, - { - 'callsign': {'help': 'Message recipient callsign'}, - '--subject': {'help': 'Message subject'}, - '--message': {'help': 'Message'}, - },),] - - # Send all the commands defined above to the parser - return Parser(commands).parser - - # - # Input and output - # - - def _read_line(self, prompt): - """Read a single line of input, with an optional prompt, - until we get something. - """ - output = None - while not output: - if prompt: - self._write_output(prompt) - input = sys.stdin.readline().strip() - if input != "": - output = input - return output - - def _read_multiline(self, prompt): - """Read multiple lines of input, with an optional prompt, - until the user enters '/ex' by itself on a line. - """ - output = [] - if prompt: - self._write_output(prompt) - while True: - line = sys.stdin.readline() - if line.lower().strip() == "/ex": - break - else: - output.append(line) - return ''.join(output) - - def _write_output(self, output): - """Write something to stdout.""" - sys.stdout.write(output + '\r\n') - - def print_message_list(self, messages): - """Print a list of messages.""" - # Print the column headers - self._write_output(f"{'MSG#': <{5}} " - f"{'TO': <{9}} " - f"{'FROM': <{9}} " - f"{'DATE': <{11}} " - f"SUBJECT") - # Print the messages - for message in messages: - datetime_ = message.Message.datetime.strftime('%Y-%m-%d') - self._write_output(f"{message.Message.id: <{5}} " - f"{message.Message.recipient: <{9}} " - f"{message.Message.sender: <{9}} " - f"{datetime_: <{11}} " - f"{message.Message.subject}") - - def print_message(self, message): - """Print an individual message.""" - # Format the big ol' date and time string - datetime = message.Message.datetime.strftime( - '%A, %B %-d, %Y at %-H:%M %p UTC') - # Print the message - self._write_output(f"") - self._write_output(f"Message: {message.Message.id}") - self._write_output(f"Date: {datetime}") - self._write_output(f"From: {message.Message.sender}") - self._write_output(f"To: {message.Message.recipient}") - self._write_output(f"Subject: {message.Message.subject}") - self._write_output(f"") - self._write_output(f"{message.Message.message}") - - def print_greeting(self): - # Show greeting - greeting = [] - greeting.append(f"[RSBBS-{__version__}] listening on " - f"{self.bbs.config['callsign']} ") - - greeting.append(f"Welcome to {self.bbs.config['bbs_name']}, " - f"{self.bbs.calling_station}") - - greeting.append(self.bbs.config['banner_message']) - - greeting.append("For help, enter 'h'") - - self._write_output('\r\n'.join(greeting))