# --- 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) # 全域非同步鎖：確保同一時間只有一人能執行設定寫入 cmts_config_lock = asyncio.Lock()