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
|
#!/usr/bin/env python3
import os
import sys
import uuid
import base64
import getopt
import requests
import tempfile
from libs.pysignald.signald import Signal
from configparser import ConfigParser
def usage():
print(os.path.basename(sys.argv[0]) + " [-h|--help] [-r|--register]")
print("Runs the signal frontend to N. Reads the number to use from n_signal.ini")
print(" -h|--help prints this help and exits")
print(" -r|--register register the number with signal")
def read_config():
config_object = ConfigParser()
# config_object.read(os.environ.get('XDG_CONFIG_HOME') + "/n_signal.ini")
config_object.read("n_signal.ini")
info = config_object["INFO"]
return info["number"]
def register_signald(number):
s = Signal(number)
s.register(voice=False)
code = input(f"Please input register code send via sms to {number}: ")
s.verify(code)
def handle_replies(replies):
answer = ""
if type(replies) is not list:
replies = [replies]
for reply in replies:
l_answer = reply['text']
for annotation in reply["annotations"]:
a_type = annotation["type"]
a_extra, is_attachment = handle_replies(annotation["extra"])
if a_type == "command":
l_answer = f"/{l_answer}: {a_extra}\n"
elif a_type == "link":
l_answer = f"{a_extra} ({l_answer})"
elif a_type == "bold":
l_answer = f"*{l_answer}*"
elif a_type == "italic":
l_answer = f"_{l_answer}_"
elif a_type == "strikethrough":
l_answer = f"~{l_answer}~"
elif a_type == "attachment":
[file_name, file_type, file_content] = a_extra.split(';')
try:
tmp_path = "/tmp/" + str(uuid.uuid1())
os.mkdir(tmp_path)
# hack, do clean
file_content = base64.b64decode(file_content)
file_path = tmp_path + '/' + file_name
with open(file_path, "wb") as f:
f.write(file_content)
return file_path, True
except Exception as e:
print(str(e))
answer += l_answer
return answer, False
def handle_response(response):
replies = response.json()['reply']
return handle_replies(replies)
def startup(number):
s = Signal(number)
# with args
@s.chat_handler("^/([^\\s]+)\\s+([^$]*)$", order=10) # This is case-insensitive.
def klinger(message, match):
# Returning `False` as the first argument will cause matching to continue
# after this handler runs.
stop = True
# TODO: try catch
response = requests.post('http://localhost:18080',
json={"command": match.group(1), "arguments": match.group(2).strip()})
answer, is_attachment = handle_response(response)
return stop, answer, is_attachment
# no args
@s.chat_handler("^/(.+)$", order=20) # This is case-insensitive.
def klinger(message, match):
# Returning `False` as the first argument will cause matching to continue
# after this handler runs.
stop = True
# TODO: try catch
response = requests.post('http://localhost:18080',
json={"command": match.group(1), "arguments": ""})
answer, is_attachment = handle_response(response)
return stop, answer, is_attachment
s.run_chat()
def main():
number = read_config()
try:
opts, args = getopt.getopt(sys.argv[1:], "hr", ["help", "register"])
except getopt.GetoptError as err:
# print help information and exit:
print(err) # will print something like "option -a not recognized"
usage()
sys.exit(2)
for o, a in opts:
if o in ("-h", "--help"):
usage()
sys.exit()
elif o in ("-r", "--register"):
register_signald(number)
sys.exit()
else:
assert False, "unhandled option: " + o
startup(number)
if __name__ == "__main__":
main()
|
