Оркестрація RPA за допомогою Flask + Celery + Redis + Streamlit: Практичний посібник

Автоматизація процесів, також відома як RPA (Robotic Process Automation), є потужною технікою, яка дозволяє автоматизувати повторювані завдання в комп'ютерних системах. Хоча існує багато комерційних інструментів, таких як BotCity, BluePrism і UiPath, багато компаній і розробників віддають перевагу створенню власних оркестраторів для RPA.

У цій статті наша мета — побудувати власний оркестратор RPA, використовуючи популярні технології екосистеми Python: Flask, Celery, Redis і Streamlit. Ми створимо просте, але потужне рішення, яке оркеструє виконання завдань Python у фоновому режимі, надаючи при цьому інтерактивний інтерфейс моніторингу для відслідковування прогресу автоматизації.

Чому не використовувати готові фреймворки?

Створення власного оркестратора може бути швидким і зручним рішенням, особливо коли існує велика бюрократія для затвердження бюджету або впровадження зовнішніх систем у компанії.

У багатьох випадках, особливо в організаціях з жорсткими вимогами безпеки або процесами відповідності (compliance), використання комерційних інструментів може зіштовхнутися з серйозними перешкодами, такими як необхідність внутрішніх затверджень, аналізу ризиків і перевірок безпеки. Ці процеси можуть зайняти місяці, що затримує впровадження необхідних рішень для підвищення продуктивності.

Більше того, створюючи власну систему, ви отримуєте повну гнучкість для адаптації її під конкретні потреби вашого бізнесу. Замість того щоб мати справу з обмеженнями інструментів сторонніх розробників, можна спроектувати оркестратор, що відповідає специфіці внутрішніх процесів компанії, не жертвуючи безпекою чи продуктивністю.

Інструменти, які ми будемо використовувати

• Flask: Веб-фреймворк, який використовуватиметься для створення API, що прийматиме запити користувачів та взаємодіятиме зі скриптами Python для автоматизації. Він направляє запити до Celery та керує комунікацією з frontend.
• Celery: Бібліотека для асинхронного виконання завдань. Вона керує та обробляє завдання автоматизації у фоновому режимі, не блокуючи основний потік програми.
• Redis: Система зберігання даних в пам'яті, яка використовується як бекенд для Celery, керуючи чергами завдань та зберігаючи їхній стан.
• Streamlit: Інструмент для створення інтерактивних панелей моніторингу. Він надає простий інтерфейс для відслідковування прогресу завдань автоматизації в реальному часі.

Робочий процес

1. Користувач взаємодіє з інтерфейсом Streamlit, відправляючи запит на запуск завдання.
2. Flask отримує запит та пересилає його до Celery.
3. Celery виконує завдання автоматизації (RPA) у фоновому режимі.
4. Redis зберігає статус завдання.
5. Streamlit оновлює інтерфейс в реальному часі, показуючи прогрес завдання.

Структура проекту

Перед тим як розпочати реалізацію, давайте визначимо структуру папок та файлів нашого проекту. Організація структури коду важлива для того, щоб система була легкою для розуміння та розширення з часом. Ось рекомендована структура для нашого проекту:

rpa-orquestrador/  
│  
├── backend/ # Папка для Flask та Celery  
│ ├── app/  
│ │ ├── __init__.py # Ініціалізація Flask додатку  
│ │ ├── tasks.py # Визначає завдання для Celery  
│ │ ├── app.py # Керує маршрутами Flask (API)  
│ │ ├── celery_config.py # Конфігурація Celery  
│ ├── scripts/ # Папка для Python скриптів автоматизації (RPA)  
│ │ └── automacao1.py # Приклад скрипту автоматизації  
│ └── requirements.txt # Залежності backend (Flask, Celery, Redis)  
│  
├── frontend/ # Папка для Streamlit  
│ └── app.py # Основний файл Streamlit  
│  
└── requirements.txt # Спільні залежності для проекту

Крок за кроком

1. Встановлення Flask

Встановлення: Перше, що потрібно зробити, це встановити Flask.

Автоматизація процесів, також відома як RPA (Robotic Process Automation), є потужною технікою, що дозволяє автоматизувати повторювані завдання в комп'ютерних системах. Хоча існують різні комерційні інструменти, такі як BotCity, BluePrism та UiPath, багато компаній і розробників віддають перевагу створенню власних оркестраторів для RPA.

У цій статті ми ставимо за мету побудувати власний оркестратор RPA, використовуючи популярні технології з екосистеми Python: Flask, Celery, Redis і Streamlit. Ми створимо просте, але потужне рішення, яке оркеструватиме виконання завдань Python у фоновому режимі, водночас надаючи інтерактивний інтерфейс для моніторингу прогресу автоматизації.

Чому не використовувати готові фреймворки?

Створення власного оркестратора може бути швидким і зручним рішенням, особливо коли є велика бюрократія для затвердження бюджету чи впровадження зовнішніх систем у компанії.

У багатьох випадках, особливо в організаціях із суворими вимогами до безпеки чи процесами відповідності (compliance), використання комерційних інструментів може зіткнутися з серйозними перешкодами, такими як необхідність внутрішніх затверджень, аналізу ризиків і перевірок безпеки. Ці процеси можуть займати місяці, що затримує впровадження рішень, необхідних для покращення продуктивності.

До того ж, створюючи власну систему, ви отримуєте повну гнучкість для її адаптації відповідно до потреб вашого бізнесу. Замість того, щоб стикатися з обмеженнями інструментів сторонніх розробників, ви можете спроектувати оркестратор, який відповідатиме специфіці внутрішніх процесів компанії, не жертвуючи безпекою чи продуктивністю.

Інструменти, які ми будемо використовувати

• Flask: Веб-фреймворк, який буде використано для створення API, що прийматиме запити користувачів і взаємодіятиме з Python-скриптами автоматизації. Він направляє запити до Celery та керує комунікацією з frontend.
• Celery: Бібліотека для асинхронного виконання завдань. Вона керує і обробляє завдання автоматизації у фоновому режимі, не блокуючи основний потік програми.
• Redis: Система зберігання даних в пам'яті, що працює як бекенд для Celery, керуючи чергами завдань і зберігаючи їхні стани.
• Streamlit: Інструмент для створення інтерактивних панелей моніторингу. Він надає простий інтерфейс для відслідковування прогресу завдань автоматизації в реальному часі.

Робочий процес

1. Користувач взаємодіє з інтерфейсом Streamlit, відправляючи запит на запуск завдання.
2. Flask отримує запит і передає його до Celery.
3. Celery виконує завдання автоматизації (RPA) у фоновому режимі.
4. Redis зберігає статус завдання.
5. Streamlit оновлює інтерфейс в реальному часі, показуючи прогрес завдання.

Структура проекту

Перед тим, як розпочати реалізацію, давайте визначимо структуру папок і файлів нашого проекту. Організація структури коду є важливою для того, щоб система була легкою для розуміння та розширення з часом. Ось рекомендована структура для нашого проекту:

rpa-orquestrador/  
│  
├── backend/ # Папка для Flask та Celery  
│ ├── app/  
│ │ ├── __init__.py # Ініціалізація Flask додатку  
│ │ ├── tasks.py # Визначення завдань для Celery  
│ │ ├── app.py # Керування маршрутами Flask (API)  
│ │ ├── celery_config.py # Конфігурація Celery  
│ ├── scripts/ # Папка для Python скриптів автоматизації (RPA)  
│ │ └── automacao1.py # Приклад скрипту автоматизації  
│ └── requirements.txt # Залежності backend (Flask, Celery, Redis)  
│  
├── frontend/ # Папка для Streamlit  
│ └── app.py # Основний файл Streamlit  
│  
└── requirements.txt # Спільні залежності для проекту

Крок за кроком

1. Встановлення Flask

Встановлення: Перше, що потрібно зробити — це встановити Flask.

Це можна легко зробити за допомогою pip:

pip install Flask

Базова структура додатку Flask: Далі створимо простий додаток Flask, який визначатиме маршрут для запуску завдання автоматизації:

from flask import Flask, jsonify, request  
from celery.result import AsyncResult  
from tasks import run_task # Імпортує функцію завдання з Celery  

app = Flask(__name__)  

@app.route('/start_task', methods=['POST'])  
def start_task():  
 # Запускає асинхронне завдання та повертає ID завдання  
 task = run_task.apply_async()  
 return jsonify({"task_id": task.id}), 202  

@app.route('/task_status/', methods=['GET'])  
def task_status(task_id):  
 task = AsyncResult(task_id)  
 return jsonify({"task_id": task.id, "status": task.status})  

if __name__ == '__main__':  
 app.run(debug=True)

Пояснення процесу: Коли користувач надсилає запит POST до ендпоїнту /start_task, Flask створює асинхронне завдання в Celery. Потім Celery обробляє це завдання у фоновому режимі, поки Flask продовжує відповідати на нові запити. Flask повертає ID завдання, щоб користувач міг відслідковувати його статус.

Ви можете запустити сервер Flask через термінал за допомогою таких команд:

cd backend  
python app.py

2. Налаштування Celery

Встановлення: Для використання Celery необхідно встановити бібліотеку:

pip install Celery

Базова конфігурація Celery: Celery потребує браузера для керування чергою завдань. Код нижче показує, як налаштувати Celery в проєкті:

from celery import Celery  

# Конфігурація Celery: вказуємо Redis як брокер  
app = Celery('tasks', broker='redis://localhost:6379/0')  

# Функція завдання для Celery  
@app.task  
def run_task():  
 # Тут буде логіка завдання автоматизації  
 print("Виконую завдання автоматизації")  
 return "Завдання успішно завершено"

• Celery('tasks', broker='redis://localhost:6379/0'): Цей рядок налаштовує Celery на використання Redis як брокера. Redis працює на localhost на порту 6379, що є стандартом.
• @app.task: Цей декоратор реєструє функцію run_task як асинхронне завдання Celery. Тому, коли Flask відправляє запит для цього завдання, Celery виконує його у фоновому режимі.

Запуск Celery Worker: Щоб Celery виконував завдання, потрібно запустити worker, який буде чекати і обробляти завдання з черги. Для цього виконайте таку команду в терміналі:

celery -A tasks worker --loglevel=info

-A tasks вказує, що Celery повинен використовувати файл tasks.py для пошуку завдань. Worker буде чекати нові завдання та виконувати їх, коли вони надійдуть.

3. Налаштування Redis

Встановлення Redis: Redis можна встановити на вашу систему або використовувати в контейнері Docker. Особисто я віддаю перевагу використанню Docker, ось команда для запуску контейнера:

docker run --name redis -p 6379:6379 -d redis

Налаштування Celery для використання Redis: У коді Celery ми вже бачили, що Redis використовується як брокер. Конфігурація у Celery виглядає ось так:

app = Celery('tasks', broker='redis://localhost:6379/0')

Щоб перевірити, чи правильно працює Redis, ви можете виконати команду redis-cli в терміналі, а потім ввести команду ping. Якщо все налаштовано правильно, він поверне "PONG".

4. Налаштування Streamlit

Встановлення Streamlit: Щоб встановити Streamlit в своєму середовищі, достатньо виконати таку команду:

pip install streamlit

Створення додатку Streamlit: Далі створимо простий додаток для взаємодії з Flask і перегляду статусу завдань.

Це можна легко зробити за допомогою pip:

pip install Flask

Базова структура додатку Flask: Далі створимо простий додаток Flask, який визначатиме маршрут для запуску завдання автоматизації:

from flask import Flask, jsonify, request  
from celery.result import AsyncResult  
from tasks import run_task # Імпортує функцію завдання з Celery  

app = Flask(__name__)  

@app.route('/start_task', methods=['POST'])  
def start_task():  
 # Запускає асинхронне завдання та повертає ID завдання  
 task = run_task.apply_async()  
 return jsonify({"task_id": task.id}), 202  

@app.route('/task_status/', methods=['GET'])  
def task_status(task_id):  
 task = AsyncResult(task_id)  
 return jsonify({"task_id": task.id, "status": task.status})  

if __name__ == '__main__':  
 app.run(debug=True)

Пояснення процесу: Коли користувач надсилає запит POST до ендпоїнту /start_task, Flask створює асинхронне завдання в Celery. Потім Celery обробляє це завдання у фоновому режимі, поки Flask продовжує відповідати на нові запити. Flask повертає ID завдання, щоб користувач міг відслідковувати його статус.

Ви можете запустити сервер Flask через термінал за допомогою таких команд:

cd backend  
python app.py

2. Налаштування Celery

Встановлення: Для використання Celery необхідно встановити бібліотеку:

pip install Celery

Базова конфігурація Celery: Celery потребує браузера (broker) для керування чергою завдань. Код нижче показує, як налаштувати Celery в проєкті:

from celery import Celery  

# Конфігурація Celery: вказуємо Redis як брокер  
app = Celery('tasks', broker='redis://localhost:6379/0')  

# Функція завдання для Celery  
@app.task  
def run_task():  
 # Тут буде логіка завдання автоматизації  
 print("Виконую завдання автоматизації")  
 return "Завдання успішно завершено"

• Celery('tasks', broker='redis://localhost:6379/0'): Цей рядок налаштовує Celery на використання Redis як брокера. Redis працює на localhost на порту 6379, що є стандартом.
• @app.task: Цей декоратор реєструє функцію run_task як асинхронне завдання Celery. Тому, коли Flask відправляє запит для цього завдання, Celery виконує його у фоновому режимі.

Запуск Celery Worker: Щоб Celery виконував завдання, потрібно запустити worker, який буде чекати і обробляти завдання з черги. Для цього виконайте таку команду в терміналі:

celery -A tasks worker --loglevel=info

-A tasks вказує, що Celery повинен використовувати файл tasks.py для пошуку завдань. Worker буде чекати нові завдання та виконувати їх, коли вони надійдуть.

3. Налаштування Redis

Встановлення Redis: Redis можна встановити на вашу систему або використовувати в контейнері Docker. Особисто я віддаю перевагу використанню Docker, ось команда для запуску контейнера:

docker run --name redis -p 6379:6379 -d redis

Налаштування Celery для використання Redis: У коді Celery ми вже бачили, що Redis використовується як брокер. Конфігурація у Celery виглядає ось так:

app = Celery('tasks', broker='redis://localhost:6379/0')

Щоб перевірити, чи правильно працює Redis, ви можете виконати команду redis-cli в терміналі, а потім ввести команду ping. Якщо все налаштовано правильно, він поверне "PONG".

4. Налаштування Streamlit

Встановлення Streamlit: Щоб встановити Streamlit в своєму середовищі, достатньо виконати таку команду:

pip install streamlit

Створення додатку Streamlit: Далі створимо простий додаток для взаємодії з Flask і перегляду статусу завдань.

Файл app.py буде основним кодом для Streamlit:

import streamlit as st  
import requests  
import time  

# URL API Flask, яка буде керувати завданнями  
FLASK_API_URL = "http://localhost:5000/start-task"  

# Функція для запуску завдання  
def start_task():  
 response = requests.post(FLASK_API_URL)  
 if response.status_code == 200:  
 task_id = response.json()['task_id']  
 st.write(f"Завдання успішно запущено! ID: {task_id}")  
 return task_id  
 else:  
 st.write("Помилка при запуску завдання")  
 return None  

# Функція для моніторингу статусу завдання  
def monitor_task(task_id):  
 status_url = f"http://localhost:5000/status/{task_id}"  
 while True:  
 response = requests.get(status_url)  
 if response.status_code == 200:  
 status = response.json()['status']  
 st.write(f"Статус завдання {task_id}: {status}")  
 if status == 'Завершено':  
 break  
 time.sleep(2) # Чекати 2 секунди перед наступною перевіркою  

# Інтерфейс Streamlit  
st.title("Моніторинг завдань RPA")  

if st.button("Запустити завдання"):  
 task_id = start_task()  
 if task_id:  
 monitor_task(task_id)

• При натисканні кнопки "Запустити завдання", вона надсилає запит POST до Flask, який потім запускає завдання автоматизації.
• Після запуску завдання, система моніторить статус завдання, перевіряючи його кожні 2 секунди, чи було воно завершене. Коли завдання завершується, статус оновлюється на "Завершено".

Запуск Streamlit: Після створення файлу, запустіть додаток Streamlit за допомогою наступної команди:

cd frontend  
streamlit run app.py

Покращення та фінальні зауваження

Готово! Тепер у вас є функціональний оркестратор для ваших RPA, використовуючи Flask, Celery, Redis та Streamlit. З цією базовою структурою ви можете почати автоматизувати ваші процеси та моніторити їх просто та ефективно.

Звісно, є багато покращень, які можна реалізувати, щоб розширити та персоналізувати вашу систему. Ось кілька ідей:

• Додавання нових скриптів автоматизації: Ви можете легко додавати нові Python скрипти, які виконують різні завдання RPA.
• Створення панелі моніторингу для Celery: Використання API Celery для моніторингу виконуваних завдань, помилок та історії завдань.
• Аутентифікація в Streamlit: Реалізація системи аутентифікації, щоб лише авторизовані користувачі могли запускати та моніторити завдання.
• Масштабованість: Якщо обсяг завдань зросте, можна налаштувати кілька worker'ів Celery або навіть використовувати більш потужне рішення для оркестрації.
• Покращення інтерфейсу: Зробіть інтерфейс Streamlit більш насиченим, з графічними візуалізаціями прогресу завдань, а також додайте сповіщення про помилки.

З такими доповненнями ваш оркестратор може стати ще потужнішим і здатним ефективно управляти складними автоматизаціями.

Якщо у вас виникли питання або проблеми, ви можете знайти мене на LinkedIn! 😉
Файл app.py буде основним кодом для Streamlit:

import streamlit as st  
import requests  
import time  

# URL API Flask, яка буде керувати завданнями  
FLASK_API_URL = "http://localhost:5000/start-task"  

# Функція для запуску завдання  
def start_task():  
 response = requests.post(FLASK_API_URL)  
 if response.status_code == 200:  
 task_id = response.json()['task_id']  
 st.write(f"Завдання успішно запущено! ID: {task_id}")  
 return task_id  
 else:  
 st.write("Помилка при запуску завдання")  
 return None  

# Функція для моніторингу статусу завдання  
def monitor_task(task_id):  
 status_url = f"http://localhost:5000/status/{task_id}"  
 while True:  
 response = requests.get(status_url)  
 if response.status_code == 200:  
 status = response.json()['status']  
 st.write(f"Статус завдання {task_id}: {status}")  
 if status == 'Завершено':  
 break  
 time.sleep(2) # Чекати 2 секунди перед наступною перевіркою  

# Інтерфейс Streamlit  
st.title("Моніторинг завдань RPA")  

if st.button("Запустити завдання"):  
 task_id = start_task()  
 if task_id:  
 monitor_task(task_id)

• При натисканні кнопки "Запустити завдання", вона надсилає запит POST до Flask, який потім запускає завдання автоматизації.
• Після запуску завдання система моніторить його статус, перевіряючи кожні 2 секунди, чи завершено воно. Коли завдання завершується, статус оновлюється на "Завершено".

Запуск Streamlit: Після створення файлу, запустіть додаток Streamlit за допомогою наступної команди:

cd frontend  
streamlit run app.py

Покращення та фінальні зауваження

Готово! Тепер у вас є функціональний оркестратор для ваших RPA, використовуючи Flask, Celery, Redis та Streamlit. З цією базовою структурою ви можете почати автоматизувати ваші процеси та моніторити їх просто та ефективно.

Звісно, є багато покращень, які можна реалізувати, щоб розширити та персоналізувати вашу систему. Ось кілька ідей:

• Додавання нових скриптів автоматизації: Ви можете легко додавати нові Python скрипти, які виконують різні завдання RPA.
• Створення панелі моніторингу для Celery: Використання API Celery для моніторингу виконуваних завдань, помилок та історії завдань.
• Аутентифікація в Streamlit: Реалізація системи аутентифікації, щоб лише авторизовані користувачі могли запускати та моніторити завдання.
• Масштабованість: Якщо обсяг завдань зросте, можна налаштувати кілька worker'ів Celery або навіть використовувати більш потужне рішення для оркестрації.
• Покращення інтерфейсу: Зробіть інтерфейс Streamlit більш насиченим, з графічними візуалізаціями прогресу завдань, а також додайте сповіщення про помилки.

З такими доповненнями ваш оркестратор може стати ще потужнішим і здатним ефективно управляти складними автоматизаціями.

Якщо у вас виникли питання або проблеми, ви можете знайти мене на LinkedIn! 😉

Перекладено з: Orquestração de RPA com Flask + Celery + Redis + Streamlit: Um Guia Prático