ESP32 門禁感應端 與 電腦管理中控端 透過 MQTT 通訊協定 所架構出的 遠端智慧門禁與 Telegram 雙向控制系統 (2)
優化 Telegram 警報訊息，最直觀且便利的方式是利用 Telegram 的
Inline Code (單行程式碼塊) 語法。
在 Telegram 中，被反引號 ` 包裹的文字，使用者在手機或電腦上點擊該區塊時，系統會自動將內容複製到剪貼簿。這樣管理員收到警報時，只需按一下指令文字，接著在輸入框貼上、補上姓名與部門，即可完成授權！
一、 修改 Python 發送 Telegram 警報邏輯
在 process_swipe() 處理 DENIED: 非法刷卡時，將原本的純文字格式改為包含 Inline Code 快速複製指令
的格式：
修改處（位於 AccessControlApp.process_swipe 內）：
Python
# 1. 收到非法刷卡訊號 (DENIED:)
if
raw_data.startswith("DENIED:"):
uid =
raw_data[7:]
user_name
= get_user_name(uid)
# 寫入歷史日誌
cursor.execute(
"INSERT
INTO access_logs (card_uid, user_name, action,"
"
timestamp) VALUES (?, ?, ?, ?)",
(uid,
user_name, "DENIED (非法刷卡拒絕)",
now_str),
)
# 組裝含有單擊複製指令的警報通知
# 提示：`...` 包裹的內容在 Telegram 點擊即可自動複製！
telegram_msg = (
f"🚨 ⚠️ 偵測到未知非法卡片嘗試開鎖
⚠️\n"
f"卡號 UID: `{uid}`\n"
f"狀態: 拒絕通行 / 警報聲響起\n"
f"時間: {now_str}\n\n"
f"👉 **點擊下方指令快速複製並授權：**\n"
f"`/add
{uid} 姓名 部門`"
)
二、 修改 send_telegram_msg 工具函式 (支援 Markdown 解析)
為了讓 Telegram 能正確解析 ` `/add
UID 姓名 部門` 的複製格式，我們需要在
POST 送出 Payload 時加入 "parse_mode": "Markdown"：
Python
# 通用 Telegram 訊息推播 (加入 parse_mode="Markdown")
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,
"parse_mode":
"Markdown", # 讓 Telegram 支援點擊複製等 Markdown 格式
}
try:
requests.post(url,
json=payload, timeout=5)
except
Exception as e:
print(f"[Telegram
發送失敗]: {e}")
threading.Thread(target=_send, daemon=True).start()
三、 實際 Telegram 警報效果與操作體驗
當 Wokwi 或實體 ESP32 感應到非法卡片（例如 UID 為 99:88:77:66）時，管理員的手機 Telegram 會收到如下訊息：
🚨 ⚠️ 偵測到未知非法卡片嘗試開鎖
⚠️
卡號 UID: 99:88:77:66
狀態: 拒絕通行 / 警報聲響起
時間: 2026-10-31 20:00:00
👉 點擊下方指令快速複製並授權：
/add 99:88:77:66 姓名 部門
📱 管理員快速授權操作步驟：
1.
點擊一下 訊息底部的 /add 99:88:77:66 姓名 部門（手機畫面會提示「已複製到剪貼簿」）。
2.
在 Telegram 輸入框貼上。
3.
將預設的 姓名 與 部門 替換成真實資料（例：/add 99:88:77:66 黃大明 資安部）並送出。
4.
Bot 立刻寫入 SQLite 並透過 MQTT 發送 ADD:99:88:77:66 寫入 ESP32 的 Flash，卡片即刻生效開鎖！
Python程式
import asyncio
import json
import sqlite3
import threading
import tkinter as tk
from datetime import datetime
from tkinter import messagebox, ttk
import paho.mqtt.client as mqtt
import requests
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.ext import (
Application,
CallbackQueryHandler,
CommandHandler,
ContextTypes,
)
# --- Telegram Bot 設定 ---
TELEGRAM_BOT_TOKEN = "8845708831:AAEuRreXFRyv1m--jGK7OCn8MiEsWXqpMT0"
TELEGRAM_CHAT_ID = "7965218469"
global_app = None
# --- MQTT 工具函式 (Broker: broker.emqx.io) ---
def send_mqtt_cmd(topic, payload):
def _pub():
try:
pub_client = mqtt.Client()
pub_client.connect("broker.emqx.io", 1883, 60)
pub_client.publish(topic, payload)
pub_client.disconnect()
print(f"📡 [MQTT 發送成功] Topic: {topic} -> {payload}")
except Exception as e:
print(f"❌ [MQTT 發送失敗]: {e}")
threading.Thread(target=_pub, daemon=True).start()
# --- 通用 Telegram 推播 (支援 Markdown 解析與 InlineKeyboard 按鈕) ---
def send_telegram_msg(message, reply_markup=None):
def _send():
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "Markdown",
}
if reply_markup:
payload["reply_markup"] = reply_markup.to_json()
try:
requests.post(url, json=payload, timeout=5)
except Exception as e:
print(f"[Telegram 發送失敗]: {e}")
threading.Thread(target=_send, daemon=True).start()
# --- 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,
user_name TEXT,
action TEXT,
timestamp TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS inside_cards (
card_uid TEXT PRIMARY KEY,
user_name TEXT,
entry_time TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS users (
card_uid TEXT PRIMARY KEY,
name TEXT NOT NULL,
department TEXT
)
"""
)
# 舊資料庫欄位補齊防護
try:
cursor.execute(
"ALTER TABLE access_logs ADD COLUMN user_name TEXT DEFAULT ''"
)
except sqlite3.OperationalError:
pass
try:
cursor.execute(
"ALTER TABLE inside_cards ADD COLUMN user_name TEXT DEFAULT ''"
)
except sqlite3.OperationalError:
pass
default_users = [
("01:02:03:04", "張小明", "研發部"),
("11:22:33:44", "李美玲", "行政部"),
("55:66:77:88", "王大衛", "工程部"),
("AA:11:22:33", "陳經理", "管理部"),
("BB:44:55:66", "林專員", "業務部"),
]
cursor.executemany(
"INSERT OR IGNORE INTO users (card_uid, name, department) VALUES (?, ?, ?)",
default_users,
)
conn.commit()
conn.close()
init_db()
def get_user_name(card_uid):
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute(
"SELECT name, department FROM users WHERE card_uid = ?", (card_uid,)
)
row = cursor.fetchone()
conn.close()
if row:
return f"{row[0]} ({row[1]})"
return "未註冊卡片/訪客"
def send_full_whitelist_sync():
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute("SELECT card_uid FROM users")
rows = cursor.fetchall()
conn.close()
uid_list = [row[0] for row in rows]
payload_data = {"count": len(uid_list), "uids": uid_list}
json_payload = json.dumps(payload_data)
send_mqtt_cmd("wokwi/rfid/sync_res", json_payload)
print(
f"🔄 [開機全量同步] 已回應 ESP32 白名單，包含 {len(uid_list)} 張卡號。"
)
# =====================================================================
# Telegram 指令與按鈕 Event Handlers
# =====================================================================
async def handle_start_command(
update: Update, _context: ContextTypes.DEFAULT_TYPE
):
welcome_text = (
"🤖 門禁管理機器人已上線！\n\n"
"👉 /status : 查詢場內人員名單與人數\n"
"👉 /add [UID] [姓名] [部門] : 新增白名單成員\n"
"👉 /del [UID] : 刪除白名單成員\n"
"👉 /clear : 清空場內名單"
)
await update.message.reply_text(welcome_text)
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, user_name, entry_time FROM inside_cards ORDER BY entry_time DESC"
)
records = cursor.fetchall()
count = len(records)
conn.close()
reply_msg = f"📊 【門禁場內狀態回報】\n當前場內總人數：{count} 人\n------------------------------------\n"
if count == 0:
reply_msg += "目前場內無任何人員。"
else:
reply_msg += "📋 【在場人員清單】:\n"
for idx, (uid, name, entry_time) in enumerate(records, start=1):
reply_msg += (
f"{idx}. {name}\n UID: {uid}\n 入場時間: {entry_time}\n"
)
await update.message.reply_text(reply_msg)
async def handle_add_command(
update: Update, context: ContextTypes.DEFAULT_TYPE
):
if str(update.effective_chat.id) != str(TELEGRAM_CHAT_ID):
await update.message.reply_text("⛔ 【權限不足】您無法執行此指令！")
return
args = context.args
if len(args) < 2:
await update.message.reply_text(
"⚠️ 【格式錯誤】格式：\n/add [UID] [姓名] [部門(選填)]"
)
return
uid = args[0].strip().upper()
name = args[1].strip()
dept = args[2].strip() if len(args) >= 3 else "未設定"
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute(
"INSERT OR REPLACE INTO users (card_uid, name, department) VALUES (?, ?, ?)",
(uid, name, dept),
)
conn.commit()
conn.close()
send_mqtt_cmd("wokwi/rfid/cmd", f"ADD:{uid}")
if global_app and global_app.root:
global_app.root.after(0, global_app.refresh_ui)
reply_msg = (
f"✅ 【新增成功】\nUID: {uid}\n姓名: {name}\n部門: {dept}\n📡"
" 已同步至 ESP32 Flash。"
)
await update.message.reply_text(reply_msg)
async def handle_del_command(
update: Update, context: ContextTypes.DEFAULT_TYPE
):
if str(update.effective_chat.id) != str(TELEGRAM_CHAT_ID):
await update.message.reply_text("⛔ 【權限不足】您無法執行此指令！")
return
args = context.args
if len(args) < 1:
await update.message.reply_text("⚠️ 【格式錯誤】格式：\n/del [UID]")
return
uid = args[0].strip().upper()
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute("SELECT name FROM users WHERE card_uid = ?", (uid,))
row = cursor.fetchone()
user_name = row[0] if row else "未知人員"
cursor.execute("DELETE FROM users WHERE card_uid = ?", (uid,))
conn.commit()
conn.close()
send_mqtt_cmd("wokwi/rfid/cmd", f"DEL:{uid}")
if global_app and global_app.root:
global_app.root.after(0, global_app.refresh_ui)
reply_msg = (
f"🗑️ 【刪除成功】\nUID: {uid}\n原姓名: {user_name}\n📡"
" 已同步通知 ESP32 移除。"
)
await update.message.reply_text(reply_msg)
async def handle_clear_command(
update: Update, _context: ContextTypes.DEFAULT_TYPE
):
if str(update.effective_chat.id) != str(TELEGRAM_CHAT_ID):
await update.message.reply_text("⛔ 【權限不足】您無法執行此指令！")
return
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")
cursor.execute(
"INSERT INTO access_logs (card_uid, user_name, action, timestamp) VALUES (?, ?, ?, ?)",
("SYSTEM", "系統管理員", f"SYSTEM_CLEAR ({cleared_count}人)", now_str),
)
conn.commit()
conn.close()
reply_msg = f"🧹 【場內名單已清空】\n已清除人數: {cleared_count} 人"
await update.message.reply_text(reply_msg)
if global_app and global_app.root:
global_app.root.after(0, global_app.refresh_ui)
# 處理一鍵授權按鈕點擊事件
async def handle_button_click(
update: Update, _context: ContextTypes.DEFAULT_TYPE
):
query = update.callback_query
await query.answer()
if str(update.effective_chat.id) != str(TELEGRAM_CHAT_ID):
await query.message.reply_text("⛔ 【權限不足】您非管理者！")
return
data = query.data
if data.startswith("AUTH:"):
uid = data[5:]
name = "臨時授權"
dept = "訪客"
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute(
"INSERT OR REPLACE INTO users (card_uid, name, department) VALUES (?, ?, ?)",
(uid, name, dept),
)
conn.commit()
conn.close()
send_mqtt_cmd("wokwi/rfid/cmd", f"ADD:{uid}")
if global_app and global_app.root:
global_app.root.after(0, global_app.refresh_ui)
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
updated_msg = (
f"✅ **【卡片一鍵授權成功】**\n"
f"卡號 UID: `{uid}`\n"
f"授權身份: {name} ({dept})\n"
f"授權時間: {now_str}\n"
f"📡 已同步發送 MQTT 指令通知 ESP32 寫入 Flash！"
)
await query.edit_message_text(text=updated_msg, parse_mode="Markdown")
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("add", handle_add_command))
app.add_handler(CommandHandler("del", handle_del_command))
app.add_handler(CommandHandler("clear", handle_clear_command))
# 註冊 InlineKeyboard 按鈕監聽器
app.add_handler(CallbackQueryHandler(handle_button_click))
print("Telegram Bot 指令與一鍵授權按鈕監聽服務已啟動...")
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())
# =====================================================================
# Tkinter GUI 主介面類別
# =====================================================================
class AccessControlApp:
def __init__(self, root):
self.root = root
self.root.title("RFID 門禁人員監控與 Telegram 系統")
self.root.geometry("820x520")
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", "Name", "EntryTime"),
show="headings",
)
self.inside_tree.heading("UID", text="卡片 UID")
self.inside_tree.heading("Name", text="人員姓名/部門")
self.inside_tree.heading("EntryTime", text="入場時間")
self.inside_tree.column("UID", width=180, anchor="center")
self.inside_tree.column("Name", width=220, anchor="center")
self.inside_tree.column("EntryTime", width=250, 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", "Name", "Action", "Time"),
show="headings",
)
self.log_tree.heading("ID", text="編號")
self.log_tree.heading("UID", text="卡片 UID")
self.log_tree.heading("Name", text="人員姓名")
self.log_tree.heading("Action", text="動作")
self.log_tree.heading("Time", text="記錄時間")
self.log_tree.column("ID", width=50, anchor="center")
self.log_tree.column("UID", width=150, anchor="center")
self.log_tree.column("Name", width=180, anchor="center")
self.log_tree.column("Action", width=140, anchor="center")
self.log_tree.column("Time", width=200, anchor="center")
self.log_tree.pack(fill=tk.BOTH, expand=True)
# ------------------ 分頁 3: 人員卡號管理 ------------------
self.users_frame = ttk.Frame(notebook)
notebook.add(self.users_frame, text="人員卡號管理")
form_frame = ttk.LabelFrame(
self.users_frame, text="新增 / 編輯卡片成員", padding=10
)
form_frame.pack(fill=tk.X, padx=10, pady=5)
ttk.Label(form_frame, text="卡號 UID:").grid(
row=0, column=0, padx=5, pady=5, sticky=tk.W
)
self.uid_entry = ttk.Entry(form_frame, width=20)
self.uid_entry.grid(row=0, column=1, padx=5, pady=5)
ttk.Label(form_frame, text="姓名:").grid(
row=0, column=2, padx=5, pady=5, sticky=tk.W
)
self.name_entry = ttk.Entry(form_frame, width=15)
self.name_entry.grid(row=0, column=3, padx=5, pady=5)
ttk.Label(form_frame, text="部門:").grid(
row=0, column=4, padx=5, pady=5, sticky=tk.W
)
self.dept_entry = ttk.Entry(form_frame, width=15)
self.dept_entry.grid(row=0, column=5, padx=5, pady=5)
btn_add = ttk.Button(
form_frame, text="新增 / 更新人員", command=self.add_or_update_user
)
btn_add.grid(row=0, column=6, padx=10, pady=5)
self.users_tree = ttk.Treeview(
self.users_frame,
columns=("UID", "Name", "Department"),
show="headings",
)
self.users_tree.heading("UID", text="卡片 UID")
self.users_tree.heading("Name", text="姓名")
self.users_tree.heading("Department", text="部門")
self.users_tree.column("UID", width=200, anchor="center")
self.users_tree.column("Name", width=200, anchor="center")
self.users_tree.column("Department", width=200, anchor="center")
self.users_tree.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
action_frame = ttk.Frame(self.users_frame)
action_frame.pack(fill=tk.X, padx=10, pady=5)
btn_delete = ttk.Button(
action_frame, text="刪除選取人員", command=self.delete_user
)
btn_delete.pack(side=tk.RIGHT, padx=5)
self.refresh_ui()
def add_or_update_user(self):
uid = self.uid_entry.get().strip().upper()
name = self.name_entry.get().strip()
dept = self.dept_entry.get().strip()
if not uid or not name:
messagebox.showwarning("輸入錯誤", "卡號 UID 與姓名為必填項目！")
return
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute(
"INSERT OR REPLACE INTO users (card_uid, name, department) VALUES (?, ?, ?)",
(uid, name, dept),
)
conn.commit()
conn.close()
send_mqtt_cmd("wokwi/rfid/cmd", f"ADD:{uid}")
self.uid_entry.delete(0, tk.END)
self.name_entry.delete(0, tk.END)
self.dept_entry.delete(0, tk.END)
self.refresh_ui()
def delete_user(self):
selected = self.users_tree.selection()
if not selected:
messagebox.showwarning("選取錯誤", "請先點選欲刪除的人員項目！")
return
item = self.users_tree.item(selected[0])
uid = item["values"][0]
conn = sqlite3.connect("access_control.db")
cursor = conn.cursor()
cursor.execute("DELETE FROM users WHERE card_uid = ?", (uid,))
conn.commit()
conn.close()
send_mqtt_cmd("wokwi/rfid/cmd", f"DEL:{uid}")
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 = ""
reply_markup = None
# 1. 收到非法刷卡訊號 (DENIED:)
if raw_data.startswith("DENIED:"):
uid = raw_data[7:]
user_name = get_user_name(uid)
cursor.execute(
"INSERT INTO access_logs (card_uid, user_name, action, timestamp) VALUES (?, ?, ?, ?)",
(uid, user_name, "DENIED (非法刷卡拒絕)", now_str),
)
# 內嵌按鈕設定
keyboard = [[
InlineKeyboardButton(
"✅ 一鍵授權新增 (預設訪客)", callback_data=f"AUTH:{uid}"
)
]]
reply_markup = InlineKeyboardMarkup(keyboard)
telegram_msg = (
f"🚨 ⚠️ **偵測到未知非法卡片嘗試開鎖** ⚠️\n"
f"卡號 UID: `{uid}`\n"
f"狀態: 拒絕通行 / 警報聲響起\n"
f"時間: {now_str}\n\n"
f"👉 **快速授權：** 可直接點擊下方按鈕，或複製指令修改姓名：\n"
f"`/add {uid} 姓名 部門`"
)
# 2. 入口合法進場 (AA:)
elif raw_data.startswith("AA:"):
uid = raw_data[3:]
user_name = get_user_name(uid)
cursor.execute(
"INSERT INTO access_logs (card_uid, user_name, action, timestamp) VALUES (?, ?, ?, ?)",
(uid, user_name, "IN (進場)", now_str),
)
cursor.execute(
"INSERT OR REPLACE INTO inside_cards (card_uid, user_name, entry_time) VALUES (?, ?, ?)",
(uid, user_name, now_str),
)
cursor.execute("SELECT COUNT(*) FROM inside_cards")
current_count = cursor.fetchone()[0]
telegram_msg = (
f"🟢 [門禁進場通知]\n"
f"姓名: {user_name}\n"
f"卡號: `{uid}`\n"
f"地點: 入口處 (CARD1)\n"
f"時間: {now_str}\n"
f"📊 當前場內總人數: {current_count} 人"
)
# 3. 出口合法出場 (BB:)
elif raw_data.startswith("BB:"):
uid = raw_data[3:]
user_name = get_user_name(uid)
cursor.execute(
"INSERT INTO access_logs (card_uid, user_name, action, timestamp) VALUES (?, ?, ?, ?)",
(uid, user_name, "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"姓名: {user_name}\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, reply_markup=reply_markup)
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, user_name, entry_time FROM inside_cards ORDER BY entry_time DESC"
)
for row in cursor.fetchall():
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, user_name, action, timestamp FROM access_logs ORDER BY id DESC LIMIT 50"
)
for row in cursor.fetchall():
self.log_tree.insert("", tk.END, values=row)
for item in self.users_tree.get_children():
self.users_tree.delete(item)
cursor.execute(
"SELECT card_uid, name, department FROM users ORDER BY card_uid ASC"
)
for row in cursor.fetchall():
self.users_tree.insert("", tk.END, values=row)
cursor.execute("SELECT COUNT(*) FROM inside_cards")
count = cursor.fetchone()[0]
self.status_label.config(text=f"場內當前總人數: {count} 人")
conn.close()
# --- MQTT 背景監聽服務 ---
def start_mqtt(app):
def on_connect(client, _userdata, _flags, _rc):
print("MQTT 監聽服務已啟動 (Broker: broker.emqx.io)...")
client.subscribe("wokwi/rfid/access")
client.subscribe("wokwi/rfid/sync_req")
def on_message(_client, _userdata, msg):
topic = msg.topic
payload = msg.payload.decode("utf-8").strip()
if topic == "wokwi/rfid/sync_req" and payload == "REQ_ALL":
print("\n📥 收到 ESP32 開機全量白名單同步請求 (REQ_ALL)")
send_full_whitelist_sync()
elif topic == "wokwi/rfid/access":
app.process_swipe(payload)
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.emqx.io", 1883, 60)
client.loop_forever()
if __name__ == "__main__":
send_telegram_msg(
"🚀 【門禁監控後台系統已啟動】\n"
"👉 /status : 查詢場內人數\n"
"👉 /add [UID] [姓名] [部門] : 新增成員與白名單\n"
"👉 /del [UID] : 刪除成員與白名單\n"
"👉 /clear : 清空場內名單"
)
root = tk.Tk()
app = AccessControlApp(root)
global_app = app
threading.Thread(target=start_mqtt, args=(app,), daemon=True).start()
threading.Thread(target=start_telegram_bot_listener, daemon=True).start()
root.mainloop()
