================================================================================ PROJECT SOURCE CODE EXPORT (FULL) ================================================================================ ================================================================================ FILE: .env.example ================================================================================ # --- .env.example --- # Database Configuration (請填入你的本地端或正式機設定) DB_NAME=cmts_nms DB_USER= DB_PASS= DB_HOST=127.0.0.1 DB_PORT=5432 # God Mode Secret GOD_MODE_SECRET= # Default CMTS Device DEFAULT_CMTS_HOST= DEFAULT_CMTS_USER= DEFAULT_CMTS_PASS= ================================================================================ FILE: .env ================================================================================ # --- .env --- # Database Configuration DB_NAME=cmts_nms DB_USER=swpa DB_PASS=swpa4920 DB_HOST=127.0.0.1 DB_PORT=5432 # God Mode Secret GOD_MODE_SECRET=swpa@Serc0mm # Default CMTS Device (本地開發用，正式環境可留空) DEFAULT_CMTS_HOST=10.14.110.4 DEFAULT_CMTS_USER=admin DEFAULT_CMTS_PASS=nsgadmin ================================================================================ FILE: database.py ================================================================================ import asyncpg import json import uuid import os from dotenv import load_dotenv # 🌟 1. 引入 load_dotenv from typing import Dict, List, Optional, Any from logger import get_logger # 🌟 2. 明確指示 Python 讀取同目錄下的 .env 檔案 load_dotenv() # ========================================== # 💡 PostgreSQL 連線與操作 (高可用性版) # ========================================== logger = get_logger("app.database") # 🌟 2. 同步改用 os.getenv 讀取環境變數 DB_CONFIG = { "database": os.getenv("DB_NAME", "cmts_nms"), "user": os.getenv("DB_USER", "postgres"), # 本地開發常用的預設帳號，或留空 "" "password": os.getenv("DB_PASS", ""), # 🌟 絕對機密：預設留空！ "host": os.getenv("DB_HOST", "127.0.0.1"), "port": os.getenv("DB_PORT", "5432") } _pool: Optional[asyncpg.Pool] = None async def init_db_pool(): """初始化非同步連線池""" global _pool try: if _pool is None: _pool = await asyncpg.create_pool( database=DB_CONFIG["database"], user=DB_CONFIG["user"], password=DB_CONFIG["password"], host=DB_CONFIG["host"], port=int(DB_CONFIG["port"]), min_size=1, max_size=10 ) logger.info("✅ PostgreSQL Connection Pool Initialized.") except Exception as e: logger.error(f"❌ Failed to initialize DB Pool: {e}") async def close_db_pool(): """關閉非同步連線池""" global _pool if _pool: await _pool.close() _pool = None logger.info("🔌 PostgreSQL Connection Pool Closed.") async def get_pool() -> Optional[asyncpg.Pool]: if _pool is None: await init_db_pool() return _pool # ------------------------------------------ # CRUD Functions for cmts_options # ------------------------------------------ async def upsert_leaf_option(host: str, path: str, data: dict) -> bool: pool = await get_pool() if not pool: return False query = """ INSERT INTO cmts_options (host, path, data) VALUES ($1, $2, $3) ON CONFLICT (host, path) DO UPDATE SET data = EXCLUDED.data, updated_at = CURRENT_TIMESTAMP; """ try: async with pool.acquire() as conn: await conn.execute(query, host, path, json.dumps(data)) return True except Exception as e: logger.error(f"❌ DB Error (upsert_leaf_option): {e}") return False async def get_all_leaf_options(host: str) -> Optional[Dict[str, Any]]: pool = await get_pool() if not pool: return None query = "SELECT path, data FROM cmts_options WHERE host = $1;" try: async with pool.acquire() as conn: records = await conn.fetch(query, host) result = {} for record in records: data_val = record['data'] result[record['path']] = json.loads(data_val) if isinstance(data_val, str) else data_val return result except Exception as e: logger.error(f"❌ DB Error (get_all_leaf_options): {e}") return None async def delete_leaf_options(host: str, paths: List[str]) -> int: pool = await get_pool() if not pool or not paths: return -1 query = "DELETE FROM cmts_options WHERE host = $1 AND path = ANY($2);" try: async with pool.acquire() as conn: status = await conn.execute(query, host, paths) return int(status.split()[-1]) except Exception as e: logger.error(f"❌ DB Error (delete_leaf_options): {e}") return -1 # ------------------------------------------ # CRUD Functions for device_status # ------------------------------------------ async def upsert_device_status(host: str, metadata: dict) -> bool: pool = await get_pool() if not pool: return False cmts_version = metadata.get("cmts_version") last_scanned = metadata.get("last_scanned", None) try: async with pool.acquire() as conn: if cmts_version and cmts_version != "unknown": query = """ INSERT INTO device_status (host, cmts_version, last_scanned) VALUES ($1, $2, $3) ON CONFLICT (host) DO UPDATE SET cmts_version = EXCLUDED.cmts_version, last_scanned = EXCLUDED.last_scanned, updated_at = CURRENT_TIMESTAMP; """ await conn.execute(query, host, cmts_version, last_scanned) else: query = """ INSERT INTO device_status (host, last_scanned) VALUES ($1, $2) ON CONFLICT (host) DO UPDATE SET last_scanned = EXCLUDED.last_scanned, updated_at = CURRENT_TIMESTAMP; """ await conn.execute(query, host, last_scanned) return True except Exception as e: logger.error(f"❌ DB Error (upsert_device_status): {e}") return False async def get_device_status(host: str) -> Optional[Dict[str, Any]]: pool = await get_pool() if not pool: return None query = "SELECT cmts_version, last_scanned FROM device_status WHERE host = $1;" try: async with pool.acquire() as conn: record = await conn.fetchrow(query, host) if record: return {"cmts_version": record["cmts_version"], "last_scanned": record["last_scanned"]} return None except Exception as e: logger.error(f"❌ DB Error (get_device_status): {e}") return None # ------------------------------------------ # CRUD Functions for system_filters (Tree Filters) # ------------------------------------------ async def upsert_tree_filters(config_type: str, hidden_keys: List[str]) -> bool: """寫入全域的樹狀圖隱藏節點名單""" pool = await get_pool() if not pool: return False # Use 'global' as a dummy host to keep schema simple if needed, # but a dedicated system_filters table is better. query = """ INSERT INTO system_filters (config_type, hidden_keys) VALUES ($1, $2) ON CONFLICT (config_type) DO UPDATE SET hidden_keys = EXCLUDED.hidden_keys, updated_at = CURRENT_TIMESTAMP; """ try: async with pool.acquire() as conn: await conn.execute(query, config_type, hidden_keys) return True except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (upsert_tree_filters): {e}") return False except Exception as e: logger.error(f"❌ Unknown Error (upsert_tree_filters): {e}") return False async def get_tree_filters(config_type: str) -> Optional[List[str]]: """讀取全域的樹狀圖隱藏節點名單""" pool = await get_pool() if not pool: return None query = """ SELECT hidden_keys FROM system_filters WHERE config_type = $1; """ try: async with pool.acquire() as conn: record = await conn.fetchrow(query, config_type) if record: return record["hidden_keys"] return [] except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (get_tree_filters): {e}") return None except Exception as e: logger.error(f"❌ Unknown Error (get_tree_filters): {e}") return None # ========================================== # CRUD Functions for config_backups (Phase 1 & 2) # ========================================== async def insert_config_backup( host: str, config_type: str, raw_cli: str, parsed_tree: dict, snapshot_name: Optional[str] = None, description: str = "", # 🟢 新增描述參數 (預設為空字串) is_auto: bool = False ) -> Optional[str]: """新增一筆設備配置備份，回傳產生的 Backup ID""" pool = await get_pool() if not pool: return None backup_id = str(uuid.uuid4()) # 🟢 SQL 語句加入 description 與對應的 $5 query = """ INSERT INTO config_backups (id, host, config_type, snapshot_name, description, is_auto, raw_cli, parsed_tree) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) """ try: async with pool.acquire() as conn: await conn.execute( query, backup_id, host, config_type, snapshot_name, description, # 🟢 傳入描述參數 is_auto, raw_cli, json.dumps(parsed_tree) ) return backup_id except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (insert_config_backup): {e}") return None except Exception as e: logger.error(f"❌ Unknown Error (insert_config_backup): {e}") return None async def get_config_backup_list(host: str, config_type: str) -> Optional[List[Dict[str, Any]]]: """取得歷史快照列表 (支援 config_type='all' 撈取全部，包含 is_pinned 狀態)""" pool = await get_pool() if not pool: return None try: async with pool.acquire() as conn: if config_type == "all": # 🌟 SQL 查詢補上 is_pinned query = """ SELECT id, host, config_type, timestamp, snapshot_name, description, is_auto, is_pinned FROM config_backups WHERE host = $1 ORDER BY timestamp DESC; """ records = await conn.fetch(query, host) else: # 🌟 SQL 查詢補上 is_pinned query = """ SELECT id, host, config_type, timestamp, snapshot_name, description, is_auto, is_pinned FROM config_backups WHERE host = $1 AND config_type = $2 ORDER BY timestamp DESC; """ records = await conn.fetch(query, host, config_type) return [ { "id": str(r["id"]), "host": r["host"], "config_type": r["config_type"], "timestamp": r["timestamp"].isoformat(), "snapshot_name": r["snapshot_name"], "description": r["description"], "is_auto": r["is_auto"], "is_pinned": r["is_pinned"] # 🌟 放入回傳字典 } for r in records ] except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (get_config_backup_list): {e}") return None except Exception as e: logger.error(f"❌ Unknown Error (get_config_backup_list): {e}") return None async def get_config_backup_detail(backup_id: str) -> Optional[Dict[str, Any]]: """取得特定快照的完整內容 (包含 parsed_tree，供 Phase 3 還原使用)""" pool = await get_pool() if not pool: return None # 🟢 SELECT 加入 description query = """ SELECT id, host, timestamp, snapshot_name, description, is_auto, parsed_tree FROM config_backups WHERE id = $1; """ try: async with pool.acquire() as conn: record = await conn.fetchrow(query, backup_id) if record: data_val = record['parsed_tree'] parsed_tree = json.loads(data_val) if isinstance(data_val, str) else data_val return { "id": str(record["id"]), "host": record["host"], "timestamp": record["timestamp"].isoformat(), "snapshot_name": record["snapshot_name"], "description": record["description"], # 🟢 將資料庫的描述放入回傳字典 "is_auto": record["is_auto"], "parsed_tree": parsed_tree } return None except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (get_config_backup_detail): {e}") return None except Exception as e: logger.error(f"❌ Unknown Error (get_config_backup_detail): {e}") return None async def delete_config_backup(backup_id: str) -> bool: """刪除指定的快照""" pool = await get_pool() if not pool: return False query = "DELETE FROM config_backups WHERE id = $1;" try: async with pool.acquire() as conn: status = await conn.execute(query, backup_id) # status 會是 'DELETE 1' 或 'DELETE 0' return int(status.split()[-1]) > 0 except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (delete_config_backup): {e}") return False except Exception as e: logger.error(f"❌ Unknown Error (delete_config_backup): {e}") return False # ============================================================================ # 🧹 備份滾動淘汰機制 (Retention Policy) # ============================================================================ async def cleanup_old_backups(pool, host: str) -> None: """ 執行手動備份的滾動淘汰機制 (Retention Policy) - 每個設備 (host) 保留最新 20 筆未釘選 (is_pinned=FALSE) 的手動備份 - 釘選的備份 (is_pinned=True) 永遠不刪除 """ if not pool: logger.warning("⚠️ [Retention] 無法執行備份清理：Database Pool 未初始化。") return # 使用 PostgreSQL CTE (Common Table Expression) 語法 # 先選出該設備最新 20 筆需要保留的 ID，再將其餘未釘選的舊備份一次性刪除 query = """ WITH kept_backups AS ( SELECT id FROM config_backups WHERE host = $1 AND is_pinned = FALSE ORDER BY timestamp DESC LIMIT 20 ) DELETE FROM config_backups WHERE host = $1 AND is_pinned = FALSE AND id NOT IN (SELECT id FROM kept_backups) RETURNING id; """ try: async with pool.acquire() as conn: deleted_records = await conn.fetch(query, host) total_deleted = len(deleted_records) if total_deleted > 0: logger.info(f"🧹 [Retention Policy] 設備 {host} 清理完畢：已自動淘汰 {total_deleted} 筆過期手動備份。") else: logger.debug(f"ℹ️ [Retention Policy] 設備 {host} 備份數量未達 20 筆上限，無需清理。") except Exception as e: logger.error(f"❌ [Retention Policy] 清理設備 {host} 過期備份時發生錯誤: {e}", exc_info=True) # ============================================================================ # 📌 釘選防護與容量指標計算 (Pinning & Metrics) # ============================================================================ async def toggle_config_backup_pin(backup_id: str) -> Optional[bool]: """ 切換特定備份的釘選狀態 (True -> False, False -> True) 回傳更新後的 is_pinned 狀態，若失敗則回傳 None """ pool = await get_pool() if not pool: return None # 使用 PostgreSQL 的 UPDATE ... RETURNING 語法，一步完成切換與讀取，保證原子性 query = """ UPDATE config_backups SET is_pinned = NOT is_pinned WHERE id = $1 RETURNING is_pinned; """ try: async with pool.acquire() as conn: new_state = await conn.fetchval(query, backup_id) return new_state except Exception as e: logger.error(f"❌ DB Error (toggle_config_backup_pin): {e}") return None async def get_backup_metrics(pool, host: str) -> dict: """ 計算特定設備的備份容量指標 - total: 總備份數 (含釘選與未釘選) - pinned: 已釘選保護的數量 - unpinned: 未釘選的數量 (上限為 20 筆) - remaining: 剩餘可用手動備份額度 (20 - unpinned) """ if not pool: return {"total": 0, "pinned": 0, "unpinned": 0, "remaining": 20} query = """ SELECT COUNT(*) as total, COUNT(*) FILTER (WHERE is_pinned = TRUE) as pinned, COUNT(*) FILTER (WHERE is_pinned = FALSE) as unpinned FROM config_backups WHERE host = $1; """ try: async with pool.acquire() as conn: r = await conn.fetchrow(query, host) total = r["total"] or 0 pinned = r["pinned"] or 0 unpinned = r["unpinned"] or 0 remaining = max(0, 20 - unpinned) return { "total": total, "pinned": pinned, "unpinned": unpinned, "remaining": remaining } except Exception as e: logger.error(f"❌ DB Error (get_backup_metrics): {e}") return {"total": 0, "pinned": 0, "unpinned": 0, "remaining": 20} ================================================================================ FILE: shared.py ================================================================================ # --- shared.py --- import asyncio import json import os import re from collections import defaultdict DEBUG_MODE = False def debug_print(msg: str): if DEBUG_MODE: print(msg) # 🌟 2. 淨化預設的 CMTS 連線樣板 CMTS_DEVICE = { 'device_type': 'cisco_ios', 'host': os.getenv("DEFAULT_CMTS_HOST", ""), 'username': os.getenv("DEFAULT_CMTS_USER", ""), 'password': os.getenv("DEFAULT_CMTS_PASS", ""), 'port': 22, 'fast_cli': False, 'global_delay_factor': 2 } def parse_cli_to_tree(cli_text: str) -> dict: """ 將 CMTS 的 CLI 配置文字轉換為完美的邏輯巢狀 JSON (兩階段解析法) """ # ========================================== # 🌟 階段 0：預處理 (清理雜訊，移除危險的自動黏合) # ========================================== cli_text = cli_text.replace('\r\n', '\n').replace('\r', '\n') ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') raw_lines = cli_text.split('\n') fixed_lines = [] for line in raw_lines: line = ansi_escape.sub('', line) line = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]', '', line) if not line.strip(): continue # 💡 乾淨俐落：直接加入陣列，不再做危險的字串長度黏合判斷 fixed_lines.append(line) lines = [line.rstrip() for line in fixed_lines if line.strip()] # ========================================== # 🌟 階段 1：建立實體樹 (嚴格比對縮排與 '!' 的對應關係) # ========================================== def build_raw_tree(start_idx, current_indent): nodes = [] i = start_idx while i < len(lines): line = lines[i] stripped = line.strip() indent = len(line) - len(line.lstrip()) if indent < current_indent and stripped != '!': return nodes, i if stripped == '!': i += 1 continue has_children = False next_i = i + 1 while next_i < len(lines) and lines[next_i].strip() == '!': next_i += 1 if next_i < len(lines): next_line = lines[next_i] next_indent = len(next_line) - len(next_line.lstrip()) if next_indent > indent: has_children = True if has_children: children, next_i = build_raw_tree(i + 1, next_indent) nodes.append({"type": "folder", "content": stripped, "children": children}) i = next_i else: nodes.append({"type": "leaf", "content": stripped}) i += 1 return nodes, i raw_tree, _ = build_raw_tree(0, 0) # ========================================== # 🌟 階段 2：邏輯群組化與字典合併 (保留嚴格型別防護) # ========================================== def deep_merge(dict1, dict2): for k, v in dict2.items(): if k in dict1: if isinstance(dict1[k], dict) and isinstance(v, dict): deep_merge(dict1[k], v) else: idx = 1 while f"{k} [{idx}]" in dict1: idx += 1 dict1[f"{k} [{idx}]"] = v else: dict1[k] = v def nest_folder_key(key_string, value_dict, target_dict): parts = key_string.split() current = target_dict for i, part in enumerate(parts): if i == len(parts) - 1: if part not in current: current[part] = value_dict else: if isinstance(current[part], dict) and isinstance(value_dict, dict): deep_merge(current[part], value_dict) else: idx = 1 while f"{part} (衝突 {idx})" in current: idx += 1 current[f"{part} (衝突 {idx})"] = value_dict else: if part not in current or not isinstance(current[part], dict): if part in current and not isinstance(current[part], dict): old_val = current[part] current[part] = {"[0]": old_val} else: current[part] = {} current = current[part] def group_leaves(leaf_strings): grouped = {} first_word_map = defaultdict(list) for s in leaf_strings: parts = s.split(maxsplit=1) first_word = parts[0] remainder = parts[1] if len(parts) > 1 else "" first_word_map[first_word].append(remainder) for prefix, remainders in first_word_map.items(): if len(remainders) == 1: grouped[prefix] = remainders[0] else: valid_remainders = [r for r in remainders if r] empty_count = len(remainders) - len(valid_remainders) group_dict = {} if valid_remainders: sub_grouped = group_leaves(valid_remainders) is_all_empty_vals = all(not isinstance(v, dict) and v == "" for v in sub_grouped.values()) if is_all_empty_vals: for i, k in enumerate(sub_grouped.keys()): group_dict[f"[{i}]"] = k else: idx = 0 for k, v in sub_grouped.items(): if isinstance(v, dict): group_dict[k] = v elif v == "": group_dict[f"[{idx}]"] = k idx += 1 else: group_dict[k] = v for _ in range(empty_count): idx = 0 while f"[{idx}]" in group_dict: idx += 1 group_dict[f"[{idx}]"] = "" grouped[prefix] = group_dict return grouped def fold_nodes(nodes): result = {} for node in [n for n in nodes if n["type"] == "folder"]: folded_children = fold_nodes(node["children"]) nest_folder_key(node["content"], folded_children, result) leaves = [n["content"] for n in nodes if n["type"] == "leaf"] if leaves: leaf_dict = group_leaves(leaves) deep_merge(result, leaf_dict) return result return fold_nodes(raw_tree) def deep_split_tree(data): """將平坦的 CLI 字典，依照空白字元拆分成深層巢狀結構 (具備終極型態衝突保護)""" if not isinstance(data, dict): return data nested_tree = {} for key, value in data.items(): processed_value = deep_split_tree(value) parts = str(key).strip().split() if not parts: continue current = nested_tree for i, part in enumerate(parts): if i == len(parts) - 1: if part not in current: current[part] = processed_value else: # 💡 終極衝突保護：安全轉型，絕不遺失資料 if isinstance(current[part], dict) and isinstance(processed_value, dict): current[part].update(processed_value) elif isinstance(current[part], dict) and not isinstance(processed_value, dict): # 原本是資料夾，新來的是字串 -> 塞入虛擬 key [0], [1] idx = 0 while f"[{idx}]" in current[part]: idx += 1 current[part][f"[{idx}]"] = processed_value elif not isinstance(current[part], dict) and isinstance(processed_value, dict): # 原本是字串，新來的是資料夾 -> 升級成資料夾，並保留原字串至 [0] old_val = current[part] current[part] = processed_value idx = 0 while f"[{idx}]" in current[part]: idx += 1 current[part][f"[{idx}]"] = old_val else: # 兩者都是字串 -> 升級成資料夾，包含兩個字串 old_val = current[part] current[part] = {"[0]": old_val, "[1]": processed_value} else: if part not in current: current[part] = {} elif not isinstance(current[part], dict): # 中間節點原本是字串，必須升級為資料夾，並保留原字串 old_val = current[part] current[part] = {"[0]": old_val} current = current[part] return nested_tree # ========================================== # 💡 系統設定管理 (System Settings Manager) # ========================================== SETTINGS_FILE = "system_settings.json" def load_settings(): """讀取系統設定，若檔案不存在則回傳預設的黑名單""" if os.path.exists(SETTINGS_FILE): try: with open(SETTINGS_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass # 預設隱藏這些比較雜亂且不常修改的系統層級設定 return {"hidden_keys": ["logging", "privilege", "aaa", "certificate", "line"]} def save_settings(settings_data): """將系統設定寫入 JSON 檔案""" with open(SETTINGS_FILE, "w", encoding="utf-8") as f: json.dump(settings_data, f, indent=4, ensure_ascii=False) # 🌟 [Priority 1 修復] 將全域非同步鎖改為依設備 IP (Host) 隔離的鎖 (安全動態獲取版) _cmts_config_locks = {} def get_cmts_lock(host: str) -> asyncio.Lock: if host not in _cmts_config_locks: _cmts_config_locks[host] = asyncio.Lock() return _cmts_config_locks[host] ================================================================================ FILE: index.html ================================================================================ Harmonic CMTS Manager

🎙️ Harmonic CMTS Admin

狀態：未連線

Cable 狀態綜合查詢 (show cable modem...)

RPD 狀態綜合查詢 (show cable rpd...)

CM 深度診斷與 MER 分析

⚠️ MAC Domain 狀態感知配置精靈

請先讀取設備狀態...

⚙️ 全域系統設定 (God Mode Filters)

請勾選您希望在「完整設備配置樹狀圖」中 隱藏 的設定項目。
💡 點擊下方按鈕載入設備實時配置，您可以深入展開並勾選任何層級的節點進行過濾。

🎛️ 伺服器日誌管控 (Server Log Management)

在此動態調整各個後端模組的日誌輸出等級。設定會立即生效，無需重啟伺服器。
💡 建議平時保持在 INFO 或 ERROR，僅在需要排查問題時開啟 DEBUG。

⏳ 正在載入日誌設定...

🎙️ 系統實時日誌 (System Live Logs)
狀態：已暫停 ⚪

📸 建立新快照

📁 歷史備份紀錄

至
時間 保護 快照名稱 描述 類型 操作
請點擊「重新整理」載入歷史紀錄...
================================================================================ FILE: init_db.py ================================================================================ import asyncio import asyncpg import logging import sys import os from dotenv import load_dotenv # 🌟 1. 引入 load_dotenv # 🌟 2. 明確指示 Python 讀取同目錄下的 .env 檔案 load_dotenv() logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 🌟 2. 改用 os.getenv 讀取環境變數，並保留原本的設定作為安全預設值 DB_CONFIG = { "database": os.getenv("DB_NAME", "cmts_nms"), "user": os.getenv("DB_USER", "postgres"), # 本地開發常用的預設帳號，或留空 "" "password": os.getenv("DB_PASS", ""), # 🌟 絕對機密：預設留空！ "host": os.getenv("DB_HOST", "127.0.0.1"), "port": os.getenv("DB_PORT", "5432") } async def init_database(force_reset: bool = False): try: logger.info("🔄 正在連線到 PostgreSQL (asyncpg)...") conn = await asyncpg.connect( database=DB_CONFIG["database"], user=DB_CONFIG["user"], password=DB_CONFIG["password"], host=DB_CONFIG["host"], port=int(DB_CONFIG["port"]) ) # ========================================== # 💣 核彈模式：清除舊有資料表 # ========================================== if force_reset: logger.warning("⚠️ 警告：啟動強制重建模式，正在刪除現有資料表...") await conn.execute("DROP TABLE IF EXISTS cmts_options;") await conn.execute("DROP TABLE IF EXISTS device_status;") await conn.execute("DROP TABLE IF EXISTS system_filters;") # 💡 備份表通常極度重要，即使 reset 也不建議輕易 DROP，除非你確定要連備份一起砍 # await conn.execute("DROP TABLE IF EXISTS config_backups;") logger.info("🗑️ 舊資料表已清除完畢。") # ========================================== # 🏗️ 安全建置模式：建立資料表 (IF NOT EXISTS) # ========================================== # 1. 建立選項快取表 cmts_options (無 config_type) logger.info("🛠️ 正在檢查/建立 cmts_options 資料表...") await conn.execute(""" CREATE TABLE IF NOT EXISTS cmts_options ( host VARCHAR(255) NOT NULL, path VARCHAR(500) NOT NULL, data JSONB NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (host, path) ); """) # 2. 建立設備狀態表 device_status (無 config_type) logger.info("🛠️ 正在檢查/建立 device_status 資料表...") await conn.execute(""" CREATE TABLE IF NOT EXISTS device_status ( host VARCHAR(255) NOT NULL PRIMARY KEY, cmts_version VARCHAR(100) DEFAULT 'unknown', last_scanned VARCHAR(100), updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # 3. 建立過濾器設定表 system_filters logger.info("🛠️ 正在檢查/建立 system_filters 資料表...") await conn.execute(""" CREATE TABLE IF NOT EXISTS system_filters ( config_type VARCHAR(50) PRIMARY KEY, hidden_keys TEXT[] DEFAULT '{}', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # 4. 建立設備配置備份表 config_backups (手動備份 + 釘選防護版) logger.info("🛠️ 正在檢查/建立 config_backups 資料表與索引...") await conn.execute(""" CREATE TABLE IF NOT EXISTS config_backups ( id UUID PRIMARY KEY, host VARCHAR(255) NOT NULL, config_type VARCHAR(50) NOT NULL DEFAULT 'running', timestamp TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, snapshot_name VARCHAR(255), description TEXT DEFAULT '', is_auto BOOLEAN NOT NULL DEFAULT FALSE, is_pinned BOOLEAN NOT NULL DEFAULT FALSE, -- 🌟 新增：釘選防護欄位 raw_cli TEXT, parsed_tree JSONB, CONSTRAINT uq_snapshot_name UNIQUE (host, config_type, snapshot_name) ); """) # 建立基礎查詢索引 await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_config_backups_host_type_ts ON config_backups (host, config_type, timestamp DESC); """) # 🌟 新增：建立滾動淘汰專用索引 await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_config_backups_retention ON config_backups (host, is_pinned, timestamp DESC); """) await conn.close() logger.info("✅ 資料庫初始化完成！所有資料表已準備就緒。") except Exception as e: logger.error(f"❌ 資料庫初始化失敗: {e}") if __name__ == "__main__": # 偵測命令列參數是否包含 --reset is_reset = "--reset" in sys.argv if is_reset: print("\n" + "="*50) print("🚨 你正在執行資料庫重置 (--reset) 🚨") print("這將會清空所有的選項快取與系統設定！") print("="*50 + "\n") confirm = input("確定要繼續嗎？(輸入 yes 繼續): ") if confirm.lower() != "yes": print("已取消操作。") sys.exit(0) asyncio.run(init_database(force_reset=is_reset)) ================================================================================ FILE: cmts_scraper.py ================================================================================ # --- cmts_scraper.py --- import asyncio import asyncssh import re import json import os import time import database from logger import get_logger logger = get_logger("app.scraper") def parse_question_mark_output(output: str) -> dict: """解析 '?' 回傳內容，無差別掃描支援所有標題排列組合""" hint_lines = [] options = [] current_value = None format_desc = None is_format_only = False has_bracket_value = False if "dhcp-relay" in output.lower(): is_format_only = True format_desc = "IP address or options" for line in output.splitlines(): line = line.strip() # ========================================== # 🛡️ 1. 終極雜訊與 Echo 過濾器 # ========================================== if not line or line.startswith('%') or line.startswith('^'): continue # 攔截 Echo 的問號指令 (例如 "logging buffered ?") if line.endswith('?'): continue # 攔截終端機 Prompt (例如 "admin@SERCOMM-COS-02(config)# ..." 或 "admin@SERCOMM-COS-02>") if re.search(r"[a-zA-Z0-9_.@-]+\(.*\)#", line) or re.search(r"^[a-zA-Z0-9_.@-]+>", line): continue # ========================================== # 2. 無差別收集所有有效行作為 Hint (保留最完整的說明給使用者看) hint_lines.append(line) # 3. 略過純標題行，不進行選項解析 if line.startswith('Description:') or line.startswith('Possible completions:'): continue # 4. 解析格式與選項 (無差別掃描每一行) # 格式 A: <格式說明>[當前值] (例如: [Cold Start]) match_format = re.search(r"<([^>]+)>\s*(?:\[([^\]]+)\])?", line) if match_format: is_format_only = True format_desc = match_format.group(1).strip() if match_format.group(2): current_value = match_format.group(2).strip() continue # 格式 B: 型態, 最小值 .. 最大值 (例如: unsignedInt, 1 .. 3000) match_range = re.search(r"^\s*([a-zA-Z0-9_]+),\s*(\d+)\s*\.\.\s*(\d+)\s*$", line) if match_range: is_format_only = True format_desc = f"{match_range.group(1)} ({match_range.group(2)} .. {match_range.group(3)})" continue # 格式 C: 純文字關鍵字 match_keyword = re.search(r"^(IP address|IPv4 address|IPv6 address|MAC address)$", line, re.IGNORECASE) if match_keyword: is_format_only = True format_desc = match_keyword.group(1) continue # 格式 D: 選項列表 [現值] 選項1 選項2 match_current = re.search(r"^\[([^\]]+)\]", line) if match_current: has_bracket_value = True if not current_value: current_value = match_current.group(1).strip() clean_line = re.sub(r"^\[[^\]]+\]", "", line).strip() else: clean_line = line.strip() # 分離選項與說明 (完美支援水平列表，如 severity size-mb) parts = re.split(r'\s{2,}', clean_line) valid_options = [] for p in parts: p = p.strip() if not p: continue # 如果這個片段包含空白，代表它是說明文字 (Description)，停止解析後續片段 if ' ' in p: break valid_options.append(p) for p in valid_options: if p not in ["|", ".."]: options.append(p) # 子命令防呆：如果抓到一堆單字，但沒有 [現值]，且不是已知格式，很可能是子命令列表 if not has_bracket_value and not is_format_only and options: is_format_only = True options = [] if current_value and options and current_value not in options: options.append(current_value) return { "hint": "\n".join(hint_lines).strip(), "options": list(dict.fromkeys(options)) if not is_format_only else [], "current_value": current_value, "format_desc": format_desc, "is_format_only": is_format_only } def parse_device_response(output: str) -> dict: # 完美支援 () (Cold Start): 格式 match = re.search(r"(?:\[(.*?)\]|\(<(.*?)>\))\s*\((.*?)\):", output) if match: enum_content = match.group(1) desc_content = match.group(2) current_value = match.group(3).strip() if enum_content: if enum_content.strip().lower() == "list": return {"type": "list", "options": [], "current_value": current_value} else: return { "type": "enum", "options": [opt.strip() for opt in enum_content.split(",") if opt.strip()], "current_value": current_value } elif desc_content: return { "type": "string_or_number", "options": [], "current_value": current_value, "format_desc": desc_content.strip() } return {"type": "unknown", "options": [], "current_value": None, "raw_output": output.strip()} async def sync_cmts_leaves_async(host, username, password, leaf_paths: list): try: total_paths = len(leaf_paths) processed_count = 0 result_data = {} BATCH_SIZE = 30 batches = [leaf_paths[i:i + BATCH_SIZE] for i in range(0, len(leaf_paths), BATCH_SIZE)] cmts_version = "unknown" for batch_idx, batch in enumerate(batches): logger.info(f"🔄 正在處理第 {batch_idx + 1}/{len(batches)} 批次 (共 {len(batch)} 個路徑)...") try: # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) conn = await asyncio.wait_for( asyncssh.connect(host, username=username, password=password, known_hosts=None), timeout=10.0 ) # 再進入 async with 確保資源會被自動關閉 async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=1.0, prompt_pattern: str = None): output = "" while True: try: chunk = await asyncio.wait_for(process.stdout.read(4096), timeout=timeout) if not chunk: break output += chunk if "--More--" in chunk or "More" in chunk: process.stdin.write(" ") await process.stdin.drain() if prompt_pattern and re.search(prompt_pattern, output): break except asyncio.TimeoutError: break return output # 🌟 關鍵修復 1：等待登入歡迎詞 (MOTD) 結束，確保設備準備好接收指令 await read_until_quiet(timeout=1.5, prompt_pattern=r"(?:#|>)") if cmts_version == "unknown": process.stdin.write("show version | nomore\n") await process.stdin.drain() version_output = await read_until_quiet(timeout=2.0, prompt_pattern=r"(?:#|>)") match = re.search(r"(?:infra|vcmts-cd-0|CableOS)\s+([\w\.\-]+)", version_output, re.IGNORECASE) if match: cmts_version = match.group(1) else: cmts_version = "parse_failed" # 🌟 避免正則失敗導致每批次都重查 # 🌟 關鍵修復 2：確保成功進入 config 模式 process.stdin.write("config\n") await process.stdin.drain() await read_until_quiet(timeout=1.5, prompt_pattern=r"\(config.*\)#") for path in batch: await asyncio.sleep(0.01) command_to_send = f"{path} ?" process.stdin.write(command_to_send) await process.stdin.drain() output_question = await read_until_quiet(timeout=0.8) q_data = parse_question_mark_output(output_question) # 🌟 頂級優雅解法：使用 Ctrl+U (\x15) 瞬間清空整行輸入緩衝區 process.stdin.write("\x15") await process.stdin.drain() await read_until_quiet(timeout=0.2) parsed_data = { "hint": q_data["hint"], "options": q_data["options"], "type": "list" if q_data["options"] else "unknown", "current_value": q_data.get("current_value") } if q_data.get("is_format_only"): parsed_data["type"] = "string_or_number" elif not q_data["options"]: process.stdin.write(f"{path}\n") await process.stdin.drain() output_enter = await read_until_quiet(timeout=0.8) enter_data = parse_device_response(output_enter) parsed_data["type"] = enter_data["type"] parsed_data["options"] = enter_data["options"] parsed_data["current_value"] = enter_data["current_value"] if enter_data["type"] != "unknown": # 如果進入了互動式輸入 (例如 prompt 變成 (val): )，按 Enter 接受預設值並退出 process.stdin.write("\n") await process.stdin.drain() else: # 如果只是印出錯誤，我們不需要做什麼 pass await read_until_quiet(timeout=0.5) result_data[path] = parsed_data processed_count += 1 yield json.dumps({ "event": "progress", "current": processed_count, "total": total_paths, "current_path": path }) + "\n" process.stdin.write("exit\n") await process.stdin.drain() await read_until_quiet(timeout=1.0) except Exception as e: yield json.dumps({"event": "error", "message": f"批次錯誤: {str(e)}"}) + "\n" continue try: current_ts = int(time.time()) formatted_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) # 寫入資料庫 if cmts_version not in ["unknown", "parse_failed"]: await database.upsert_device_status(host, {"cmts_version": cmts_version, "last_scanned": formatted_time}) else: await database.upsert_device_status(host, {"last_scanned": formatted_time}) for p in batch: if p in result_data: result_data[p]["updated_at"] = current_ts await database.upsert_leaf_option(host, p, result_data[p]) logger.info(f"💾 [DB] 第 {batch_idx + 1}/{len(batches)} 批次已寫入資料庫！") except Exception as e: logger.error(f"⚠️ 資料庫寫入失敗: {e}") if batch_idx < len(batches) - 1: await asyncio.sleep(2) yield json.dumps({"event": "done", "message": "快取更新完成"}) + "\n" except Exception as e: yield json.dumps({"event": "error", "message": f"爬蟲嚴重錯誤: {str(e)}"}) + "\n" async def fetch_raw_config(host: str, username: str, password: str, config_type: str = "running") -> str: try: # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) conn = await asyncio.wait_for( asyncssh.connect(host, username=username, password=password, known_hosts=None), timeout=10.0 ) # 再進入 async with 確保資源會被自動關閉 async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=2.0, prompt_pattern: str = None): output = "" while True: try: chunk = await asyncio.wait_for(process.stdout.read(4096), timeout=timeout) if not chunk: break output += chunk if "--More--" in chunk or "More" in chunk: process.stdin.write(" ") await process.stdin.drain() if prompt_pattern and re.search(prompt_pattern, output): break except asyncio.TimeoutError: break return output await read_until_quiet(timeout=1.0, prompt_pattern=r"(?:#|>)") if config_type == "full": process.stdin.write("config\n") await process.stdin.drain() await read_until_quiet(timeout=1.5, prompt_pattern=r"\(config\)#") cmd = "show full-configuration | nomore" process.stdin.write(f"{cmd}\n") await process.stdin.drain() raw_output = await read_until_quiet(timeout=3.0, prompt_pattern=r"\(config\)#") process.stdin.write("exit\n") await process.stdin.drain() else: cmd = "show running-config | nomore" process.stdin.write(f"{cmd}\n") await process.stdin.drain() raw_output = await read_until_quiet(timeout=3.0) process.stdin.write("exit\n") await process.stdin.drain() cleaned_output = re.sub(r'\x1b\[[0-9;?]*[a-zA-Z]|\x08', '', raw_output) cleaned_output = re.sub(r'\[7m\s*--More--\s*\[27m\[\d+D\[K', '', cleaned_output) cleaned_output = re.sub(r'\s*--More--\s*', '', cleaned_output) cleaned_output = re.sub(r'\s*\(END\)\s*', '', cleaned_output) lines = cleaned_output.splitlines() clean_lines = [ line for line in lines if not line.strip().startswith(cmd) and not line.strip().startswith("admin@") ] return "\n".join(clean_lines).strip() except Exception as e: raise Exception(f"SSH 連線或抓取設定失敗: {str(e)}") def parse_config_to_tree(raw_cli: str) -> dict: return {"_raw_length": len(raw_cli), "status": "pending_parser"} ================================================================================ FILE: main.py ================================================================================ # --- main.py --- import os from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from fastapi.responses import HTMLResponse from contextlib import asynccontextmanager from starlette.middleware.base import BaseHTTPMiddleware # 🌟 新增引入 import database import asyncio # 引入我們剛剛拆分出來的路由模組 from routers import query, config, terminal, lock, leaf_options, backup, diagnostics, auth, logs @asynccontextmanager async def lifespan(app: FastAPI): # Startup await database.init_db_pool() # 🌟 新增：初始化 Log Broadcaster 的 Event Loop 參考，確保跨執行緒排程安全 logs.log_broadcaster.loop = asyncio.get_running_loop() yield # Shutdown await database.close_db_pool() app = FastAPI(title="Harmonic CMTS Manager", version="2.0", lifespan=lifespan) # 🌟 新增：全域安全與快取 Middleware class SecurityAndCacheMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): response = await call_next(request) # 1. 補上安全性標頭 (圖片 7 的警告) response.headers["X-Content-Type-Options"] = "nosniff" # 2. 針對靜態檔案補上 Cache-Control (圖片 2 的警告) # 讓 JS/CSS 快取 1 小時 (3600秒)，提升前端載入效能 if request.url.path.startswith("/static/"): response.headers["Cache-Control"] = "public, max-age=3600" else: # API 請求不快取，確保拿到最新資料 response.headers["Cache-Control"] = "no-store" return response app.add_middleware(SecurityAndCacheMiddleware) # 🌟 掛載 Middleware # 掛載靜態檔案目錄 (對應 static/style.css 與 static/app.js) app.mount("/static", StaticFiles(directory="static"), name="static") # 🌟 統一標準：所有 REST API 都在這裡掛載 /api/v1 前綴 app.include_router(query.router, prefix="/api/v1") app.include_router(config.router, prefix="/api/v1") app.include_router(leaf_options.router, prefix="/api/v1") app.include_router(lock.router, prefix="/api/v1") app.include_router(backup.router, prefix="/api/v1") app.include_router(diagnostics.router, prefix="/api/v1") app.include_router(auth.router, prefix="/api/v1") # WebSocket 通常獨立於 API 版本之外，所以不加前綴 app.include_router(terminal.router) app.include_router(logs.router) # 🌟 新增：掛載日誌 WebSocket 路由 # 根目錄路由：回傳前端 UI @app.get("/", response_class=HTMLResponse, tags=["UI"]) async def read_root(): # 讀取同層級的 index.html html_path = os.path.join(os.path.dirname(__file__), "index.html") with open(html_path, "r", encoding="utf-8") as f: return f.read() if __name__ == "__main__": import uvicorn # 啟動伺服器 uvicorn.run("main:app", host="0.0.0.0", port=8001, reload=True) ================================================================================ FILE: test_scraper.py ================================================================================ from cmts_scraper import sync_cmts_leaves # 💡 直接套用您在 shared.py 中驗證過的完美設定 CMTS_DEVICE = { 'device_type': 'cisco_ios', 'host': '10.14.110.4', 'username': 'admin', 'password': 'nsgadmin', 'port': 22, 'fast_cli': False, 'global_delay_factor': 2 } # 測試用的葉節點路徑 TEST_LEAVES = [ "clock timezone", "cable mac-domain 13:0/0.0 admin-state" ] if __name__ == "__main__": print("🚀 開始執行 CMTS 爬蟲測試...") result = sync_cmts_leaves(CMTS_DEVICE, TEST_LEAVES) print("\n📊 最終測試結果:") import json print(json.dumps(result, indent=4, ensure_ascii=False)) ================================================================================ FILE: logger.py ================================================================================ # --- logger.py --- import logging import sys # 定義終端機輸出的 ANSI 顏色 COLORS = { "DEBUG": "\033[36m", # 青色 (Cyan) "INFO": "\033[32m", # 綠色 (Green) "WARNING": "\033[33m", # 黃色 (Yellow) "ERROR": "\033[31m", # 紅色 (Red) "CRITICAL": "\033[1;31m",# 粗體紅色 (Bold Red) "RESET": "\033[0m" # 重置顏色 } class ColoredFormatter(logging.Formatter): def format(self, record): log_color = COLORS.get(record.levelname, COLORS["RESET"]) # 格式：時間 | 等級 | 模組名稱 | 訊息 format_str = f"{log_color}%(asctime)s | %(levelname)-7s | %(name)-15s | %(message)s{COLORS['RESET']}" formatter = logging.Formatter(format_str, datefmt="%Y-%m-%d %H:%M:%S") return formatter.format(record) # 定義系統中受管控的模組清單 MANAGED_LOGGERS = [ "app.database", # 資料庫操作 "app.scraper", # 爬蟲與快取 "app.diagnostics", # CM 診斷 "app.backup", # 備份與還原 "app.ssh", # 🔌 SSH 連線引擎 (注意這裡要有逗號！) "app.auth" # 🔐 上帝模式與安全授權 ] def setup_logger(name: str, level=logging.INFO): logger = logging.getLogger(name) logger.setLevel(level) # 避免重複添加 Handler 導致日誌印兩次 if not logger.handlers: handler = logging.StreamHandler(sys.stdout) handler.setFormatter(ColoredFormatter()) logger.addHandler(handler) logger.propagate = False # 防止日誌往上傳遞給 root logger (避免被 FastAPI 預設格式覆蓋) return logger # 系統啟動時，初始化所有模組，預設等級為 INFO for name in MANAGED_LOGGERS: setup_logger(name, logging.INFO) def get_logger(name: str): """供各個檔案引入 logger 使用""" if name not in MANAGED_LOGGERS: return setup_logger(name) return logging.getLogger(name) def get_all_log_levels(): """取得目前所有模組的日誌等級 (供前端 UI 顯示)""" return {name: logging.getLevelName(logging.getLogger(name).level) for name in MANAGED_LOGGERS} def set_log_level(name: str, level_str: str): """動態設定特定模組的日誌等級""" if name in MANAGED_LOGGERS: level = getattr(logging, level_str.upper(), logging.INFO) logging.getLogger(name).setLevel(level) return True return False def register_websocket_handler(handler: logging.Handler): """🌟 新增：註冊自訂 Handler 到所有受管控的 Logger，並防止重複添加""" for name in MANAGED_LOGGERS: logger = logging.getLogger(name) if handler not in logger.handlers: logger.addHandler(handler) ================================================================================ FILE: PROJECT_STATE.md ================================================================================ # 📖 Harmonic CMTS Manager - Project State (Living Document) > ⚠️ **AI 助手請注意**：本專案的核心架構與開發規範已移至 `.clinerules`。此檔案僅作為「專案進度存檔」與「待辦事項追蹤」使用。 --- ## ✅ 已完成開發階段 (Completed Phases) ### Phase 1: PostgreSQL 高可用性架構升級 - [x] 成功導入 `asyncpg`，建立 `database.py` 管理非同步資料庫連線池。 - [x] 建立 `cmts_options`, `device_status`, `system_filters` 資料表，嚴格遵守 `config_type` 雙軌隔離的主鍵設計。 - [x] 實踐完整的「**Zero-Downtime Fallback 機制**」：資料庫連線異常時，自動退回使用 JSON 檔案讀寫。 ### Phase 2: 設備配置備份與快照機制 - [x] **資料庫擴充**：在 PostgreSQL 中建立 `config_backups` 資料表 (包含 `id`, `host`, `timestamp`, `raw_cli`, `parsed_tree`, `snapshot_name`, `description`)。 - [x] **前端 UI 實作**：完成「設備備份與還原」頁籤，包含建立快照表單與歷史紀錄列表。 - [x] **前端 UI 優化與防禦性編程**：實作具備 Null-Safety 與 `.trim()` 容錯的多維度前端搜尋過濾器 (支援快照名稱 + 描述雙欄位比對)。 ### Phase 3: 智慧差異還原與歷史預覽 - [x] **前端安全還原防呆**：實作三階段安全還原流程 UI (包含 Diff 預覽與確認寫入按鈕)，並加入跨設備還原阻斷機制。 - [x] **後端 Diff 引擎實作**：完成 `/api/v1/backups/{id}/diff`，成功生成絕對路徑指令陣列。 - [x] **後端 SSH 交易寫入管道 (Transactional SSH Pipeline)**：實作 `/api/v1/backups/{id}/restore` API，採用 `StreamingResponse` (NDJSON 串流回應)，並具備 Fail-safe `abort` 撤銷機制。 ### Phase 3.5: 企業級前端效能重構 (Extreme Performance Optimization) - [x] **記憶體遞迴渲染 (In-Memory Recursive Rendering)**：徹底重構 `tree-ui.js`，將數千次 DOM 寫入壓縮為單次 `innerHTML` 寫入，解決「展開全部」導致瀏覽器卡死的問題。 - [x] **非同步 UI 保護機制**：在所有大型渲染場景 (初始載入、單點展開、全部展開) 導入 `setTimeout` 讓出主執行緒，並搭配沙漏游標與橘色讀取提示，確保 UI 絕對滑順。 - [x] **CSS 渲染隔離**：導入 `content-visibility: auto`，讓不在可視範圍內的 DOM 節點暫停渲染計算。 - [x] **精準 DOM 查詢**：將鎖定狀態輪詢 (`startLockStatusPolling`) 的搜尋範圍限縮於當前啟用的視圖內，消除全域搜尋造成的卡頓 (Jank)。 --- ## 🚀 即將到來的里程碑 (Upcoming Milestones Summary) ### Phase 4: 自動化防護、進階管理與指令精準度 (Automation & Advanced Management) - [x] **智能視覺診斷 (Visual Diagnostics)**： - CM 一鍵診斷中心：整合基礎狀態、PHY 射頻指標與 OFDM MER 頻譜。引入 Chart.js 將 ASCII 報表轉化為紅黃綠狀態圖與長條圖。 - [x] **備份保留策略 (Retention Policy)**： - **手動快照配額 (Manual Quota)**：限制每台設備最多保留 20 份手動快照，達上限時，採用 FIFO (先進先出) 機制自動清理舊資料，防止資料庫無限膨脹。 - [ ] **系統日誌動態儀表板 (Log Viewer UI)** : - 基於 FastAPI 實作一個 WebSocket Log Streamer。 1. **Custom Log Handler**: 在現有的 Python `logging` 模組中，撰寫一個自訂的 Handler，能夠攔截系統的 Log 訊息（包含 ANSI 色碼）。 2. **WebSocket Endpoint**: 建立一個 FastAPI WebSocket 路由 `/ws/logs`。 3. **Broadcaster**: 實作一個簡單的機制，當 Custom Log Handler 收到新日誌時，能非同步地將訊息推播給所有連線中的 WebSocket 客戶端。 ### Phase 5: RPD 快速擴容與部署精靈 (Rapid RPD Provisioning Wizard) - [ ] **RPD 樣板克隆引擎 (Template Cloning)**： - 於「MAC Domain 狀態感知配置精靈」中新增 RPD 擴容模式。允許使用者選擇現有設備上已配置完成的 RPD (支援 FDX, FDD, D3.1+) 作為基準樣板。 - [ ] **Tree 節點複製與參數替換 (Node Duplication & Modification)**： - 透過底層 Tree 架構，完整複製複雜的 RF 與通道設定，並提供 UI 介面供使用者修改唯一識別碼 (如 MAC Address、RPD Name)。 - [ ] **安全寫入與校驗 (Safe Provisioning)**： - 結合 Phase 4 的防護機制，在將全新 RPD 配置寫入 CMTS 前，進行參數衝突檢查（避免 MAC 或 IP 重複），實現零錯誤的設備擴容。 - [ ] **Diff 引擎指令精準度強化 (Diff Logic Hardening)**： - 重新 Review `generate_diff_commands` 演算法。 - 實作「指令截斷機制」，確保生成 `no` 移除指令時，能精準剝離多餘的 Value 或參數，避免 CMTS 拒絕執行或引發非預期刪除。 --- ## 🐛 已知問題與技術債 (Known Issues & Tech Debt) - 目前系統運行極度穩定，前端效能瓶頸已徹底消除，各項併發鎖定與 UI 狀態連動皆已完善。準備進入 Phase 4 的 Diff 引擎強化開發。 ================================================================================ FILE: static/api.js ================================================================================ // --- static/api.js --- // ========================================== // 1. 鎖定管理 (Lock Management) // ========================================== export async function apiAcquireLock(path, userId, username, host) { const res = await fetch('/api/v1/locks/acquire', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ path, user_id: userId, username, host }) }); const data = await res.json(); return { status: res.status, data }; } export async function apiReleaseLock(path, userId, host) { return fetch('/api/v1/locks/release', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ path, user_id: userId, host }) }); } export async function apiSendHeartbeat(path, userId, host) { return fetch('/api/v1/locks/heartbeat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ path, user_id: userId, host }) }); } export async function apiGetLockStatus(host) { // 帶入 host 參數，實現 IP 隔離查詢 const url = host ? `/api/v1/locks/status?host=${encodeURIComponent(host)}` : '/api/v1/locks/status'; const response = await fetch(url); return response.json(); } // ========================================== // 2. 樹狀圖與選項快取 (Tree & Options) // ========================================== export async function apiGetFullConfig(host, username, password, skipFilter = false, configType = 'running') { // 🌟 將 config_type 加入 URL 參數中 let url = `/api/v1/cmts-full-config?host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}&config_type=${configType}`; if (skipFilter) url += '&skip_filter=true'; const response = await fetch(url); return response.json(); } export async function apiGetLeafOptions(host) { const response = await fetch(`/api/v1/cmts-leaf-options?host=${encodeURIComponent(host)}`); return response.json(); } export async function apiSyncLeafOptions(host, paths) { return fetch('/api/v1/cmts-leaf-options/sync', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ host, leaf_paths: paths }) }); } export async function apiClearCache(host, paths) { const response = await fetch(`/api/v1/clear_cache?host=${encodeURIComponent(host)}`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(paths) }); if (!response.ok) throw new Error(`伺服器回應錯誤: ${response.status}`); return response.json(); } export async function apiGetCmtsVersion(host, username, password) { const url = `/api/v1/cmts-version?host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}`; const response = await fetch(url); return response.json(); } export async function apiGetScanStatus(host) { try { const response = await fetch(`/api/v1/scan-status?host=${encodeURIComponent(host)}`); const data = await response.json(); return data.is_scanning; } catch (e) { return false; } } // ========================================== // 3. 指令生成與執行 (CLI Generation & Execution) // ========================================== export async function apiGenerateCli(diffs, interfaces) { const response = await fetch('/api/v1/cmts-generate-cli', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ diffs, interfaces_with_admin_state: interfaces }) }); return response.json(); } export async function apiExecuteConfig(script, host, username, password) { const response = await fetch('/api/v1/cmts-config', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ script, host, username, password }) }); return response.json(); } // ========================================== // 4. 系統設定與過濾器 (System Settings) // ========================================== // 🌟 加上 configType 參數 export async function apiGetTreeFilters(configType = 'running') { const response = await fetch(`/api/v1/settings/tree-filters?config_type=${configType}`); return response.json(); } // 🌟 加上 configType 參數 export async function apiSaveTreeFilters(hiddenKeys, configType = 'running') { const response = await fetch(`/api/v1/settings/tree-filters?config_type=${configType}`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ hidden_keys: hiddenKeys, config_type: configType }) }); return response.json(); } // ========================================== // 5. MAC Domain 與綜合查詢 (MAC Domain & Query) // ========================================== export async function apiGetMacDomainList(host, username, password) { const url = `/api/v1/cmts-mac-domain-list?host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}`; const response = await fetch(url); return response.json(); } export async function apiGetMacDomainConfig(target, host, username, password) { const url = `/api/v1/cmts-mac-domain-config?target=${encodeURIComponent(target)}&host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}`; const response = await fetch(url); return response.json(); } export async function apiExecuteQuery(queryType, target, host, username, password) { const url = `/api/v1/cmts-query?query_type=${queryType}&target=${encodeURIComponent(target)}&host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}`; const response = await fetch(url); return response.json(); } // 🌟 新增：動態抓取設備清單 API export async function apiListCms(host, username, password) { const url = `/api/v1/list-cms?host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}`; const response = await fetch(url); return response.json(); } export async function apiListRpds(host, username, password) { const url = `/api/v1/list-rpds?host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}`; const response = await fetch(url); return response.json(); } // ========================================== // 6. 專門處理串流的 API 呼叫函數 (UI 可以即時更新) // ========================================== export async function apiSyncLeafOptionsStream(host, paths, onProgress) { const response = await fetch('/api/v1/cmts-leaf-options/sync', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ host, leaf_paths: paths }) }); if (!response.body) throw new Error("瀏覽器不支援 ReadableStream"); const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8"); while (true) { const { done, value } = await reader.read(); if (done) break; const chunk = decoder.decode(value, { stream: true }); const lines = chunk.split("\n").filter(line => line.trim() !== ""); for (const line of lines) { try { const data = JSON.parse(line); onProgress(data); } catch (e) {} } } } export async function apiGetCmDiagnostics(host, username, password, mac) { const url = `/api/v1/cm-diagnostics/?host=${encodeURIComponent(host)}&username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}&mac=${encodeURIComponent(mac)}`; const response = await fetch(url); if (!response.ok) { const err = await response.json(); throw new Error(err.detail || "伺服器發生錯誤"); } return response.json(); } export async function apiGetLogLevels() { const response = await fetch('/api/v1/settings/logs'); return response.json(); } export async function apiSetLogLevel(module, level) { const response = await fetch('/api/v1/settings/logs', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ module, level }) }); return response.json(); } // ========================================== // 7. 授權與驗證 API (Auth) // ========================================== export async function apiVerifyGodMode(password) { const response = await fetch('/api/v1/auth/god-mode', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ password }) }); if (!response.ok) { const err = await response.json(); throw new Error(err.detail || "驗證失敗"); } return response.json(); } // ========================================== // 📌 釘選防護 API (Pinning) // ========================================== export async function apiToggleBackupPin(backupId) { const response = await fetch(`/api/v1/backups/${backupId}/toggle-pin`, { method: 'POST', headers: { 'Content-Type': 'application/json' } }); return response.json(); } ================================================================================ FILE: static/app.js ================================================================================ /* --- static/app.js --- */ // ============================================================================ // 📦 模組匯入區 (Imports) // ============================================================================ // 1. 終端機模組 (Terminal) import { initTerminal, connectWebSocket, disconnectWebSocket, injectCommand, downloadTerminal, isConnected } from './terminal.js'; // 2. API 網路請求模組 (API) import { apiGetFullConfig, apiClearCache, apiExecuteQuery, apiGetTreeFilters, apiSaveTreeFilters, apiSyncLeafOptionsStream, apiGetCmtsVersion, apiGetScanStatus, apiGetLockStatus, apiGetCmDiagnostics, apiListCms, apiListRpds, apiGetLogLevels, apiSetLogLevel, apiVerifyGodMode, apiGetLeafOptions, // 🌟 補上這個 apiSyncLeafOptions, // 🌟 補上這個 apiToggleBackupPin // 🌟 新增這行 } from './api.js'; // 3. 共用工具模組 (Utils) import { getGlobalConnectionInfo } from './utils.js'; // 4. 樹狀圖 UI 模組 (Tree UI) import { buildTree, buildRealFilterTree, expandAll, collapseAll, clearTreeCache } from './tree-ui.js'; // 5. 編輯模式與鎖定模組 (Edit Mode) import { releaseAllLocks, startEditFolder, startEditLeaf, cancelEditFolder, cancelEditLeaf, previewFolderCLI, previewLeafCLI, hideSideCLI, executeSideCLI, previewNodeCLI, SESSION_USER_ID, currentEditElementId } from './edit-mode.js'; // 6. MAC Domain 配置模組 (MAC Domain) import { hasMacDomainData, resetMacDomainData, loadMacDomainList, fetchMacDomainConfig, populateFormFromData, revertFormChanges, toggleGroupVisibility, loadDsGroupData, loadUsGroupData, generateMacDomainCLI, executeBondingConfig } from './mac-domain.js'; // ============================================================================ // 🌟 全域記憶體 (Dual-Track Memory) // ============================================================================ window.treeDataStore = { running: null, full: null }; // ============================================================================ // 🌟 1. 全域生命週期、閒置偵測與 SSE 廣播監聽 // ============================================================================ let idleTimer; const IDLE_TIMEOUT = 300000; // 5 分鐘 (毫秒) let globalSSE = null; let logTerm = null; let logWs = null; let isLogEnabled = false; // 預設關閉，不主動消耗資源 // 初始化全域 SSE 監聽器 (設備級別) function initGlobalSSE() { // 🌟 新增防線：如果尚未連線，強制關閉舊的 SSE 並退出 if (!isConnected) { if (globalSSE) { globalSSE.close(); globalSSE = null; } return; } const connInfo = getGlobalConnectionInfo(); if (!connInfo || !connInfo.host) return; if (globalSSE) globalSSE.close(); // 🌟 拔除 config_type，純粹監聽該設備的廣播 globalSSE = new EventSource(`/api/v1/cmts-leaf-options/stream?host=${encodeURIComponent(connInfo.host)}`); globalSSE.onmessage = (event) => { const data = JSON.parse(event.data); const statusEl = document.getElementById('scan-status'); if (data.event === 'scan_start' || data.event === 'progress') { setAdminButtonsState(true, "⏳ 系統更新中..."); } if (data.event === 'scan_start') { if (statusEl) { statusEl.innerHTML = `⏳ 系統正在背景更新快取，請稍候...`; statusEl.style.color = "#f39c12"; } } else if (data.event === 'progress') { if (statusEl) { statusEl.innerHTML = `⏳ 背景掃描進度：${data.current} / ${data.total} (正在處理: ${data.current_path})`; statusEl.style.color = "#f39c12"; } } else if (data.event === 'done') { if (statusEl) { statusEl.innerHTML = `✅ 掃描完美達成！快取已全面更新。`; statusEl.style.color = "#27ae60"; setTimeout(() => statusEl.textContent = "", 5000); } setAdminButtonsState(false); localStorage.removeItem('scanLockUntil'); } else if (data.event === 'error') { if (statusEl) { statusEl.innerHTML = `❌ 錯誤: ${data.message}`; statusEl.style.color = "#e74c3c"; } setAdminButtonsState(false); localStorage.removeItem('scanLockUntil'); } }; } // 🌟 關鍵修復：當使用者按 F5 重整或關閉網頁時，主動切斷 SSE 連線 window.addEventListener('beforeunload', () => { if (globalSSE) { globalSSE.close(); globalSSE = null; } }); // 🌟 特務的記憶體：記錄「上一次輪詢時，處於鎖定狀態的路徑」(加入 Host 隔離) let activeLockState = new Set(); let lockPollingTimer = null; let currentPollingHost = null; // 記錄當前輪詢的設備 IP window.globalActiveLocks = {}; // 🌟 升級為二維結構：{ "10.14.110.4": { "alias": {...} } } window.recentlyReleasedLocks = {}; // 🌟 新增：防閃爍冷卻表 (Optimistic UI Cooldown) function startLockStatusPolling() { if (lockPollingTimer) clearInterval(lockPollingTimer); lockPollingTimer = setInterval(async () => { if (!isConnected) return; // 🌟 新增防線：未連線時不進行輪詢 const connInfo = getGlobalConnectionInfo(); if (!connInfo || !connInfo.host) return; const host = connInfo.host; // 🛡️ 跨設備防呆：如果使用者切換了 IP，必須清空舊設備的鎖定記憶，防止幽靈解鎖 if (currentPollingHost !== host) { activeLockState.clear(); currentPollingHost = host; } try { const result = await apiGetLockStatus(host); if (result.status === 'success') { const currentLocks = result.data; window.globalActiveLocks[host] = currentLocks; const newLockState = new Set(); const now = Date.now(); // 🎯 任務 1：處理「新增鎖定」或「持續鎖定」的節點 for (const [path, lockInfo] of Object.entries(currentLocks)) { const lockKey = `${host}@@${path}`; // 🛡️ 防閃爍機制 if (window.recentlyReleasedLocks[lockKey]) { if (now - window.recentlyReleasedLocks[lockKey] < 8000) { continue; } else { delete window.recentlyReleasedLocks[lockKey]; } } newLockState.add(lockKey); const safePath = path.replace(/\\/g, '\\\\').replace(/"/g, '\\"'); // 🌟 效能優化：只在當前啟用的視圖中搜尋，避免掃描隱藏的龐大 DOM const currentMode = document.getElementById('configTask')?.value === 'form-full-config' ? 'full' : 'running'; const activeContainer = document.getElementById(`tree-container-${currentMode}`); if (activeContainer) { const targetBtns = activeContainer.querySelectorAll(`.edit-btn[data-path="${safePath}"], .edit-btn[data-path^="${safePath}::"]`); targetBtns.forEach(btn => { if (btn.id === `edit-btn-${currentEditElementId}`) return; if (btn.innerText !== "🔒") { btn.style.pointerEvents = 'none'; btn.style.opacity = '0.2'; const btnPath = btn.getAttribute('data-path'); if (btnPath === path) { btn.title = lockInfo.user_id !== SESSION_USER_ID ? `🔒 此區塊正由 [${lockInfo.username}] 編輯中` : `🔒 您正在另一個視圖編輯此項目`; } else { btn.title = `🔒 父層級已被鎖定，無法編輯此項目`; } btn.innerText = "🔒"; } }); } } // 🔓 任務 2：處理「解除鎖定」的節點 activeLockState.forEach(oldKey => { const [oldHost, oldPath] = oldKey.split('@@'); if (oldHost === host && !newLockState.has(oldKey)) { const safePath = oldPath.replace(/\\/g, '\\\\').replace(/"/g, '\\"'); // 🌟 效能優化：同樣只在當前啟用的視圖中搜尋 const currentMode = document.getElementById('configTask')?.value === 'form-full-config' ? 'full' : 'running'; const activeContainer = document.getElementById(`tree-container-${currentMode}`); if (activeContainer) { const targetBtns = activeContainer.querySelectorAll(`.edit-btn[data-path="${safePath}"], .edit-btn[data-path^="${safePath}::"]`); targetBtns.forEach(btn => { if (btn.innerText === "🔒") { const btnPath = btn.getAttribute('data-path'); let stillLockedByOther = false; if (currentLocks[btnPath]) { stillLockedByOther = true; } else { for (const lockedPath of Object.keys(currentLocks)) { if (btnPath.startsWith(lockedPath + "::")) { stillLockedByOther = true; break; } } } if (!stillLockedByOther) { btn.style.pointerEvents = 'auto'; btn.style.opacity = '0.3'; btn.title = "鎖定並編輯此項目"; btn.innerText = "✏️"; } } }); } } }); // 💾 任務 3：更新特務的記憶 activeLockState = newLockState; } } catch (error) { console.warn("獲取鎖定狀態失敗:", error); } }, 15000); } // 頁面載入與縮放初始化 window.onload = () => { initTerminal(); resetIdleTimer(); // 頁面載入時啟動閒置計時器 // initGlobalSSE(); <--- 🌟 刪除這行！不要在網頁剛開時就偷聽 startLockStatusPolling(); // 啟動鎖定狀態輪詢 // 🌟 新增：網頁載入時，自動觸發一次「選擇配置任務」的切換，確保畫面與選單同步 const configTaskSelect = document.getElementById('configTask'); if (configTaskSelect) { configTaskSelect.dispatchEvent(new Event('change')); } }; window.onresize = () => { if (typeof fitAddon !== 'undefined' && fitAddon) fitAddon.fit(); }; // 閒置計時器重置邏輯 (超時將自動釋放所有編輯鎖定) function resetIdleTimer() { clearTimeout(idleTimer); idleTimer = setTimeout(releaseAllLocks, IDLE_TIMEOUT); } // 🌟 效能急救：節流閥 (Throttle) 機制 let throttleTimer = false; function throttledReset() { if (!throttleTimer) { resetIdleTimer(); throttleTimer = true; setTimeout(() => { throttleTimer = false; }, 2000); // 每 2 秒最多只觸發一次重設 } } window.addEventListener('mousemove', throttledReset); window.addEventListener('keydown', throttledReset); window.addEventListener('click', throttledReset); // ============================================================================ // 🌟 2. 頁籤切換與 UI 狀態控制 (Tabs & UI State) // ============================================================================ // 切換主頁籤 (終端機 / 設備查詢 / 設備配置 / 系統設定) function switchTab(tabId, element) { document.querySelectorAll('.tab-content').forEach(tab => tab.classList.remove('active')); document.querySelectorAll('.tab-btn').forEach(btn => btn.classList.remove('active')); const targetTab = document.getElementById(tabId); if (targetTab) targetTab.classList.add('active'); if (element) element.classList.add('active'); if(tabId === 'cli-tab' && typeof fitAddon !== 'undefined' && fitAddon) setTimeout(() => fitAddon.fit(), 50); } // 切換「設備查詢」內的子任務表單 // ============================================================================ // 🌟 設備查詢動態清單快取 (新增) // ============================================================================ let cachedDeviceLists = { cms: null, rpds: null, host: null }; // 切換「設備查詢」內的子任務表單 function switchQueryTask() { document.querySelectorAll('#query-tab .task-form').forEach(form => form.classList.remove('active')); const selectedTaskId = document.getElementById('queryTask').value; const targetForm = document.getElementById(selectedTaskId); if (targetForm) targetForm.classList.add('active'); // 🌟 觸發背景動態載入 Datalist loadDynamicDatalist(selectedTaskId); } // 🌟 新增：背景載入 Datalist 邏輯 async function loadDynamicDatalist(taskType) { // 若尚未連線，則不觸發背景抓取 (不跳警告，默默 return) if (!isConnected) return; const host = document.getElementById('cmtsHost').value.trim(); const user = document.getElementById('cmtsUser').value.trim(); const pass = document.getElementById('cmtsPass').value.trim(); if (!host || !user || !pass) return; // 若切換了設備 IP，清空舊快取 if (cachedDeviceLists.host !== host) { cachedDeviceLists = { cms: null, rpds: null, host: host }; } if (taskType === 'form-cm-query' || taskType === 'form-cm-diagnostics') { const inputId = taskType === 'form-cm-query' ? 'queryTargetCm' : 'diagMacInput'; const listId = taskType === 'form-cm-query' ? 'cm-mac-list' : 'diag-cm-mac-list'; const defaultPlaceholder = taskType === 'form-cm-query' ? "例: 6467.7240.4076 (留白則查詢全體)" : "例如: 6467.7240.4076"; await fetchAndBindList('cms', inputId, listId, defaultPlaceholder, host, user, pass); } else if (taskType === 'form-rpd-query') { await fetchAndBindList('rpds', 'queryTargetRpd', 'rpd-vcvs-list', "例: 13:0 (留白則查詢全體)", host, user, pass); } } // 🌟 新增：執行 API 呼叫與 UX 狀態切換 async function fetchAndBindList(type, inputId, listId, defaultPlaceholder, host, user, pass) { const inputEl = document.getElementById(inputId); const datalistEl = document.getElementById(listId); if (!inputEl || !datalistEl) return; // 如果已經有快取，直接綁定並結束 if (cachedDeviceLists[type]) { populateDatalist(datalistEl, cachedDeviceLists[type]); return; } // 狀態 1：發出請求前 (鎖定輸入框，提示載入中) inputEl.disabled = true; inputEl.placeholder = "🔄 載入設備清單中..."; inputEl.style.backgroundColor = "#fdfefe"; try { let data; if (type === 'cms') { data = await apiListCms(host, user, pass); cachedDeviceLists.cms = data.cms || []; } else { data = await apiListRpds(host, user, pass); cachedDeviceLists.rpds = data.rpds || []; } // 狀態 2：請求成功 (塞入選項，恢復輸入框) populateDatalist(datalistEl, cachedDeviceLists[type]); inputEl.placeholder = defaultPlaceholder; } catch (error) { console.error(`[Background Task] Failed to load ${type}:`, error); // 狀態 3：請求失敗 (提示失敗，但仍解鎖讓使用者手動輸入) inputEl.placeholder = "⚠️ 清單載入失敗，請手動輸入"; } finally { inputEl.disabled = false; inputEl.style.backgroundColor = ""; // 恢復預設背景色 } } // 🌟 新增：將陣列轉換為