import datetime
import enum
import json
from typing import List, Dict, Optional
import requests
from twilio.rest import Client
from twilio.base import values
from platypush.message import Message
from platypush.plugins import Plugin, action
[docs]
class TwilioPhoneNumberType(enum.Enum):
MOBILE = 'mobile'
LOCAL = 'local'
TOLL_FREE = 'toll_free'
[docs]
class TwilioPlugin(Plugin):
"""
The Twilio plugin allows you to send messages and WhatsApp texts and make
programmable phone call by using a Twilio account. Note that some features
may require a Premium account.
"""
_api_base_url = 'https://api.twilio.com'
[docs]
def __init__(
self,
account_sid: str,
auth_token: str,
address_sid: Optional[str] = None,
phone_number: Optional[str] = None,
address_book: Optional[Dict[str, str]] = None,
**kwargs
):
"""
:param account_sid: Account SID.
:param auth_token: Account authentication token.
:param address_sid: SID of the default physical address - required to
register a new number in some countries.
:param phone_number: Default phone number associated to the account to
be used for messages and calls.
:param address_book: ``name``->``phone_number`` mapping of contacts.
You can use directly these names to send messages and make calls
instead of the full phone number.
"""
super().__init__(**kwargs)
self.account_sid = account_sid
self.auth_token = auth_token
self.address_sid = address_sid
self.phone_number = phone_number
self.address_book = address_book or {}
self.client = Client(account_sid, auth_token)
[docs]
@action
def get_available_phone_numbers(self, country: str, number_type: str) -> List[dict]:
"""
Get a list of phone numbers of a certain type available for a certain country.
:param country: Country code (e.g. ``US`` or ``NL``).
:param number_type: Phone number type - e.g. ``mobile``, ``local`` or
``toll_free``.
:return: A list of the available phone numbers with their properties
and capabilities. Example:
.. code-block:: json
[
{
"friendly_name": "+311234567890",
"phone_number": "+311234567890",
"lata": null,
"rate_center": null,
"latitude": null,
"longitude": null,
"locality": null,
"region": null,
"postal_code": null,
"iso_country": "NL",
"address_requirements": "any",
"beta": false,
"capabilities": {
"voice": true,
"SMS": true,
"MMS": false,
"fax": false
}
}
]
"""
phone_numbers = self.client.available_phone_numbers(country.upper()).fetch()
resp = requests.get(
self._api_base_url + phone_numbers.uri,
auth=(self.account_sid, self.auth_token),
).json()
assert (
'subresource_uris' in resp
), 'No available phone numbers found for the country {}'.format(country)
assert (
number_type in resp['subresource_uris']
), 'No "{}" phone numbers available - available types: {}'.format(
number_type, list(resp['subresource_uris'].keys())
)
resp = requests.get(
self._api_base_url + resp['subresource_uris'][number_type],
auth=(self.account_sid, self.auth_token),
).json()
phone_numbers = resp['available_phone_numbers']
assert len(phone_numbers), 'No phone numbers available'
return phone_numbers
[docs]
@action
def create_address(
self,
customer_name: str,
street: str,
city: str,
region: str,
postal_code: str,
iso_country: str,
):
"""
Create a new address associated to your account.
:param customer_name: Full name of the customer.
:param street: Street name.
:param city: City name.
:param region: Region name.
:param postal_code: Postal code.
:param iso_country: ISO code of the country.
:return: Details of the newly created address. Example:
.. code-block:: json
{
"account_sid": "ACXXX",
"city": "city",
"customer_name": "customer_name",
"date_created": "Tue, 18 Aug 2015 17:07:30 +0000",
"date_updated": "Tue, 18 Aug 2015 17:07:30 +0000",
"emergency_enabled": false,
"friendly_name": null,
"iso_country": "US",
"postal_code": "postal_code",
"region": "region",
"sid": "ADXXX",
"street": "street",
"validated": false,
"verified": false,
"uri": "/2010-04-01/Accounts/ACXXX/Addresses/ADXXX.json"
}
"""
address = self.client.addresses.create(
customer_name=customer_name,
street=street,
city=city,
region=region,
postal_code=postal_code,
iso_country=iso_country,
)
return address._properties
[docs]
@action
def register_phone_number(
self,
phone_number: str,
friendly_name: Optional[str] = None,
address_sid: Optional[str] = None,
sms_url: Optional[str] = None,
sms_fallback_url: Optional[str] = None,
status_callback: Optional[str] = None,
voice_caller_id_lookup: bool = True,
voice_url: Optional[str] = None,
voice_fallback_url: Optional[str] = None,
area_code: Optional[str] = None,
) -> dict:
"""
Request to allocate a phone number on your Twilio account. The phone
number should first be displayed as available in
:meth:`.get_available_phone_numbers`.
:param phone_number: Phone number to be allocated.
:param friendly_name: A string used to identify your new phone number.
:param address_sid: Address SID. NOTE: some countries may require you
to specify a valid address in order to register a new phone number
(see meth:`create_address`). If none is specified then the
configured ``address_sid`` (if available) will be applied.
:param sms_url: URL to call when an SMS is received.
:param sms_fallback_url: URL to call when an error occurs on SMS
delivery/receipt.
:param status_callback: URL to call when a status change occurs.
:param voice_caller_id_lookup: Whether to perform ID lookup for
incoming caller numbers.
:param voice_url: URL to call when the number receives a call.
:param voice_fallback_url: URL to call when a call fails.
:param area_code: Override the area code for the new number.
:return: Status of the newly created number. Example:
.. code-block:: json
{
"account_sid": "ACXXX",
"address_requirements": "none",
"address_sid": "ADXXX",
"api_version": "2010-04-01",
"beta": false,
"capabilities": {
"voice": true,
"sms": false,
"mms": true,
"fax": false
},
"date_created": "Thu, 30 Jul 2015 23:19:04 +0000",
"date_updated": "Thu, 30 Jul 2015 23:19:04 +0000",
"emergency_status": "Active",
"emergency_address_sid": "ADXXX",
"friendly_name": "friendly_name",
"identity_sid": "RIXXX",
"origin": "origin",
"phone_number": "+18089255327",
"sid": "PNXXX",
"sms_application_sid": "APXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
"sms_fallback_method": "GET",
"sms_fallback_url": "https://example.com",
"sms_method": "GET",
"sms_url": "https://example.com",
"status_callback": "https://example.com",
"status_callback_method": "GET",
"trunk_sid": null,
"uri": "/2010-04-01/Accounts/ACXXX/IncomingPhoneNumbers/PNXXX.json",
"voice_application_sid": "APXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
"voice_caller_id_lookup": false,
"voice_fallback_method": "GET",
"voice_fallback_url": "https://example.com",
"voice_method": "GET",
"voice_url": "https://example.com",
"bundle_sid": "BUXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
"voice_receive_mode": "voice",
"status": "in-use"
}
"""
status = self.client.incoming_phone_numbers.create(
phone_number=phone_number,
friendly_name=friendly_name,
sms_url=sms_url,
sms_fallback_url=sms_fallback_url,
status_callback=status_callback,
voice_caller_id_lookup=voice_caller_id_lookup,
voice_url=voice_url,
voice_fallback_url=voice_fallback_url,
area_code=area_code,
address_sid=address_sid or self.address_sid,
)
return status._properties
[docs]
@action
def send_message(
self,
body: str,
to: str,
from_: Optional[str] = None,
status_callback: Optional[str] = None,
max_price: Optional[str] = None,
attempt: Optional[int] = None,
validity_period: Optional[int] = None,
smart_encoded: bool = True,
media_url: Optional[str] = None,
) -> dict:
"""
Send an SMS/MMS.
Note: WhatsApp messages are also supported (and free of charge),
although the functionality is currently quite limited. Full support is
only available to WhatsApp Business profiles and indipendent software
vendors approved by WhatsApp. If that's not the case, you can send
WhatsApp messages through the Twilio Test account/number - as of now
the ``from_`` field should be ``whatsapp:+14155238886`` and the ``to``
field should be ``whatsapp:+<phone_number_here>``. More information
`here <https://www.twilio.com/docs/whatsapp>`_.
:param body: Message body.
:param to: Recipient number or address book name.
:param from_: Sender number. If none is specified then the default
configured ``phone_number`` will be used if available.
:param status_callback: The URL to call to send status information to
the application.
:param max_price: The total maximum price up to 4 decimal places in US
dollars acceptable for the message to be delivered.
:param attempt: Total numer of attempts made , this inclusive to send
out the message.
:param validity_period: The number of seconds that the message can
remain in our outgoing queue.
:param smart_encoded: Whether to detect Unicode characters that have a
similar GSM-7 character and replace them.
:param media_url: The URL of the media to send with the message.
:return: A mapping representing the status of the delivery. Example:
.. code-block:: json
{
"account_sid": "ACXXX",
"api_version": "2010-04-01",
"body": "Sent from your Twilio trial account - It works!",
"date_created": "2020-08-17T16:32:09.341",
"date_updated": "2020-08-17T16:32:09.526",
"date_sent": null,
"direction": "outbound-api",
"error_code": null,
"error_message": null,
"from_": "+XXXXXXXXX",
"messaging_service_sid": null,
"num_media": "0",
"num_segments": "1",
"price": null,
"price_unit": "USD",
"sid": "XXXXXXXXXXXXX",
"status": "queued",
"subresource_uris": {
"media": "/2010-04-01/Accounts/ACXXX/Messages/SMXXX/Media.json"
},
"to": "+XXXXXXXXXXXXXX",
"uri": "/2010-04-01/Accounts/ACXXX/Messages/SMXXX.json"
}
"""
assert (
from_ or self.phone_number
), 'No valid sender phone number specified nor configured'
to = self.address_book.get(to, to)
status = self.client.messages.create(
body=body,
from_=from_ or self.phone_number,
to=to,
media_url=media_url or values.unset,
max_price=max_price or values.unset,
attempt=str(attempt) if attempt else values.unset,
status_callback=status_callback or values.unset,
validity_period=str(validity_period) if validity_period else values.unset,
smart_encoded=smart_encoded,
)
return status._properties
[docs]
@action
def list_messages(
self,
to: Optional[str] = None,
from_: Optional[str] = None,
date_sent_before: Optional[str] = None,
date_sent: Optional[str] = None,
date_sent_after: Optional[str] = None,
limit: Optional[int] = None,
page_size: Optional[int] = None,
) -> List[dict]:
"""
List all messages matching the specified criteria.
:param to: Recipient phone number or address book name.
:param from_: Sender phone number.
:param date_sent_before: Maximum date filter (ISO format: YYYYMMDD with or without time).
:param date_sent: Date filter (ISO format: YYYYMMDD with or without time).
:param date_sent_after: Minimum date filter (ISO format: YYYYMMDD with or without time).
:param limit: Maximum number of messages to be returned.
:param page_size: Maximum number of messages per page.
:return: List of selected messages. Example:
.. code-block:: json
{
"account_sid": "ACXXX",
"api_version": "2010-04-01",
"body": "testing",
"date_created": "Fri, 24 May 2019 17:44:46 +0000",
"date_sent": "Fri, 24 May 2019 17:44:50 +0000",
"date_updated": "Fri, 24 May 2019 17:44:50 +0000",
"direction": "outbound-api",
"error_code": null,
"error_message": null,
"from": "+12019235161",
"messaging_service_sid": null,
"num_media": "0",
"num_segments": "1",
"price": "-0.00750",
"price_unit": "USD",
"sid": "SMded05904ccb347238880ca9264e8fe1c",
"status": "sent",
"subresource_uris": {
"media": "/2010-04-01/Accounts/ACXXX/Messages/SMXXX/Media.json",
"feedback": "/2010-04-01/Accounts/ACXXX/Messages/SMXXX/Feedback.json"
},
"to": "+18182008801",
"uri": "/2010-04-01/Accounts/ACXXX/Messages/SMXXX.json"
}
"""
if to:
to = self.address_book.get(to, to)
messages = self.client.messages.list(
to=to,
from_=from_,
date_sent_before=datetime.datetime.fromisoformat(date_sent_before)
if date_sent_before
else None,
date_sent_after=datetime.datetime.fromisoformat(date_sent_after)
if date_sent_after
else None,
date_sent=datetime.datetime.fromisoformat(date_sent) if date_sent else None,
limit=limit,
page_size=page_size,
)
return json.loads(
json.dumps(
[msg._properties for msg in messages], indent=2, cls=Message.Encoder
)
)
[docs]
@action
def get_message(self, message_sid: str) -> dict:
"""
Get the details of a stored message.
:param message_sid: Message SID to be retrieved.
:return: Message with its properties - see :meth:`.send_message`.
"""
msg = self.client.messages(message_sid).fetch()
return msg._properties
[docs]
@action
def update_message(self, message_sid: str, body: str) -> dict:
"""
Update/redact the body of a message.
:param message_sid: Message SID to be updated.
:param body: New message body.
:return: Updated message with its properties - see :meth:`.send_message`.
"""
msg = self.client.messages(message_sid).update(body)
return msg._properties
[docs]
@action
def delete_message(self, message_sid: str):
"""
Delete a send/received message.
:param message_sid: Message SID to be deleted.
"""
self.client.messages(message_sid).delete()
[docs]
@action
def make_call(
self,
twiml: str,
to: str,
from_: Optional[str] = None,
method: Optional[str] = None,
status_callback: Optional[str] = None,
status_callback_event: Optional[str] = None,
status_callback_method: Optional[str] = None,
fallback_url: Optional[str] = None,
fallback_method: Optional[str] = None,
send_digits: Optional[str] = None,
timeout: Optional[int] = 30,
record: bool = False,
recording_channels: Optional[int] = None,
recording_status_callback: Optional[str] = None,
recording_status_callback_method: Optional[str] = None,
recording_status_callback_event: Optional[str] = None,
sip_auth_username: Optional[str] = None,
sip_auth_password: Optional[str] = None,
caller_id: Optional[str] = None,
call_reason: Optional[str] = None,
) -> dict:
"""
Make an automated phone call from a registered Twilio number.
:param twiml: TwiML containing the logic to be executed in the call
(see https://www.twilio.com/docs/voice/twiml).
:param to: Recipient phone number or address book name.
:param from_: Registered Twilio phone number that will perform the call
(default: default configured phone number).
:param method: HTTP method to use to fetch TwiML if it's provided remotely.
:param status_callback: The URL that should be called to send status
information to your application.
:param status_callback_event: The call progress events to be sent to
the ``status_callback`` URL.
:param status_callback_method: HTTP Method to use with status_callback.
:param fallback_url: Fallback URL in case of error.
:param fallback_method: HTTP Method to use with fallback_url.
:param send_digits: The digits to dial after connecting to the number.
:param timeout: Number of seconds to wait for an answer.
:param record: Whether to record the call.
:param recording_channels: The number of channels in the final recording.
:param recording_status_callback: The URL that we call when the
recording is available to be accessed.
:param recording_status_callback_method: The HTTP method to use when
calling the `recording_status_callback` URL.
:param recording_status_callback_event: The recording status events
that will trigger calls to the URL specified in
`recording_status_callback`
:param sip_auth_username: The username used to authenticate the caller
making a SIP call.
:param sip_auth_password: The password required to authenticate the
user account specified in `sip_auth_username`.
:param caller_id: The phone number, SIP address, or Client identifier
that made this call. Phone numbers are in E.164 format (e.g.,
+16175551212). SIP addresses are formatted as `name@company.com`.
:param call_reason: Reason for the call (Branded Calls Beta).
:return: The call properties and details, as a dictionary. Example:
.. code-block:: json
{
"account_sid": "ACXXX",
"annotation": null,
"answered_by": null,
"api_version": "2010-04-01",
"caller_name": null,
"date_created": "Tue, 31 Aug 2010 20:36:28 +0000",
"date_updated": "Tue, 31 Aug 2010 20:36:44 +0000",
"direction": "inbound",
"duration": "15",
"end_time": "Tue, 31 Aug 2010 20:36:44 +0000",
"forwarded_from": "+141586753093",
"from": "+15017122661",
"from_formatted": "(501) 712-2661",
"group_sid": null,
"parent_call_sid": null,
"phone_number_sid": "PNXXX",
"price": "-0.03000",
"price_unit": "USD",
"sid": "CAXXX",
"start_time": "Tue, 31 Aug 2010 20:36:29 +0000",
"status": "completed",
"subresource_uris": {
"notifications": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Notifications.json",
"recordings": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Recordings.json",
"feedback": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Feedback.json",
"feedback_summaries": "/2010-04-01/Accounts/ACXXX/Calls/FeedbackSummary.json",
"payments": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Payments.json"
},
"to": "+14155551212",
"to_formatted": "(415) 555-1212",
"trunk_sid": null,
"uri": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX.json",
"queue_time": "1000"
}
"""
to = self.address_book.get(to, to)
call = self.client.calls.create(
to=to,
from_=from_ or self.phone_number,
twiml=twiml,
method=method,
status_callback=status_callback,
status_callback_event=status_callback_event,
status_callback_method=status_callback_method,
fallback_url=fallback_url,
fallback_method=fallback_method,
send_digits=send_digits,
timeout=str(timeout) if timeout else None,
record=record,
recording_channels=str(recording_channels) if recording_channels else None,
recording_status_callback=recording_status_callback,
recording_status_callback_method=recording_status_callback_method,
recording_status_callback_event=recording_status_callback_event,
sip_auth_username=sip_auth_username,
sip_auth_password=sip_auth_password,
caller_id=caller_id,
call_reason=call_reason,
)
return call._properties
[docs]
@action
def list_calls(
self,
to: Optional[str] = None,
from_: Optional[str] = None,
parent_call_sid: Optional[str] = None,
status: Optional[str] = None,
start_time_before: Optional[str] = None,
start_time: Optional[str] = None,
start_time_after: Optional[str] = None,
end_time_before: Optional[str] = None,
end_time: Optional[str] = None,
end_time_after: Optional[str] = None,
limit: Optional[int] = None,
page_size: Optional[int] = None,
) -> List[dict]:
"""
List the calls performed by the account, either the full list or those
that match some filter.
:param to: Phone number or Client identifier of calls to include
:param from_: Phone number or Client identifier to filter `from` on
:param parent_call_sid: Parent call SID to filter on
:param status: The status of the resources to read
:param start_time_before: Only include calls that started on this date
:param start_time: Only include calls that started on this date
:param start_time_after: Only include calls that started on this date
:param end_time_before: Only include calls that ended on this date
:param end_time: Only include calls that ended on this date
:param end_time_after: Only include calls that ended on this date
:param limit: Upper limit for the number of records to return. list() guarantees
never to return more than limit. Default is no limit
:param page_size: Number of records to fetch per request, when not set will use
the default value of 50 records. If no page_size is defined
but a limit is defined, list() will attempt to read the limit
with the most efficient page size, i.e. min(limit, 1000)
:return: A list of dictionaries, each representing the information of a
call. Example:
.. code-block:: json
[
{
"account_sid": "ACXXX",
"annotation": "billingreferencetag1",
"answered_by": "machine_start",
"api_version": "2010-04-01",
"caller_name": "callerid1",
"date_created": "Fri, 18 Oct 2019 17:00:00 +0000",
"date_updated": "Fri, 18 Oct 2019 17:01:00 +0000",
"direction": "outbound-api",
"duration": "4",
"end_time": "Fri, 18 Oct 2019 17:03:00 +0000",
"forwarded_from": "calledvia1",
"from": "+13051416799",
"from_formatted": "(305) 141-6799",
"group_sid": "GPXXX",
"parent_call_sid": "CAXXX",
"phone_number_sid": "PNXXX",
"price": "-0.200",
"price_unit": "USD",
"sid": "CAXXX",
"start_time": "Fri, 18 Oct 2019 17:02:00 +0000",
"status": "completed",
"subresource_uris": {
"feedback": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Feedback.json",
"feedback_summaries": "/2010-04-01/Accounts/ACXXX/Calls/FeedbackSummary.json",
"notifications": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Notifications.json",
"recordings": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Recordings.json",
"payments": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX/Payments.json"
},
"to": "+13051913581",
"to_formatted": "(305) 191-3581",
"trunk_sid": "TKXXX",
"uri": "/2010-04-01/Accounts/ACXXX/Calls/CAXXX.json",
"queue_time": "1000"
}
]
"""
call_list = self.client.calls.list(
to=to,
from_=from_,
parent_call_sid=parent_call_sid,
status=status,
start_time_before=datetime.datetime.fromisoformat(start_time_before)
if start_time_before
else None,
start_time=datetime.datetime.fromisoformat(start_time)
if start_time
else None,
start_time_after=datetime.datetime.fromisoformat(start_time_after)
if start_time_after
else None,
end_time_before=datetime.datetime.fromisoformat(end_time_before)
if end_time_before
else None,
end_time=datetime.datetime.fromisoformat(end_time) if end_time else None,
end_time_after=datetime.datetime.fromisoformat(end_time_after)
if end_time_after
else None,
limit=limit,
page_size=page_size,
)
return json.loads(
json.dumps(
[call._properties for call in call_list], indent=2, cls=Message.Encoder
)
)
# vim:sw=4:ts=4:et:
