296 lines
14 KiB
Python
296 lines
14 KiB
Python
# --- routers/query.py ---
|
||
import re
|
||
import asyncssh
|
||
import asyncio
|
||
from fastapi import APIRouter, HTTPException, Query
|
||
from shared import CMTS_DEVICE
|
||
|
||
router = APIRouter()
|
||
|
||
# ==========================================
|
||
# 🛠️ AsyncSSH 共用執行引擎 (取代 Netmiko)
|
||
# ==========================================
|
||
|
||
async def execute_single_command(host, username, password, command, timeout=15.0):
|
||
"""
|
||
執行單一查詢指令 (適用於 90% 的標準 show 指令)
|
||
自動防呆：確保指令帶有取消分頁的後綴，防止 Event Loop 卡死
|
||
"""
|
||
try:
|
||
async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn:
|
||
# 確保指令包含取消分頁的參數
|
||
if "| nomore" not in command.lower():
|
||
command += " | nomore"
|
||
|
||
result = await conn.run(command, check=False, timeout=timeout)
|
||
return result.stdout or ""
|
||
|
||
except asyncssh.Error as e:
|
||
raise Exception(f"SSH Authentication/Connection Error: {str(e)}")
|
||
except asyncio.TimeoutError:
|
||
raise Exception("SSH Timeout Error: 設備無回應 (Timeout)")
|
||
except Exception as e:
|
||
raise Exception(f"SSH Unexpected Error: {str(e)}")
|
||
|
||
async def execute_interactive_command(host, username, password, commands: list, timeout=15.0):
|
||
"""
|
||
執行互動式指令 (適用於需要進入 config 模式，或使用 '?' 觸發補全的指令)
|
||
動態處理終端機的 --More-- 提示
|
||
"""
|
||
try:
|
||
async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn:
|
||
async with conn.create_process(term_type='xterm-256color', term_size=(200, 24)) as process:
|
||
|
||
async def read_until_quiet(wait_time):
|
||
output = ""
|
||
while True:
|
||
try:
|
||
chunk = await asyncio.wait_for(process.stdout.read(4096), timeout=wait_time)
|
||
if not chunk: break
|
||
output += chunk
|
||
# 動態處理分頁符號
|
||
if "--More--" in chunk or "More" in chunk:
|
||
process.stdin.write(" ")
|
||
await process.stdin.drain()
|
||
except asyncio.TimeoutError:
|
||
break
|
||
return output
|
||
|
||
# 1. 清除登入 MOTD
|
||
await read_until_quiet(1.5)
|
||
|
||
# 2. 執行取消分頁指令 (雙重保險)
|
||
process.stdin.write("terminal length 0\n")
|
||
await process.stdin.drain()
|
||
await read_until_quiet(0.5)
|
||
|
||
# 3. 執行目標指令
|
||
final_output = ""
|
||
for cmd in commands:
|
||
process.stdin.write(cmd)
|
||
await process.stdin.drain()
|
||
final_output += await read_until_quiet(timeout)
|
||
|
||
# 4. 退出 session
|
||
process.stdin.write("exit\n")
|
||
await process.stdin.drain()
|
||
|
||
# 清理 ANSI 控制碼與退格鍵雜訊
|
||
clean_output = re.sub(r'\x1b\[[0-9;?]*[a-zA-Z]|\x08', '', final_output)
|
||
return clean_output
|
||
|
||
except asyncssh.Error as e:
|
||
raise Exception(f"SSH Authentication/Connection Error: {str(e)}")
|
||
except asyncio.TimeoutError:
|
||
raise Exception("SSH Timeout Error: 設備無回應 (Timeout)")
|
||
except Exception as e:
|
||
raise Exception(f"SSH Unexpected Error: {str(e)}")
|
||
|
||
# ==========================================
|
||
# 🚀 API 路由
|
||
# ==========================================
|
||
|
||
@router.get("/cmts-query")
|
||
async def get_cmts_query(
|
||
query_type: str = Query(..., description="查詢動作類型"),
|
||
target: str = Query("", description="MAC Address 或 VC:VS (選填)"),
|
||
host: str = Query(...),
|
||
username: str = Query(...),
|
||
password: str = Query(...)
|
||
):
|
||
try:
|
||
target_str = f" {target.strip()}" if target.strip() else ""
|
||
|
||
commands = {
|
||
"base": f"show cable modem{target_str}",
|
||
"cpe": f"show cable modem{target_str} cpe",
|
||
"cpe_dhcp": f"show cable modem{target_str} cpe dhcp",
|
||
"cpe_ipv6": f"show cable modem{target_str} cpe ipv6",
|
||
"bonding": f"show cable modem{target_str} bonding",
|
||
"bonding_ds": f"show cable modem{target_str} bonding downstream",
|
||
"bonding_us": f"show cable modem{target_str} bonding upstream",
|
||
"phy": f"show cable modem{target_str} phy",
|
||
"verbose": f"show cable modem{target_str} verbose",
|
||
"ofdm_profile": f"show cable modem{target_str} ofdm-profile",
|
||
"ofdma_profile": f"show cable modem{target_str} ofdma-profile",
|
||
"dhcp_verbose": f"show cable modem{target_str} dhcp verbose",
|
||
"service_flow": f"show cable modem{target_str} service-flow",
|
||
"service_flow_verbose": f"show cable modem{target_str} service-flow verbose",
|
||
"qos": f"show cable modem{target_str} qos",
|
||
"uptime": f"show cable modem{target_str} uptime",
|
||
"ugs": f"show cable modem{target_str} ugs",
|
||
"cm_status": f"show cable modem{target_str} cm-status",
|
||
"partial_mode": "show cable modem partial-mode",
|
||
"hop": "show cable hop",
|
||
"flap_list": "show cable flap-list",
|
||
"flap_sum": "show cable flap-sum",
|
||
"rpd_base": f"show cable rpd{target_str}",
|
||
"rpd_verbose": f"show cable rpd{target_str} verbose",
|
||
"rpd_ptp_time": f"show cable rpd{target_str} ptp time-property",
|
||
"rpd_ptp_verbose": f"show cable rpd{target_str} ptp verbose",
|
||
"rpd_counters_map": f"show cable rpd{target_str} counters map",
|
||
"rpd_capabilities": f"show cable rpd{target_str} capabilities",
|
||
"rpd_video_counters": f"show cable rpd{target_str} video-channel counters",
|
||
"rpd_env_temp": f"show cable rpd{target_str} environment temperature",
|
||
"rpd_env_volt": f"show cable rpd{target_str} environment voltage",
|
||
"rpd_session": f"show cable rpd{target_str} session",
|
||
"rpd_reset_history": f"show cable rpd{target_str} reset-history",
|
||
"rpd_port_transceiver": f"show cable rpd{target_str} port-transceiver",
|
||
}
|
||
|
||
if query_type not in commands:
|
||
raise ValueError(f"未知的查詢類型: {query_type}")
|
||
|
||
cli_command = commands[query_type] + " | nomore"
|
||
raw_output = await execute_single_command(host, username, password, cli_command, timeout=15.0)
|
||
|
||
return {"status": "success", "command": cli_command, "data": raw_output}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"CMTS Query Error: {str(e)}")
|
||
|
||
@router.get("/cmts-mac-domain-config")
|
||
async def get_mac_domain_config(target: str, host: str, username: str, password: str):
|
||
try:
|
||
cli_command = f"show running-config cable mac-domain {target.strip()} | nomore"
|
||
raw_output = await execute_single_command(host, username, password, cli_command, timeout=15.0)
|
||
|
||
# 💡 正名工程：更新字典的 Key 為一致性的命名
|
||
config = {
|
||
"common_settings": { "ip-provisioning-mode": "dual-stack", "diplexer-band-edge": "enabled", "cm-battery-mode-31-support": "disabled", "cm-battery-mode-30-support": "disabled", "docsis40": "disabled", "ds-dynamic-bonding-group": "disabled", "us-dynamic-bonding-group": "disabled" },
|
||
"basic_channel_sets": { "admin-state": "down", "ds-primary-set": "", "ds-non-primary-set": "", "us-phy-channel-set": "", "ds-ofdm-set": "", "us-ofdma-set": "" },
|
||
"static_ds_bonding_groups": {},
|
||
"static_us_bonding_groups": {}
|
||
}
|
||
|
||
current_group_type = None
|
||
current_group_name = None
|
||
|
||
for line in raw_output.splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("cable mac-domain"): continue
|
||
|
||
if line.startswith("ds-bonding-group"):
|
||
current_group_type = "static_ds_bonding_groups"
|
||
current_group_name = line.split()[1]
|
||
config["static_ds_bonding_groups"][current_group_name] = {"admin-state": "down"}
|
||
continue
|
||
elif line.startswith("us-bonding-group"):
|
||
current_group_type = "static_us_bonding_groups"
|
||
current_group_name = line.split()[1]
|
||
config["static_us_bonding_groups"][current_group_name] = {"admin-state": "down"}
|
||
continue
|
||
elif line == "!":
|
||
current_group_type = None
|
||
current_group_name = None
|
||
continue
|
||
|
||
if current_group_type and current_group_name:
|
||
parts = line.split(maxsplit=1)
|
||
if len(parts) == 2: config[current_group_type][current_group_name][parts[0]] = parts[1]
|
||
else:
|
||
# 💡 正名工程：將解析結果存入對應的新 Key 中
|
||
if line.startswith("ip-provisioning-mode"): config["common_settings"]["ip-provisioning-mode"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("diplexer-band-edge control"): config["common_settings"]["diplexer-band-edge"] = line.split(maxsplit=2)[2]
|
||
elif line.startswith("cm-battery-mode-31-support"): config["common_settings"]["cm-battery-mode-31-support"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("cm-battery-mode-30-support"): config["common_settings"]["cm-battery-mode-30-support"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("docsis40"): config["common_settings"]["docsis40"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("ds-dynamic-bonding-group"): config["common_settings"]["ds-dynamic-bonding-group"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("us-dynamic-bonding-group"): config["common_settings"]["us-dynamic-bonding-group"] = line.split(maxsplit=1)[1]
|
||
|
||
# 💡 正名工程：將原本的 dynamic_bonding_groups 改為 basic_channel_sets
|
||
elif line.startswith("admin-state"): config["basic_channel_sets"]["admin-state"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("ds-primary-set"): config["basic_channel_sets"]["ds-primary-set"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("ds-non-primary-set"): config["basic_channel_sets"]["ds-non-primary-set"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("us-phy-channel-set"): config["basic_channel_sets"]["us-phy-channel-set"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("ds-ofdm-set"): config["basic_channel_sets"]["ds-ofdm-set"] = line.split(maxsplit=1)[1]
|
||
elif line.startswith("us-ofdma-set"): config["basic_channel_sets"]["us-ofdma-set"] = line.split(maxsplit=1)[1]
|
||
|
||
return {"status": "success", "data": config}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Parse Error: {str(e)}")
|
||
|
||
@router.get("/cmts-mac-domain-list")
|
||
async def get_cmts_mac_domain_list(
|
||
host: str = Query(...),
|
||
username: str = Query(...),
|
||
password: str = Query(...)
|
||
):
|
||
try:
|
||
# 這裡使用 '?' 觸發補全，必須使用互動式引擎
|
||
commands = ["show running-config cable mac-domain ?"]
|
||
raw_output = await execute_interactive_command(host, username, password, commands, timeout=5.0)
|
||
|
||
matches = re.findall(r'\b\d+:\d+/\d+\.\d+\b', raw_output)
|
||
mac_domains = list(dict.fromkeys(matches))
|
||
|
||
return {"status": "success", "data": mac_domains}
|
||
except Exception as e:
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
@router.get("/cmts-version")
|
||
async def get_cmts_version(
|
||
host: str = Query(...),
|
||
username: str = Query(...),
|
||
password: str = Query(...)
|
||
):
|
||
try:
|
||
raw_output = await execute_single_command(host, username, password, "show version", timeout=15.0)
|
||
|
||
# 🌟 使用 Regex 尋找 infra 或 vcmts-cd-0 後面的版本號
|
||
match = re.search(r"(?:infra|vcmts-cd-0)\s+([\w\.\-]+)", raw_output)
|
||
|
||
if match:
|
||
return {"status": "success", "version": match.group(1)}
|
||
else:
|
||
return {"status": "error", "message": "無法解析版本號", "raw_output": raw_output}
|
||
|
||
except Exception as e:
|
||
return {"status": "error", "message": f"CMTS Connection Error: {str(e)}"}
|
||
|
||
@router.get("/list-cms", summary="獲取設備上所有的 CM MAC 清單")
|
||
async def list_cms(host: str, username: str, password: str):
|
||
try:
|
||
raw_output = await execute_single_command(host, username, password, "show cable modem", timeout=15.0)
|
||
|
||
if not raw_output:
|
||
return {"cms": []}
|
||
|
||
# 嚴謹的 Regex：匹配 xxxx.xxxx.xxxx 或 xx:xx:xx:xx:xx:xx
|
||
mac_pattern = re.compile(r"([0-9a-fA-F]{4}\.[0-9a-fA-F]{4}\.[0-9a-fA-F]{4}|[0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5})")
|
||
macs = []
|
||
|
||
for line in raw_output.splitlines():
|
||
match = mac_pattern.search(line)
|
||
if match:
|
||
macs.append(match.group(1).lower())
|
||
|
||
# 去重複並排序
|
||
return {"cms": sorted(list(set(macs)))}
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ [Background Task] 獲取 CM 清單失敗: {e}")
|
||
return {"cms": []} # 發生錯誤 (如 Timeout, 空行) 安全回傳空陣列
|
||
|
||
@router.get("/list-rpds", summary="獲取設備上所有的 RPD VC:VS 清單")
|
||
async def list_rpds(host: str, username: str, password: str):
|
||
try:
|
||
raw_output = await execute_single_command(host, username, password, "show cable rpd", timeout=15.0)
|
||
|
||
if not raw_output:
|
||
return {"rpds": []}
|
||
|
||
# 嚴謹的 Regex：匹配 數字:數字 (例如 13:0)，並使用 Negative Lookbehind/Lookahead 避免匹配到 MAC 或 IPv6
|
||
vcvs_pattern = re.compile(r"(?<![:\w])(\d{1,3}:\d{1,3})(?![:\w])")
|
||
rpds = []
|
||
|
||
for line in raw_output.splitlines():
|
||
match = vcvs_pattern.search(line)
|
||
if match:
|
||
rpds.append(match.group(1))
|
||
|
||
return {"rpds": sorted(list(set(rpds)))}
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ [Background Task] 獲取 RPD 清單失敗: {e}")
|
||
return {"rpds": []}
|