scm-harmonic-cmts-admin/shared.py

# --- shared.py ---
import asyncio
import json
import os
import re
from collections import defaultdict

DEBUG_MODE = False 

def debug_print(msg: str):
    if DEBUG_MODE:
        print(msg)

# 🌟 2. 淨化預設的 CMTS 連線樣板
CMTS_DEVICE = {
    'device_type': 'cisco_ios',
    'host': os.getenv("DEFAULT_CMTS_HOST", ""),
    'username': os.getenv("DEFAULT_CMTS_USER", ""),
    'password': os.getenv("DEFAULT_CMTS_PASS", ""),
    'port': 22,
    'fast_cli': False,
    'global_delay_factor': 2
}

def parse_cli_to_tree(cli_text: str) -> dict:
    """
    將 CMTS 的 CLI 配置文字轉換為完美的邏輯巢狀 JSON (兩階段解析法)
    """
    # ==========================================
    # 🌟 階段 0：預處理 (清理雜訊，移除危險的自動黏合)
    # ==========================================
    cli_text = cli_text.replace('\r\n', '\n').replace('\r', '\n')
    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')

    raw_lines = cli_text.split('\n')
    fixed_lines = []
    
    for line in raw_lines:
        line = ansi_escape.sub('', line)
        line = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]', '', line)

        if not line.strip():
            continue
            
        # 💡 乾淨俐落：直接加入陣列，不再做危險的字串長度黏合判斷
        fixed_lines.append(line)
        
    lines = [line.rstrip() for line in fixed_lines if line.strip()]

    # ==========================================
    # 🌟 階段 1：建立實體樹 (嚴格比對縮排與 '!' 的對應關係)
    # ==========================================
    def build_raw_tree(start_idx, current_indent):
        nodes = []
        i = start_idx
        while i < len(lines):
            line = lines[i]
            stripped = line.strip()
            indent = len(line) - len(line.lstrip())
            
            if indent < current_indent and stripped != '!':
                return nodes, i
                
            if stripped == '!':
                i += 1
                continue
            
            has_children = False
            next_i = i + 1
            while next_i < len(lines) and lines[next_i].strip() == '!':
                next_i += 1
                
            if next_i < len(lines):
                next_line = lines[next_i]
                next_indent = len(next_line) - len(next_line.lstrip())
                if next_indent > indent:
                    has_children = True
            
            if has_children:
                children, next_i = build_raw_tree(i + 1, next_indent)
                nodes.append({"type": "folder", "content": stripped, "children": children})
                i = next_i
            else:
                nodes.append({"type": "leaf", "content": stripped})
                i += 1
        return nodes, i

    raw_tree, _ = build_raw_tree(0, 0)

    # ==========================================
    # 🌟 階段 2：邏輯群組化與字典合併 (保留嚴格型別防護)
    # ==========================================
    def deep_merge(dict1, dict2):
        for k, v in dict2.items():
            if k in dict1:
                if isinstance(dict1[k], dict) and isinstance(v, dict):
                    deep_merge(dict1[k], v)
                else:
                    idx = 1
                    while f"{k} [{idx}]" in dict1:
                        idx += 1
                    dict1[f"{k} [{idx}]"] = v
            else:
                dict1[k] = v

    def nest_folder_key(key_string, value_dict, target_dict):
        parts = key_string.split()
        current = target_dict
        for i, part in enumerate(parts):
            if i == len(parts) - 1:
                if part not in current:
                    current[part] = value_dict
                else:
                    if isinstance(current[part], dict) and isinstance(value_dict, dict):
                        deep_merge(current[part], value_dict)
                    else:
                        idx = 1
                        while f"{part} (衝突 {idx})" in current:
                            idx += 1
                        current[f"{part} (衝突 {idx})"] = value_dict
            else:
                if part not in current or not isinstance(current[part], dict):
                    if part in current and not isinstance(current[part], dict):
                        old_val = current[part]
                        current[part] = {"[0]": old_val}
                    else:
                        current[part] = {}
                current = current[part]

    def group_leaves(leaf_strings):
        grouped = {}
        first_word_map = defaultdict(list)
        
        for s in leaf_strings:
            parts = s.split(maxsplit=1)
            first_word = parts[0]
            remainder = parts[1] if len(parts) > 1 else ""
            first_word_map[first_word].append(remainder)
            
        for prefix, remainders in first_word_map.items():
            if len(remainders) == 1:
                grouped[prefix] = remainders[0]
            else:
                valid_remainders = [r for r in remainders if r]
                empty_count = len(remainders) - len(valid_remainders)
                
                group_dict = {}
                if valid_remainders:
                    sub_grouped = group_leaves(valid_remainders)
                    is_all_empty_vals = all(not isinstance(v, dict) and v == "" for v in sub_grouped.values())
                    
                    if is_all_empty_vals:
                        for i, k in enumerate(sub_grouped.keys()):
                            group_dict[f"[{i}]"] = k
                    else:
                        idx = 0
                        for k, v in sub_grouped.items():
                            if isinstance(v, dict):
                                group_dict[k] = v
                            elif v == "":
                                group_dict[f"[{idx}]"] = k
                                idx += 1
                            else:
                                group_dict[k] = v
                                
                for _ in range(empty_count):
                    idx = 0
                    while f"[{idx}]" in group_dict:
                        idx += 1
                    group_dict[f"[{idx}]"] = ""
                    
                grouped[prefix] = group_dict
                
        return grouped

    def fold_nodes(nodes):
        result = {}
        for node in [n for n in nodes if n["type"] == "folder"]:
            folded_children = fold_nodes(node["children"])
            nest_folder_key(node["content"], folded_children, result)
        
        leaves = [n["content"] for n in nodes if n["type"] == "leaf"]
        if leaves:
            leaf_dict = group_leaves(leaves)
            deep_merge(result, leaf_dict)
        return result

    return fold_nodes(raw_tree)

def deep_split_tree(data):
    """將平坦的 CLI 字典，依照空白字元拆分成深層巢狀結構 (具備終極型態衝突保護)"""
    if not isinstance(data, dict):
        return data
        
    nested_tree = {}
    for key, value in data.items():
        processed_value = deep_split_tree(value)
        
        parts = str(key).strip().split()
        if not parts:
            continue
            
        current = nested_tree
        for i, part in enumerate(parts):
            if i == len(parts) - 1:
                if part not in current:
                    current[part] = processed_value
                else:
                    # 💡 終極衝突保護：安全轉型，絕不遺失資料
                    if isinstance(current[part], dict) and isinstance(processed_value, dict):
                        current[part].update(processed_value)
                    elif isinstance(current[part], dict) and not isinstance(processed_value, dict):
                        # 原本是資料夾，新來的是字串 -> 塞入虛擬 key [0], [1]
                        idx = 0
                        while f"[{idx}]" in current[part]: idx += 1
                        current[part][f"[{idx}]"] = processed_value
                    elif not isinstance(current[part], dict) and isinstance(processed_value, dict):
                        # 原本是字串，新來的是資料夾 -> 升級成資料夾，並保留原字串至 [0]
                        old_val = current[part]
                        current[part] = processed_value
                        idx = 0
                        while f"[{idx}]" in current[part]: idx += 1
                        current[part][f"[{idx}]"] = old_val
                    else:
                        # 兩者都是字串 -> 升級成資料夾，包含兩個字串
                        old_val = current[part]
                        current[part] = {"[0]": old_val, "[1]": processed_value}
            else:
                if part not in current:
                    current[part] = {}
                elif not isinstance(current[part], dict):
                    # 中間節點原本是字串，必須升級為資料夾，並保留原字串
                    old_val = current[part]
                    current[part] = {"[0]": old_val}
                current = current[part]
                
    return nested_tree

# ==========================================
# 💡 系統設定管理 (System Settings Manager)
# ==========================================
SETTINGS_FILE = "system_settings.json"

def load_settings():
    """讀取系統設定，若檔案不存在則回傳預設的黑名單"""
    if os.path.exists(SETTINGS_FILE):
        try:
            with open(SETTINGS_FILE, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            pass
    # 預設隱藏這些比較雜亂且不常修改的系統層級設定
    return {"hidden_keys": ["logging", "privilege", "aaa", "certificate", "line"]}

def save_settings(settings_data):
    """將系統設定寫入 JSON 檔案"""
    with open(SETTINGS_FILE, "w", encoding="utf-8") as f:
        json.dump(settings_data, f, indent=4, ensure_ascii=False)

# 🌟 [Priority 1 修復] 將全域非同步鎖改為依設備 IP (Host) 隔離的鎖 (安全動態獲取版)
_cmts_config_locks = {}

def get_cmts_lock(host: str) -> asyncio.Lock:
    if host not in _cmts_config_locks:
        _cmts_config_locks[host] = asyncio.Lock()
    return _cmts_config_locks[host]