auth.py

# auth.py
# 此檔案負責身份驗證相關功能
# 包含登入處理、會話管理、角色切換等

import datetime
import threading
import secrets
from config import USER_DB, FIXED_TEST_EMAIL
from models import record_log, global_stats
from utils import get_current_time
import gradio as gr

# --- Identity verification session cache ---
# key: user_email
# value: { status(active/verifying/blocked), pending_token, anomaly_timestamp }
session_cache = {}
session_cache_lock = threading.Lock()

def _get_or_create_session_entry(user_email: str):
    """獲取或創建會話條目"""
    user_email = user_email.strip().replace("\xa0", " ")
    with session_cache_lock:
        if user_email not in session_cache:
            session_cache[user_email] = {
                "status": "active",
                "pending_token": None,
                "anomaly_timestamp": None,
            }
        return session_cache[user_email]

def check_session_timeouts(timeout_seconds=300):
    """檢查會話超時，如果 verifying 超過 timeout_seconds 則改為 blocked"""
    now = datetime.datetime.now()
    with session_cache_lock:
        for user_email, entry in session_cache.items():
            if entry.get("status") == "verifying" and entry.get("anomaly_timestamp"):
                try:
                    # 嘗試解析多種可能的格式
                    ts_str = entry["anomaly_timestamp"]
                    try:
                        ts = datetime.datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S")
                    except ValueError:
                        # 處理可能帶有時區或其他格式的情況
                        import pandas as pd
                        ts = pd.to_datetime(ts_str).to_pydatetime().replace(tzinfo=None)
                    
                    if (now - ts).total_seconds() > timeout_seconds:
                        entry["status"] = "blocked"
                        entry["pending_token"] = None
                        print(f"⏰ [Timeout] User {user_email} status changed to blocked due to verification timeout.")
                except Exception as e:
                    print(f"❌ [Timeout Check Error] {e}")

def get_user_status(user_email: str):
    """獲取用戶當前狀態"""
    user_email = user_email.strip().replace("\xa0", " ")
    with session_cache_lock:
        if user_email not in session_cache:
            return "active"
        return session_cache[user_email].get("status", "active")

def unblock_user(user_email: str):
    """手動解封用戶（以帳戶名稱作為 key）"""
    user_email = user_email.strip().replace("\xa0", " ")
    with session_cache_lock:
        if user_email in session_cache:
            session_cache[user_email]["status"] = "active"
            session_cache[user_email]["pending_token"] = None
            session_cache[user_email]["anomaly_timestamp"] = None
            return True
        else:
            # 就算 session_cache 中沒有此 key，也視為解鎖成功（帳戶原本就是 active）
            return True

def _verify_token_to_user(token: str):
    """根據 token 驗證用戶"""
    token = token.strip()
    with session_cache_lock:
        for user_email, entry in session_cache.items():
            if entry.get("pending_token") == token:
                return user_email
    return None

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)
        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)
        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)
        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
    else:
        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 internal_process_login(ip, cookie, account, password, captcha, time_mode, custom_time):
    """內部處理登入邏輯，回傳 JSON 友善格式"""
    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)
        return {"status": "error", "message": "帳號不存在！"}

    if get_user_status(account) == "blocked":
        df_all, df_abn, inp_upd = record_log(ip, cookie, account, USER_DB[account][1], "login_attempt", "failed_blocked", time_mode, custom_time)
        return {"status": "error", "message": "此帳號已被封鎖，無法登入！"}

    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)
        return {"status": "error", "message": "驗證碼錯誤！"}

    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)
        return {
            "status": "success", 
            "role": "admin" if role == "管理" else "student",
            "name": name,
            "message": "登入成功"
        }
    else:
        df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_password", time_mode, custom_time)
        return {"status": "error", "message": "密碼錯誤！"}

def internal_logout(ip, cookie, account, time_mode, custom_time):
    """內部處理登出邏輯"""
    role = USER_DB.get(account, ["", "Unknown"])[1]
    record_log(ip, cookie, account, role, "logout", "success", time_mode, custom_time)
    return {"status": "success", "message": "已登出"}

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)
    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 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)
    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):
    """登出"""
    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)
    return gr.update(selected="login_route"), "", "", "", "", df_all, df_abn, inp_upd

def simulate_click(ip, cookie, account, button_name, time_mode, custom_time):
    """模擬點擊操作"""
    # 以帳戶名稱查詢封鎖狀態（不再使用全域 FIXED_TEST_EMAIL 作為 key）
    status = get_user_status(account)
    if status == "verifying":
        gr.Info("偵測到異常，請於5分鐘內確認 Email 以繼續使用")
        return gr.update(), gr.update(), gr.update()
    elif status == "blocked":
        gr.Warning("帳號已被封鎖，請聯絡系統管理員或寄驗證信進行解封")
        return gr.update(), gr.update(), gr.update()

    role = USER_DB.get(account, ["", "Unknown"])[1]
    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):
    """測試異常偵測"""
    # 測試異常時不阻擋，以便觀察結果
    role = USER_DB.get(account, ["", "Unknown"])[1]
    return record_log(ip, cookie, account, role, action_name, "success", "自訂時間 (模擬過去/未來)", test_time_str)