日常工作中，由于装维人员疏忽经常在现场留下未注册的“僵尸ONU”，影响业务指标的判断，此时要手动将 ONU 进行注册，于是编写了一个自动批量注册“僵尸ONU”的脚本
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
import re
import logging
import argparse
# # 开启 Netmiko 底层调试日志，输出到 netmiko_debug.log 文件
# logging.basicConfig(
# filename='netmiko_debug.log',
# level=logging.DEBUG,
# format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
# )
# logger = logging.getLogger("netmiko")
def parse_autofind_output(output):
"""解析 display ont autofind all 的输出"""
onus = []
pattern = r"Number\s+:\s*(\d+)\s+.*?F/S/P\s+:\s*(\S+)\s+.*?ONT MAC\s+:\s*(\S+)"
matches = re.findall(pattern, output, re.DOTALL)
for match in matches:
onus.append({'number': match[0], 'fsp': match[1], 'mac': match[2]})
return onus
def get_next_ont_id(info_output):
"""
解析 display ont info 输出，获取下一个可用的 ONT ID。
优先填补中间的空洞，如果没有空洞则取最大值 + 1。
"""
pattern = r"\d+/\s*\d+/\s*\d+\s+(\d+)\s+[A-Fa-f0-9-]+"
ids = re.findall(pattern, info_output)
if not ids:
return 1 # 如果没有任何 ONT，直接返回 1
existing_ids = set(int(ont_id) for ont_id in ids)
next_id = 1
while next_id in existing_ids:
next_id += 1
return next_id
def test_olt_autoreg(host, username, password, port=23):
"""
OLT 自动注册主逻辑
"""
# 1. 动态构建设备连接参数
olt_device = {
'device_type': 'huawei_olt_telnet',
'host': host,
'username': username,
'password': password,
'port': port,
'timeout': 30,
'global_delay_factor': 3, # 增加延迟，防止 OLT 处理慢
}
net_connect = None
try:
# 1. 建立 Telnet 连接并提权到配置模式
print(f"[1/5] 正在通过 Telnet 连接到 {olt_device['host']}...")
net_connect = ConnectHandler(**olt_device)
net_connect.set_base_prompt()
# 进入特权模式
net_connect.send_command_timing('enable', delay_factor=2)
# 进入全局配置模式
net_connect.send_command_timing('config', delay_factor=2)
print("[1/5] 提权成功，已进入全局配置模式！\n")
# 2. 获取未注册的 ONU 列表
print("[2/5] 正在获取未注册ONU列表...")
autofind_output = net_connect.send_command("display ont autofind all")
onus = parse_autofind_output(autofind_output)
if not onus:
print("[提示] 未发现未注册的ONU，脚本正常退出。")
return
print(f"[2/5] 共发现 {len(onus)} 个未注册ONU。\n")
# 3. 遍历每个未注册的 ONU 进行注册
print("[3/5] 开始遍历并自动注册ONU...")
for onu in onus:
fsp = onu['fsp']
mac = onu['mac']
slot, pon, port_id = fsp.split('/')
print(f"\n>>> 正在处理 PON口 {fsp} 的 ONU (MAC: {mac})...")
# 3.1 计算下一个可用的 ONT ID
info_output = net_connect.send_command(f"display ont info {slot} {pon} {port_id} all")
next_id = get_next_ont_id(info_output)
# PON口容量校验：EPON ONT ID 范围为 0-63，如果下一个可用 ID 为 64，说明 PON 口已满
if next_id >= 64:
print(f" [警告] PON口 {fsp} 的 ONT ID 资源已耗尽 (当前计算ID: {next_id})，无法自动注册。")
print(f" >>> 请人工介入，检查该 PON 口是否需要扩容或清理僵尸 ONU！")
continue # 跳过当前 ONU，处理下一个
print(f" 当前PON口下一个可用 ID 为: {next_id}")
# 3.2 逐条发送命令并匹配提示符
interface_prompt = rf'.*?\(config-if-epon-{slot}/{pon}\)#'
enter_interface_cmd = f"interface epon {slot}/{pon}"
print(f" 正在进入接口视图: {enter_interface_cmd}")
net_connect.send_command(
enter_interface_cmd,
expect_string=interface_prompt,
cmd_verify=False,
read_timeout=30,
delay_factor=2
)
# 下发注册配置
add_cmd = f"ont add {port_id} {next_id} mac-auth {mac} oam ont-lineprofile-id 1 ont-srvprofile-id 201 des autoreg"
print(f" 下发配置: {add_cmd}")
output = net_connect.send_command_timing(add_cmd, delay_factor=3)
# 退出接口视图
net_connect.send_command("quit", expect_string=r'.*?\(config\)#')
# 3.3 核心逻辑：检查是否报 MAC 已存在错误
if "Failure: ONT MAC is already exist" in output:
print(f" [冲突] MAC {mac} 已存在，正在自动清理旧数据...")
mac_info = net_connect.send_command(f"display ont info by-mac {mac}")
match = re.search(r"F/S/P\s+:\s*(\S+)\s+.*?ONT-ID\s+:\s*(\d+)", mac_info, re.DOTALL)
if match:
real_fsp = match.group(1)
real_ont_id = match.group(2)
real_slot, real_pon, real_port_id = real_fsp.split('/')
print(f" [定位] 发现旧数据位于 {real_fsp} ONT-ID: {real_ont_id}，准备删除...")
# 进入接口视图并尝试删除
real_interface_prompt = rf'.*?\(config-if-epon-{real_slot}/{real_pon}\)#'
net_connect.send_command(f"interface epon {real_slot}/{real_pon}", expect_string=real_interface_prompt)
del_output = net_connect.send_command(f"ont delete {real_port_id} {real_ont_id}", expect_string=real_interface_prompt)
# 核心新增逻辑：如果提示有业务数据流，则先清理数据流
if "Failure" in del_output:
print(f" [拦截] ONU 绑定了业务数据流，无法直接删除。正在清理数据流...")
# 1. 退回全局配置模式
net_connect.send_command("quit", expect_string=r'.*?\(config\)#')
# 2. 查询该 ONU 绑定的所有 service-port
# 注意：这里使用 real_slot/real_pon/real_port_id，对应 F/S/P 和 ONT-ID
sp_cmd = f"display service-port port {real_slot}/{real_pon}/{real_port_id} ont {real_ont_id}"
sp_output = net_connect.send_command(sp_cmd)
# 3. 使用正则提取开头的 INDEX 列（仅匹配数字，绝不匹配 VLAN ID）
# 正则解释：^\s*(\d+) 匹配行首的空白字符后紧跟的数字（即 INDEX）
index_pattern = r"^\s*(\d+)\s+\d+\s+\w+\s+epon"
sp_indexes = re.findall(index_pattern, sp_output, re.MULTILINE)
if sp_indexes:
print(f" [解析] 发现 {len(sp_indexes)} 条业务数据流，开始逐一清理...")
for idx in sp_indexes:
# 4. 执行 undo 命令，匹配全局配置提示符
undo_output = net_connect.send_command(f"undo service-port {idx}", expect_string=r'.*?\(config\)#')
print(f" [清理] 已删除数据流 INDEX: {idx}")
else:
print(f" [警告] 拦截提示有数据流，但未解析到具体 INDEX，请检查输出:\n{sp_output}")
# 5. 数据流清理完毕后，重新进入接口视图并删除 ONU
print(f" [重试] 数据流清理完成，重新尝试删除 ONU...")
net_connect.send_command(f"interface epon {real_slot}/{real_pon}", expect_string=real_interface_prompt)
del_output = net_connect.send_command(f"ont delete {real_port_id} {real_ont_id}", expect_string=real_interface_prompt)
# 退出接口视图
net_connect.send_command("quit", expect_string=r'.*?\(config\)#')
# 检查最终删除是否成功
if "success: 1" in del_output or "success: 1" not in del_output:
# 只要执行了删除，就尝试重新注册（如果删除失败，重新注册也会报错，交由后续逻辑处理）
if "success: 1" in del_output:
print(f" [清理] 旧数据删除成功，正在重新注册...")
else:
print(f" [警告] 重新删除 ONU 仍失败，回显: {del_output.strip()}")
# 如果删除依然失败，跳过该 ONU 的重新注册
continue
net_connect.send_command_timing(f"interface epon {slot}/{pon}", delay_factor=3)
retry_output = net_connect.send_command_timing(add_cmd, delay_factor=3)
net_connect.send_command_timing("quit", delay_factor=2)
if "success: 1" in retry_output:
print(f" [成功] ONU {mac} 重新注册完成！")
else:
print(f" [失败] 重新注册失败，设备回显：\n{retry_output}")
else:
print(f" [警告] 无法解析 MAC {mac} 的真实位置，跳过该ONU。")
elif "success: 1" in output:
print(f" [成功] ONU {mac} 注册完成！")
elif "% Error" in output or "Error:" in output:
print(f" [警告] 注册失败，设备返回: {output.strip()}")
else:
print(f" [未知状态] 设备未返回明确结果，请检查输出: {output.strip()}")
except Exception as e:
print(f"[未知错误] 发生异常: {e}")
finally:
if net_connect and net_connect.is_alive():
net_connect.disconnect()
print("\n[结束] Telnet 连接已安全关闭。")
if __name__ == "__main__":
# 配置命令行参数解析
parser = argparse.ArgumentParser(description="华为 OLT ONU 自动注册脚本")
parser.add_argument('-H', '--host', required=True, help='OLT 设备的 IP 地址或主机名')
parser.add_argument('-u', '--username', required=True, help='登录用户名')
parser.add_argument('-p', '--password', required=True, help='登录密码')
parser.add_argument('-P', '--port', type=int, default=23, help='Telnet 端口号 (默认: 23)')
args = parser.parse_args()
# 将解析到的参数传递给主函数
test_olt_autoreg(
host=args.host,
username=args.username,
password=args.password,
port=args.port,
)