Upload app.py

app.py

@@ -0,0 +1,201 @@
+import gradio as gr
+import datetime
+import pandas as pd
+import joblib
+from sklearn.cluster import HDBSCAN
+import numpy as np
+import logger_service
+
+# --- 1. 載入模型與歷史資料 (實現動態 Rule 的關鍵) ---
+try:
+    le_action = joblib.load('le_action.pkl')
+    le_ip = joblib.load('le_ip.pkl')
+    
+    # 🌟 載入歷史資料作為 HDBSCAN 即時比對的基準
+    baseline_df = pd.read_csv('baseline_logs.csv')
+    baseline_df['hour'] = pd.to_datetime(baseline_df['timestamp']).dt.hour
+    
+    # 預先將歷史資料轉成特徵矩陣
+    baseline_X = np.column_stack((
+        baseline_df['hour'].values,
+        le_ip.transform(baseline_df['ip_address'].values),
+        le_action.transform(baseline_df['action'].values)
+    ))
+    
+    AI_READY = True
+    print("✅ AI 模組與歷史基準資料載入成功！")
+except Exception as e:
+    AI_READY = False
+    print(f"⚠️ AI 模組載入失敗，請確認 .pkl 與 .csv 檔案是否已上傳。錯誤: {e}")
+
+USER_DB = {
+    "student": ["1234", "學生", "同學"], 
+    "teacher": ["1234", "老師", "教授"],
+    "admin": ["1234", "行政", "行政人員"],
+    "ta": ["1234", "TA", "助教"],
+    "sysadmin": ["1234", "系統管理員", "管理員"]
+}
+
+log_history = baseline_df.sort_values(by='timestamp', ascending=False).head(20).to_dict('records')
+
+# --- 3. 核心邏輯與真實 HDBSCAN 異常偵測 ---
+def check_anomaly(ip, action, log_time):
+    if not AI_READY:
+        return "success" 
+        
+    if action in ["login_attempt", "logout"]:
+        return "success"
+
+    try:
+        # 1. 將新進來的行為轉為代碼
+        ip_code = le_ip.transform([ip])[0]
+        action_code = le_action.transform([action])[0]
+        
+        # 2. 擷取當下行為的時間 (Hour)
+        current_hour = pd.to_datetime(log_time).hour
+        
+        # 3. 組合這筆新資料的特徵向量
+        new_data_point = np.array([[current_hour, ip_code, action_code]])
+        
+        # 4. 🌟 將新資料與歷史資料合併 (這就是流程圖中的「即時丟入 AI 進行判斷」)
+        combined_X = np.vstack((baseline_X, new_data_point))
+        
+        # 5. 執行 HDBSCAN 分群
+        model = HDBSCAN(min_cluster_size=5, min_samples=3)
+        clusters = model.fit_predict(combined_X)
+        
+        # 6. 檢查最後一筆 (也就是剛剛送進來的新動作) 是否被分類為 -1 (雜訊/異常)
+        if clusters[-1] == -1:
+            return "⚠️異常_作息或行為不符"
+        else:
+            return "success"
+            
+    except ValueError:
+        # 如果遇到沒看過的 IP 或動作，連算都不用算，絕對是異常
+        return "⚠️異常_未知特徵"
+
+def record_log(ip, cookie, account, role, action, status, time_mode, custom_time):
+    global baseline_X, log_history
+    # 決定紀錄的時間
+    if time_mode == "自訂時間 (模擬過去/未來)" and custom_time.strip() != "":
+        log_time = custom_time
+    else:
+        tw_tz = datetime.timezone(datetime.timedelta(hours=8))
+        log_time = datetime.datetime.now(tw_tz).strftime("%Y-%m-%d %H:%M:%S")
+
+    # 執行 AI 檢查 (現在把 log_time 也傳進去了)
+    if status == "success":
+        ai_judgment = check_anomaly(ip, action, log_time)
+        if ai_judgment != "success":
+            status = ai_judgment
+
+    new_log = {
+        "timestamp": log_time,
+        "ip_address": ip,
+        "cookie_id": cookie,
+        "account": account,
+        "role": role,
+        "action": action,
+        "status": status
+    }
+
+    try:
+        # 將新行為轉為特徵數值
+        new_hour = pd.to_datetime(log_time).hour
+        new_ip_code = le_ip.transform([ip])[0]
+        new_action_code = le_action.transform([action])[0]
+        new_feature = np.array([[new_hour, new_ip_code, new_action_code]])
+        
+        # 🌟 把新特徵壓入全域矩陣中
+        baseline_X = np.vstack((baseline_X, new_feature))
+    except Exception as e:
+        print(f"⚠️ AI 矩陣更新跳過 (可能是未知特徵): {e}")
+
+    # 呼叫外部 Service 進行存檔與上傳
+    logger_service.push_new_log(new_log)
+
+    log_history.insert(0, new_log)
+    return pd.DataFrame(log_history)
+
+def process_login(ip, cookie, account, password, captcha, time_mode, custom_time):
+    valid_captcha = "FEIK"
+    if account not in USER_DB: return gr.update(visible=True), gr.update(visible=False), "帳號不存在！", gr.update(), record_log(ip, cookie, account, "Unknown", "login_attempt", "failed_no_account", time_mode, custom_time)
+    role, name = USER_DB[account][1], USER_DB[account][2]
+    if captcha.upper() != valid_captcha: return gr.update(visible=True), gr.update(visible=False), "驗證碼錯誤！", gr.update(), record_log(ip, cookie, account, role, "login_attempt", "failed_captcha", time_mode, custom_time)
+    if USER_DB[account][0] == password:
+        welcome_msg = f"**{name}** 歡迎您  |  線上人數: 938"
+        df = record_log(ip, cookie, account, role, "login_attempt", "success", time_mode, custom_time)
+        return gr.update(visible=False), gr.update(visible=True), "", welcome_msg, df
+    return gr.update(visible=True), gr.update(visible=False), "密碼錯誤！", gr.update(), record_log(ip, cookie, account, role, "login_attempt", "failed_password", time_mode, custom_time)
+
+def simulate_click(ip, cookie, account, button_name, time_mode, custom_time):
+    role = USER_DB.get(account, ["", "Unknown"])[1]
+    return record_log(ip, cookie, account, role, button_name, "success", time_mode, custom_time)
+
+def logout(ip, cookie, account, time_mode, custom_time):
+    role = USER_DB.get(account, ["", "Unknown"])[1]
+    df = record_log(ip, cookie, account, role, "logout", "success", time_mode, custom_time)
+    return gr.update(visible=True), gr.update(visible=False), "", "", "", "", df
+
+def toggle_time_input(mode):
+    return gr.update(visible=(mode == "自訂時間 (模擬過去/未來)"))
+
+# --- 4. UI 介面設計 ---
+custom_css = """
+.mock-panel { background-color: var(--background-fill-secondary); padding: 15px; border-radius: 8px; border: 1px dashed var(--border-color-primary); }
+.login-container { max-width: 500px; margin: 0 auto; padding: 20px; }
+.placeholder-box { background-color: var(--background-fill-secondary); border: 2px dashed var(--border-color-primary); text-align: center; color: var(--body-text-color); padding: 40px 0; margin-bottom: 15px; border-radius: 5px; }
+.ntut-title { color: #d9534f; font-size: 22px; font-weight: bold; text-align: center; margin-bottom: 20px;}
+"""
+
+with gr.Blocks(title="模擬校園入口與資安防護系統", css=custom_css) as demo:
+    with gr.Row():
+        with gr.Column(scale=1, elem_classes="mock-panel"):
+            gr.Markdown("### ⚙️ 環境變數模擬器")
+            mock_ip = gr.Dropdown(choices=["140.124.71.55 (校內預設)", "140.124.18.22 (宿舍)", "45.33.22.11 (國外異常IP)"], value="140.124.71.55 (校內預設)", allow_custom_value=True, label="模擬來源 IP")
+            mock_cookie = gr.Textbox(label="模擬 Cookie Session ID", value="sess_baseline")
+            gr.Markdown("---")
+            time_mode = gr.Radio(label="時間設定模式", choices=["真實時間 (目前時間)", "自訂時間 (模擬過去/未來)"], value="真實時間 (目前時間)")
+            custom_time_input = gr.Textbox(label="輸入自訂時間", placeholder="格式: YYYY-MM-DD HH:MM:SS", value="2026-03-17 03:00:00", visible=False)
+            gr.Markdown("---")
+            gr.Markdown("### 📊 即時 Log 與 AI 判定區")
+            log_display = gr.Dataframe(headers=["timestamp", "ip_address", "cookie_id", "account", "role", "action", "status"], datatype=["str", "str", "str", "str", "str", "str", "str"], interactive=False, wrap=True)
+
+        with gr.Column(scale=2):
+            with gr.Column(visible=True, elem_classes="login-container") as login_view:
+                gr.HTML("<div class='placeholder-box'>預留位置：台北科大 Logo 與 e化校園圖片</div>")
+                gr.Markdown("<div class='ntut-title'>校園入口網站 Taipei Tech Portal</div>")
+                login_msg = gr.Markdown(value="", visible=True)
+                with gr.Row(): acc_input = gr.Textbox(label="使用者帳號 (Account)", placeholder="例如: student, admin...")
+                with gr.Row(): pwd_input = gr.Textbox(label="使用者密碼 (Password)", type="password", placeholder="密碼皆為 1234")
+                with gr.Row(): gr.HTML("<div style='background:var(--background-fill-secondary); color:var(--body-text-color); padding:10px; font-weight:bold; letter-spacing: 3px; border:1px solid var(--border-color-primary); text-align:center;'>F E I K</div>")
+                with gr.Row(): captcha_input = gr.Textbox(label="請輸入驗證碼 (Keyin Code)")
+                login_btn = gr.Button("登入 Login", variant="primary")
+
+            with gr.Column(visible=False) as dashboard_view:
+                gr.HTML("<div class='placeholder-box'>預留位置：學生學習歷程入口網站橫幅</div>")
+                with gr.Row():
+                    gr.Markdown("### 資訊系統")
+                    welcome_text = gr.Markdown(value="", elem_classes="text-right")
+                    logout_btn = gr.Button("登出", size="sm")
+                gr.Markdown("---")
+                with gr.Accordion("▼ 1. 教務系統", open=True):
+                    with gr.Row(): btn_eval = gr.Button("▶ 期末網路教學評量", size="sm"); btn_pre_select = gr.Button("▶ 期末網路預選系統", size="sm"); btn_add_drop = gr.Button("▶ 開學後加退選系統", size="sm")
+                    with gr.Row(): btn_istudy = gr.Button("▶ 北科i學園PLUS", size="sm"); btn_card = gr.Button("▶ 學生證掛失及補發系統", size="sm"); btn_course = gr.Button("▶ 課程系統", size="sm")
+                    with gr.Row(): btn_score = gr.Button("▶ 學業成績查詢系統", size="sm")
+                with gr.Accordion("▼ 2. 學務系統", open=True):
+                    with gr.Row(): btn_leave = gr.Button("▶ 學生請假系統", size="sm"); btn_dorm = gr.Button("▶ 學生宿舍登錄(抽籤)系統", size="sm"); btn_scholarship = gr.Button("▶ 獎助學金申請系統", size="sm")
+                with gr.Accordion("▼ 3. 資訊服務", open=True):
+                    with gr.Row(): btn_webmail = gr.Button("▶ 網路郵局 WebMail", size="sm"); btn_vdesk = gr.Button("▶ 北科軟體雲", size="sm")
+                with gr.Accordion("▼ 惡意操作測試區", open=True):
+                    btn_hack_score = gr.Button("💀 嘗試偷改成績 (SQL Injection)", variant="stop", size="sm")
+
+    time_mode.change(fn=toggle_time_input, inputs=time_mode, outputs=custom_time_input)
+    login_btn.click(fn=process_login, inputs=[mock_ip, mock_cookie, acc_input, pwd_input, captcha_input, time_mode, custom_time_input], outputs=[login_view, dashboard_view, login_msg, welcome_text, log_display])
+    logout_btn.click(fn=logout, inputs=[mock_ip, mock_cookie, acc_input, time_mode, custom_time_input], outputs=[login_view, dashboard_view, acc_input, pwd_input, captcha_input, login_msg, log_display])
+
+    for btn, action_name in [(btn_eval, "click_eval_system"), (btn_pre_select, "click_pre_select_system"), (btn_add_drop, "click_add_drop_system"), (btn_istudy, "click_istudy"), (btn_card, "click_card_loss"), (btn_course, "click_course_system"), (btn_score, "click_score_query"), (btn_leave, "click_leave_system"), (btn_dorm, "click_dorm_system"), (btn_scholarship, "click_scholarship"), (btn_webmail, "click_webmail"), (btn_vdesk, "click_vdesk"), (btn_hack_score, "malicious_sql_injection")]:
+        btn.click(fn=lambda ip, c, acc, t_mode, t_custom, n=action_name: simulate_click(ip, c, acc, n, t_mode, t_custom), inputs=[mock_ip, mock_cookie, acc_input, time_mode, custom_time_input], outputs=[log_display])
+
+if __name__ == "__main__":
+    demo.launch()