1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
#!/usr/bin/env python3
# Copyright (c) 2023 Florian Fischer. All rights reserved.
#
# This file is part of geldschieberbot.
#
# geldschieberbot 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.
#
# geldschieberbot 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
# geldschieberbot found in the LICENSE file. If not,
# see <https://www.gnu.org/licenses/>.
"""Run geldschieberbot on all received messages, send replies, and exit"""
import argparse
from datetime import datetime
import json
import os
import subprocess
import sys
import typing as T
from geldschieberbot import Geldschieberbot
from models import Reply, Quote
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',
help='do not persist changes',
action='store_true')
parser.add_argument('-f', '--state-file', help='the state file')
parser.add_argument('-g', '--group-id', help='the group id to listen to')
parser.add_argument('--send-cmd',
help='the shell command used to send messages')
parser.add_argument('-nq',
'--no-quote',
help='not quote the message causing the reply',
action='store_true')
args = parser.parse_args()
if args.dry_run:
print("Dry Run no changes will apply!")
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:
envelope = json.loads(line)['envelope']
except json.JSONDecodeError:
print(datetime.now(), line, "not valid json")
continue
quote = None
if not args.no_quote:
quote = Quote(timestamp=envelope['timestamp'],
author=envelope['source'])
send(send_cmd, bot.handle(envelope), quote)
send(send_cmd, bot.run_scheduled_cmds())
if __name__ == "__main__":
main()
|
