Instantly share code, notes, and snippets.

@revmischa revmischa/handler.py
Created Jan 15, 2019

Embed
What would you like to do?
Twilio call handler lambda
from twilio.twiml.voice_response import VoiceResponse, Dial
from twilio.rest import Client
from urllib.parse import parse_qsl
import requests
import json
from pprint import pprint
import boto3
session = boto3.session.Session()
secrets = session.client(service_name='secretsmanager')
get_secret_value_response = secrets.get_secret_value(SecretId='twilio')
twil_config = json.loads(get_secret_value_response['SecretString'])
TWILIO_ACCT_SID = twil_config['TWILIO_ACCT_SID']
TWILIO_AUTH_TOKEN = twil_config['TWILIO_AUTH_TOKEN']
twil = Client(TWILIO_ACCT_SID, TWILIO_AUTH_TOKEN)
FORWARD_NUM = '+15101234567'
HOOK_URL = 'https://hooks.slack.com/services/T0000000/B324584/Zgatq2AS5FccT'
def incoming_callme(event, context):
call = dict(parse_qsl(event['body']))
from_num = call['From']
print(f"call from {from_num}")
lookup = lookup_number(call)
reputation = check_rep(call)
is_robo = check_is_robo(call)
caller = call['CallerName'] if 'CallerName' in call else call['Caller']
attachments = [{
'author_name': f"Call from {from_num}",
'text': f"Caller: {caller}",
}, {
'author_name': f"Reputation",
'text': reputation,
}, {
'author_name': f"Is Robo",
'text': "Yes" if is_robo else "No",
}]
attachments.append({"text": json.dumps(lookup, indent=4)})
slack(
text=f"Call from {from_num}",
icon=":telephone:",
attachments=attachments,
)
response = VoiceResponse()
# spam?
if reputation >= 3 or is_robo:
response.hangup()
return twiml_res(response)
dial = Dial()
dial.number(FORWARD_NUM)
response.append(dial)
return twiml_res(response)
def twiml_res(response):
return {
"statusCode": 200,
"headers": {"Content-Type": 'application/xml'},
"body": str(response),
}
return str(response)
def format_stuff(rec):
if len(rec) == 1:
rec = rec[0]
return json.dumps(rec, indent=2)
def check_rep(call):
res = twil.lookups.phone_numbers(call['From']).fetch(add_ons='whitepages_pro_phone_rep')
res = res.add_ons['results']['whitepages_pro_phone_rep']['result']
if 'reputation_level' not in res:
return -1
rep = res['reputation_level']
return rep
def check_is_robo(call):
res = twil.lookups.phone_numbers(call['From']).fetch(add_ons='nomorobo_spamscore')
res = res.add_ons['results']['nomorobo_spamscore']['result']
if 'score' not in res:
return False
return bool(res['score'])
def lookup_number(call):
res = twil.lookups.phone_numbers(call['From']).fetch(add_ons='whitepages_pro_caller_id')
res = res.add_ons['results']['whitepages_pro_caller_id']
pprint(res)
return res['result']
def slack(text=None, title=None, attachments=[], icon=None):
if not attachments:
return
# emoji icon
icon = 'telephone'
message = {
"username": title,
# "channel": channel,
"icon_emoji": f":{icon}:",
"attachments": attachments,
"text": text,
}
slack_raw(message)
def slack_raw(payload):
response = requests.post(
HOOK_URL, data=json.dumps(payload),
headers={'Content-Type': 'application/json'}
)
if response.status_code != 200:
raise ValueError(
'Request to slack returned an error %s, the response is:\n%s'
% (response.status_code, response.text)
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment