# --- 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(): net_connect = None try: net_connect = ConnectHandler(**CMTS_DEVICE) raw_output = str(net_connect.send_command("show cable modem")) 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)}") finally: if net_connect: net_connect.disconnect() @router.get("/configuration") async def get_configuration(): net_connect = None 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") return {"status": "success", "data": raw_output} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Error: {str(e)}") finally: if net_connect: net_connect.disconnect() @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(...) ): net_connect = None 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) 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)}") finally: if net_connect: net_connect.disconnect() @router.get("/cmts-mac-domain-config") async def get_mac_domain_config(target: str, host: str, username: str, password: str): net_connect = None 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 = str(net_connect.send_command(cli_command, read_timeout=15)) # 💡 正名工程：更新字典的 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)}") finally: if net_connect: net_connect.disconnect() @router.get("/cmts-mac-domain-list") async def get_cmts_mac_domain_list( host: str = Query(...), username: str = Query(...), password: str = Query(...) ): net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command_timing("show running-config cable mac-domain ?")) 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)} finally: if net_connect: net_connect.disconnect() @router.get("/cmts-version") async def get_cmts_version( host: str = Query(...), username: str = Query(...), password: str = Query(...) ): net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command("show version", read_timeout=15)) import re # 🌟 使用 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)}"} finally: if net_connect: net_connect.disconnect()