В одной сказочной стране пять седых старцев не захотели уйти на заслуженную пенсию да и открыли собственный бизнес.

В далеком, далеком, волшебном лесу, где совсем не ловит интернет они собирают дикоросы чудные. А когда соберут, значится к ним приезжает старичок-грузовичок. 

Да вот незадача, память то уже не та у всех. Бывало грузовичок забудет приехать, а бывало, что и сами старички забудут его позвать. Так бы и маялись они все от неэффективности, да на помощь к ним пришли технологии современные — Camunda BPM и МТС Exolve. А что из этого вышло мой дорогой друг ты узнаешь под катом.

Оглавление:

1. Проблема и решение

2. Процесс в Camunda

3. Скрипт отправки SMS

4. Результат

Проблема и решение

Несмотря на стилизованное вступление, задача которую мы разберем вовсе не сказочная. При соответствующей доработке решение может вполне пригодится для малого и среднего бизнеса. Например, в процессах комплектации заказа на складе, автоматизации курьерской доставки, или как способ проконтролировать выполнение периодических задач подчиненными. Camunda версии 7 бесплатна для коммерческого использования, для автоматизации требуются минимальные навыки программирования, а возможности МТС Exolve не ограничиваются отправкой и получением SMS.

Начнем с бизнес-процесса, который устаканился у наших инициативных старичков.

Предусловия:

1. Дикоросы из волшебного леса быстро портятся, если их не консервировать.

Изначально сценарий следующий:

1. Старички собирают дикоросы.

2. Докладывают бригадиру, что собрали достаточно.

3. Бригадир вызывает грузовик.

4. Грузовик приезжает и забирает товар.

5. Производство готовит продукцию – все счастливы.

Негативные сценарии:

1. Старички забывают доложить бригадиру, что пора вызывать грузовик — продукция портится.

2. Бригадир забывает вызвать грузовик — продукция портится.

3. Грузовик забывает, что его никто не вызывал и приезжает сам по себе — простой грузовика.

А мы хотим добится результата как на картинке ниже:

Для решения поставленной задачи воспользуемся Camunda 7.22.0 Open Source Community Edition и МТС Exolve API.

Если вы раньше не работали с МТС Exolve, пришло самое время зарегистрировать личный кабинет и получить 300 бонусных рублей, которых как раз с головой хватит, чтобы протестировать кейс. Подробнее о процедуре регистрации см. в документации.

Процесс в Camunda

Прежде чем приступить непосредственно к реализации замыслов, озвучу традиционный дисклеймер. Я не разработчик и не эксперт по Camunda, решения представленные в статье, не рекомендуется внедрять в продуктовой среде без анализа и доработок. 

Также сразу скажу, я специально выбрал задачу с волшебными старичками, потому что мы рассмотрим только концептуальную демонстрацию интеграции SMS API и Camunda. По сравнению с реальной автоматизацией у нас будет совсем простой кейс, но надо же с чего-то начинать.

В рамках статьи мы подготовим бизнес-процесс в camunda modeller и автоматизируем в нем отправку и получение SMS с помощью обработки внешних задач в Python скрипте.

Для наглядности я подготовил достаточно простую BPMN диаграмму.

Сценарий работы следующий:

1. Запускаем бизнес-процесс

2. Ждем по таймеру пока соберут дикоросы. В реальности можно ожидать пол-дня, но в демонстрационных целях мы будем ждать буквально несколько секунд.

3. Когда таймер сработал с помощью Service task передадим управление в скрипт на Python, который вызовет метод для отправки SMS бригадиру, чтобы понимать нужно ли высылать грузовик. Если отправка SMS прошла успешно пойдем дальше. Иначе передадим управление оператору с помощью блока user task. N.B. Как альтернативу user task в Camunda, можно напрямую из скрипта реализовать созвон между оператором и бригадиром с помощью Callback API.

4. После успешной отправки, подождем когда отработает второй таймер. В реальности могло бы быть несколько часов, но в целях демонстрации. также подождем пару секунд.

5. Передадим управление обратно в скрипт. Скрипт проверит есть ли от бригадира SMS, отправленная не раньше чем запустили процесс. Если ответ от бригадира «да», значит вызываем машину (данный блок мы не автоматизируем в рамках статьи). Если ответ не пришел или бригадир ответил что-то кроме «да», то мы передаем управление оператору, который должен решить проблему в ручном режиме.

Все материалы как обычно можно скачать на GitHub. 

Поэтому я не буду очень подробно останавливаться на всех блоках диаграммы.

Однако, с настройками ключевых блоков и потоков стоит ознакомиться:

Процесс

Для процесса необходимо установить срок хранения истории. Иначе не получится запустить процесс в Camunda.

В редакторе, нажмите на пустое пространство и установите в поле History cleanup значение P5D (каждые 5 дней). Также нам важно задать id процесса (sendReceiveSms), чтобы позже обратиться к нему из скрипта.

Остальные настройки на ваше усмотрение.

Таймеры

Оба таймера настроены на период в 2 секунды (PT2S).

Service task – Отправить SMS с текстом «Отправлять машину»?

Нас в первую очередь волнует блок Implementation.

Выберите в нем Type = External
И дайте название топику  «send_sms». Его мы будем использовать в скрипте.

Service task — Получить SMS

Настройки аналогичные, но название топика «receive_sms»

Условия потов после шлюзов.

Для потока SMS отправлено — установите условие

${sent ==  true}

Переменную для условий мы отправим в процесс из скрипта.

Для потока «Да». Настройте условие

${need_car == true}

Потоки по умолчанию.

Я не увидел в настройках пункта для установки значения потока по умолчанию.

Поэтому для того, чтобы перечеркнуть палочкой поток по умолчанию, придется открыть XML редактор и добавить шлюзу поток по умолчанию в атрибуте default.

Код для первого ответвления:

<bpmn:exclusiveGateway id="Gateway_sent_sms" default="Flow_not_sent">

      <bpmn:incoming>Flow_10zcue0</bpmn:incoming>

      <bpmn:outgoing>Flow_not_sent</bpmn:outgoing>

      <bpmn:outgoing>Flow_sent</bpmn:outgoing>

 </bpmn:exclusiveGateway>

Код для второго ответвления:

    <bpmn:exclusiveGateway id="Gateway_ready" name="Отправлять машину?" default="Flow_not_ready">

      <bpmn:incoming>Flow_1ptvosm</bpmn:incoming>

      <bpmn:outgoing>Flow_not_ready</bpmn:outgoing>

      <bpmn:outgoing>Flow_ready</bpmn:outgoing>

    </bpmn:exclusiveGateway>

В качестве альтернативы потокам можно прописать условия.

Пользовательские задачи

Их настройки по сути одинаковые. Для нас важно указать в Form reference – id формы (Form_problem), которую мы создадим ниже.

Форма для пользовательской задачи.

Нажмите в правом верхнем углу “+” и создайте новую форму для Camunda 7.

Поскольку у нас демонстрационный пример, форма будет простейшая. Я добавил строку с текстом и кнопку, которая по сути отправляет результат выполнения. Если на неё нажать, задача выполнится успешно. Будем считать, что вся логика решения проблем с вывозом дикоросов лежит на совести оператора. Не забудьте сохранить форму (не удержался от каламбура актуального после новогодних праздников).

Осталось задеплоить процесс и форму в Camunda.

Я надеюсь, что вы уже запустили её заранее.

Не забудьте добавить форму при деплое процесса.

Скрипт отправки SMS

Для скрипта я использовал окружение с Python 3.13.1, но думаю подойдет любая более менее свежая версия Python 3. Также я установил библиотеку PyCamunda

Полный код скрипта можно посмотреть в листинге ниже, а также на GitHub. 

СПОЙЛЕР!!!

import pycamunda.externaltask
import pycamunda.variable
import pycamunda.processdef
import pycamunda.processinst
import time
import http.client
import json
import datetime

# settings

SYSTEM_PHONE = "{your phone}"
COLLECTOR_PHONE = "{collector phone}"
API_KEY = "{API token}"
MESSAGE = "Отправлять машину?"
 
url = 'http://localhost:8080/engine-rest' # Correct port (assuming standard Camunda)
worker_id = 'my-worker'


def get_formatted_timestamp_utc():
    """Generates a timestamp in the format: YYYY-MM-DDTHH:MM:SS.ffffff+00:00 (UTC)."""
    # Get current UTC time
    utc_now = datetime.datetime.utcnow()
    # Format and append the UTC timezone offset
    formatted_timestamp = utc_now.strftime('%Y-%m-%dT%H:%M:%S.%f') + '+00:00'
    return formatted_timestamp
 
def send_sms():
  #send sms via MTS Exolve SMS API
  conn = http.client.HTTPSConnection("api.exolve.ru")
  payload = json.dumps({
    "number": SYSTEM_PHONE,
    "destination": COLLECTOR_PHONE,
    "text": MESSAGE
  })
  headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer '+API_KEY}
  conn.request("POST", "/messaging/v1/SendSMS", payload, headers)
  res = conn.getresponse()
  return res

def recieve_sms():
  #recieve SMS via MTS Exolve SMS API
  conn = http.client.HTTPSConnection("api.exolve.ru")
  payload = json.dumps({
    "receiver": SYSTEM_PHONE,
    "direction": 1,
    "date_gte": str(sms_date)
  })
  headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer '+API_KEY}
  conn.request("POST", "/messaging/v1/GetList", payload, headers)
  res = conn.getresponse()
  return res


# setup Camunda process and topics
start_instance = pycamunda.processdef.StartInstance(url=url, key='sendReceiveSms')
process_instance = start_instance()



fetch_and_lock_sent = pycamunda.externaltask.FetchAndLock(url=url, worker_id=worker_id, max_tasks=10)
fetch_and_lock_sent.add_topic(name='send_sms', lock_duration=10000) 

fetch_and_lock_receive = pycamunda.externaltask.FetchAndLock(url=url, worker_id=worker_id, max_tasks=10)
fetch_and_lock_receive.add_topic(name='receive_sms', lock_duration=10000) 

# get current timestamp to filter SMS
sms_date= get_formatted_timestamp_utc()

# Wait for process execute first task
time.sleep(5)

# Main loop for Camunda process
while (True):
  try:
      # get task
      tasks_sent = fetch_and_lock_sent()
      if tasks_sent:
          for task in tasks_sent:
            complete = pycamunda.externaltask.Complete(url=url, id_=task.id_, worker_id=worker_id)
            sms_res = send_sms()
            if (sms_res.status):
              complete.add_variable(name='sent', value=True)
            else:
              complete.add_variable(name='sent', value=False)
            complete()
            print("Send tasks complited")
      else:
        print("No tasks in send SMS block available")
  except Exception as e:
    print(f"An error has occurred: {e}") 

  try:
      # Wait for process execute second task
      time.sleep(20)


      # get task
      tasks_receive = fetch_and_lock_receive()
      if tasks_receive:
          for task in tasks_receive:
            # get opened tasks
            complete = pycamunda.externaltask.Complete(url=url, id_=task.id_, worker_id=worker_id)

            #get sms
            sms_res = recieve_sms()

            #get message from response body
            res_json = json.loads(sms_res.read().decode("utf-8"))
            if ("messages" in res_json) and ( len(res_json["messages"])):
              message_text = res_json['messages'][0]['text'].lower().strip()
            else:
              message_text = ""
            if (message_text == "да"):
             # if get approved from sms, go to main flow 
              complete.add_variable(name='need_car', value=True)
            else:
              # if get approved from sms, go to alternative flow 
              complete.add_variable(name='need_car', value=False)
            complete()     
            print("Recieve tasks complited")
      else:
        print("No tasks in recieve SMS block available")
  except Exception as e:
    print(f"An error has occurred: {e}") 

  # Get status of current process
  is_process_active = pycamunda.processinst.GetList(url=url, process_instance_ids=[process_instance.id_], active = True).__call__()
  #End loop if Camunda process finished
  if (not is_process_active):
     print ("Process finished")
     break

Давайте разберем основные блоки кода.

import pycamunda.externaltask
import pycamunda.variable
import pycamunda.processdef
import pycamunda.processinst
import time
import http.client
import json
import datetime

Импорт библиотек, не забудьте установить pycamunda, c помощью команды:

$ pip3 install pycamunda

SYSTEM_PHONE = "{your phone}"
COLLECTOR_PHONE = "{collector phone}"
API_KEY = "{API token}"
MESSAGE = "Отправлять машину?"

Настройки для отправки и получения SMS, с помощью МТС Exolve

• SYSTEM_PHONE – телефон из ЛК МТС Exolve, с него мы отправляем SMS и на него будет отвечать бригадир.

• COLLECTOR_PHONE – телефон бригадира

• API_KEY – ключ приложения ЛК МТС Exolve

• MESSAGE – текст отправляемого SMS

url = 'http://localhost:8080/engine-rest' # Correct port (assuming standard Camunda)
worker_id = 'my-worker'

• url – адрес Camunda API, в нашем случае для локального развертывания.

• worker_id  – идентификатор нашего воркера, которым мы позже заблокируем выполнение внешних задач в Camunda. Можно вписать любой.

def get_formatted_timestamp_utc():
   """Generates a timestamp in the format: YYYY-MM-DDTHH:MM:SS.ffffff+00:00 (UTC)."""
   # Get current UTC time
   utc_now = datetime.datetime.utcnow()
   # Format and append the UTC timezone offset
   formatted_timestamp = utc_now.strftime('%Y-%m-%dT%H:%M:%S.%f') + '+00:00'
   return formatted_timestamp

Данная функция нужна для генерации правильной даты в фильтрах SMS.

Функция send_sms() — это по сути автоматически сгенерированный с помощью Postman код, для вызова метода SendSMS в SMS API (см. документацию метода).

Функция receive_sms() — аналогично сгенерирована, но уже для вызова метода GetList в SMS API (см. документацию метода)

start_instance = pycamunda.processdef.StartInstance(url=url, key='sendReceiveSms')
process_instance = start_instance()

Запускаем новый экземпляр процесса, который мы задеплоили в Camunda на прошлом этапе.

fetch_and_lock_sent = pycamunda.externaltask.FetchAndLock(url=url, worker_id=worker_id, max_tasks=10)
fetch_and_lock_sent.add_topic(name='send_sms', lock_duration=10000)
fetch_and_lock_receive = pycamunda.externaltask.FetchAndLock(url=url, worker_id=worker_id, max_tasks=10)
fetch_and_lock_receive.add_topic(name='receive_sms', lock_duration=10000)

Готовимся обрабатывать топики из двух внешних задач процесса.

# get current timestamp to filter SMS
sms_date= get_formatted_timestamp_utc()
# Wait for process execute first task
time.sleep(5)

Получаем текущее время, чтобы при фильтре полученных SMS не получить те, что были отправлены раньше старта процесса. Это не самое надежное решение, но для демонстрационных целей подойдет, поскольку я предполагаю, что одновременно запускается только один экземпляр процесса.

Таймер на 5 секунд нужен, чтобы к моменту первого цикла процесс уже дошел до выполнения первой внешней задачи.

while (True):

Старт бесконечного цикла.

try:
     # get task
     tasks_sent = fetch_and_lock_sent()
     if tasks_sent:
         for task in tasks_sent:
           complete = pycamunda.externaltask.Complete(url=url, id_=task.id_, worker_id=worker_id)
           sms_res = send_sms()
           if (sms_res.status):
             complete.add_variable(name='sent', value=True)
           else:
             complete.add_variable(name='sent', value=False)
           complete()
           print("Send tasks complited")
     else:
       print("No tasks in send SMS block available")
 except Exception as e:
   print(f"An error has occurred: {e}")

Блок для вычитки первой внешней задачи. Если задача запустилась у нас не пустой топик, и мы проходимся по всем задачам (в нашем случае будет всего одна). Для каждой задачи отправляем SMS бригадиру. Если сообщение отправилось, возвращаем в процесс переменную sent=True, чтобы процесс пошел по положительному пути. Если были проблемы с отправкой, отправляем sent=False, чтобы у оператора создалась задача на ручное решение проблемы Затем завершаем обработку задач.

try:
     # Wait for process execute second task
     time.sleep(20)
     # get task
     tasks_receive = fetch_and_lock_receive()
     if tasks_receive:
         for task in tasks_receive:
           # get opened tasks
           complete = pycamunda.externaltask.Complete(url=url, id_=task.id_, worker_id=worker_id)
           #get sms
           sms_res = recieve_sms()
           #get message from response body
           res_json = json.loads(sms_res.read().decode("utf-8"))
           if ("messages" in res_json) and ( len(res_json["messages"])):
             message_text = res_json['messages'][0]['text'].lower().strip()
           else:
             message_text = ""
           if (message_text == "да"):
            # if get approved from sms, go to main flow
             complete.add_variable(name='need_car', value=True)
           else:
             # if get approved from sms, go to alternative flow
             complete.add_variable(name='need_car', value=False)
           complete()    
           print("Recieve tasks complited")
     else:
       print("No tasks in recieve SMS block available")
 except Exception as e:
   print(f"An error has occurred: {e}")

Ждем 20 секунд, этого времени нам хватит, чтобы в демонстрационном примере ответить на SMS.

Затем аналогично обрабатываем вторую внешнюю задачу. Только в этот раз вместо отправки SMS мы проверяем список входящих SMS на наш номер. Мы берем самое последнее входящее SMS,и парсим ответ. Если в тексте SMS было сообщение равное «да», то продолжаем процесс по успешному пути. Если ответа не было, или ответ любой другой, то создаем оператору задачу на ручное решение проблемы.

В реальности и время, которое мы даем на ответ, регулируется таймером в процессе перед второй внешней задачей. По факту это может быть и несколько часов.

 # Get status of current process
 is_process_active = pycamunda.processinst.GetList(url=url, process_instance_ids=[process_instance.id_], active = True).__call__()
 #End loop if Camunda process finished

 if (not is_process_active):
    print ("Process finished")
    break

Чтобы наш цикл не был бесконечным, мы проверяем является ли текущий процесс еще активным. Если его уже нет в списке активных процессов, завершаем цикл.

Результаты

Осталось только собрать все вместе и насладиться результатом.

Убедимся, что процесс загружен в Camunda.

Спасибо, что прочитали статью. Буду рад вашим конструктивным комментариям.