MQTT + SQLite + Tkinter + Telegram 雙向門禁管理後台系統。
它透過多執行緒（Multithreading）同時處理 MQTT 訊息接收、GUI 視窗介面更新、SQLite 資料庫讀寫，以及 Telegram Bot 的推播與指令監聽。
一、 系統四大核心模組與作用
┌── [1. SQLite 資料庫] (儲存 access_logs / inside_cards)
│
[ Wokwi ESP32 ] ──► [2. MQTT 接收執行緒] ──► [3. Tkinter GUI 介面] (即時顯示人數與清單)
│
└── [4. Telegram 雙向互動模組]
├── 主動推播：進出場通知 & 系統開機提示
└── 反向指令：接收 /status 與 /clear 指令
二、 程式碼逐區功能解析
1. Telegram 主動發送模組 (send_telegram_msg)
作用：利用 HTTP POST 發送 Telegram 訊息。
特點：使用獨立背景 Thread (
threading.Thread) 執行，即使 Telegram 網路連線延遲或斷線，也完全不會造成 Tkinter GUI 視窗凍結或卡頓。
2. Telegram 反向指令處理器 (handle_status_command & handle_clear_command)
/status指令：收到指令時，連線至 SQLite 查詢inside_cards資料表，格式化整理後回傳「目前場內總人數」與「所有在場卡片 UID 與入場時間名單」。/clear指令：清空 SQLite 的
inside_cards資料表（將場內人數歸零）。在
access_logs寫入一筆SYSTEM_CLEAR歷史日誌（記錄清場時間與當時被清除的人數）。透過
global_app.root.after()跨 Thread 即時更新 Tkinter GUI 視窗介面，並回覆 Telegram 操作結果。
3. Telegram 指令監聽服務 (start_telegram_bot_listener)
作用：在背景開啟獨立的
asyncio事件迴圈，啟動python-telegram-bot的 Polling 服務，持續監聽使用者在 Telegram 輸入的/start、/status與/clear指令。
4. SQLite 資料庫模組 (init_db)
作用：自動建立並管理
access_control.db本地資料庫，包含兩張資料表：access_logs（歷史刷卡日誌）：記錄所有刷卡細節（流水號、卡號 UID、動作IN/OUT、時間戳記）。inside_cards（場內動態清單）：僅保留當前人在場內的卡片 UID 與入場時間。
5. Tkinter 視覺化 GUI 介面 (AccessControlApp)
作用：提供直觀的電腦後台管理視窗，分為：
頂部狀態列：醒目顯示「場內當前總人數：X 人」。
「場內卡號清單」頁籤：即時條列當前場內的所有卡片與入場時間。
「歷史進出紀錄」頁籤：列出最近 50 筆進出紀錄。
process_swipe()核心邏輯：當收到 MQTT 刷卡封包（如AA:11:22:33或BB:44:55:66）時：自動判斷進/出場並更新 SQLite。
計算最新場內人數。
發送 Telegram 帶有精確時間與人數的推播通知。
排程刷新 Tkinter UI 表格。
6. Background MQTT 監聽服務 (start_mqtt)
作用：連線至公共 MQTT Broker (
broker.hivemq.com)，訂閱wokwi/rfid/access主題，即時接收 ESP32 傳送過來的刷卡資料。
三、 系統特色與優勢
完全非阻塞（Non-blocking）：MQTT 接收、GUI 繪製與 Telegram 網路傳輸各自獨立在不同的 Thread 運作，系統回應非常順暢。
雙向互動能力：不只能接收 ESP32 刷卡自動通知 Telegram，管理者還能隨時透過 Telegram 發送
/status查詢現況或/clear遠端清場。資料持久化與可追溯性：即使後台程式重啟，SQLite 資料庫仍完整保存場內狀態與所有歷史門禁日誌。
WOKWI ESP32程式
Python程式
# --- Telegram Bot 設定 (請填入您的 Token 與 Chat ID) ---
TELEGRAM_BOT_TOKEN = "882452708831:AAEuRreXFRyv1m--jGK7OCn8M1iEsWXqpMT0" # // 請替換為您的 Bot Token
TELEGRAM_CHAT_ID = "792652184679" # // 請替換為您的 Chat ID
下面程式碼需修改成自己的 TOKEN權杖與CHAT ID
import asyncio
import sqlite3
import threading
import tkinter as tk
from datetime import datetime
from tkinter import ttk
import paho.mqtt.client as mqtt
import requests
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
# --- Telegram Bot 設定 (請填入您的 Token 與 Chat ID) ---
TELEGRAM_BOT_TOKEN = "882452708831:AAEuRreXFRyv1m--jGK7OCn8M1iEsWXqpMT0" # // 請替換為您的 Bot Token
TELEGRAM_CHAT_ID = "792652184679" # // 請替換為您的 Chat ID
# 全域參考變數，供 Telegram 指令背景 Thread 安全呼叫 refresh_ui
global_app = None
# 1. 刷卡時主動發送 Telegram 訊息 (背景 Thread)
def send_telegram_msg(message):
def _send():
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message}
try:
requests.post(url, json=payload, timeout=5)
except Exception as e:
print(f"[Telegram 發送失敗]: {e}")
threading.Thread(target=_send, daemon=True).start()
# 2. Telegram /status 指令處理函式 (反向查詢 SQLite)
async def handle_status_command(
update: Update, context: ContextTypes.DEFAULT_TYPE
):
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
# 查詢目前在場內的所有卡片與進場時間
cursor.execute(
"SELECT card_uid, entry_time FROM inside_cards ORDER BY entry_time DESC"
)
records = cursor.fetchall()
count = len(records)
conn.close()
# 組裝回覆訊息
reply_msg = f"📊 【門禁場內狀態回報】\n"
reply_msg += f"當前場內總人數：{count} 人\n"
reply_msg += "------------------------------------\n"
if count == 0:
reply_msg += "目前場內無任何人員/卡片。"
else:
reply_msg += "📋 【在場卡號清單】:\n"
for idx, (uid, entry_time) in enumerate(records, start=1):
reply_msg += f"{idx}. UID: {uid}\n 入場時間: {entry_time}\n"
# 回覆 Telegram 使用者
await update.message.reply_text(reply_msg)
# 3. Telegram /clear 指令處理函式 (清空 SQLite 場內名單並刷新 GUI)
async def handle_clear_command(
update: Update, context: ContextTypes.DEFAULT_TYPE
):
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
# 統計清場前的數量
cursor.execute("SELECT COUNT(*) FROM inside_cards")
cleared_count = cursor.fetchone()[0]
# 清空場內名單
cursor.execute("DELETE FROM inside_cards")
# 於歷史紀錄下記錄系統清場事件
log_msg = f"SYSTEM_CLEAR (管理員 Telegram 手動清場 {cleared_count} 人)"
cursor.execute(
"INSERT INTO access_logs (card_uid, action, timestamp) VALUES (?, ?, ?)",
("SYSTEM", log_msg, now_str),
)
conn.commit()
conn.close()
reply_msg = (
f"🧹 【門禁場內名單已清空】\n"
f"執行時間: {now_str}\n"
f"已清除卡片數: {cleared_count} 張\n"
f"當前場內人數已重置為 0 人。"
)
await update.message.reply_text(reply_msg)
# 跨 Thread 即時更新 Tkinter GUI
if global_app and global_app.root:
global_app.root.after(0, global_app.refresh_ui)
# 4. Telegram /start 指令歡迎訊息
async def handle_start_command(
update: Update, context: ContextTypes.DEFAULT_TYPE
):
welcome_text = (
"🤖 門禁管理機器人已上線！\n\n"
"您可以使用以下指令：\n"
"👉 /status : 查詢當前場內總人數與卡號名單\n"
"👉 /clear : 手動清空場內名單並重置人數"
)
await update.message.reply_text(welcome_text)
# 5. 在背景啟動 Telegram Bot Polling (使用獨立 asyncio 事件迴圈)
def start_telegram_bot_listener():
async def run_bot():
app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
# 註冊指令處理器
app.add_handler(CommandHandler("start", handle_start_command))
app.add_handler(CommandHandler("status", handle_status_command))
app.add_handler(CommandHandler("clear", handle_clear_command))
print("Telegram Bot 指令監聽服務已啟動 (/status 與 /clear 就緒)...")
await app.initialize()
await app.start()
await app.updater.start_polling()
# 保持事件迴圈運作
while True:
await asyncio.sleep(3600)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(run_bot())
# --- SQLite 資料庫初始化 ---
def init_db():
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS access_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
card_uid TEXT,
action TEXT,
timestamp TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS inside_cards (
card_uid TEXT PRIMARY KEY,
entry_time TEXT
)
"""
)
conn.commit()
conn.close()
init_db()
# --- Tkinter GUI 主介面類別 ---
class AccessControlApp:
def __init__(self, root):
self.root = root
self.root.title("RFID 門禁監控與 Telegram 雙向系統")
self.root.geometry("750x450")
self.status_label = tk.Label(
root,
text="場內當前總人數: 0 人",
font=("Helvetica", 16, "bold"),
bg="#2c3e50",
fg="white",
pady=10,
)
self.status_label.pack(fill=tk.X)
notebook = ttk.Notebook(root)
notebook.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
# 分頁 1: 當前場內卡號
self.inside_frame = ttk.Frame(notebook)
notebook.add(self.inside_frame, text="場內卡號清單")
self.inside_tree = ttk.Treeview(
self.inside_frame,
columns=("UID", "EntryTime"),
show="headings",
)
self.inside_tree.heading("UID", text="卡片 UID")
self.inside_tree.heading("EntryTime", text="入場時間")
self.inside_tree.column("UID", width=250, anchor="center")
self.inside_tree.column("EntryTime", width=350, anchor="center")
self.inside_tree.pack(fill=tk.BOTH, expand=True)
# 分頁 2: 歷史進出日誌
self.log_frame = ttk.Frame(notebook)
notebook.add(self.log_frame, text="歷史進出紀錄")
self.log_tree = ttk.Treeview(
self.log_frame,
columns=("ID", "UID", "Action", "Time"),
show="headings",
)
self.log_tree.heading("ID", text="編號")
self.log_tree.heading("UID", text="卡片 UID")
self.log_tree.heading("Action", text="動作")
self.log_tree.heading("Time", text="記錄時間")
self.log_tree.column("ID", width=60, anchor="center")
self.log_tree.column("UID", width=200, anchor="center")
self.log_tree.column("Action", width=120, anchor="center")
self.log_tree.column("Time", width=250, anchor="center")
self.log_tree.pack(fill=tk.BOTH, expand=True)
self.refresh_ui()
def process_swipe(self, raw_data):
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
telegram_msg = ""
if raw_data.startswith("AA:"): # 入口
uid = raw_data[3:]
cursor.execute(
"INSERT INTO access_logs (card_uid, action, timestamp) VALUES (?, ?, ?)",
(uid, "IN (進場)", now_str),
)
cursor.execute(
"INSERT OR REPLACE INTO inside_cards (card_uid, entry_time) VALUES (?, ?)",
(uid, now_str),
)
cursor.execute("SELECT COUNT(*) FROM inside_cards")
current_count = cursor.fetchone()[0]
telegram_msg = (
f"🟢 [門禁進場通知]\n"
f"卡號: {uid}\n"
f"地點: 入口處 (CARD1)\n"
f"時間: {now_str}\n"
f"📊 當前場內總人數: {current_count} 人"
)
elif raw_data.startswith("BB:"): # 出口
uid = raw_data[3:]
cursor.execute(
"INSERT INTO access_logs (card_uid, action, timestamp) VALUES (?, ?, ?)",
(uid, "OUT (出場)", now_str),
)
cursor.execute(
"DELETE FROM inside_cards WHERE card_uid = ?", (uid,)
)
cursor.execute("SELECT COUNT(*) FROM inside_cards")
current_count = cursor.fetchone()[0]
telegram_msg = (
f"🔴 [門禁出場通知]\n"
f"卡號: {uid}\n"
f"地點: 出口處 (CARD2)\n"
f"時間: {now_str}\n"
f"📊 當前場內總人數: {current_count} 人"
)
conn.commit()
conn.close()
if telegram_msg:
send_telegram_msg(telegram_msg)
self.root.after(0, self.refresh_ui)
def refresh_ui(self):
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
for item in self.inside_tree.get_children():
self.inside_tree.delete(item)
cursor.execute(
"SELECT card_uid, entry_time FROM inside_cards ORDER BY entry_time DESC"
)
inside_records = cursor.fetchall()
for row in inside_records:
self.inside_tree.insert("", tk.END, values=row)
for item in self.log_tree.get_children():
self.log_tree.delete(item)
cursor.execute(
"SELECT id, card_uid, action, timestamp FROM access_logs ORDER BY id DESC LIMIT 50"
)
for row in cursor.fetchall():
self.log_tree.insert("", tk.END, values=row)
count = len(inside_records)
self.status_label.config(text=f"場內當前總人數: {count} 人")
conn.close()
# --- MQTT 背景服務 ---
def start_mqtt(app):
def on_connect(client, userdata, flags, rc):
print("MQTT 監聽服務已啟動...")
client.subscribe("wokwi/rfid/access")
def on_message(client, userdata, msg):
payload = msg.payload.decode("utf-8")
app.process_swipe(payload)
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.hivemq.com", 1883, 60)
client.loop_forever()
if __name__ == "__main__":
# 啟動時推播開機簡介
send_telegram_msg(
"🚀 【門禁監控後台系統已啟動】\n"
"👉 /status : 查詢場內人數與卡號名單\n"
"👉 /clear : 手動清空場內名單"
)
root = tk.Tk()
app = AccessControlApp(root)
global_app = app # 綁定全域參考供 handle_clear_command 呼叫
# 1. 啟動 MQTT 接收背景執行緒
mqtt_thread = threading.Thread(target=start_mqtt, args=(app,), daemon=True)
mqtt_thread.start()
# 2. 啟動 Telegram Bot /status 與 /clear 指令監聽背景執行緒
telegram_thread = threading.Thread(
target=start_telegram_bot_listener, daemon=True
)
telegram_thread.start()
# 3. 運行 GUI 主迴圈
root.mainloop()
- 在 Telegram 搜尋並加入官方帳號 @BotFather。 [1]
- 傳送指令
/newbot。 [1] - 依照提示依序輸入：
- 顯示名稱（例如：我的通知機器人）
- 使用者名稱（必須以
bot結尾，例如：my_alert_bot） [1]
- BotFather 會回傳一組類似
123456789:ABCdefGhIJKlmNoPQRsTUVwxYZ1234567890的字串，這就是你的 Bot Token（請妥善保管，勿公開）。 [1] - 搜尋剛建立好的 Bot 名稱並點選 「開始 / Start」，隨便傳送一封訊息（不先傳訊息機器人將無法主動發訊給你）。 [1]
- 個人 Chat ID： 在 Telegram 搜尋並啟動 @userinfobot 或 @WhatChatIDBot，機器人回傳的
Id數字即為你的個人 Chat ID。 [1, 2]
getUpdates) 取得 [1]- 確保你已經向自己的 Bot 傳送過一則測試訊息。
- 在瀏覽器網址列輸入以下網址（將
YOUR_TOKEN替換為你的 Bot Token）：https://api.telegram.org/bot{YOUR_TOKEN}/getUpdates
https://api.telegram.org/bot{YOUR_TOKEN}/sendMessage?chat_id={CHAT_ID}&text=Hello%20World [1]參考 https://ahha.com.tw/blog/telegram-bot-setup
