107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
#!/data/data/com.termux/files/usr/bin/python3
|
|
|
|
from subprocess import Popen, DEVNULL, PIPE
|
|
from json import loads as read_json
|
|
from threading import Thread
|
|
from time import sleep
|
|
|
|
sms_list_program = '/data/data/com.termux/files/usr/bin/termux-sms-list'
|
|
sms_send_program = '/data/data/com.termux/files/usr/bin/termux-sms-send'
|
|
|
|
|
|
class SMS_Service (Thread):
|
|
def __init__ (self, sms_list = None):
|
|
Thread.__init__ (self)
|
|
if type (sms_list) == list:
|
|
self.sms_list = sms_list
|
|
else:
|
|
self.sms_list = []
|
|
|
|
def build_database (self):
|
|
limit_fast_get = 1000000
|
|
offset = 0
|
|
while True:
|
|
msgs = self.get_sms (limit = limit_fast_get, offset = offset)
|
|
if len (msgs) != 0:
|
|
self.sms_list += msgs
|
|
if len (msgs) < limit_fast_get:
|
|
break
|
|
offset += limit_fast_get
|
|
for msg in self.sms_list:
|
|
del (msg ['read'])
|
|
|
|
def update_database (self):
|
|
limit_normal_get = 20
|
|
getting_new = True
|
|
offset = 0
|
|
while getting_new:
|
|
msgs = self.get_sms (limit = limit_normal_get, offset = offset)
|
|
getting_new = False
|
|
for msg in msgs:
|
|
del (msg ['read'])
|
|
for msg in msgs:
|
|
if msg not in self.sms_list:
|
|
getting_new = True
|
|
self.sms_list.append (msg)
|
|
offset += limit_normal_get
|
|
|
|
def run (self):
|
|
self.build_database ()
|
|
while True:
|
|
sleep (5)
|
|
self.update_database ()
|
|
|
|
def get_sms (self, limit = 10, offset = 0):
|
|
while True:
|
|
try:
|
|
if type (limit) != int or type (offset) != int or limit < 1 or offset < 0:
|
|
raise (TypeError ('limit and offset must be positive integers'))
|
|
p = Popen (
|
|
[sms_list_program, '-l', str (limit), '-o', str (offset)],
|
|
stdin = DEVNULL,
|
|
stdout = PIPE
|
|
)
|
|
return (read_json (p.communicate () [0]))
|
|
except Exception:
|
|
pass
|
|
|
|
def send_sms (self, dest, message):
|
|
p = Popen (
|
|
[sms_send_program, '-n', dest.replace (' ', '')],
|
|
stdin = PIPE,
|
|
stdout = DEVNULL
|
|
)
|
|
p.communicate (message.encode ('utf-8'))
|
|
|
|
def sorted_list (self):
|
|
return (sorted (self.sms_list, key = lambda item: item ['received']))
|
|
|
|
def thread_list (self):
|
|
l = {}
|
|
for i in self.sms_list:
|
|
if i ['threadid'] in l:
|
|
if l [i ['threadid']] [1] < i ['received']:
|
|
l [i ['threadid']] [1] = i ['received']
|
|
l [i ['threadid']] [2] = i ['body']
|
|
l [i ['threadid']] [3] += 1
|
|
if len (l [i ['threadid']]) == 4 and 'sender' in i and i ['type'] == 'inbox':
|
|
l [i ['threadid']].append (i ['sender'])
|
|
else:
|
|
if 'sender' in i and i ['type'] == 'inbox':
|
|
l [i ['threadid']] = [i ['number'], i ['received'], i ['body'], 1, i ['sender']]
|
|
else:
|
|
l [i ['threadid']] = [i ['number'], i ['received'], i ['body'], 1]
|
|
return (l)
|
|
|
|
def get_thread (self, id):
|
|
l = []
|
|
number = None
|
|
name = None
|
|
for i in self.sms_list:
|
|
if str (i ['threadid']) == str (id):
|
|
l.append ((i ['received'], True if i ['type'] == 'inbox' else False, i ['body']))
|
|
if number is None:
|
|
number = i ['number']
|
|
if name is None and i ['type'] == 'inbox' and 'sender' in i:
|
|
name = i ['sender']
|
|
return (number, name, sorted (l, key = lambda item: item [0])) |