H3ConuMS-v2/guide.md

为 H3ConuMS2 添加 ONU 远程重启 & 光功率查询功能 — 后端开发指南

目标：将 H3C iMC 平台的 ONU 远程重启和光功率查询功能集成到 H3ConuMS2 项目中 技术栈：FastAPI + SQLAlchemy + requests (HTTP Digest Auth) 源项目参考：/home/v6ole/pyproject/H3ConuMS

---

1. 整体架构

┌──────────────────┐      REST API 调用       ┌──────────────────┐
│   H3ConuMS2 后端  │  ──────────────────────►  │   H3C iMC 平台   │
│   (FastAPI)       │  ◄──────────────────────  │                  │
│   IMCService      │                           │  /imcrs/epon/... │
└──────────────────┘                           └──────────────────┘

后端提供的 API 端点：

端点	方法	说明	iMC 后端接口
/api/devices/{id}/reboot	POST	远程重启 ONU	/imcrs/epon/onu/reboot?mac={mac} (POST)
/api/devices/{id}/optical-power	GET	获取 ONU 光功率	/imcrs/epon/onu/onuLightWaneInfo?mac={mac} (GET)

数据流（以重启为例）：

1. 客户端 POST /api/devices/{id}/reboot
2. 后端控制器验证设备存在 + 权限 → 调用 IMCService.reboot_onu()
3. IMCService 构建 HTTP Digest 认证头 → POST 到 iMC REST API
4. iMC 返回结果 → 后端解析错误码 → 返回 JSON

---

2. Digest 认证原理解析

iMC 的 REST API 使用 HTTP Digest Access Authentication（RFC 2617），不是普通的 Cookie/Session 登录。

认证流程

客户端                          iMC 服务器
  │                                │
  │──── GET /imcrs/... (无认证) ────│
  │                                │──── 401 + WWW-Authenticate header
  │                                │    (包含 nonce, realm, qop)
  │                                │
  │  ── 解析 WWW-Authenticate ──►  │
  │     提取 nonce 和 realm        │
  │                                │
  │  ── 计算 Digest 响应 ────────►  │
  │     HA1 = MD5(user:realm:pass) │
  │     HA2 = MD5(method:uri)      │
  │     response = MD5(HA1:nonce:nc:cnonce:qop:HA2) │
  │                                │
  │──── POST /imcrs/... ──────────►│
  │    Authorization: Digest ...   │
  │                                │──── 200 OK (成功)

核心 MD5 计算

cnonce = md5(str(time.time())).hexdigest()[:16]
ha1 = md5(f"{username}:{realm}:{password}").hexdigest()
ha2 = md5(f"{method}:{uri}").hexdigest()
response = md5(f"{ha1}:{nonce}:{nc:08d}:{cnonce}:auth:{ha2}").hexdigest()

---

3. 需要修改/新增的文件清单

文件	操作	说明
backend/app/services/imc_service.py	新增	iMC API 服务（Digest 认证 + 重启 + 光功率）
backend/app/services/__init__.py	修改	导出 IMCService
backend/app/schemas/device.py	修改	添加重启/光功率响应 Schema
backend/app/api/v1/devices.py	修改	添加重启和光功率 API 路由
backend/app/core/config.py	修改	添加 iMC 配置项
.env 或 backend/.env	修改	添加 iMC 环境变量

---

4. 后端实现

4.1 配置项 — backend/app/core/config.py

在 Settings 类中添加 iMC 相关配置：

# ===== iMC API 配置（用于 ONU 远程重启和光功率查询）=====
IMC_API_URL: str = ""          # 例如 https://172.16.1.252:8443
IMC_API_USERNAME: str = ""     # iMC 用户名
IMC_API_PASSWORD: str = ""     # iMC 密码（明文，Digest认证需要原始密码）
IMC_API_VERIFY_SSL: bool = False  # 是否验证 SSL 证书
IMC_CONNECT_TIMEOUT: float = 5.0
IMC_READ_TIMEOUT: float = 20.0

4.2 .env 配置

在 backend/.env（或项目根目录 .env）中添加：

# iMC API 配置（用于 ONU 远程重启和光功率查询）
IMC_API_URL=https://172.16.1.252:8443
IMC_API_USERNAME=admin
IMC_API_PASSWORD=Pwd@12345
IMC_API_VERIFY_SSL=false
IMC_CONNECT_TIMEOUT=5
IMC_READ_TIMEOUT=20

4.3 IMCService — backend/app/services/imc_service.py

完整代码，包含 Digest 认证 + 重启 ONU + 光功率查询三大功能：

"""
iMC REST API 服务
- 使用 HTTP Digest Access Authentication (RFC 2617)
- 支持 nonce 过期自动续约（401 时自动重新握手）
- 功能：ONU 远程重启、光功率查询
"""
import hashlib
import re
import json
import time
import logging
import requests
from app.core.config import settings

logger = logging.getLogger(__name__)

# iMC 重启错误码映射
REBOOT_ERROR_CODES = {
    '103': 'ONU不存在',
    '119': 'SNMP连接超时',
    '120': '业务割接失败',
    '121': 'ONU未运行',
    '122': '重启失败',
}


class IMCService:
    """iMC REST API 服务封装"""

    def __init__(self):
        self.base_url = settings.IMC_API_URL.rstrip('/')
        self.username = settings.IMC_API_USERNAME
        self.password = settings.IMC_API_PASSWORD
        self.verify_ssl = settings.IMC_API_VERIFY_SSL
        self.session = requests.Session()
        self.realm = "iMC RESTful Web Services"
        self.connect_timeout = getattr(settings, 'IMC_CONNECT_TIMEOUT', 5)
        self.read_timeout = getattr(settings, 'IMC_READ_TIMEOUT', 20)
        # Digest 认证状态（每次重新初始化时清空，让首次请求自动获取 nonce）
        self.nonce = None
        self.nc = 1

    # ═══════════════════════════════════════════════
    # 内部：Digest 认证
    # ═══════════════════════════════════════════════

    def _get_digest_auth_header(self, method: str, uri: str) -> str | None:
        """
        构建 HTTP Digest 认证头

        首次调用时会自动发一个请求获取 nonce（服务器返回 401 + WWW-Authenticate），
        后续复用 nonce 并递增 nc 值。
        nonce 过期时调用方捕获 401 后清空 self.nonce，下次自动重新握手。
        """
        if not self.nonce:
            try:
                resp = self.session.get(
                    f"{self.base_url}{uri}",
                    verify=self.verify_ssl,
                    headers={"Accept": "application/json"},
                    timeout=(self.connect_timeout, self.read_timeout),
                )
                if resp.status_code == 401 and 'WWW-Authenticate' in resp.headers:
                    auth_header = resp.headers['WWW-Authenticate']
                    auth_parts = {}
                    for part in auth_header.split(','):
                        if '=' in part:
                            key, value = part.split('=', 1)
                            auth_parts[key.strip()] = value.strip(' "')
                    self.nonce = auth_parts.get('nonce', '')
                    self.realm = auth_parts.get('realm', self.realm)
                    logger.info(f"获取 nonce 成功: {self.nonce}")
                else:
                    logger.error(f"获取 nonce 失败, 状态码: {resp.status_code}")
                    return None
            except requests.Timeout:
                logger.error("获取 nonce 超时")
                raise TimeoutError("iMC 认证超时")
            except Exception as e:
                logger.error(f"获取 nonce 异常: {e}")
                return None

        # 计算 Digest 响应
        cnonce = hashlib.md5(str(time.time()).encode()).hexdigest()[:16]
        ha1 = hashlib.md5(
            f"{self.username}:{self.realm}:{self.password}".encode()
        ).hexdigest()
        ha2 = hashlib.md5(f"{method}:{uri}".encode()).hexdigest()
        response_hash = hashlib.md5(
            f"{ha1}:{self.nonce}:{self.nc:08d}:{cnonce}:auth:{ha2}".encode()
        ).hexdigest()

        auth_value = (
            f'Digest username="{self.username}", '
            f'realm="{self.realm}", '
            f'nonce="{self.nonce}", '
            f'uri="{uri}", '
            f'response="{response_hash}", '
            f'qop=auth, '
            f'nc={self.nc:08d}, '
            f'cnonce="{cnonce}"'
        )
        self.nc += 1
        return auth_value

    def _clear_auth(self):
        """清除认证状态（nonce 过期时调用）"""
        self.nonce = None
        self.nc = 1

    # ═══════════════════════════════════════════════
    # 公共：重启 ONU
    # ═══════════════════════════════════════════════

    def reboot_onu(self, mac: str) -> dict:
        """
        远程重启 ONU 设备

        Args:
            mac: MAC 地址，格式如 "1484-7790-4840"

        Returns:
            {"success": True, "message": "设备正在重启，请稍后..."}
            或 {"success": False, "message": "重启失败: ..."}
        """
        max_retries = 1
        for retry in range(max_retries + 1):
            try:
                uri = f"/imcrs/epon/onu/reboot?mac={mac}"
                auth = self._get_digest_auth_header("POST", uri)
                if not auth:
                    return {"success": False, "message": "认证失败，无法发送重启请求"}

                headers = {
                    "Accept": "application/xml",
                    "Content-Type": "application/xml",
                    "Content-Length": "0",
                    "Authorization": auth,
                }

                resp = self.session.post(
                    f"{self.base_url}{uri}",
                    headers=headers,
                    verify=self.verify_ssl,
                    timeout=(self.connect_timeout, self.read_timeout),
                )

                if resp.status_code == 200:
                    # 检查 XML 响应中是否有错误码
                    if "<errorCode>" in resp.text:
                        m = re.search(r"<errorCode>(\d+)</errorCode>", resp.text)
                        if m:
                            code = m.group(1)
                            msg = REBOOT_ERROR_CODES.get(
                                code, f"未知错误(代码: {code})"
                            )
                            return {"success": False, "message": f"重启失败: {msg}"}
                    return {"success": True, "message": "设备正在重启，请稍后..."}

                elif resp.status_code == 401:
                    # nonce 过期，清空后重试
                    self._clear_auth()
                    continue
                else:
                    return {
                        "success": False,
                        "message": f"重启请求失败(HTTP {resp.status_code})",
                    }

            except TimeoutError:
                return {"success": False, "message": "iMC 接口超时，请稍后重试"}
            except Exception as e:
                logger.error(f"重启异常: {e}")
                if retry < max_retries:
                    time.sleep(3)
                    continue
                return {"success": False, "message": f"重启异常: {e}"}

        return {"success": False, "message": "重启失败，已达最大重试次数"}

    # ═══════════════════════════════════════════════
    # 公共：获取光功率
    # ═══════════════════════════════════════════════

    def get_optical_power(self, mac: str) -> dict | None:
        """
        获取 ONU 设备光功率信息

        接口: /imcrs/epon/onu/onuLightWaneInfo?mac={mac}
        响应 JSON 字段：powerIn(接收光功率), powerOut(发送光功率),
                       bindMac, devId, eponDevName, oltIfName, onuIfDesc

        Args:
            mac: MAC 地址，格式如 "1484-7790-4840"

        Returns:
            dict: {
                "powerIn": "-18.5",   # dBm，接收光功率
                "powerOut": "2.3",    # dBm，发送光功率
                "bindMac": "...",
                "devId": ...,
                "eponDevName": "...",
                "oltIfName": "...",
                "onuIfDesc": "..."
            }
            或 None（失败时）
        """
        max_retries = 1
        for retry in range(max_retries + 1):
            try:
                uri = f"/imcrs/epon/onu/onuLightWaneInfo?mac={mac}"
                auth = self._get_digest_auth_header("GET", uri)
                if not auth:
                    logger.error("生成认证头失败，无法获取光功率")
                    return None

                headers = {
                    "Accept": "application/json",
                    "Content-Type": "application/json",
                    "Authorization": auth,
                }

                logger.info(f"获取光功率: {self.base_url}{uri}")
                resp = self.session.get(
                    f"{self.base_url}{uri}",
                    headers=headers,
                    verify=self.verify_ssl,
                    timeout=(self.connect_timeout, self.read_timeout),
                )
                logger.info(f"光功率API响应状态码: {resp.status_code}")

                if resp.status_code == 200:
                    try:
                        data = resp.json()
                        logger.info(
                            f"光功率响应: {json.dumps(data, ensure_ascii=False)}"
                        )
                        return {
                            "powerIn": data.get("powerIn"),
                            "powerOut": data.get("powerOut"),
                            "bindMac": data.get("bindMac"),
                            "devId": data.get("devId"),
                            "eponDevName": data.get("eponDevName"),
                            "oltIfName": data.get("oltIfName"),
                            "onuIfDesc": data.get("onuIfDesc"),
                        }
                    except json.JSONDecodeError as e:
                        logger.error(f"解析光功率 JSON 失败: {e}, 内容: {resp.text}")

                elif resp.status_code == 401:
                    self._clear_auth()
                    continue
                else:
                    logger.error(
                        f"光功率API请求失败, 状态码: {resp.status_code}, "
                        f"内容: {resp.text}"
                    )
                break  # 非401不重试

            except TimeoutError:
                logger.error("获取光功率超时")
                raise
            except requests.Timeout:
                logger.error("光功率接口请求超时")
                raise TimeoutError("iMC 光功率接口请求超时，请稍后重试")
            except Exception as e:
                logger.error(f"获取光功率异常: {e}")
                if retry < max_retries:
                    time.sleep(3)
                    continue
        return None

4.4 Schema — backend/app/schemas/device.py

添加重启和光功率的响应模型：

class RebootResponse(BaseModel):
    success: bool
    message: str


class OpticalPowerResponse(BaseModel):
    power_in: Optional[str] = None      # 接收光功率 (dBm)
    power_out: Optional[str] = None     # 发送光功率 (dBm)
    bind_mac: Optional[str] = None
    dev_id: Optional[int] = None
    epon_dev_name: Optional[str] = None
    olt_if_name: Optional[str] = None
    onu_if_desc: Optional[str] = None

4.5 API 路由 — backend/app/api/v1/devices.py

在文件顶部导入新 Schema：

from app.schemas.device import (
    DeviceListResponse,
    ONUDeviceResponse,
    RebootResponse,
    OpticalPowerResponse,
)

在文件末尾添加两个新端点：

# ═══════════════════════════════════════════════
# 重启 ONU
# ═══════════════════════════════════════════════

@router.post("/{device_id}/reboot", response_model=RebootResponse)
def reboot_device(
    device_id: int,
    db: Session = Depends(get_db),
    current: dict = Depends(require_permission('device.check')),
):
    """
    远程重启 ONU 设备（通过 iMC REST API）

    权限要求：device.check
    区域/学校管理员只能操作自己范围内的设备。
    """
    device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
    if not device:
        raise HTTPException(status_code=404, detail="设备不存在")

    # 数据范围权限过滤
    role = current.get('role', 'user')
    if role == 'area_admin':
        assigned = current.get('assigned_area') or ''
        areas = [a.strip() for a in assigned.split(',') if a.strip()]
        if device.region not in areas:
            raise HTTPException(status_code=403, detail="无权限操作此区域的设备")
    elif role == 'school_admin':
        assigned = current.get('assigned_school') or ''
        schools = [s.strip() for s in assigned.split(',') if s.strip()]
        if device.school_name not in schools:
            raise HTTPException(status_code=403, detail="无权限操作此学校的设备")

    try:
        from app.services.imc_service import IMCService
        service = IMCService()
        mac = device.mac_address
        result = service.reboot_onu(mac)
        return RebootResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"重启失败: {str(e)}")


# ═══════════════════════════════════════════════
# 获取光功率
# ═══════════════════════════════════════════════

@router.get("/{device_id}/optical-power", response_model=OpticalPowerResponse)
def get_device_optical_power(
    device_id: int,
    db: Session = Depends(get_db),
    current: dict = Depends(require_permission('device.view')),
):
    """
    获取 ONU 设备光功率信息（通过 iMC REST API）

    返回接收光功率(power_in)和发送光功率(power_out)，单位 dBm。
    权限要求：device.view（只读操作）
    """
    device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
    if not device:
        raise HTTPException(status_code=404, detail="设备不存在")

    # 数据范围权限过滤（同上）
    role = current.get('role', 'user')
    if role == 'area_admin':
        assigned = current.get('assigned_area') or ''
        areas = [a.strip() for a in assigned.split(',') if a.strip()]
        if device.region not in areas:
            raise HTTPException(status_code=403, detail="无权限操作此区域的设备")
    elif role == 'school_admin':
        assigned = current.get('assigned_school') or ''
        schools = [s.strip() for s in assigned.split(',') if s.strip()]
        if device.school_name not in schools:
            raise HTTPException(status_code=403, detail="无权限操作此学校的设备")

    try:
        from app.services.imc_service import IMCService
        service = IMCService()
        mac = device.mac_address
        result = service.get_optical_power(mac)
        if result is None:
            raise HTTPException(status_code=502, detail="获取光功率失败，iMC 接口无响应")

        # 字段名转换：下划线转驼峰前先映射
        from app.schemas.device import OpticalPowerResponse
        return OpticalPowerResponse(
            power_in=result.get("powerIn"),
            power_out=result.get("powerOut"),
            bind_mac=result.get("bindMac"),
            dev_id=result.get("devId"),
            epon_dev_name=result.get("eponDevName"),
            olt_if_name=result.get("oltIfName"),
            onu_if_desc=result.get("onuIfDesc"),
        )
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取光功率失败: {str(e)}")

4.6 注册 Service — backend/app/services/__init__.py

from .imc_service import IMCService

__all__ = ["IMCService"]

---

5. 关键陷阱与注意事项

⚠️ Digest 认证的 nonce 过期问题

iMC 的 nonce 有有效期（通常 5-10 分钟）。过期后服务器返回 401。

• 代码中 _clear_auth() 清空 nonce，下次请求自动重新握手
• 重启和光功率方法都在 for 循环中捕获 401 并 continue 重试

⚠️ MAC 地址格式

iMC REST API 要求 MAC 地址格式为 1484-7790-4840（连字符分隔，大写十六进制）。 如果数据库存储格式不同，需要做格式转换：

def normalize_mac(mac: str) -> str:
    """标准化 MAC 为 iMC 要求的格式：1484-7790-4840"""
    clean = mac.replace(':', '').replace('-', '').replace('.', '').upper()
    return f"{clean[0:4]}-{clean[4:8]}-{clean[8:12]}"

⚠️ 重启接口需要 Content-Length: 0

即使请求体为空，也必须显式设置 Content-Length: 0 头，否则 iMC 会报错。

⚠️ 光功率接口返回空数据的情况

当 ONU 离线或光模块故障时，iMC 返回的 powerIn / powerOut 可能是 " --"（两个空格+两个横线）或 None。前端需做占位符处理。

⚠️ 重启操作较慢

从发起请求到设备实际重启完成约需 30-60 秒（取决于 SNMP 响应）。建议：

• 前端按钮显示 loading 状态
• 后端设置合理超时（connect=5s, read=20s）
• 不要在短时间内对同一设备重复操作

⚠️ 并发控制

建议对重启操作添加简单的并发控制，避免同一设备被多次重启：

import threading

_reboot_locks = {}
_reboot_lock = threading.Lock()

def reboot_onu(self, mac):
    with _reboot_lock:
        if mac not in _reboot_locks:
            _reboot_locks[mac] = threading.Lock()
    lock = _reboot_locks[mac]

    if not lock.acquire(blocking=False):
        return {"success": False, "message": "该设备正在重启中，请稍后"}
    try:
        # ... 重启逻辑 ...
    finally:
        lock.release()

⚠️ Docker 部署注意

• 在 docker-compose.yml 的 backend 服务中新增环境变量：

  environment:
    - IMC_API_URL=https://172.16.1.252:8443
    - IMC_API_USERNAME=admin
    - IMC_API_PASSWORD=Pwd@12345
    - IMC_API_VERIFY_SSL=false
  

• backend 和 celery-worker 容器都需要这些变量
• 修改后必须重新构建镜像：

  docker compose build --no-cache backend
  docker compose rm -f backend && docker compose up -d backend
  

---

6. 测试验证

手动测试

# 1. 重启设备
curl -X POST "http://localhost:8000/api/devices/1/reboot" \
  -H "Authorization: Bearer <token>"

# 2. 获取光功率
curl "http://localhost:8000/api/devices/1/optical-power" \
  -H "Authorization: Bearer <token>"

# 3. 查看后端日志
docker compose logs backend | grep IMCService

# 4. 直接测试 iMC API（验证认证是否工作）
curl -k -v "https://172.16.1.252:8443/imcrs/epon/onu/onuLightWaneInfo?mac=1484-7790-4840"

测试响应示例

重启成功：

{"success": true, "message": "设备正在重启，请稍后..."}

重启失败（ONU不存在）：

{"success": false, "message": "重启失败: ONU不存在"}

光功率获取成功：

{
  "power_in": "-18.5",
  "power_out": "2.3",
  "bind_mac": "1484-7790-4840",
  "dev_id": 123,
  "epon_dev_name": "OLT-1-1",
  "olt_if_name": "1/0/2",
  "onu_if_desc": "ONU-学校A"
}

光功率获取失败（设备离线）：

{
  "power_in": null,
  "power_out": null,
  "bind_mac": null,
  "dev_id": null,
  "epon_dev_name": null,
  "olt_if_name": null,
  "onu_if_desc": null
}

---

7. 完整调用时序图

客户端                         后端 FastAPI                      iMC 平台
 │                                │                                │
 │  POST /api/devices/1/reboot    │                                │
 │ ────────────────────────────►  │                                │
 │                                │  ── 查数据库：设备存在？──►     │
 │                                │  ◄── 返回设备信息 ──────────    │
 │                                │  ── 权限检查 ─────────────     │
 │                                │                                │
 │                                │  ── GET /imcrs/epon/onu/reboot │
 │                                │     (无认证，获取 nonce)       │
 │                                │ ────────────────────────────►  │
 │                                │  ◄── 401 + WWW-Authenticate  ──│
 │                                │     nonce=xxx, realm=...       │
 │                                │                                │
 │                                │  ── POST 同 URI + Digest ────► │
 │                                │     Authorization: Digest ...  │
 │                                │  ◄── 200 OK (XML) ────────────│
 │                                │                                │
 │  ◄── {success: true,           │                                │
 │        message: "设备重启中"}   │                                │
 │                                │                                │
 │  ── 或 ──                      │                                │
 │                                │                                │
 │  GET /api/devices/1/optical-power                               │
 │ ────────────────────────────►  │                                │
 │                                │  ── 查 + 权限 (同上) ────      │
 │                                │                                │
 │                                │  ── GET /imcrs/epon/onu/       │
 │                                │     onuLightWaneInfo?mac=...   │
 │                                │     (+ Digest Auth)            │
 │                                │ ────────────────────────────►  │
 │                                │  ◄── 200 OK (JSON) ───────────│
 │                                │     {powerIn, powerOut, ...}   │
 │                                │                                │
 │  ◄── {power_in: "-18.5",       │                                │
 │        power_out: "2.3", ...}  │                                │

---

附录：源项目参考文件位置

内容	路径
IMCService 完整实现	/home/v6ole/pyproject/H3ConuMS/app/services/imc_service.py
重启控制器	/home/v6ole/pyproject/H3ConuMS/app/controllers/device.py (第1059行)
优化版控制器	/home/v6ole/pyproject/H3ConuMS/app/controllers/optimized_device.py (第157行)
iMC 配置项	/home/v6ole/pyproject/H3ConuMS/app/config.py (第61-67行)