Hướng dẫn xây dựng chatbot truy vấn dữ liệu trên Telegram bằng Python

pip install psycopg2
pip install python-telegram-bot
pip install python-dotenv

DB_HOST=
DB_PORT=
DB_NAME=
DB_USER=
DB_PASSWORD=
#Thông tin kết nối Postgresql, không cần đặt trong ""

BOT_TOKEN=
#API Token của chatbot được botfather telegram cung cấp khi tạo

import logging
#Được sử dụng để ghi lại thông tin hoạt động của ứng dụng.

import psycopg2
#Dùng để kết nối và tương tác với cơ sở dữ liệu PostgreSQL.

from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
#Update: Chứa thông tin về một cập nhật từ người dùng trên Telegram. Nó bao gồm thông tin về người gửi, tin nhắn

#InlineKeyboardButton & InlineKeyboardMarkup: Nút và bàn phím tương tác của chatbot ( ví dụ click tùy chọn )

from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackQueryHandler
#Updater: Quản lý các bộ xử lý cho các cập nhật từ Telegram. Nó được sử dụng để thiết lập kết nối với Telegram APi

#CommandHandler: Được sử dụng để xử lý các lệnh được gửi đến bot từ người dùng trên Telegram.

#MessageHandler: Được sử dụng để xử lý các tin nhắn văn bản từ người dùng trên Telegram.

#Filters: Cung cấp các bộ lọc để kiểm tra và lọc các loại tin nhắn, cho phép bạn chỉ xử lý các tin nhắn thỏa mãn các điều kiện nhất định.

#CallbackQueryHandler: Được sử dụng để xử lý các sự kiện callback từ người dùng, chẳng hạn như khi người dùng nhấn vào các nút tương tác.

import os
#Tương tác với hệ điều hành, đường dẫn file, biến môi trường.

from dotenv import load_dotenv
#Được sử dụng để tải các biến môi trường từ tệp tin .env, cho phép bạn lưu trữ thông tin nhạy cảm như API token một cách bảo mật.

import threading
#Cung cấp các phương thức để tạo và quản lý các luồng (threads), cho phép ứng dụng của bạn thực hiện các tác vụ đồng thời.

from datetime import datetime
#Làm việc với ngày và giờ.

env_path = 'C:/Users/admin/...'
#Đường dẫn đến file .env chứa thông tin kết nối

load_dotenv(dotenv_path=env_path)
DB_HOST = os.getenv('DB_HOST')
DB_PORT = os.getenv('DB_PORT')
DB_NAME = os.getenv('DB_NAME')
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')

BOT_TOKEN = os.getenv('BOT_TOKEN')

#Kịch bản mong muốn của mình như sau, khi chat 1 ký tự bất kỳ cho chatbot thì chatbot sẽ phản hồi lại 2 câu lệnh để lựa chọn

COMMAND_1 = 'total_revenue'
#Khi click chọn câu lệnh này, chatbot sẽ hỏi tiếp channel muốn xem doanh thu ( online hay offline, chúng ta sẽ xử lý sau ) và sau đó điền tiếp khoảng thời gian muốn xem ( gồm ngày bắt đầu và ngày kết thúc )

COMMAND_2 = 'customer_order'
#Khi click chọn command này và cung cấp số điện thoại thì sẽ trả ra 10 đơn hàng gần nhất

#Ngoài ra các bạn có thể tạo thêm nhiều câu command khác tùy mục dích sử dụng như truy vấn thông tin khách hàng, truy vấn sản phẩm...

QUERY_1 = "select sum(total_amount) as revenue from customer_bill where completed_date::date between %s and %s and channel = %s"
QUERY_2 = "select * from customer_bill where phone = %s limit 10"

# Thiết lập logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

# Hàm kết nối PostgreSQL
def connect_database():
    return psycopg2.connect(host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASSWORD)

# Hàm đóng kết nối PostgreSQL
def close_database(connection):
    connection.close()

# Kết nối PostgreSQL
conn = connect_database()

# Hàm đóng kết nối cơ sở dữ liệu sau 60s nếu không có tương tác từ người dùng
def close_database_after_timeout(conn, timeout):
    threading.Timer(timeout, close_database, args=[conn]).start()

# Hàm xử lý command /start
# Khi người dùng lần đầu tiên truy cập vào chatbot hoặc dùng câu lệnh /start 
def start(update: Update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="Xin chào, tôi là trợ lý của bạn, bạn cần gì?")
    close_database(conn)
    context.user_data.clear()
#Khi gõ command /start thì chatbot sẽ phản hồi nội dung như trên, đồng thời xóa mọi dữ liệu người dùng đã lưu trữ trong user_data và đóng hết các kết nối và lệnh thực thi đang treo ( ví dụ đang truy vấn nửa chừng xong thoát không hỏi nữa ) hoặc có bug thì có thể clear hết bắt đầu hỏi lại

def customer_order(update: Update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="Số điện thoại cần kiểm tra")
    context.user_data['next_handler'] = 'check_customer_order'
#Khi người dùng chọn command /customer_order thì chatbot sẽ hỏi tiếp với nội dung như trên nhằm lấy tiếp số điện thoại để thực thi câu truy vấn QUERY_2 ở trên ( phone = %s thì %s chính là số điện thoại người dùng nhập vào )

 #Hàm xử lý command /total_revenue
 def total_revenue(update: Update, context):
    keyboard = [
        [InlineKeyboardButton("STORE", callback_data='STORE')],
        [InlineKeyboardButton("ONLINE", callback_data='ONLINE')]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    context.bot.send_message(chat_id=update.effective_chat.id, text="Vui lòng chọn tùy chọn", reply_markup=reply_markup)
    context.user_data['next_handler'] = 'handle_total_revenue_option'
  #Ở đây chúng ta sẽ thấy khác 1 chút so với ở trên vì bước này mình muốn hiện ra kết quả để người dùng chọn luôn thay vì input ( do chỉ có 2 tùy chọn là STORE và ONLINE còn ở trên thì phải nhập số điện thoại do có rất nhièu số điện thoại trong database )


   # Hàm xử lý callback từ InlineKeyboard
   def handle_callback(update: Update, context):
    query = update.callback_query
    chat_id = query.message.chat_id
    callback_data = query.data

    if callback_data == 'STORE' or callback_data == 'ONLINE':
        context.user_data['total_revenue_option'] = callback_data
        context.bot.send_message(chat_id=chat_id, text="Nhập ngày bắt đầu (YYYY-MM-DD)")
        context.user_data['next_handler'] = 'handle_total_revenue_start_date'
    else:
        response = "Tùy chọn không hợp lệ."
        context.bot.send_message(chat_id=chat_id, text=response)
   #Sau khi người dùng chọn button "STORE" & "ONLINE" thì chatbot sẽ hỏi tiếp "Nhập ngày bắt đầu (YYYY-MM-DD)"

def execute_query(query, params=None):
    connection = connect_database()
    cur = connection.cursor()
    cur.execute(query, params)
    results = cur.fetchall()
    cur.close()
    close_database(connection)
    return results, cur.description

def is_valid_date(date_string):
    try:
        datetime.strptime(date_string, '%Y-%m-%d')
        return True
    except ValueError:
        return False
#Dùng để kiểm tra format thời gian khách hàng nhập vào

def format_query_result(result, field_names):
    formatted_result = '\n'.join(f"{field_names[i]}: {format_number(field, field_names[i])}" for i, field in enumerate(result))
    return formatted_result
#Dùng để tách kết quả trả lời của chatbot thành từng dòng, mỗi dòng sẽ là field + giá trị, nếu không có hàm này thì kết quả sẽ thành 1 chuỗi văn bản dài rất khó đọc ( ví dụ: customer A dob 2023-01-01 name D city HCM )

def format_number(number, field_name):
    if field_name == 'revenue':
        return "{:,.0f}".format(number)
    return str(number)
    #Dùng để  chỉnh số thập phân cho total revenue ( ví dụ từ 1000000 sẽ thành 1,000,000 )

def send_query_results(results, chat_id, context):
    field_names = [desc[0] for desc in results[1]]
    has_results = False
    for result in results[0]:
        formatted_result = format_query_result(result, field_names)
        context.bot.send_message(chat_id=chat_id, text=formatted_result)
        has_results = True

    if not has_results:
        response = "Không tìm thấy kết quả."
        context.bot.send_message(chat_id=chat_id, text=response)

def handle_text(update: Update, context):
    user_input = update.message.text
    chat_id = update.effective_chat.id

    if context.user_data.get('next_handler') == 'check_customer_order':
        phone = user_input.strip()
        results = execute_query(QUERY_2, (phone,))
        send_query_results(results, chat_id, context)
        del context.user_data['next_handler']
        close_database(conn)
    elif context.user_data.get('next_handler') == 'handle_total_revenue_option':
        context.bot.send_message(chat_id=chat_id, text="Vui lòng chọn tùy chọn hợp lệ (STORE hoặc ONLINE)")
    elif context.user_data.get('next_handler') == 'handle_total_revenue_start_date':
        start_date = user_input.strip()
        if not is_valid_date(start_date):
            context.bot.send_message(chat_id=chat_id, text="Lỗi sai định dạng ngày. Vui lòng nhập lại (YYYY-MM-DD)")
            return
        context.user_data['total_revenue_start_date'] = start_date
        context.bot.send_message(chat_id=chat_id, text="Nhập ngày kết thúc (YYYY-MM-DD)")
        context.user_data['next_handler'] = 'handle_total_revenue_end_date'
    elif context.user_data.get('next_handler') == 'handle_total_revenue_end_date':
        end_date = user_input.strip()
        if not is_valid_date(end_date):
            context.bot.send_message(chat_id=chat_id, text="Lỗi sai định dạng ngày. Vui lòng nhập lại (YYYY-MM-DD)")
            return
        start_date = context.user_data['total_revenue_start_date']
        total_revenue_option = context.user_data['total_revenue_option']
        results = execute_query(QUERY_1, (start_date, end_date, total_revenue_option))
        send_query_results(results, chat_id, context)
        del context.user_data['next_handler']
        del context.user_data['total_revenue_start_date']
        del context.user_data['total_revenue_option']
        close_database(conn)
    else:
        response = "Vui lòng chọn\n\n/customer_order : Kiểm tra lịch sử 10 đơn hàng gần nhất của khách hàng\n\n/total_revenue : Tính tổng doanh thu"
        context.bot.send_message(chat_id=chat_id, text=response)
        
#Hàm này hiểu đơn giản như sau nếu người dùng chọn /customer_order ở chatbot thì next_handler = "check_customer_order" ( xem lại def customer_order ở trên ). Sau đó sẽ thực hiện câu lệnh QUERY_2 để truy vấn dữ liệu và thực hiện 1 hàm send_query_result để gửi kết quả ra khung chat telegram và đóng kết nối cơ sở dữ liệu

#Nếu người dùng chọn command /total_revenue thì chatbot sẽ hỏi tiếp "Vui lòng chọn tùy chọn hợp lệ (STORE hoặc ONLINE)" dưới dạng button, sau khi chọn thì chatbot lại hỏi tiếp ngày bắt đầu sau đó là ngày kết thúc, nếu thời gian nhập không đúng định dạng của db thì sẽ báo "Lỗi sai định dạng ngày. Vui lòng nhập lại (YYYY-MM-DD)"

#Nếu người dùng gõ 1 ký tự bất kỳ thì chatbot sẽ phản hồi với nội dung như sau
#Vui lòng chọn

#/customer_order: Kiểm tra lịch sử 10 đơn hàng gần nhất của khách hàng

#/total_revenue : Tính tổng doanh thu
#-> Mục đích để người dùng có thể chọn nhanh command từ gợi ý để click chọn mà không phải ghi nhớ command gì

def main():
    updater = Updater(token=BOT_TOKEN, use_context=True)

    dispatcher = updater.dispatcher

    start_handler = CommandHandler('start', start)
    dispatcher.add_handler(start_handler)

    customer_order_handler = CommandHandler(COMMAND_2, customer_order)
    dispatcher.add_handler(customer_order_handler)

    total_revenue_handler = CommandHandler(COMMAND_1, total_revenue)
    dispatcher.add_handler(total_revenue_handler)

    text_handler = MessageHandler(Filters.text & (~Filters.command), handle_text)
    dispatcher.add_handler(text_handler)

    callback_handler = CallbackQueryHandler(handle_callback)
    dispatcher.add_handler(callback_handler)

    updater.start_polling()

    close_database_after_timeout(conn, 60)

    updater.idle()

if __name__ == '__main__':
    main()