update passing weight

app.py

@@ -26,13 +26,26 @@ if os.path.exists(font_path):
     plt.rc('font', family='SimHei')
 plt.rcParams['axes.unicode_minus'] = False
 
-# --- 全域統計狀態變數 ---
+# --- 全域統計與狀態變數 ---
 global_stats = {
     "total_count": 0,
     "abnormal_count": 0,
     "start_time": datetime.datetime.now()
 }
 
+# 系統防護權重全域變數 (預設值)
+system_weights = {
+    "ow": 75.0, "tw": 80.0, "iw": 65.0, 
+    "dw": 45.0, "gw": 50.0, "fw": 60.0
+}
+
+# 專屬更新權重的副程式
+def update_system_weights(ow, tw, iw, dw, gw, fw):
+    system_weights.update({
+        "ow": float(ow), "tw": float(tw), "iw": float(iw), 
+        "dw": float(dw), "gw": float(gw), "fw": float(fw)
+    })
+
 # --- 13 個系統按鈕的對應字典 ---
 ACTION_MAP = {
     "教務系統: 期末網路教學評量": "click_eval_system",
@@ -66,7 +79,7 @@ try:
     le_action = joblib.load('le_action.pkl')
     
     print("📂 正在讀取本地歷史資料庫...")
-    # 👉 修改點 1: 直接讀取本地的 baseline_logs.csv，不再向雲端同步
+    # 直接讀取本地的 baseline_logs.csv，不再向雲端同步
     if os.path.exists('baseline_logs.csv'): 
         baseline_df = pd.read_csv('baseline_logs.csv')
     else: 
@@ -108,7 +121,7 @@ try:
     ))
     
     AI_READY = True
-    print("✅ AI 模組與動態歷史基準資料載入成功！")
+    print("✅ AI 模組與歷史基準資料載入成功！")
 except Exception as e:
     AI_READY = False
     global_stats["total_count"] = 0; global_stats["abnormal_count"] = 0
@@ -130,7 +143,7 @@ def get_dashboard_html():
         <div class="stat-card card-total">
             <div class="card-icon">📊</div>
             <div class="card-content">
-                <div class="card-label">總監測量 (靜態+動態資料庫)</div>
+                <div class="card-label">總監測量 (本地歷史資料庫)</div>
                 <div class="card-value">{total:,} <span class="card-unit">筆 Log</span></div>
                 <div class="card-sub-text">自系統啟動起統計 (+今日即時)</div>
             </div>
@@ -244,15 +257,20 @@ def explain_abnormal_log(log_text):
     except Exception as e: return f"⚠️ 呼叫 API 發生錯誤: {str(e)}"
 
 # --- 2. 核心邏輯與真實 HDBSCAN 異常偵測 ---
-def check_anomaly(ip, account, action, log_time, ow, tw, iw, dw, gw, fw):
+def check_anomaly(ip, account, action, log_time):
     if not AI_READY: return "success" 
     if action in ["login_attempt", "logout"]: return "success"
 
     try:
-        ow, tw, iw, dw = float(ow), float(tw), float(iw), float(dw)
-        gw, fw = float(gw), float(fw) 
+        # 直接從全域變數提取當前權重
+        ow = float(system_weights["ow"])
+        tw = float(system_weights["tw"])
+        iw = float(system_weights["iw"])
+        dw = float(system_weights["dw"])
+        gw = float(system_weights["gw"])
+        fw = float(system_weights["fw"])
         
-        # --- 🌟 修改：第一道防線 (限縮為單一按鈕判斷，並套用精確的閾值) ---
+        # --- 第一道防線 (限縮為單一按鈕判斷，並套用精確的閾值) ---
         current_dt = pd.to_datetime(str(log_time))
         cutoff_dt = current_dt - pd.Timedelta(seconds=60) # 往前回溯 60 秒
         
@@ -260,13 +278,11 @@ def check_anomaly(ip, account, action, log_time, ow, tw, iw, dw, gw, fw):
         for log in log_history:
             log_dt = pd.to_datetime(log['timestamp'])
             if log_dt >= cutoff_dt:
-                # 🌟 修改：同時檢查「同一個帳號」且「同一個動作 (按鈕)」
                 if log.get('account') == account and log.get('action') == action:
                     recent_clicks += 1
             else:
                 break
                 
-        # 🌟 修改：依據你要求的階梯式敏感度設定最高容忍次數
         if fw <= 0: max_allowed_clicks = 100
         elif fw <= 10: max_allowed_clicks = 90
         elif fw <= 20: max_allowed_clicks = 80
@@ -279,7 +295,6 @@ def check_anomaly(ip, account, action, log_time, ow, tw, iw, dw, gw, fw):
         elif fw <= 90: max_allowed_clicks = 9
         else: max_allowed_clicks = 3
 
-        # 觸發條件：若包含這一次的點擊，超過了容忍次數，直接攔截！
         if (recent_clicks + 1) > max_allowed_clicks:
             return f"⚠️異常_單一操作頻率過高 ({recent_clicks+1}次/分)"
             
@@ -342,17 +357,18 @@ def check_anomaly(ip, account, action, log_time, ow, tw, iw, dw, gw, fw):
     except Exception as e:
         return f"⚠️系統錯誤: {str(e)}"
 
-def record_log(ip, cookie, account, role, action, status, time_mode, custom_time, ow, tw, iw, dw, gw, fw):
+def record_log(ip, cookie, account, role, action, status, time_mode, custom_time):
     global baseline_X 
     if time_mode == "自訂時間 (模擬過去/未來)" and custom_time and str(custom_time).strip() != "": log_time = str(custom_time)
     else: log_time = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S")
 
     if status == "success":
-        ai_judgment = check_anomaly(ip, account, action, log_time, ow, tw, iw, dw, gw, fw)
+        ai_judgment = check_anomaly(ip, account, 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}
 
+    # 將產生的新 Log 直接 append (附加) 到本地的 baseline_logs.csv 中
     new_df = pd.DataFrame([new_log])
     new_df.to_csv('baseline_logs.csv', mode='a', header=not os.path.exists('baseline_logs.csv'), index=False)
     
@@ -379,45 +395,45 @@ def record_log(ip, cookie, account, role, action, status, time_mode, custom_time
     input_update = gr.update(value=f"時間: {log_time} | IP: {ip} | 帳號: {account} | 動作: {action} | 狀態: {status}") if "異常" in str(status) else gr.update()
     return df_all, df_abnormal, input_update
 
-def fast_login(target_account, ip, cookie, time_mode, custom_time, ow, tw, iw, dw, gw, fw):
+def fast_login(target_account, ip, cookie, time_mode, custom_time):
     role, name = USER_DB[target_account][1], USER_DB[target_account][2]
-    df_all, df_abn, inp_upd = record_log(ip, cookie, target_account, role, "login_attempt", "success", time_mode, custom_time, ow, tw, iw, dw, gw, fw)
+    df_all, df_abn, inp_upd = record_log(ip, cookie, target_account, role, "login_attempt", "success", time_mode, custom_time)
     target_tab = "student_tab" if role == "學生" else "admin_tab"
     return gr.update(selected="dashboard_route"), gr.update(selected=target_tab), "", f"**{name}** 歡迎您  |  線上人數: 938 (⚡快速登入模式)", df_all, df_abn, inp_upd, target_account
 
-def process_login(ip, cookie, account, password, captcha, time_mode, custom_time, ow, tw, iw, dw, gw, fw):
+def process_login(ip, cookie, account, password, captcha, time_mode, custom_time):
     def fail_response(msg, df_all, df_abn, inp_upd): return gr.update(selected="login_route"), gr.update(), msg, gr.update(), df_all, df_abn, inp_upd
     if account not in USER_DB: 
-        df_all, df_abn, inp_upd = record_log(ip, cookie, account, "Unknown", "login_attempt", "failed_no_account", time_mode, custom_time, ow, tw, iw, dw, gw, fw)
+        df_all, df_abn, inp_upd = record_log(ip, cookie, account, "Unknown", "login_attempt", "failed_no_account", time_mode, custom_time)
         return fail_response("帳號不存在！", df_all, df_abn, inp_upd)
     role, name = USER_DB[account][1], USER_DB[account][2]
     if captcha.upper() != "FEIK": 
-        df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_captcha", time_mode, custom_time, ow, tw, iw, dw, gw, fw)
+        df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_captcha", time_mode, custom_time)
         return fail_response("驗證碼錯誤！", df_all, df_abn, inp_upd)
     if USER_DB[account][0] == password:
-        df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "success", time_mode, custom_time, ow, tw, iw, dw, gw, fw)
+        df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "success", time_mode, custom_time)
         return gr.update(selected="dashboard_route"), gr.update(selected="student_tab" if role == "學生" else "admin_tab"), "", f"**{name}** 歡迎您  |  線上人數: 938", df_all, df_abn, inp_upd
-    df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_password", time_mode, custom_time, ow, tw, iw, dw, gw, fw)
+    df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_password", time_mode, custom_time)
     return fail_response("密碼錯誤！", df_all, df_abn, inp_upd)
 
-def toggle_role(current_account, ip, cookie, time_mode, custom_time, ow, tw, iw, dw, gw, fw):
+def toggle_role(current_account, ip, cookie, time_mode, custom_time):
     target_account = "student" if current_account == "admin" else "admin"
     role, name = USER_DB[target_account][1], USER_DB[target_account][2]
-    df_all, df_abn, inp_upd = record_log(ip, cookie, target_account, role, "login_attempt", "success", time_mode, custom_time, ow, tw, iw, dw, gw, fw)
+    df_all, df_abn, inp_upd = record_log(ip, cookie, target_account, role, "login_attempt", "success", time_mode, custom_time)
     return gr.update(selected="student_tab" if role == "學生" else "admin_tab"), f"**{name}** 歡迎您  |  線上人數: 938 (⚡單鍵切換模式)", df_all, df_abn, inp_upd, target_account
 
-def logout(ip, cookie, account, time_mode, custom_time, ow, tw, iw, dw, gw, fw):
+def logout(ip, cookie, account, time_mode, custom_time):
     role = USER_DB.get(account, ["", "Unknown"])[1]
-    df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "logout", "success", time_mode, custom_time, ow, tw, iw, dw, gw, fw)
+    df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "logout", "success", time_mode, custom_time)
     return gr.update(selected="login_route"), "", "", "", "", df_all, df_abn, inp_upd
 
-def simulate_click(ip, cookie, account, button_name, time_mode, custom_time, ow, tw, iw, dw, gw, fw):
+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, ow, tw, iw, dw, gw, fw)
+    return record_log(ip, cookie, account, role, button_name, "success", time_mode, custom_time)
 
-def test_anomaly(ip, cookie, account, action_name, test_time_str, ow, tw, iw, dw, gw, fw):
+def test_anomaly(ip, cookie, account, action_name, test_time_str):
     role = USER_DB.get(account, ["", "Unknown"])[1]
-    return record_log(ip, cookie, account, role, action_name, "success", "自訂時間 (模擬過去/未來)", test_time_str, ow, tw, iw, dw, gw, fw)
+    return record_log(ip, cookie, account, role, action_name, "success", "自訂時間 (模擬過去/未來)", test_time_str)
 
 def toggle_time_input(mode): return gr.update(visible=(mode == "自訂時間 (模擬過去/未來)"))
 
@@ -565,18 +581,22 @@ with gr.Blocks(title="模擬校園入口與資安防護系統") as demo:
     time_mode.change(fn=toggle_time_input, inputs=time_mode, outputs=custom_time_input)
     btn_explain_log.click(fn=explain_abnormal_log, inputs=[abnormal_log_input], outputs=[ai_explanation_output])
     
+    # 綁定權重拉桿與更新函式
     weight_inputs = [overall_weight, time_weight, ip_weight, device_weight, geo_weight, freq_weight]
+    for w_input in weight_inputs:
+        w_input.change(fn=update_system_weights, inputs=weight_inputs, outputs=[])
+
     update_outputs = [main_router, role_tabs, login_msg, welcome_text, store_df_all, store_df_abn, abnormal_log_input]
     
-    login_btn.click(fn=process_login, inputs=[mock_ip, mock_cookie, acc_input, pwd_input, captcha_input, time_mode, custom_time_input] + weight_inputs, outputs=update_outputs)
-    logout_btn.click(fn=logout, inputs=[mock_ip, mock_cookie, acc_input, time_mode, custom_time_input] + weight_inputs, outputs=[main_router, acc_input, pwd_input, captcha_input, login_msg, store_df_all, store_df_abn, abnormal_log_input])
+    login_btn.click(fn=process_login, inputs=[mock_ip, mock_cookie, acc_input, pwd_input, captcha_input, time_mode, custom_time_input], outputs=update_outputs)
+    logout_btn.click(fn=logout, inputs=[mock_ip, mock_cookie, acc_input, time_mode, custom_time_input], outputs=[main_router, acc_input, pwd_input, captcha_input, login_msg, store_df_all, store_df_abn, abnormal_log_input])
     
-    fast_inputs = [mock_ip, mock_cookie, time_mode, custom_time_input] + weight_inputs
+    fast_inputs = [mock_ip, mock_cookie, time_mode, custom_time_input]
     fast_outputs = update_outputs + [acc_input] 
     
-    btn_fast_student.click(fn=lambda ip, c, tm, ct, ow, tw, iw, dw, gw, fw: fast_login("student", ip, c, tm, ct, ow, tw, iw, dw, gw, fw), inputs=fast_inputs, outputs=fast_outputs)
-    btn_fast_admin.click(fn=lambda ip, c, tm, ct, ow, tw, iw, dw, gw, fw: fast_login("admin", ip, c, tm, ct, ow, tw, iw, dw, gw, fw), inputs=fast_inputs, outputs=fast_outputs)
-    btn_toggle_role.click(fn=toggle_role, inputs=[acc_input, mock_ip, mock_cookie, time_mode, custom_time_input] + weight_inputs, outputs=[role_tabs, welcome_text, store_df_all, store_df_abn, abnormal_log_input, acc_input])
+    btn_fast_student.click(fn=lambda ip, c, tm, ct: fast_login("student", ip, c, tm, ct), inputs=fast_inputs, outputs=fast_outputs)
+    btn_fast_admin.click(fn=lambda ip, c, tm, ct: fast_login("admin", ip, c, tm, ct), inputs=fast_inputs, outputs=fast_outputs)
+    btn_toggle_role.click(fn=toggle_role, inputs=[acc_input, mock_ip, mock_cookie, time_mode, custom_time_input], outputs=[role_tabs, welcome_text, store_df_all, store_df_abn, abnormal_log_input, acc_input])
 
     action_mapping = [
         (btn_eval, "click_eval_system"), (btn_pre_select, "click_pre_select_system"), (btn_add_drop, "click_add_drop_system"),
@@ -588,8 +608,8 @@ with gr.Blocks(title="模擬校園入口與資安防護系統") as demo:
 
     for btn, action_name in action_mapping:
         btn.click(
-            fn=lambda ip, c, acc, tm, ct, ow, tw, iw, dw, gw, fw, n=action_name: simulate_click(ip, c, acc, n, tm, ct, ow, tw, iw, dw, gw, fw), 
-            inputs=[mock_ip, mock_cookie, acc_input, time_mode, custom_time_input] + weight_inputs, 
+            fn=lambda ip, c, acc, tm, ct, n=action_name: simulate_click(ip, c, acc, n, tm, ct), 
+            inputs=[mock_ip, mock_cookie, acc_input, time_mode, custom_time_input], 
             outputs=[store_df_all, store_df_abn, abnormal_log_input] 
         )
         
@@ -603,8 +623,8 @@ with gr.Blocks(title="模擬校園入口與資安防護系統") as demo:
     
     for btn, action_name, test_time_str in test_mapping:
         btn.click(
-            fn=lambda ip, c, acc, ow, tw, iw, dw, gw, fw, n=action_name, t_str=test_time_str: test_anomaly(ip, c, acc, n, t_str, ow, tw, iw, dw, gw, fw),
-            inputs=[mock_ip, mock_cookie, acc_input, overall_weight, time_weight, ip_weight, device_weight, geo_weight, freq_weight],
+            fn=lambda ip, c, acc, n=action_name, t_str=test_time_str: test_anomaly(ip, c, acc, n, t_str),
+            inputs=[mock_ip, mock_cookie, acc_input],
             outputs=[store_df_all, store_df_abn, abnormal_log_input] 
         )