'''
通过python获取win屏幕媒体流传递给js
'''
import asyncio
import ctypes
import threading
import numpy as np
import win32gui
import win32ui
import win32con
import webview
from aiortc import MediaStreamTrack, RTCPeerConnection
from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription
# -------------------------- 1. Windows高DPI适配 --------------------------
ctypes.windll.shcore.SetProcessDpiAwareness(2)
# -------------------------- 2. 兼容VideoFrame导入 --------------------------
try:from aiortc.contrib.media import VideoFrame,AudioFrame
except ImportError:from aiortc import VideoFrame,AudioFrame
# -------------------------- 2. AIORTC屏幕采集轨道 --------------------------
class ScreenVideoTrack(MediaStreamTrack):
kind = "video"
def __init__(self, fps=30):
super().__init__()
self.fps = fps
self.stop_flag = False
self.pts = 0
self.frame_interval = 1.0 / fps
# 获取真实屏幕分辨率（适配高DPI）
self.user32 = ctypes.windll.user32
self.width = self.user32.GetSystemMetrics(0)
self.height = self.user32.GetSystemMetrics(1)
def capture_screen(self):
"""Windows原生GDI采集屏幕（低延迟）"""
hdesktop = win32gui.GetDesktopWindow()
hwnd_dc = win32gui.GetWindowDC(hdesktop)
mfc_dc = win32ui.CreateDCFromHandle(hwnd_dc)
save_dc = mfc_dc.CreateCompatibleDC()
save_bitmap = win32ui.CreateBitmap()
save_bitmap.CreateCompatibleBitmap(mfc_dc, self.width, self.height)
save_dc.SelectObject(save_bitmap)
save_dc.BitBlt((0, 0), (self.width, self.height), mfc_dc, (0, 0), win32con.SRCCOPY)
# 转换为numpy数组（BGR格式，适配aiortc）
bmp_data = save_bitmap.GetBitmapBits(True)
frame = np.frombuffer(bmp_data, dtype=np.uint8).reshape((self.height, self.width, 4))
frame = frame[:, :, :3][:, :, ::-1] # BGRA → BGR
# 释放资源，避免内存泄漏
win32gui.DeleteObject(save_bitmap.GetHandle())
save_dc.DeleteDC()
mfc_dc.DeleteDC()
win32gui.ReleaseDC(hdesktop, hwnd_dc)
return frame
async def recv(self):
"""aiortc核心方法：返回视频帧"""
if self.stop_flag:
raise StopAsyncIteration
await asyncio.sleep(self.frame_interval)
frame_data = self.capture_screen()
# 转换为aiortc VideoFrame
video_frame = VideoFrame.from_ndarray(frame_data, format="bgr24")
video_frame.pts = int(self.pts)
video_frame.time_base = np.array([1, self.fps])
self.pts += 1
return video_frame
def stop(self):
"""停止屏幕采集"""
self.stop_flag = True
# -------------------------- 3. WebRTC服务（异步线程运行） --------------------------
class WebRTCService:
def __init__(self):
self.pc = None
self.screen_track = None
self.loop = asyncio.new_event_loop()
self.thread = None
def start_async_loop(self):
"""在子线程启动asyncio循环，避免阻塞pywebview UI"""
asyncio.set_event_loop(self.loop)
self.loop.run_forever()
async def create_offer(self):
"""生成Offer信令，供前端调用"""
# 初始化PeerConnection（配置STUN服务器，解决NAT穿透）
self.pc = RTCPeerConnection()
self.pc.addIceServers([{"urls": "stun:stun.l.google.com:19302"}])
# 添加屏幕采集轨道
self.screen_track = ScreenVideoTrack(fps=30)
self.pc.addTrack(self.screen_track)
print("✅ Python侧：已添加屏幕采集轨道")
# 监听连接状态
@self.pc.on("connectionstatechange")
def on_connectionstatechange():
print(f"🔌 WebRTC连接状态：{self.pc.connectionState}")
if self.pc.connectionState in ["failed", "closed", "disconnected"]:
if self.screen_track:
self.screen_track.stop()
# 生成Offer并设置本地SDP
offer = await self.pc.createOffer()
await self.pc.setLocalDescription(offer)
# 返回序列化的Offer（供前端解析）
return {
"sdp": self.pc.localDescription.sdp,
"type": self.pc.localDescription.type
}
async def set_answer(self, answer_data):
"""接收前端的Answer信令并应用"""
if not self.pc:
raise Exception("PeerConnection未初始化")
# 反序列化Answer
answer = RTCSessionDescription(
sdp=answer_data["sdp"],
type=answer_data["type"]
)
await self.pc.setRemoteDescription(answer)
print("✅ Python侧：已应用前端Answer信令，P2P连接建立")
def start(self):
"""启动WebRTC服务（子线程）"""
self.thread = threading.Thread(target=self.start_async_loop, daemon=True)
self.thread.start()
def stop(self):
"""停止服务并清理资源"""
if self.screen_track:
self.screen_track.stop()
if self.pc:
asyncio.run_coroutine_threadsafe(self.pc.close(), self.loop)
self.loop.stop()
if self.thread:
self.thread.join()
# -------------------------- 4. PyWebView API通信层 --------------------------
class PyWebViewAPI:
def __init__(self, webrtc_service):
self.webrtc_service = webrtc_service
# 供前端JS调用：获取Offer信令
async def get_offer(self):
return await asyncio.wrap_future(
asyncio.run_coroutine_threadsafe(self.webrtc_service.create_offer(), self.webrtc_service.loop)
)
# 供前端JS调用：发送Answer信令到Python
async def send_answer(self, answer_data):
await asyncio.wrap_future(
asyncio.run_coroutine_threadsafe(self.webrtc_service.set_answer(answer_data), self.webrtc_service.loop)
)
# -------------------------- 5. 前端页面（向日葵风格） --------------------------
HTML_CONTENT = """
Windows屏幕共享
"""
# -------------------------- 6. 主程序入口 --------------------------
if __name__ == "__main__":
# 初始化WebRTC服务并启动异步线程
webrtc_service = WebRTCService()
webrtc_service.start()
# 初始化PyWebView API
api = PyWebViewAPI(webrtc_service)
# 创建PyWebView窗口（必须用Edge Chromium引擎，支持WebRTC）
window = webview.create_window(
title="Windows屏幕共享",
html=HTML_CONTENT,
width=1400,
height=900,
resizable=True,
confirm_close=True,
easy_drag=False,
js_api=api
)
# 启动PyWebView（关键：指定edgechromium引擎）
try:
webview.start(
gui='edgechromium', # 强制使用WebView2引擎
private_mode=False, # WebRTC必需关闭私有模式
)
except KeyboardInterrupt:
print("\n🛑 程序被手动终止")
finally:
# 清理资源
webrtc_service.stop()
print("✅ 资源已清理完成")