Spaces:
Sleeping
Sleeping
| # 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""" | |
| <div class="stat-dashboard"> | |
| <!-- 卡片 1: 總監測量 --> | |
| <div class="stat-card card-total"> | |
| <div class="card-icon-container"> | |
| <div class="icon-bg" style="background-color: #f0fdf4;"> | |
| <svg width="24" height="24" viewBox="0 0 24 24" fill="none" stroke="#10b981" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M18 20V10"></path><path d="M12 20V4"></path><path d="M6 20v-6"></path></svg> | |
| </div> | |
| </div> | |
| <div class="card-content"> | |
| <div class="card-label">總監測量 (本地歷史資料庫)</div> | |
| <div class="card-value-group"> | |
| <span class="card-value">{total:,}</span> | |
| <span class="card-unit">筆 Log</span> | |
| </div> | |
| <div class="card-sub-text">自系統啟動起統計 (+今日即時)</div> | |
| </div> | |
| </div> | |
| <!-- 卡片 2: 異常率 --> | |
| <div class="stat-card card-anomaly"> | |
| <div class="card-icon-container"> | |
| <div class="icon-bg" style="background-color: #fffbeb;"> | |
| <svg width="24" height="24" viewBox="0 0 24 24" fill="none" stroke="#f59e0b" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M10.29 3.86L1.82 18a2 2 0 0 0 1.71 3h16.94a2 2 0 0 0 1.71-3L13.71 3.86a2 2 0 0 0-3.42 0z"></path><line x1="12" y1="9" x2="12" y2="13"></line><line x1="12" y1="17" x2="12" y2="17.01"></line></svg> | |
| </div> | |
| </div> | |
| <div class="card-content"> | |
| <div class="card-label">異常率 (平均數值)</div> | |
| <div class="card-value-group"> | |
| <span class="card-value" style="color: #ef4444;">{rate:.2f} %</span> | |
| <span class="card-trend trend-up">↑即時</span> | |
| </div> | |
| <div class="card-sub-text">共 {abnormal:,} 筆監測到異常行為</div> | |
| </div> | |
| </div> | |
| <!-- 卡片 3: 系統狀態 --> | |
| <div class="stat-card card-status"> | |
| <div class="card-icon-container"> | |
| <div class="status-dot-container"> | |
| <div class="status-dot {pulse_class}"></div> | |
| </div> | |
| </div> | |
| <div class="card-content"> | |
| <div class="card-label">系統狀態</div> | |
| <div class="card-value-group"> | |
| <span class="card-value" style="color: {status_color};">{status_text}</span> | |
| </div> | |
| <div class="card-sub-text">{status_desc}</div> | |
| </div> | |
| </div> | |
| </div> | |
| """ | |
| 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'<div style="text-align:center; padding:40px; color:#94a3b8;">【{selected_key}】系統目前尚無資料</div>' | |
| # 取得代碼 | |
| 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'<div style="text-align:center; padding:40px; color:#94a3b8;">【{selected_key}】系統目前尚無資料</div>' | |
| 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""" | |
| <div style="width: 100%; height: 450px; background: white; border-radius: 12px; padding: 20px; box-shadow: 0 4px 6px -1px rgba(0,0,0,0.1); margin-top: 10px;"> | |
| <canvas id="{chart_id}"></canvas> | |
| <script> | |
| (function() {{ | |
| const initChart = () => {{ | |
| const canvas = document.getElementById('{chart_id}'); | |
| if (!canvas) return; | |
| const ctx = canvas.getContext('2d'); | |
| if (typeof Chart === 'undefined') {{ | |
| setTimeout(initChart, 500); | |
| return; | |
| }} | |
| console.log("Rendering Gradio Chart: {selected_key}"); | |
| new Chart(ctx, {{ | |
| type: 'scatter', | |
| data: {{ | |
| datasets: [ | |
| {{ | |
| label: '正常行為', | |
| data: {normal_json}, | |
| backgroundColor: 'rgba(16, 185, 129, 0.4)', | |
| pointRadius: 5 | |
| }}, | |
| {{ | |
| label: '異常行為', | |
| data: {abnormal_json}, | |
| backgroundColor: 'rgba(239, 68, 68, 0.8)', | |
| pointRadius: 7, | |
| pointStyle: 'rectRot' | |
| }} | |
| ] | |
| }}, | |
| options: {{ | |
| responsive: true, | |
| maintainAspectRatio: false, | |
| animation: {{ duration: 800 }}, | |
| scales: {{ | |
| x: {{ title: {{ display: true, text: '月份' }}, min: 0.5, max: 12.5, ticks: {{ stepSize: 1 }} }}, | |
| y: {{ title: {{ display: true, text: '時間 (0-24h)' }}, min: 0, max: 24, reverse: true, ticks: {{ stepSize: 4 }} }} | |
| }}, | |
| plugins: {{ | |
| legend: {{ position: 'top' }} | |
| }} | |
| }} | |
| }}); | |
| }}; | |
| if (typeof Chart === 'undefined') {{ | |
| const script = document.createElement('script'); | |
| script.src = 'https://cdn.jsdelivr.net/npm/chart.js'; | |
| script.onload = initChart; | |
| document.head.appendChild(script); | |
| }} else {{ | |
| initChart(); | |
| }} | |
| }})(); | |
| </script> | |
| </div> | |
| """ | |
| return html | |