# --- 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：預處理 (修復終端機 200 字元自動折行問題) # ========================================== raw_lines = cli_text.split('\n') fixed_lines = [] # 定義常見的 Root 指令關鍵字，避免把正常的指令誤判為斷行 root_keywords = ( 'alias', 'ssh', 'hostname', 'logging', 'cable', 'ipdr', 'snmp-server', 'aaa', 'radius-server', 'privilege', 'cli', 'packetcable', 'system', 'network', 'clock', 'no' ) for line in raw_lines: line = line.replace('\r', '') if not line.strip(): continue # 💡 判斷是否為被截斷的行： # 1. 前一行長度非常長 (大於 190 字元，根據您的截圖，極限是 199) # 2. 當前行「沒有縮排」(終端機自動折行會把字元推到最左邊第 0 格) # 3. 當前行不是註解 '!' if fixed_lines and len(fixed_lines[-1]) > 190 and not line.startswith(' ') and not line.startswith('!'): # 取得當前行的第一個單字 first_word = line.strip().split()[0] if line.strip() else "" # 確保它不是一個正常的 Root 指令 if first_word not in root_keywords: # ✅ 觸發黏合邏輯：將這行直接接在上一行後面 fixed_lines[-1] = fixed_lines[-1] + line continue fixed_lines.append(line) # 統一去除右側多餘空白，準備進入階段 1 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()) # 🌟 嚴格判斷 1：遇到「真實指令」縮排退回，才代表區塊結束 # (加上 stripped != '!'，防止設備偶發的「0 縮排 !」導致區塊被誤殺) if indent < current_indent and stripped != '!': return nodes, i # 🌟 嚴格判斷 2：處理 ! (無論縮排多少，都當純分隔符消耗掉) 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): """將 'cable mac-domain 13' 拆解為巢狀字典""" 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: current[f"{part} (資料夾)"] = value_dict else: if part not in current or not isinstance(current[part], dict): 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) # 判斷是否全為純陣列值 (例如 IP 列表) 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 = {} # 1. 處理資料夾 (Folders) 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) # 2. 處理單行指令 (Leaves) 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) # 全域非同步鎖：確保同一時間只有一人能執行設定寫入 cmts_config_lock = asyncio.Lock()