Введение
В этой статье мы собираемся создать приложение, которое отправляет уведомления о назначении пользователям через WhatsApp через Twilio. Согласно статистике, 90,4% молодого поколения активны в социальных сетях, поэтому этот подход отправки уведомлений будет более эффективным в установлении убежистов без навязчивого.
Пользовательский путь
Запрос API отправляется в заявку, содержащую данные о назначении (заголовок, описание, телефон и время). Как только время назначения достигнуто, пользователь получает уведомление о WhatsApp о встрече.
Учебные требования
Чтобы следовать этому уроку, вы ожидаете:
- Иметь достаточное понимание Python и Flask
- Установите Python 3 на машине
- Установите MongoDB на вашей машине
- Иметь смартфон с активным номером телефона и WhatsApp установлены
- Иметь аккаунт Twilio. Вы можете создать бесплатную учетную запись, если вы новичок в Twilio.
Настройка песочницы Twilio WhatsApp
Прежде всего, вам нужно создать учетную запись Twilio, если у вас ее нет или войти, если у вас есть. Вам нужно будет активировать песочницу Twilio WhatsApp, так как мы будем работать с WhatsApp. Песочница позволяет вам немедленно проверить с WhatsApp, используя номер телефона, не ожидая, пока вас будет утвержден номер Twilio. Вы можете Запросите добычи производства для вашего номера телефона Twilio здесь Анкет Чтобы подключиться к песочнице, войдите в свой Twilio Console и выберите Программируемые SMS в боковом меню. После этого нажмите на WhatsApp Анкет Это откроет страницу песочницы WhatsApp, которая содержит номер телефона песочницы и код соединения. Отправьте код соединения, начиная со слова «соединение» на номер телефона песочницы, чтобы включить песочницу WhatsApp для вашего телефона. Затем вы получите сообщение, подтверждающее активацию номера телефона для использования песочницы.
Настройка приложения
Мы собираемся использовать флажок для создания приложения и MongoDB в качестве предпочтительной базы данных.
Создание каталога приложений и виртуальной среды запускают следующие команды в вашем терминале для создания папки проекта и виртуальной среды для этого проекта.
- Создать каталог проектов с именем
WhatsApp_papointmentsmkdir whatsapp_ppointments - Вступите в компакт -диск Directory Project WhatsApp_ppointments
- Создайте виртуальную среду для Project Python3 -m Venv
- Активировать виртуальную среду
Для пользователей MacOS и Linux:
source venv/bin/activate
Для пользователей Windows:
venv\Scripts\activate
Виртуальная среда помогает создать изолированную среду, отделенную от глобальной среды Python для запуска проекта.
Структурирование проекта В каталоге приложений запустите команды ниже, чтобы настроить структуру папки:
mkdir controllers jobs services utils touch app.py
Приведенные выше команды создают 4 папки и файл с именем app.py В каталоге приложений. Ваш каталог приложений должен выглядеть так:
Установка зависимостей проекта Наконец, давайте установим все зависимости, используемые в этом проекте.
- Колба: эта библиотека будет использоваться для запуска нашего веб -сервера.
- Pimongo: Эта библиотека будет использоваться для взаимодействия с базой данных MongoDB на вашем компьютере.
- Библиотека Twilio Python Helper: Эта библиотека будет использоваться для отправки сообщений WhatsApp пользователям.
Запустите команду ниже, чтобы установить эти зависимости:
pip install Flask pymongo twilio
Построение напоминаний о назначении
В этом разделе мы собираемся написать логику для напоминаний о назначении. Мы собираемся использовать Mutprocessing Модуль в Python для создания логики напоминаний. Основная причина, по которой мы используем процессы по потокам в этом разделе, заключается в том, что в Python нет безопасного способа прекратить потоки, и нам нужно будет отредактировать и отменить встречи 🤷🏿 ♂️. Прежде всего, создайте 2 файла с названием __init__.py и Назначения.py в Джобс/ каталог. Также создайте файл с именем WhatsApp.py в utils/ каталог. Скопируйте код ниже в WhatsApp.py файл:
import os
from twilio.rest import Client
ACCOUNT_SID = os.getenv('ACCOUNT_SID')
AUTH_TOKEN = os.getenv('AUTH_TOKEN')
TWILIO_NUMBER = os.getenv('TWILIO_NUMBER')
client = Client(ACCOUNT_SID, AUTH_TOKEN)
def format_message(appointment):
message = "You have an appointment\n"
message += f"Title: {appointment['title']}\n"
message += f"Description: {appointment['description']}\n"
message += f"Time: {appointment['time']}"
return message
def send_message(appointment):
message = client.messages.create(
from_=f'whatsapp:{TWILIO_NUMBER}',
body=format_message(appointment),
to=f"whatsapp:{appointment['phone']}"
)
return message.sid
В приведенном выше коде у нас есть логика для отправки сообщений WhatsApp. Обратите внимание, что Account_sid , Auth_token , Twilio_number устанавливаются в качестве переменных среды, и мы используем Os.getenv Чтобы получить доступ к ним. Чтобы отправить сообщения WhatsApp, вы должны префикс номера телефонов с WhatsApp: иначе это будет отправлено как SMS. format_message Функция принимает объект встречи и представляет его в следующем формате:
You have an appointment Title:Description: Time:
Далее мы собираемся написать логику напоминания. Откройте файл jobs/ewingsments.py и скопируйте следующий код внутри:
import time
from datetime import datetime
from utils.whatsapp import send_message
def start_appointment_job(appointment):
print(f"=====> Scheduled Appointment {appointment['_id']} for :> {appointment['time']}")
if datetime.now() > appointment['time']:
return
diff = appointment['time'] - datetime.now()
time.sleep(diff.seconds)
send_message(appointment)
start_appointment_job Функция сначала проверяет, если назначение находится в прошлом, и если да, он возвращается. Мы используем Time.sleep Функция, чтобы сообщить заявлению приостановить выполнение в течение количества секунд, необходимых для достижения времени встречи. Как только он заканчивает сон, он вызывает send_message Функция, написанная ранее, чтобы уведомить пользователя о встрече через WhatsApp. Затем мы собираемся написать сервисные функции для создания, получения, обновления и удаления встреч. Создайте файл с именем Назначения.py в Услуги/ папка и скопируйте следующее внутри:
from bson.objectid import ObjectId
from pymongo import MongoClient
client = MongoClient('mongodb://localhost/')
appointments = client['twilio']['appointments']
def create_appointment(data):
appointments.insert_one(data)
def get_appointment(appointment_id):
result = appointments.find_one({'_id': ObjectId(appointment_id)})
return dict(result) if result else None
def get_appointments(conditions):
if '_id' in conditions:
conditions['_id'] = ObjectId(conditions['_id'])
results = appointments.find(conditions)
data = []
for result in results:
data.append(dict(result))
return data
def update_appointment(appointment_id, data):
appointment = appointments.find_one_and_update({'_id': ObjectId(appointment_id)}, {'$set': data},
return_document=True)
return appointment
def delete_appointment(appointment_id):
appointments.find_one_and_delete({'_id': ObjectId(appointment_id)})
В приведенном выше коде мы пишем функции для выполнения операций CRUD на назначения Коллекции в MongoDB. Мы используем Pimongo Библиотека для подключения к нашему местному экземпляру MongoDB. Обратите внимание, что _id Поле не может быть отправлено как строка, иначе данные не будут возвращены, мы должны отбрасывать его на ObjectId экземпляр, который использует MongoDB.
Наконец, мы собираемся написать логику для создания процессов для встречи и изменения их в зависимости от изменения назначений. Open jobs/__ init__.py и скопируйте следующий код внутри:
from multiprocessing import Process
from datetime import datetime
from .appointments import start_appointment_job
from services.appointments import get_appointments
WORKERS = {}
def terminate_worker(worker):
try:
worker.terminate()
worker.join()
worker.close()
except Exception as err:
print('====> Error occurred terminating process', err)
def schedule_appointment(appointment):
appointment_id = str(appointment['_id'])
worker = Process(target=start_appointment_job, args=(appointment,))
worker.start()
WORKERS[appointment_id] = worker
def update_scheduled_appointment(appointment_id, updated_appt):
worker = WORKERS[appointment_id]
terminate_worker(worker)
new_worker = Process(target=start_appointment_job, args=(updated_appt,))
new_worker.start()
WORKERS[appointment_id] = new_worker
def delete_scheduled_appointment(appointment_id):
worker = WORKERS[appointment_id]
terminate_worker(worker)
del WORKERS[appointment_id]
def init_workers():
print('=====> Initializing workers')
appts = get_appointments({})
for appt in appts:
if datetime.now() > appt['time']:
continue
schedule_appointment(appt)
def close_workers():
for appointment_id, worker in WORKERS.items():
terminate_worker(worker)
В приведенном выше коде, когда мы хотим назначить встречу, мы создаем Процесс нацеливание start_appointment_job Функция с назначение объект как аргумент. Обратите внимание, что у нас есть возражение в словаре по имени Рабочие где мы храним объект процесса с ebinement_id Как ключ, это то, что мы используем, чтобы получить Процесс объект в случае обновления. Чтобы обновить встречу, мы получаем объект процесса через ebinement_id и прекратить это. Затем мы создаем новый Процесс объект с обновленными деталями и обновите ссылку в Рабочие толковый словарь. Чтобы удалить встречу, мы получаем Процесс объект и прекратите его. Затем мы удаляем ключ из Рабочие Словарь, чтобы у него снова не было ссылки. Обратите внимание, что завершить Процесс Объект, мы сначала называем прекратить Функция на нем при отправке Sigterm сигнал на процесс. Присоединяйтесь Функция ждет Процесс объект прекратить и Закрыть Функция сообщает процессу выпустить ресурсы, которые он удерживает. Допустим, наше приложение перезагружается, мы хотим иметь возможность воссоздать наши объекты процесса, и именно здесь init_workers Функция вступает в игру. Он получает все встречи и звонит PADECTION_APPOINTMENT Функция на встречах, которые все еще находятся на рассмотрении.
Построение API
В этом разделе мы собираемся создать маршруты API, которые мы собираемся использовать для создания, обновления, получения и удаления встреч. Прежде всего, мы должны создать функции утилиты для обработки как наших запросов, так и наших ответов API. Создать 2 файла с названием request.py и response.py в utils/ папка. Скопируйте код ниже в request.py файл:
from datetime import datetime
def validate_body(data, required_fields):
for field in required_fields:
if field not in data:
return False, field
return True, None
def parse_appointment(body):
try:
if body.get('time'):
time_obj = datetime.strptime(body['time'], '%Y-%m-%d %H:%M')
body['time'] = time_obj
return True, None
except Exception as err:
return False, str(err)
Из приведенного выше кода мы видим, что validate_body Функция принимает словарь данных и список необходимых полей. Он проверяет требуемые поля против словаря данных и возвращает статус false, если поле отсутствует. parse_papointment Функция используется для преобразования время Поле в объекте встречи в объект DateTime с использованием DateTime.Strptime функция, которая принимает строку времени и формат времени. Функция возвращает false, если она не может преобразовать строку в объект DateTime. Далее скопируйте код ниже в response.py файл:
import json
import time
from flask import jsonify
def stringify_objectid(data):
str_data = json.dumps(data, default=str)
return json.loads(str_data)
def response(status, message, data, status_code=200):
"""
:param status : Boolean Status of the request
:param status_code: Status Code of response
:param message : String message to be sent out as description of the message
:param data : dictionary representing extra data
"""
if data:
data = stringify_objectid(data)
res = {'status': status, 'message': message, 'data': data, 'timestamp': timestamp()}
return jsonify(res), status_code
def error_response(message, status='error', code='R0', status_code=400):
res = {'message': message, 'status': status, 'code': code}
return jsonify(res), status_code
def timestamp():
"""
Helper Function to generate the current time
"""
return time.time()
В приведенном выше коде у нас есть вспомогательные функции для возврата ответов API и сообщений об ошибках. stringify_objectid Функция используется для преобразования объекта MongoDB ObjectId в строку.
Далее мы собираемся написать наши функции контроллера. Создайте файл с именем Назначения.py в контроллеры/ папка и скопируйте следующее внутри:
from flask import Blueprint, request
from services import appointments as apt_service
from utils.request import validate_body, parse_appointment
from utils.response import response, error_response
from utils.whatsapp import send_sms
import jobs
appointments = Blueprint('appointments', __name__)
@appointments.route('/', methods=['POST'])
def create():
body = request.get_json()
status, missing_field = validate_body(body, ['title', 'phone', 'description', 'time'])
if not status:
return error_response(f'{missing_field} is required')
status, error = parse_appointment(body)
if not status:
return error_response(error)
try:
apt_service.create_appointment(body)
jobs.schedule_appointment(body)
return response(True, 'Appointment created successfully', body)
except Exception as err:
print('=====> Error', err)
return error_response(str(err))
@appointments.route('/')
def view():
conditions = dict(request.args)
try:
data = apt_service.get_appointments(conditions)
return response(True, 'Appointments', data)
except Exception as err:
print('=====> Error', err)
return error_response(str(err))
@appointments.route('/')
def view_one(appointment_id):
try:
data = apt_service.get_appointment(appointment_id)
return response(True, 'Appointment', data)
except Exception as err:
print('=====> Error', err)
return error_response(str(err))
@appointments.route('/', methods=['PUT'])
def update(appointment_id):
body = request.get_json()
try:
parse_appointment(body)
appointment = apt_service.update_appointment(appointment_id, body)
jobs.update_scheduled_appointment(appointment_id, appointment)
return response(True, 'Updated Appointment', appointment)
except Exception as err:
print('=====> Error', err)
return error_response(str(err))
@appointments.route('/', methods=['DELETE'])
def delete(appointment_id):
try:
apt_service.delete_appointment(appointment_id)
jobs.delete_scheduled_appointment(appointment_id)
return response(True, 'Appointment deleted successfully', None)
except Exception as err:
print('====> Error', err)
return error_response(str(err))
В приведенном выше коде мы создали новый план Flask и функции для создания встреч, получения встреч, обновления встреч и удаления встреч. Как вы можете видеть из приведенного выше кода, мы называем функцию назначения расписания, чтобы запустить процесс планирования при создании встречи. Функция обновления вызывает update_scheduled_papointment Функция для пересмотра назначения и функции DELETE вызывает DELETE_SCHEDULED_APPOINTMENT Функция удаления процесса планирования.
Наконец, откройте app.py файл и скопируйте следующее внутри:
import atexit
from flask import Flask
from controllers.appointments import appointments
from jobs import close_workers, init_workers
def create_app():
init_workers()
atexit.register(close_workers)
return Flask(__name__)
app = create_app()
app.register_blueprint(appointments, url_prefix='/api/appointments')
if __name__ == '__main__':
app.run()
В приведенном выше коде используйте Register_blueprint Функция, чтобы убедиться, что URL -адреса, начинающиеся с /API/встречи направляются через план встреч, и наши функции контроллера выполняются для соответствующих URL -адресов. Мы используем atexit модуль для выполнения задачи при закрытии приложения. В create_app Функция, мы говорим системе сначала расписание, ожидающие назначения, а затем закрываем все запуска процессов, когда приложение закрывается.
Тестирование приложения
Чтобы запустить приложение, откройте свой терминал и выполните Python app.py в каталоге вашего приложения. Вы должны увидеть что -то вроде изображения ниже:
Теперь вы можете отправить запрос сообщения в http://127.0.0.1:5000/api/appointments Чтобы создать встречу, и вы получите сообщение WhatsApp, когда будет достигнуто время встречи. Ниже приведен образец запроса на корпус:
{
"title": "Data Structures & Algorithms",
"phone": "+2349094739283",
"description": "Watch Youtube videos on Data Structures and Algorithms",
"time":"2020-02-23 18:10"
}
Вы должны получить сообщение WhatsApp, когда время назначения достигнуто, как показано на изображении ниже
Вывод
Теперь мы подошли к концу учебника. Мы успешно создали заявку, которая отправляет уведомление о встрече с WhatsApp, используя Twilio, Python и Flask. Вы найдете исходный код для этого урока Здесь Анкет
Оригинал: “https://dev.to/favouroked/sending-app-notifications-over-whatsapp-using-twilio-dm6”