From 800d7de809e405522144fbfebdb2f1eaf27c6dae Mon Sep 17 00:00:00 2001 From: John Burwell Date: Mon, 24 Apr 2023 14:23:49 -0500 Subject: [PATCH] refactor to make linter happy --- rsbbs/__init__.py | 3 +- rsbbs/bbs.py | 381 +++++++++++++++++++++++++++++------------- rsbbs/message.py | 7 +- rsbbs/parser.py | 4 +- rsbbs/project_root.py | 28 ---- rsbbs/rsbbs.py | 18 +- setup.py | 3 +- 7 files changed, 286 insertions(+), 158 deletions(-) delete mode 100644 rsbbs/project_root.py diff --git a/rsbbs/__init__.py b/rsbbs/__init__.py index db87668..77a8d1d 100644 --- a/rsbbs/__init__.py +++ b/rsbbs/__init__.py @@ -1,5 +1,4 @@ #!/usr/bin/env python -# -*- coding: utf-8 -*- # # Really Simple BBS - a really simple BBS for ax.25 packet radio. # Copyright (C) 2023 John Burwell @@ -17,4 +16,4 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . -__all__ = ["bbs", "message", "parser", "project_root"] \ No newline at end of file +__all__ = ["rsbbs", "bbs", "message", "parser"] diff --git a/rsbbs/bbs.py b/rsbbs/bbs.py index 7c81742..7b3d279 100644 --- a/rsbbs/bbs.py +++ b/rsbbs/bbs.py @@ -1,5 +1,4 @@ #!/usr/bin/env python -# -*- coding: utf-8 -*- # # Really Simple BBS - a really simple BBS for ax.25 packet radio. # Copyright (C) 2023 John Burwell @@ -22,15 +21,14 @@ import os import subprocess import sys import yaml +import platformdirs -from sqlalchemy import create_engine, delete, select +from sqlalchemy import create_engine, delete, select, or_ from sqlalchemy.orm import Session from rsbbs.message import Message, Base from rsbbs.parser import Parser -import platformdirs - # Main BBS class @@ -38,14 +36,14 @@ class BBS(): def __init__(self, sysv_args): - self.sysv_args = sysv_args + self._sysv_args = sysv_args - self.config = self.load_config(sysv_args.config_file) + self.config = self._load_config(sysv_args.config_file) self.calling_station = sysv_args.calling_station.upper() - self.engine = self.init_engine() - self.parser = self.init_parser() + self.engine = self._init_engine() + self.parser = self._init_parser() logging.config.dictConfig(self.config['logging']) @@ -57,100 +55,189 @@ class BBS(): def config(self, value): self._config = value - # Load the config file - def load_config(self, config_file): + # + # Config file + # - config_dir = platformdirs.user_config_dir(appname='rsbbs', ensure_exists=True) - - config_path = os.path.join(config_dir, 'config.yaml') - config_template_path = os.path.join(os.path.dirname(__file__), 'config_default.yaml') + 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 the file doesn't exist there, create it if not os.path.exists(config_path): - 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) - + config_template_path = os.path.join( + os.path.dirname(__file__), + '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 yaml.YAMLError as e: - print(f"Could not load configuration file. Error: {e}") - exit(1) - except FileNotFoundError as e: - print(f'Configuration file full path: {os.path.abspath(config_file)}') - print(f"Configuration file {config_file} could not be found. Error: {e}") - exit(1) - except Exception as msg: - print(f"Error while loading configuration file {config_file}. Error: {e}") + except Exception as e: + print(f"Error loading configuration file: {e}") exit(1) - logging.info(f"Configuration file was successfully loaded. File name: {config_file}") + logging.info(f"Configuration file was successfully loaded." + f"File name: {config_path}") return config - - # Set up the BBS command parser + # + # BBS command parser + # - def init_parser(self): + 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.bye, {}), - ('delete', ['d', 'k'], 'Delete a message', self.delete, - {'number': {'help': 'The numeric index of the message to delete'}}, - ), - ('deletem', ['dm', 'km'], 'Delete all your messages', self.delete_mine, {}), - ('help', ['h', '?'], 'Show help', self.help, {}), - ('heard', ['j'], 'Show heard stations log', self.heard, {}), - ('list', ['l'], 'List all messages', self.list, {}), - ('listm', ['lm'], 'List only messages addressed to you', self.list_mine, {}), - ('read', ['r'], 'Read messages', self.read, - {'number': {'help': 'Message number to read'}} - ), - ('readm', ['rm'], 'Read only messages addressed to you', self.read_mine, {}), - ('send', ['s'], 'Send a new message to a user', self.send, + # (name, + # aliases, + # helpmsg, + # function, + # {arg: + # {arg attributes}, + # ...}) + ('bye', + ['b', 'q'], + 'Sign off and disconnect', + self.bye, + {}), + + ('delete', + ['d', 'k'], + 'Delete a message', + self.delete, + {'number': + {'help': + 'The numeric index of the message to delete'}},), + + ('deletem', + ['dm', 'km'], + 'Delete all your messages', + self.delete_mine, + {}), + + ('help', + ['h', '?'], + 'Show help', + self.help, + {}), + + ('heard', + ['j'], + 'Show heard stations log', + self.heard, + {}), + + ('list', + ['l'], + 'List all messages', + self.list, + {}), + + ('listm', + ['lm'], + 'List only messages addressed to you', + self.list_mine, + {}), + + ('read', + ['r'], + 'Read messages', + self.read, + {'number': {'help': 'Message number to read'}}), + + ('readm', + ['rm'], + 'Read only messages addressed to you', + self.read_mine, + {}), + + ('send', + ['s'], + 'Send a new message to a user', + self.send, { 'callsign': {'help': 'Message recipient callsign'}, '--subject': {'help': 'Message subject'}, '--message': {'help': 'Message'}, - }, - ), - ('sendp', ['sp'], 'Send a private message to a user', self.send_private, + },), + + ('sendp', + ['sp'], + 'Send a private message to a user', + self.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 - + # # Database + # - def init_engine(self): - 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) + 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 - + # # Input and output - - def read_line(self, prompt): + # + + def _read_line(self, prompt): + """Read a single line of input, with an optional prompt, + until we get something. + """ output = None - while output == None: + while not output: if prompt: - self.write_output(prompt) + self._write_output(prompt) input = sys.stdin.readline().strip() if input != "": output = input return output - def read_multiline(self, prompt): + 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) + self._write_output(prompt) while True: line = sys.stdin.readline() if line.lower().strip() == "/ex": @@ -159,118 +246,166 @@ class BBS(): output.append(line) return ''.join(output) - def write_output(self, output): + def _write_output(self, output): + """Write something to stdout.""" sys.stdout.write(output + '\r\n') - def print_message_list(self, results): - self.write_output(f"{'MSG#': <{5}} {'TO': <{9}} {'FROM': <{9}} {'DATE': <{10}} SUBJECT") - for result in results: - self.write_output(f"{result.Message.id: <{5}} {result.Message.recipient: <{9}} {result.Message.sender: <{9}} {result.Message.datetime.strftime('%Y-%m-%d')} {result.Message.subject}") + 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): - self.write_output(f"") - self.write_output(f"Message: {message.Message.id}") - self.write_output(f"Date: {message.Message.datetime.strftime('%A, %B %-d, %Y at %-H:%M %p UTC')}") - # self.write_output(f"Date: {message.Message.datetime.strftime('%Y-%m-%dT%H:%M:%S+0000')}") - 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"\r\n{message.Message.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}") + # # BBS command functions + # def bye(self, args): - '''Disconnect and exit''' - self.write_output("Bye!") + """Close the connection and exit.""" + self._write_output("Bye!") exit(0) def delete(self, args): - '''Delete message specified by numeric index''' + """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.write_output(f"Deleted message #{args.number}") + self._write_output(f"Deleted message #{args.number}") except Exception as e: - self.write_output(f"Unable to delete message #{args.number}") + self._write_output(f"Unable to delete message #{args.number}") def delete_mine(self, args): - '''Delete all messages addressed to user''' - self.write_output("Delete all messages addressed to you? Y/N:") + """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: with Session(self.engine) as session: try: - statement = delete(Message).where(Message.recipient == self.calling_station).returning(Message) + statement = delete(Message).where( + Message.recipient == self.calling_station + ).returning(Message) results = session.execute(statement) count = len(results.all()) if count > 0: - self.write_output(f"Deleted {count} messages") + self._write_output(f"Deleted {count} messages") session.commit() else: - self.write_output(f"No messages to delete.") + self._write_output(f"No messages to delete.") except Exception as e: - self.write_output(f"Unable to delete messages: {e}") + self._write_output(f"Unable to delete messages: {e}") def heard(self, args): - '''Show heard stations log''' - self.write_output(f"Heard stations:") + """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 = subprocess.run(['mheard'], capture_output=True, text=True) - self.write_output(result.stdout) + self._write_output(result.stdout) def help(self, args): - '''Print help''' + """Show help.""" self.parser.print_help() def list(self, args): - '''List all messages''' + """List all messages.""" with Session(self.engine) as session: - statement = select(Message).where((Message.is_private == False) | (Message.recipient == self.calling_station)) + statement = select(Message).where(or_( + (Message.is_private.is_not(False)), + (Message.recipient.__eq__(self.calling_station))) + ) results = session.execute(statement) self.print_message_list(results) def list_mine(self, args): - '''List only messages addressed to user''' + """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) + statement = select(Message).where( + Message.recipient == self.calling_station) results = session.execute(statement) self.print_message_list(results) def read(self, args): - '''Read messages''' + """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.print_message(result) def read_mine(self, args): - '''Read only messages addressed to user''' + """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) + statement = select(Message).where( + Message.recipient == self.calling_station) result = session.execute(statement) messages = result.all() count = len(messages) if count > 0: - self.write_output(f"Reading {count} messages:") + self._write_output(f"Reading {count} messages:") for message in messages: self.print_message(message) - self.write_output("Enter to continue...") + self._write_output("Enter to continue...") sys.stdin.readline() else: - self.write_output(f"No messages to read.") + self._write_output(f"No messages to read.") def send(self, args, is_private=False): - '''Create a message addressed to another user''' + """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:") + args.callsign = self._read_line("Callsign:") if not args.subject: - args.subject = self.read_line("Subject:") + args.subject = self._read_line("Subject:") if not args.message: - args.message = self.read_multiline("Message - end with /ex on a single line:") + 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(), @@ -281,25 +416,41 @@ class BBS(): )) try: session.commit() - self.write_output("Message saved!") + self._write_output("Message saved!") except Exception as e: session.rollback() - self.write_output("Error saving message. Contact the sysop for assistance.") + 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 loop + # def run(self): # Show greeting - self.write_output(f"[RSBBS-1.0.0] listening on {self.config['callsign']} ") - self.write_output(f"Welcome to {self.config['bbs_name']}, {self.calling_station}") - self.write_output(self.config['banner_message']) - self.write_output("For help, enter 'h'") + greeting = [] + greeting.append(f"[RSBBS-1.0.0] listening on " + f"{self.config['callsign']}") + greeting.append(f"Welcome to {self.config['bbs_name']}, " + f"{self.calling_station}") + greeting.append(self.config['banner_message']) + greeting.append("For help, enter 'h'") + self._write_output('\r\n'.join(greeting)) # Show initial prompt to the calling user - self.write_output(self.config['command_prompt']) + self._write_output(self.config['command_prompt']) # Parse the BBS interactive commands for the rest of time for line in sys.stdin: @@ -310,4 +461,4 @@ class BBS(): pass # Show our prompt to the calling user again - self.write_output(self.config['command_prompt']) + self._write_output(self.config['command_prompt']) diff --git a/rsbbs/message.py b/rsbbs/message.py index 85927ab..43281c5 100644 --- a/rsbbs/message.py +++ b/rsbbs/message.py @@ -1,5 +1,4 @@ #!/usr/bin/env python -# -*- coding: utf-8 -*- # # Really Simple BBS - a really simple BBS for ax.25 packet radio. # Copyright (C) 2023 John Burwell @@ -22,9 +21,11 @@ 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) @@ -32,6 +33,6 @@ class Message(Base): 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)) + 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 360fc50..1bdc78e 100644 --- a/rsbbs/parser.py +++ b/rsbbs/parser.py @@ -1,5 +1,4 @@ #!/usr/bin/env python -# -*- coding: utf-8 -*- # # Really Simple BBS - a really simple BBS for ax.25 packet radio. # Copyright (C) 2023 John Burwell @@ -19,6 +18,7 @@ import argparse + # We want to override the error and exit methods of ArgumentParser # to prevent it exiting unexpectedly or spewing error data over the air @@ -61,4 +61,4 @@ class Parser(BBSArgumentParser): subparser.add_argument(arg_name, **options) subparser.set_defaults(func=func) - return parser + return parser diff --git a/rsbbs/project_root.py b/rsbbs/project_root.py deleted file mode 100644 index 92dd62d..0000000 --- a/rsbbs/project_root.py +++ /dev/null @@ -1,28 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -# -# 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 . - -''' -This is a placeholder to be able to access the location of project root directory. -This is needed to be able to read config.yaml from this package, when it is distributed with setuptools and installed. -''' - -import os - -def path(): - return os.path.dirname(__file__) diff --git a/rsbbs/rsbbs.py b/rsbbs/rsbbs.py index 6c712a0..0291692 100755 --- a/rsbbs/rsbbs.py +++ b/rsbbs/rsbbs.py @@ -1,5 +1,4 @@ #!/usr/bin/env python -# -*- coding: utf-8 -*- # # Really Simple BBS - a really simple BBS for ax.25 packet radio. # Copyright (C) 2023 John Burwell @@ -22,6 +21,7 @@ import sys from rsbbs.bbs import BBS + def main(): # Parse and handle the system invocation arguments @@ -30,13 +30,18 @@ def main(): # Configure args: args_list = [ - #[ short, long, action, default, dest, help, required ] - ['-d', '--debug', 'store_true', None, 'debug', 'Enable debugging output to stdout', False], - ['-s', '--calling-station', 'store', 'N0CALL', 'calling_station', 'The callsign of the calling station', True], - ['-f', '--config-file', 'store', 'config.yaml', 'config_file', 'specify path to config.yaml file', False], + # [ short, long, action, default, dest, help, required ] + ['-d', '--debug', 'store_true', None, 'debug', + 'Enable debugging output to stdout', False], + ['-s', '--calling-station', 'store', 'N0CALL', 'calling_station', + 'The callsign of the calling station', True], + ['-f', '--config-file', 'store', 'config.yaml', 'config_file', + 'specify path to config.yaml file', False], ] for arg in args_list: - sysv_parser.add_argument(arg[0], arg[1], action=arg[2], default=arg[3], dest=arg[4], help=arg[5], required=arg[6]) + sysv_parser.add_argument( + arg[0], arg[1], action=arg[2], default=arg[3], dest=arg[4], + help=arg[5], required=arg[6]) # Version arg is special: sysv_parser.add_argument('-v', '--version', @@ -52,5 +57,6 @@ def main(): # Start the main BBS loop bbs.run() + if __name__ == "__main__": main() diff --git a/setup.py b/setup.py index 7f6b57b..f911ed6 100644 --- a/setup.py +++ b/setup.py @@ -1,5 +1,4 @@ #!/usr/bin/env python -# -*- coding: utf-8 -*- # # Really Simple BBS - a really simple BBS for ax.25 packet radio. # Copyright (C) 2023 John Burwell @@ -37,7 +36,7 @@ setup( license=license, packages=find_packages(exclude=('tests', 'docs')), data_files=[('', ['config_default.yaml'])], - entry_points = ''' + entry_points=''' [console_scripts] rsbbs=rsbbs.rsbbs:main '''