Files
python----/工具测试/监听usb接口.ipynb
2025-10-20 12:32:18 +08:00

12 KiB

In [ ]:
import wmi
import subprocess
import time
from threading import Lock

# 配置部分
KEY_FILE = ".key"        # U盘中的密码文件
EXPECTED_PASSWORD = "1234"  # 预设密码
DEVCON_PATH = r"E:\下载器\Devcon\x64\devcon.exe"
DISABLED_DEVICES = set()
device_lock = Lock()

def disable_device(dev_id):
    subprocess.run([DEVCON_PATH, "disable", dev_id], shell=True, capture_output=True)
    with device_lock:
        DISABLED_DEVICES.add(dev_id)

def enable_device(dev_id):
    subprocess.run([DEVCON_PATH, "enable", dev_id], shell=True, capture_output=True)
    with device_lock:
        if dev_id in DISABLED_DEVICES:
            DISABLED_DEVICES.remove(dev_id)

def get_usb_storage_drives():
    """获取所有可移动USB存储设备的盘符"""
    c = wmi.WMI()
    drives = []
    for disk in c.Win32_DiskDrive(InterfaceType="USB"):
        for p in disk.associators("Win32_DiskDriveToDiskPartition"):
            for d in p.associators("Win32_LogicalDiskToPartition"):
                drives.append(d.DeviceID)
    return drives

def check_usb_key(drive):
    """检查U盘根目录的密钥文件"""
    try:
        key_path = fr"{drive}\{KEY_FILE}"
        with open(key_path, "r") as f:
            return f.read().strip() == EXPECTED_PASSWORD
    except:
        return False

def handle_storage_insert():
    """处理U盘插入事件"""
    time.sleep(2)  # 等待系统分配盘符
    for drive in get_usb_storage_drives():
        if check_usb_key(drive):
            print(f"{drive} 发现有效密钥")
            with device_lock:
                for dev_id in list(DISABLED_DEVICES):
                    enable_device(dev_id)
            return True
    return False

def monitor_devices():
    c = wmi.WMI()
    
    # 存储设备监听线程
    def storage_watcher():
        storage_monitor = c.Win32_DeviceChangeEvent.watch_for(
            event_type=2,  # 设备插入
            delay_secs=2
        )
        while True:
            try:
                storage_monitor()
                if handle_storage_insert():
                    print("成功启用被禁用的键盘设备")
            except Exception as e:
                print(f"存储设备监控异常: {e}")

    # 键盘设备监听
    keyboard_monitor = c.Win32_DeviceChangeEvent.watch_for(
        EventType=2,
        delay_secs=1
    )
    
    # 启动存储监控线程
    import threading
    threading.Thread(target=storage_watcher, daemon=True).start()

    while True:
        try:
            keyboard_monitor()
            print("检测到新输入设备...")
            c = wmi.WMI()
            for dev in c.Win32_PnPEntity():
                if "Keyboard" in str(dev.Description) and "USB" in str(dev.Description):
                    dev_id = dev.DeviceID
                    if dev_id not in DISABLED_DEVICES:
                        print(f"禁用USB键盘: {dev_id}")
                        disable_device(dev_id)
        except Exception as e:
            print(f"键盘监控异常: {e}")

# if __name__ == "__main__":
#     # 检查管理员权限
#     import ctypes, sys
#     if ctypes.windll.shell32.IsUserAnAdmin() == 0:
#         ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, __file__, None, 1)
#     else:
#         monitor_devices()
In [ ]:
c = wmi.WMI()
In [ ]:
watcher = c.Win32_VolumeChangeEvent.watch_for(
    EventType=2  # 2表示卷载入(设备插入)
)
event = watcher()
drive_letter = event.DriveName
In [ ]:
drive_letter
In [ ]:
import wmi
import time

def wait_for_usb():
    c = wmi.WMI()

    # 监听USB存储设备插入事件
    watcher = c.Win32_VolumeChangeEvent.watch_for(
        EventType=2  # 2表示卷载入(设备插入)
    )

    print("等待U盘插入...")
    while True:
        try:
            event = watcher()
            drive_letter = event.DriveName
            print(f"检测到U盘插入,盘符: {drive_letter}")
            return drive_letter
        except wmi.x_wmi_timed_out:
            # 超时后继续等待
            time.sleep(0.1)
        except KeyboardInterrupt:
            print("\n已停止监听")
            break

if __name__ == "__main__":
    usb_drive = wait_for_usb()
    if usb_drive:
        print(f"接下来可以使用 {usb_drive} 访问U盘文件")
        # 示例:列出U盘根目录文件
        import os
        print("根目录文件:", os.listdir(usb_drive))
In [ ]:
keyboard_monitor = c.Win32_DeviceChangeEvent.watch_for(
    EventType=2,
    delay_secs=1
)
keyboard_monitor()
for dev in c.Win32_PnPEntity(fields=["Description","DeviceID"]):
    if "Keyboard" in str(dev.Description):
        print(f"检测到键盘设备: {dev.Description} - ID: {dev.DeviceID}")
        os.system(rf'''E:\下载器\Devcon\x64\devcon.exe remove @"{dev.DeviceID}"''')
In [ ]:

    # if "Keyboard" in str(dev.Description) and "USB" in str(dev.Description):
    #     dev_id = dev.DeviceID
    #     if dev_id not in DISABLED_DEVICES:
    #         print(f"禁用USB键盘: {dev_id}")
    #         disable_device(dev_id)
In [ ]:
subprocess.run([DEVCON_PATH, "remove", r'''@"HID\VID_0C45&PID_8071&MI_00\7&612DDAA&0&0000"'''], shell=True, capture_output=True)
In [ ]:
import os   
In [ ]:
for disk in c.Win32_DiskDrive():
    print(f"USB设备: {disk.InterfaceType}")
In [ ]:
import tkinter as tk

def show_auto_close_message(title, message, duration=3):
    # 创建主窗口(隐藏)
    root = tk.Tk()
    root.withdraw()

    # 创建弹窗
    popup = tk.Toplevel()
    popup.title(title)
    
    # 设置消息内容
    label = tk.Label(popup, text=message, padx=30, pady=20)
    label.pack()
    
    # 禁止窗口缩放
    popup.resizable(False, False)
    
    # 计算居中位置
    popup.update_idletasks()  # 强制更新窗口尺寸
    width = popup.winfo_width()
    height = popup.winfo_height()
    x = (popup.winfo_screenwidth() - width) // 2
    y = (popup.winfo_screenheight() - height) // 2
    popup.geometry(f"+{x}+{y}")

    # 设置自动关闭
    def close():
        popup.destroy()
        root.destroy()  # 同时销毁主窗口

    popup.after(duration * 1000, close)
    popup.mainloop()

# 使用示例
if __name__ == "__main__":
    show_auto_close_message("操作成功", "设备已启用!", 3)
In [7]:
import tkinter as tk

def close_window():
    # 关闭弹窗和主窗口(确保程序退出)
    popup.destroy()
    root.destroy()

# 创建主窗口并隐藏
root = tk.Tk()
root.withdraw()

# 创建弹窗
popup = tk.Toplevel(root)
popup.title("提示")

# 设置弹窗大小和居中显示
window_width = 300
window_height = 200
screen_width = root.winfo_screenwidth()
screen_height = root.winfo_screenheight()
x = (screen_width - window_width) // 2
y = (screen_height - window_height) // 2
popup.geometry(f"{window_width}x{window_height}+{x}+{y}")

# 添加文本标签
label = tk.Label(popup, text="3秒后自动关闭", font=("微软雅黑", 12))
label.pack(expand=True, pady=50)

# 设置3秒后关闭窗口
popup.after(3000, close_window)

# 启动主循环
root.mainloop()
---------------------------------------------------------------------------
KeyboardInterrupt                         Traceback (most recent call last)
Cell In[7], line 33
     30 popup.after(3000, close_window)
     32 # 启动主循环
---> 33 root.mainloop()

File d:\Anaconda3\lib\tkinter\__init__.py:1429, in Misc.mainloop(self, n)
   1427 def mainloop(self, n=0):
   1428     """Call the mainloop of Tk."""
-> 1429     self.tk.mainloop(n)

KeyboardInterrupt: