设备密钥认证

# Windows PowerShell
python -m pip install "paho-mqtt>=2.1,<3.0"

# Linux && macOS Shell
python3 -m pip install "paho-mqtt>=2.1,<3.0"

# Windows PowerShell
python -m pip show paho-mqtt

# Linux && macOS Shell
python3 -m pip show paho-mqtt

# 输出中的 Version 应为 2.1.0。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import getpass
import sys
from importlib.metadata import PackageNotFoundError, version

try:
    import paho.mqtt.client as mqtt
except ImportError:
    mqtt = None


def required_input(prompt):
    """读取必填参数并清除首尾空格。"""
    value = input(prompt).strip()
    if not value:
        raise ValueError("输入内容不能为空")
    return value


def get_port():
    """读取并校验MQTT端口号。"""
    port_text = input(
        "端口号（直接回车使用1883）："
    ).strip()

    if not port_text:
        return 1883

    try:
        port = int(port_text)
    except ValueError as exc:
        raise ValueError("端口号必须是整数") from exc

    if not 1 <= port <= 65535:
        raise ValueError("端口号必须在1～65535范围内")

    return port


def check_paho_version():
    """检查Paho MQTT是否安装且支持Callback API VERSION2。"""
    install_command = (
        f'"{sys.executable}" -m pip install "paho-mqtt==2.1.0"'
    )

    if mqtt is None:
        raise ImportError(
            "未安装Paho MQTT客户端。\\n"
            f"请使用当前Python环境执行：\\n{install_command}"
        )

    try:
        installed_version = version("paho-mqtt")
    except PackageNotFoundError:
        installed_version = "未知"

    if not hasattr(mqtt, "CallbackAPIVersion"):
        raise RuntimeError(
            f"当前Paho MQTT版本为{installed_version}，"
            "不支持Callback API VERSION2。\\n"
            f"请使用当前Python环境执行：\\n{install_command}"
        )

    print(f"Paho MQTT版本：{installed_version}")


def main():
    check_paho_version()

    print("\\n请输入设备详情页中的MQTT连接参数。")
    print("输入内容仅在本次程序运行期间使用，不会保存到文件。\\n")

    broker = required_input("MQTT服务器地址：")

    if "://" in broker:
        raise ValueError(
            "MQTT服务器地址不能包含协议前缀，"
            "请勿输入tcp://或mqtt://"
        )

    if any(char in broker for char in "${}"):
        raise ValueError(
            "MQTT服务器地址不能包含${}，"
            "请将${ProductID}替换为真实ProductID"
        )

    port = get_port()
    client_id = required_input("ClientId：")
    username = required_input("Username：")

    # Password输入时不会显示在终端中。
    password = getpass.getpass(
        "Password（输入内容不会显示）："
    ).strip()

    if not password:
        raise ValueError("Password不能为空")

    def on_connect(
        client,
        userdata,
        connect_flags,
        reason_code,
        properties,
    ):
        if reason_code == 0:
            print("\\n认证成功，设备已连接物联网开发平台。")
            print(f"MQTT服务器：{broker}")
            print(f"端口号：{port}")
            print("请前往控制台查看设备在线状态和上下线日志。")
        else:
            print(
                f"\\nMQTT连接失败，返回码：{reason_code}。"
                "\\n请检查ClientId、Username、Password及设备状态。"
            )
            client.disconnect()

    def on_connect_fail(client, userdata):
        print(
            "\\n网络连接失败，请检查以下内容："
            f"\\n- MQTT服务器地址：{broker}"
            f"\\n- 端口号：{port}"
            "\\n- DNS解析及网络访问状态"
        )

    def on_disconnect(
        client,
        userdata,
        disconnect_flags,
        reason_code,
        properties,
    ):
        if reason_code != 0:
            print(f"\\n连接异常断开，返回码：{reason_code}")

    client = mqtt.Client(
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        client_id=client_id,
        clean_session=True,
        protocol=mqtt.MQTTv311,
    )

    client.username_pw_set(
        username=username,
        password=password,
    )

    client.on_connect = on_connect
    client.on_connect_fail = on_connect_fail
    client.on_disconnect = on_disconnect

    # 不输出Username和Password等认证信息。
    print("\\n连接配置：")
    print(f"MQTT服务器：{broker}")
    print(f"端口号：{port}")
    print(f"ClientId：{client_id}")
    print("\\n正在建立MQTT连接。")
    print("程序将保持连接，按Ctrl+C退出。")

    client.connect(
        host=broker,
        port=port,
        keepalive=60,
    )

    try:
        client.loop_forever()
    finally:
        client.disconnect()


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\\n用户终止连接。")
        sys.exit(0)
    except Exception as exc:
        print(f"\\n运行失败：{exc}")
        sys.exit(1)

# Windows PowerShell
python mqtt_connect.py

# Linux && macOS Shell
python3 mqtt_connect.py

# 运行成功后依次输入
MQTT服务器地址：
端口号（直接回车使用1883）：
ClientId：
Username：
Password（输入内容不会显示）：

正在连接：******.iotcloud.tencentdevices.com:1883
ClientId：******
程序将保持连接，按Ctrl+C退出。
认证成功，设备已连接物联网开发平台。
请前往控制台查看设备在线状态。