回调系统 (Utils.callbacks)

回调系统模块提供了一套灵活的回调机制，用于处理异步操作的事件通知和状态更新。

模块概述

该模块包含以下主要组件：

• Callbacks - 基础回调容器类

• CallbackGroup - 回调组管理类

• InstallationCallbackGroup - 类型化的安装器回调组

• 多个协议接口定义不同场景下的回调签名

基础回调类

class Utils.callbacks.Callbacks(**kwargs: Callable[[Any], Any])

基类：object

基础回调容器类，用于存储和管理一组回调函数。

__init__(**kwargs)

初始化回调容器。

参数:

kwargs -- 命名回调函数，键为回调名称，值为可调用对象

__getitem__(key: str) → Callable[[Any], Any]

通过键获取回调函数。

参数:

key -- 回调函数名称

返回:

对应的回调函数，如果不存在则返回空操作函数

__getattr__(name: str) → Callable[[Any], Any]

通过属性访问回调函数。

参数:

name -- 回调函数名称

返回:

对应的回调函数，如果不存在则返回空操作函数

__init__(**kwargs: Callable[[Any], Any])

回调组管理

class Utils.callbacks.CallbackGroup(**kwargs: Callbacks)

基类：object

回调组管理类，用于组织多个相关的回调容器。

__init__(**kwargs)

初始化回调组。

参数:

kwargs -- 命名的回调容器，键为组名，值为Callbacks实例

__getitem__(key: str) → Callbacks

通过键获取回调容器。

参数:

key -- 回调组名称

返回:

对应的回调容器，如果不存在则返回空回调容器

__getattr__(name: str) → Callbacks

通过属性访问回调容器。

参数:

name -- 回调组名称

返回:

对应的回调容器，如果不存在则返回空回调容器

__init__(**kwargs: Callbacks)

类型化回调组

class Utils.callbacks.InstallationCallbackGroup(**kwargs: Callbacks)

基类：CallbackGroup

类型化的安装器回调组

专门用于安装过程的类型化回调组，提供类型提示支持。

协议接口

下载相关协议

class Utils.callbacks.IDownloadSingle(*args, **kwargs)

单一下载任务信号

单一下载任务的回调协议接口。

start()

任务开始信号。

progress(progress: int)

任务进度更新。

参数:

progress -- 进度百分比 (0-100)

bytes_downloaded(downloaded: int, total: int)

字节级进度更新。

参数:

• downloaded -- 已下载字节数

• total -- 总字节数

speed(speed: int)

下载速度更新。

参数:

speed -- 下载速度，单位：字节/秒

finished()

任务完成信号。

error(error: Exception)

任务错误信号。

参数:

error -- 发生的异常

start()

任务开始信号

progress(progress: int)

任务进度；0-100

bytes_downloaded(downloaded: int, total: int)

字节级进度；已下载字节数和总字节数

speed(speed: int)

任务下载速度；单位：字节/秒

finished()

任务完成信号

error(error: Exception)

任务错误信号，传递错误

__init__(*args, **kwargs)

class Utils.callbacks.IDownloadMultiThread(*args, **kwargs)

多线程任务调度

多线程下载任务的回调协议接口。

start()

任务开始信号。

tasks_progress(progress: Dict[str, int])

各任务进度更新。

参数:

progress -- 任务名称到进度百分比的映射

size(size: int)

任务数据大小。

参数:

size -- 数据大小，单位：字节

downloaded_size(task: str, size: int)

任务已下载大小。

参数:

• task -- 任务名称

• size -- 已下载大小，单位：字节

speed(speed: int)

下载速度更新。

参数:

speed -- 下载速度，单位：字节/秒

progress(progress: int)

总体进度更新。

参数:

progress -- 总进度百分比 (0-100)

finished(task: str)

单个任务完成信号。

参数:

task -- 完成的任务名称

error(task: str, error: Exception)

任务错误信号。

参数:

• task -- 出错的任务名称

• error -- 发生的异常

start()

任务开始信号

tasks_progress(progress: Dict[str, int])

任务进度；0-100

size(size: int)

任务数据大小；单位：字节

downloaded_size(size: int)

任务已下载的总大小；单位：字节

speed(speed: int)

任务下载速度；单位：字节/秒

progress(progress: int)

任务总进度；0-100

finished()

整个下载任务完成信号

error(error: Exception)

整个下载任务错误信号，传递错误

__init__(*args, **kwargs)

安装相关协议

class Utils.callbacks.IInstallGame(*args, **kwargs)

安装游戏信号

游戏安装任务的回调协议接口。

start()

安装开始信号。

finished()

安装完成信号。

error(error: Exception)

安装错误信号。

参数:

error -- 发生的异常

start()

任务开始信号

finished()

任务完成信号

error(error: Exception)

任务错误信号，传递错误

__init__(*args, **kwargs)

class Utils.callbacks.IVerifyGameFile(*args, **kwargs)

验证游戏文件信号

游戏文件验证任务的回调协议接口。

start()

验证开始信号。

finished()

验证完成信号。

error(error: Exception)

验证错误信号。

参数:

error -- 发生的异常

start()

任务开始信号

finished()

任务完成信号

error(error: Exception)

任务错误信号，传递错误

__init__(*args, **kwargs)

使用示例

基础使用:

from Utils.callbacks import Callbacks, CallbackGroup

# 创建回调函数
def on_start():
    print("任务开始")

def on_progress(progress):
    print(f"进度: {progress}%")

# 创建回调容器
callbacks = Callbacks(
    start=on_start,
    progress=on_progress
)

# 调用回调
callbacks.start()  # 输出: 任务开始
callbacks.progress(50)  # 输出: 进度: 50%

回调组使用:

# 创建回调组
download_callbacks = Callbacks(
    start=lambda: print("下载开始"),
    progress=lambda p: print(f"下载进度: {p}%")
)

install_callbacks = Callbacks(
    start=lambda: print("安装开始"),
    finished=lambda: print("安装完成")
)

callback_group = CallbackGroup(
    download=download_callbacks,
    install=install_callbacks
)

# 使用回调组
callback_group.download.start()  # 输出: 下载开始
callback_group.install.finished()  # 输出: 安装完成

注意事项

• 当访问不存在的回调时，会返回一个空操作函数而不是抛出异常

• 在DEBUG日志级别下，调用空回调时会记录调用堆栈信息，便于调试

• 协议接口使用了Python的Protocol类型，提供静态类型检查支持

• InstallationCallbackGroup提供了类型化的回调访问，增强IDE支持