# 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