437 lines
19 KiB
Python
437 lines
19 KiB
Python
# --- 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, get_cmts_lock, parse_cli_to_tree, deep_split_tree
|
||
from collections import defaultdict
|
||
from fastapi.responses import StreamingResponse
|
||
from logger import get_all_log_levels, set_log_level
|
||
|
||
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:
|
||
try:
|
||
db_filters = await database.get_tree_filters(config_type)
|
||
return db_filters if db_filters is not None else []
|
||
except Exception as e:
|
||
print(f"資料庫讀取過濾器發生例外: {e}")
|
||
return []
|
||
|
||
async def save_tree_filters(config_type: str, hidden_keys: list):
|
||
try:
|
||
await database.upsert_tree_filters(config_type, hidden_keys)
|
||
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 = get_cmts_lock[req.host]
|
||
async with host_lock:
|
||
try:
|
||
yield json.dumps({"status": "progress", "message": f"🔌 準備連線至設備 {req.host}..."}) + "\n"
|
||
|
||
# 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError)
|
||
conn = await asyncio.wait_for(
|
||
asyncssh.connect(host, username=username, password=password, known_hosts=None),
|
||
timeout=10.0
|
||
)
|
||
# 再進入 async with 確保資源會被自動關閉
|
||
async with 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}) 已更新"}
|
||
|
||
# ==========================================
|
||
# 🌟 伺服器日誌管控 API (Server Log Management)
|
||
# ==========================================
|
||
class LogLevelRequest(BaseModel):
|
||
module: str
|
||
level: str
|
||
|
||
@router.get("/settings/logs")
|
||
async def api_get_log_levels():
|
||
"""獲取目前所有模組的日誌等級"""
|
||
return {"status": "success", "data": get_all_log_levels()}
|
||
|
||
@router.post("/settings/logs")
|
||
async def api_set_log_level(req: LogLevelRequest):
|
||
"""動態修改特定模組的日誌等級"""
|
||
success = set_log_level(req.module, req.level)
|
||
if success:
|
||
return {"status": "success", "message": f"已將 {req.module} 的日誌等級設為 {req.level}"}
|
||
raise HTTPException(status_code=400, detail="未知的模組名稱") |