-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtexter.py
58 lines (49 loc) · 1.85 KB
/
texter.py
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
import random
import time
import os
from datetime import datetime, timedelta
from config import (TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN,
TWILIO_PHONE_NUMBER, MESSAGES, REST_HOURS)
import daemon
from twilio.rest import TwilioRestClient
random.seed(os.urandom(10))
def start_texting():
twilio_client = TwilioRestClient(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)
start_time = datetime.now()
next_message = None
while True:
get_next_send_times(start_time)
next_message = next_send_message()
if next_message:
time.sleep((next_message['next_send'] - datetime.now()).total_seconds())
now = datetime.now()
if now < today_at_hour(now, REST_HOURS['start']) and now > today_at_hour(now, REST_HOURS['end']):
twilio_client.sms.messages.create(body=random.choice(next_message['messages']),
to=next_message['phone_number'],
from_=TWILIO_PHONE_NUMBER)
def today_at_hour(now, hour):
return datetime(now.year, now.month, now.day, hour)
def next_send_message():
time_asc = sorted(MESSAGES, key=lambda x: x.get('next_send', datetime(4000, 1, 1)))
for t in time_asc:
if t['next_send'] > datetime.now():
return t
return None
def get_next_send_times(start_time):
for message in MESSAGES:
next_send = message.get('next_send')
if not next_send:
message['send_num'] = 0
message['next_send'] = pick_next_send_time(message, start_time)
elif next_send < datetime.now():
message['send_num'] += 1
message['next_send'] = pick_next_send_time(message, start_time)
def pick_next_send_time(message, start_time):
''' Randomly picks a send time between start_time+interval*send_number
and start_time+interval*(send_number+1)
'''
low = start_time + timedelta(seconds=message['interval']*message['send_num'])
return low + timedelta(seconds=random.randint(0, message['interval']))
if __name__ == '__main__':
with daemon.DaemonContext():
start_texting()