for API

app.py

@@ -73,15 +73,14 @@ def get_ip_category(ip_str):
     elif ip.startswith("45.33."): return 3.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                           # 未知
+    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:
@@ -136,6 +135,74 @@ except Exception as e:
 
 USER_DB = {"student": ["1234", "學生", "同學"], "admin": ["1234", "管理", "管理員"]}
 
+
+# =====================================================================
+# 👉 新增：純淨 API 端點專用函式 (只回傳 JSON 字典，給隊友抓資料用)
+# =====================================================================
+def api_get_stats():
+    """取得系統狀態與權重"""
+    total = global_stats["total_count"]
+    abnormal = global_stats["abnormal_count"]
+    rate = round((abnormal / total * 100), 2) if total > 0 else 0
+    return {
+        "status": "success",
+        "ai_ready": AI_READY,
+        "data": {
+            "total_logs": total,
+            "abnormal_logs": abnormal,
+            "anomaly_rate_percent": rate,
+            "current_weights": system_weights
+        }
+    }
+
+def api_get_logs(limit=20):
+    """取得最新的 N 筆 Log"""
+    try: limit = int(limit)
+    except: limit = 20
+    return {
+        "status": "success", 
+        "count": len(log_history[:limit]), 
+        "data": log_history[:limit]
+    }
+
+def api_get_chart_data(action_name):
+    """取得特定系統的圖表原始數據 (供前端自行繪圖)"""
+    if not AI_READY: return {"status": "error", "message": "AI 模型未載入"}
+    action_code = ACTION_MAP.get(action_name)
+    if not action_code: return {"status": "error", "message": f"找不到系統: {action_name}"}
+    
+    df_hist = baseline_df[baseline_df['action'] == action_code].copy()
+    df_live = pd.DataFrame(log_history)
+    if not df_live.empty:
+        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()
+        
+    if not df_live.empty:
+        df_combined = pd.concat([df_hist, df_live], ignore_index=True)
+        df_combined = df_combined.drop_duplicates(subset=['timestamp', 'account', 'action'])
+    else:
+        df_combined = df_hist.copy()
+        
+    if df_combined.empty:
+        return {"status": "success", "action": action_code, "data": {"normal": [], "abnormal": []}}
+        
+    df_combined['month'] = df_combined['timestamp'].dt.month
+    df_combined['time_of_day'] = df_combined['timestamp'].dt.hour + df_combined['timestamp'].dt.minute / 60.0
+    
+    is_abnormal = df_combined['status'].astype(str).str.contains("異常", na=False)
+    
+    normal_data = df_combined[~is_abnormal][['month', 'time_of_day', 'timestamp']].astype(str).to_dict('records')
+    abnormal_data = df_combined[is_abnormal][['month', 'time_of_day', 'timestamp', 'status']].astype(str).to_dict('records')
+    
+    return {
+        "status": "success", 
+        "action_code": action_code, 
+        "data": {"normal": normal_data, "abnormal": abnormal_data}
+    }
+# =====================================================================
+
+
 def get_dashboard_html():
     total = global_stats["total_count"]
     abnormal = global_stats["abnormal_count"]
@@ -277,7 +344,6 @@ 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:
@@ -288,9 +354,7 @@ def check_anomaly(ip, account, action, log_time):
             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)
         
@@ -318,7 +382,6 @@ def check_anomaly(ip, account, action, log_time):
         if (recent_clicks + 1) > max_allowed_clicks:
             return f"⚠️異常_單一操作頻率過高 ({recent_clicks+1}次/分)"
             
-        # --- 第三道防線：HDBSCAN AI 空間異常判斷 ---
         ip_code = float(get_ip_category(ip))
         action_code = float(le_action.transform([action])[0])
         
@@ -655,5 +718,24 @@ with gr.Blocks(title="模擬校園入口與資安防護系統") as demo:
             outputs=[store_df_all, store_df_abn, abnormal_log_input] 
         )
 
+    # =====================================================================
+    # 👉 新增：隱藏的 UI 綁定區 (讓外部能透過 api_name 存取資料)
+    # =====================================================================
+    with gr.Group(visible=False):
+        api_stats_btn = gr.Button("API_Stats")
+        api_stats_out = gr.JSON()
+        api_stats_btn.click(fn=api_get_stats, inputs=[], outputs=[api_stats_out], api_name="get_stats")
+        
+        api_logs_btn = gr.Button("API_Logs")
+        api_logs_in = gr.Number(value=20)
+        api_logs_out = gr.JSON()
+        api_logs_btn.click(fn=api_get_logs, inputs=[api_logs_in], outputs=[api_logs_out], api_name="get_logs")
+        
+        api_chart_btn = gr.Button("API_Chart")
+        api_chart_in = gr.Textbox()
+        api_chart_out = gr.JSON()
+        api_chart_btn.click(fn=api_get_chart_data, inputs=[api_chart_in], outputs=[api_chart_out], api_name="get_chart_data")
+    # =====================================================================
+
 if __name__ == "__main__":
     demo.launch(css=custom_css)