debounce()Description: 防抖函数
def debounce(wait):
"""
防抖函数
"""
def decorator(func):
def wrapper(*args, **kwargs):
nonlocal last_call_time
current_time = time.time()
if current_time - last_call_time > wait:
last_call_time = current_time
return func(*args, **kwargs)
last_call_time = None
return wrapper
return decorator
on_file_system_event(directories: tuple[str] = True, recursive: bool = None) -> Callable[[CALLBACK_FUNC], CALLBACK_FUNC]Description: 注册文件系统变化监听器
Return: 装饰器，装饰一个函数在接收到数据后执行
Arguments:
- directories: 监听目录们
- recursive: 是否递归监听子目录
- event_filter: 事件过滤器, 返回True则执行回调函数
def on_file_system_event(directories: tuple[str], recursive: bool=True, event_filter: FILTER_FUNC=None) -> Callable[[CALLBACK_FUNC], CALLBACK_FUNC]:
"""
注册文件系统变化监听器
Args:
directories: 监听目录们
recursive: 是否递归监听子目录
event_filter: 事件过滤器, 返回True则执行回调函数
Returns:
装饰器，装饰一个函数在接收到数据后执行
"""
def decorator(func: CALLBACK_FUNC) -> CALLBACK_FUNC:
def wrapper(event: FileSystemEvent):
if event_filter is not None and (not event_filter(event)):
return
func(event)
code_modified_handler = CodeModifiedHandler()
code_modified_handler.on_modified = wrapper
for directory in directories:
observer.schedule(code_modified_handler, directory, recursive=recursive)
return func
return decorator
CodeModifiedHandler(FileSystemEventHandler)@debounce(1)on_modified(self, event)@debounce(1)
def on_modified(self, event):
raise NotImplementedError('on_modified must be implemented')
on_created(self, event)def on_created(self, event):
self.on_modified(event)
on_deleted(self, event)def on_deleted(self, event):
self.on_modified(event)
on_moved(self, event)def on_moved(self, event):
self.on_modified(event)
on_any_event(self, event)def on_any_event(self, event):
self.on_modified(event)
CALLBACK_FUNC = Callable[[FileSystemEvent], None]Type: TypeAlias
Description: 位置1为FileSystemEvent
FILTER_FUNC = Callable[[FileSystemEvent], bool]Type: TypeAlias
Description: 位置1为FileSystemEvent