PIM
物理输入监控器(Physical Input Monitor),提供人机协同保护机制。
类变量
PIM 是全局单例,所有状态存储在类变量上。
构造方法
python
PIM(idle_wait=0, lock_input=False)- idle_wait
float— 等待物理空闲的秒数。0表示不等待, 此时所有@PIM.guard都退化为透传 - lock_input
bool— 操作时是否锁定物理输入(需管理员权限)
上下文管理器
python
with PIM(idle_wait=3):
# 在此范围内的操作会等待物理空闲
wx = Weixin()
wx.send_text("张三", "hello")类方法
start(timeout=3.0)
启动物理输入监控,等待钩子安装完成并检查结果。
- timeout
float— 等待钩子安装结果的超时秒数 - 返回
bool—True监控已启动(至少一个钩子生效),False启动失败
两个钩子都安装失败时不会假装在保护:记录警告并把监控标记为未启动,后续 @PIM.guard 不再等待物理空闲。只有一个失败时保留另一个,并警告说明保护已不完整。已启动时重复调用直接返回 True。
stop()
停止监控,卸载钩子。
get_idle_duration()
获取当前物理空闲时长。
- 返回
float— 空闲秒数
wait_for_idle(min_idle=3.0, check_interval=0.5)
等待物理输入空闲达到指定秒数。
- min_idle
float— 需要的最小空闲时长 - check_interval
float— 检查间隔
ensure_idle()
等待物理空闲,秒数取全局 PIM.idle_wait。@PIM.guard 内部走的就是这里;供生成器函数在函数体开头手动调用。
block_input()
锁定物理键盘鼠标输入(需管理员权限)。
unblock_input()
解除锁定。
装饰器
@PIM.guard
为方法添加物理空闲等待保护,执行期间可选锁定物理输入。
python
@PIM.guard
def my_operation():
# 等待 PIM.idle_wait 秒物理空闲;全局未配置时不等待
...等待秒数由全局 PIM.idle_wait 决定,是否锁定由全局 PIM.lock_input 决定,装饰器不提供逐方法开关。需要在某个函数里临时加长等待,直接调 PIM.wait_for_idle(秒数):
python
def my_critical_operation():
PIM.wait_for_idle(5)
...WARNING
配了 idle_wait 但监控未启动(没调 PIM.start(),也没通过 Weixin(idle_wait=...) 间接启动)时,等待会被跳过,此时会输出一条警告而不是静默忽略。
生成器函数不能用这个装饰器(装饰器只在创建迭代器时执行,等待时机不对),改在函数体开头调用 PIM.ensure_idle():
python
def iter_items(self):
PIM.ensure_idle() # 等同于 @PIM.guard
...
yield item工作原理
- 通过 Windows 低级钩子 (
SetWindowsHookExW) 监控物理键盘/鼠标事件 - 排除程序自身注入的虚拟输入(通过
LLKHF_INJECTED/LLMHF_INJECTED标志) - 所有带
@PIM.guard的方法执行前会等待物理输入空闲
注意事项
- PIM 仅在
idle_wait > 0时自动启动;只在装饰器上写秒数不会自动启动监控 - 钩子安装失败(权限不足、被安全软件拦截等)会输出警告并关闭监控,不会静默降级为空保护;
print(PIM())可以查看当前的钩子状态 lock_input需要管理员权限,否则BlockInput会静默失败block_input()/unblock_input()带引用计数 + 线程归属,嵌套调用时只有最外层那次真正解锁; Win32 只允许加锁线程解锁,锁被其他线程持有时本线程不参与计数- 监控线程以 daemon 模式运行,主进程退出时自动结束
示例
python
from pywxauto import Weixin, PIM
# 方式一:通过 Weixin 参数启用
wx = Weixin(idle_wait=3, lock_input=False)
wx.send_text("张三", "hello") # 自动等待物理空闲
# 方式二:手动管理
PIM.start()
PIM.wait_for_idle(5.0)
idle = PIM.get_idle_duration()
print(f"已空闲 {idle:.1f} 秒")
PIM.stop()
# 方式三:上下文管理器
with PIM(idle_wait=3):
wx = Weixin()
wx.send_text("张三", "hello")