将电脑接入米家，远程、语音开关机，推送消息

viklion

1.前言
1.1 要求：电脑主板支持网络唤醒，并且需要另一台设备运行服务，理论上可以跑python、24小时运行、能上网在同一局域网内的都行，比如路由器、nas、不用的手机等（我是运行在华硕路由器上）
1.2 门槛不低，并非拿来就能用，需要有点基础，因为要亿点点设置，所以应该也算不上是教程
1.3 搜索了各种实现方法，参考内容写在最后
1.4 并不是python高手，AI指导，欢迎给出各种意见，讨论学习

2.功能
2.1 通过小爱音箱、天猫精灵等语音开关电脑，解放双手(?)，可加入自动化
2.2 通过巴法云的app或微信小程序远程开关电脑
2.3 只要有人开了电脑或关掉电脑，会推送消息

3.实现
3.1 电脑设置：开启网络唤醒，可能需要从bios里设置，记录网卡mac地址
3.2 巴法云中新建TCP创客云虚拟设备，主题名最后以001结尾，昵称可以设置为电脑，记录下私钥和主题名
3.3 米家为例，添加第三方平台设备，选择巴法，同步设备
3.4 设备python环境安装模块

[Python] 纯文本查看 复制代码

?

1	pip3 install wakeonlan pythonping pywinrm

3.5 代码中有4个方法可以自由选择是否启用
3.5.1 ping检测更新电脑状态：用于手动开关电脑后，同步更新巴法云虚拟设备的状态，如果开启消息推送，还会告诉你电脑开机/关机了
3.5.2 关机指令：如果想要语音或远程关机，需要电脑配置好winRM，并填写好参数
3.5.3 日志记录：指定目录内生成日志文件记录
3.5.4 消息推送：用的方糖推送，也就是Server酱（ServerChan），需要填写参数
3.6 填写必要参数，运行python代码
3.7 可以百度最后的参考内容获得更详细的教程

4.代码

[Python] 纯文本查看 复制代码

?

001 002 003 004 005 006 007 008 009 010 011 012 013 014 015 016 017 018 019 020 021 022 023 024 025 026 027 028 029 030 031 032 033 034 035 036 037 038 039 040 041 042 043 044 045 046 047 048 049 050 051 052 053 054 055 056 057 058 059 060 061 062 063 064 065 066 067 068 069 070 071 072 073 074 075 076 077 078 079 080 081 082 083 084 085 086 087 088 089 090 091 092 093 094 095 096 097 098 099 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297

# -*- coding: utf-8 import re import socket import threading import time from datetime import datetime import winrm import requests from pythonping import ping from wakeonlan import send_magic_packet   class PCpower():     '''     PCpower类，已测试：python3.8.18     :param uid: 巴法云用户私钥，必填     :param topic: 巴法云设备主题，必填     :param pc_mac: 电脑mac地址，必填（格式：xx:xx:xx……或xx-xx-xx……）     :param local_ip: 本机ip地址，非电脑ip地址，必填（运行python的设备局域网ip地址，或填写'auto'尝试自动获取）     :param use_ping: 是否开启ping检测更新电脑状态，True为开启，False为关闭     :param use_shutdown: 是否需要关机指令，True为开启，False为关闭     :param pc_ip: 电脑局域网ip地址，非必填，如开启ping或关机指令，则必填     :param pc_account: 电脑登录账户，非必填，如开启关机指令，则必填     :param pc_password: 电脑登录密码，非必填，如开启关机指令，则必填     :param shutdown_time: 延迟关机时间，非必填，如开启关机指令，则必填（单位：秒，立即关机填0）     :param use_write_log: 是否开启日志记录，True为开启，False为关闭     :param log_path: 日志文件路径，非必填，如开启日志记录，则必填     :param use_send_message: 是否开启消息推送[方糖推送]，True为开启，False为关闭     :param url_send_message: [方糖推送]的个人接口，非必填，如开启消息推送，则必填     :param channel_send_message: [方糖推送]的频道，非必填，如开启消息推送，则必填     '''     def __init__(self, uid:str, topic:str, pc_mac:str, local_ip:str, use_ping:bool=False, use_shutdown:bool=False, pc_ip:str ='', pc_account:str='', pc_password:str='', shutdown_time:int=0, use_write_log:bool=False, log_path:str='', use_send_message:bool=False, url_send_message:str='', channel_send_message:str=''):         self.__uid = uid         self.__topic = topic         self.__pc_mac = pc_mac         self.__local_ip = local_ip if not local_ip == 'auto' else self.get_ip_address()         self.__use_ping = use_ping         self.__use_shutdown = use_shutdown         self.__pc_ip = pc_ip         self.__pc_account = pc_account         self.__pc_password = pc_password         self.__shutdown_time = shutdown_time         self.__use_write_log = use_write_log         self.__log_path = log_path         self.__use_send_message = use_send_message         self.__url_send_message = url_send_message         self.__channel_send_message = channel_send_message                   self.__pc_state = None         self.__t_check = None                   self.__check_correct(self.__uid, 'uid')         self.__check_correct(self.__topic, 'topic')         self.__check_correct(self.__pc_mac, 'pc_mac', is_mac=True)         self.__check_correct(self.__local_ip, 'local_ip', is_ip=True)         self.__check_correct(self.__use_ping, 'use_ping', is_bool=True)         self.__check_correct(self.__use_shutdown, 'use_shutdown', is_bool=True)         self.__check_correct(self.__use_write_log, 'use_write_log', is_bool=True)         self.__check_correct(self.__use_send_message, 'use_send_message', is_bool=True)           if self.__use_ping or self.__use_shutdown:             self.__check_correct(self.__pc_ip, 'pc_ip', is_ip=True)         if self.__use_shutdown:             self.__check_correct(self.__pc_account, 'pc_account')             self.__check_correct(self.__pc_password, 'pc_password')             self.__check_correct(self.__shutdown_time, 'shutdown_time', is_int=True)         if self.__use_write_log:             self.__check_correct(self.__log_path, 'log_path')         if self.__use_send_message:             self.__check_correct(self.__url_send_message, 'url_send_message')             self.__check_correct(self.__channel_send_message, 'channel_send_message')                       print("初始化成功")         self.write_log("初始化成功")         if local_ip == 'auto':             print(f"自动获取本机局域网ip地址：{self.__local_ip}")             self.write_log(f"自动获取本机局域网ip地址：{self.__local_ip}")           def __check_correct(self, self_var, var_name, is_mac=False, is_ip=False, is_bool=False, is_int=False):         if is_mac:              if not isinstance(self_var, str) or not self.check_mac_address(self_var):                  raise ValueError(f"{var_name}必须是有效的MAC地址，当前为：{self_var}")          elif is_ip:              if not isinstance(self_var, str) or not self.check_ip_address(self_var):                  raise ValueError(f"{var_name}必须是有效的IP地址，当前为：{self_var}")          elif is_bool:              if not isinstance(self_var, bool):                  raise TypeError(f"{var_name}必须是True或False，当前为：{self_var}，类型：{type(self_var)}")          elif is_int:              if not isinstance(self_var, int) or self_var < 0:                  raise ValueError(f"{var_name}必须是大于等于0的整数，当前为：{self_var}，类型：{type(self_var)}")          else:              if not isinstance(self_var, str) or not self_var.strip():                  raise ValueError(f"{var_name}必须是非空字符串")                @staticmethod     def check_mac_address(mac):         # 定义正则表达式模式         pattern = r'^(([0-9A-Fa-f]{2}:){5}([0-9A-Fa-f]{2})|([0-9A-Fa-f]{2}-){5}([0-9A-Fa-f]{2}))$'         # 利用re模块进行匹配         return bool(re.match(pattern, mac))           @staticmethod     def check_ip_address(ip):         pattern = r'^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'         return bool(re.match(pattern, ip))               @classmethod     def get_time(cls):         # 获取当前时间         time_now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')         return time_now           @classmethod     def get_ip_address(cls):         try:             # 创建一个socket对象             s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)             # 连接到目标网址             s.connect(("192.168.255.255", 80))             # 获取本地IP地址             ip_address = s.getsockname()[0]             # 关闭socket连接             s.close()             return ip_address         except socket.error:             return "无法获取IP地址"       #订阅     def __connTCP(self):         # 创建socket         self.__tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)         # IP和端口         server_ip = 'bemfa.com'         server_port = 8344         try:             # 连接服务器             self.__tcp_client_socket.connect((server_ip, server_port))             # 发送订阅指令             substr = f'cmd=1&uid={self.__uid}&topic={self.__topic}\r\n'             self.__tcp_client_socket.send(substr.encode('utf-8'))             # 写入日志             self.write_log("订阅成功")         except:             self.write_log("订阅失败，正在重试")             time.sleep(2)             self.__connTCP()       #心跳     def __Ping(self):         # 发送心跳         try:             keeplive = 'ping\r\n'             self.__tcp_client_socket.send(keeplive.encode('utf-8'))             self.write_log("已发送心跳")         except:             time.sleep(2)             self.write_log("发送心跳出现问题,重新订阅")             self.__connTCP()         #开启定时，60秒发送一次心跳         t = threading.Timer(60, self.__Ping)         t.start()       #定时ping电脑检测开关机状态     def __check_pc_state(self, is_power_off):         if self.__use_ping:             ping_result = str(ping(self.__pc_ip, timeout = 1, count = 1))  #timeout:超时时间，count:包的个数             self.write_log('\n' + ping_result + '\n' + '---------------------', '_ping_result')             if 'Reply' in ping_result:                 is_power_off = False                 new_state = "电脑已开机"                 message = "开机"                 update_state = "on"             elif 'timed out' in ping_result:                 if not is_power_off:                     self.__check_pc_state(True)                     return                 new_state = "电脑已关机"                 message = "关机"                 update_state = "off"             if self.__pc_state != new_state:                 self.__pc_state = new_state                 substr = f'cmd=2&uid={self.__uid}&topic={self.__topic}/up&msg={update_state}\r\n'                 self.__tcp_client_socket.send(substr.encode("utf-8"))                 self.write_log(f"电脑状态更新为：{message}")                 self.send_message(f"电脑状态更新为：{message}")             #开启定时，120秒ping一次pc             self.__t_check = threading.Timer(120, self.__check_pc_state, args=(is_power_off,))             self.__t_check.start()       #重置ping定时     def __restart_pc_check(self, value):         if self.__use_ping:             try:                 self.__t_check.cancel()                 self.__t_check = threading.Timer(120, self.__check_pc_state, args=(value,))                 self.__t_check.start()             except Exception as e:                 self.write_log(" error: " + str(e))       #推送消息     def send_message(self, content):         if self.__use_send_message:             data = {                     'title': content,                     'desp': self.get_time(),                     'channel': self.__channel_send_message                 }             try:                 _ = requests.post(self.__url_send_message, data=data)             except Exception as e:                 self.write_log("发送消息出错：" + str(e))       #写入日志     def write_log(self, content, nameadd = ''):         if self.__use_write_log:             # 打开日志文件，"a"追加写入             time_day = datetime.now().strftime('%Y-%m-%d')             try:                 with open(rf'{self.__log_path}/{time_day}{nameadd}.log', 'a') as file:                     file.write(self.get_time() + ' ' + content + '\n')             except Exception as e:                 print(self.get_time() + " 写入日志出错了：\n" + str(e))           #网络唤醒     def wake_on_lan(self):         # subprocess.Popen([f'ether-wake', '-i', 'br0', '-b', '{self.__pc_mac}'])         send_magic_packet(self.__pc_mac,  interface = self.__local_ip)         self.__pc_state = "电脑已开机"         self.__restart_pc_check(False)         self.write_log(self.__pc_state)         self.send_message(self.__pc_state)       #收到消息后执行开关机     def pc_power_control(self, state):         if state == 'on' and (self.__use_ping and self.__pc_state != "电脑已开机" or not self.__use_ping):             self.wake_on_lan()         elif state == 'off' and (self.__use_ping and self.__pc_state != "电脑已关机" or not self.__use_ping):             self.__restart_pc_check(True)             if self.__use_shutdown:                 try:                     session = winrm.Session(self.__pc_ip, auth=(self.__pc_account, self.__pc_password))                     _ = session.run_cmd(f'shutdown -s -t {self.__shutdown_time}')                 except Exception as e:                     print(self.get_time() + " 电脑可能已经关机，error: " + str(e))                     self.write_log(str(e) + "电脑可能已经关机")             self.__pc_state = "电脑已关机"             self.write_log(self.__pc_state)             self.send_message(self.__pc_state)       #启动     def start(self):         self.__connTCP()         self.__Ping()         self.__check_pc_state(False)         while True:             try:                 # 接收服务器发送过来的数据                 __recv_Data = self.__tcp_client_socket.recv(1024)                 __recv_str = str(__recv_Data.decode('utf-8'))                 self.write_log(__recv_str)             except Exception as e:                 self.write_log("接收消息出错，可能网络断开，error: " + str(e))                 time.sleep(5)                 continue                           if not __recv_Data:                 self.write_log("服务器未返回任何数据，重新订阅")                 self.__connTCP()                           elif f'uid={self.__uid}' in __recv_str:                 #收到开机消息                 if 'msg=on' in __recv_str:                     self.pc_power_control('on')                 #收到关机消息                 elif 'msg=off' in __recv_str:                     self.pc_power_control('off')   if __name__ == '__main__':     c = PCpower(         #以下填写为示例         uid = 'abcdefghijk'                        #巴法云用户私钥，必填         ,topic = 'mypc001'                     #巴法云设备主题，必填         ,pc_mac = '11:22:33:44:55:66'                    #电脑mac地址，必填（格式：xx:xx:xx……或xx-xx-xx……）         ,local_ip = 'auto'                  #本机ip地址，非电脑ip地址，必填（运行python的设备局域网ip地址，或填写'auto'尝试自动获取）         ,use_ping = False               #是否开启ping检测更新电脑状态，True为开启，False为关闭         ,use_shutdown = False           #是否需要关机指令，True为开启，False为关闭         ,pc_ip = '192.168.1.23'                     #电脑局域网ip地址，非必填，如开启ping或关机指令，则必填         ,pc_account = 'admin'                #电脑登录账户，非必填，如开启关机指令，则必填         ,pc_password = '123456'               #电脑登录密码，非必填，如开启关机指令，则必填         ,shutdown_time = 60             #延迟关机时间，非必填，如开启关机指令，则必填（单位：秒，立即关机填0）         ,use_write_log = False          #是否开启日志记录，True为开启，False为关闭         ,log_path = r'/opt/log'                  #日志文件路径，非必填，如开启日志记录，则必填         ,use_send_message = False       #是否开启消息推送[方糖推送]，True为开启，False为关闭         ,url_send_message = r'https://*******'          #[方糖推送]的个人接口，非必填，如开启消息推送，则必填         ,channel_send_message = '9'      #[方糖推送]的频道，非必填，如开启消息推送，则必填     )     c.start()

5.参考内容
5.1 《用小爱同学控制台式机睡眠和唤醒的思路》，来自知乎，作者：Comzyh
https://zhuanlan.zhihu.com/p/412298207
5.2 《电脑接入米家，控制电脑开关机，并且无需购买米家外设》，来自cnblogs，作者：hackyo
https://www.cnblogs.com/hackyo/p/18000627
5.3 《python 接入，mqtt和tcp》，来自巴法开放论坛
https://bbs.bemfa.com/81
5.4 巴法文档中心
https://cloud.bemfa.com/docs/src/
5.5 Server酱 Key&API
https://sct.ftqq.com/sendkey

viklion

传闻中的喜哥哥 发表于 2024-6-5 21:23
可以搞成docker吗？

12月31号前发布一版docker

Nemo92

如果单纯只是语音开电脑的话，直接买个接入米家或者天猫精灵的智能插座（便宜的第三方十几块钱），然后电脑主板打开通电开机就行了（有些旧电脑不支持），非局域网也能控制，开机之后的功能用远程控制软件

淼先森

aigo891 发表于 2024-5-8 17:03
只能开关机，能像控制智能音箱那样就好了，想放音乐就放音乐，想查什么就查什么

可以搭配神秘鸭试下，我现在是这样弄的

ZhjhJZ

牛，家庭设备智能化了

redapple2015

这个功能强大，现在全设备向智能化发展。

心伤的天堂

这个不错

dingqh

666666   将电脑接入米家  这你是怎么想到的，太神奇了...

aigo891

只能开关机，能像控制智能音箱那样就好了，想放音乐就放音乐，想查什么就查什么

squallzcy

确实神奇，但是我一开始理解错了，哈哈哈哈，感谢大佬无私分享

海是倒过来的天

手残党在网上买了个几十的开机卡解决的

fast001

先收藏后续有用