jmbwell/rsbbs
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor to make linter happy

Browse Source
This commit is contained in:
John Burwell 2023-04-24 14:23:49 -05:00
parent 3f27a1bd7a
commit 800d7de809
7 changed files with 286 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
rsbbs/__init__.py
View File

@ -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 <john@atatdotdot.com>
@ -17,4 +16,4 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
__all__ = ["bbs", "message", "parser", "project_root"]
__all__ = ["rsbbs", "bbs", "message", "parser"]

381
rsbbs/bbs.py
View File

@ -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 <john@atatdotdot.com>
@ -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'])

7
rsbbs/message.py
View File

@ -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 <john@atatdotdot.com>
@ -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)

4
rsbbs/parser.py
View File

@ -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 <john@atatdotdot.com>
@ -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

28
rsbbs/project_root.py
View File

@ -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 <john@atatdotdot.com>
#
# 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 <https://www.gnu.org/licenses/>.
'''
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__)

18
rsbbs/rsbbs.py
View File

@ -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 <john@atatdotdot.com>
@ -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()

3
setup.py
View File

@ -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 <john@atatdotdot.com>
@ -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
'''