203 lines
10 KiB
Python
203 lines
10 KiB
Python
# --- routers/query.py ---
|
|
import re
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from netmiko import ConnectHandler
|
|
from shared import CMTS_DEVICE
|
|
|
|
router = APIRouter()
|
|
|
|
def parse_cm_output(raw_text: str) -> list:
|
|
parsed_data = []
|
|
lines = raw_text.splitlines()
|
|
data_started = False
|
|
for line in lines:
|
|
if not line.strip(): continue
|
|
if line.strip().startswith('---'):
|
|
data_started = True
|
|
continue
|
|
if line.strip().startswith('==='): break
|
|
if data_started:
|
|
columns = re.split(r'\s{2,}', line.strip())
|
|
if len(columns) >= 8:
|
|
cm_info = {
|
|
"downstream": columns[0], "upstream": columns[1],
|
|
"bond_cap": columns[2], "ofdm_cap": columns[3],
|
|
"mac_address": columns[4], "ip_address": columns[5],
|
|
"num_cpe": int(columns[6]), "state": columns[7]
|
|
}
|
|
parsed_data.append(cm_info)
|
|
return parsed_data
|
|
|
|
@router.get("/cable-modems")
|
|
async def get_cable_modems():
|
|
try:
|
|
net_connect = ConnectHandler(**CMTS_DEVICE)
|
|
raw_output = net_connect.send_command("show cable modem")
|
|
net_connect.disconnect()
|
|
structured_data = parse_cm_output(raw_output)
|
|
return {"status": "success", "total_count": len(structured_data), "data": structured_data}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"CMTS Connection Error: {str(e)}")
|
|
|
|
@router.get("/configuration")
|
|
async def get_configuration():
|
|
try:
|
|
net_connect = ConnectHandler(**CMTS_DEVICE)
|
|
net_connect.send_command_timing("config")
|
|
raw_output = net_connect.send_command("show full-configuration | nomore", expect_string=r"#", read_timeout=120)
|
|
net_connect.send_command_timing("exit")
|
|
net_connect.disconnect()
|
|
return {"status": "success", "data": raw_output}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"CMTS Error: {str(e)}")
|
|
|
|
@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:
|
|
device = CMTS_DEVICE.copy()
|
|
device.update({'host': host, 'username': username, 'password': password})
|
|
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"
|
|
net_connect = ConnectHandler(**device)
|
|
raw_output = net_connect.send_command(cli_command, read_timeout=15)
|
|
net_connect.disconnect()
|
|
|
|
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:
|
|
device = CMTS_DEVICE.copy()
|
|
device.update({'host': host, 'username': username, 'password': password})
|
|
|
|
cli_command = f"show running-config cable mac-domain {target.strip()} | nomore"
|
|
net_connect = ConnectHandler(**device)
|
|
raw_output = net_connect.send_command(cli_command, read_timeout=15)
|
|
net_connect.disconnect()
|
|
|
|
# 💡 正名工程：更新字典的 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:
|
|
device = CMTS_DEVICE.copy()
|
|
device.update({'host': host, 'username': username, 'password': password})
|
|
net_connect = ConnectHandler(**device)
|
|
|
|
raw_output = net_connect.send_command_timing("show running-config cable mac-domain ?")
|
|
net_connect.disconnect()
|
|
|
|
import re
|
|
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)}
|