2026年10月30日 星期五

MQTT + SQLite + Tkinter + Telegram 雙向門禁管理後台系統

 MQTT + SQLite + Tkinter + Telegram 雙向門禁管理後台系統





#include <WiFi.h>
#include <PubSubClient.h>
#include <SPI.h>
#include <MFRC522.h>


// Wi-Fi 與 MQTT 伺服器設定 (Wokwi 虛擬 Wi-Fi)
const char* ssid = "Wokwi-GUEST";
const char* password = "";
const char* mqtt_server = "broker.hivemq.com";
const char* mqtt_topic = "wokwi/rfid/access";

WiFiClient espClient;
PubSubClient client(espClient);

// RFID 腳位定義
#define SS_1_PIN   5
#define RST_1_PIN 21
#define SS_2_PIN   4
#define RST_2_PIN 22

MFRC522 rfid1(SS_1_PIN, RST_1_PIN); // CARD1 (入口)
MFRC522 rfid2(SS_2_PIN, RST_2_PIN); // CARD2 (出口)

void setup_wifi() {
  delay(10);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println("\nWiFi 已連線");
}

void reconnect() {
  while (!client.connected()) {
    String clientId = "ESP32-RFID-" + String(random(0xffff), HEX);
    if (client.connect(clientId.c_str())) {
      Serial.println("MQTT 已連線");
    } else {
      delay(2000);
    }
  }
}

void readAndPublish(MFRC522 &mfrc522, String prefix) {
  if (!mfrc522.PICC_IsNewCardPresent() || !mfrc522.PICC_ReadCardSerial()) {
    return;
  }

  String fullCode = prefix;
  for (byte i = 0; i < mfrc522.uid.size; i++) {
    if (i > 0) fullCode += ":";
    if (mfrc522.uid.uidByte[i] < 0x10) fullCode += "0";
    fullCode += String(mfrc522.uid.uidByte[i], HEX);
  }
  fullCode.toUpperCase();

  Serial.print("刷卡發送: ");
  Serial.println(fullCode);

  // 發送卡號至 MQTT 主題
  client.publish(mqtt_topic, fullCode.c_str());

  mfrc522.PICC_HaltA();
  mfrc522.PCD_StopCrypto1();
  delay(1000);
}

void setup() {
  Serial.begin(115200);
  SPI.begin();

  setup_wifi();
  client.setServer(mqtt_server, 1883);

  rfid1.PCD_Init();
  rfid2.PCD_Init();
  Serial.println("雙 MFRC522 + MQTT 系統已啟動");
}

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();

  readAndPublish(rfid1, "AA:"); // 入口
  readAndPublish(rfid2, "BB:"); // 出口

  delay(20);
}

import sqlite3
import threading
import tkinter as tk
from datetime import datetime
from tkinter import ttk
import paho.mqtt.client as mqtt
import requests

# --- Telegram Bot 設定 (請替換為您的 Token 與 Chat ID) ---
# TELEGRAM_BOT_TOKEN = "YOUR_TELEGRAM_BOT_TOKEN"
# TELEGRAM_CHAT_ID = "YOUR_TELEGRAM_CHAT_ID"
TELEGRAM_BOT_TOKEN = "88415708831:AAEuRreXFRyv1m--jGK7OC1n8MiEsWXqpMT0"  #// 請替換為您的 Bot Token
TELEGRAM_CHAT_ID = "791652218169"    #// 請替換為您的 Chat ID

# 發送 Telegram 訊息的函式
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}")

    # 使用背景 Thread 發送，避免 Telegram 網路延遲阻塞 Python 主程式/GUI
    threading.Thread(target=_send, daemon=True).start()


# --- 1. 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()


# --- 2. 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()

    # 處理卡片刷卡邏輯與 Telegram 發送
    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:"):  # 入口 (CARD1)
            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:"):  # 出口 (CARD2)
            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()

        # 發送 Telegram 通知
        if telegram_msg:
            send_telegram_msg(telegram_msg)

        # 更新 GUI 畫面
        self.root.after(0, self.refresh_ui)

    # 刷新表格與統計人數
    def refresh_ui(self):
        conn = sqlite3.connect("access_control.db")
        cursor = conn.cursor()

        # 1. 刷新場內表格
        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)

        # 2. 刷新歷史紀錄
        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)

        # 3. 更新人數計數
        count = len(inside_records)
        self.status_label.config(text=f"場內當前總人數: {count} 人")

        conn.close()


# --- 3. MQTT 接收執行緒 ---
def start_mqtt(app):
    def on_connect(client, userdata, flags, rc):
        print("已成功連線至 MQTT Broker，監聽門禁事件中...")
        client.subscribe("wokwi/rfid/access")

    def on_message(client, userdata, msg):
        payload = msg.payload.decode("utf-8")
        print(f"收到 MQTT 刷卡紀錄: {payload}")
        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__":
    root = tk.Tk()
    app = AccessControlApp(root)

    # 將 MQTT 服務放在背景 Thread 執行
    mqtt_thread = threading.Thread(target=start_mqtt, args=(app,), daemon=True)
    mqtt_thread.start()

    root.mainloop()

這套 Python 後台程式是 門禁系統的核心控制塔（Management Operations Center）。它利用多執行緒（Multithreading）技術，同時扮演 MQTT 訂閱者、SQLite 資料庫管理器、Tkinter GUI 圖形化介面，以及 Telegram Bot 自動推播機器人 4 個角色。

一、 系統四大核心組件與作用

[ Wokwi ESP32 雙讀卡機 ]
       │ (MQTT Topic: wokwi/rfid/access)
       ▼
[ Python MQTT 接收執行緒 ] ──(背景監聽)
       │
       ▼
[ process_swipe() 核心邏輯處理 ]
   ├── 1. 更新 SQLite 資料庫 (access_logs / inside_cards)
   ├── 2. 觸發 Telegram 背景 Thread 發送即時推播
   └── 3. 呼叫 refresh_ui() 更新 Tkinter GUI 畫面

二、 程式碼分區功能詳細說明

1. Telegram 非同步發送模組 (send_telegram_msg)

  • 作用：負責透過 HTTP POST 將格式化訊息發送到指定的 Telegram Chat ID。

  • 關鍵技術：使用 threading.Thread(target=_send, daemon=True) 在獨立的背景執行緒發送。

  • 效益：即便 Telegram 伺服器回應較慢或網路波動，也完全不會凍結（Lag）Tkinter GUI 介面或阻塞 MQTT 訊息接收。

2. SQLite 資料庫初始化與管理 (init_db)

  • 作用：自動建立與維護 access_control.db 本地資料庫，包含兩張關鍵資料表：

    • access_logs（歷史日誌）：記錄每一筆刷卡紀錄（流水號 id、卡片 card_uid、進出動作 action、時間戳記 timestamp）。

    • inside_cards（場內清單）：僅儲存目前人在場內的卡片 UID 與進場時間。

  • 邏輯轉化：

    • 入口刷卡 (AA:)：將 UID 寫入 access_logs（動作：IN），同時以 INSERT OR REPLACE 存入 inside_cards。

    • 出口刷卡 (BB:)：將 UID 寫入 access_logs（動作：OUT），並執行 DELETE FROM inside_cards 將其從場內名單移除。

3. Tkinter GUI 介面物件 (AccessControlApp)

  • 作用：提供直觀的視覺化管理視窗，分為三大區塊：

    • 頂部狀態欄 (status_label)：醒目顯示「場內當前總人數：X 人」，隨刷卡動態更新。

    • 「場內卡號清單」頁籤 (inside_tree)：使用 Treeview 表格，即時條列目前在場內的所有卡片及其入場時間。

    • 「歷史進出紀錄」頁籤 (log_tree)：顯示最新的 50 筆進出刷卡日誌。

  • 介面更新機制：process_swipe() 處理完資料後，使用 self.root.after(0, self.refresh_ui) 將 GUI 刷新動作安全地排入主執行緒執行，避免跨執行緒操作 UI 造成的崩潰。

4. 背景 MQTT 監聽服務 (start_mqtt)

  • 作用：在背景執行緒啟動 paho.mqtt.client，連線至公共 MQTT Broker (broker.hivemq.com) 並訂閱 wokwi/rfid/access 主題。

  • 自動觸發：當 ESP32 刷卡並 Publish 訊息時，on_message 毀調函式會第一時間擷取卡號 Payload（如 AA:01:02:03:04），並丟給 app.process_swipe() 進行處理。

三、 系統特色與架構優勢

  1. 職責分離（Decoupling）：ESP32 只專注於讀取 RFID 並發送 MQTT 極短字串，所有複雜的邏輯判斷、SQLite 寫入、時間計算與 Telegram 網路傳送全由 Python 後台承擔。

  2. 狀態即時同步：刷卡感應後，SQLite 日誌寫入、Tkinter 介面人數更新、Telegram 訊息推播 三者會在 1 秒內同步完成。

  3. 資料持久化（Persistence）：就算電腦關機或程式重啟，SQLite 內的歷史紀錄與場內清單依然完整保留，不會遺失。






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 = "8845708831:AAEuRreXFRyv1m--jGK7OCn8MiEsWXqpMT0"  # // 請替換為您的 Bot Token
TELEGRAM_CHAT_ID = "7965218469"  # // 請替換為您的 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()





它透過多執行緒（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）時：

    1. 自動判斷進/出場並更新 SQLite。

    2. 計算最新場內人數。

    3. 發送 Telegram 帶有精確時間與人數的推播通知。

    4. 排程刷新 Tkinter UI 表格。

6. Background MQTT 監聽服務 (start_mqtt)

  • 作用：連線至公共 MQTT Broker (broker.hivemq.com)，訂閱 wokwi/rfid/access 主題，即時接收 ESP32 傳送過來的刷卡資料。

三、 系統特色與優勢

  1. 完全非阻塞（Non-blocking）：MQTT 接收、GUI 繪製與 Telegram 網路傳輸各自獨立在不同的 Thread 運作，系統回應非常順暢。

  2. 雙向互動能力：不只能接收 ESP32 刷卡自動通知 Telegram，管理者還能隨時透過 Telegram 發送 /status 查詢現況或 /clear 遠端清場。

  3. 資料持久化與可追溯性：即使後台程式重啟，SQLite 資料庫仍完整保存場內狀態與所有歷史門禁日誌。








於
標籤：

沒有留言:

張貼留言

訂閱： 張貼留言 (Atom)

MQTT + SQLite + Tkinter + Telegram 雙向門禁管理後台系統 (作業3)

MQTT + SQLite + Tkinter + Telegram 雙向門禁管理後台系統 。 它透過多執行緒（Multithreading）同時處理 MQTT 訊息接收、GUI 視窗介面更新、SQLite 資料庫讀寫，以及 Telegram Bot 的推播與指令監聽。 一、 系...

  • Line 發報機
    Line 發報機 Python TKinter (CONFIG_LINE Message API_2.py) import serial import serial.tools.list_ports import tkinter as tk from tkinter import...
  • python pip 不是内部或外部命令--解決方法
    python pip 不是内部或外部命令 -- 解決方法 要安裝 Pyqt5 1. 首先，開啟命令提示字元。 2. 輸入 pip3 install pyqt5 好像不能執行 ! ! 錯誤顯示 : ‘ pip3 ’ 不是內部或外部命令、可執行的程式或批...
  • 2017 勤益科大 課程講義 下載
    課程講義 下載 11/20 1)  PPT 下載  +  程式下載    http://www.mediafire.com/file/cru4py7e8pptfda/106%E5%8B%A4%E7%9B%8A2-1.rar 11/27 2)  PPT 下載...