Created
June 1, 2015 23:48
-
-
Save iL3D/38d85c3734cebdfbf725 to your computer and use it in GitHub Desktop.
photon demos
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import re | |
| import sys | |
| import traceback | |
| from flask import Flask | |
| from flask import render_template | |
| from flask import url_for | |
| from flask import request | |
| from flask import redirect | |
| from urllib.parse import quote | |
| from urllib.parse import urlencode | |
| from twilio import twiml | |
| from twilio.rest import TwilioRestClient | |
| from twilio.util import TwilioCapability | |
| # Declare and configure application | |
| app = Flask(__name__, static_url_path='/static') | |
| app.config.from_pyfile('local_settings.py') | |
| # ======================== util/endpoints | |
| class EndPoint(object): | |
| numbers = {} | |
| def __init__(self, tag=None): | |
| self.tag_=tag | |
| def is_sip (self): return 'sip' == self.tag_ | |
| def is_pstn (self): return (('simcard' == self.tag_) or ('land' == self.tag_)) | |
| def is_client(self): return 'app' == self.tag_ | |
| def tag(self, t=None): | |
| if t and (t in numbers): self.tag_=t | |
| return self.tag_ | |
| def number(self,tag=None): | |
| if not tag: tag=self.tag_ | |
| try: number = self.numbers[tag] | |
| except: number = None | |
| finally: return number | |
| def number_sanitized(number): | |
| n=None | |
| if re.match(r'^(\+1\d{10})$', number): | |
| n=number | |
| return n | |
| def alias_sanitized(alias): | |
| a=None | |
| if re.match(r'^(\w{3,5})$', alias): | |
| a=alias | |
| return a | |
| # util/endpoints ========================= | |
| EndPoint.numbers = { | |
| 'simcard':app.config['PHOTON_SIMCARD'], | |
| 'sip' :app.config['PHOTON_SIPADDR'], | |
| 'land' :app.config['PHOTON_LNDLINE'], | |
| 'app' :app.config['PHOTON_TIPNAME'],} | |
| app.designated_voice = EndPoint('sip' ) | |
| app.designated_message = EndPoint('simcard') | |
| app.designatedd_mobile = EndPoint('app' ) | |
| app.contact_codes = { 'abc':'+15555551234' } | |
| # Voice Request URL | |
| @app.route('/voice', methods=['GET', 'POST']) | |
| def voice(): | |
| response = twiml.Response() | |
| if request.values.get('AccountSid') != app.config['TWILIO_ACCOUNT_SID']: | |
| return str(response) | |
| url_voicemail = "http://twimlets.com/{}/vmail".format(app.config['TWILIO_ACCOUNT_SID']) | |
| url_callme = url_for('.callme',_external=True) | |
| url_menu = url_for('.menu' ,_external=True) | |
| message_menu = "To ring {}'s current location, please press 2. To connect, press 2. Or, to leave a voice mail, please press 5.".format(app.config['PHOTON_CALLER_NAME']) | |
| params_menu = urlencode({"Message":message_menu, "Option[2]":url_callme, "Option[5]":url_voicemail}) | |
| url_menu_with_params = "{}?{}".format(url_menu, params_menu) | |
| response.say("Please hold while your call is routed. Thank you.") | |
| response.redirect( url_menu_with_params ) | |
| return str(response) | |
| # SMS Request URL | |
| @app.route('/sms', methods=['GET', 'POST']) | |
| def sms(): | |
| response = twiml.Response() | |
| try: | |
| # if a stranger is knocking, don't respond. "goto" finally clause | |
| if (request.values.get('AccountSid') != app.config['TWILIO_ACCOUNT_SID']): | |
| raise Exception('bad-key') | |
| hq_phone_internal = app.config['TWILIO_CALLER_ID'] | |
| hq_phone_external = app.config['PHOTON_CID_HQ2'] | |
| live_message = app.designated_message.number() | |
| # parse incoming message body | |
| body_in = request.form['Body'] | |
| command = re.compile(r"""((?P<cmd>^photon)\ # command code then space | |
| (?P<arg1>\S+)\ # arg 1 then space | |
| (?P<arg2>\S+)\ # arg 2 then space | |
| (?P<arg3>\S*)$) # arg 3 then end | |
| | # OR match | |
| ((?P<send>^qq=) # send code NO space | |
| (?P<alias>\d{4})\ # alias code then space | |
| (?P<body>.+)$) # body then end""" | |
| ,re.VERBOSE).match(body_in) | |
| # got a regular incoming message, not a command; so | |
| # forward internally treating body as normal content | |
| if command == None: | |
| # relay sms body to an live phone, prepend original callee | |
| from_number = request.values.get('From') | |
| body_out = from_number + " - " + body_in | |
| response.message(body_out, to=live_message, sender=hq_phone_internal) | |
| # block commands from strangers | |
| elif request.values.get('From') != live_message: | |
| raise Exception('bad-commander') | |
| # qq=1234 content |send outgoing message by forwarding externally | |
| elif command.group('send') == 'qq=': | |
| alias = Endpoint.alias_sanitized( command.group('alias') ) | |
| if alias in app.contact_codes: | |
| phone_them = app.contact_codes[alias] | |
| body_out = command.group('body') | |
| response.message(body_out, to=phone_them, sender=hq_phone_internal) | |
| else: | |
| body_out = "Unknown alias {}".format(command.group('contact')) | |
| response.message(body_out, to=live_message, sender=hq_phone_internal) | |
| # photon connect us_tag them_phone | |
| elif command.group('arg1')== 'connect' and command.group('arg2') and command.group('arg3'): | |
| # Make it look like caller dialed callee. Use two outgoing voice | |
| # legs to dial caller and callee into same conference room. | |
| api_talker = TwilioRestClient( app.config['TWILIO_ACCOUNT_SID'], | |
| app.config['TWILIO_AUTH_TOKEN' ]) | |
| #url_conference_create = url_for('.conference_create', _external=True) | |
| url_conference_join = url_for('.conference_join' , _external=True) | |
| phone_us = EndPoint.number( command.group('arg2') ) | |
| if not phone_us: raise Exception('bad-tag') | |
| # don't trust inputs bearing gifts | |
| phone_them = EndPoint.number_sanitized( command.group('arg3') ) | |
| if not phone_them: raise Exception('bad-number') | |
| #print("{} {} {}".format(url_conference_join,phone_us ,hq_phone_internal)) | |
| #print("{} {} {}".format(url_conference_join,phone_them,hq_phone_external)) | |
| api_talker.calls.create(url=url_conference_join, to=phone_us , from_=hq_phone_internal) | |
| api_talker.calls.create(url=url_conference_join, to=phone_them, from_=hq_phone_external) | |
| # response.nop() ...just return empty response | |
| # photon recv voice tag_phone | |
| elif command.group('arg1')== 'recv' and command.group('arg2')=='voice' and command.group('arg3'): | |
| tag = app.designated_voice.tag(command.group('arg3')) | |
| if tag == command.group('arg3'): | |
| body_out = "Using " + tag + " for receiving voice" | |
| else: | |
| body_out = "Unchanged: using previous setting for receiving voice." | |
| response.message(body_out, to=live_message, sender=hq_phone_internal) | |
| # photon alias code phone | |
| elif command.group('arg1')=='alias' and command.group('arg2') and command.group('arg3'): | |
| alias = Endpoint.alias_sanitized ( command.group('arg2') ) | |
| contact = Endpoint.number_sanitized( command.group('arg3') ) | |
| if alias and contact: | |
| app.contact_codes[alias] = contact | |
| body_out = "Using alias {} for {}".format(alias,contact) | |
| else: | |
| body_out = "Alias:none set" | |
| response.message(body_out, to=live_message, sender=hq_phone_internal) | |
| # shouldn't get here, but the code gremlins will find a way... can't wait | |
| else: | |
| response.message("Please refrain from feeding gizmo after midnight") | |
| except Exception as e: | |
| # if got a bad key presumably from a stranger or other bad-ness, | |
| # don't say anything. at most maybe spout gibberish | |
| bads = ['bad-key','bad-number','bad-tag','bad-commander'] | |
| ex = e.args[0] | |
| if not ex in bads: | |
| body_out = "gREM Error. Code green {}-360A. {}".format(46,"Wet gremlins") | |
| response.say(body_out) | |
| finally: | |
| return str(response) | |
| # Birthday app | |
| @app.route('/birthday', methods=['GET','POST']) | |
| def birthday(): | |
| if request.values.get('AccountSid') != app.config['TWILIO_ACCOUNT_SID']: | |
| return str('<?xml version="1.0" encoding="UTF-8"?><Response />') | |
| url_domain = url_for('.index',_external=True) | |
| who = 'WHO' | |
| response = '<?xml version="1.0" encoding="UTF-8"?><Response><Say voice="alice">Hello. And Happy Birthday. This message is from your {}. For your birthday, some important people are waiting to sing to you.</Say><Gather numDigits="1" action="{}/static/assets/ivr2.xml" timeout="7" method="GET"><Say voice="alice">To listen, please press one</Say></Gather><Hangup/></Response>'.format(who,url_domain) | |
| return response | |
| # Callme ====================== | |
| # Call me at a designated number/phone | |
| @app.route('/callme', methods=['GET','POST']) | |
| def callme(): | |
| response = twiml.Response() | |
| if request.values.get('AccountSid') != app.config['TWILIO_ACCOUNT_SID']: | |
| return str(response) | |
| dial_call_status = request.values.get('DialCallStatus') | |
| dial_status = request.values.get('DialStatus' ) | |
| dial = request.values.get('Dial' ) | |
| # Second pass only, after initial call | |
| if dial and (dial_status or dial_call_status): | |
| url_next = request.values.get('FailUrl') | |
| if dial_call_status=='completed' or dial_status=='answered' or not url_next: | |
| response.hangup() | |
| else: | |
| response.redirect(url_next) | |
| # First pass, making initial call | |
| else: | |
| # We heard you like to pass urls in your urls, so we made sure to urlencode your... url | |
| url_voicemail = quote("http://twimlets.com/{}/vmail".format(app.config['TWILIO_ACCOUNT_SID'])) | |
| # Repeat with Dial flag set in order to trigger the proper voicemail or hangup | |
| url_after_call = "{}?{}".format( url_for('.callme', _external=True), | |
| urlencode({'Dial':'true','FailUrl':url_voicemail})) | |
| # Do a whisper on "my" end to prevent unintended pickup by greedy voicemail say of powered-off mobile phone | |
| url_upon_pickup="http://twimlets.com/whisper?HumanCheck&Message=Please+press+1+to+connect+incoming+caller" | |
| to_dial = app.designated_voice.number() | |
| with response.dial(action=url_after_call,timeout="20") as d: | |
| if app.designated_voice.is_pstn() : d.number(to_dial, url=url_upon_pickup) | |
| elif app.designated_voice.is_sip() : d.sip (to_dial, url=url_upon_pickup) | |
| elif app.designated_voice.is_client(): d.client(to_dial, url=url_upon_pickup) | |
| return str(response) | |
| # ===================== /Callme | |
| # Conference ================== | |
| @app.route('/conference/create', methods=['GET','POST']) | |
| def conference_create(): | |
| if request.values.get('AccountSid') != app.config['TWILIO_ACCOUNT_SID']: | |
| return str('<?xml version="1.0" encoding="UTF-8"?><Response />') | |
| url_domain = url_for('.index',_external=True) | |
| response = '<?xml version="1.0" encoding="UTF-8"?><Response><Dial><Conference beep="true" waitUrl="{}/static/assets/ClockworkWaltz.mp3" startConferenceOnEnter="false" endConferenceOnExit="true">MinConf</Conference></Dial></Response>'.format(url_domain) | |
| return response | |
| @app.route('/conference/join', methods=['GET','POST']) | |
| def conference_join(): | |
| if request.values.get('AccountSid') != app.config['TWILIO_ACCOUNT_SID']: | |
| return str('<?xml version="1.0" encoding="UTF-8"?><Response />') | |
| response = '<?xml version="1.0" encoding="UTF-8"?><Response><Dial><Conference beep="false" waitUrl="" startConferenceOnEnter="true" endConferenceOnExit="true">MinConf</Conference></Dial></Response>' | |
| return response | |
| # ================= /Conference | |
| # Menu ======================== | |
| @app.route('/menu', methods=['GET','POST']) | |
| def menu(): | |
| response = twiml.Response() | |
| if request.values.get('AccountSid') != app.config['TWILIO_ACCOUNT_SID']: | |
| return str(response) | |
| digits = request.values.get('Digits') | |
| if digits: | |
| url_submenu = request.values.get('Options[' + digits + ']') | |
| if url_submenu: | |
| response.redirect(url_submenu) | |
| else: | |
| response.say("I'm sorry, that is not a valid option.") | |
| response.hangup() | |
| else: | |
| max_digits = 1 | |
| # for gathering multiple digits, uncomment the following | |
| #keycodes = re.findall(r'Options\[(\S+)\]',str(request.values)) | |
| #for keycode in keycodes: | |
| # max_digits = max(max_digits, len(keycode)) | |
| with response.gather(numDigits=max_digits) as g: | |
| message = request.values.get('Message') | |
| if ( re.match(r'^http*',message) ): g.play(message) | |
| elif ( len(message) > 0 ): g.say( message) | |
| response.say("You did not make a selection. Good-bye.") | |
| response.hangup() | |
| return str(response) | |
| # ======================= /Menu | |
| # Installation success page | |
| @app.route('/') | |
| def index(): | |
| params = { | |
| 'Voice Request URL': url_for('.voice', _external=True), | |
| 'SMS Request URL': url_for('.sms', _external=True),} | |
| return render_template('index.html', params=params, | |
| configuration_error=None) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment