В современном мире услуги доставки становятся всё более популярными и востребованными, поэтому любая возможность автоматизации в этой сфере принесёт большую пользу как бизнесу, так и пользователям. В прошлых статьях нашего блога мы рассказывали о применении машинного зрения и нейронных сетей для распознавания ценников товаров в магазине, а также для распознавания комплектующих деталей. В этой статье мы расскажем о менее амбициозной (но не менее интересной) задаче – автоматизации оповещения клиентов о статусе их заказов с использованием чат-бота в Telegram, QR-кодов и реляционной СУБД SAP SQL Anywhere.
Нужно отметить, что такая задача разбиралась исключительно в образовательных целях, и решение пока ограничилось лишь игрушечным прототипом. Тем не менее, мы снова прибегли к машинному зрению, обработке QR-кодов, а также Telegram ботам. Мы прекрасно понимаем, что тема ботов в Telegram на Хабре практически полностью раскрыта, поэтому постараемся не заострять на этой части особого внимания. Отметим только, что выбор пал на Telegram из-за его популярности в России и удобного API. Также в конце статьи будет ссылка на Github репозиторий этого проекта, так что любой желающий сможет изучить его исходный код. Поэтому, если вам интересна тема уведомлений пользователей Telegram по QR-коду с live-трансляции веб-камеры и их потенциального применения, то милости просим под кат.
Постановка задачи
Как вы знаете, обычно уведомления о статусе онлайн-заказа приходят по e-mail, но сейчас интернет-магазины всё чаще информируют своих клиентов через мессенджеры. Когда мы начинали работать над этой задачей, то тоже решили, что через бот в мессенджере можно гораздо активнее информировать о статусе («принят», «прибыл на склад», «взят со склада», «в пути» и т.д.), не прибегая к рассылке спама по почте. Разумеется, важным условием является предоставление клиенту выбора «уровня» уведомлений – какие статусы своего заказа он хотел бы отслеживать, а также возможности и вовсе полностью от них отписаться. Как можно видеть на обобщённой блок-схеме ниже, мы подключили бота к веб-камере и сделали триггером отправки сообщения появление в эфире определённого QR-кода.
Итак, пусть заказы наших клиентов хранятся в базе данных SQL Anywhere, и у нас есть оборудование, оснащённое веб-камерой (Raspberry Pi с PiCam, например) и подключённое к этой базе. Допустим, что заказанный клиентом товар доставлен на склад. На товаре наклеен QR-код, по которому однозначно определяется заказ в базе. Мы хотим, чтобы, как только QR-код товара появлялся в кадре складской веб-камеры, клиенту, заказавшему этот товар, отсылалось уведомление в Telegram о прибытии его заказа на склад. Причём в нём должны учитываться предпочитаемый язык текста и временная зона, в которой находится клиент, а также детали товара.
Для нашего прототипа считается, что QR-код кодирует физический адрес клиента, но он вполне может кодировать что угодно, что позволяет идентифицировать заказ. Конечно же на практике такой QR-код должен кодировать нечто уникальное (UPC, например), но в нашей учебной модели каждый адрес соответствует единственному заказу. Вообще, подобный проект относительно легко переделать под другие задачи, связанные с распознаванием QR-кодов с эфира камеры и отсылкой сообщений в Telegram, в зависимости от содержания этих QR-кодов.
Инструментарий и структура проекта
Мы написали бота на Python, используя асинхронный фреймворк AIOgram для работы с Telegram Bot API и sqlanydb в качестве драйвера для SQL Anywhere. Для взаимодействия с веб-камерой, поиска и обработки QR-кодов мы воспользовались модулями OpenCV и NumPy. Таким образом, наш проект состоит из трёх основных частей:
· работа с базой данных SQL Anywhere (через модуль sqlanydb);
· параллельное подключение к стриму веб-камеры, поиск и обработка QR-кодов, которые в нём появляются (через модули OpenCV и NumPy);
· работа самого Telegram бота для связи с клиентами (через фреймворк AIOgram).
Далее мы более детально рассмотрим каждый из перечисленных пунктов и продемонстрируем работу нашего приложения.
База данных SQL Anywhere
SAP предоставляет реляционную СУБД SQL Anywhere. Среди её плюсов – масштабируемость на десятки тысяч соединений, эффективный менеджмент потребляемых ресурсов и низкая нагрузка на память. Мы очень хотели её немного протестировать, потому выбор базы данных для приложения пал именно на неё.
Сначала нам необходимо создать сам файл с базой. Это можно сделать через CLI-утилиту dbinit:
dbinit -dba admin,password123 -p 4k -z UTF8 -ze UTF8 -zn UTF8 orders.dbЭта команда создаёт базу данных с администратором «admin» (и паролем «password123»), размером страницы в 4 килобайта и схемой, заданной в кодировке UTF-8. Теперь можем подключиться к созданной базе «orders.db» через SQL Central (программа устанавливается в комплекте с SQL Anywhere) и создать таблицу для хранения заказов. В нашем случае таблица создаётся следующим SQL-запросом:
CREATE TABLE Orders (
-- ID of an order
id UNSIGNED INT PRIMARY KEY NOT NULL IDENTITY,
-- Product's name
product NVARCHAR(24) NOT NULL,
-- Product's model
model NVARCHAR(20),
-- Product's price (in Euros)
price DECIMAL(10,2) NOT NULL,
-- Amount of the product
amount UNSIGNED INT NOT NULL DEFAULT 1,
-- Weight of the product (in kilograms)
weight DECIMAL(8,3) NOT NULL,
-- Customer's first name
first_name NVARCHAR(16) NOT NULL,
-- Customer's last name
last_name NVARCHAR(20),
-- Customer's physical address
address NVARCHAR(48) NOT NULL,
-- Customer's Telegram ID
telegram_id UNSIGNED INT NOT NULL,
-- Customer's timezone
timezone NVARCHAR(16) DEFAULT 'UTC',
-- Customer's prefered locale
locale NVARCHAR(5) DEFAULT 'en_US'
);
В итоге пример заказа в такой таблице может выглядеть примерно так:
Эта база данных затем легко подключается к нашему приложению через модуль sqlanydb: достаточно лишь добавить наши credentials (в нашем случае admin как UID и password123 как пароль) в переменные среды (в проекте мы их храним в .env файле и подцепляем через модуль dotenv). Теперь у бота появился полный доступ к таблице Orders:
conn = sqlanydb.connect(uid=config.DB_UID, pwd=config.DB_PASSWORD)
curs = conn.cursor()Поиск и анализ QR-кодов с эфира
После успешного подключения к базе данных бот пытается найти и открыть камеру на устройстве, чтобы начать анализировать происходящее на эфире:
cap = cv2.VideoCapture(0)Ниже приведена основная функция, отвечающая за рендер UI на окне с эфиром, поиск и обработку QR-кодов на эфире и передачу информации с найденного QR-кода дальше по цепочке.
async def scan_qr(area: int = 300, color: int = 196, side: int = 240, lang: str = "en", debug: bool = False) -> None:
"""Main function that creates a screen with the capture, monitors the web-cam's stream, searches for a QR-code in a squared area and passes the decoded QR-code to the notify module.
Args:
[optional] area (int): Minimal area of a detected object to be consider a QR-code.
[optional] color (int): Minimal hue of gray of a detected object to be consider a QR-code.
[optional] side (int): Length of the side of a square to be drawn in the center of the screen.
[optional] lang (str): Language of a text to be written above the square.
[optional] debug (bool): Crops and outputs an image containing inside the square at potential detection.
"""
if (cap is None) or (not cap.isOpened()):
logger.critical("No video stream detected. "
"Make sure that you've got a webcam connected and enabled")
return
kernel = np.ones((2, 2), np.uint8)
square = create_square(cap.read()[1], side=side)
while cap.isOpened():
ret, frame = cap.read()
key = cv2.waitKey(1)
if not ret or square is None or ((key & 0xFF) in {27, ord("Q"), ord("q")}):
exit(1)
image = draw_bounds(frame, square, lang=lang)
detected, cropped = detect_inside_square(frame, square, kernel, area_min=area, color_lower=color, debug=debug)
if detected:
address = detect_qr(cropped)
if address:
logger.debug("Detected: \"{}\"", address)
await notify.start(address)
cv2.imshow("Live Capture", image)
await asyncio.sleep(0.1)
Немного поясним работу этой функции. Сперва мы проверяем, что устройство захвата изображения (она же веб-камера) открыто модулем OpenCV. Затем, так как мы следим, чтобы QR-код появился в определённой области кадра (а именно внутри квадрата с заданной стороной в центре), мы создаём массив из пар координат (x, y) четырёх вершин нашего квадрата со стороной `side` функцией create_square(). Далее мы функцией draw_bounds() рендерим на нашем кадре границы искомого квадрата, а также текст в зависимости от выбранного языка `lang`. То есть, draw_bounds() накладывает поверх стрима веб-камеры минимальный UI:
Наконец, непосредственный анализ эфира и поиск объекта, похожего на QR-код происходит в следующей функции detect_inside_square():
def detect_inside_square(frame: Any, square: np.ndarray, kernel: np.ndarray, area_min: int = 300, color_lower: int = 212, color_upper: int = 255, debug: bool = False) -> Tuple[bool, Any]:
"""Detects and analyzes contours and shapes on the frame. If the detected shape's area is >= :area_min:, its color hue is >= :color_lower and a rectangle that encloses the shape contains inside the square returns True and the cropped image of the frame.
Args:
frame (Union[Mat, UMat]): A frame of the webcam's captured stream.
square (np.ndarray): A numpy array of the square's (x,y)-coordinates on the frame.
kernel (np.ndarray): A kernel for the frame dilation and transformation (to detect contours of shapes in the frame).
[optional] area_min (int): Minimal area of a detected object to be consider a QR-code.
[optional] color_lower (int): Minimal hue of gray of a detected object to be consider a QR-code.
[optional] color_upper (int): Maximal hue of gray of a detected object to be consider a QR-code.
[optional] debug (bool): Crops and outputs an image containing inside the square at potential detection.
Returns:
A tuple where the first element is whether a potential shape has been detected inside the square or not.
If it was then the second element is the square-cropped image with the detected shape, None otherwise.
"""
filter_lower = np.array(color_lower, dtype="uint8")
filter_upper = np.array(color_upper, dtype="uint8")
mask = cv2.inRange(frame, filter_lower, filter_upper)
dilation = cv2.dilate(mask, kernel, iterations=3)
closing = cv2.morphologyEx(dilation, cv2.MORPH_GRADIENT, kernel)
closing = cv2.morphologyEx(dilation, cv2.MORPH_CLOSE, kernel)
closing = cv2.GaussianBlur(closing, (3, 3), 0)
edge = cv2.Canny(closing, 175, 250)
if debug:
cv2.imshow("Edges", edge)
contours, hierarchy = cv2.findContours(edge, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
for contour in contours:
area = cv2.contourArea(contour)
if area < area_min:
continue
rect = cv2.minAreaRect(contour)
box = cv2.boxPoints(rect)
box = np.int0(box)
rect = order_points(box)
cv2.drawContours(frame, [box], 0, (0, 0, 255), 1)
if contains_in_area(rect, square):
cropped = frame[square[0][1]:square[2][1], square[0][0]:square[2][0]]
if debug:
cv2.imshow("Cropped", cropped)
return (True, cropped)
return (False, None)
Как вы успели заметить, по ходу написания кода мы старались его документировать с помощью докстрингов, поэтому ограничимся лишь описанием логики работы этой функции. Считается, что вокруг наших QR-кодов есть белый бордюр. Следовательно, мы начинаем с того, что отфильтровываем объекты в кадре по их «светлости» (яркости). Затем мы извлекаем контуры отфильтрованных объектов и вычисляем площади каждого из них. Если площадь объекта меньше минимального порога, заданного в переменной `area_min`, то нас такой объект не интересует, и мы переходим к следующему. Уточним, что все эти параметры можно передать скрипту через CLI-аргументы. Если же объект подходит по площади, то остаётся проверить, что он целиком содержится внутри квадрата, определённого выше. Для этого мы описываем вокруг объекта прямоугольник, сортируем его вершины так, чтобы первой вершиной в массиве была левая верхняя, а последней – левая нижняя, и проверяем довольно тривиальной функцией:
def contains_in_area(rectangle: np.ndarray, square: np.ndarray) -> bool:
"""Checks whether a rectangle fully contains inside the area of a square.
Args:
rectangle (np.array): An ordered numpy array of a rectangle's coordinates.
square (np.array): An ordered numpy array of a square's coordinates.
Returns:
Whether the rectangle contains inside the square. Since the both arrays are ordered it's suffice
to check that the top-left and the bottom-right points of the rectangle are both in the square.
"""
if ((rectangle[0][0] < square[0][0]) or (rectangle[0][1] < square[0][1])) or (
(rectangle[2][0] > square[2][0]) or (rectangle[2][1] > square[2][1])
):
return False
return True
Если все условия выполнены, то возможно, что наш объект является QR-кодом, и мы передаём изображение в квадрате (переменная `cropped`) в функцию detect_qr() для дальнейшего анализа.
Если объект действительно оказывается QR-кодом, который можно расшифровать, то мы посылаем соответствующий запрос в нашу базу (напомним, что QR-код кодирует значения поля `address`), извлекаем информацию о заказе и отсылаем уведомление по Telegram ID клиента:
async def start(address: str, pause_success: int = 5, pause_fail: int = 1) -> None:
"""Checks whether the :address: string contains in the set of all different addresses saved in the table.
If it does, gets the record containing :address: in its "address" field.
Sends the record to the notification function.
Args:
address (str): The decoded address to check the table with.
[optional] pause_success (int): Time in seconds to standby for after the notification was sent.
[optional] pause_fail (int): Time in seconds to standby for after detecting an invalid QR-code.
"""
try:
query_addresses = "SELECT address FROM %s.%s;"
curs.execute(
query_addresses
% (
config.DB_UID,
config.DB_TABLE_NAME,
)
)
response_addresses = curs.fetchall()
addresses = set([res[0] for res in response_addresses])
if not (address in addresses):
logger.warning('Address "{}" not found among the available addresses. Skipping', address)
logger.info("Standing by for {} second(s)", pause_fail)
await asyncio.sleep(pause_fail)
return
query = "SELECT * FROM %s.%s WHERE address='%s';"
curs.execute(
query
% (
config.DB_UID,
config.DB_TABLE_NAME,
address,
)
)
response = curs.fetchone()
logger.debug('Got response for address "{}": "{}"', address, response)
except sqlanydb.Error:
logger.exception("Encountered an error while handling query to the database. See below for the details")
return
res_row = {}
for (i, field) in zip(range(len(response)), config.FIELDS):
res_row[field] = response[i]
await notify_user(res_row)
logger.info("Standing by for {} second(s)", pause_success)
await asyncio.sleep(pause_success)
async def notify_user(row: Dict[str, str]) -> None:
"""Sends a notification about the order contained in :row: to a user with a Telegram ID from :row:.
Args:
row (dict): A dict containing full record about the user's order.
"""
try:
user_id = row["telegram_id"]
timestamp = datetime.now(pytz.timezone(row["timezone"])).strftime("%d/%m/%Y %H:%M:%S %Z")
lang = row.get("locale", "en_US")
info = constants.MSG_NOTIFY_EN if lang.startswith("en") else constants.MSG_NOTIFY_RU
info = info.format(
first_name=row["first_name"],
timestamp=timestamp,
id=row["id"],
address=row["address"],
product=row["product"],
model=row["model"],
price=float(row["price"]),
amount=row["amount"],
weight=float(row["weight"])
).replace(".", "\.").replace("-", "\-")
except KeyError:
logger.exception("Got invalid query response. See below for the details")
try:
await bot.send_message(user_id, info)
logger.success("Order notification message has been successfully sent to user {}", user_id)
except CantParseEntities as ex:
logger.error("Notification failed. AIOgram couldn't properly parse the following text:\n"
"\"{}\"\n"
"Exception: {}",
info, ex)
except ChatNotFound:
logger.error("Notification failed. User {} hasn\'t started the bot yet", user_id)
except BotBlocked:
logger.error("Notification failed. User {} has blocked the bot", user_id)
except UserDeactivated:
logger.error("Notification failed. User {}\'s account has been deactivated", user_id)
except NetworkError:
logger.critical("Could not access https://api.telegram.org/. Check your internet connection")
Заметим, во-первых, что строчка
timestamp = datetime.now(pytz.timezone(row["timezone"])).strftime("%d/%m/%Y %H:%M:%S %Z")отвечает за автоматическую конвертацию даты прибытия товара в часовой пояс клиента, а во-вторых, присутствует обработка исключений на те случаи, когда клиент, например, отключил бота или деактивировал свой аккаунт в Telegram.
Стоит также упомянуть, что на данный момент в модуле sqlanydb, к сожалению, не предусмотрена sanitization SQL-запросов. Но модуль сейчас довольно активно разрабатывается, поэтому эта фича может появиться в ближайших релизах. А пока мы в функции `start()` воспользовались «костылём», который проверяет, что строка `address` содержится во множестве всех адресов из таблицы. Таким образом, если в полученном QR-коде (внезапно) закодировано нечто вроде «’; DROP TABLE Orders;», то наши данные останутся нетронутыми.
После отправки уведомления наш бот будет ждать появления следующего QR-кода. Повторим, что это лишь демонстрация конкретного примера применения подобного приложения, и что можно относительно просто переписать код под решение смежных задач.
Работа самого Telegram бота
Как мы обещали в начале, мы не станем особо вдаваться в детали. На Хабре более чем достаточно статей, посвящённых созданию ботов в Telegram (хотя, к нашему удивлению, статей, которые описывают работу именно с AIOgram на Хабре, мы не нашли). Отметим лишь, что мы использовали фреймворк AIOgram из-за асинхронности HTTP-запросов и качественной документации. Если читателям будет интересно, то мы как-нибудь подробно напишем о работе с этим фреймворком для создания Telegram ботов. На данный же момент мы ограничимся лишь демонстрацией хэндлеров, которые отвечают за установку предпочитаемого языка через ботовую команду /lang:
@dp.message_handler(commands=["lang"])
async def cmd_lang(message: Message) -> None:
"""Handles the "/lang" command from a Telegram user. Allows the user to change the locale from the chosen one.
Outputs the message in the language that was initially chosen by the user.
Args:
message (Message): User's Telegram message that is sent to the bot.
"""
query = "SELECT locale FROM %s.%s WHERE telegram_id=%d;"
curs.execute(
query
% (
config.DB_UID,
config.DB_TABLE_NAME,
message.from_user.id,
)
)
(lang,) = curs.fetchone()
logger.debug('Got user\'s {} current language "{}"', message.from_user.id, lang)
str_lang = "Please choose your language\." if lang.startswith("en") else "Пожалуйста, выберите язык\."
btn_en = InlineKeyboardButton("?? English", callback_data="lang_en")
btn_ru = InlineKeyboardButton("?? Русский", callback_data="lang_ru")
inline_kb = InlineKeyboardMarkup().add(btn_en, btn_ru)
await bot.send_message(message.chat.id, str_lang, reply_markup=inline_kb)
logger.info("User {} called /lang", message.from_user.id)
@dp.callback_query_handler(lambda c: c.data.startswith("lang"))
async def set_lang(cb_query: CallbackQuery) -> None:
"""Handles the callback that sets the user preferred locale. Updates the locale in the table.
Args:
cb_query (CallbackQuery): User's Telegram callback query that is sent to the bot.
"""
lang = "en_US" if cb_query.data.endswith("en") else "ru_RU"
info = "Setting your language..." if lang.startswith("en") else "Настраиваю язык..."
await bot.answer_callback_query(cb_query.id, text=info)
try:
query = "UPDATE %s.%s SET locale='%s' WHERE telegram_id=%d;"
curs.execute(
query
% (
config.DB_UID,
config.DB_TABLE_NAME,
lang,
cb_query.from_user.id,
)
)
logger.debug("Commiting the changes")
conn.commit()
except sqlanydb.Error as ex:
logger.exception(ex)
return
str_setlang = (
"Language is set to English\.\nCall /lang to change it\."
if lang.startswith("en")
else "Ваш язык Русский\.\nВызовите команду /lang, чтобы его изменить\."
)
logger.info('User {} set the language to "{}"', cb_query.from_user.id, lang)
await bot.send_message(cb_query.from_user.id, str_setlang)
Заметим, что возможности фреймворка AIOgram позволяют в перспективе настроить полноценную интернационализацию бота, но на текущем этапе (когда у нас всего три возможных сообщения и только два языка) вполне можно обойтись и заданием фраз вручную. Также обратим внимание, что для этих операций с нашей базой данных мы не делали никаких дополнительных проверок, так как единственное «незахардкоденное» значение для них – это ID пользователя в Telegram, а оно всегда является натуральным числом.
Результат и дальнейшие шаги
В конечном итоге мы создали игрушечный прототип системы уведомления через популярный мессенджер с использованием легковесной СУБД от компании SAP. Разумеется, описанный пример очень далёк от внедрения в реальную логистику, но подобной цели перед нами и не стояло. Мы хотели в первую очередь исследовать возможности чат-ботов и связать их с машинным зрением. Поэтому и эту статью следует рассматривать в подобном ключе. Но если наша идея будет кому-то полезна для конкретной задачи, будем рады.
Этот проект был представлен ещё летом 2019 года, задолго до текущей ситуации. И, как мы упомянули в самом начале статьи, в связи с нынешними реалиями службы доставки продолжают совершенствовать свою эффективность и повышать степень удобства для клиентов. Как показывают некоторые исследования и прогнозы, потенциал чат-ботов в этой области ещё совсем не исчерпан, и они продолжат бурно развиваться в течение ближайших нескольких лет.
Поэтому мы тоже постараемся в перспективе улучшить текущий прототип и исследовать дополнительные случаи, в которых подобные приложения смогут себя проявить. Самое очевидное, что можно сейчас сделать – это разделить логику бота и попытаться интегрировать его с SAP Cloud Platform (SCP). Более того, можно воспользоваться средствами Conversational AI и подключить элементы машинного обучения.
Исходный код нашего проекта открыт и доступен на Github. Мы постарались оставить подробные инструкции в README файле, на случай если вы захотите запустить нашего бота у себя. Если же у вас возникнут какие-либо замечания или предложения относительно самого кода, то, пожалуйста, напишите issue или откройте pull request. Мы будем рады вашему фидбэку!
Автор - Артемий Геворков, стажёр отдела Co-Innovation Labs, SAP Labs CIS