#!/usr/bin/env python3 import json, os, sys, shutil, fcntl from datetime import datetime BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_FILE = os.path.join(BASE_DIR, "pbbs_data.json") HEARD_FILE = os.path.join(BASE_DIR, "pbbs_heard.json") CONFIG_FILE = os.path.join(BASE_DIR, "pbbs_config.cfg") USER_CALL = sys.argv[1].upper() if len(sys.argv) > 1 else "GUEST" def load_json(path, default): """Loads JSON with Shared Locking (LOCK_SH) and self-healing logic.""" if not os.path.exists(path): return default try: with open(path, "r") as f: fcntl.flock(f, fcntl.LOCK_SH) data = json.load(f) fcntl.flock(f, fcntl.LOCK_UN) return data except Exception: print(f"*** Warning: {os.path.basename(path)} is corrupted.") bak_path = path + ".bak" if os.path.exists(bak_path): print(f"*** Restoring from backup: {os.path.basename(bak_path)}") try: with open(bak_path, "r") as f_bak: fcntl.flock(f_bak, fcntl.LOCK_SH) recovered_data = json.load(f_bak) fcntl.flock(f_bak, fcntl.LOCK_UN) save_json(path, recovered_data) return recovered_data except Exception: print("*** Critical: Backup also corrupted.") return default def save_json(path, data): """Saves JSON with Exclusive Locking (LOCK_EX) and Atomic Backup.""" if os.path.exists(path): shutil.copy2(path, path + ".bak") with open(path, "w") as f: try: fcntl.flock(f, fcntl.LOCK_EX) json.dump(data, f, indent=4) f.flush() os.fsync(f.fileno()) finally: fcntl.flock(f, fcntl.LOCK_UN) def load_config(): """Initial setup wizard with detailed per-setting comments in the config file.""" if not os.path.exists(CONFIG_FILE): print("--- PBBS Initial Setup ---") node_name = input("Enter node name or callsign (e.g., MYNODE): ").strip().upper() or "MYNODE" print("\nSYSOPS are users with full administrative rights (Delete/Pin).") user_input = input(f"Enter SYSOPS (comma separated, e.g., KK6SEN,W6XYZ): ") sysop_list = [c.strip().upper() for c in user_input.split(",") if c.strip()] if not sysop_list: sysop_list = [USER_CALL] print("\nPrivacy Mode: If YES, users can only see mail to/from them or public groups.") print("If NO, all messages are public to everyone.") privacy_input = input("Enable Privacy Mode? (Y/N): ").strip().upper() privacy_mode = "TRUE" if privacy_input == 'Y' else "FALSE" aliases = ["ALL"] if privacy_mode == "TRUE": print("\nPublic Aliases are group destinations anyone can read (e.g., TECH, PKTNET).") print("'ALL' is included by default.") alias_input = input("Enter additional Public Aliases (comma separated): ") extra_aliases = [c.strip().upper() for c in alias_input.split(",") if c.strip()] for a in extra_aliases: if a not in aliases: aliases.append(a) with open(CONFIG_FILE, "w") as f: f.write("# PBBS Configuration File\n") f.write("# Written by Josh KK6SEN\n\n") f.write("# The station name or callsign displayed in the BBS prompt and banner.\n") f.write(f"node_name={node_name}\n\n") f.write("# A comma-separated list of users with administrative (Delete/Pin) privileges.\n") f.write(f"sysops={','.join(sysop_list)}\n\n") f.write("# Set to TRUE to restrict private mail visibility; FALSE makes all mail public.\n") f.write(f"privacy_enabled={privacy_mode}\n\n") f.write("# Destinations that bypass privacy filters (e.g., ALL, PKTNET, or groups).\n") f.write(f"public_aliases={','.join(aliases)}\n") return { "node_name": node_name, "sysops": sysop_list, "privacy_enabled": privacy_mode == "TRUE", "public_aliases": aliases } config = {"sysops": [], "node_name": "PBBS", "privacy_enabled": False, "public_aliases": ["ALL"]} with open(CONFIG_FILE, "r") as f: for line in f: if "=" in line and not line.startswith("#"): key, val = line.strip().split("=", 1) if key == "sysops": config["sysops"] = [c.strip().upper() for c in val.split(",")] elif key == "node_name": config["node_name"] = val.upper() elif key == "privacy_enabled": config["privacy_enabled"] = val.upper() == "TRUE" elif key == "public_aliases": config["public_aliases"] = [c.strip().upper() for c in val.split(",")] return config CONFIG = load_config() NODE_NAME, SYSOP_CALLS, PRIVACY_ON, PUBLIC_DESTS = CONFIG["node_name"], CONFIG["sysops"], CONFIG["privacy_enabled"], CONFIG["public_aliases"] def is_sysop(call): return call in SYSOP_CALLS def can_view(msg, user): if is_sysop(user) or not PRIVACY_ON: return True if msg.get('to') in PUBLIC_DESTS: return True if user == "GUEST": return False return msg.get('to') == user or msg.get('from') == user def list_messages(data, limit=None, mine_only=False, call_filter=None, dir_filter=None): msgs = [m for m in data.get("messages", []) if can_view(m, USER_CALL) and 'date' in m] if not msgs: print("*** No messages\n\nend of messages") return if mine_only: msgs = [m for m in msgs if m.get('to') == USER_CALL and not m.get('read')] elif call_filter: if dir_filter == "FROM": msgs = [m for m in msgs if m.get('from') == call_filter] else: msgs = [m for m in msgs if m.get('to') == call_filter] if limit: msgs = msgs[-limit:] # Sort: Pinned first, then by date descending (newest on top) sorted_msgs = sorted(msgs, key=lambda x: (not x.get('pinned', False), x.get('date', ''), x.get('id', 0)), reverse=True) # Special adjustment: Ensure pinned messages stay at the very top despite global reverse pinned = [m for m in sorted_msgs if m.get('pinned')] unpinned = [m for m in sorted_msgs if not m.get('pinned')] final_list = pinned + unpinned print(f"{'MSG#':<5} {'ST':<2} {'SIZE':<6} {'TO':<8} {'FROM':<8} {'DATE':<18} {'SUBJECT'}") for m in final_list: st = "!" if m.get('pinned') else m.get('st', ' ') if m.get('to') == USER_CALL and not m.get('read'): st += "N" print(f"{m.get('id', 0):<5} {st:<2} {len(m.get('body', '')):<6} {m.get('to', '')[:8]:<8} {m.get('from', '')[:8]:<8} {m.get('date', '')[:18]:<18} {m.get('subject', '')[:26]}") print("\nend of messages") def main(): data = load_json(DATA_FILE, {"messages": [], "next_id": 1}) heard = load_json(HEARD_FILE, {}) heard[USER_CALL] = datetime.now().strftime("%m/%d/%y %H:%M:%S") save_json(HEARD_FILE, heard) print(f"*** Welcome {USER_CALL} to the {NODE_NAME} PBBS ***") total, used, free = shutil.disk_usage(BASE_DIR) print(f"{free // (2**30)}GB AVAILABLE ({used // (2**30)}GB USED OF {total // (2**30)}GB TOTAL) :)") list_messages(data) while True: try: prompt_cmds = "B,J,K,L,P,R,S, or Help" if is_sysop(USER_CALL) else "B,J,K,L,R,S, or Help" inp = input(f"\n[{NODE_NAME}] {prompt_cmds} > ").strip().split() if not inp: continue cmd, arg = inp[0].upper(), (inp[1].upper() if len(inp) > 1 else None) if cmd == 'B': break elif cmd == 'J': for c, t in sorted(heard.items(), key=lambda x: x[1], reverse=True): print(f"{c:<10} {t}") elif cmd == 'L': if not arg: list_messages(data) elif arg in ['<', '>'] and len(inp) > 2: list_messages(data, call_filter=inp[2].upper(), dir_filter="FROM" if arg == '<' else "TO") elif cmd == 'LL' and arg and arg.isdigit(): list_messages(data, limit=int(arg)) elif cmd == 'LM': list_messages(data, mine_only=True) elif cmd == 'R' and arg: m = next((x for x in data["messages"] if str(x.get('id')) == arg), None) if m and can_view(m, USER_CALL): print(f"MSG#{m['id']} {m['date']} FROM {m['from']} TO {m['to']}\nSUBJECT: {m['subject']}\n\n{m['body']}") if m['to'] == USER_CALL: m['read'] = True; save_json(DATA_FILE, data) elif cmd == 'RM': mine = [m for m in data["messages"] if m['to'] == USER_CALL and not m.get('read')] for m in mine: print(f"MSG#{m['id']} {m['date']} FROM {m['from']} TO {m['to']}\nSUBJECT: {m['subject']}\n\n{m['body']}") m['read'] = True if mine: save_json(DATA_FILE, data) elif cmd in ['S', 'SB', 'SP']: dest = arg if arg else "ALL" subj = input("SUBJECT: ").strip()[:50] or "(No Subject)" mid = data["next_id"]; data["next_id"] += 1 print(f"\nENTER MESSAGE {mid}--END WITH /EX"); body = [] while True: line = input() if line.upper() == "/EX": break body.append(line) data["messages"].append({"id": mid, "st": "B" if cmd == 'SB' else "P", "to": dest, "from": USER_CALL, "date": datetime.now().strftime("%m/%d/%y %H:%M:%S"), "subject": subj, "body": "\n".join(body), "read": False, "pinned": False}) save_json(DATA_FILE, data); print("MESSAGE SAVED") elif cmd == 'K' and arg: msg = next((x for x in data["messages"] if str(x.get('id')) == arg), None) if msg and (msg['from'] == USER_CALL or msg.get('to') == USER_CALL or is_sysop(USER_CALL)): if input(f"Delete message {arg}? (y/N): ").upper() == 'Y': data["messages"] = [x for x in data["messages"] if str(x.get('id')) != arg] save_json(DATA_FILE, data); print("MESSAGE DELETED") elif cmd == 'KM': if input("Delete ALL your read messages? (y/N): ").upper() == 'Y': data["messages"] = [m for m in data["messages"] if not (m.get('to') == USER_CALL and m.get('read'))] save_json(DATA_FILE, data); print("Read messages killed.") elif cmd == 'P' and is_sysop(USER_CALL) and arg: m = next((x for x in data["messages"] if str(x.get('id')) == arg), None) if m: m['pinned'] = not m.get('pinned', False); save_json(DATA_FILE, data); print("Toggled Pin.") elif cmd == 'HELP' or cmd == 'H': print(f"--- {NODE_NAME} PBBS Help ---") print("B(ye) PBBS WILL DISCONNECT") print("J(heard) CALLSIGNS WITH DAYSTAMP") print("K(ill) n DELETE MESSAGE NUMBER n") print("KM(ine) DELETE ALL READ MESSAGES ADDRESSED TO YOU") print("L LIST ALL ACCESSIBLE MESSAGES") print("L < call LIST MESSAGES FROM CALL") print("L > call LIST MESSAGES TO CALL") print("LL n LIST LAST n MESSAGES") print("LM(ine) LIST UNREAD MESSAGES ADDRESSED TO YOU") print("R(ead) n DISPLAY MESSAGE NUMBER n") print("RM(ine) READ ALL MESSAGES ADDRESSED TO YOU") print("S(end) call SEND PRIVATE MESSAGE (LEAVE BLANK FOR ALL)") print("SB call SEND BULLETIN") if is_sysop(USER_CALL): print("\nSYSOP COMMANDS:") print("P n PIN MESSAGE n (Forces to top of list with '!')") else: print("Not a valid command") except Exception as e: print(f"*** Error: {e}") if __name__ == '__main__': main()