我用 Python 编写了一个 Windows 服务,需要能够检测用户活动。在正常情况下,我会调用 Windows
GetLastInputInfo
方法,但该方法在由服务调用时不起作用。以下是此方法文档中的相关信息:
此功能对于输入空闲检测非常有用。但是,GetLastInputInfo 不提供所有正在运行的会话中的系统范围的用户输入信息。相反,GetLastInputInfo 仅为调用该函数的会话提供特定于会话的用户输入信息。
重点是:“仅适用于调用该函数的会话”
如果由服务调用,
GetLastInputInfo
将always返回0,因为服务在会话0中运行并且不接收输入!我的服务如何从控制台会话检测用户活动?
预先警告 - 另一位用户评论说此方法似乎不适用于 Windows 11。我仅在 Windows 10 中成功测试了此解决方案。 YMMV!
幸运的是,这个问题有一个解决方法!虽然您无法直接从服务轮询用户活动,但您可以通过查询 Windows Client Server Runtime 进程(也称为“csrss.sys”)的 I/O 信息来检查系统当前是否正在处理用户输入。 exe”)。
通过利用Python的
psutil
模块,您可以检查csrss.exe的read_bytes
属性。该值应该在用户输入(即击键或鼠标事件)时随时更改。
首先,您需要获取csrss.exe进程的进程ID(PID):
import psutil
def get_csrss_pids() -> list[int]:
"""Get the PID(s) for the Windows Client Server Runtime process"""
# NOTE: more than one instance of csrss.exe may be running on your
# machine, so you'll want to gather all of the matching PIDs!
return [
proc.pid for proc in psutil.process_iter(attrs=['name'])
if proc.name() == 'csrss.exe'
]
一旦获得了 csrss.exe PID,您就可以使用
psutil
的 io_counters
方法来获取 read_bytes
信息
def get_io(pids: list[int]) -> list[int]:
"""Returns the last `read_bytes` value for the given csrss.exe PID(s)"""
# NOTE: if multiple PIDs are given, it's likely that only one of the PIDs
# 'read_bytes' values will be changing on user input because one of these
# processes is for your current session and the others aren't
return [psutil.Process(pid).io_counters().read_bytes for pid in pids]
get_io
函数将返回与每个给定 csrss.exe 进程 ID 的 read_bytes
值相对应的整数列表。要检查用户活动,应定期将此列表与以前存储的值进行比较 - 任何更改都意味着有来自用户的输入!
这是一个快速演示:
import psutil
def get_csrss_pids() -> list[int]:
"""Get the PID(s) for the Windows Client Server Runtime process"""
return [
proc.pid for proc in psutil.process_iter(attrs=['name'])
if proc.name() == 'csrss.exe'
]
def get_io(pids: list[int]) -> list[int]:
"""Returns the last `read_bytes` value for the given csrss.exe PID(s)"""
return [psutil.Process(pid).io_counters().read_bytes for pid in pids]
pids = get_csrss_pids()
last_io = get_io(pids) # store an initial value to compare against
while True:
try:
if (tick := get_io(pids)) != last_io: # if activity is detected...
print(tick) # do something
last_io = tick # store the new value to compare against
except KeyboardInterrupt:
break
要将这些功能合并到您的服务中,只需将它们包含在您的主类中(子类
ServiceFramework
) - 不要忘记添加 self
参数!
您需要调用
get_csrss_pids
并将 last_io
的初始值设置为 __init__
,然后从那里开始:
class MyService(ServiceFramework):
_svc_name_ = 'my_service'
_svc_display_name_ = 'My Service'
def __init__(self, *args):
super().__init__(*args)
self.csrss_pids = self.get_csrss_pids()
self.last_io = self.get_io()
...