文章目录
- import mss import mss.tools import socket import threading import time import zlib class ScreenServer: def __init__(self, host=’0.0.0.0′, port=5000): self.host = host self.port = port self.running = False self.server_socket = None self.quality = 50 # 图像质量 (0-100) def start_server(self): self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind((self.host, self.port)) self.server_socket.listen(5) print(f”屏幕共享服务器启动在 {self.host}:{self.port}”) self.running = True while self.running: try: client_socket, addr = self.server_socket.accept() print(f”新的连接来自: {addr}”) # 为每个客户端创建新线程 client_thread = threading.Thread( target=self.handle_client, args=(client_socket,) ) client_thread.daemon = True client_thread.start() except Exception as e: if self.running: print(f”接受连接时出错: {e}”) def handle_client(self, client_socket): try: with mss.mss() as sct: # 获取所有显示器 monitors = sct.monitors # 使用主显示器 monitor = monitors[1] # 索引0是所有显示器的合并区域,索引1是主显示器 while self.running: # 捕获屏幕 screenshot = sct.grab(monitor) # 转换为PNG并压缩 png_data = mss.tools.to_png(screenshot.rgb, screenshot.size) compressed_data = zlib.compress(png_data, level=zlib.Z_BEST_COMPRESSION) # 发送数据大小和数据 size_data = len(compressed_data).to_bytes(4, byteorder=’big’) client_socket.send(size_data + compressed_data) # 控制帧率 time.sleep(0.05) # 约20FPS except Exception as e: print(f”处理客户端时出错: {e}”) finally: client_socket.close() def stop_server(self): self.running = False if self.server_socket: self.server_socket.close() if __name__ == “__main__”: server = ScreenServer() try: server.start_server() except KeyboardInterrupt: server.stop_server() print(“服务器已停止”)
- import socket import cv2 import numpy as np import zlib import threading class ScreenClient: def __init__(self, server_ip, server_port=5000): self.server_ip = server_ip self.server_port = server_port self.running = False self.client_socket = None def connect(self): try: self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.client_socket.connect((self.server_ip, self.server_port)) self.running = True print(f”已连接到服务器 {self.server_ip}:{self.server_port}”) return True except Exception as e: print(f”连接失败: {e}”) return False def receive_screen(self): try: while self.running: # 接收数据大小 size_data = self.client_socket.recv(4) if not size_data: break data_size = int.from_bytes(size_data, byteorder=’big’) # 接收压缩的图像数据 received_data = b”” while len(received_data) < data_size: chunk = self.client_socket.recv(min(4096, data_size – len(received_data))) if not chunk: break received_data += chunk if len(received_data) != data_size: print(“数据接收不完整”) continue # 解压并解码图像 decompressed_data = zlib.decompress(received_data) nparr = np.frombuffer(decompressed_data, np.uint8) frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR) if frame is not None: cv2.imshow(‘远程桌面’, frame) # 按’q’退出 if cv2.waitKey(1) & 0xFF == ord(‘q’): break except Exception as e: print(f”接收数据时出错: {e}”) finally: self.close() def close(self): self.running = False if self.client_socket: self.client_socket.close() cv2.destroyAllWindows() if __name__ == “__main__”: # 替换为服务端的IP地址 server_ip = “192.168.1.100” client = ScreenClient(server_ip) if client.connect(): client.receive_screen()
目录
- 方法一:使用mss和socket实现屏幕共享
- 服务端(被监控端)
- 客户端(监控端)
- 方法二:使用PyAutoGUI和Flask实现Web界面监控
- 服务端(被监控端)
- 安装必要的库
- 使用说明
- 注意事项
- 优化建议
使用Python可以在局域网内实现远程监控电脑屏幕。以下是两种实现方法:
import mss
import mss.tools
import socket
import threading
import time
import zlib
class ScreenServer:
def __init__(self, host='0.0.0.0', port=5000):
self.host = host
self.port = port
self.running = False
self.server_socket = None
self.quality = 50 # 图像质量 (0-100)
def start_server(self):
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.server_socket.bind((self.host, self.port))
self.server_socket.listen(5)
print(f"屏幕共享服务器启动在 {self.host}:{self.port}")
self.running = True
while self.running:
try:
client_socket, addr = self.server_socket.accept()
print(f"新的连接来自: {addr}")
# 为每个客户端创建新线程
client_thread = threading.Thread(
target=self.handle_client,
args=(client_socket,)
)
client_thread.daemon = True
client_thread.start()
except Exception as e:
if self.running:
print(f"接受连接时出错: {e}")
def handle_client(self, client_socket):
try:
with mss.mss() as sct:
# 获取所有显示器
monitors = sct.monitors
# 使用主显示器
monitor = monitors[1] # 索引0是所有显示器的合并区域,索引1是主显示器
while self.running:
# 捕获屏幕
screenshot = sct.grab(monitor)
# 转换为PNG并压缩
png_data = mss.tools.to_png(screenshot.rgb, screenshot.size)
compressed_data = zlib.compress(png_data, level=zlib.Z_BEST_COMPRESSION)
# 发送数据大小和数据
size_data = len(compressed_data).to_bytes(4, byteorder='big')
client_socket.send(size_data + compressed_data)
# 控制帧率
time.sleep(0.05) # 约20FPS
except Exception as e:
print(f"处理客户端时出错: {e}")
finally:
client_socket.close()
def stop_server(self):
self.running = False
if self.server_socket:
self.server_socket.close()
if __name__ == "__main__":
server = ScreenServer()
try:
server.start_server()
except KeyboardInterrupt:
server.stop_server()
print("服务器已停止")
import socket
import cv2
import numpy as np
import zlib
import threading
class ScreenClient:
def __init__(self, server_ip, server_port=5000):
self.server_ip = server_ip
self.server_port = server_port
self.running = False
self.client_socket = None
def connect(self):
try:
self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_socket.connect((self.server_ip, self.server_port))
self.running = True
print(f"已连接到服务器 {self.server_ip}:{self.server_port}")
return True
except Exception as e:
print(f"连接失败: {e}")
return False
def receive_screen(self):
try:
while self.running:
# 接收数据大小
size_data = self.client_socket.recv(4)
if not size_data:
break
data_size = int.from_bytes(size_data, byteorder='big')
# 接收压缩的图像数据
received_data = b""
while len(received_data) < data_size:
chunk = self.client_socket.recv(min(4096, data_size - len(received_data)))
if not chunk:
break
received_data += chunk
if len(received_data) != data_size:
print("数据接收不完整")
continue
# 解压并解码图像
decompressed_data = zlib.decompress(received_data)
nparr = np.frombuffer(decompressed_data, np.uint8)
frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
if frame is not None:
cv2.imshow('远程桌面', frame)
# 按'q'退出
if cv2.waitKey(1) & 0xFF == ord('q'):
break
except Exception as e:
print(f"接收数据时出错: {e}")
finally:
self.close()
def close(self):
self.running = False
if self.client_socket:
self.client_socket.close()
cv2.destroyAllWindows()
if __name__ == "__main__":
# 替换为服务端的IP地址
server_ip = "192.168.1.100"
client = ScreenClient(server_ip)
if client.connect():
client.receive_screen()
from flask import Flask, Response
import pyautogui
import cv2
import numpy as np
import threading
import time
app = Flask(__name__)
class ScreenStreamer:
def __init__(self):
self.frame = None
self.running = True
def capture_screen(self):
while self.running:
# 截取屏幕
screenshot = pyautogui.screenshot()
# 转换为OpenCV格式
frame = cv2.cvtColor(np.array(screenshot), cv2.COLOR_RGB2BGR)
# 编码为JPEG
_, jpeg = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 70])
self.frame = jpeg.tobytes()
time.sleep(0.1) # 控制帧率
def get_frame(self):
return self.frame
def stop(self):
self.running = False
streamer = ScreenStreamer()
def generate_frames():
while True:
frame = streamer.get_frame()
if frame:
yield (b'--framern'
b'Content-Type: image/jpegrnrn' + frame + b'rn')
@app.route('/')
def index():
return '''
<html>
<head>
<title>远程桌面监控</title>
</head>
<body>
<h1>远程桌面监控</h1>
<img src="/video_feed" width="80%">
</body>
</html>
'''
@app.route('/video_feed')
def video_feed():
return Response(generate_frames(),
mimetype='multipart/x-mixed-replace; boundary=frame')
if __name__ == '__main__':
# 启动屏幕捕获线程
capture_thread = threading.Thread(target=streamer.capture_screen)
capture_thread.daemon = True
capture_thread.start()
# 启动Flask服务器
app.run(host='0.0.0.0', port=5000, threaded=True)
pip install mss opencv-python pyautogui flask pillow zlib
方法一:
- 在被监控电脑上运行服务端代码
- 在监控电脑上运行客户端代码,并设置正确的服务端IP地址
- 客户端会显示一个窗口展示远程桌面
方法二:
- 在被监控电脑上运行Flask服务端
- 在浏览器中访问
http://被监控电脑IP:5000即可查看远程桌面
- 安全性:这些代码仅用于学习和授权的监控用途,请确保你有权限监控目标电脑
- 性能:屏幕共享可能会占用较多网络带宽,请根据网络情况调整图像质量和帧率
- 防火墙:确保相关端口(如5000)在防火墙中已开放
- 可以根据网络状况动态调整图像质量
- 可以只传输屏幕变化的部分以减少数据量
- 可以添加身份验证机制增强安全性
- 对于高分辨率屏幕,可以考虑缩放图像以减少数据量
以上就是使用Python实现局域网远程监控电脑屏幕的方法的详细内容,更多关于Python远程监控电脑屏幕的资料请关注风君子博客其它相关文章!
您可能感兴趣的文章:
- 使用Python一键实现屏幕监控软件
- 使用Python监控屏幕变化并截图的方法详解
- Python实现监控屏幕界面内容变化并发送通知