En el mundo moderno, los servicios de entrega son cada vez más populares y demandados, por lo que cualquier posibilidad de automatización en esta área será de gran beneficio tanto para las empresas como para los usuarios. En artículos anteriores de nuestro blog, hablamos sobre el uso de la visión artificial y las redes neuronales para reconocer las etiquetas de precios de los productos en una tienda, así como para reconocer los componentes. En este artículo, hablaremos de una tarea menos ambiciosa (pero no menos interesante): automatizar la notificación al cliente del estado de sus pedidos mediante un chatbot de Telegram, códigos QR y el DBMS relacional 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. , , . - . . - , .
, . , – SAP Cloud Platform (SCP). , Conversational AI .
Github. README , . - , , , issue pull request. !
- , Co-Innovation Labs, SAP Labs CIS