app.py

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 = logger_service.sync_from_hf()
    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(value=pd.DataFrame(log_history),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(allowed_paths=["*"])