```
feat(api): 添加OLT端口管理和设备发现功能 - 在check.py中更新discover_olt方法,使用scan_and_discover替代check_olt_devices - 在olt.py中添加get_olt_ports和toggle_olt_port接口,支持获取和控制OLT端口状态 - 添加TogglePortRequest模型用于端口操作请求验证 - 更新SSHService类,新增get_olt_ports和toggle_olt_port方法 - 优化CheckService中的MAC地址索引逻辑,支持模型信息更新 ```
This commit is contained in:
@@ -276,6 +276,66 @@ class SSHService:
|
||||
model=model
|
||||
)
|
||||
|
||||
def get_olt_ports(self) -> list:
|
||||
"""获取所有 Olt 端口状态,返回 [{'name': 'Olt1/0/1', 'status': 'up'}, ...]"""
|
||||
output = self.execute_command("display interface brief")
|
||||
ports = []
|
||||
for line in output.splitlines():
|
||||
line = line.strip()
|
||||
if not line.startswith("Olt"):
|
||||
continue
|
||||
parts = line.split()
|
||||
if len(parts) < 2:
|
||||
continue
|
||||
name = parts[0]
|
||||
link = parts[1].upper()
|
||||
# ADM 表示 administratively down(手动关闭)
|
||||
if link == "ADM":
|
||||
status = "adm-down"
|
||||
elif link == "UP":
|
||||
status = "up"
|
||||
else:
|
||||
status = "down"
|
||||
ports.append({"name": name, "status": status})
|
||||
return ports
|
||||
|
||||
def toggle_olt_port(self, port_name: str, action: str) -> bool:
|
||||
"""开启或关闭 OLT 端口
|
||||
action: 'shutdown' 或 'undo shutdown'
|
||||
流程: system-view -> interface {port_name} -> shutdown/undo shutdown -> quit -> quit
|
||||
"""
|
||||
if not self.shell:
|
||||
raise Exception("SSH 未连接")
|
||||
|
||||
def send_and_wait(cmd: str, expect: str, timeout: int = 10) -> str:
|
||||
self.shell.send(cmd + "\n")
|
||||
buf = ""
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
if self.shell.recv_ready():
|
||||
buf += self.shell.recv(4096).decode('utf-8', errors='ignore')
|
||||
if expect in buf:
|
||||
return buf
|
||||
else:
|
||||
time.sleep(0.2)
|
||||
return buf
|
||||
|
||||
out = send_and_wait("system-view", "]")
|
||||
if "]" not in out:
|
||||
raise Exception("进入 system-view 失败")
|
||||
|
||||
out = send_and_wait(f"interface {port_name}", "]")
|
||||
if "]" not in out:
|
||||
raise Exception(f"进入端口 {port_name} 失败")
|
||||
|
||||
out = send_and_wait(action, "]")
|
||||
if "]" not in out:
|
||||
raise Exception(f"执行 {action} 失败")
|
||||
|
||||
send_and_wait("quit", "]", timeout=5)
|
||||
send_and_wait("quit", ">", timeout=5)
|
||||
return True
|
||||
|
||||
def close(self):
|
||||
"""关闭 SSH 连接"""
|
||||
if self.client:
|
||||
|
||||
Reference in New Issue
Block a user