diff options
| -rw-r--r-- | geldschieberbot.py | 126 | ||||
| -rwxr-xr-x | test.py | 19 |
2 files changed, 83 insertions, 62 deletions
diff --git a/geldschieberbot.py b/geldschieberbot.py index c58887b..f876461 100644 --- a/geldschieberbot.py +++ b/geldschieberbot.py @@ -65,6 +65,13 @@ class GeldschieberbotJSONEncoder(json.JSONEncoder): return json.JSONEncoder.default(self, o) +@dataclass +class Reply: + """Class representing a reply from the bot""" + msg: str + attachment: T.Optional[str] = None + + def to_cent(euro) -> int: """Parse string containing euros into a cent value""" if '.' in euro: @@ -211,20 +218,6 @@ class Geldschieberbot: return Modification(mod.donor, mod.recipient, mod.amount) - def send(self, - msg, - attachment=None, - cmd='', - quote: T.Optional[Quote] = None): - """Send a message with optional attachment""" - cmd = cmd or self.send_cmd - if not self.quiet: - if attachment: - cmd += f' -a {attachment}' - if quote: - cmd += f' --quote-timestamp={quote.timestamp} --quote-author={quote.author}' - subprocess.run(cmd.split(' '), input=msg.encode(), check=False) - def create_summary(self, user, include=None) -> str: """Create a summary for a user""" msg = '' @@ -1082,22 +1075,12 @@ class Geldschieberbot: self.aliases[alias] = users return {'msg': f'New alias "{alias}" registered'} - def __init__(self, - state_path='', - group_id='', - send_cmd='', - dry_run=False, - quote_cmd=True): - self.state_path = state_path or os.environ['GSB_STATE_FILE'] - - self.group_id = group_id or os.environ["GSB_GROUP_ID"] - - self.send_cmd = send_cmd or os.environ["GSB_SEND_CMD"] - self.group_send_cmd = self.send_cmd + self.group_id + def __init__(self, state_path, group_id, dry_run=False): + """Create a new Geldschieberbot""" + self.state_path = state_path + self.group_id = group_id self.dry_run = dry_run # Run without changing the stored state - self.quote_cmd = quote_cmd # Quote the message causing the reply - self.quiet = False # Run without sending messages self.record_changes = True # Should changes be recorded self.load_state() @@ -1165,24 +1148,25 @@ class Geldschieberbot: if self.record_changes and not self.dry_run: self.changes[user].append(change) - def handle(self, envelope: dict): + def handle(self, envelope: dict) -> list[Reply]: """Parse and respond to a message""" sender_number = envelope["source"] if not "dataMessage" in envelope or not envelope[ "dataMessage"] or not envelope["dataMessage"]["message"]: - return + return [] message = envelope["dataMessage"] if message["groupInfo"] and message["groupInfo"][ "groupId"] != self.group_id: - return + return [] body = [l.strip() for l in message["message"].lower().splitlines()] if len(body) == 0 or not body[0].startswith('!'): - return + return [] + + replies: list[Reply] = [] - quote = Quote(timestamp=message["timestamp"], author=sender_number) args = body[0].split(' ') msg_context = MessageContext(sender_number, self.num2name.get(sender_number, None), @@ -1191,21 +1175,23 @@ class Geldschieberbot: if cmd in self.cmds: ret = self.cmds[cmd](msg_context) if 'err' in ret: - self.send(f'ERROR: {ret["err"]}') + replies.append(Reply(f'ERROR: {ret["err"]}', None)) else: if not 'msg' in ret: print(ret) - self.send(ret['msg'], - attachment=ret.get('attachment', None), - quote=quote if self.quote_cmd else None) + replies.append(Reply(ret['msg'], ret.get('attachment', None))) else: - self.send( - 'ERROR: unknown cmd. Enter !help for a list of commands.') + replies.append( + Reply( + 'ERROR: unknown cmd. Enter !help for a list of commands.', + None)) self.save_state() + return replies - def run_scheduled_cmds(self): + def run_scheduled_cmds(self) -> list[Reply]: """Progress the scheduled commands""" + replies: list[Reply] = [] self.record_changes = False now = datetime.now().date() @@ -1239,21 +1225,44 @@ class Geldschieberbot: out = f'Running {interval} command {name} for {runas} triggered on {d.isoformat()}' # TODO: use d as timestamp cmd_ctx = MessageContext(self.name2num[runas], runas, - cmd_args, [], None) + cmd_args, [], '') ret = self.cmds[cmd_args[0]](cmd_ctx) if 'err' in ret: - self.send(f'{out}\nERROR: {ret["err"]}') + replies.append( + Reply(f'{out}\nERROR: {ret["err"]}', None)) else: - self.send(f'{out}\n{ret["msg"]}', - attachment=ret.get('attachment', None)) + replies.append( + Reply(f'{out}\n{ret["msg"]}', + ret.get('attachment', None))) scheduled_cmd["last_time"] = d.isoformat() else: break + return replies + + +def send(_cmd, msgs: list[Reply], quote: T.Optional[Quote] = None): + """Send a message with optional attachment""" + for msg in msgs: + cmd = _cmd + if msg.attachment: + cmd += f' -a {msg.attachment}' + + if quote: + cmd += f' --quote-timestamp={quote.timestamp} --quote-author={quote.author}' + + subprocess.run(cmd.split(' '), input=msg.msg.encode(), check=False) + + +def die(msg: str, status=1): + """Exit because an error ocurred""" + print(msg, file=sys.stderr) + sys.exit(status) def main(): + """Read messages from stdin and send the bot's replies back""" parser = argparse.ArgumentParser() parser.add_argument('-d', '--dry-run', @@ -1272,23 +1281,36 @@ def main(): if args.dry_run: print("Dry Run no changes will apply!") - bot = Geldschieberbot(state_path=args.state_file, - group_id=args.group_id, - send_cmd=args.send_cmd, - dry_run=args.dry_run, - quote_cmd=not args.no_quote) + state_path = args.state_file or os.environ['GSB_STATE_FILE'] + if not state_path: + die('A state path must be provided') + + send_cmd = args.send_cmd or os.environ['GSB_SEND_CMD'] + if not send_cmd: + die('A send command must be provided') + + group_id = args.group_id or os.environ['GSB_GROUP_ID'] + if not group_id: + die('A group id must be provided') + + bot = Geldschieberbot(state_path, group_id, dry_run=args.dry_run) # Read cmds from stdin for line in sys.stdin.read().splitlines(): try: - message = json.loads(line)["envelope"] + envelope = json.loads(line)['envelope'] except json.JSONDecodeError: print(datetime.now(), line, "not valid json") continue - bot.handle(message) + quote = None + if not args.no_quote: + quote = Quote(timestamp=envelope['timestamp'], + author=envelope['source']) + + send(send_cmd, bot.handle(envelope), quote) - bot.run_scheduled_cmds() + send(send_cmd, bot.run_scheduled_cmds()) if __name__ == "__main__": @@ -12,8 +12,12 @@ from geldschieberbot import Geldschieberbot, GeldschieberbotJSONEncoder alice, bob, charlie = "alice", "bob", "charlie" num = {alice: "+49123456", bob: "+49654321", charlie: "+49615243"} -os.environ["GSB_GROUP_ID"] = "test" -os.environ["GSB_STATE_FILE"] = "test/state.json" + +DEFAULT_STATE_FILE = 'test/state.json' +DEFAULT_GROUP_ID = 'test' + +os.environ["GSB_GROUP_ID"] = DEFAULT_GROUP_ID +os.environ["GSB_STATE_FILE"] = DEFAULT_STATE_FILE os.environ["GSB_SEND_CMD"] = "cat" os.environ["GSB_SEND_GROUP"] = "cat" os.environ["GSB_SEND_USER"] = "cat" @@ -1391,19 +1395,14 @@ class TestConvertState(unittest.TestCase): class TestStateLoadStore(unittest.TestCase): - def test_default_init(self): - bot = Geldschieberbot() - self.assertTrue( - compare_state(os.environ['GSB_STATE_FILE'], state=bot.state)) - - def test_explicit_init(self): + def test_init(self): sp = "test/state_3users.json" - bot = Geldschieberbot(sp) + bot = Geldschieberbot(sp, DEFAULT_GROUP_ID) self.assertTrue(compare_state(sp, state=bot.state)) def test_explicit_load(self): sp = "test/state_3users.json" - bot = Geldschieberbot() + bot = Geldschieberbot(DEFAULT_STATE_FILE, DEFAULT_GROUP_ID) bot.load_state(sp) self.assertTrue(compare_state(sp, state=bot.state)) |