模块开发最佳实践

本文档提供了 ErisPulse 模块开发的最佳实践建议。

模块设计

1. 单一职责原则

每个模块应该只负责一个核心功能：

# 好的设计：每个模块只负责一个功能
class WeatherModule(BaseModule):
    """天气查询模块"""
    pass

class NewsModule(BaseModule):
    """新闻查询模块"""
    pass

# 不好的设计：一个模块负责多个不相关的功能
class UtilityModule(BaseModule):
    """包含天气、新闻、笑话等多个功能"""
    pass

2. 模块命名规范

[project]
name = "ErisPulse-ModuleName"  # 使用 ErisPulse- 前缀

3. 清晰的配置管理

推荐使用声明式配置（ConfigClass + BaseConfig），获得类型安全、自动模板生成、WebUI 表单支持等能力：

from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_url: str = field(default="https://api.example.com", metadata={
        "description": {"i18n": "my_module.api_url", "default": "API 地址"},
    })
    timeout: int = field(default=30, metadata={
        "description": {"i18n": "my_module.timeout", "default": "超时时间（秒）"},
    })
    cache_ttl: int = field(default=3600, metadata={
        "description": {"i18n": "my_module.cache_ttl", "default": "缓存存活时间（秒）"},
    })

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    async def do_something(self):
        cfg = self.cfg  # 类型安全，实时读取
        await self._fetch(cfg.api_url, timeout=cfg.timeout)

也可以在继续使用手动方式读写配置存储（见模块核心概念）。

声明式翻译键（v2.7.0+）

模块可以通过 I18nClass 集中声明翻译键，框架自动注册到 i18n 系统，无需手动调用 i18n.register()。

from ErisPulse.Core.Bases import BaseI18n, I18nKey

class MyModule(BaseModule):
    class I18nClass(BaseI18n):
        # 带占位符的业务翻译键
        welcome: I18nKey = I18nKey(
            default="Welcome, {name}!",
            zh_CN="欢迎你，{name}！",
            zh_TW="歡迎你，{name}！",
            en="Welcome, {name}!",
            ja="ようこそ、{name}！",
            ru="Добро пожаловать, {name}!",
        )
        # 配置字段描述的翻译
        api_url: I18nKey = I18nKey(
            default="API URL",
            zh_CN="API 地址",
            zh_TW="API 位址",
            en="API URL",
            ja="API URL",
            ru="API URL",
        )

详细用法见 i18n 文档。

异步编程

1. 使用异步库

# 推荐使用 SDK 内置 HTTP 客户端（异步，自动日志和统计）
from ErisPulse.Core import client

class MyModule(BaseModule):
    async def fetch_data(self, url):
        resp = await client.get(url)
        return await resp.json()

# 也可通过 sdk.client 使用（效果相同）
from ErisPulse import sdk

class MyModule(BaseModule):
    async def fetch_data(self, url):
        resp = await sdk.client.get(url)
        return await resp.json()

# 不要使用 aiohttp 直接导入（不便于框架统一管理）
import aiohttp

class MyModule(BaseModule):
    async def fetch_data(self, url):
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.json()

# 不要使用 requests（同步，会阻塞事件循环）
import requests

class MyModule(BaseModule):
    def fetch_data(self, url):
        return requests.get(url).json()  # 会阻塞事件循环

2. 正确的异步操作

from ErisPulse.Core.Event import Event  # event: Event 注解可获得 IDE 补全

async def handle_command(self, event: Event):
    # 需要等待结果的耗时操作：直接 await（生命周期明确）
    result = await self._long_operation()

async def on_load(self, event: dict):
    # 后台任务（轮询/定时/fire-and-forget）：用 self.spawn()，
    # 模块卸载时框架在 on_unload 之后兜底取消，避免持有 self 导致泄漏
    self.spawn(self._poll())

3. 资源管理

async def on_load(self, event):
    # SDK 客户端已自动管理连接池，无需手动创建 session
    pass
    
async def on_unload(self, event):
    # 如需自定义客户端，记得清理资源
    pass

事件处理

1. 使用 Event 包装类

# 使用 Event 包装类的便捷方法
@command("info")
async def info_command(event: Event):
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    await event.reply(f"你好，{nickname}！")

# 而非直接访问字典
@command("info")
async def info_command(event: Event):
    user_id = event["user_id"]  # 不够清晰，容易出错

2. 合理使用懒加载

# 低频命令模块：声明 activate_on 触发器，首个匹配命令到达时自动激活（保持懒加载）
class CommandModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True, activate_on=[
            {"command": {"name": "dice", "help": "掷一个骰子", "aliases": ["d"]}},
        ])

# 低频监听器模块：声明事件触发器，事件到达时自动激活
class ListenerModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True, activate_on=[
            {"notice": "group_member_increase"},
        ])

# 高频触发（每条消息都要处理）或启动时就必须就绪的模块：立即加载
class HotListenerModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=False)

# 工具模块适合懒加载
class UtilityModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True)

activate_on 的完整语法（事件三形式 / 命令简写与 dict 声明 / help 回退链）见 懒加载模块系统。

3. 事件处理器注册

async def on_load(self, event):
    # 在 on_load 中注册事件处理器
    @command("hello")
    async def hello_handler(event: Event):
        await event.reply("你好！")
    
    @message.on_group_message()
    async def group_handler(event: Event):
        self.logger.info("收到群消息")
    
    # 不需要手动注销，框架会自动处理

工具模块：托管别人东西时要接住"卸载通知"

什么时候需要：你的模块替其他模块保管东西（定时回调、订阅者、连接、缓存条目……）。这些引用在对方模块卸载后如果一直不丢弃，对方实例就永远无法被回收——这是工具模块最常见的内存泄漏来源。

from ErisPulse.Core.Bases import BaseModule
from ErisPulse.runtime import off_cleanup, on_cleanup

class MyToolModule(BaseModule):
    def __init__(self):
        self._entries = {}  # {模块名: 托管的东西}

    def register(self, entry):
        owner = on_cleanup(self._drop)   # ① 登记时挂入清理链，自动识别调用方
        self._entries.setdefault(owner, []).append(entry)

    def _drop(self, owner: str):
        self._entries.pop(owner, None)   # ② 对方卸载时框架自动调用：丢弃它的东西

    async def on_unload(self, event):
        off_cleanup(self._drop)          # ③ 自己卸载前注销钩子

就这么多，框架保证：

• 对方模块被卸载 / 禁用（或适配器关闭）时，_drop("对方模块名") 一定会被调用
• 调用方识别全自动：对方在 on_load 里直接调 sdk.MyToolModule.register(...)，或经 sdk.module.call("MyToolModule", "register", ...) 调用，都能正确识别是谁
• 不用操心时机——钩子在框架清理链内触发，早于泄漏诊断，不会误报

不接入的后果：对方 purge 彻底卸载时实例无法回收（泄漏诊断报"不可回收"）；若对方自己也不在 on_unload 里向你注销，泄漏就是永久性的。

普通模块（不托管别人东西）不需要关心这个——框架资源（命令 / 处理器 / 路由 / 后台任务……）的卸载清理是全自动的。

触发时机、调用方识别规则、超时与容错等细节见 归属权系统 · 工具模块指南。

错误处理

1. 分类异常处理

from ErisPulse.Core.Bases.errors import ClientError

async def handle_event(self, event: Event):
    try:
        result = await self._process(event)
    except ValueError as e:
        # 预期的业务错误
        self.logger.warning(f"业务警告: {e}")
        await event.reply(f"参数错误: {e}")
    except ClientError as e:
        # 网络错误（sdk.client 的底层 aiohttp 异常已自动转换）
        self.logger.error(f"网络错误 {e.method} {e.url}: {e}")
        await event.reply("网络请求失败，请稍后重试")
    except Exception as e:
        # 未预期的错误
        self.logger.error(f"未知错误: {e}", exc_info=True)
        await event.reply("处理失败，请联系管理员")
        raise

2. 超时处理

# 推荐使用 SDK 内置客户端（自带超时和重试）
from ErisPulse.Core import client
from ErisPulse.Core.Bases.errors import ClientTimeoutError

async def fetch_with_timeout(self, url, timeout=30):
    try:
        resp = await client.get(url, timeout=timeout)
        return await resp.json()
    except ClientTimeoutError:
        self.logger.warning(f"请求超时: {url}")
        raise

存储系统

1. 使用事务

# 使用事务确保数据一致性
async def update_user(self, user_id, data):
    with self.sdk.storage.transaction():
        self.sdk.storage.set(f"user:{user_id}:profile", data["profile"])
        self.sdk.storage.set(f"user:{user_id}:settings", data["settings"])

# ❌ 不使用事务可能导致数据不一致
async def update_user(self, user_id, data):
    self.sdk.storage.set(f"user:{user_id}:profile", data["profile"])
    # 如果这里出错，上面的设置无法回滚
    self.sdk.storage.set(f"user:{user_id}:settings", data["settings"])

2. 批量操作

# 使用批量操作提高性能
def cache_multiple_items(self, items):
    self.sdk.storage.set_multi({
        f"item:{k}": v for k, v in items.items()
    })

# ❌ 多次调用效率低
def cache_multiple_items(self, items):
    for k, v in items.items():
        self.sdk.storage.set(f"item:{k}", v)

日志记录

1. 合理使用日志级别

# DEBUG: 详细的调试信息（仅开发时）
self.logger.debug(f"输入参数: {params}")

# INFO: 正常运行信息
self.logger.info("模块已加载")
self.logger.info(f"处理请求: {request_id}")

# WARNING: 警告信息，不影响主要功能
self.logger.warning(f"配置项 {key} 未设置，使用默认值")
self.logger.warning("API 响应慢，可能需要优化")

# ERROR: 错误信息
self.logger.error(f"API 请求失败: {e}")
self.logger.error(f"处理事件失败: {e}", exc_info=True)

# CRITICAL: 致命错误，需要立即处理
self.logger.critical("数据库连接失败，机器人无法正常运行")

2. 结构化日志

# 使用结构化日志，便于解析
self.logger.info(f"处理请求: request_id={request_id}, user_id={user_id}, duration={duration}ms")

# ❌ 使用非结构化日志
self.logger.info(f"处理请求了，来自用户 {user_id}，用时 {duration} 毫秒")

性能优化

1. 使用缓存

class MyModule(BaseModule):
    def __init__(self):
        self._cache = {}
        self._cache_lock = asyncio.Lock()
    
    async def get_data(self, key):
        async with self._cache_lock:
            if key in self._cache:
                return self._cache[key]
            
            # 从数据库获取
            data = await self._fetch_from_db(key)
            
            # 缓存数据
            self._cache[key] = data
            return data

2. 避免阻塞操作

# 使用异步操作
async def process_message(self, event: Event):
    # 异步处理
    await self._async_process(event)

# ❌ 阻塞操作
async def process_message(self, event: Event):
    # 同步操作，阻塞事件循环
    result = self._sync_process(event)

安全性

1. 敏感数据保护

# 敏感数据存储在配置中（声明式 ConfigClass，secret 字段不进入日志/导出）
from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseModule, BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_key: str = field(
        default="",
        metadata={"description": "API 密钥", "secret": True},
    )

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    def check_api_key(self):
        if not self.cfg.api_key or self.cfg.api_key == "YOUR_API_KEY_HERE":
            raise ValueError("请在 config.toml 中配置有效的 API 密钥")

# ❌ 敏感数据硬编码
class MyModule(BaseModule):
    API_KEY = "sk-1234567890"  # 不要这样做！

2. 输入验证

# 验证用户输入
async def process_command(self, event: Event):
    user_input = event.get_text()
    
    # 验证输入长度
    if len(user_input) > 1000:
        await event.reply("输入过长，请重新输入")
        return
    
    # 验证输入格式
    if not re.match(r'^[a-zA-Z0-9]+$', user_input):
        await event.reply("输入格式不正确")
        return

测试

1. 单元测试

import pytest
from ErisPulse.Core.Bases import BaseModule

class TestMyModule:
    def test_config_defaults(self):
        """测试配置默认值"""
        config = MyModule.ConfigClass()
        assert config.timeout == 30

2. 集成测试

@pytest.mark.asyncio
async def test_command_handling():
    """测试命令处理"""
    module = MyModule()
    await module.on_load({})
    
    # 模拟命令事件
    event = create_test_command_event("hello")
    await module.handle_command(event)

部署

1. 版本管理

[project]
name = "ErisPulse-MyModule"
version = "1.0.0"

遵循语义化版本：

• MAJOR.MINOR.PATCH
• 主版本：不兼容的 API 变更
• 次版本：向下兼容的功能新增
• 修订号：向下兼容的问题修正

2. README 头部

epsdk create 生成的 README 已内置 ErisPulse 头部标识（Logo + 徽章行）。两种推荐模式：

模式 A — 仅 ErisPulse Logo（默认）：

<div align="center">

<img src="https://raw.githubusercontent.com/ErisPulse/ErisPulse/main/.github/assets/ErisPulseLogo.png" width="180" alt="MyModule" />

# MyModule

**一句话描述**

<p>
  <a href="https://pypi.org/project/ErisPulse-MyModule/"><img src="https://img.shields.io/pypi/v/ErisPulse-MyModule?style=for-the-badge&logo=pypi&logoColor=white" alt="PyPI"></a>
  <a href="https://pypi.org/project/ErisPulse-MyModule/"><img src="https://img.shields.io/badge/Python-3.10+-FFD43B?style=for-the-badge&logo=python&logoColor=blue" alt="Python"></a>
  <a href="./LICENSE"><img src="https://img.shields.io/badge/License-MIT-blue?style=for-the-badge" alt="License"></a>
  <a href="https://github.com/ErisPulse/ErisPulse"><img src="https://img.shields.io/badge/Powered_by-ErisPulse-FF6B9D?style=for-the-badge&logo=bookstack&logoColor=white" alt="ErisPulse"></a>
</p>

</div>

模式 B — 模块图标 × ErisPulse Logo（有自定义图标时）：

<div align="center">

<img src=".github/assets/MyModuleIcon.svg" width="120" alt="MyModule" />
<span style="font-size:44px;color:#c8c8c8;margin:0 18px;vertical-align:middle;">×</span>
<img src="https://raw.githubusercontent.com/ErisPulse/ErisPulse/main/.github/assets/ErisPulseLogo.png" height="120" alt="ErisPulse" />

# MyModule
（徽章行同上）
</div>

可按需追加 GitHub Stars、Downloads 等徽章。Logo 也可下载到项目本地（.github/assets/ErisPulseLogo.png）改为相对路径引用。

相关文档

• 模块开发入门 - 创建第一个模块
• 模块核心概念 - 理解模块架构
• Event 包装类 - 事件处理详解