Для кого эта статья
Предназначено для тех, кто задумывается о цифровизации предприятий. Телеграм-бот - это отличный способ сделать интерфейс между базой данных и сотрудником, у которого нет в данный момент доступа в корпоративную сеть (к своему компьютеру). Безусловно, мобильное приложение решает этот вопрос лучше, но затраты на создание/поддержание мобильного приложения несопоставимы с затратами на простого ТГ-бота (или даже средней сложности), то же относится и ко времени разработки/внедрения. Надеюсь, данная статья будет полезной для энтузиастов широкого внедрения ТГ-ботов как технологии цифровизации промышленных (и не только) предприятий.
Легенда
Мы находимся внутри корпоративной сети предприятия, у нас есть выход в Интернет, но снаружи нет открытых интерфейсов. Есть учетная система предприятия на основе базы данных MS SQL, к ней можно подключиться произвольному клиенту и выполнять запросы в рамках прав, определенных для этого клиента. Наша задача - позволить рабочему сообщить о поломке станка через ТГ-бот (у рабочего нет логина в корпоративную сеть и подключения к учетной системе).
Технологии
Бот будем разрабатывать на python с помощью библиотеки aiogram3, как это описано в серии обучающих статей @yakvenalex, например, https://habr.com/ru/companies/amvera/articles/820527/
Если эта практика цифровизации приживется, ботов будет много; чтобы как-то управлять всем этим и ориентироваться в сервисах, сразу проектируем их в микросервисной архитектуре - упаковываем в docker container, и запускать будем именно конейтенер.
Для демонстрации работоспособности MS SQL Server также запустим в docker container, согласно https://hub.docker.com/r/microsoft/mssql-server. Это просто эмуляция существующей базы данных учетной системы, на этом подробно останавливаться не будем, но поговорим о правах доступа.
Схема работы
Два слова по поводу слоя хранимых процедур, посредством которых бот будет получать и записывать данные в таблицы. Он необходим: 1) как уровень абстракции (чтобы отвязаться от структуры таблиц и иметь возможность ее менять не меняя кода бота); 2) чтобы ограничить права пользователя, под которым бот будет авторизоваться в рабочей базе данных (таким образом, даже при ошибке программиста, бот не сможет испортить произвольные данные в базе)
Подготовка MS SQL - эмуляция рабочей базы данных
Запускаем контейнер с SQL Server
docker run -e 'ACCEPT_EULA=Y' -e 'MSSQL_SA_PASSWORD=<password>' \
-p 1433:1433 \
--name mssql-server \
-v sqlvolume:/var/opt/mssql \
-d mcr.microsoft.com/mssql/server:2022-latestДалее, подключаемся SSMS (или любым другим инструментом) как пользователь sa и выполняем следующий скрипт
-- Стартовый скрипт
create database testdb; -- Тестовая база данных для демонстрации
go;
use testdb;
-- Создаем таблицы для работы бота, возможно, их придется создать
-- и в продуктивной базе
create table bot_user_type( -- типы пользователей
id int identity primary key, -- первичный ключ
user_type varchar(32), -- тип пользователя
);
insert into bot_user_type(user_type)
values (N'Рабочий'), (N'Механик'), (N'Электрик'), (N'Администратор');
create table bot_user( -- пользователи бота
id int identity primary key, -- первичный ключ
user_name varchar(128), -- имя пользователя в ТГ
tg_id bigint, -- ИД в телеграм
f_bot_user_type int foreign key references bot_user_type(id) --тип пользователя
);
insert into bot_user(user_name, f_bot_user_type) -- фейковые пользователи для демонстрации
values(N'Петр Иванов', 2), (N'Василий Кузецов', 3);
create table issue( -- Поломки
id int identity primary key, -- первичный ключ
f_bot_user int foreign key references bot_user(id), -- Подавший заявку
created smalldatetime, -- Дата и время подачи
f_bot_user_type int foreign key references bot_user_type(id), -- 1 - нужен механик, 2 - электрик
f_bot_user_resp int foreign key references bot_user(id), -- ответственный исполнитель
closed smalldatetime -- дата и время закрытия
);
go;
create schema bot; -- специальная схема, в которой будет находиться слой хранимых процедур
go;
create login tgbot with password = '<VeryStrongPassword>'; -- пользователь, под которым бот будет авторизоваться
create user tgbot for login tgbot;
go;
grant execute on schema ::bot to tgbot; -- разрешаем запуск любых хранимых процедур в схеме bot,
-- других прав не даем!
go;Следующий этап - создадим слой хранимых процедур, к которым будет обращаться бот для обмена данными. Слой будет содержать 3 хранимые процедуры - bot.get_or_create_user будет вызываться при регистрации пользователя и возвращать recordset, bot.create_issue будет создавать новую поломку или имитировать назначение ответственного (при повторном вызове того же пользователя) и возвращать recordset, bot.close_all_issues будет закрывать все поломки пользователя и возвращать число закрытых событий
-- Запоминаем в базе данных пользователя ТГ
create procedure bot.get_or_create_user @user_tg_id bigint, -- ИД пользователя ТГ
@user_name varchar(128) -- имя пользователя ТГ
as
begin
set nocount on
if not exists(select 1 from bot_user where tg_id = @user_tg_id) -- если нет такого ID - создаем запись
insert into bot_user(tg_id, user_name, f_bot_user_type)
values (@user_tg_id, @user_name, 1); -- по умолчанию ставим роль Рабочий
select bu.id, bu.tg_id, bu.user_name, ut.user_type
from bot_user bu
join dbo.bot_user_type ut on ut.id = bu.f_bot_user_type
where bu.tg_id = @user_tg_id;
end;
go;
-- Создаем новую поломку, если есть незакрытая для данного пользователя и данного типа - не создаем
create procedure bot.create_issue @user_tg_id bigint, -- ИД пользователя ТГ
@issue_type int -- тип поломки, должен быть 2 или 3
as
begin
set nocount on
if @issue_type not in (2, 3)
throw 50011, N'Неизвестный тип поломки', 1;
declare @last_issue int;
select top 1 @last_issue = i.id -- ищем последнюю открытую поломку у этого пользователя
from issue i
join dbo.bot_user b on i.f_bot_user = b.id
where b.tg_id = @user_tg_id
and i.f_bot_user_type = @issue_type
and closed is null
order by i.id desc
if @@rowcount = 0 -- не нашли
begin
insert into issue(f_bot_user, created, f_bot_user_type)
select u.id, getdate(), @issue_type
from bot_user u
where u.tg_id = @user_tg_id
set @last_issue = scope_identity()
end
else -- есть такая поломка, сымитируем назначение ответственного
begin
update issue
set f_bot_user_resp = (select top 1 id
from bot_user
where bot_user.f_bot_user_type = @issue_type
order by id desc) -- выбираем последнего зарегистрированного механика или электрика
where id = @last_issue
end;
-- Наконец, возвращаем результат
select i.id, i.created, bu.user_name, but.user_type, coalesce(resp.user_name, N'Не назначено') as responsible
from issue i
join dbo.bot_user bu on i.f_bot_user = bu.id
join dbo.bot_user_type but on i.f_bot_user_type = but.id
left join dbo.bot_user resp on resp.id = i.f_bot_user_resp
where i.id = @last_issue
end;
go;
-- Закрываем все открытые поломки для пользователя
create procedure bot.close_all_issues
@user_tg_id bigint
as
begin
set nocount on
update i set i.closed = getdate()
from issue i
join dbo.bot_user bu on i.f_bot_user = bu.id
where bu.tg_id = @user_tg_id and i.closed is null
select @@rowcount -- возвращаем число закрытых поломок
end;
go;Телеграм-бот
Цель статьи - продемонстрировать рабочую связку бота и MS SQL Server, поэтому бот будет примитивный. Задачи бота:
По команде /start зарегистрировать пользователя в БД
По команде /issue спросить причину (механика или электрика) и создать инцидент, при повторном создании инцидента с той же причиной имитировать назначение ответственного сотрудника.
По команде /close закрываем все открытые инциденты для данного инициатора
Создаем бота и получаем токен как описано выше, https://habr.com/ru/companies/amvera/articles/820527/
Постарался уложить код проекта в минимум файлов. Получилось 2:
# bot.py - все, что связано с ботом
import asyncio
import logging
from re import Match
import os
from aiogram import Bot, F, Dispatcher, Router
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import Command
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message
from aiogram.types.bot_command import BotCommand
from dotenv import load_dotenv
from db import DbConn
# Для вывода логов
LOG = logging.getLogger(__name__)
# Читаем .env файл в переменные среды
dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
load_dotenv(dotenv_path)
# Установки программы
BOT_TOKEN = os.getenv("BOT_TOKEN")
SERVICE_CHAT_ID = os.getenv("SERVICE_CHAT_ID")
# Создаем объекты бота
bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(
parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())
router = Router()
# Онлайн-клавиатура при вводе инцидента
def get_inline_keyboard() -> InlineKeyboardMarkup:
keyboard_list = [
[
InlineKeyboardButton(text="Механика", callback_data="issue#2"),
InlineKeyboardButton(text="Электрика", callback_data="issue#3"),
],
[
InlineKeyboardButton(text="Отмена", callback_data="issue#0"),
]
]
return InlineKeyboardMarkup(inline_keyboard=keyboard_list)
# Команды бота
@router.message(Command("start"))
async def start_command(message: Message):
user_info = DbConn.get_or_create_user(
message.from_user.id, message.from_user.full_name)
if isinstance(user_info, dict):
await message.answer(f"😪 Что-то пошло не так... Ошибка {user_info.get('error')}")
else:
await message.answer(f"Привет, {user_info.user_name}, Ваша роль - {user_info.user_type}!")
@router.message(Command("issue"))
async def create_issue_command(message: Message):
await message.answer("Укажите вероятную причину инцидента", reply_markup=get_inline_keyboard())
@router.message(Command("close"))
async def close_issues(message: Message):
count = DbConn.close_all_issues(message.from_user.id)
if count == 0:
await message.answer("Нет открытых инцидентов")
elif count < 0:
await message.answer(f"😪 Что-то пошло не так...")
else:
await message.answer(f"Все в порядке, {count} событи{'е' if count == 1 else 'я'} закрыто")
@router.callback_query(F.data.regexp("^issue#(\d+)").as_("match_type"))
async def choose_issue_type(call: CallbackQuery, match_type: Match[str]):
issue_type = int(match_type.group(1))
# Убираем онлайн-клавиатуру
await call.message.edit_reply_markup(reply_markup=None)
if issue_type == 0:
# Операция отменена
await call.message.answer("Операция отменена 🥶")
else:
# Создаем новый инцидент
issue = DbConn.create_issue(call.from_user.id, issue_type)
if isinstance(issue, dict):
await call.message.answer(
f"😪 Что-то пошло не так... Ошибка - {issue.get('error')}")
else:
await call.message.answer(f" 👩✈️ Инцидент зарегистрирован, время: {issue.created}, нужен: "
f"{issue.user_type}, ответственный: {issue.responsible}, ожидайте.")
async def on_startup():
await bot.set_my_commands(
[
BotCommand(command="start", description="Начать работу с ботом"),
BotCommand(command="issue", description="Открыть инцидент"),
BotCommand(command="close", description="Закрыть все инциденты")
]
)
# Проверим соединение с базой данных
is_db_ready = DbConn.test_connection()
if SERVICE_CHAT_ID:
await bot.send_message(SERVICE_CHAT_ID, f"Бот запущен, соединение с базой данных {'🆗' if is_db_ready else '❌'}")
async def main():
dp.include_router(router)
dp.startup.register(on_startup)
await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
# db.py - связь с базой данных
import os
import logging
import pyodbc
from pyodbc import Cursor
LOG = logging.getLogger(__name__)
class DbConn:
_conn_str = None
@staticmethod
def connection_string() -> str:
if not isinstance(DbConn._conn_str, str):
DbConn._conn_str = "Driver={ODBC Driver 18 for SQL Server};" \
f"Server={os.environ.get('DB_SERVER')};" \
f"Database={os.environ.get('DB_DATABASE')};" \
f"UID={os.environ.get('DB_USER')};" \
f"PWD={os.environ.get('DB_PASSWORD')};" \
f"TrustServerCertificate=yes;"
return DbConn._conn_str
@staticmethod
def test_connection() -> bool:
try:
with pyodbc.connect(DbConn.connection_string()) as conn:
cusror = conn.cursor()
cusror.execute("select @@version")
records = cusror.fetchall()
LOG.warning(records[0])
return len(records) == 1
except BaseException as e:
LOG.error(e)
return False
@staticmethod
def build_records(cursor: Cursor) -> list:
"""
Преобразуем данные к виду списка записей
"""
columns = [column[0] for column in cursor.description]
all_data = cursor.fetchall()
data = []
for rec in all_data:
dct = {columns[i]: v for i, v in enumerate(rec)}
data.append(dct)
return data
@staticmethod
def get_or_create_user(tg_user_id: int, tg_user_name: str) -> dict:
try:
with pyodbc.connect(DbConn.connection_string(), autocommit=True) as conn:
cursor = conn.cursor()
cursor.execute("exec bot.get_or_create_user ?, ?",
tg_user_id, tg_user_name[:128])
return cursor.fetchall()[0]
except BaseException as e:
LOG.error(e)
return {"error": e}
@staticmethod
def create_issue(tg_user_id: int, issue_type: int) -> dict:
try:
with pyodbc.connect(DbConn.connection_string(), autocommit=True) as conn:
cursor = conn.cursor()
cursor.execute("exec bot.create_issue ?, ?", tg_user_id, issue_type)
return cursor.fetchall()[0]
except BaseException as e:
LOG.error(e)
return {"error": e}
@staticmethod
def close_all_issues(tg_user_id: int) -> int:
try:
with pyodbc.connect(DbConn.connection_string(), autocommit=True) as conn:
cursor = conn.cursor()
cursor.execute("exec bot.close_all_issues ?", tg_user_id)
return cursor.fetchall()[0][0]
except BaseException as e:
LOG.error(e)
return -1 Нужен еще файл с настройками переменных окружения, .env
BOT_TOKEN=455656336:AAHCw22T_qiA391bvLYOcSGMzg-4AADD346
SERVICE_CHAT_ID=436568544
DB_SERVER=localhost
DB_DATABASE=testdb
DB_USER=tgbot
DB_PASSWORD=VeryStrongPasswordУпаковка в Docker Container
Во-первых, нужно добавить в проект Dockerfile
FROM --platform=linux/amd64 public.ecr.aws/docker/library/python:3.10-slim-buster
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
ENV PYTHONPATH /
RUN apt-get update && apt-get install -y curl gnupg
RUN sh -c "curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add -"
RUN apt-get update
RUN sh -c "curl https://packages.microsoft.com/config/ubuntu/20.04/prod.list > /etc/apt/sources.list.d/mssql-release.list"
RUN apt-get update
RUN ACCEPT_EULA=Y apt-get install -y msodbcsql18
RUN apt-get install -y netcat gcc unixodbc-dev
COPY . /app
WORKDIR /app
RUN pip install --upgrade pip
RUN pip3 install -r requirements.txt
ENTRYPOINT ["python3", "run.py"]Несколько слов пояснений: базовый образ public.ecr.aws/docker/library/python:3.10-slim-buster получился после некоторого количества экспериментов и попыток установить MS pyodbc в linux-container. С этим образом получились самые стабильные результаты, поэтому рекомендую его.
Строки ## 7-14 взяты с сайта Microsoft - из пособия по установке pyodbc
Все остальное стандартно - просто копируем содержимое проекта в папку /app контейнера, устанавливаем библиотеки и объявляем запускаемым при старте файл run.py
Нужно добавить файл requirements.txt - для корректного выполнения строки #20 предыдущего файла:
aiofiles==24.1.0
aiogram==3.17.0
aiohappyeyeballs==2.4.4
aiohttp==3.11.11
aiosignal==1.3.2
annotated-types==0.7.0
attrs==25.1.0
certifi==2025.1.31
frozenlist==1.5.0
idna==3.10
magic-filter==1.0.12
multidict==6.1.0
propcache==0.2.1
pydantic==2.10.6
pydantic_core==2.27.2
pyodbc==5.2.0
python-dotenv==1.0.1
typing_extensions==4.12.2
yarl==1.18.3Наконец, соберем контейнер, но есть один момент. Так как мы запустили эмулятор MS SQL Server на локальном компьютере, придется в файле .env заменить значение DB_SERVER с localhost на host.docker.internal
Возможно объединить контейнеры MS SQL и бота в одну docker network, тогда значение DB_SERVER нужно будет поменять на имя контейнера с базой данных. Но если ваш сервер - внешний (а на практике так и будет), ничего менять не нужно!
# Собираем контейнер
docker build -t tgbot .
# и запускаем его
docker run --rm -d tgbotНажмите все пункты меню бота и убедитесь, что все работает как надо. Иллюстрация:
Выводы
Проиллюстрирована технология построения интерфейса между пользователем без доступа в корпоративную сеть и корпоративным сервисом (базой данных) посредством Телеграм-бота. Данная технология позволяет построить микросервисную архитектуру ботов на основе контейнеров docker, повысить надежность и управляемость.
