import{_ as i,c as e,a7 as a,o as n}from"./chunks/framework.Z6UMW4JZ.js";const c=JSON.parse('{"title":"liteyuki.dev.observer","description":"","frontmatter":{"title":"liteyuki.dev.observer"},"headers":[],"relativePath":"dev/api/dev/observer.md","filePath":"zh/dev/api/dev/observer.md","lastUpdated":null}'),t={name:"dev/api/dev/observer.md"};function l(h,s,p,r,k,d){return n(),e("div",null,s[0]||(s[0]=[a(`
liteyuki.dev.observer
此模块用于注册观察者函数,使用watchdog监控文件变化并重启bot 启用该模块需要在配置文件中设置dev_mode
为True
CALLBACK_FUNC
说明: 位置1为FileSystemEvent
类型: TypeAlias
默认值: Callable[[FileSystemEvent], None]
FILTER_FUNC
说明: 位置1为FileSystemEvent
类型: TypeAlias
默认值: Callable[[FileSystemEvent], bool]
debounce(wait)
说明: 防抖函数
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
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)
on_file_system_event(directories: tuple[str], recursive: bool = True, event_filter: FILTER_FUNC = None) -> Callable[[CALLBACK_FUNC], CALLBACK_FUNC]
说明: 注册文件系统变化监听器
参数:
- 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]:
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