Python twilio.rest 模块，TwilioRestClient() 实例源码

def send_sms(self, receiver, text_body, sender, log_failure):
        """Asynchronous call to Twilio.

        We execute the actual SMS sending (calling Twilio HTTP API) in the Celery worker process, so that we do not block HTTP response head. This is especially important if we send more than one SMS message per HTTP request.
        """
        registry = self.request.registry

        account = registry.settings.get("sms.twilio_account")
        token = registry.settings.get("sms.twilio_token")

        if (not account) or (not token):
            raise RuntimeError("Missing sms.twilio_account and sms.twilio_token settings")

        logger.info("Sending Twilio SMS to: %s, body: %s", receiver, text_body)

        if account and token:
            try:
                client = TwilioRestClient(account, token)
                client.messages.create(to=receiver, from_=sender, body=text_body)
            except Exception as e:
                if log_failure:
                    logger.error("Could not send SMS from %s to %s, content %s", sender, receiver, text_body)
                    logger.exception(e)
                else:
                    raise

def __init__(
            self,
            sid,
            auth_token,
            twilio_number,
            ring_time=30,
            twiml='message',
            **kwargs
            ):

        if _has_twilio is False:
            raise ImportError('The twilio package is required for this Notifier')

        self.client = TwilioRestClient(sid, auth_token)
        self.twilio_number = twilio_number
        self.ring_time = ring_time
        self.twiml = 'message'

        super().__init__(**kwargs)

def post(self, request, *args, **kwargs):
        """Post response."""
        self.object = None
        self.form = self.get_form(self.form_class)

        # Here ou may consider creating a new instance of form_class(),
        # so that the form will come clean.
        text = self.form.save()
        text.sender = 'you'
        text.contact = Contact.objects.get(pk=int(self.kwargs.get('pk')))
        account_sid = os.environ["ACCOUNT_SID"]
        auth_token = os.environ["AUTH_TOKEN"]
        twilio_number = os.environ["TWILIO_NUMBER"]
        client = TwilioRestClient(account_sid, auth_token)
        client.messages.create(
            to=str(text.contact.number),
            from_=twilio_number,
            body=text.body
        )
        text.save()
        return self.get(request, *args, **kwargs)

def sms(name, message):
    name = name.lower()

    accountSID = config['SMS']['accountSID']
    authToken = config['SMS']['authToken']
    twilioCli = TwilioRestClient(accountSID, authToken)
    myTwilioNumber = config['SMS']['myTwilioNumber']

    success = "Message sent to "
    numbers = {}
    for key in config['numbers']:
        num = config['numbers'][key]
        numbers.update({key:num})

    print("SENDING MESSAGE")
    number = numbers[name]
    success += name
    message = twilioCli.messages.create(body=message, from_=myTwilioNumber, to=number)
    return success

def send_sms():
    """Sends a simple SMS message."""
    to = request.args.get('to')
    if not to:
        return ('Please provide the number to message in the "to" query string'
                ' parameter.'), 400

    client = TwilioRestClient(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)
    rv = client.messages.create(
        to=to,
        from_=TWILIO_NUMBER,
        body='Hello from Twilio!')
    return str(rv)
# [END send_sms]


# [START receive_sms]

def main():
    # Initialize Twilio Client
    twilio_client = TwilioRestClient(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)

    # Open JSON files
    try:
        with open('db.json') as fp:
            db = json.load(fp)
    except IOError:  # File doesn't exist
        db = {}

    try:
        with open('user.json') as fp:
            users = json.load(fp)
    except IOError:  # File doesn't exist
        users = {}

    for user_name, user_data in users.iteritems():
        process_user(user_name, user_data, db, twilio_client)

def send_email_link(phone_number):
    client = TwilioRestClient(ACCOUNT_SID, AUTH_TOKEN)
    orig_get_cert_file = base.get_cert_file
    try:
        # We monkey patch the cert file to not use anything
        base.get_cert_file = lambda: None
        logging.info("Sending SMS to %s", phone_number)
        client.messages.create(
            to=phone_number,
            from_="+12566932623",
            body="Download the DanceDeets App at https://www.dancedeets.com/mobile_apps?action=download",
        )
    except TwilioRestException as e:
        if 'not a valid phone number' in e.msg:
            raise InvalidPhoneNumberException(e.msg)
        else:
            raise
    finally:
        base.get_cert_file = orig_get_cert_file

def contact(intent, session):
    card_title = intent['name']
    session_attributes = {}
    reprompt_text = None
    account_sid = "AC39d9722bca359c3885bd8e876492859d"
    auth_token  = "222f8028aa78ffbdecbb558badf6db93"
    client = TwilioRestClient(account_sid, auth_token)
    speech_output = "Something went wrong, sorry."
    should_end_session = True
    try:
        message = client.messages.create(body="Save me it's an emergency #mhacks", 
        to="+16478367351",    # Replace with your phone number
            from_="+15675100423") # Replace with your Twilio number
        print(message.sid)
        speech_output = "I contacted your physician"
    except twilio.TwilioRestException as e:
        print(e)
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))

def _validate_configure_twilio(self):
        """configure twilio client

        we will provide to ways to configure clients :
         - One, you can configure twilio keys from environment variables if not,
         - Then Second, you ca send keys as function arguments too,
         - Priority wil be given to function arguments

        :return: twilio client instance
        """

        if self.account_sid is None:
            self.account_sid = getattr(settings, 'TWILIO_ACCOUNT_SID', None)
        if self.auth_token is None:
            self.auth_token = getattr(settings, 'TWILIO_AUTH_TOKEN', None)

        if self.account_sid is None or self.auth_token is None:
            raise RuntimeWarning(
                'to send sms via {0} you need to configure TWILIO_ACCOUNT_SID & TWILIO_AUTH_TOKEN in \n'
                'environment variables or send account_sid & auth_token as function arguments.'.format(self.name)
            )

        self.twilio_client = TwilioRestClient(self.account_sid, self.auth_token)

def send_sms(alert=None, recepients=None, template=None):
    details = sms_model.get()

    try:
        client = TwilioRestClient(details['account'], details['token'])
    except:
        client = None


    if client != None:
        body = render_template(alert=alert, template=template)

        for recepient in recepients:
            t = threading.Thread(target=_send_sms_in_thread, 
                kwargs={"client": client,
                    "from_": details['from_'],
                    "to": recepient['phone'],
                    "body": body
                }
            )
            t.start()

def list(self, dateSent):
        if self.isMock:
            twilioMessages = self.mockMessages
        elif self.twilioRestClient is not None:
            twilioMessages = self.twilioRestClient.messages.list(date_sent=dateSent)
        else:
            raise ValueError('TwilioRestClient not installed, unable to make sms calls')

        return twilioMessages

#     def __mock_messages(self):
#        messages = []
#        messages.append(MockMessage({'sid': '1-close', 'body': 'Close', 'status': 'received', 'from_': '+15551113333', 'to': '+15552229999', 'direction': 'outbound-api', 'date_created': datetime.datetime.now() - datetime.timedelta(minutes=15), 'date_sent': datetime.datetime.now() - datetime.timedelta(minutes=15)}))
#        messages.append(MockMessage({'sid': '2-lock', 'body': 'Lock', 'status': 'received', 'from_': '+15551113333', 'to': '+15552229999', 'direction': 'outbound-api', 'date_created': datetime.datetime.now() - datetime.timedelta(minutes=12), 'date_sent': datetime.datetime.now() - datetime.timedelta(minutes=12)}))
#        messages.append(MockMessage({'sid': '22-lock', 'body': 'Lock', 'status': 'received', 'from_': '+15551113333', 'to': '+15552229999', 'direction': 'outbound-api', 'date_created': datetime.datetime.now() - datetime.timedelta(minutes=12), 'date_sent': datetime.datetime.now() - datetime.timedelta(minutes=12)}))
#        messages.append(MockMessage({'sid': '3-open', 'body': 'Open', 'status': 'received', 'from_': '+15551113333', 'to': '+15552229999', 'direction': 'outbound-api', 'date_created': datetime.datetime.now() - datetime.timedelta(minutes=11), 'date_sent': datetime.datetime.now() - datetime.timedelta(minutes=11)}))
#        messages.append(MockMessage({'sid': '4-unlock', 'body': 'Unlock', 'status': 'received', 'from_': '+15551113333', 'to': '+15552229999', 'direction': 'outbound-api', 'date_created': datetime.datetime.now() - datetime.timedelta(minutes=10), 'date_sent': datetime.datetime.now() - datetime.timedelta(minutes=10)}))
#        messages.append(MockMessage({'sid': '5-open', 'body': 'Open', 'status': 'received', 'from_': '+15551113333', 'to': '+15552229999', 'direction': 'outbound-api', 'date_created': datetime.datetime.now() - datetime.timedelta(minutes=9), 'date_sent': datetime.datetime.now() - datetime.timedelta(minutes=9)}))
#        messages.append(MockMessage({'sid': '6-ignore-me', 'body': 'This is just an arbitrary text', 'status': 'received', 'from_': '+15551113333', 'to': '+15552229999', 'direction': 'outbound-api', 'date_created': datetime.datetime.now() - datetime.timedelta(minutes=9), 'date_sent': datetime.datetime.now() - datetime.timedelta(minutes=9)}))
#
#        for message in messages:
#            logging.info("Mock Message: {}".format(vars(message)))
#
#        return messages

def index(request):
    todos = Todo.objects.all()[:10]
    if(request.method == 'POST'):
        account = "ACbadfc40befbac0c3827fc116ab0fc0b8"
        token = "0fa9b25fb15e3009bec7fe6bcf786aa9"
        client = TwilioRestClient(account, token)
        title = request.POST['title']
        text = request.POST['text']
        hour = request.POST['hour']
        minute  = request.POST['minute']
        # while(1):
        #       c=datetime.now()
        #       print c.hour , c.minute 
        #       print hour , minute
        #       if(int(hour)==c.hour and int(minute) ==c.minute):
        #           message = client.sms.messages.create(to="+919176409201",
     #                                     from_="+1 559-512-7609 ",
     #                                     body="master you have to complete"+title+':'+text)
        #           break


    context = {
    'todos' : todos
    }
    return render(request, 'index.html',context)

def startJoinMeetingAction(startJoinRoomIdVal):
    getRoomDetailsResponse = sparkApi.get('rooms', {'roomId': startJoinRoomIdVal, 'showSipAddress': 'true'})
    if getRoomDetailsResponse != 'Error':
        roomSipVal = getRoomDetailsResponse['sipAddress']
        try:
            client = TwilioRestClient(main.twilio_AccountSid, main.twilio_AuthToken)
            call = client.calls.create(url=main.twilioXmlPath + '/' + roomSipVal, to=main.cellPhoneE164, from_=main.twilioNumber)
            callStatus = client.calls.get(call.sid).status
            if callStatus != 'failed':
                speechOutput = "Calling your cellphone now"
            else:
                speechOutput = "There is a problem connecting you to the Spark room. Please try again in a few minutes."
        except TwilioRestException as e:
            speechOutput = "There is a problem connecting to Twilio. Please try again in a few minutes."

    else:
        speechOutput = "There is a problem connecting to Cisco Spark. Please try again in a few minutes."
    return speechOutput

def Retrieving():
    client = TwilioRestClient(ACCOUNT_SID, AUTH_TOKEN)
    for message in client.messages.list():
        Historico = (message.body)
        print ("Historico de mensagens Enviadas > ", Historico )

def _send(recipients, text_content=None, html_content=None, sent_from=None, subject=None, extra_data=None,
              attachments=None):
        try:
            # twilio version 6
            from twilio.rest import Client
        except ImportError:
            try:
                # twillio version < 6
                from twilio.rest import TwilioRestClient as Client
            except ImportError:
                raise Exception('Twilio is required for sending a TwilioTextNotification.')

        try:
            account_sid = settings.TWILIO_ACCOUNT_SID
            auth_token = settings.TWILIO_AUTH_TOKEN
        except AttributeError:
            raise Exception(
                'TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN settings are required for sending a TwilioTextNotification'
            )

        client = Client(account_sid, auth_token)

        client.messages.create(
            body=text_content,
            to=recipients[0],
            from_=sent_from
        )

def connect(self):
        self.__client = TwilioRestClient(self.__account_sid, self.__auth_token)

    # Send a message letting the channel know that this alarm started

def test_creds_error(creds):
    creds.return_value = (None, None)
    assert_raises(TwilioException, TwilioRestClient)

def setUp(self):
        self.client = TwilioRestClient("ACCOUNT_SID", "AUTH_TOKEN")
        self.task_router_client = TwilioTaskRouterClient("ACCOUNT_SID",
                                                         "AUTH_TOKEN")

def setUp(self):
        self.client = TwilioRestClient("ACCOUNT_SID", "AUTH_TOKEN",
                                       timeout=sentinel.timeout)

def send_sms(contact, content):
    """Send sms via twilio api."""
    account_sid = os.environ["TWILIO_SID"]
    auth_token = os.environ["TWILIO_TOKEN"]
    twilio_number = os.environ["TWILIO_NUMBER"]
    client = TwilioRestClient(account_sid, auth_token)
    client.messages.create(
        to='+1' + contact.phone,
        from_=twilio_number,
        body=content
    )

def setUp(self):
        """Setup for tests."""
        self.client = Client()
        self.request = RequestFactory()
        self.contact = ContactFactory.create()
        self.twilio_client = TwilioRestClient(
            settings.TEST_ACCOUNT_SID,
            settings.TEST_AUTH_TOKEN)

# ----------------------------- BASE PAGE --------------------------------

def send_sms(message, number):
    logger.info('Sending sms to: ' + number)

    details = dict(
        to=str(number),
        from_=settings.TWILIO_FROM_NUMBER,
        body=str(message),
    )

    if os.environ.get('DEBUG', True) not in ('True', 'true', True,):
        client = TwilioRestClient(settings.TWILIO_SID, 
            settings.TWILIO_TOKEN)
        client.messages.create(**details)
    else:
        logger.info(details)

def _get_twilio_client(cls):
        """
        A singleton method for getting a twilio rest client.

        Returns:
            TwilioRestClient: The twilio client we'll use for sending text
            messages.
        """
        if not cls.twilio_client:
            account_sid = os.environ["TWILIO_ACCOUNT_SID"]
            auth_token = os.environ["TWILIO_AUTH_TOKEN"]
            cls.twilio_client = TwilioRestClient(account_sid, auth_token)

        return cls.twilio_client

def __init__(self):
        try:
            number, account_sid, auth_token = utils.load_twilio_config()
            self.number = number
            self.twilio_client = TwilioRestClient(account_sid, auth_token)
        except TwilioException:
            print('\U0001F6AB  Current twilio credentials invalid. '
                  'Please reset.')
            utils.initialize_twiolio()
            number, account_sid, auth_token = utils.load_twilio_config()
            self.number = number
            self.twilio_client = TwilioRestClient(account_sid, auth_token)

def authenticate(errorFile):

    try:
        keyFile = open("/home/pi/frosti/alertSrc/keys.txt",'r')
    except IOError:
        date = datetime.datetime.now()
        errorFile.write(date.isoformat(" ") + ": Could not find key file\n")
        return -1

    date = datetime.datetime.now()
    errorFile.write(date.isoformat(" ") + ": Alert module called\n")

    #read keys from key file and strip newlines (hence [:-1])
    test_sid   = keyFile.readline()[:-1]
    test_token = keyFile.readline()[:-1]

    prod_sid   = keyFile.readline()[:-1]
    prod_token = keyFile.readline()[:-1]

    #select keys to use (test or production)
    sid   = prod_sid
    token = prod_token

    try:
        client = TwilioRestClient(sid, token)
        return sid, token, client

    except TwilioException as e:
        date = datetime.datetime.now()
        errorFile.write(date.isoformat(" ") + ": Could not register Twilio client\n")
        return -1

def textmyself(message):
    twilioClient=TwilioRestClient(sid,token)
    twilioClient.messages.create(body=message,from_=twilioNumber,to=myNumber)

# textmyself("Hello There")  will send "Hello There" message on my mobile number

def sendsms_task():
    client = TwilioRestClient(account_sid, auth_token)

    msg = 'Toast is DONE!'
    mJson = request.get_json()
    if mJson is not None and 'message' in mJson:
        msg = mJson['message']
    for pn in phoneNumbers:

        message = client.messages.create(body=msg,
            to='+'+str(pn['phoneNumber']),
            from_="+16783355213")

    return jsonify({'task': 'sms sent'}), 201

def __init__(self, account_sid, auth_token):
        self.twilio = TwilioRestClient(account_sid, auth_token)

def _make_client(configuration):
    account_sid = configuration['twilio']['account_sid']
    auth_token = configuration['twilio']['auth_token']
    number = configuration['twilio']['number']
    client = TwilioRestClient(account_sid, auth_token)
    client.from_ = number
    return client


# *****************************************************
# SERVICE METHODS
# *****************************************************

def _sendText(message):
    account_sid = "TWILIO SID"
    auth_token = "TWILIO AUTH_TOKEN"
    client = TwilioRestClient(account_sid, auth_token)
    message = client.messages.create(to='+RECV NUMBER', from_='+TWILIO NUMBER', body=message)

def check_and_run() :
    imap_conn = imapclient.IMAPClient (domain_name, ssl = True)
    print (imap_conn)
    try :
        print (imap_conn.login (emailaddress, password))
    except :
        print ('Wrong username password combination! Please try again.')
        sys.exit()
    print (imap_conn.select_folder ('INBOX', readonly = False))
    uids = imap_conn.gmail_search ('label:' + label + ' label:Inbox ' + emailaddress + ' subject:' + mysubject + ' label:unread')
    print (uids)
    messages = imap_conn.fetch (uids, ['BODY[]'])
    for x in messages.keys() :
        lstcomm = pyzmail.PyzMessage.factory (messages[x][b'BODY[]'])
        commands.append (lstcomm.text_part.get_payload().decode (lstcomm.text_part.charset))   
    print (imap_conn.logout())
    twilioClient = TwilioRestClient (twilioSID, twilioAuthToken)
    for x in commands[::-1] :
        newfile = open ('commandstorun.py', 'w')
        newfile.write ('#! /usr/bin/python3\n')
        newfile.write (x)
        newfile.close()
        os.system ('chmod +x commandstorun.py')
        os.system ('./commandstorun.py')
        print ('Executed script :\n' + x)
        msg = twilioClient.messages.create (body = 'Your script has been executed! ' + x, from_ = sender, to = receiver)        
        print (msg.status)

def textmyself(message):
    twilioCli = TwilioRestClient(accountSID, authToken)
    twilioCli.messages.create(body=message, from_=twilioNumber, to=myNumber)

def test_creds_error(creds):
    creds.return_value = (None, None)
    assert_raises(TwilioException, TwilioRestClient)

def setUp(self):
        self.client = TwilioRestClient("ACCOUNT_SID", "AUTH_TOKEN")
        self.task_router_client = TwilioTaskRouterClient("ACCOUNT_SID",
                                                         "AUTH_TOKEN")

def setUp(self):
        self.client = TwilioRestClient("ACCOUNT_SID", "AUTH_TOKEN",
                                       timeout=sentinel.timeout)

def getClient(self):
        if self.smsclient:
            return self.smsclient
        assert 'twilio_sid' in self.settings
        self.smsclient = TwilioRestClient(self.settings['twilio_sid'],
                                          self.settings['twilio_token'])
        return self.smsclient

def __init__(self):
        self.map = {}
        self.brd = {}
        exec(open('./twilio.json').read())
        print
        sid = x['twilio']['account_sid']
        sec = x['twilio']['secret']

        self.twilio = TwilioRestClient(sid, sec)

def _get_twilio_client():
    """Return twilio rest client with variables"""
    return TwilioRestClient(
        current_app.config.get("TWILIO_ACCOUNT_SID"),
        current_app.config.get("TWILIO_AUTH_TOKEN"))

def alert(self, matches):
        client = TwilioRestClient(self.twilio_accout_sid, self.twilio_auth_token)

        try:
            client.messages.create(body=self.rule['name'],
                                   to=self.twilio_to_number,
                                   from_=self.twilio_to_number)

        except TwilioRestException as e:
            raise EAException("Error posting to twilio: %s" % e)

        elastalert_logger.info("Trigger sent to Twilio")

def send_phone_code(user_id, verify_token, phone_number):
    customer = Customer.objects.get(id=user_id)

    # simulating a short delay
    sleep(randint(3, 9))

    client = TwilioRestClient(  # noqa
        settings.TWILLO_API_KEY,
        settings.TWILLO_AUTH_TOKEN,
    )

    try:
        client.messages.create(
            to=phone_number,
            from_=settings.TWILLO_FROM_NUMBER,
            body="Your Verify Code is {}".format(
                verify_token
            )
        )
        # Notify the FE task has completed
        Group('phone_verify-%s' % customer.username).send({
            'text': json.dumps({
                'success': True,
                'msg': 'new message sent'
            })
        })
    except TwilioRestException as e:
        Group('phone_verify-%s' % customer.username).send({
            'text': json.dumps({
                'success': False,
                'msg': e.msg
            })
        })

def send_message(request, source, destination, menu_text):
    client = TwilioRestClient(ACCOUNT_SID, AUTH_TOKEN) 

    client.messages.create(
        to=destination, 
        from_=source, 
        body=menu_text,  
    )
    request.session['last_message'] = menu_text

def textmyself(self, message):
        twilioCli = TwilioRestClient(self.accountsid, self.authtoken)
        twilioCli.messages.create(body=message, from_=self.twiliocell, to=self.mycellphone)
        print("successfully sent the text")

def textmyself(message):
    twilioCli = TwilioRestClient(accountSID, authToken)
    twilioCli.messages.create(body = message, from_ = twilioNumber, to = myNumber)

def sendMessage(mediaURL):
    client = TwilioRestClient(ACCOUNT_SID, AUTH_TOKEN) 
    client.messages.create(
        to="+14254298436", 
        from_="+19284874422 ", 
        media_url=mediaURL, 
    )    
    return

def get_twilio_client(self):
        return TwilioRestClient(self.config['account_sid'],
                                self.config['auth_token'])

def test_creds_error(creds):
    creds.return_value = (None, None)
    assert_raises(TwilioException, TwilioRestClient)

def setUp(self):
        self.client = TwilioRestClient("ACCOUNT_SID", "AUTH_TOKEN")
        self.task_router_client = TwilioTaskRouterClient("ACCOUNT_SID",
                                                         "AUTH_TOKEN")

def setUp(self):
        self.client = TwilioRestClient("ACCOUNT_SID", "AUTH_TOKEN",
                                       timeout=sentinel.timeout)

def send_test_sms(recepient=None):
    details = sms_model.get()
    client = TwilioRestClient(details['account'], details['token'])

    t = threading.Thread(target=_send_sms_in_thread, 
                kwargs={"client": client,
                    "from_": details['from_'],
                    "to": recepient,
                    "body": "Amon alert!"
                })
    t.start()