================================================================================ PROJECT SOURCE CODE EXPORT ================================================================================ ================================================================================ FILE: database.py ================================================================================ import asyncpg import json import logging import uuid from typing import Dict, List, Optional, Any # ========================================== # 💡 PostgreSQL 連線與操作 (高可用性版) # ========================================== logger = logging.getLogger(__name__) # DB 設定 (從您的 init_db.py 中提取) DB_CONFIG = { "database": "cmts_nms", "user": "swpa", "password": "swpa4920", "host": "127.0.0.1", "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, config_type: str, path: str, data: dict) -> bool: """將選項寫入或更新至資料庫""" pool = await get_pool() if not pool: return False query = """ INSERT INTO cmts_options (host, config_type, path, data) VALUES ($1, $2, $3, $4) ON CONFLICT (host, config_type, path) DO UPDATE SET data = EXCLUDED.data, updated_at = CURRENT_TIMESTAMP; """ try: async with pool.acquire() as conn: await conn.execute(query, host, config_type, path, json.dumps(data)) return True except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (upsert_leaf_option): {e}") return False except Exception as e: logger.error(f"❌ Unknown Error (upsert_leaf_option): {e}") return False async def get_all_leaf_options(host: str, config_type: str) -> Optional[Dict[str, Any]]: """取得特定 host 與 config_type 的所有選項""" pool = await get_pool() if not pool: return None query = """ SELECT path, data FROM cmts_options WHERE host = $1 AND config_type = $2; """ try: async with pool.acquire() as conn: records = await conn.fetch(query, host, config_type) result = {} for record in records: # asyncpg returns strings for JSON if not explicitly configured with type mapping, # but usually it's fine to just json.loads it. data_val = record['data'] if isinstance(data_val, str): result[record['path']] = json.loads(data_val) else: result[record['path']] = data_val return result except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (get_all_leaf_options): {e}") return None except Exception as e: logger.error(f"❌ Unknown Error (get_all_leaf_options): {e}") return None async def delete_leaf_options(host: str, config_type: 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 config_type = $2 AND path = ANY($3); """ try: async with pool.acquire() as conn: # Execute returns the command tag, e.g., 'DELETE 5' status = await conn.execute(query, host, config_type, paths) deleted_count = int(status.split()[-1]) return deleted_count except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (delete_leaf_options): {e}") return -1 except Exception as e: logger.error(f"❌ Unknown Error (delete_leaf_options): {e}") return -1 # ------------------------------------------ # CRUD Functions for device_status # ------------------------------------------ async def upsert_device_status(host: str, config_type: str, metadata: dict) -> bool: """更新設備的 metadata""" pool = await get_pool() if not pool: return False # Extract known fields cmts_version = metadata.get("cmts_version", "unknown") last_scanned = metadata.get("last_scanned", None) query = """ INSERT INTO device_status (host, config_type, cmts_version, last_scanned) VALUES ($1, $2, $3, $4) ON CONFLICT (host, config_type) DO UPDATE SET cmts_version = EXCLUDED.cmts_version, last_scanned = EXCLUDED.last_scanned, updated_at = CURRENT_TIMESTAMP; """ try: async with pool.acquire() as conn: await conn.execute(query, host, config_type, cmts_version, last_scanned) return True except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (upsert_device_status): {e}") return False except Exception as e: logger.error(f"❌ Unknown Error (upsert_device_status): {e}") return False async def get_device_status(host: str, config_type: str) -> Optional[Dict[str, Any]]: """獲取設備 metadata""" pool = await get_pool() if not pool: return None query = """ SELECT cmts_version, last_scanned FROM device_status WHERE host = $1 AND config_type = $2; """ try: async with pool.acquire() as conn: record = await conn.fetchrow(query, host, config_type) if record: return { "cmts_version": record["cmts_version"], "last_scanned": record["last_scanned"] } return None except asyncpg.PostgresError as e: logger.error(f"❌ DB Error (get_device_status): {e}") return None except Exception as e: logger.error(f"❌ Unknown 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' 撈取全部)""" pool = await get_pool() if not pool: return None try: async with pool.acquire() as conn: if config_type == "all": # 🟢 SELECT 加入 description query = """ SELECT id, host, config_type, timestamp, snapshot_name, description, is_auto FROM config_backups WHERE host = $1 ORDER BY timestamp DESC; """ records = await conn.fetch(query, host) else: # 🟢 SELECT 加入 description query = """ SELECT id, host, config_type, timestamp, snapshot_name, description, is_auto 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"] } 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 ================================================================================ FILE: shared.py ================================================================================ # --- shared.py --- import asyncio import json import os import re from collections import defaultdict DEBUG_MODE = False USE_DB = True # PostgreSQL Feature Toggle def debug_print(msg: str): if DEBUG_MODE: print(msg) # 預設的 CMTS 連線樣板 CMTS_DEVICE = { 'device_type': 'cisco_ios', 'host': '10.14.110.4', 'username': 'admin', 'password': 'nsgadmin', '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 = defaultdict(asyncio.Lock) ================================================================================ FILE: index.html ================================================================================ Harmonic CMTS Manager

🎙️ Harmonic CMTS Admin

狀態：未連線

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

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

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

請先讀取設備狀態...

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

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

📸 建立新快照

📁 歷史備份紀錄

至
時間 快照名稱 描述 類型 操作
請點擊「重新整理」載入歷史紀錄...
================================================================================ FILE: init_db.py ================================================================================ import asyncio import asyncpg import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # DB 設定 DB_CONFIG = { "database": "cmts_nms", "user": "swpa", "password": "swpa4920", "host": "127.0.0.1", "port": "5432" } async def init_database(): 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"]) ) # 1. 建立選項快取表 cmts_options logger.info("🛠️ 正在建立 cmts_options 資料表...") await conn.execute(""" CREATE TABLE IF NOT EXISTS cmts_options ( host VARCHAR(255) NOT NULL, config_type VARCHAR(50) NOT NULL, path VARCHAR(500) NOT NULL, data JSONB NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (host, config_type, path) ); """) # 2. 建立設備狀態表 device_status logger.info("🛠️ 正在建立 device_status 資料表...") await conn.execute(""" CREATE TABLE IF NOT EXISTS device_status ( host VARCHAR(255) NOT NULL, config_type VARCHAR(50) NOT NULL, cmts_version VARCHAR(100) DEFAULT 'unknown', last_scanned VARCHAR(100), updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (host, config_type) ); """) # 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 (Phase 1 新增) 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), is_auto 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.close() logger.info("✅ 資料庫初始化完成！所有資料表已準備就緒。") except Exception as e: logger.error(f"❌ 資料庫初始化失敗: {e}") if __name__ == "__main__": asyncio.run(init_database()) ================================================================================ FILE: cmts_scraper.py ================================================================================ # --- cmts_scraper.py --- import asyncio import asyncssh import re import json import os import time import database from shared import USE_DB def parse_question_mark_output(output: str) -> dict: """解析 '?' 回傳內容，支援無 Description、子命令判定與 dhcp-relay 混合欄位""" hint_lines = [] options = [] current_value = None format_desc = None is_format_only = False state = "INIT" has_description = False has_bracket_value = False # 🌟 Case-5 特殊處理：dhcp-relay 混合型欄位 (選項 + IP輸入) # 直接在開頭掃描完整輸出，若包含 dhcp-relay，強制轉為純文字輸入框 if "dhcp-relay" in output.lower(): is_format_only = True format_desc = "IP address or options" for line in output.splitlines(): line = line.strip() # 過濾雜訊與終端機提示字元 if not line or line.startswith('admin@') or line.startswith('cable') or line.startswith('%') or line.startswith('^'): continue if line.startswith('Description:'): state = "DESC" has_description = True hint_lines.append(line) continue if line.startswith('Possible completions:'): state = "COMPLETIONS" # 🌟 解除限制：無論有沒有 Description，都把這行加進 hint hint_lines.append(line) continue if state == "DESC": hint_lines.append(line) elif state == "COMPLETIONS": # Case 1~4: 無 Description 時，將後續選項說明也納入提示 # 🌟 解除限制：無論有沒有 Description，都把這行加進 hint hint_lines.append(line) # 規則 1: <格式說明>[當前值] 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 # 規則 1.5: 型態, 最小值 .. 最大值 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 # 規則 1.6: IP address 等純文字關鍵字 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 # 規則 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() # 🌟 Case-6 防呆：過濾垂直列表的說明文字 # 利用「3 個以上的連續空白」作為選項與說明文字的分界線 if re.search(r"\s{3,}", clean_line): clean_line = re.split(r"\s{3,}", clean_line)[0] parts = clean_line.split() for p in parts: if p and p not in ["|", ".."]: options.append(p) # 🌟 Case 2, 3, 4: 子命令防呆機制 # 若在選項區塊從未發現 [現值]，且非已知格式，判定為子命令，強制轉純文字 if state == "COMPLETIONS" and 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) hint_text = "\n".join(hint_lines).strip() return { "hint": hint_text, # 若為純文字模式，強制回傳空陣列，確保前端正確渲染為 input "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: """支援多種編輯狀態格式解析""" 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()} # 🌟 1. 函數簽名加上 config_type 參數，預設為 "running" async def sync_cmts_leaves_async(host, username, password, leaf_paths: list, config_type: str = "running"): 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): print(f"🔄 正在處理第 {batch_idx + 1}/{len(batches)} 批次 (共 {len(batch)} 個路徑)...") try: # 🧹 [穩定性修復] 全面改用 async with 管理生命週期，確保連線與 process 絕對釋放 async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as 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() # 🌟 精準 Prompt 偵測 if prompt_pattern and re.search(prompt_pattern, output): break except asyncio.TimeoutError: break return output # 🌟 新增：在第一批次連線時，先抓取設備版本 if cmts_version == "unknown": process.stdin.write("show version\n") await process.stdin.drain() version_output = await read_until_quiet(timeout=1.5, prompt_pattern=r"(?:#|>)") match = re.search(r"(?:infra|vcmts-cd-0)\s+([\w\.\-]+)", version_output) if match: cmts_version = match.group(1) process.stdin.write("config\n") await process.stdin.drain() await read_until_quiet(timeout=1.5, prompt_pattern=r"\(config\)#") for path in batch: # 🌟 優化 1：強迫讓出事件迴圈控制權，讓 FastAPI 去處理其他使用者的請求 await asyncio.sleep(0.01) # --- 步驟 1：發送 '?' 查詢 --- command_to_send = f"{path} ?" process.stdin.write(command_to_send) await process.stdin.drain() output_question = await read_until_quiet(timeout=1.0) q_data = parse_question_mark_output(output_question) backspaces = "\x08" * (len(command_to_send) + 5) process.stdin.write(backspaces) await process.stdin.drain() await read_until_quiet(timeout=0.5) 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") } # --- 步驟 2：判斷是否需要發送 Enter --- 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=1.0) 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": process.stdin.write("\x03") await process.stdin.drain() else: print(f"⚠️ [Debug] 未知格式 ({path}):\n{enter_data['raw_output']}") process.stdin.write("\n") await process.stdin.drain() await read_until_quiet(timeout=0.5) result_data[path] = parsed_data processed_count += 1 event_data = { "event": "progress", "current": processed_count, "total": total_paths, "current_path": path } yield json.dumps(event_data) + "\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 # --- 步驟 3：寫入快取 (DB or JSON) --- try: db_success = False current_ts = int(time.time()) formatted_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) # 1. 嘗試寫入資料庫 if USE_DB: try: # 寫入 metadata if cmts_version != "unknown": await database.upsert_device_status(host, config_type, {"cmts_version": cmts_version, "last_scanned": formatted_time}) else: await database.upsert_device_status(host, config_type, {"last_scanned": formatted_time}) # 寫入 options db_write_count = 0 for p in batch: if p in result_data: result_data[p]["updated_at"] = current_ts success = await database.upsert_leaf_option(host, config_type, p, result_data[p]) if success: db_write_count += 1 if db_write_count > 0: db_success = True print(f"💾 [DB] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入資料庫！") else: print("⚠️ [Fallback] 資料庫寫入 0 筆，退回寫入 JSON 快取...") except Exception as e: print(f"⚠️ [Fallback] 資料庫寫入發生例外: {e}，自動切換至 JSON 快取寫入...") # 2. 如果 DB 寫入失敗或 USE_DB=False，則 Fallback 寫入 JSON if not db_success: # 🌟 動態決定快取檔名 (加入 IP 隔離) safe_host = host.replace(".", "_") cache_file = f"{safe_host}_{config_type}_cache.json" def read_json_from_file(filepath): if os.path.exists(filepath): with open(filepath, "r", encoding="utf-8") as f: return json.load(f) return {} cache_data = await asyncio.to_thread(read_json_from_file, cache_file) # 🌟 新增：寫入 Metadata (版本與掃描時間) if "__metadata__" not in cache_data: cache_data["__metadata__"] = {} if cmts_version != "unknown": cache_data["__metadata__"]["cmts_version"] = cmts_version cache_data["__metadata__"]["last_scanned"] = formatted_time for p in batch: if p in result_data: result_data[p]["updated_at"] = current_ts cache_data[p] = result_data[p] def write_json_to_file(filepath, data): with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, indent=4, ensure_ascii=False) await asyncio.to_thread(write_json_to_file, cache_file, cache_data) print(f"💾 [JSON] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入快取檔！(版本: {cmts_version}, 檔案: {cache_file})") except Exception as e: print(f"⚠️ 寫入快取失敗 (DB 與 JSON 皆失敗): {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: # 🧹 [穩定性修復] 全面改用 async with 管理生命週期 async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as 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() # 🌟 精準 Prompt 偵測 if prompt_pattern and re.search(prompt_pattern, output): break except asyncio.TimeoutError: break return output # 🌟 新增這行：剛連線成功後，先清空終端機的登入歡迎詞 (MOTD) 與雜訊 await read_until_quiet(timeout=1.0, prompt_pattern=r"(?:#|>)") # 關鍵修改：根據 config_type 切換模式與指令，並加上 | nomore 關閉分頁 if config_type == "full": # 1. 先進入 config 模式 process.stdin.write("config\n") await process.stdin.drain() await read_until_quiet(timeout=1.5, prompt_pattern=r"\(config\)#") # 等待提示字元變成 (config)# # 2. 下達 full-configuration 指令 (🌟 加上 | nomore) 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\)#") # 3. 抓完後退回上一層 (保持良好習慣) process.stdin.write("exit\n") await process.stdin.drain() else: # running-config 在一般模式即可下達 (🌟 加上 | nomore) 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() # 簡單清理頭尾的雜訊 (例如指令本身的 echo) # 🌟 關鍵修正 1：強化 ANSI 正規表達式，加入對 '?' 的支援 (精準捕捉 \x1b[?7h) cleaned_output = re.sub(r'\x1b\[[0-9;?]*[a-zA-Z]|\x08', '', raw_output) # 2. 清除失去 \x1b 殘留的字面分頁符號 (精準捕捉 [7m--More--[27m[8D[K) cleaned_output = re.sub(r'\[7m\s*--More--\s*\[27m\[\d+D\[K', '', cleaned_output) # 3. 清除純文字的 --More-- (防呆) cleaned_output = re.sub(r'\s*--More--\s*', '', cleaned_output) # 🌟 新增 4：清除 CableOS 終端機特有的 (END) 結尾標記 cleaned_output = re.sub(r'\s*\(END\)\s*', '', cleaned_output) # 關鍵修正：這裡改用 cleaned_output 來切行！ lines = cleaned_output.splitlines() # 🌟 關鍵修正 2：加入 strip() 避免空白干擾，確保精準踢掉提示字元 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)}") finally: if conn: conn.close() def parse_config_to_tree(raw_cli: str) -> dict: """將純文字設定檔轉換為簡單的階層式 JSON 樹狀圖 (Phase 3 會用到)""" # 這裡先實作一個基礎的縮排解析器，未來可依據您的設備格式優化 tree = {} # 暫時回傳空字典，確保 Phase 1 & 2 能順利走通 # 真正的樹狀解析邏輯我們可以在 Phase 3 完善 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 import database # 引入我們剛剛拆分出來的路由模組 from routers import query, config, terminal, lock, leaf_options, backup @asynccontextmanager async def lifespan(app: FastAPI): # Startup await database.init_db_pool() yield # Shutdown await database.close_db_pool() app = FastAPI(title="Harmonic CMTS Manager", version="2.0", lifespan=lifespan) # 掛載靜態檔案目錄 (對應 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") # WebSocket 通常獨立於 API 版本之外，所以不加前綴 app.include_router(terminal.router) # 根目錄路由：回傳前端 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: PROJECT_STATE.md ================================================================================ # 📖 Harmonic CMTS Manager - Project State (Living Document) > ⚠️ **AI 助手請注意**：本專案的核心架構與開發規範已移至 `.clinerules`。此檔案僅作為「專案進度存檔」與「待辦事項追蹤」使用。 --- ## ✅ 已完成開發階段 (Completed Phases) ### Phase 1: PostgreSQL 高可用性架構升級 (Completed) - [x] 成功導入 `asyncpg`，建立 `database.py` 管理非同步資料庫連線池。 - [x] 建立 `cmts_options`, `device_status`, `system_filters` 資料表，嚴格遵守 `config_type` 雙軌隔離的主鍵設計。 - [x] 實踐完整的「**Zero-Downtime Fallback 機制**」：資料庫連線異常時，自動退回使用 JSON 檔案讀寫。 - [x] 調整 `init_db.py` 為非同步啟動腳本。 ### Phase 2: 設備配置備份與快照機制 (Completed) - [x] **資料庫擴充**：在 PostgreSQL 中建立 `config_backups` 資料表 (包含 `id`, `host`, `timestamp`, `raw_cli`, `parsed_tree`, `snapshot_name`, `description`)。 - [x] **前端 UI 實作**：完成「設備備份與還原」頁籤，包含建立快照表單與歷史紀錄列表 (馬卡龍色系與 Flexbox 佈局)。 - [x] **後端 API 實作**：完成手動建立快照與取得歷史快照列表的 API 路由。 - [x] **前端 UI 優化與防禦性編程**：完成滿版視覺重構、原生日期選擇器整合，並實作具備 Null-Safety 與 `.trim()` 容錯的多維度前端搜尋過濾器 (支援快照名稱 + 描述雙欄位比對)。 ### Phase 3: 智慧差異還原與歷史預覽 (Completed) - [x] **歷史預覽 UI**：前端支援點擊歷史快照的「檢視」按鈕。 - [x] **前端安全還原防呆**：實作三階段安全還原流程 UI (包含 Diff 預覽與確認寫入按鈕)，並加入跨設備還原阻斷機制。 - [x] **後端 Diff 引擎實作**：完成 `/api/v1/backups/{id}/diff`，成功生成絕對路徑指令陣列。 - [x] **前端串流接收器 (Streaming UI)**：升級 `executeSmartRestore`，使用 `ReadableStream` 即時渲染後端傳來的 SSH 逐行執行 Log，並具備自動捲動功能。 - [x] **後端 SSH 交易寫入管道 (Transactional SSH Pipeline)**： - 實作 `/api/v1/backups/{id}/restore` API，採用 `StreamingResponse` (NDJSON 串流回應)。 - 建立安全的逐行寫入機制 (Fail-safe)，遇錯立即停止、發送 `abort` 撤銷變更，並回傳錯誤 Chunk。 - [x] **效能與體驗優化**：完成 DOM 陣列渲染優化、Terminal 資源回收 (防殭屍連線)、精準 Prompt 偵測，以及 UI 視窗連動防呆機制。 --- ## 🚧 目前開發階段 (Current Phase) ### Phase 4: 自動化防護與進階管理 (Automation & Advanced Management) - [ ] **自動備份攔截 (Auto-Backup Interceptor)**：在執行任何 `generate_cli` (寫入設備變更) 之前，實作自動觸發背景備份 `running config` 的防呆機制，確保每次變更前都有還原點。 - [ ] **備份保留策略 (Retention Policy)**：實作定期清理過期或過多歷史快照的機制，防止資料庫無限膨脹。 --- ## 🐛 已知問題與未來計畫 (Known Issues & Backlog) - 目前系統運行穩定，各項併發鎖定與 UI 狀態連動皆已完善。等待進入 Phase 4 開發。 ================================================================================ 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(); } // 🌟 1. 取得選項快取 (加上 configType 查詢參數) export async function apiGetLeafOptions(host, configType = 'running') { const response = await fetch(`/api/v1/cmts-leaf-options?host=${encodeURIComponent(host)}&config_type=${configType}`); return response.json(); } // 🌟 2. 同步選項快取 (將 config_type 塞入 Body) export async function apiSyncLeafOptions(host, paths, configType = 'running') { return fetch('/api/v1/cmts-leaf-options/sync', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ host, leaf_paths: paths, config_type: configType }) }); } // 🌟 3. 清除快取 (加上 configType 查詢參數) export async function apiClearCache(host, paths, configType = 'running') { const response = await fetch(`/api/v1/clear_cache?host=${encodeURIComponent(host)}&config_type=${configType}`, { 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, configType = 'running') { try { const response = await fetch(`/api/v1/scan-status?host=${encodeURIComponent(host)}&config_type=${configType}`); 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(); } // ========================================== // 6. 專門處理串流的 API 呼叫函數 (UI 可以即時更新) // ========================================== // 🌟 4. 串流掃描 API (將 config_type 塞入 Body) export async function apiSyncLeafOptionsStream(host, paths, onProgress, configType = 'running') { const response = await fetch('/api/v1/cmts-leaf-options/sync', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ host, leaf_paths: paths, config_type: configType }) }); 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 }); // 因為一次可能收到多行 JSON，我們用換行符號切開 const lines = chunk.split("\n").filter(line => line.trim() !== ""); for (const line of lines) { try { const data = JSON.parse(line); onProgress(data); // 將解析後的資料傳給 UI 介面 } catch (e) { console.error("JSON 解析錯誤:", e, line); } } } } ================================================================================ 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 } 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'; // ============================================================================ // 🌟 1. 全域生命週期、閒置偵測與 SSE 廣播監聽 // ============================================================================ let idleTimer; const IDLE_TIMEOUT = 300000; // 5 分鐘 (毫秒) let globalSSE = null; // 初始化全域 SSE 監聽器，加上 mode 參數，並在連線前關閉舊頻道 function initGlobalSSE(mode = 'running') { const connInfo = getGlobalConnectionInfo(); if (!connInfo || !connInfo.host) return; // 🌟 確保有連線才建立 SSE if (globalSSE) globalSSE.close(); globalSSE = new EventSource(`/api/v1/cmts-leaf-options/stream?host=${encodeURIComponent(connInfo.host)}&config_type=${mode}`); 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 () => { 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}`; // 🛡️ 防閃爍機制：如果這個鎖是我們剛剛才主動釋放的，忽略後端傳來的舊狀態 (冷卻 8 秒) if (window.recentlyReleasedLocks[lockKey]) { if (now - window.recentlyReleasedLocks[lockKey] < 8000) { continue; // 跳過，不當作鎖定 } else { delete window.recentlyReleasedLocks[lockKey]; // 超過冷卻時間，清除記憶 } } newLockState.add(lockKey); // 記憶體加入 Host 標籤 const safePath = path.replace(/\\/g, '\\\\').replace(/"/g, '\\"'); // 🌟 擴大選擇器：同時選取完全匹配的節點，以及所有子節點 const targetBtns = document.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) { // 判斷是別人鎖的，還是自己在另一個視圖鎖的 if (lockInfo.user_id !== SESSION_USER_ID) { btn.title = `🔒 此區塊正由 [${lockInfo.username}] 編輯中`; } else { btn.title = `🔒 您正在另一個視圖編輯此項目`; } } 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 targetBtns = document.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); } // 切換「設備查詢」內的子任務表單 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'); } // 🌟 新增一個變數來記錄上一次的樹狀圖模式 let lastTreeMode = null; // 切換「設備配置」內的子任務表單 function switchConfigTask() { document.querySelectorAll('#config-tab .task-form').forEach(form => { form.classList.remove('active'); form.style.display = 'none'; }); const selectedTaskId = document.getElementById('configTask').value; let targetFormId = selectedTaskId; if (selectedTaskId === 'form-running-config' || selectedTaskId === 'form-full-config') { targetFormId = 'form-tree-config'; } const targetForm = document.getElementById(targetFormId); if (targetForm) { targetForm.classList.add('active'); targetForm.style.display = 'block'; const titleSpan = document.getElementById('tree-form-title'); const currentMode = selectedTaskId === 'form-full-config' ? 'full' : 'running'; if (titleSpan) { titleSpan.textContent = currentMode === 'running' ? '🌳 設備配置樹狀圖 (running-config)' : '🌳 完整設備配置樹狀圖 (full configuration)'; } if (targetFormId === 'form-tree-config') { // 🌟 魔法所在：純視覺切換，不銷毀資料！ const runningContainer = document.getElementById('tree-container-running'); const fullContainer = document.getElementById('tree-container-full'); if (runningContainer) runningContainer.style.display = currentMode === 'running' ? 'block' : 'none'; if (fullContainer) fullContainer.style.display = currentMode === 'full' ? 'block' : 'none'; if (lastTreeMode !== currentMode) { const statusEl = document.getElementById('scan-status'); if (statusEl) statusEl.textContent = ''; // 🌟 新增：如果使用者在載入途中切換模式，強制把沙漏文字隱藏！ const loadingMsg = document.getElementById('loading-message'); if (loadingMsg) loadingMsg.style.display = 'none'; lastTreeMode = currentMode; } } initGlobalSSE(currentMode); } } // 啟動配置任務 (樹狀圖載入 或 MAC Domain 精靈) function startConfigTask() { if (!isConnected) { alert("請先點擊畫面上方的「連線至 CMTS」成功後，再執行載入任務！"); return; } switchConfigTask(); const selectedTaskId = document.getElementById('configTask').value; if (selectedTaskId === 'form-bonding-config') { if (hasMacDomainData() && !confirm("重新載入將會清除您目前未套用的設定，確定要繼續嗎？")) return; resetAllManagerData(); loadMacDomainList(); } // 🌟 修正：正確的括號閉合，並根據下拉選單傳遞 configType else if (selectedTaskId === 'form-running-config') { fetchFullConfig('running'); } else if (selectedTaskId === 'form-full-config') { fetchFullConfig('full'); // 🌟 補上 full 參數 } } // 切換查詢輸入框的啟用狀態 (全域指令不需輸入目標) function toggleQueryInputs(mode) { const type = document.getElementById('queryType' + mode).value; const targetInput = document.getElementById('queryTarget' + mode); const globalTypes = ['partial_mode', 'hop', 'flap_list', 'flap_sum']; if (globalTypes.includes(type)) { targetInput.disabled = true; targetInput.style.backgroundColor = '#e8ecef'; targetInput.placeholder = "此查詢為全域指令，不需輸入目標"; targetInput.value = ""; } else { targetInput.disabled = false; targetInput.style.backgroundColor = '#f8f9fa'; targetInput.placeholder = mode === 'Cm' ? "例: 6467.7240.4076 (留白則查詢全體)" : "例: 13:0 (留白則查詢全體)"; } } // 重置所有管理介面的資料與 UI 狀態 function resetAllManagerData() { resetMacDomainData(); // 🧹 [修復 Leak] 清空前端樹狀圖快取，避免切換設備時發生 OOM clearTreeCache(); // 🛡️ [跨設備防護] 切換設備時，清空鎖定輪詢的記憶體 if (typeof activeLockState !== 'undefined' && activeLockState.clear) { activeLockState.clear(); } const configArea = document.getElementById('macDomainConfigArea'); if (configArea) configArea.style.display = 'none'; const previewEl = document.getElementById('cliPreviewContainer'); if (previewEl) { previewEl.textContent = ''; previewEl.style.display = 'none'; } const deployBtn = document.getElementById('btnDeployConfig'); if (deployBtn) { deployBtn.style.display = 'none'; deployBtn.disabled = true; } const mdSelect = document.getElementById('cfgMacDomain'); if (mdSelect) mdSelect.innerHTML = ''; const fetchStatus = document.getElementById('fetchStatus'); if (fetchStatus) { fetchStatus.textContent = "請先讀取設備狀態..."; fetchStatus.style.color = "#7f8c8d"; } document.getElementById('queryTargetCm').value = ''; document.getElementById('queryTargetRpd').value = ''; // 🌟 新增：切換設備連線後，自動重整備份歷史紀錄 if (typeof loadBackupHistory === 'function') { loadBackupHistory(); } } // ============================================================================ // 🌟 3. 設備查詢與全域配置載入 (Query & Full Config) // ============================================================================ // 執行綜合查詢 (CM / RPD) - 🌟 導入前端模擬動態訊息流 async function executeQuery(mode) { if (!isConnected) return alert("請先點擊畫面上方的「連線至 CMTS」按鈕！"); const connInfo = getGlobalConnectionInfo(); if (!connInfo) return; const queryType = document.getElementById('queryType' + mode).value; const target = document.getElementById('queryTarget' + mode).value.trim(); const isDebug = document.getElementById('debugQuery' + mode).checked; openModal(connInfo.host); const outputEl = document.getElementById('modalOutput'); // 🌟 模擬動態訊息流邏輯 const states = [ `🔌 正在建立 SSH 連線至 ${connInfo.host}...`, `📥 正在向設備發送查詢指令...`, `🧩 正在等待設備回傳資料...` ]; let progressIdx = 0; outputEl.innerHTML = `${states[0]}`; const progressInterval = setInterval(() => { progressIdx++; if (progressIdx < states.length) { const textEl = document.getElementById('query-progress-text'); if (textEl) textEl.innerText = states[progressIdx]; } }, 1500); // 每 1.5 秒切換一次狀態 try { const result = await apiExecuteQuery(queryType, target, connInfo.host, connInfo.user, connInfo.pass); clearInterval(progressInterval); // 收到結果，停止輪播 if (result.status === 'success') { outputEl.style.color = "#e0e0e0"; outputEl.textContent = isDebug ? `[DEBUG] 實際發送指令: ${result.command}\n==================================================\n${result.data}` : result.data; } else { outputEl.style.color = "#e74c3c"; outputEl.textContent = `查詢失敗:\n${result.detail || result.message}`; } } catch (error) { clearInterval(progressInterval); // 發生錯誤，停止輪播 outputEl.style.color = "#e74c3c"; outputEl.textContent = `連線失敗: ${error}`; } } // 載入完整設備配置並生成樹狀圖，預設為 running (平順過渡終極版 🚀) async function fetchFullConfig(configType = 'running') { const loadingMsg = document.getElementById('loading-message'); const treeContainer = document.getElementById(`tree-container-${configType}`); const connInfo = getGlobalConnectionInfo(); if (!connInfo) return; if (loadingMsg) { loadingMsg.style.display = 'inline-block'; loadingMsg.style.color = '#f39c12'; // 🌟 確保初始狀態為橘色 loadingMsg.innerHTML = '⏳ 準備連線至設備...'; } if (treeContainer) treeContainer.innerHTML = ''; let needAutoScan = false; let progressInterval = null; try { try { const versionRes = await apiGetCmtsVersion(connInfo.host, connInfo.user, connInfo.pass); if (versionRes.status === 'success') { const currentVersion = versionRes.version; const optRes = await fetch(`/api/v1/cmts-leaf-options?host=${encodeURIComponent(connInfo.host)}&config_type=${configType}`); const optData = await optRes.json(); const cachedVersion = optData.__metadata__?.cmts_version; if (cachedVersion && cachedVersion !== currentVersion) { const doClear = confirm(`⚠️ 偵測到 CMTS 韌體版本已變更！\n\n設備當前版本：${currentVersion}\n快取紀錄版本：${cachedVersion}\n\n為確保指令正確，系統強烈建議清空舊版快取。是否立即清空？`); if (doClear) { const allKeys = Object.keys(optData); await apiClearCache(connInfo.host, allKeys, configType); needAutoScan = true; } } } } catch (vErr) { console.warn("版本檢查失敗，略過", vErr); } const response = await fetch(`/api/v1/cmts-full-config/stream?host=${encodeURIComponent(connInfo.host)}&username=${encodeURIComponent(connInfo.user)}&password=${encodeURIComponent(connInfo.pass)}&config_type=${configType}`); if (!response.ok) throw new Error(`伺服器回應錯誤: ${response.status}`); const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8"); let buffer = ""; let finalTreeData = null; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); let lines = buffer.split('\n'); buffer = lines.pop(); for (let line of lines) { if (!line.trim()) continue; try { const data = JSON.parse(line); if (data.status === 'progress') { if (data.message.includes('正在下載')) { let progress = 0; if (progressInterval) clearInterval(progressInterval); progressInterval = setInterval(() => { progress += (95 - progress) * 0.06; if (loadingMsg) { loadingMsg.innerHTML = `📥 正在下載 ${configType} 配置檔 (${Math.round(progress)}%)...`; } }, 500); } else { if (progressInterval) { clearInterval(progressInterval); progressInterval = null; if (loadingMsg) { loadingMsg.innerHTML = `📥 正在下載 ${configType} 配置檔 (100%) - 下載完成！`; loadingMsg.style.color = "#27ae60"; } await new Promise(resolve => setTimeout(resolve, 600)); if (loadingMsg) loadingMsg.style.color = "#f39c12"; // 🌟 恢復橘色 } if (loadingMsg) { loadingMsg.style.color = "#f39c12"; // 🌟 確保橘色 loadingMsg.innerHTML = data.message; } } } else if (data.status === 'success') { finalTreeData = data.data; } else if (data.status === 'error') { throw new Error(data.message); } } catch (e) { console.warn("解析串流 JSON 失敗:", line); } } } if (!finalTreeData) throw new Error("未收到完整的配置資料，連線可能中斷。"); window.currentTreeData = finalTreeData; const currentSelectedTask = document.getElementById('configTask').value; const expectedTask = configType === 'running' ? 'form-running-config' : 'form-full-config'; if (currentSelectedTask !== expectedTask) return; if (treeContainer) { treeContainer.innerHTML = buildTree(finalTreeData, '', false, configType); if (loadingMsg) loadingMsg.style.display = 'none'; const isSomeoneScanning = await apiGetScanStatus(connInfo.host, configType); if (isSomeoneScanning) { alert("💡 提示：系統正在背景更新設備選項快取，部分選單題示內容可能稍後才會顯示完整。"); lockScanButton(60000); } else { enableGlobalScanButton(); if (needAutoScan) { setTimeout(() => scanGlobalMissingOptions(configType), 500); } } } } catch (error) { if (treeContainer) treeContainer.innerHTML = `
❌ 連線或解析失敗: ${error.message}
`; } finally { if (progressInterval) clearInterval(progressInterval); if (loadingMsg) { loadingMsg.style.display = 'none'; loadingMsg.style.color = "#f39c12"; // 🌟 確保重置為橘色 } } } // ============================================================================ // 🌟 4. 選項快取與背景掃描 (Cache & Background Scanning) // ============================================================================ // 為現有的 input 加上下拉選項 (不破壞原有結構)，加上 mode 參數 async function enhanceInputWithOptions(path, elementId, mode = 'running') { // 🌟 1. 取得當前連線的 IP const connInfo = getGlobalConnectionInfo(); if (!connInfo || !connInfo.host) return; try { // 🌟 2. GET 請求：網址加上 host 參數 const optRes = await fetch(`/api/v1/cmts-leaf-options?host=${encodeURIComponent(connInfo.host)}&config_type=${mode}`); const optData = await optRes.json(); const cacheKey = path.replace(/::/g, ' ').replace(/\s*\[\d+\]/g, ''); const leafCache = optData[cacheKey]; if (leafCache && leafCache.options && leafCache.options.length > 0) { const container = document.getElementById(`container-${elementId}`); const input = container.querySelector('.edit-input'); if (input) { const listId = `dl-${elementId}`; input.setAttribute('list', listId); let dataList = document.getElementById(listId); if (!dataList) { dataList = document.createElement('datalist'); dataList.id = listId; container.appendChild(dataList); } dataList.innerHTML = leafCache.options.map(opt => `'; statusSpan.textContent = "正在自動抓取現有 MAC Domain 清單..."; statusSpan.style.color = "#f39c12"; try { const result = await apiGetMacDomainList(connInfo.host, connInfo.user, connInfo.pass); if (result.status === 'success' && result.data.length > 0) { mdSelect.innerHTML = ''; result.data.forEach(md => { const option = document.createElement('option'); option.value = md; option.textContent = md; mdSelect.appendChild(option); }); statusSpan.textContent = "✅ 清單抓取成功！請選擇目標並點擊讀取配置。"; statusSpan.style.color = "#27ae60"; } else { mdSelect.innerHTML = ''; statusSpan.textContent = "無法解析清單，請確認設備狀態。"; statusSpan.style.color = "#e74c3c"; } } catch (error) { mdSelect.innerHTML = ''; statusSpan.textContent = "連線失敗：" + error; statusSpan.style.color = "#e74c3c"; } } export async function fetchMacDomainConfig() { const target = document.getElementById('cfgMacDomain').value.trim(); const connInfo = getGlobalConnectionInfo(); if (!connInfo || !target) return alert("請確認設備連線資訊與 MAC Domain！"); document.getElementById('fetchStatus').textContent = "⏳ 正在讀取並解析設定中..."; document.getElementById('fetchStatus').style.color = "#f39c12"; try { const result = await apiGetMacDomainConfig(target, connInfo.host, connInfo.user, connInfo.pass); if (result.status === 'success') { currentMacDomainData = result.data; populateFormFromData(); document.getElementById('macDomainConfigArea').style.display = 'block'; document.getElementById('fetchStatus').textContent = "✅ 讀取成功！"; document.getElementById('fetchStatus').style.color = "#27ae60"; } else { alert("讀取失敗：" + result.detail); document.getElementById('fetchStatus').textContent = "❌ 讀取失敗"; } } catch (error) { alert("連線錯誤：" + error); } } export function populateFormFromData() { const d = currentMacDomainData; if (!d) return; document.getElementById('g_ip_prov').value = d.common_settings['ip-provisioning-mode'] || 'dual-stack'; document.getElementById('g_diplexer').value = d.common_settings['diplexer-band-edge'] || 'enabled'; document.getElementById('g_bat31').value = d.common_settings['cm-battery-mode-31-support'] || 'disabled'; document.getElementById('g_bat30').value = d.common_settings['cm-battery-mode-30-support'] || 'disabled'; document.getElementById('g_docsis40').value = d.common_settings['docsis40'] || 'disabled'; document.getElementById('g_ds_dyn').value = d.common_settings['ds-dynamic-bonding-group'] || 'disabled'; document.getElementById('g_us_dyn').value = d.common_settings['us-dynamic-bonding-group'] || 'disabled'; document.getElementById('b_admin').value = d.basic_channel_sets['admin-state'] || 'down'; document.getElementById('b_ds_pri').value = d.basic_channel_sets['ds-primary-set'] || ''; document.getElementById('b_ds_non_pri').value = d.basic_channel_sets['ds-non-primary-set'] || ''; document.getElementById('b_us_phy').value = d.basic_channel_sets['us-phy-channel-set'] || ''; document.getElementById('b_ds_ofdm').value = d.basic_channel_sets['ds-ofdm-set'] || ''; document.getElementById('b_us_ofdma').value = d.basic_channel_sets['us-ofdma-set'] || ''; const dsSelect = document.getElementById('select_ds_group'); dsSelect.innerHTML = ''; Object.keys(d.static_ds_bonding_groups).forEach(grp => dsSelect.innerHTML += ``); if (Object.keys(d.static_ds_bonding_groups).length > 0) dsSelect.value = Object.keys(d.static_ds_bonding_groups)[0]; const usSelect = document.getElementById('select_us_group'); usSelect.innerHTML = ''; Object.keys(d.static_us_bonding_groups).forEach(grp => usSelect.innerHTML += ``); if (Object.keys(d.static_us_bonding_groups).length > 0) usSelect.value = Object.keys(d.static_us_bonding_groups)[0]; toggleGroupVisibility(); loadDsGroupData(); loadUsGroupData(); } export function revertFormChanges() { if (!currentMacDomainData) return; if (!confirm("確定要放棄目前的修改，還原回設備原始的設定值嗎？")) return; populateFormFromData(); const previewEl = document.getElementById('cliPreviewContainer'); if (previewEl) { previewEl.textContent = ''; previewEl.style.display = 'none'; } const deployBtn = document.getElementById('btnDeployConfig'); if (deployBtn) { deployBtn.style.display = 'none'; deployBtn.disabled = true; } const fetchStatus = document.getElementById('fetchStatus'); fetchStatus.textContent = "🔄 已還原為原始設定！"; fetchStatus.style.color = "#2980b9"; setTimeout(() => { fetchStatus.textContent = "✅ 讀取成功！"; fetchStatus.style.color = "#27ae60"; }, 3000); } export function toggleGroupVisibility() { const dsDyn = document.getElementById('g_ds_dyn').value; const usDyn = document.getElementById('g_us_dyn').value; document.getElementById('section_group_ds').style.display = (dsDyn === 'disabled') ? 'block' : 'none'; document.getElementById('section_group_us').style.display = (usDyn === 'disabled') ? 'block' : 'none'; } export function loadDsGroupData() { const grp = document.getElementById('select_ds_group').value; const newGrpInput = document.getElementById('input_new_ds_group'); if (grp === '_new_') { newGrpInput.style.display = 'inline-block'; document.getElementById('ds_g_admin').value = 'down'; document.getElementById('ds_g_down').value = ''; document.getElementById('ds_g_ofdm').value = ''; document.getElementById('ds_g_fdx').value = ''; } else { newGrpInput.style.display = 'none'; const data = currentMacDomainData.static_ds_bonding_groups[grp]; document.getElementById('ds_g_admin').value = data['admin-state'] || 'down'; document.getElementById('ds_g_down').value = data['down-channel-set'] || ''; document.getElementById('ds_g_ofdm').value = data['ofdm-channel-set'] || ''; document.getElementById('ds_g_fdx').value = data['fdx-ofdm-channel-set'] || ''; } } export function loadUsGroupData() { const grp = document.getElementById('select_us_group').value; const newGrpInput = document.getElementById('input_new_us_group'); if (grp === '_new_') { newGrpInput.style.display = 'inline-block'; document.getElementById('us_g_admin').value = 'down'; document.getElementById('us_g_us').value = ''; document.getElementById('us_g_ofdma').value = ''; document.getElementById('us_g_fdx').value = ''; } else { newGrpInput.style.display = 'none'; const data = currentMacDomainData.static_us_bonding_groups[grp]; document.getElementById('us_g_admin').value = data['admin-state'] || 'down'; document.getElementById('us_g_us').value = data['us-channel-set'] || ''; document.getElementById('us_g_ofdma').value = data['ofdma-channel-set'] || ''; document.getElementById('us_g_fdx').value = data['fdx-ofdma-channel-set'] || ''; } } export function generateMacDomainCLI() { const md = document.getElementById('cfgMacDomain').value.trim(); let cli = `! --- Harmonic MAC Domain Configuration SOP ---\n`; cli += `! 1. [區塊] MAC Domain 全域與 Base 設定\n`; cli += `cable mac-domain ${md} admin-state down\ncommit\n`; const baseCmds = [ `ip-provisioning-mode ${document.getElementById('g_ip_prov').value}`, `diplexer-band-edge control ${document.getElementById('g_diplexer').value}`, `cm-battery-mode-31-support ${document.getElementById('g_bat31').value}`, `cm-battery-mode-30-support ${document.getElementById('g_bat30').value}`, `docsis40 ${document.getElementById('g_docsis40').value}`, `ds-dynamic-bonding-group ${document.getElementById('g_ds_dyn').value}`, `us-dynamic-bonding-group ${document.getElementById('g_us_dyn').value}`, `ds-primary-set ${document.getElementById('b_ds_pri').value}`, `ds-non-primary-set ${document.getElementById('b_ds_non_pri').value}`, `us-phy-channel-set ${document.getElementById('b_us_phy').value}`, `ds-ofdm-set ${document.getElementById('b_ds_ofdm').value}`, `us-ofdma-set ${document.getElementById('b_us_ofdma').value}` ]; baseCmds.forEach(cmd => { if (!cmd.endsWith(" ")) cli += `cable mac-domain ${md} ${cmd}\n`; }); cli += `cable mac-domain ${md} admin-state ${document.getElementById('b_admin').value}\ncommit\n`; if (document.getElementById('g_ds_dyn').value === 'disabled') { const isNewDs = document.getElementById('select_ds_group').value === '_new_'; let dsName = isNewDs ? document.getElementById('input_new_ds_group').value.trim() : document.getElementById('select_ds_group').value; if (dsName) { cli += `\n! 2. [區塊] DS Bonding Group [${dsName}] 設定\n`; if (!isNewDs) cli += `cable mac-domain ${md} ds-bonding-group ${dsName} admin-state down\ncommit\n`; if (document.getElementById('ds_g_down').value) cli += `cable mac-domain ${md} ds-bonding-group ${dsName} down-channel-set ${document.getElementById('ds_g_down').value}\n`; if (document.getElementById('ds_g_ofdm').value) cli += `cable mac-domain ${md} ds-bonding-group ${dsName} ofdm-channel-set ${document.getElementById('ds_g_ofdm').value}\n`; if (document.getElementById('ds_g_fdx').value) cli += `cable mac-domain ${md} ds-bonding-group ${dsName} fdx-ofdm-channel-set ${document.getElementById('ds_g_fdx').value}\n`; cli += `cable mac-domain ${md} ds-bonding-group ${dsName} admin-state ${document.getElementById('ds_g_admin').value}\ncommit\n`; } } if (document.getElementById('g_us_dyn').value === 'disabled') { const isNewUs = document.getElementById('select_us_group').value === '_new_'; let usName = isNewUs ? document.getElementById('input_new_us_group').value.trim() : document.getElementById('select_us_group').value; if (usName) { cli += `\n! 3. [區塊] US Bonding Group [${usName}] 設定\n`; if (!isNewUs) cli += `cable mac-domain ${md} us-bonding-group ${usName} admin-state down\ncommit\n`; if (document.getElementById('us_g_us').value) cli += `cable mac-domain ${md} us-bonding-group ${usName} us-channel-set ${document.getElementById('us_g_us').value}\n`; if (document.getElementById('us_g_ofdma').value) cli += `cable mac-domain ${md} us-bonding-group ${usName} ofdma-channel-set ${document.getElementById('us_g_ofdma').value}\n`; if (document.getElementById('us_g_fdx').value) cli += `cable mac-domain ${md} us-bonding-group ${usName} fdx-ofdma-channel-set ${document.getElementById('us_g_fdx').value}\n`; cli += `cable mac-domain ${md} us-bonding-group ${usName} admin-state ${document.getElementById('us_g_admin').value}\ncommit\n`; } } const previewEl = document.getElementById('cliPreviewContainer'); previewEl.textContent = cli; previewEl.style.display = 'block'; const deployBtn = document.getElementById('btnDeployConfig'); deployBtn.style.display = 'inline-block'; deployBtn.disabled = false; } export async function executeBondingConfig() { if (!confirm("⚠️ 警告：此操作將導致 MAC Domain 重新啟動，影響用戶服務。確定要執行嗎？")) return; const cliScript = document.getElementById('cliPreviewContainer').textContent; const connInfo = getGlobalConnectionInfo(); if (!connInfo) return; // 開啟 Modal 顯示執行進度 document.getElementById('outputModal').classList.add('active'); document.getElementById('modalTargetInfo').textContent = connInfo.host ? `(${connInfo.host})` : ''; document.body.style.overflow = 'hidden'; const outputEl = document.getElementById('modalOutput'); outputEl.innerHTML = `🚀 正在排隊並執行配置腳本，這可能需要幾秒鐘，請勿關閉視窗...\n\n${cliScript}`; // 鎖定關閉按鈕 (Execution Lock) const closeBtn = document.getElementById('btn-modal-close'); if (closeBtn) { closeBtn.disabled = true; closeBtn.style.opacity = '0.5'; closeBtn.style.cursor = 'not-allowed'; closeBtn.style.pointerEvents = 'none'; } try { const result = await apiExecuteConfig(cliScript, connInfo.host, connInfo.user, connInfo.pass); if (result.status === 'success') { outputEl.style.color = "#e0e0e0"; outputEl.textContent = `✅ 設定執行完成！\n==================================================\n${result.data}`; } else { outputEl.style.color = "#e74c3c"; outputEl.textContent = `❌ 設定執行失敗:\n${result.message}`; } } catch (error) { outputEl.style.color = "#e74c3c"; outputEl.textContent = `❌ 連線或伺服器錯誤: ${error}`; } finally { // 恢復關閉按鈕 if (closeBtn) { closeBtn.disabled = false; closeBtn.style.opacity = '1'; closeBtn.style.cursor = 'pointer'; closeBtn.style.pointerEvents = 'auto'; } } } ================================================================================ FILE: static/utils.js ================================================================================ // --- static/utils.js --- // 加上 export，讓其他檔案可以使用這個函數 export function getGlobalConnectionInfo() { 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) { alert("請在畫面上方完整輸入目標設備的 IP、帳號與密碼！"); return null; } return { host, user, pass }; } ================================================================================ FILE: static/tree-ui.js ================================================================================ // --- static/tree-ui.js --- import { SESSION_USER_ID, currentEditElementId } from './edit-mode.js'; // ========================================== // 🌟 效能終極優化：全域快取與延遲渲染 (Lazy Rendering) // ========================================== export const folderDataCache = {}; export const filterFolderCache = {}; // 🧹 [修復 Leak] 徹底清空樹狀圖快取，釋放記憶體 export function clearTreeCache() { for (let key in folderDataCache) delete folderDataCache[key]; for (let key in filterFolderCache) delete filterFolderCache[key]; } // 魔法函數：當資料夾被點擊展開時，才將 HTML 渲染進去 window.lazyLoadFolder = function(elementId, isFilter = false) { const detailsEl = document.getElementById(`details-${elementId}`); // 如果已經載入過，就不重複渲染 if (!detailsEl || detailsEl.dataset.loaded === 'true') return; // 從 dataset 讀取當初存下來的屬性 const currentPath = detailsEl.dataset.path; const isCommandGroup = detailsEl.dataset.isGroup === 'true'; const mode = detailsEl.dataset.mode; const contentDiv = document.getElementById(isFilter ? `filter-content-${elementId}` : `content-${elementId}`); const cache = isFilter ? filterFolderCache[elementId] : folderDataCache[elementId]; if (contentDiv && cache) { if (isFilter) { contentDiv.innerHTML = buildRealFilterTree(cache.data, currentPath, cache.hiddenKeys, isCommandGroup, mode); } else { contentDiv.innerHTML = buildTree(cache, currentPath, isCommandGroup, mode); } detailsEl.dataset.loaded = 'true'; // 標記為已載入 } }; // ========================================== // 🌟 輔助函數：全部展開與全部收合 (支援延遲渲染) // ========================================== export function expandAll(elementId) { const details = document.getElementById(`details-${elementId}`); if (details) { // 如果還沒載入，先強制載入它 if (!details.dataset.loaded) { const isFilter = details.id.includes('filter-'); window.lazyLoadFolder(elementId, isFilter); } details.open = true; // 找出剛剛渲染出來的第一層子資料夾，遞迴展開 const contentDiv = details.querySelector(':scope > div'); if (contentDiv) { const childDetails = contentDiv.querySelectorAll(':scope > details'); childDetails.forEach(d => { const subId = d.id.replace('details-', ''); expandAll(subId); // 遞迴展開 }); } } } export function collapseAll(elementId) { const details = document.getElementById(`details-${elementId}`); if (details) { const subDetails = details.querySelectorAll('details'); subDetails.forEach(d => d.open = false); details.open = false; } } // ========================================== // 🌟 核心函數：生成完整設備配置樹狀圖 (Lazy 版) // ========================================== export function buildTree(node, path = '', isParentCommandGroup = false, mode = 'running') { const htmlParts = []; if (!node || typeof node !== 'object') return htmlParts.join(''); for (const [key, value] of Object.entries(node)) { const currentPath = path ? `${path}::${key}` : key; let cliCommand = currentPath.replace(/::/g, ' ').replace(/\s*\[\d+\]/g, ''); if (typeof value === 'object' && value !== null) { const hasNestedObject = Object.values(value).some(v => typeof v === 'object' && v !== null); const isCommandGroup = isParentCommandGroup || !hasNestedObject; const elementId = `${mode}-folder-${currentPath.replace(/[^a-zA-Z0-9]/g, '-')}`; const folderSuffix = isCommandGroup ? '(指令群組)' : ''; // 🌟 將資料存入快取，供延遲渲染使用 folderDataCache[elementId] = value; const svgFolderClosed = ``; const svgFolderOpen = ``; const svgGroupClosed = ``; const svgGroupOpen = ``; const iconClosed = isCommandGroup ? svgGroupClosed : svgFolderClosed; const iconOpen = isCommandGroup ? svgGroupOpen : svgFolderOpen; const expandAllIcon = ``; const collapseAllIcon = ``; const expandCollapseBtns = ` ${expandAllIcon} ${collapseAllIcon} `; // 🌟 延遲渲染防護：取得當前設備 IP，並檢查此節點是否已被鎖定 (跨設備隔離) const currentHost = document.getElementById('cmtsHost')?.value.trim(); let isLocked = false; let lockTitle = "鎖定並編輯此群組"; let lockText = "✏️"; let lockOpacity = "0.3"; let lockPointer = "auto"; if (currentHost && window.globalActiveLocks && window.globalActiveLocks[currentHost]) { const hostLocks = window.globalActiveLocks[currentHost]; // 1. 檢查自己是否被鎖定 if (hostLocks[currentPath]) { const lockInfo = hostLocks[currentPath]; if (!(lockInfo.user_id === SESSION_USER_ID && elementId === currentEditElementId)) { isLocked = true; lockText = "🔒"; lockOpacity = "0.2"; lockPointer = "none"; lockTitle = lockInfo.user_id !== SESSION_USER_ID ? `🔒 此區塊正由 [${lockInfo.username}] 編輯中` : `🔒 您正在另一個視圖編輯此項目`; } } else { // 2. 檢查父層級是否被鎖定 (Cascading Lock) for (const [lockedPath, lockInfo] of Object.entries(hostLocks)) { if (currentPath.startsWith(lockedPath + "::")) { isLocked = true; lockText = "🔒"; lockOpacity = "0.2"; lockPointer = "none"; lockTitle = `🔒 父層級已被鎖定，無法編輯此項目`; break; } } } } htmlParts.push(`
${iconClosed} ${key}${folderSuffix} ${expandCollapseBtns}
${lockText}
`); } else { const isVirtualIndex = /^\[\d+\]$/.test(key); const isNoCommand = (key === 'no'); const safeValue = (value === null ? '' : value).toString().replace(/"/g, """); const elementId = `${mode}-leaf-${currentPath.replace(/[^a-zA-Z0-9]/g, '-')}`; if (isNoCommand) { let baseCmd = cliCommand.replace(/(^|\s)no$/, '').trim(); cliCommand = baseCmd ? `${baseCmd} no ${safeValue}` : `no ${safeValue}`; } const svgLeaf = ``; const svgDisabled = ``; const iconHtml = (icon) => `${icon}`; const valueStyle = `color: #16a085; margin-left: 5px; font-family: Courier New, monospace; display: inline-block; transform: translateY(1.5px);`; // 🌟 延遲渲染防護：取得當前設備 IP，並檢查此節點是否已被鎖定 (跨設備隔離) const currentHost = document.getElementById('cmtsHost')?.value.trim(); let isLocked = false; let lockTitle = "鎖定並編輯此項目"; let lockText = "✏️"; let lockOpacity = "0.3"; let lockPointer = "auto"; if (currentHost && window.globalActiveLocks && window.globalActiveLocks[currentHost]) { const hostLocks = window.globalActiveLocks[currentHost]; // 1. 檢查自己是否被鎖定 if (hostLocks[currentPath]) { const lockInfo = hostLocks[currentPath]; if (!(lockInfo.user_id === SESSION_USER_ID && elementId === currentEditElementId)) { isLocked = true; lockText = "🔒"; lockOpacity = "0.2"; lockPointer = "none"; lockTitle = lockInfo.user_id !== SESSION_USER_ID ? `🔒 此區塊正由 [${lockInfo.username}] 編輯中` : `🔒 您正在另一個視圖編輯此項目`; } } else { // 2. 檢查父層級是否被鎖定 (Cascading Lock) for (const [lockedPath, lockInfo] of Object.entries(hostLocks)) { if (currentPath.startsWith(lockedPath + "::")) { isLocked = true; lockText = "🔒"; lockOpacity = "0.2"; lockPointer = "none"; lockTitle = `🔒 父層級已被鎖定，無法編輯此項目`; break; } } } } let displayContent = ''; if (isVirtualIndex) { displayContent = ` ${iconHtml(svgLeaf)} ${safeValue} `; } else if (isNoCommand) { displayContent = ` ${iconHtml(svgDisabled)} no: ${safeValue} `; } else { displayContent = ` ${iconHtml(svgLeaf)} ${key}: ${safeValue} `; } htmlParts.push(`
${displayContent}
${lockText}
`); } } return htmlParts.join(''); } // ========================================== // 🌟 核心函數：生成系統設定的過濾樹狀圖 (Lazy 版) // ========================================== export function buildRealFilterTree(data, parentPath = '', hiddenKeys = [], isParentCommandGroup = false, mode = 'running') { if (typeof data !== 'object' || data === null) return ''; const htmlParts = []; htmlParts.push(`
`); for (const key in data) { const value = data[key]; const currentPath = parentPath ? `${parentPath}::${key}` : key; const isChecked = hiddenKeys.includes(currentPath) ? "checked" : ""; let cliCommand = currentPath.replace(/::/g, ' ').replace(/\s*\[\d+\]/g, ''); const svgFolderClosed = ``; const svgFolderOpen = ``; const svgGroupClosed = ``; const svgGroupOpen = ``; const svgLeaf = ``; const isFolder = typeof value === 'object' && value !== null && Object.keys(value).length > 0; const onChangeEvent = isFolder ? 'toggleChildCheckboxes(this)' : 'updateParentCheckboxState(this)'; const checkboxHtml = ``; if (isFolder) { const hasNestedObject = Object.values(value).some(v => typeof v === 'object' && v !== null); const isCommandGroup = isParentCommandGroup || !hasNestedObject; const elementId = `filter-${mode}-folder-${currentPath.replace(/[^a-zA-Z0-9]/g, '-')}`; // 🌟 存入過濾器專用快取 filterFolderCache[elementId] = { data: value, hiddenKeys: hiddenKeys }; const iconClosed = isCommandGroup ? svgGroupClosed : svgFolderClosed; const iconOpen = isCommandGroup ? svgGroupOpen : svgFolderOpen; const folderSuffix = isCommandGroup ? `(指令群組)` : ''; const expandAllIcon = ``; const collapseAllIcon = ``; const expandCollapseBtns = ` ${expandAllIcon} ${collapseAllIcon} `; // 🌟 加入 lazyLoadFolder 觸發 htmlParts.push(`
${checkboxHtml} ${iconClosed} ${key}${folderSuffix} ${expandCollapseBtns}
`); } else { const isVirtualIndex = /^\[\d+\]$/.test(key); let displayName = isVirtualIndex ? `項目 ${key}` : `${key}`; const safeValue = (value === null ? '' : value).toString().replace(/"/g, """); const isNoCommand = (key === 'no'); let valueHtml = `${safeValue}`; let colonHtml = `:`; let currentIcon = ``; if (isNoCommand) { let baseCmd = cliCommand.replace(/(^|\s)no$/, '').trim(); cliCommand = baseCmd ? `${baseCmd} no ${safeValue}` : `no ${safeValue}`; displayName = `no`; colonHtml = `:`; valueHtml = `${safeValue}`; currentIcon = ``; } htmlParts.push(`
${checkboxHtml} ${currentIcon} ${displayName}${colonHtml} ${valueHtml}
`); } } htmlParts.push('
'); return htmlParts.join(''); } ================================================================================ FILE: routers/backup.py ================================================================================ # --- routers/backup.py --- import logging import asyncssh import asyncio import re import json from fastapi import APIRouter, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import Optional, List # 引入 DB 函數 from database import ( insert_config_backup, get_config_backup_list, get_config_backup_detail, delete_config_backup ) from cmts_scraper import fetch_raw_config # 🌟 [Priority 1 修復] 引入 cmts_config_locks from shared import parse_cli_to_tree, cmts_config_locks logger = logging.getLogger(__name__) router = APIRouter( prefix="/backups", tags=["Backups"] ) # ========================================== # 📦 Pydantic Models # ========================================== class SnapshotRequest(BaseModel): host: str username: str password: str snapshot_name: str description: Optional[str] = "" # 🟢 新增這行，接收前端傳來的描述 config_type: str = "running" class RestoreRequest(BaseModel): host: str username: str password: str class ExecuteRestoreRequest(BaseModel): host: str username: str password: str commands: List[str] # 接收前端確認後的差異指令清單 # ========================================== # 🛠️ 核心演算法：深度差異比對 (Deep Diff) # ========================================== # 🌟 新增小工具：拔除 [0], [1] 虛擬後綴，還原真實 CLI def clean_virtual_key(key: str) -> str: return re.sub(r'\s*\[\d+\]$', '', key) def build_add_commands(node: dict, current_path: str) -> list: """遞迴展開所有需要新增的絕對路徑指令""" commands = [] if not node: commands.append(current_path) return commands for key, val in node.items(): if key.startswith('_'): continue # 🌟 使用乾淨的 key 來組合路徑 real_key = clean_virtual_key(key) path = f"{current_path} {real_key}".strip() if isinstance(val, dict): commands.extend(build_add_commands(val, path)) else: val_str = f" {val}" if val else "" commands.append(f"{path}{val_str}") return commands def generate_diff_commands(current_node: dict, snapshot_node: dict, current_path: str = "") -> list: """比對兩棵樹，產生帶有 no 與絕對路徑的差異指令清單""" commands = [] for key, curr_val in current_node.items(): if key.startswith('_'): continue # 🌟 使用乾淨的 key real_key = clean_virtual_key(key) path = f"{current_path} {real_key}".strip() if key not in snapshot_node: commands.append(f"no {path}") else: snap_val = snapshot_node[key] if isinstance(curr_val, dict) and isinstance(snap_val, dict): commands.extend(generate_diff_commands(curr_val, snap_val, path)) elif isinstance(curr_val, dict) or isinstance(snap_val, dict): commands.append(f"no {path}") if isinstance(snap_val, dict): commands.extend(build_add_commands(snap_val, path)) else: val_str = f" {snap_val}" if snap_val else "" commands.append(f"{path}{val_str}") elif curr_val != snap_val: val_str = f" {snap_val}" if snap_val else "" commands.append(f"{path}{val_str}") for key, snap_val in snapshot_node.items(): if key.startswith('_'): continue # 🌟 使用乾淨的 key real_key = clean_virtual_key(key) path = f"{current_path} {real_key}".strip() if key not in current_node: if isinstance(snap_val, dict): commands.extend(build_add_commands(snap_val, path)) else: val_str = f" {snap_val}" if snap_val else "" commands.append(f"{path}{val_str}") return commands # ========================================== # 🚀 API 路由 # ========================================== @router.get("/", summary="取得歷史快照列表") async def list_backups(host: str, config_type: str = "running"): records = await get_config_backup_list(host, config_type) if records is None: return {"status": "error", "message": "資料庫查詢失敗"} return {"status": "success", "data": records} @router.get("/{backup_id}", summary="取得特定快照詳細內容") async def get_backup_detail(backup_id: str): record = await get_config_backup_detail(backup_id) if not record: return {"status": "error", "message": "找不到該筆備份資料"} return {"status": "success", "data": record} @router.delete("/{backup_id}", summary="刪除特定快照") async def delete_backup(backup_id: str): success = await delete_config_backup(backup_id) if not success: return {"status": "error", "message": "刪除失敗或找不到該筆資料"} return {"status": "success", "message": "快照已成功刪除"} @router.post("/snapshot", summary="手動建立設備快照 (串流版)") async def create_snapshot(req: SnapshotRequest): async def backup_streamer(): try: # 1. 廣播：正在連線 yield json.dumps({"status": "progress", "message": f"🔌 正在建立 SSH 連線至 {req.host}..."}) + "\n" await asyncio.sleep(0.1) # 2. 廣播：正在下載 yield json.dumps({"status": "progress", "message": f"📥 正在下載 {req.config_type} 配置檔 (可能需要 30~60 秒)..."}) + "\n" raw_cli = await fetch_raw_config(req.host, req.username, req.password, req.config_type) # 3. 廣播：正在解析 yield json.dumps({"status": "progress", "message": "🧩 正在解析配置並建構樹狀圖結構..."}) + "\n" parsed_tree = await asyncio.to_thread(parse_cli_to_tree, raw_cli) # 4. 廣播：寫入資料庫 yield json.dumps({"status": "progress", "message": "💾 正在將快照寫入資料庫..."}) + "\n" backup_id = await insert_config_backup( host=req.host, config_type=req.config_type, raw_cli=raw_cli, parsed_tree=parsed_tree, snapshot_name=req.snapshot_name, description=req.description, is_auto=False ) if not backup_id: yield json.dumps({"status": "error", "message": "資料庫寫入失敗，請檢查系統日誌"}) + "\n" return # 5. 廣播：成功 yield json.dumps({ "status": "success", "message": f"快照 '{req.snapshot_name}' 建立成功！", "backup_id": backup_id, "host": req.host }) + "\n" except Exception as e: logger.error(f"❌ 建立快照失敗: {e}") yield json.dumps({"status": "error", "message": f"設備連線或解析失敗: {str(e)}"}) + "\n" return StreamingResponse(backup_streamer(), media_type="application/x-ndjson") @router.post("/{backup_id}/diff", summary="分析設備當前配置與快照的差異") async def analyze_backup_diff(backup_id: str, req: RestoreRequest): try: backup_record = await get_config_backup_detail(backup_id) if not backup_record: return {"status": "error", "message": "找不到指定的備份紀錄"} snapshot_tree = backup_record.get("parsed_tree") if not snapshot_tree: return {"status": "error", "message": "此備份紀錄缺乏樹狀結構資料，無法進行智慧比對"} # 🌟 關鍵修正：動態取得該快照的 config_type，避免蘋果比橘子 snapshot_config_type = backup_record.get("config_type", "running") # 🌟 關鍵修正：將硬編碼的 "running" 改為 snapshot_config_type raw_current = await fetch_raw_config(req.host, req.username, req.password, snapshot_config_type) if not raw_current: return {"status": "error", "message": "無法從設備取得當前配置，請檢查連線狀態"} current_tree = parse_cli_to_tree(raw_current) diff_commands = generate_diff_commands(current_tree, snapshot_tree) return { "status": "success", "data": { "commands": diff_commands } } except Exception as e: logger.error(f"差異分析失敗: {str(e)}") return {"status": "error", "message": f"分析過程發生例外錯誤: {str(e)}"} @router.post("/{backup_id}/restore", summary="執行差異還原 (串流即時回報)") async def execute_restore(backup_id: str, req: ExecuteRestoreRequest): commands = req.commands if not commands: async def empty_success(): yield json.dumps({"status": "success", "message": "沒有需要執行的指令，設備狀態已同步。"}) + "\n" return StreamingResponse(empty_success(), media_type="application/x-ndjson") async def restore_generator(): # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 host_lock = cmts_config_locks[req.host] # 1. 嘗試取得全域寫入鎖 (避免多人同時打指令) if host_lock.locked(): yield json.dumps({"status": "error", "message": "❌ 設備目前正由其他使用者進行配置，請稍後再試。"}) + "\n" return async with host_lock: try: yield json.dumps({"status": "progress", "message": "🔄 正在建立 SSH 安全連線..."}) + "\n" # 🌟 [Priority 2 提前修復] 使用 async with 確保連線與 Process 絕對會被釋放 async with asyncssh.connect(req.host, username=req.username, password=req.password, known_hosts=None) as 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() # 🌟 精準 Prompt 偵測 if prompt_pattern and re.search(prompt_pattern, output): break except asyncio.TimeoutError: break return output # 2. 進入設定模式 process.stdin.write("config\n") await process.stdin.drain() await read_until_quiet(timeout=1.5, prompt_pattern=r"\(config\)#") yield json.dumps({"status": "progress", "message": "✅ 成功進入 Global Configuration 模式"}) + "\n" # 3. 逐行寫入並進行 Fail-safe 檢查 for cmd in commands: process.stdin.write(f"{cmd}\n") await process.stdin.drain() out = await read_until_quiet(timeout=0.2, prompt_pattern=r"\(config.*\)#") clean_out = re.sub(r'\x1b\[[0-9;]*[mGK]', '', out).strip() # 🚨 Fail-safe 攔截機制與安全撤銷 (Rollback) if "% Invalid" in clean_out or "% Incomplete" in clean_out or "% Ambiguous" in clean_out: # 1. 先通知前端正在撤銷 yield json.dumps({ "status": "progress", "message": "⚠️ 偵測到無效指令，正在執行 abort 放棄所有變更..." }) + "\n" # 2. 對設備下達 abort 指令 process.stdin.write("abort\n") await process.stdin.drain() await read_until_quiet(timeout=1.0, prompt_pattern=r"(?:#|>)") # 等待設備處理 abort 並退出 config 模式 # 3. 回報最終錯誤並關閉連線 yield json.dumps({ "status": "error", "message": f"❌ 寫入中斷且已安全撤銷！失敗指令: {cmd}", "output": clean_out }) + "\n" return yield json.dumps({ "status": "progress", "message": f"執行: {cmd}", "output": clean_out }) + "\n" # 稍微讓出控制權，確保串流順暢 await asyncio.sleep(0.05) # 4. 提交變更 (Commit) yield json.dumps({"status": "progress", "message": "⏳ 正在寫入 Commit 保存設定..."}) + "\n" process.stdin.write("commit\n") await process.stdin.drain() commit_out = await read_until_quiet(timeout=6.0, prompt_pattern=r"\(config\)#") # 5. 退出並關閉連線 process.stdin.write("exit\n") await process.stdin.drain() # 🌟 移除手動 conn.close()，交由 async with 處理 yield json.dumps({ "status": "success", "message": "✅ 所有差異還原指令已成功送出並 Commit！", "output": clean_commit }) + "\n" except Exception as e: logger.error(f"還原執行失敗: {str(e)}") yield json.dumps({"status": "error", "message": f"SSH 連線或執行發生例外錯誤: {str(e)}"}) + "\n" return StreamingResponse(restore_generator(), media_type="application/x-ndjson") ================================================================================ FILE: routers/terminal.py ================================================================================ import asyncio import asyncssh import traceback from fastapi import APIRouter, WebSocket, WebSocketDisconnect router = APIRouter() @router.websocket("/ws/terminal") async def websocket_terminal(websocket: WebSocket, host: str, username: str, password: str): await websocket.accept() conn = None process = None ws_task = None ssh_task = None try: # 建立連線 conn = await asyncssh.connect(host, username=username, password=password, known_hosts=None) # 💡 關鍵修復：使用 term_size 參數來指定寬高 (width, height) process = await conn.create_process( term_type='xterm-256color', term_size=(80, 24), # 正確的 asyncssh 參數格式 encoding=None ) async def forward_to_ws(): try: while True: data_bytes = await process.stdout.read(8192) if not data_bytes: print("[DEBUG] 設備端主動關閉了 stdout 通道") break safe_text = data_bytes.decode('utf-8', errors='replace') await websocket.send_text(safe_text) except asyncio.CancelledError: pass except Exception as e: print("\n❌ [WS Forward Error] 讀取設備畫面時發生錯誤:") traceback.print_exc() async def forward_to_ssh(): try: while True: data_text = await websocket.receive_text() data_bytes = data_text.encode('utf-8') # 攔截 xterm.js 預設的 DEL (\x7f)，轉換為 BS (\x08) if b'\x7f' in data_bytes: data_bytes = data_bytes.replace(b'\x7f', b'\x08') process.stdin.write(data_bytes) await process.stdin.drain() except WebSocketDisconnect: # 主動拋出，讓外層捕捉以進行資源回收 raise except asyncio.CancelledError: pass except Exception as e: print("\n❌ [SSH Forward Error] 寫入指令到設備時發生錯誤:") traceback.print_exc() ws_task = asyncio.create_task(forward_to_ws()) ssh_task = asyncio.create_task(forward_to_ssh()) done, pending = await asyncio.wait( [ws_task, ssh_task], return_when=asyncio.FIRST_COMPLETED ) for task in pending: task.cancel() except WebSocketDisconnect: print("[DEBUG] WebSocket 正常斷線，正在終止背景任務...") if ws_task and not ws_task.done(): ws_task.cancel() if ssh_task and not ssh_task.done(): ssh_task.cancel() except Exception as e: error_msg = f"\r\n\x1b[31mSSH Connection Error: {str(e)}\x1b[0m\r\n" try: await websocket.send_text(error_msg) except: pass print("\n❌ [Connection Error] 建立連線或執行過程中發生錯誤:") traceback.print_exc() finally: # 🌟 確保 process 與 conn 絕對被關閉，防止 Memory Leak if process: process.close() if conn: conn.close() try: await websocket.close() except: pass ================================================================================ FILE: routers/lock.py ================================================================================ # --- routers/lock.py --- import time from fastapi import APIRouter, HTTPException, Query from pydantic import BaseModel from typing import Dict, Optional router = APIRouter(prefix="/locks", tags=["Lock Management"]) # ========================================== # 💡 全域鎖定表 (In-Memory Lock Table) # 🌟 結構升級: { "host@@path": {"user_id": "...", "username": "...", "expires_at": ...} } # ⚠️ 嚴格規則：Lock Key 絕對不包含 config_type，確保 running 與 full 視圖共用同一把鎖！ # ========================================== ACTIVE_LOCKS: Dict[str, dict] = {} LOCK_TIMEOUT = 120 # 🌟 請求模型嚴格排除 config_type class LockAcquireReq(BaseModel): host: str path: str user_id: str username: str class LockActionReq(BaseModel): host: str path: str user_id: str def clean_expired_locks(): current_time = time.time() expired_keys = [k for k, lock in ACTIVE_LOCKS.items() if lock["expires_at"] < current_time] for k in expired_keys: del ACTIVE_LOCKS[k] def is_path_locked_by_others(target_host: str, target_path: str, user_id: str) -> tuple[Optional[dict], str]: clean_expired_locks() prefix = f"{target_host}@@" for locked_key, lock_info in ACTIVE_LOCKS.items(): # 🌟 只檢查同一台設備 (IP) 的鎖定 if not locked_key.startswith(prefix): continue locked_path = locked_key.split("@@", 1)[1] if lock_info["user_id"] == user_id: continue if target_path == locked_path: return lock_info, f"此區塊正由 [{lock_info['username']}] 編輯中" if target_path.startswith(locked_path + "::"): return lock_info, f"父層級 [{locked_path}] 已被 [{lock_info['username']}] 鎖定，無法編輯子區塊" if locked_path.startswith(target_path + "::"): return lock_info, f"子層級 [{locked_path}] 已被 [{lock_info['username']}] 鎖定，無法鎖定整個父區塊" return None, "" @router.post("/acquire") async def acquire_lock(req: LockAcquireReq): conflict_lock, error_msg = is_path_locked_by_others(req.host, req.path, req.user_id) if conflict_lock: raise HTTPException(status_code=409, detail=error_msg) # ⚠️ 嚴格綁定 host 與 path，跨視圖共用 lock_key = f"{req.host}@@{req.path}" ACTIVE_LOCKS[lock_key] = { "user_id": req.user_id, "username": req.username, "expires_at": time.time() + LOCK_TIMEOUT } return {"status": "success", "message": f"成功鎖定路徑: {req.path}"} @router.post("/heartbeat") async def heartbeat_lock(req: LockActionReq): clean_expired_locks() lock_key = f"{req.host}@@{req.path}" if lock_key not in ACTIVE_LOCKS: raise HTTPException(status_code=404, detail="鎖定已失效，請重新獲取") if ACTIVE_LOCKS[lock_key]["user_id"] != req.user_id: raise HTTPException(status_code=403, detail="無權更新他人的鎖定") ACTIVE_LOCKS[lock_key]["expires_at"] = time.time() + LOCK_TIMEOUT return {"status": "success", "message": "心跳更新成功"} @router.post("/release") async def release_lock(req: LockActionReq): lock_key = f"{req.host}@@{req.path}" if lock_key in ACTIVE_LOCKS and ACTIVE_LOCKS[lock_key]["user_id"] == req.user_id: del ACTIVE_LOCKS[lock_key] return {"status": "success", "message": "鎖定已釋放"} @router.get("/status") async def get_lock_status(host: str = Query(None)): """獲取特定設備的鎖定狀態""" clean_expired_locks() if host: prefix = f"{host}@@" # 🌟 貼心設計：拔除 host@@ 前綴，讓前端收到的依然是乾淨的 path，UI 不用改！ filtered_locks = {k.split("@@", 1)[1]: v for k, v in ACTIVE_LOCKS.items() if k.startswith(prefix)} return {"status": "success", "data": filtered_locks} return {"status": "success", "data": {}} ================================================================================ FILE: routers/query.py ================================================================================ # --- routers/query.py --- import re from fastapi import APIRouter, HTTPException, Query from netmiko import ConnectHandler from shared import CMTS_DEVICE router = APIRouter() def parse_cm_output(raw_text: str) -> list: parsed_data = [] lines = raw_text.splitlines() data_started = False for line in lines: if not line.strip(): continue if line.strip().startswith('---'): data_started = True continue if line.strip().startswith('==='): break if data_started: columns = re.split(r'\s{2,}', line.strip()) if len(columns) >= 8: cm_info = { "downstream": columns[0], "upstream": columns[1], "bond_cap": columns[2], "ofdm_cap": columns[3], "mac_address": columns[4], "ip_address": columns[5], "num_cpe": int(columns[6]), "state": columns[7] } parsed_data.append(cm_info) return parsed_data @router.get("/cable-modems") async def get_cable_modems(): net_connect = None try: net_connect = ConnectHandler(**CMTS_DEVICE) raw_output = str(net_connect.send_command("show cable modem")) structured_data = parse_cm_output(raw_output) return {"status": "success", "total_count": len(structured_data), "data": structured_data} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Connection Error: {str(e)}") finally: if net_connect: net_connect.disconnect() @router.get("/configuration") async def get_configuration(): net_connect = None try: net_connect = ConnectHandler(**CMTS_DEVICE) net_connect.send_command_timing("config") raw_output = net_connect.send_command("show full-configuration | nomore", expect_string=r"#", read_timeout=120) net_connect.send_command_timing("exit") return {"status": "success", "data": raw_output} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Error: {str(e)}") finally: if net_connect: net_connect.disconnect() @router.get("/cmts-query") async def get_cmts_query( query_type: str = Query(..., description="查詢動作類型"), target: str = Query("", description="MAC Address 或 VC:VS (選填)"), host: str = Query(...), username: str = Query(...), password: str = Query(...) ): net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) target_str = f" {target.strip()}" if target.strip() else "" commands = { "base": f"show cable modem{target_str}", "cpe": f"show cable modem{target_str} cpe", "cpe_dhcp": f"show cable modem{target_str} cpe dhcp", "cpe_ipv6": f"show cable modem{target_str} cpe ipv6", "bonding": f"show cable modem{target_str} bonding", "bonding_ds": f"show cable modem{target_str} bonding downstream", "bonding_us": f"show cable modem{target_str} bonding upstream", "phy": f"show cable modem{target_str} phy", "verbose": f"show cable modem{target_str} verbose", "ofdm_profile": f"show cable modem{target_str} ofdm-profile", "ofdma_profile": f"show cable modem{target_str} ofdma-profile", "dhcp_verbose": f"show cable modem{target_str} dhcp verbose", "service_flow": f"show cable modem{target_str} service-flow", "service_flow_verbose": f"show cable modem{target_str} service-flow verbose", "qos": f"show cable modem{target_str} qos", "uptime": f"show cable modem{target_str} uptime", "ugs": f"show cable modem{target_str} ugs", "cm_status": f"show cable modem{target_str} cm-status", "partial_mode": "show cable modem partial-mode", "hop": "show cable hop", "flap_list": "show cable flap-list", "flap_sum": "show cable flap-sum", "rpd_base": f"show cable rpd{target_str}", "rpd_verbose": f"show cable rpd{target_str} verbose", "rpd_ptp_time": f"show cable rpd{target_str} ptp time-property", "rpd_ptp_verbose": f"show cable rpd{target_str} ptp verbose", "rpd_counters_map": f"show cable rpd{target_str} counters map", "rpd_capabilities": f"show cable rpd{target_str} capabilities", "rpd_video_counters": f"show cable rpd{target_str} video-channel counters", "rpd_env_temp": f"show cable rpd{target_str} environment temperature", "rpd_env_volt": f"show cable rpd{target_str} environment voltage", "rpd_session": f"show cable rpd{target_str} session", "rpd_reset_history": f"show cable rpd{target_str} reset-history", "rpd_port_transceiver": f"show cable rpd{target_str} port-transceiver", } if query_type not in commands: raise ValueError(f"未知的查詢類型: {query_type}") cli_command = commands[query_type] + " | nomore" net_connect = ConnectHandler(**device) raw_output = net_connect.send_command(cli_command, read_timeout=15) return {"status": "success", "command": cli_command, "data": raw_output} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Query Error: {str(e)}") finally: if net_connect: net_connect.disconnect() @router.get("/cmts-mac-domain-config") async def get_mac_domain_config(target: str, host: str, username: str, password: str): net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) cli_command = f"show running-config cable mac-domain {target.strip()} | nomore" net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command(cli_command, read_timeout=15)) # 💡 正名工程：更新字典的 Key 為一致性的命名 config = { "common_settings": { "ip-provisioning-mode": "dual-stack", "diplexer-band-edge": "enabled", "cm-battery-mode-31-support": "disabled", "cm-battery-mode-30-support": "disabled", "docsis40": "disabled", "ds-dynamic-bonding-group": "disabled", "us-dynamic-bonding-group": "disabled" }, "basic_channel_sets": { "admin-state": "down", "ds-primary-set": "", "ds-non-primary-set": "", "us-phy-channel-set": "", "ds-ofdm-set": "", "us-ofdma-set": "" }, "static_ds_bonding_groups": {}, "static_us_bonding_groups": {} } current_group_type = None current_group_name = None for line in raw_output.splitlines(): line = line.strip() if not line or line.startswith("cable mac-domain"): continue if line.startswith("ds-bonding-group"): current_group_type = "static_ds_bonding_groups" current_group_name = line.split()[1] config["static_ds_bonding_groups"][current_group_name] = {"admin-state": "down"} continue elif line.startswith("us-bonding-group"): current_group_type = "static_us_bonding_groups" current_group_name = line.split()[1] config["static_us_bonding_groups"][current_group_name] = {"admin-state": "down"} continue elif line == "!": current_group_type = None current_group_name = None continue if current_group_type and current_group_name: parts = line.split(maxsplit=1) if len(parts) == 2: config[current_group_type][current_group_name][parts[0]] = parts[1] else: # 💡 正名工程：將解析結果存入對應的新 Key 中 if line.startswith("ip-provisioning-mode"): config["common_settings"]["ip-provisioning-mode"] = line.split(maxsplit=1)[1] elif line.startswith("diplexer-band-edge control"): config["common_settings"]["diplexer-band-edge"] = line.split(maxsplit=2)[2] elif line.startswith("cm-battery-mode-31-support"): config["common_settings"]["cm-battery-mode-31-support"] = line.split(maxsplit=1)[1] elif line.startswith("cm-battery-mode-30-support"): config["common_settings"]["cm-battery-mode-30-support"] = line.split(maxsplit=1)[1] elif line.startswith("docsis40"): config["common_settings"]["docsis40"] = line.split(maxsplit=1)[1] elif line.startswith("ds-dynamic-bonding-group"): config["common_settings"]["ds-dynamic-bonding-group"] = line.split(maxsplit=1)[1] elif line.startswith("us-dynamic-bonding-group"): config["common_settings"]["us-dynamic-bonding-group"] = line.split(maxsplit=1)[1] # 💡 正名工程：將原本的 dynamic_bonding_groups 改為 basic_channel_sets elif line.startswith("admin-state"): config["basic_channel_sets"]["admin-state"] = line.split(maxsplit=1)[1] elif line.startswith("ds-primary-set"): config["basic_channel_sets"]["ds-primary-set"] = line.split(maxsplit=1)[1] elif line.startswith("ds-non-primary-set"): config["basic_channel_sets"]["ds-non-primary-set"] = line.split(maxsplit=1)[1] elif line.startswith("us-phy-channel-set"): config["basic_channel_sets"]["us-phy-channel-set"] = line.split(maxsplit=1)[1] elif line.startswith("ds-ofdm-set"): config["basic_channel_sets"]["ds-ofdm-set"] = line.split(maxsplit=1)[1] elif line.startswith("us-ofdma-set"): config["basic_channel_sets"]["us-ofdma-set"] = line.split(maxsplit=1)[1] return {"status": "success", "data": config} except Exception as e: raise HTTPException(status_code=500, detail=f"Parse Error: {str(e)}") finally: if net_connect: net_connect.disconnect() @router.get("/cmts-mac-domain-list") async def get_cmts_mac_domain_list( host: str = Query(...), username: str = Query(...), password: str = Query(...) ): net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command_timing("show running-config cable mac-domain ?")) import re matches = re.findall(r'\b\d+:\d+/\d+\.\d+\b', raw_output) mac_domains = list(dict.fromkeys(matches)) return {"status": "success", "data": mac_domains} except Exception as e: return {"status": "error", "message": str(e)} finally: if net_connect: net_connect.disconnect() @router.get("/cmts-version") async def get_cmts_version( host: str = Query(...), username: str = Query(...), password: str = Query(...) ): net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command("show version", read_timeout=15)) import re # 🌟 使用 Regex 尋找 infra 或 vcmts-cd-0 後面的版本號 match = re.search(r"(?:infra|vcmts-cd-0)\s+([\w\.\-]+)", raw_output) if match: return {"status": "success", "version": match.group(1)} else: return {"status": "error", "message": "無法解析版本號", "raw_output": raw_output} except Exception as e: return {"status": "error", "message": f"CMTS Connection Error: {str(e)}"} finally: if net_connect: net_connect.disconnect() ================================================================================ FILE: routers/config.py ================================================================================ # --- routers/config.py --- import re import json import os import database import asyncio import asyncssh from fastapi import APIRouter, HTTPException from pydantic import BaseModel from typing import List from netmiko import ConnectHandler # 🌟 [Priority 1 修復] 引入 cmts_config_locks from shared import CMTS_DEVICE, cmts_config_locks, parse_cli_to_tree, deep_split_tree, USE_DB from collections import defaultdict from fastapi.responses import StreamingResponse router = APIRouter() # ========================================== # 🌟 雙軌過濾器檔案讀寫輔助函數 (加入高可用性 Fallback) # ========================================== def get_filter_file_path(config_type: str) -> str: # 確保檔名安全，只允許 'running' 或 'full' safe_type = "full" if config_type == "full" else "running" return f"filters_{safe_type}.json" async def load_tree_filters(config_type: str) -> list: if USE_DB: try: db_filters = await database.get_tree_filters(config_type) if db_filters is not None: return db_filters else: print("⚠️ [Fallback] 資料庫無法取得過濾器，自動切換至 JSON 快取讀取...") except Exception as e: print(f"⚠️ [Fallback] 資料庫讀取過濾器發生例外: {e}，自動切換至 JSON 快取讀取...") file_path = get_filter_file_path(config_type) if os.path.exists(file_path): try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) return data.get("hidden_keys", []) except Exception as e: print(f"讀取過濾器檔案失敗: {e}") return [] return [] async def save_tree_filters(config_type: str, hidden_keys: list): db_success = False if USE_DB: try: if await database.upsert_tree_filters(config_type, hidden_keys): db_success = True else: print("⚠️ [Fallback] 資料庫儲存過濾器失敗，自動切換至 JSON 快取寫入...") except Exception as e: print(f"⚠️ [Fallback] 資料庫寫入過濾器發生例外: {e}，自動切換至 JSON 快取寫入...") if not db_success: file_path = get_filter_file_path(config_type) try: with open(file_path, 'w', encoding='utf-8') as f: json.dump({"hidden_keys": hidden_keys}, f, ensure_ascii=False, indent=4) except Exception as e: print(f"儲存過濾器檔案失敗: {e}") # ========================================== class ConfigRequest(BaseModel): script: str host: str username: str password: str # 定義前端傳來的 Diff 資料結構 class DiffItem(BaseModel): path: str # 例如: "cable.ds-rf-port.1:9/0.down-channel.0.frequency" old_val: str new_val: str class GenerateCliRequest(BaseModel): diffs: List[DiffItem] interfaces_with_admin_state: List[str] = [] @router.post("/cmts-config") async def execute_cmts_config(req: ConfigRequest): # 過濾掉空行與註解 (! 開頭的行) commands = [cmd.strip() for cmd in req.script.splitlines() if cmd.strip() and not cmd.strip().startswith('!')] async def config_streamer(): # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 host_lock = cmts_config_locks[req.host] async with host_lock: try: yield json.dumps({"status": "progress", "message": f"🔌 準備連線至設備 {req.host}..."}) + "\n" # 使用 asyncssh 建立非同步連線 async with asyncssh.connect( req.host, username=req.username, password=req.password, known_hosts=None ) as conn: async with conn.create_process() as process: # 🌟 建立安全的非同步讀取函數 (讀到安靜為止，避免卡死) async def read_until_quiet(timeout=0.5, prompt_pattern: str = None): out_data = "" while True: try: # 利用 wait_for 設定超時，時間內沒資料就當作設備吐完了 chunk = await asyncio.wait_for(process.stdout.read(4096), timeout=timeout) if not chunk: break out_data += str(chunk) # 🌟 精準 Prompt 偵測 if prompt_pattern and re.search(prompt_pattern, out_data): break except asyncio.TimeoutError: break return out_data # 1. 進入設定模式 process.stdin.write("config\n") await process.stdin.drain() await read_until_quiet(0.5, prompt_pattern=r"\(config\)#") # 清空歡迎訊息 # 2. 逐行送出指令並回報進度 for cmd in commands: yield json.dumps({"status": "progress", "message": f"▶️ 執行: {cmd}"}) + "\n" process.stdin.write(cmd + "\n") await process.stdin.drain() # ⏱️ 智慧等待：如果是 commit 指令，給予 5 秒讓設備存檔；其他指令等 0.5 秒 wait_time = 5.0 if cmd.strip() == "commit" else 0.5 output = await read_until_quiet(wait_time, prompt_pattern=r"\(config.*\)#") # 🛡️ 防呆撤銷機制：偵測到錯誤立刻 abort lower_output = output.lower() if "% invalid" in lower_output or "% incomplete" in lower_output or "error" in lower_output or "aborted" in lower_output: yield json.dumps({ "status": "error", "message": f"❌ 設備拒絕指令: {cmd}", "output": output.strip() }) + "\n" process.stdin.write("abort\n") await process.stdin.drain() return # 發生錯誤，提早結束串流 # 3. 退出設定模式 process.stdin.write("exit\n") await process.stdin.drain() final_output = await read_until_quiet(1.0, prompt_pattern=r"(?:#|>)") yield json.dumps({ "status": "success", "message": "✅ 所有配置已成功寫入並 Commit！", "output": final_output.strip() }) + "\n" except Exception as e: yield json.dumps({"status": "error", "message": f"SSH 執行發生例外錯誤: {str(e)}"}) + "\n" # 回傳 NDJSON 串流格式 return StreamingResponse(config_streamer(), media_type="application/x-ndjson") @router.post("/cmts-generate-cli") async def generate_cli(req: GenerateCliRequest): """ 將前端的 JSON Diff 轉譯為 Harmonic 單行 CLI 腳本 並自動處理 admin-state 的安全生命週期 (先 down 後 up) """ interface_groups = defaultdict(list) interface_admin_states = {} for diff in req.diffs: parts = diff.path.split('::') param_name = parts[-1] interface_path = " ".join(parts[:-1]) # 全域清除路徑中間可能夾帶的虛擬資料夾索引 interface_path = re.sub(r"\s*\[\d+\]", "", interface_path) if param_name == "admin-state": interface_admin_states[interface_path] = diff.new_val else: interface_groups[interface_path].append((param_name, diff.old_val, diff.new_val)) cli_lines = [] cli_lines.append("! --- Auto-Generated Configuration Script ---") for interface, changes in interface_groups.items(): cli_lines.append(f"! Configuring: {interface}") supports_admin_state = interface in req.interfaces_with_admin_state if supports_admin_state: cli_lines.append(f"{interface} admin-state down") for param, old_val, new_val in changes: if re.match(r"^\[\d+\]$", param): if new_val == "": if old_val: cli_lines.append(f"no {interface} {old_val}") else: if old_val and old_val != new_val: cli_lines.append(f"no {interface} {old_val}") cli_lines.append(f"{interface} {new_val}") else: if new_val == "": cli_lines.append(f"no {interface} {param}") else: cli_lines.append(f"{interface} {param} {new_val}") if supports_admin_state: final_state = interface_admin_states.get(interface, "up") cli_lines.append(f"{interface} admin-state {final_state}") cli_lines.append("commit") cli_lines.append("!") for interface, state in interface_admin_states.items(): if interface not in interface_groups: cli_lines.append(f"! Configuring state only: {interface}") cli_lines.append(f"{interface} admin-state {state}") cli_lines.append("commit") cli_lines.append("!") final_script = "\n".join(cli_lines) return {"status": "success", "data": final_script} # ========================================== # DS RF Port 動態樹狀查詢 API # ========================================== @router.get("/cmts-ds-rf-port-list") async def get_ds_rf_port_list(host: str, username: str, password: str = ""): """獲取設備上所有的 DS RF Port 清單""" net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) command = 'show running-config | include "cable ds-rf-port"' raw_output = str(net_connect.send_command(command, read_timeout=60)) pattern = r"cable ds-rf-port\s+([0-9:/]+)" ports = re.findall(pattern, raw_output) unique_ports = sorted(list(set(ports))) return { "status": "success", "data": unique_ports, "raw_output": raw_output } except Exception as e: return {"status": "error", "message": str(e)} finally: if net_connect: net_connect.disconnect() @router.get("/cmts-ds-rf-port-config") async def get_ds_rf_port_config(target: str, host: str, username: str, password: str = ""): """獲取特定 DS RF Port 的配置，並轉換為樹狀 JSON""" net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) command = f"show running-config interface cable ds-rf-port {target} | nomore" raw_output = str(net_connect.send_command(command, read_timeout=60)) # 🌟 [Priority 1 修復] 將 CPU-Bound 任務移至 ThreadPool base_tree = await asyncio.to_thread(parse_cli_to_tree, raw_output) perfect_tree = await asyncio.to_thread(deep_split_tree, base_tree) return { "status": "success", "data": perfect_tree, "raw_cli": raw_output.strip() } except Exception as e: return {"status": "error", "message": str(e)} finally: if net_connect: net_connect.disconnect() @router.get("/cmts-full-config") async def get_full_config(host: str, username: str, password: str = "", skip_filter: bool = False, config_type: str = "running"): """獲取整台 CMTS 的完整配置，並轉換為大型樹狀 JSON""" net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) if config_type == "full": net_connect.send_command_timing("config") command = "show full-configuration | nomore" raw_output = str(net_connect.send_command(command, read_timeout=180)) net_connect.send_command_timing("exit") else: command = "show running-config | nomore" raw_output = str(net_connect.send_command(command, read_timeout=180)) clean_output = re.sub(r"^(?:Building configuration\.\.\.|Current configuration.*?)\n+", "", raw_output, flags=re.IGNORECASE | re.MULTILINE) clean_output = clean_output.lstrip() # 🌟 [Priority 1 修復] 將 CPU-Bound 任務移至 ThreadPool base_tree = await asyncio.to_thread(parse_cli_to_tree, clean_output) perfect_tree = await asyncio.to_thread(deep_split_tree, base_tree) # ========================================== # 🌟 深度過濾攔截器：改用雙軌過濾器讀取邏輯 # ========================================== if not skip_filter: hidden_keys = await load_tree_filters(config_type) for path in hidden_keys: keys = path.split('::') current = perfect_tree # 走到倒數第二層 for k in keys[:-1]: if isinstance(current, dict) and k in current: current = current[k] else: current = None break # 刪除最後一層的目標節點 if isinstance(current, dict) and keys[-1] in current: current.pop(keys[-1], None) # ========================================== return {"status": "success", "data": perfect_tree} except Exception as e: return {"status": "error", "message": str(e)} finally: if net_connect: net_connect.disconnect() @router.get("/cmts-full-config/stream") async def get_full_config_stream(host: str, username: str, password: str = "", skip_filter: bool = False, config_type: str = "running"): """獲取整台 CMTS 的完整配置 (串流進度回報版)""" async def config_streamer(): net_connect = None try: # 1. 廣播：正在連線 yield json.dumps({"status": "progress", "message": f"🔌 正在建立 SSH 連線至 {host}..."}) + "\n" await asyncio.sleep(0.1) # 讓前端有時間渲染 device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) # 使用 asyncio.to_thread 將阻塞的 netmiko 連線放到背景執行，避免卡住其他非同步任務 net_connect = await asyncio.to_thread(ConnectHandler, **device) # 2. 廣播：正在下載 yield json.dumps({"status": "progress", "message": f"📥 正在下載 {config_type} 配置檔 (可能需要 30~60 秒)..."}) + "\n" if config_type == "full": await asyncio.to_thread(net_connect.send_command_timing, "config") command = "show full-configuration | nomore" raw_output = await asyncio.to_thread(net_connect.send_command, command, read_timeout=180) await asyncio.to_thread(net_connect.send_command_timing, "exit") else: command = "show running-config | nomore" raw_output = await asyncio.to_thread(net_connect.send_command, command, read_timeout=180) # 3. 廣播：正在解析 yield json.dumps({"status": "progress", "message": "🧩 正在解析配置並建構樹狀圖結構..."}) + "\n" await asyncio.sleep(0.1) clean_output = re.sub(r"^(?:Building configuration\.\.\.|Current configuration.*?)\n+", "", raw_output, flags=re.IGNORECASE | re.MULTILINE) clean_output = clean_output.lstrip() # 🌟 [Priority 1 修復] 將 CPU-Bound 任務移至 ThreadPool base_tree = await asyncio.to_thread(parse_cli_to_tree, clean_output) perfect_tree = await asyncio.to_thread(deep_split_tree, base_tree) # ========================================== # 🌟 深度過濾攔截器 (維持原樣) # ========================================== if not skip_filter: yield json.dumps({"status": "progress", "message": "🔍 正在套用系統過濾器規則..."}) + "\n" await asyncio.sleep(0.1) hidden_keys = await load_tree_filters(config_type) for path in hidden_keys: keys = path.split('::') current = perfect_tree for k in keys[:-1]: if isinstance(current, dict) and k in current: current = current[k] else: current = None break if isinstance(current, dict) and keys[-1] in current: current.pop(keys[-1], None) # ========================================== # 4. 廣播：完成並傳送最終資料 yield json.dumps({"status": "success", "data": perfect_tree}) + "\n" except Exception as e: yield json.dumps({"status": "error", "message": f"載入失敗: {str(e)}"}) + "\n" finally: # 🌟 [Priority 1 修復] 確保背景連線被正確釋放 if net_connect: await asyncio.to_thread(net_connect.disconnect) # 回傳 NDJSON 串流格式 return StreamingResponse(config_streamer(), media_type="application/x-ndjson") # ========================================== # 🌟 系統設定 API (System Settings) # ========================================== class SettingsRequest(BaseModel): hidden_keys: list[str] config_type: str = "running" # 🌟 加入 config_type 參數 @router.get("/settings/tree-filters") async def get_tree_filters(config_type: str = "running"): """獲取目前的樹狀圖隱藏名單""" # 🌟 根據 config_type 讀取對應的 JSON hidden_keys = await load_tree_filters(config_type) return {"status": "success", "data": hidden_keys} @router.post("/settings/tree-filters") async def update_tree_filters(req: SettingsRequest): """更新樹狀圖隱藏名單並存檔""" # 🌟 根據 config_type 寫入對應的 JSON await save_tree_filters(req.config_type, req.hidden_keys) return {"status": "success", "message": f"系統設定 ({req.config_type}) 已更新"} ================================================================================ FILE: routers/leaf_options.py ================================================================================ # --- routers/leaf_options.py --- from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, Body from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import List import json, os, asyncio, re import database from cmts_scraper import sync_cmts_leaves_async from shared import CMTS_DEVICE, USE_DB router = APIRouter(tags=["Options"]) # 🌟 1. 動態獲取快取檔名 (加入 host 隔離) def get_cache_file(host: str, config_type: str): safe_host = host.replace(".", "_") return f"{safe_host}_{config_type}_cache.json" # ========================================== # 🌟 2. 全域廣播機制 (SSE) - 升級為「IP + 模式」頻道分流 # ========================================== # 將狀態改為空字典，動態根據 host_configType 建立 SCAN_STATUS = {} active_clients = {} async def broadcast_message(message_dict: dict, host: str, config_type: str = "running"): """將訊息推播給特定頻道的連線前端""" channel_key = f"{host}_{config_type}" dead_clients = set() # SSE 協定嚴格要求必須以 "data: " 開頭，並以 "\n\n" 結尾 message_str = f"data: {json.dumps(message_dict)}\n\n" for client_queue in active_clients.get(channel_key, set()): try: await client_queue.put(message_str) except Exception: dead_clients.add(client_queue) for dead in dead_clients: active_clients[channel_key].discard(dead) @router.get("/cmts-leaf-options/stream") async def sse_stream(request: Request, host: str, config_type: str = "running"): """前端一載入就會連上這個路由，持續監聽特定頻道的廣播""" channel_key = f"{host}_{config_type}" client_queue = asyncio.Queue() if channel_key not in active_clients: active_clients[channel_key] = set() active_clients[channel_key].add(client_queue) async def event_generator(): try: while True: # 主動檢查斷線，避免死鎖 if await request.is_disconnected(): break try: # [修復 Leak] 延長 timeout 降低輪詢負擔 message = await asyncio.wait_for(client_queue.get(), timeout=5.0) yield message except asyncio.TimeoutError: yield ": keepalive\n\n" except asyncio.CancelledError: pass finally: # [修復 Leak] 確保斷線時 Queue 絕對會被移出 active_clients 釋放記憶體 if channel_key in active_clients: active_clients[channel_key].discard(client_queue) return StreamingResponse(event_generator(), media_type="text/event-stream") @router.get("/scan-status") async def get_scan_status(host: str, config_type: str = "running"): channel_key = f"{host}_{config_type}" return {"is_scanning": SCAN_STATUS.get(channel_key, False)} # ========================================== # 🌟 3. 快取讀取與背景掃描任務 # ========================================== class SyncOptionsRequest(BaseModel): host: str # 🌟 新增 host 參數 username: str = "" # 🌟 新增 username 參數 password: str = "" # 🌟 新增 password 參數 leaf_paths: List[str] config_type: str = "running" @router.get("/cmts-leaf-options") async def get_leaf_options(host: str, config_type: str = "running"): if USE_DB: try: db_options = await database.get_all_leaf_options(host, config_type) if db_options is not None: # 取得 metadata metadata = await database.get_device_status(host, config_type) if metadata: db_options["__metadata__"] = metadata return db_options else: # 若回傳 None 表示 DB 連線異常，執行 Fallback print("⚠️ [Fallback] 資料庫無法取得資料，自動切換至 JSON 快取讀取...") except Exception as e: print(f"⚠️ [Fallback] 資料庫讀取發生例外: {e}，自動切換至 JSON 快取讀取...") cache_file = get_cache_file(host, config_type) if not os.path.exists(cache_file): return {} # 🌟 優化 3：將讀取動作丟到背景執行緒 def read_json_from_file(filepath): with open(filepath, "r", encoding="utf-8") as f: return json.load(f) return await asyncio.to_thread(read_json_from_file, cache_file) async def run_scan_task(host: str, username: str, password: str, paths: List[str], config_type: str): """在背景執行的爬蟲任務，並將進度廣播出去""" global SCAN_STATUS channel_key = f"{host}_{config_type}" try: await broadcast_message({"event": "scan_start"}, host, config_type) # 呼叫爬蟲 (帶入設備連線資訊與 config_type) async for chunk in sync_cmts_leaves_async( host=host, username=username, password=password, leaf_paths=paths, config_type=config_type ): if isinstance(chunk, str): await broadcast_message(json.loads(chunk), host, config_type) else: await broadcast_message(chunk, host, config_type) await broadcast_message({"event": "done"}, host, config_type) except Exception as e: await broadcast_message({"event": "error", "message": str(e)}, host, config_type) finally: # 解除特定頻道的鎖定 SCAN_STATUS[channel_key] = False @router.post("/cmts-leaf-options/sync") async def sync_leaf_options(request: SyncOptionsRequest, background_tasks: BackgroundTasks): global SCAN_STATUS if not request.leaf_paths: raise HTTPException(status_code=400) channel_key = f"{request.host}_{request.config_type}" if SCAN_STATUS.get(channel_key, False): return {"status": "busy", "message": f"⚠️ {request.config_type} 模式的掃描任務正在執行中，請勿重複點擊！"} SCAN_STATUS[channel_key] = True # 保留您原本完美的路徑清理邏輯 cmts_query_paths = [] for p in request.leaf_paths: clean_p = p.replace("::", " ") clean_p = re.sub(r"\s*\[\d+\]", "", clean_p) cmts_query_paths.append(clean_p) cmts_query_paths = list(dict.fromkeys(cmts_query_paths)) # 決定帳密 (優先使用 request 傳來的，否則 fallback 到 shared.CMTS_DEVICE) # 🌟 加上 or "" 確保最終結果絕對是字串，消除 Pylance 警告 user = request.username or CMTS_DEVICE.get("username") or "" pwd = request.password or CMTS_DEVICE.get("password") or "" background_tasks.add_task(run_scan_task, request.host, user, pwd, cmts_query_paths, request.config_type) return {"status": "started"} # ========================================== # 🌟 4. 清除快取 API # ========================================== @router.post("/clear_cache") async def clear_specific_cache(host: str, config_type: str = "running", paths_to_clear: list = Body(...)): cleared_count = 0 # 1. 嘗試清除 DB if USE_DB: try: db_cleared = await database.delete_leaf_options(host, config_type, paths_to_clear) if db_cleared >= 0: cleared_count = db_cleared else: print("⚠️ [Fallback] 資料庫清除失敗，自動切換至 JSON 快取清除...") except Exception as e: print(f"⚠️ [Fallback] 資料庫清除發生例外: {e}，自動切換至 JSON 快取清除...") # 2. 清除 JSON (如果 DB 沒清掉，或 USE_DB=False，或者是連同舊檔一起清確保乾淨) cache_file = get_cache_file(host, config_type) if not os.path.exists(cache_file): # 如果是走 DB，且有清掉，回傳 DB 的結果 if USE_DB and cleared_count > 0: return {"status": "success", "cleared_count": cleared_count} return {"status": "success", "cleared_count": 0, "message": "快取檔案不存在"} try: with open(cache_file, "r", encoding="utf-8") as f: cache_data = json.load(f) json_cleared_count = 0 for path in paths_to_clear: if path in cache_data: del cache_data[path] json_cleared_count += 1 if json_cleared_count > 0: with open(cache_file, "w", encoding="utf-8") as f: json.dump(cache_data, f, ensure_ascii=False, indent=4) # 回傳較大的那個數值 final_count = max(cleared_count, json_cleared_count) return {"status": "success", "cleared_count": final_count} except Exception as e: # 確保回傳標準 JSON 格式 return {"status": "error", "message": f"資料庫連線與快取存取皆失敗: {str(e)}"}