# plotting.py # 此檔案負責圖表繪製和可視化功能 # 包含統計儀表板、操作時間分佈圖等 import matplotlib.pyplot as plt import seaborn as sns import pandas as pd import numpy as np import io import base64 import os import models from config import ACTION_MAP, FONT_PATH from utils import setup_chinese_font def get_dashboard_html(): """生成統計儀表板 HTML (精緻版)""" total = models.global_stats["total_count"] abnormal = models.global_stats["abnormal_count"] rate = (abnormal / total * 100) if total > 0 else 0 if not models.AI_READY: status_icon = "🔴" status_text = "系統錯誤" status_color = "#ef4444" status_desc = "AI 模型載入失敗,防護停用中。" pulse_class = "pulse-red" elif rate > 10: status_icon = "🟡" status_text = "高風險警告" status_color = "#f59e0b" status_desc = "近期異常行為頻發,請立刻檢查 Log。" pulse_class = "pulse-yellow" else: status_icon = "🟢" status_text = "正常執行" status_color = "#10b981" status_desc = "HDBSCAN 與 Agent 服務運作正常。" pulse_class = "pulse-green" return f"""
總監測量 (本地歷史資料庫)
{total:,} 筆 Log
自系統啟動起統計 (+今日即時)
異常率 (平均數值)
{rate:.2f} % ↑即時
共 {abnormal:,} 筆監測到異常行為
系統狀態
{status_text}
{status_desc}
""" def get_plot_data_for_api(selected_key): """API 專用版本,返回 Base64 圖片字符串""" if not selected_key or not models.DATA_READY: return None action_code = ACTION_MAP.get(selected_key) try: plt.close('all') df_hist = models.baseline_df[models.baseline_df['action'] == action_code].copy() df_live = pd.DataFrame(models.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: fig, ax = plt.subplots(figsize=(12, 6)) if os.path.exists(FONT_PATH): plt.rc('font', family='SimHei') ax.text(0.5, 0.5, f"【{selected_key}】目前尚無資料", ha='center', va='center', fontsize=15) # 將圖片轉換為 Base64 buf = io.BytesIO() fig.savefig(buf, format='png', dpi=100, bbox_inches='tight') buf.seek(0) img_base64 = base64.b64encode(buf.getvalue()).decode('utf-8') buf.close() plt.close(fig) return f"data:image/png;base64,{img_base64}" 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) df_normal = df_combined[~is_abnormal] df_abnormal = df_combined[is_abnormal] fig, ax = plt.subplots(figsize=(12, 6)) sns.set_theme(style="whitegrid") if os.path.exists(FONT_PATH): plt.rc('font', family='SimHei') plt.rcParams['axes.unicode_minus'] = False has_data = False months_order = list(range(1, 13)) if not df_normal.empty: has_data = True sns.stripplot(data=df_normal, x='month', y='time_of_day', jitter=0.3, alpha=0.4, size=5, color='#10b981', order=months_order, ax=ax, label='正常資料') if not df_abnormal.empty: has_data = True sns.stripplot(data=df_abnormal, x='month', y='time_of_day', jitter=0.3, alpha=0.9, size=9, color='#ef4444', marker='X', order=months_order, ax=ax, label='異常資料') if not has_data: ax.text(0.5, 0.5, f"【{selected_key}】目前尚無資料", ha='center', va='center', fontsize=15) # 將圖片轉換為 Base64 buf = io.BytesIO() fig.savefig(buf, format='png', dpi=100, bbox_inches='tight') buf.seek(0) img_base64 = base64.b64encode(buf.getvalue()).decode('utf-8') buf.close() plt.close(fig) return f"data:image/png;base64,{img_base64}" ax.set_xlabel('月份 (1 ~ 12月)', fontsize=12) ax.set_ylabel('時間 (24小時制)', fontsize=12) ax.set_yticks(np.arange(0, 25, 2)) ax.set_ylim(24.5, -0.5) handles, labels = ax.get_legend_handles_labels() by_label = dict(zip(labels, handles)) if by_label: ax.legend(by_label.values(), by_label.keys(), loc='upper right', bbox_to_anchor=(1.15, 1)) plt.tight_layout() # 將圖片轉換為 Base64 buf = io.BytesIO() fig.savefig(buf, format='png', dpi=100, bbox_inches='tight') buf.seek(0) img_base64 = base64.b64encode(buf.getvalue()).decode('utf-8') buf.close() plt.close(fig) return f"data:image/png;base64,{img_base64}" except Exception as e: print(f"繪圖失敗: {e}") # 返回一個簡單的錯誤圖片 fig, ax = plt.subplots(figsize=(8, 4)) ax.text(0.5, 0.5, f"圖表生成失敗: {str(e)}", ha='center', va='center', fontsize=12, color='red') ax.set_xlim(0, 1) ax.set_ylim(0, 1) ax.axis('off') buf = io.BytesIO() fig.savefig(buf, format='png', dpi=100, bbox_inches='tight') buf.seek(0) img_base64 = base64.b64encode(buf.getvalue()).decode('utf-8') buf.close() plt.close(fig) return f"data:image/png;base64,{img_base64}" def get_plot_data(selected_key): """給 Gradio 界面使用,返回 Figure 對象""" if not selected_key or not models.DATA_READY: return None action_code = ACTION_MAP.get(selected_key) try: plt.close('all') df_hist = models.baseline_df[models.baseline_df['action'] == action_code].copy() df_live = pd.DataFrame(models.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: fig, ax = plt.subplots(figsize=(12, 6)) if os.path.exists(FONT_PATH): plt.rc('font', family='SimHei') ax.text(0.5, 0.5, f"【{selected_key}】目前尚無資料", ha='center', va='center', fontsize=15) return fig 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) df_normal = df_combined[~is_abnormal] df_abnormal = df_combined[is_abnormal] fig, ax = plt.subplots(figsize=(12, 6)) sns.set_theme(style="whitegrid") if os.path.exists(FONT_PATH): plt.rc('font', family='SimHei') plt.rcParams['axes.unicode_minus'] = False has_data = False months_order = list(range(1, 13)) if not df_normal.empty: has_data = True sns.stripplot(data=df_normal, x='month', y='time_of_day', jitter=0.3, alpha=0.4, size=5, color='#10b981', order=months_order, ax=ax, label='正常資料') if not df_abnormal.empty: has_data = True sns.stripplot(data=df_abnormal, x='month', y='time_of_day', jitter=0.3, alpha=0.9, size=9, color='#ef4444', marker='X', order=months_order, ax=ax, label='異常資料') if not has_data: ax.text(0.5, 0.5, f"【{selected_key}】目前尚無資料", ha='center', va='center', fontsize=15) return fig ax.set_xlabel('月份 (1 ~ 12月)', fontsize=12) ax.set_ylabel('時間 (24小時制)', fontsize=12) ax.set_yticks(np.arange(0, 25, 2)) ax.set_ylim(24.5, -0.5) handles, labels = ax.get_legend_handles_labels() by_label = dict(zip(labels, handles)) if by_label: ax.legend(by_label.values(), by_label.keys(), loc='upper right', bbox_to_anchor=(1.15, 1)) plt.tight_layout() return fig except Exception as e: print(f"繪圖失敗: {e}") return None def timer_update(selected_key): """定時更新函數""" return get_dashboard_html(), get_plot_data(selected_key) def get_chart_js_html(selected_key): """生成 Chart.js HTML 內容 (Gradio 專用)""" import models import numpy as np import json from config import ACTION_MAP import pandas as pd if not selected_key or not models.DATA_READY: return f'
【{selected_key}】系統目前尚無資料
' # 取得代碼 action_code = ACTION_MAP.get(selected_key) print(f"🔍 [Chart Debug] Selected: '{selected_key}', Mapped Code: '{action_code}'") if not action_code: # 嘗試模糊匹配 for k, v in ACTION_MAP.items(): if selected_key in k or k in selected_key: action_code = v print(f"💡 [Chart Debug] Fuzzy matched to: '{action_code}'") break if not action_code: action_code = selected_key # 最後手段:直接用名稱查 # 取得資料 df_hist = models.baseline_df[models.baseline_df['action'] == action_code].copy() print(f"📊 [Chart Debug] Found {len(df_hist)} historical records for '{action_code}'") df_live = pd.DataFrame(models.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() print(f"📈 [Chart Debug] Found {len(df_live)} live records for '{action_code}'") 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: print(f"⚠️ [Chart Debug] Combined data is empty for '{action_code}'") return f'
【{selected_key}】系統目前尚無資料
' 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) # 準備資料點 (添加 Jitter) normal_pts = [] for _, row in df_combined[~is_abnormal].iterrows(): normal_pts.append({"x": float(row['month']) + (np.random.rand()-0.5)*0.4, "y": float(row['time_of_day']) + (np.random.rand()-0.5)*0.5}) abnormal_pts = [] for _, row in df_combined[is_abnormal].iterrows(): abnormal_pts.append({"x": float(row['month']) + (np.random.rand()-0.5)*0.4, "y": float(row['time_of_day']) + (np.random.rand()-0.5)*0.5}) normal_json = json.dumps(normal_pts) abnormal_json = json.dumps(abnormal_pts) # 使用隨機 ID 避免 Gradio 多次渲染衝突 import time chart_id = f"chart_{int(time.time() * 1000)}" html = f"""
""" return html