update locate weight

app.py

@@ -33,20 +33,17 @@ global_stats = {
     "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",
     "教務系統: 期末網路預選系統": "click_pre_select_system",
@@ -66,24 +63,33 @@ ACTION_MAP = {
 log_history = []
 initial_df_abn = pd.DataFrame(columns=["timestamp", "ip_address", "cookie_id", "account", "role", "action", "status"])
 
+# HDBSCAN 特徵分群用的 IP 分類
 def get_ip_category(ip_str):
     ip = str(ip_str).split(" ")[0]
     if ip.startswith("140.124."): return 0.0    
     elif ip.startswith("61.228."): return 1.0   
     elif ip.startswith("210.61."): return 2.0   
     elif ip.startswith("45.33."): return 3.0    
-    else: return 4.0                            
+    else: return 4.0  
+
+# 👉 新增：專供 Geo-Weight 權重邏輯使用的精細分級
+def get_geo_level(ip_str):
+    ip = str(ip_str).split(" ")[0]
+    if ip == "140.124.71.55": return 0       # 校內預設
+    elif ip == "140.124.18.22": return 1     # 宿舍
+    elif ip.startswith("61.228."): return 2  # 家裡
+    elif ip.startswith("210.61."): return 3  # 公共場所
+    elif ip.startswith("45.33."): return 4   # 國外異常
+    else: return 5                           # 未知
 
 # --- 1. 載入模型與歷史資料 ---
 try:
     le_action = joblib.load('le_action.pkl')
     
     print("📂 正在讀取本地歷史資料庫...")
-    # 直接讀取本地的 baseline_logs.csv，不再向雲端同步
     if os.path.exists('baseline_logs.csv'): 
         baseline_df = pd.read_csv('baseline_logs.csv')
     else: 
-        # 如果檔案不存在，建立帶有標題的空 DataFrame 並存檔
         baseline_df = pd.DataFrame(columns=["timestamp", "ip_address", "cookie_id", "account", "role", "action", "status"])
         baseline_df.to_csv('baseline_logs.csv', index=False)
 
@@ -179,7 +185,6 @@ def get_plot_data(selected_key):
         
         df_live = pd.DataFrame(log_history)
         if not df_live.empty:
-            # 👉 雙重防護 2：加入 errors='coerce' 容錯機制，剔除壞資料避免圖表破圖
             df_live['timestamp'] = pd.to_datetime(df_live['timestamp'], errors='coerce')
             df_live = df_live.dropna(subset=['timestamp']) 
             df_live = df_live[df_live['action'] == action_code].copy()
@@ -264,7 +269,6 @@ def check_anomaly(ip, account, action, log_time):
     if action in ["login_attempt", "logout"]: return "success"
 
     try:
-        # 直接從全域變數提取當前權重
         ow = float(system_weights["ow"])
         tw = float(system_weights["tw"])
         iw = float(system_weights["iw"])
@@ -272,9 +276,22 @@ def check_anomaly(ip, account, action, log_time):
         gw = float(system_weights["gw"])
         fw = float(system_weights["fw"])
         
-        # --- 第一道防線 (限縮為單一按鈕判斷，並套用精確的閾值) ---
+        # 👉 第一道防線：地理圍籬 (Geo-Fencing) 攔截
+        geo_level = get_geo_level(ip)
+        
+        if gw == 100 and geo_level > 0:
+            return "⚠️異常_地理位置受限 (100%絕對鎖定: 僅限校內專網存取)"
+        elif gw >= 90 and geo_level > 1:
+            return "⚠️異常_地理位置受限 (高敏感防護: 僅限校園網路與宿舍)"
+        elif gw >= 70 and geo_level > 2:
+            return "⚠️異常_地理位置受限 (進階防護: 禁止公共場所與海外連線)"
+        elif gw >= 50 and geo_level > 3:
+            return "⚠️異常_地理位置受限 (預設防護: 禁止海外異常 IP)"
+        # 若 gw < 50，則放行所有位置，交由後續 AI 處理
+        
+        # --- 第二道防線：頻率攔截 ---
         current_dt = pd.to_datetime(str(log_time))
-        cutoff_dt = current_dt - pd.Timedelta(seconds=60) # 往前回溯 60 秒
+        cutoff_dt = current_dt - pd.Timedelta(seconds=60)
         
         recent_clicks = 0
         for log in log_history:
@@ -300,7 +317,7 @@ def check_anomaly(ip, account, action, log_time):
         if (recent_clicks + 1) > max_allowed_clicks:
             return f"⚠️異常_單一操作頻率過高 ({recent_clicks+1}次/分)"
             
-        # --- 第二道防線 (HDBSCAN AI 空間異常判斷) 繼續執行 ---
+        # --- 第三道防線：HDBSCAN AI 空間異常判斷 ---
         ip_code = float(get_ip_category(ip))
         action_code = float(le_action.transform([action])[0])
         
@@ -362,13 +379,11 @@ def check_anomaly(ip, account, action, log_time):
 def record_log(ip, cookie, account, role, action, status, time_mode, custom_time):
     global baseline_X 
     
-    # 👉 雙重防護 1：驗證自訂時間格式
     if time_mode == "自訂時間 (模擬過去/未來)" and custom_time and str(custom_time).strip() != "": 
         try:
             valid_time = pd.to_datetime(str(custom_time))
             log_time = valid_time.strftime("%Y-%m-%d %H:%M:%S")
         except Exception:
-            # 🚨 解析失敗，退回當前時間並標記狀態
             log_time = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S")
             status = "⚠️格式錯誤_無效的自訂時間"
     else: 
@@ -380,7 +395,6 @@ def record_log(ip, cookie, account, role, action, status, time_mode, custom_time
 
     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)
     
@@ -388,7 +402,7 @@ def record_log(ip, cookie, account, role, action, status, time_mode, custom_time
     except: pass
 
     global_stats["total_count"] += 1
-    if "異常" in str(status) or "錯誤" in str(status): global_stats["abnormal_count"] += 1
+    if "異常" in str(status) or "錯誤" in str(status) or "受限" in str(status): global_stats["abnormal_count"] += 1
 
     log_history.insert(0, new_log)
 
@@ -401,10 +415,10 @@ def record_log(ip, cookie, account, role, action, status, time_mode, custom_time
         except: pass 
 
     df_all = pd.DataFrame(log_history)
-    abnormal_logs = [log for log in log_history if "異常" in str(log.get("status", "")) or "錯誤" in str(log.get("status", ""))]
+    abnormal_logs = [log for log in log_history if "異常" in str(log.get("status", "")) or "錯誤" in str(log.get("status", "")) or "受限" in str(log.get("status", ""))]
     df_abnormal = pd.DataFrame(abnormal_logs) if abnormal_logs else pd.DataFrame(columns=df_all.columns)
     
-    input_update = gr.update(value=f"時間: {log_time} | IP: {ip} | 帳號: {account} | 動作: {action} | 狀態: {status}") if "異常" in str(status) or "錯誤" in str(status) else gr.update()
+    input_update = gr.update(value=f"時間: {log_time} | IP: {ip} | 帳號: {account} | 動作: {action} | 狀態: {status}") if "異常" in str(status) or "錯誤" in str(status) or "受限" in str(status) else gr.update()
     return df_all, df_abnormal, input_update
 
 def fast_login(target_account, ip, cookie, time_mode, custom_time):
@@ -593,7 +607,6 @@ 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=[])