WakeOnLan与PoweroffOnLan
专业,作品
|
|
newgd

747 字
|
47 分钟

大多数的PC机都支持WakeOnLan，但如何PoweroffOnLan？这就要自己实现了。

以下WakeOnLan里的Server、PoweroffOnLan里的Server都运行在管理机上（比如中央主机），PoweroffOnLan部分的Client是运行在被管理的PC机上的，可实现局域网远程关机。

功能特性

  • 局域网远程开机
  • 局域网远程关机
  • 实时查看局域网内某台计算机的开关机状态

WakeOnLan

Server

import subprocess
from wakeonlan import send_magic_packet
import json
import os
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class WakeOnLanController:
    def __init__(self, config_file=None):
        """初始化WakeOnLan控制器"""
        self.computers = {}
        if config_file:
            self.load_config(config_file)
        else:
            # 默认配置
            self.computers = {
                "SE1_computer": {"mac": "00:00:00:00:00:01", "ip": "192.168.1.101"},
                "SE2_computer": {"mac": "00:00:00:00:00:02", "ip": "192.168.1.102"},
                "SE3_computer": {"mac": "00:00:00:00:00:03", "ip": "192.168.1.103"}
            }
    
    def load_config(self, config_file):
        """从配置文件加载电脑信息"""
        try:
            with open(config_file, 'r') as f:
                config = json.load(f)
                self.computers = config.get("computers", {})
            logger.info(f"配置已从 {config_file} 加载")
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
            # 使用默认配置
            self.computers = {
                "SE1_computer": {"mac": "00:00:00:00:00:01", "ip": "192.168.1.101"},
                "SE2_computer": {"mac": "00:00:00:00:00:02", "ip": "192.168.1.102"},
                "SE3_computer": {"mac": "00:00:00:00:00:03", "ip": "192.168.1.103"}
            }
    
    def wake_computer(self, computer_id):
        print(f"wake computer: {computer_id}")
        """唤醒指定的电脑"""
        if computer_id not in self.computers:
            logger.error(f"未找到电脑ID: {computer_id}")
            return False
        
        computer = self.computers[computer_id]
        mac = computer.get("mac")
        ip = computer.get("ip")
        
        if not mac:
            logger.error(f"电脑 {computer_id} 没有MAC地址")
            return False
        
        try:
            logger.info(f"正在唤醒电脑 {computer_id} (MAC: {mac})")
            send_magic_packet(mac)
            logger.info(f"唤醒包已发送到 {mac}")
            return True
        except Exception as e:
            logger.error(f"唤醒电脑 {computer_id} 失败: {e}")
            return False
    
    def check_computer_status(self, computer_id):
        """检查电脑是否在线"""
        if computer_id not in self.computers:
            logger.error(f"未找到电脑ID: {computer_id}")
            return False
        
        computer = self.computers[computer_id]
        ip = computer.get("ip")
        
        if not ip:
            logger.error(f"电脑 {computer_id} 没有IP地址")
            return False
        
        try:
            # 使用ping命令检查电脑是否在线
            response = subprocess.run(
                ["ping", "-c", "1", "-W", "1", ip],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            return response.returncode == 0
        except Exception as e:
            logger.error(f"检查电脑 {computer_id} 状态失败: {e}")
            return False
    
    def wake_all_computers(self):
        """唤醒所有电脑"""
        results = {}
        for computer_id in self.computers:
            results[computer_id] = self.wake_computer(computer_id)
        return results

# 如果直接运行此脚本
if __name__ == "__main__":
    # 尝试从配置文件加载
    config_path = os.path.join(os.path.abspath("."), "tb_configs", "wol_config.json")
    controller = WakeOnLanController(config_path)
    
    # 测试唤醒所有电脑
    results = controller.wake_all_computers()
    for computer_id, success in results.items():
        status = "成功" if success else "失败"
        logger.info(f"唤醒电脑 {computer_id}: {status}")

wol_config.json

{
    "computers": {
        "SE1_computer": {
            "mac": "00:E0:4F:1C:74:00",
            "ip": "192.168.1.101",
            "description": "工位1电脑"
        },
        "SE2_computer": {
            "mac": "00:E0:4F:1C:92:00",
            "ip": "192.168.1.102",
            "description": "工位2电脑"
        },
        "SE3_computer": {
            "mac": "00:E0:4F:29:5D:00",
            "ip": "192.168.1.103",
            "description": "工位3电脑"
        }
    }
}

PoweroffOnLan

Server

import socket
import json
import time
import sys
import threading

# 配置
SERVER_HOST = '192.168.1.10'  # 服务器IP地址，需要修改为实际地址
SERVER_PORT = 20000
ADMIN_KEY = "admin_auth_key"


class PoweroffOnLanController:
    def __init__(self, config_file=None):

        self.thread = threading.Thread(target=self.admin_thread)
        self.thread.start()

        self.admin_socket = self.connect_to_server()
        if not self.admin_socket:
            print("[-] 无法连接到服务器，退出")
            sys.exit(1)
        pass

    def admin_thread(self):
        while True:
            time.sleep(1)
    
    def connect_to_server(self):
        """连接到服务器"""
        try:
            # 创建套接字
            admin = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            admin.connect((SERVER_HOST, SERVER_PORT))
            print(f"[+] 已连接到服务器 {SERVER_HOST}:{SERVER_PORT}")
            
            # 发送管理员认证
            auth_data = {"auth_key": ADMIN_KEY}
            admin.send(json.dumps(auth_data).encode('utf-8'))
            
            # 接收认证响应
            response = json.loads(admin.recv(1024).decode('utf-8'))
            if response.get("status") != "success":
                print(f"[-] 认证失败: {response.get('message')}")
                admin.close()
                return None
            
            print("[+] 管理员认证成功")
            return admin
            
        except ConnectionRefusedError:
            print(f"[-] 无法连接到服务器 {SERVER_HOST}:{SERVER_PORT}")
        except Exception as e:
            print(f"[-] 连接错误: {e}")
        
        return None

    def shutdown_pc(self, number):
        clients = self.list_clients(self.admin_socket)
        if clients:
            try:
                number = int(number)
                if 1 <= number <= 3:
                    target_ip = ""
                    if number == 1:
                        target_ip = "192.168.1.101"
                    elif number == 2:
                        target_ip = "192.168.1.102"
                    elif number == 3:
                        target_ip = "192.168.1.103"
                    for item in clients:
                        if item["address"][0] == target_ip:
                            print(f"[+] 正在关闭 {target_ip}")
                            self.shutdown_client(self.admin_socket, target_ip)
                            break
                        else:
                            print(f"[-] 无法找到 {target_ip}")
                elif number != 0:
                    print("[-] 无效的客户端序号")
            except ValueError:
                print("[-] 请输入有效的数字")

    def list_clients(self, admin_socket):
        """获取客户端列表"""
        try:
            admin_socket.send(json.dumps({"action": "list_clients"}).encode('utf-8'))
            response = json.loads(admin_socket.recv(4096).decode('utf-8'))
            
            if response.get("status") == "success":
                clients = response.get("clients", [])
                if not clients:
                    print("[*] 当前没有客户端连接")
                    return []
                
                print("\n当前在线客户端:")
                print("-" * 80)
                print(f"{'序号':<5}{'IP地址':<15}{'MAC地址':<20}{'主机名':<20}{'开机时间':<20}{'开机时长'}")
                print("-" * 80)
                
                for i, client in enumerate(clients, 1):
                    info = client.get("info", {})
                    uptime_seconds = info.get("uptime_seconds", 0)
                    hours, remainder = divmod(uptime_seconds, 3600)
                    minutes, seconds = divmod(remainder, 60)
                    uptime_str = f"{int(hours)}小时{int(minutes)}分{int(seconds)}秒"
                    
                    print(f"{i:<5}{client['address'][0]:<15}{info.get('mac', 'N/A'):<20}"
                        f"{info.get('hostname', 'N/A'):<20}{info.get('boot_time', 'N/A'):<20}{uptime_str}")
                
                return clients
            else:
                print(f"[-] 获取客户端列表失败: {response.get('message')}")
                return []
                
        except Exception as e:
            print(f"[-] 获取客户端列表错误: {e}")
            return []
        
    def shutdown_client(self, admin_socket, target_ip):
        """关闭指定客户端"""
        try:
            admin_socket.send(json.dumps({"action": "shutdown", "target_ip": target_ip}).encode('utf-8'))
            response = json.loads(admin_socket.recv(1024).decode('utf-8'))
            
            if response.get("status") == "success":
                print(f"[+] {response.get('message')}")
            else:
                print(f"[-] {response.get('message')}")
                
        except Exception as e:
            print(f"[-] 发送关机命令错误: {e}")


        
# def main():
#     print("[*] 远程关机管理端启动")
    
#     admin_socket = connect_to_server()
#     if not admin_socket:
#         print("[-] 无法连接到服务器，退出")
#         sys.exit(1)
    
#     try:
#         while True:
#             print("\n管理菜单:")
#             print("1. 列出在线客户端")
#             print("2. 关闭指定客户端")
#             print("3. 退出")
            
#             choice = input("请选择操作 [1-3]: ")
            
#             if choice == "1":
#                 list_clients(admin_socket)
            
#             elif choice == "2":
#                 clients = list_clients(admin_socket)
#                 if clients:
#                     client_num = input("\n请输入要关闭的客户端序号 (输入0取消): ")
#                     try:
#                         client_num = int(client_num)
#                         if 1 <= client_num <= len(clients):
#                             target_ip = clients[client_num-1]["address"][0]
#                             confirm = input(f"确认关闭 {target_ip} 吗? (y/n): ")
#                             if confirm.lower() == 'y':
#                                 shutdown_client(admin_socket, target_ip)
#                         elif client_num != 0:
#                             print("[-] 无效的客户端序号")
#                     except ValueError:
#                         print("[-] 请输入有效的数字")
            
#             elif choice == "3":
#                 print("[*] 退出管理端")
#                 break
            
#             else:
#                 print("[-] 无效的选择，请重试")
    
#     except KeyboardInterrupt:
#         print("\n[*] 管理端关闭")
#     except Exception as e:
#         print(f"[!] 错误: {e}")
#     finally:
#         if admin_socket:
#             admin_socket.close()

Client

client运行环境

python-3.11.9，python官网下载

client主机配置步骤

第1步：设置每个工位的电脑IP为固定IP

第2步：安装python-3.11.9-amd64.exe

第3步：C盘根目录创建tb_pcctrl目录

第4步：win+r输入：shell:startup，将client快捷方式复制到上一步打开的窗口文件夹下

第5步：确认client快捷方式属性

  • python路径：pythonw.exe（查看python安装路径：python -c "import sys; print(sys.executable)"）
  • 窗口：最小化
  • 高级：管理员身份运行

WakeOnLan与PoweroffOnLan

import socket
import json
import time
import threading
import subprocess
import uuid
import os
from datetime import datetime

# 配置
SERVER_HOST = '192.168.1.10'  # 服务器IP地址，需要修改为实际地址
SERVER_PORT = 20000
AUTH_KEY = "secure_auth_key"
RECONNECT_DELAY = 5  # 重连延迟(秒)

# 系统信息
boot_time = datetime.now()

def get_mac_address():
    """获取MAC地址"""
    mac = ':'.join(['{:02x}'.format((uuid.getnode() >> elements) & 0xff) 
                   for elements in range(0, 8*6, 8)][::-1])
    return mac

def get_ip_address():
    """获取IP地址"""
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        # 不需要真正连接
        s.connect(('10.255.255.255', 1))
        ip = s.getsockname()[0]
    except Exception:
        ip = '127.0.0.1'
    finally:
        s.close()
    return ip

def get_uptime():
    """获取开机时长(秒)"""
    return (datetime.now() - boot_time).total_seconds()

def get_system_info():
    """获取系统信息"""
    return {
        "ip": get_ip_address(),
        "mac": get_mac_address(),
        "boot_time": boot_time.strftime("%Y-%m-%d %H:%M:%S"),
        "uptime_seconds": get_uptime(),
        "hostname": socket.gethostname()
    }

def shutdown_computer():
    """关闭计算机"""
    print("[!] 收到关机命令，系统将在5秒后关闭...")
    subprocess.call(["shutdown", "/s", "/t", "5"])

def connect_to_server():
    """连接到服务器"""
    while True:
        try:
            # 创建套接字
            client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            client.connect((SERVER_HOST, SERVER_PORT))
            print(f"[+] 已连接到服务器 {SERVER_HOST}:{SERVER_PORT}")
            
            # 发送认证
            auth_data = {"auth_key": AUTH_KEY}
            client.send(json.dumps(auth_data).encode('utf-8'))
            
            # 接收认证响应
            response = json.loads(client.recv(1024).decode('utf-8'))
            if response.get("status") != "success":
                print(f"[-] 认证失败: {response.get('message')}")
                client.close()
                time.sleep(RECONNECT_DELAY)
                continue
            
            print("[+] 认证成功")
            
            # 发送系统信息
            system_info = get_system_info()
            client.send(json.dumps(system_info).encode('utf-8'))
            
            # 接收确认
            response = json.loads(client.recv(1024).decode('utf-8'))
            print(f"[+] 服务器响应: {response.get('message')}")
            
            # 主循环处理命令
            while True:
                data = client.recv(1024).decode('utf-8')
                if not data:
                    print("[-] 服务器断开连接")
                    break
                
                command = json.loads(data)
                
                if command.get("action") == "shutdown":
                    print("[!] 收到关机命令")
                    # 发送确认
                    client.send(json.dumps({"status": "success", "message": "正在关机"}).encode('utf-8'))
                    shutdown_computer()
                
                elif command.get("action") == "query_info":
                    print("[+] 收到信息查询请求")
                    system_info = get_system_info()
                    client.send(json.dumps({"info_update": system_info}).encode('utf-8'))
            
        except ConnectionRefusedError:
            print(f"[-] 无法连接到服务器 {SERVER_HOST}:{SERVER_PORT}")
        except json.JSONDecodeError:
            print("[-] 收到无效的JSON数据")
        except Exception as e:
            print(f"[-] 连接错误: {e}")
        
        print(f"[*] {RECONNECT_DELAY}秒后尝试重新连接...")
        time.sleep(RECONNECT_DELAY)

def main():
    print("[*] 远程关机客户端启动")
    print(f"[*] 系统启动时间: {boot_time.strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 启动连接线程
    connect_thread = threading.Thread(target=connect_to_server)
    connect_thread.daemon = True
    connect_thread.start()
    
    try:
        # 保持主线程运行
        while True:
            time.sleep(60)
    except KeyboardInterrupt:
        print("[*] 客户端关闭")

if __name__ == "__main__":
    main()
暂无评论

发送评论 编辑评论 


				

			

						

				

					

						

							

								
							

							
						

					

				

				

					

						

							

								
							

							
						

					

				

				

					

						

							

								
							

							
							
													

					

				

			

			
			

				

					
				

			

				

											

							
							
						

																				
					
											
						

	

		
|´・ω・)ノ
ヾ(≧∇≦*)ゝ
(☆ω☆)
（╯‵□′）╯︵┴─┴
￣﹃￣
(/ω＼)
∠( ᐛ 」∠)＿
(๑•̀ㅁ•́ฅ)
→_→
୧(๑•̀⌄•́๑)૭
٩(ˊᗜˋ*)و
(ノ°ο°)ノ
(´இ皿இ｀)
⌇●﹏●⌇
(ฅ´ω`ฅ)
(╯°A°)╯︵○○○
φ(￣∇￣o)
ヾ(´･ ･｀｡)ノ"
( ง ᵒ̌皿ᵒ̌)ง⁼³₌₃
(ó﹏ò｡)
Σ(っ °Д °;)っ
( ,,´･ω･)ﾉ"(´っω･｀｡)
╮(╯▽╰)╭
o(*////▽////*)q
＞﹏＜
( ๑´•ω•) "(ㆆᴗㆆ)
😂
😀
😅
😊
🙂
🙃
😌
😍
😘
😜
😝
😏
😒
🙄
😳
😡
😔
😫
😱
😭
💩
👻
🙌
🖕
👍
👫
👬
👭
🌚
🌝
🙈
💊
😶
🙏
🍦
🍉
😣
Source: github.com/k4yt3x/flowerhd
	

	

		
颜文字
Emoji
小恐龙
花!
	


									

			

			
			
		

	






上一篇
下一篇 

					
				

			

		

		
						
					
		
	


	
error: Content is protected !!