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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
#!/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 and send replies"""
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 single_shot(bot: Geldschieberbot, send_cmd: str, no_quote: bool):
"""Handle all messages read from stdin and run scheduled commands"""
# Read messages 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 no_quote:
quote = Quote(timestamp=envelope['timestamp'],
author=envelope['source'])
send(send_cmd, bot.handle(envelope), quote)
send(send_cmd, bot.run_scheduled_cmds())
def dbus_loop(bot: Geldschieberbot, send_cmd: str, no_quote: bool):
"""Handle all messages read from stdin and run scheduled commands"""
from pydbus import SessionBus # type: ignore
from gi.repository import GLib
from base64 import b64encode
bus = SessionBus()
loop = GLib.MainLoop()
signal = bus.get('org.asamk.Signal')
# NOTE: when daemon was started without explicit `-u USERNAME`, replace the line above with
# signal = bus.get("org.asamk.Signal", "/org/asamk/Signal/_YOURPHONENUMBER")
def on_msg_recv(timestamp, source, groupid, message, _):
quote = None
if not no_quote:
quote = Quote(timestamp=timestamp, author=source)
envelope = {
"source": source,
"dataMessage": {
"timestamp": timestamp,
"message": message,
"groupInfo": {
"groupId": b64encode(bytes(groupid)).decode()
}
}
}
logfile_path = f'msgs/msg{datetime.now().isoformat()}.log'
with open(logfile_path, 'w', encoding='utf-8') as msglog:
print(f'{envelope}', file=msglog)
send(send_cmd, bot.handle(envelope), quote)
bot.save_state()
signal.onMessageReceived = on_msg_recv
def run_scheduled_cmds():
send(send_cmd, bot.run_scheduled_cmds())
bot.save_state()
return True
GLib.timeout_add_seconds(30, run_scheduled_cmds)
loop.run()
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('--dbus',
action='store_true',
help='receive messages via dbus in a loop')
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')
with Geldschieberbot(state_path, group_id, dry_run=args.dry_run) as bot:
if not args.dbus:
single_shot(bot, send_cmd, args.no_quote)
else:
dbus_loop(bot, send_cmd, args.no_quote)
if __name__ == "__main__":
main()
|
