DatacomShell

面向网络设备的 Python 远程管理工具库，统一封装 SSH 与 Telnet 协议交互。

功能特性

• SSH 客户端 (SyncSSHClientUtil) — 基于 paramiko，使用交互式 shell 适配网络设备（交换机/路由器等）
• Telnet 客户端 (AsyncTelnetClientUtil) — 基于 telnetlib3，异步底层，对外暴露统一同步接口
• 统一抽象接口 (AbsClient) — 两种协议使用完全一致的方法调用风格
• Builder 参数构建 — 所有配置项通过链式 Builder 构建，清晰可维护
• 会话日志记录 — 自动按时间窗口切分日志文件，支持 [start] / [end] 标记
• 多种认证方式 — 支持无认证、用户名/密码认证

安装

本项目使用 uv 作为包管理器。

# 克隆仓库
git clone <repository-url>
cd DatacomShell

# 安装依赖
uv sync

# 激活虚拟环境（Windows）
.venv\Scripts\activate

快速开始

SSH 连接示例

from datacomshell.SyncSSHClientUtil import SyncSSHClientUtil, SyncSSHClientParam

param = SyncSSHClientParam.builder()\
    .set_host('192.168.1.1')\
    .set_port(22)\
    .set_username('admin')\
    .set_password('Admin@123')\
    .set_timeout(15.0)\
    .set_log_folder_path('./ssh_logs')\
    .build()

client = SyncSSHClientUtil(param)

try:
    client.connect()
    result = client.execute('display version')
    print(result)
finally:
    client.close()

Telnet 连接示例

from datacomshell.AsyncTelnetClientUtil import AsyncTelnetClientUtil, AsyncTelnetClientParam
from datacomshell.Auth import UsernamePasswordAuth

auth = UsernamePasswordAuth('admin', 'Admin@123')

param = AsyncTelnetClientParam.builder()\
    .set_host('192.168.1.1')\
    .set_port(23)\
    .set_auth(auth)\
    .set_connect_timeout(10.0)\
    .set_login_timeout(10.0)\
    .set_command_timeout(15.0)\
    .set_log_folder_path('./telnet_logs')\
    .build()

client = AsyncTelnetClientUtil(param)

try:
    client.connect()
    result = client.execute('display version')
    print(result)
finally:
    client.close()

统一接口 — 批量管理多设备

from datacomshell.abs_client import AbsClient

def run_on_device(client: AbsClient, device_name: str):
    try:
        client.connect()
        result = client.execute('display version')
        print(f"[{device_name}] 执行成功")
        return result
    except Exception as e:
        print(f"[{device_name}] 执行失败: {e}")
    finally:
        client.close()

# SSH 设备
ssh_param = SyncSSHClientParam.builder()\
    .set_host('10.0.0.1').set_username('admin').set_password('Admin@123').build()
run_on_device(SyncSSHClientUtil(ssh_param), 'SSH-Switch-01')

# Telnet 设备
auth = UsernamePasswordAuth('admin', 'Admin@123')
telnet_param = AsyncTelnetClientParam.builder()\
    .set_host('10.0.0.2').set_auth(auth).build()
run_on_device(AsyncTelnetClientUtil(telnet_param), 'Telnet-Switch-01')

项目架构

┌─────────────────────────────────────────────┐
│              应用层 (App Layer)              │
│            运维脚本 / CLI 入口                │
├─────────────────────────────────────────────┤
│              抽象层 (AbsClient)              │
│     connect / execute / execute_bytes / close │
├─────────────────────────────────────────────┤
│              实现层 (Impl Layer)              │
│  ┌──────────────────┐  ┌──────────────────┐ │
│  │ SyncSSHClientUtil │  │AsyncTelnetClientUtil│
│  │   (paramiko)      │  │   (telnetlib3)   │ │
│  └──────────────────┘  └──────────────────┘ │
├─────────────────────────────────────────────┤
│              支撑层 (Support Layer)           │
│  Auth (认证)    SessionLogger (日志)          │
└─────────────────────────────────────────────┘

API 说明

统一接口 AbsClient

Builder 参数链

两个客户端参数均支持链式 Builder：

SyncSSHClientParam.builder()\
    .set_host(str)\
    .set_port(int)\
    .set_username(str)\
    .set_password(str)\
    .set_timeout(float)\
    .set_log_folder_path(Optional[str])\
    .set_log_new_second(float)\
    .build()

AsyncTelnetClientParam.builder()\
    .set_host(str)\
    .set_port(int)\
    .set_auth(Auth)\
    .set_connect_timeout(float)\
    .set_login_timeout(float)\
    .set_command_timeout(float)\
    .set_log_folder_path(Optional[str])\
    .set_log_new_second(float)\
    .build()

认证方式

from datacomshell.Auth import Auth, AuthState, UsernamePasswordAuth

# 无认证
no_auth = Auth()

# 用户名密码认证
user_auth = UsernamePasswordAuth('admin', 'Admin@123')

日志功能

当配置了 log_folder_path 后，客户端会自动记录每次 execute 的输入输出：

• 自动目录创建：日志目录不存在时自动创建
• 时间窗口切分：超过 log_new_second（默认 1800 秒）自动创建新日志文件
• 起止标记：每个日志文件包含 [start] 和 [end] 时间戳标记
• 空行过滤：自动过滤纯空白行

日志示例：

================ [start]:2026-07-22_14-30-00-123456.log ================
display version
Huawei Versatile Routing Platform Software
VRP (R) software, Version 5.110 (S5700 V200R001C00)
...
================ [end]:2026-07-22_14-30-05-789012 ================

目录结构

DatacomShell/
├── src/
│   ├── datacomshell/
│   │   ├── __init__.py              # 包入口
│   │   ├── abs_client.py            # 抽象客户端基类
│   │   ├── Auth.py                  # 认证模块
│   │   ├── AsyncTelnetClientUtil.py # Telnet 客户端
│   │   ├── SyncSSHClientUtil.py     # SSH 客户端
│   │   └── SessionLogger.py         # 日志管理器
│   └── test/
│       ├── test.py                  # 交互式测试菜单
│       └── example.py               # 使用示例
├── pyproject.toml                   # 项目配置与依赖
├── uv.lock                          # uv 锁定文件
└── README.md                        # 本文档

依赖

注意事项

• 网络设备 SSH：大多数交换机/路由器不支持 exec_command，本库使用 invoke_shell 交互式 shell 发送命令。
• Telnet 结束标记：命令执行后会追加 [saven-process-end] 标记，通过匹配该标记确认命令执行完成。
• 超时处理：Telnet 的 login_timeout 控制认证阶段超时，command_timeout 控制命令执行超时。

作者

Saven — 2416844857@qq.com

License

MIT License