#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ PPT遥控器电脑端应用（带GUI界面） 支持Windows和MacOS系统 使用WebSocket接收手机端控制命令，通过模拟键盘按键控制PPT翻页 """ import asyncio import platform import socket import os import threading import json import time import tkinter as tk from tkinter import scrolledtext, ttk import websockets import qrcode from PIL import Image, ImageTk # 根据操作系统选择不同的键盘控制方法 system = platform.system() # UDP广播相关配置 UDP_BROADCAST_PORT = 5000 UDP_BROADCAST_INTERVAL = 2 # 广播间隔（秒） # WebSocket服务器配置 WS_PORT = 8765 class PPTRemoteServer: def __init__(self): # 服务器状态 self.is_running = False self.local_ip = "" self.server_thread = None self.broadcast_thread = None self.stop_event = threading.Event() self.qr_image = None # 创建主窗口 self.root = tk.Tk() self.root.title("PPT遥控器服务器") self.root.geometry("700x600") self.root.resizable(True, True) # 创建界面 self.create_widgets() # 获取本地IP地址 self.local_ip = self.get_local_ip() def get_local_ip(self): """获取本地IP地址""" try: # 创建一个UDP套接字连接到外部服务器 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except Exception as e: self.log(f"获取IP地址失败: {e}") return "127.0.0.1" def press_key(self, key): """根据操作系统模拟键盘按键""" if system == "Darwin": # MacOS # 使用AppleScript模拟键盘按键 script = f''' tell application "System Events" key code {self.get_key_code(key)} end tell ''' os.system(f"osascript -e '{script}'") elif system == "Windows": # 使用ctypes调用Windows API模拟键盘按键 import ctypes user32 = ctypes.WinDLL('user32', use_last_error=True) # 定义键盘事件常量 KEYEVENTF_KEYDOWN = 0x0000 KEYEVENTF_KEYUP = 0x0002 # 获取虚拟键码 vk_code = self.get_vk_code(key) # 按下键 user32.keybd_event(vk_code, 0, KEYEVENTF_KEYDOWN, 0) # 释放键 user32.keybd_event(vk_code, 0, KEYEVENTF_KEYUP, 0) else: self.log(f"不支持的操作系统: {system}") def get_key_code(self, key): """获取MacOS的键码""" key_codes = { "right": 124, "left": 123, "f5": 96, "esc": 53 } return key_codes.get(key, 0) def get_vk_code(self, key): """获取Windows的虚拟键码""" vk_codes = { "right": 0x27, "left": 0x25, "f5": 0x74, "esc": 0x1B } return vk_codes.get(key, 0) def start_udp_broadcast(self): """启动UDP广播服务""" # 创建UDP套接字 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) # 准备广播数据 broadcast_data = json.dumps({ "type": "ppt_remote_server", "ip": self.local_ip, "port": WS_PORT, "name": socket.gethostname(), "system": system }) self.log(f"UDP广播已启动，端口: {UDP_BROADCAST_PORT}") try: while not self.stop_event.is_set(): # 发送广播 sock.sendto(broadcast_data.encode('utf-8'), ('255.255.255.255', UDP_BROADCAST_PORT)) self.stop_event.wait(UDP_BROADCAST_INTERVAL) except Exception as e: self.log(f"UDP广播异常: {e}") finally: sock.close() async def handle_client(self, websocket): """处理客户端连接""" try: # 获取客户端地址 client_address = websocket.remote_address self.log(f"新客户端已连接: {client_address}") # 发送欢迎消息 await websocket.send("电脑端已就绪，等待控制命令") # 接收消息循环 async for message in websocket: self.log(f"收到来自 {client_address} 的消息: {message}") # 根据消息执行相应的PPT控制操作 if message == "next": # 下一页：按右箭头键 self.log("执行下一页操作") self.press_key("right") elif message == "prev": # 上一页：按左箭头键 self.log("执行上一页操作") self.press_key("left") elif message == "start": # 开始播放：按F5键 self.log("执行开始播放操作") self.press_key("f5") elif message == "stop": # 停止播放：按Esc键 self.log("执行停止播放操作") self.press_key("esc") else: self.log(f"未知命令: {message}") except Exception as e: self.log(f"客户端连接异常: {e}") finally: self.log(f"客户端已断开连接: {client_address}") async def ws_server_main(self): """WebSocket服务器主函数""" try: # 创建WebSocket服务器 async with websockets.serve(self.handle_client, self.local_ip, WS_PORT): self.log(f"WebSocket服务器已启动，地址: ws://{self.local_ip}:{WS_PORT}") # 等待停止事件 while not self.stop_event.is_set(): await asyncio.sleep(0.1) except Exception as e: if not self.stop_event.is_set(): # 如果不是因为停止事件而退出 self.log(f"WebSocket服务器异常: {e}") def run_server(self): """运行服务器（在单独线程中）""" loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(self.ws_server_main()) except Exception as e: if not self.stop_event.is_set(): self.log(f"服务器运行异常: {e}") finally: loop.close() def generate_qr_code(self, url): """生成二维码""" qr = qrcode.QRCode( version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=8, border=4, ) qr.add_data(url) qr.make(fit=True) # 创建二维码图像 img = qr.make_image(fill_color="black", back_color="white") # 调整图像大小（兼容不同版本的PIL） try: img = img.resize((150, 150), Image.Resampling.LANCZOS) except AttributeError: img = img.resize((150, 150), Image.LANCZOS) # 转换为Tkinter图像 self.qr_image = ImageTk.PhotoImage(img) # 更新标签 self.qr_label.config(image=self.qr_image) self.qr_label.image = self.qr_image def start_server(self): """启动服务器""" if self.is_running: return # 重置停止事件 self.stop_event.clear() self.is_running = True # 更新界面状态 self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.NORMAL) self.status_var.set("运行中") self.status_label.config(foreground="green") # 生成二维码 ws_url = f"ws://{self.local_ip}:{WS_PORT}" self.generate_qr_code(ws_url) # 启动UDP广播线程 self.broadcast_thread = threading.Thread( target=self.start_udp_broadcast, daemon=True ) self.broadcast_thread.start() # 启动WebSocket服务器线程 self.server_thread = threading.Thread( target=self.run_server, daemon=True ) self.server_thread.start() def stop_server(self): """停止服务器""" if not self.is_running: return self.is_running = False # 设置停止事件 self.stop_event.set() # 更新界面状态 self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) self.status_var.set("已停止") self.status_label.config(foreground="red") # 清除二维码 if self.qr_image: self.qr_label.config(image="") self.qr_image = None # 等待线程结束 if self.server_thread and self.server_thread.is_alive(): self.server_thread.join(timeout=2) if self.broadcast_thread and self.broadcast_thread.is_alive(): self.broadcast_thread.join(timeout=1) self.log("服务器已停止") def log(self, message): """记录日志""" timestamp = time.strftime("%H:%M:%S") log_message = f"[{timestamp}] {message}\n" # 在日志框中显示 self.log_text.insert(tk.END, log_message) self.log_text.see(tk.END) def create_widgets(self): """创建界面控件""" # 创建主框架 main_frame = ttk.Frame(self.root, padding="10") main_frame.pack(fill=tk.BOTH, expand=True) # 创建标题 title_label = ttk.Label(main_frame, text="PPT遥控器服务器", font=("Arial", 16, "bold")) title_label.pack(pady=10) # 创建状态显示区域 status_frame = ttk.LabelFrame(main_frame, text="服务器状态", padding="10") status_frame.pack(fill=tk.X, pady=5) # 创建主内容区域，左侧为文本信息，右侧为二维码 status_content_frame = ttk.Frame(status_frame) status_content_frame.pack(fill=tk.BOTH, expand=True) # 左侧文本信息区域 text_frame = ttk.Frame(status_content_frame) text_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) # 系统信息 system_info = f"系统: {platform.system()} {platform.release()}" ttk.Label(text_frame, text=system_info).pack(anchor=tk.W) # 本地IP地址 ip_info = f"本地IP地址: {self.local_ip}" ttk.Label(text_frame, text=ip_info).pack(anchor=tk.W) # WebSocket端口 ws_info = f"WebSocket端口: {WS_PORT}" ttk.Label(text_frame, text=ws_info).pack(anchor=tk.W) # UDP广播端口 udp_info = f"UDP广播端口: {UDP_BROADCAST_PORT}" ttk.Label(text_frame, text=udp_info).pack(anchor=tk.W) # 连接地址 conn_info = f"连接地址: ws://{self.local_ip}:{WS_PORT}" ttk.Label(text_frame, text=conn_info, foreground="blue").pack(anchor=tk.W) # 状态显示 status_frame2 = ttk.Frame(text_frame) status_frame2.pack(anchor=tk.W, pady=5) ttk.Label(status_frame2, text="服务器状态:").pack(side=tk.LEFT) self.status_var = tk.StringVar(value="已停止") self.status_label = ttk.Label(status_frame2, textvariable=self.status_var, foreground="red") self.status_label.pack(side=tk.LEFT, padx=5) # 右侧二维码区域 qr_container = ttk.Frame(status_content_frame) qr_container.pack(side=tk.RIGHT, padx=10) self.qr_label = ttk.Label(qr_container) self.qr_label.pack(pady=10) # 创建按钮区域 button_frame = ttk.Frame(main_frame) button_frame.pack(pady=10) self.start_btn = ttk.Button(button_frame, text="启动服务器", command=self.start_server) self.start_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = ttk.Button(button_frame, text="停止服务器", command=self.stop_server, state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=5) # 创建日志区域 log_frame = ttk.LabelFrame(main_frame, text="日志", padding="10") log_frame.pack(fill=tk.BOTH, expand=True, pady=5) self.log_text = scrolledtext.ScrolledText(log_frame, height=15, wrap=tk.WORD) self.log_text.pack(fill=tk.BOTH, expand=True) # 添加初始日志 self.log("PPT遥控器服务器已启动") self.log(system_info) self.log(ip_info) self.log(ws_info) self.log(udp_info) self.log(conn_info) self.log("等待用户启动服务器...") def run(self): """运行GUI""" self.root.mainloop() if __name__ == "__main__": # 创建并运行服务器 server = PPTRemoteServer() server.run()