Skip to content

# Module `liteyuki.dev.observer` ​

此模块用于注册观察者函数，使用watchdog监控文件变化并重启bot 启用该模块需要在配置文件中设置`dev_mode`为True

### var `CALLBACK_FUNC` ​

- Description: 位置1为FileSystemEvent

- Type: `TypeAlias`

- Default: `Callable[[FileSystemEvent], None]`

### var `FILTER_FUNC` ​

- Description: 位置1为FileSystemEvent

- Type: `TypeAlias`

- Default: `Callable[[FileSystemEvent], bool]`

### func `debounce(wait)` ​

Description: 防抖函数

Source code or View on GitHub
python

    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

### class `CodeModifiedHandler(FileSystemEventHandler)` ​

`@debounce(1)`

#### func `on_modified(self, event)` ​

Source code or View on GitHub
python

    @debounce(1)
    def on_modified(self, event):
        raise NotImplementedError('on_modified must be implemented')

#### func `on_created(self, event)` ​

Source code or View on GitHub
python

    def on_created(self, event):
        self.on_modified(event)

#### func `on_deleted(self, event)` ​

Source code or View on GitHub
python

    def on_deleted(self, event):
        self.on_modified(event)

#### func `on_moved(self, event)` ​

Source code or View on GitHub
python

    def on_moved(self, event):
        self.on_modified(event)

#### func `on_any_event(self, event)` ​

Source code or View on GitHub
python

    def on_any_event(self, event):
        self.on_modified(event)

### func `on_file_system_event(directories: tuple[str], recursive: bool = True, event_filter: FILTER_FUNC = None) -> Callable[[CALLBACK_FUNC], CALLBACK_FUNC]` ​

Description: 注册文件系统变化监听器

Arguments:

> - directories: 监听目录们
> - recursive: 是否递归监听子目录
> - event_filter: 事件过滤器, 返回True则执行回调函数

Return: 装饰器，装饰一个函数在接收到数据后执行

Source code or View on GitHub
python

    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