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环境安装模块
pip3 install wakeonlan pythonping pywinrm3.5 代码中有4个方法可以自由选择是否启用
3.5.1 ping检测更新电脑状态：用于手动开关电脑后，同步更新巴法云虚拟设备的状态，如果开启消息推送，还会告诉你电脑开机/关机了
3.5.2 关机指令：如果想要语音或远程关机，需要电脑配置好winRM，并填写好参数
3.5.3 日志记录：指定目录内生成日志文件记录
3.5.4 消息推送：用的方糖推送，也就是Server酱（ServerChan），需要填写参数
3.6 填写必要参数，运行python代码
3.7 可以百度最后的参考内容获得更详细的教程
4.代码
# -*- 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()
#开启定时，30秒发送一次心跳
t = threading.Timer(30, 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
}
_ = requests.post(self.__url_send_message, data=data)
#写入日志
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:
# 接收服务器发送过来的数据
__recv_Data = self.__tcp_client_socket.recv(1024)
__recv_str = str(__recv_Data.decode('utf-8'))
self.__write_log(__recv_str)
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 = '' #巴法云用户私钥，必填
,topic = '' #巴法云设备主题，必填
,pc_mac = '' #电脑mac地址，必填（格式：xx:xx:xx……或xx-xx-xx……）
,local_ip = '' #本机ip地址，非电脑ip地址，必填（运行python的设备局域网ip地址，或填写'auto'尝试自动获取）
,use_ping = False #是否开启ping检测更新电脑状态，True为开启，False为关闭
,use_shutdown = False #是否需要关机指令，True为开启，False为关闭
,pc_ip = '' #电脑局域网ip地址，非必填，如开启ping或关机指令，则必填
,pc_account = '' #电脑登录账户，非必填，如开启关机指令，则必填
,pc_password = '' #电脑登录密码，非必填，如开启关机指令，则必填
,shutdown_time = 60 #延迟关机时间，非必填，如开启关机指令，则必填（单位：秒，立即关机填0）
,use_write_log = False #是否开启日志记录，True为开启，False为关闭
,log_path = '' #日志文件路径，非必填，如开启日志记录，则必填
,use_send_message = False #是否开启消息推送[方糖推送]，True为开启，False为关闭
,url_send_message = '' #[方糖推送]的个人接口，非必填，如开启消息推送，则必填
,channel_send_message = '' #[方糖推送]的频道，非必填，如开启消息推送，则必填
)
c.start()5.参考内容
5.1 《用小爱同学控制台式机睡眠和唤醒的思路》，来自知乎，作者：Comzyh
5.2 《电脑接入米家，控制电脑开关机，并且无需购买米家外设》，来自cnblogs，作者：hackyo
5.3 《python 接入，mqtt和tcp》，来自巴法开放论坛
5.4 巴法文档中心
