# --- routers/config.py --- import re from fastapi import APIRouter, HTTPException from pydantic import BaseModel from typing import List from netmiko import ConnectHandler # 💡 確保從 shared 引入 deep_split_tree from shared import CMTS_DEVICE, cmts_config_lock, parse_cli_to_tree, deep_split_tree, load_settings, save_settings from collections import defaultdict router = APIRouter() 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): # 使用非同步鎖，確保同一時間只有一個寫入任務執行 async with cmts_config_lock: try: device = CMTS_DEVICE.copy() device.update({'host': req.host, 'username': req.username, 'password': req.password}) # 過濾掉空行與註解 (! 開頭的行) commands = [cmd.strip() for cmd in req.script.splitlines() if cmd.strip() and not cmd.strip().startswith('!')] net_connect = ConnectHandler(**device) # 🌟 1. 手動進入設定模式 (恢復您原本正確的寫法) net_connect.send_command_timing("config") # 🌟 2. 批次送出設定指令 # 加上 enter_config_mode=False 與 exit_config_mode=False # 這樣 Netmiko 就不會自作主張去打 config terminal 了 output = net_connect.send_config_set( commands, read_timeout=60, cmd_verify=False, enter_config_mode=False, exit_config_mode=False ) # 🌟 3. 手動退出設定模式 net_connect.send_command_timing("exit") net_connect.disconnect() # 🌟 新增：檢查設備回傳的文字中是否包含拒絕或錯誤的關鍵字 lower_output = output.lower() if "aborted" in lower_output or "error" in lower_output or "invalid" in lower_output: # 回傳錯誤狀態，前端就會顯示紅色字體且「不會」自動關閉視窗 return { "status": "error", "message": "CMTS 拒絕了設定 (Commit Aborted)", "detail": output } return {"status": "success", "data": output} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Config Error: {str(e)}") @router.post("/cmts-generate-cli") async def generate_cli(req: GenerateCliRequest): """ 將前端的 JSON Diff 轉譯為 Harmonic 單行 CLI 腳本 並自動處理 admin-state 的安全生命週期 (先 down 後 up) """ # 用來將同一個介面的變更群組化 interface_groups = defaultdict(list) # 用來記錄該介面最終的 admin-state 應該是什麼 interface_admin_states = {} for diff in req.diffs: # 將路徑切分：例如 ['cable', 'ds-rf-port', '1:9/0', 'down-channel', '0', 'frequency'] parts = diff.path.split('::') param_name = parts[-1] interface_path = " ".join(parts[:-1]) # 🌟 新增這行：全域清除路徑中間可能夾帶的虛擬資料夾索引 (例如 [0], [1]) # 這樣 "logging evt 1024 [1] active" 就會還原成 "logging evt 1024 active" 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 腳本 cli_lines = [] cli_lines.append("! --- Auto-Generated Configuration Script ---") for interface, changes in interface_groups.items(): cli_lines.append(f"! Configuring: {interface}") # 💡 核心邏輯：判斷這個特定區塊是否支援 admin-state supports_admin_state = interface in req.interfaces_with_admin_state # 1. 安全機制：如果支援，才強制將介面 admin-state down if supports_admin_state: cli_lines.append(f"{interface} admin-state down") # 2. 寫入所有被修改的參數 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}") # 3. 恢復狀態：如果支援，才恢復為 up 或使用者指定的值 if supports_admin_state: final_state = interface_admin_states.get(interface, "up") cli_lines.append(f"{interface} admin-state {final_state}") # Harmonic 必須要有 commit 才會生效 cli_lines.append("commit") cli_lines.append("!") # 空行分隔 # 將獨立修改 admin-state (沒有修改其他參數) 的情況也補上 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 清單""" try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) # 使用 include 只抓取關鍵字，並把等待時間延長至 60 秒 command = 'show running-config | include "cable ds-rf-port"' raw_output = net_connect.send_command(command, read_timeout=60) net_connect.disconnect() # 使用正規表達式擷取 port 號碼 (例如 62:0/0) 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)} @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""" 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 = net_connect.send_command(command, read_timeout=60) net_connect.disconnect() # 1. 初步解析縮排 base_tree = parse_cli_to_tree(raw_output) # 2. 💡 深層解析空白字元，建立完美樹狀結構 perfect_tree = 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)} @router.get("/cmts-full-config") async def get_full_config(host: str, username: str, password: str = "", skip_filter: bool = False): """獲取整台 CMTS 的完整配置，並轉換為大型樹狀 JSON""" try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) command = "show running-config | nomore" raw_output = net_connect.send_command(command, read_timeout=60) net_connect.disconnect() base_tree = parse_cli_to_tree(raw_output) perfect_tree = deep_split_tree(base_tree) # ========================================== # 💡 深度過濾攔截器：支援多層級路徑 (如 cable.mac-domain) # ========================================== if not skip_filter: settings = load_settings() hidden_keys = settings.get("hidden_keys", []) 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)} # ========================================== # 💡 系統設定 API (System Settings) # ========================================== class SettingsRequest(BaseModel): hidden_keys: list[str] @router.get("/settings/tree-filters") async def get_tree_filters(): """獲取目前的樹狀圖隱藏名單""" settings = load_settings() return {"status": "success", "data": settings["hidden_keys"]} @router.post("/settings/tree-filters") async def update_tree_filters(req: SettingsRequest): """更新樹狀圖隱藏名單並存檔""" save_settings({"hidden_keys": req.hidden_keys}) return {"status": "success", "message": "系統設定已更新"}