JCbot插件开发教程

[西柚木吉]

JCbot 插件开发详细文档​

概述​

我制作的插件如何被其他用户看到?​

可以到官方gitee仓库提交issues反馈数据仓库

JCbot 是一个功能强大的机器人框架，支持通过插件系统扩展功能。本文档将详细介绍如何为 JCbot 开发插件，包括插件结构、可用接口、配置管理和最佳实践。​

插件基础结构​

1. 插件目录结构​

每个插件应该位于 Configs/JCbot/plugins/目录下的独立文件夹中：

Configs/JCbot/plugins/
└── YourPluginName/          # 插件目录（与插件名相同）
    ├── YourPluginName.py    # 插件主文件（必需）
    ├── __init__.py          # 可选的包初始化文件
    ├── config.json          # 插件配置文件（自动生成）
    ├── requirements.txt     # 依赖声明文件（可选）
    └── other_files/         # 其他资源文件

​

2. 插件基类：JCbotPlugin​

所有插件必须继承自 JCbotPlugin基类：

from typing import Dict, List, Optional, Set, Tuple, Callable, Any, Union
import os
import json
import time
import random
import datetime
import re
import aiohttp
import asyncio

# 正确的导入方式
from JCbotMain import JCbotPlugin

class YourPlugin(JCbotPlugin):
    def __init__(self, bot_instance):
        super().__init__(bot_instance)
        self.plugin_name = "YourPluginName"  # 必须与目录名相同
        self.version = "1.0.0"
        self.description = "你的插件描述"
    
    def register(self):
        """插件注册时调用"""
        self.logger.info(f"插件 {self.plugin_name} 已注册")
    
    def get_handlers(self):
        """返回命令处理器字典"""
        return {
            "your_command_handler": self.handle_your_command,
            "another_handler": self.handle_another
        }
    
    def get_commands_config(self):
        """返回要添加到命令配置中的字典"""
        return {
            "你的命令": {
                "description": "命令描述",
                "handler": "your_command_handler",
                "permission": 1,
                "fuzzy_match": False,
                "group": "你的分组",
                "options": {
                    "key": "value"
                }
            }
        }
    
    def get_default_config(self):
        """获取默认配置"""
        return {
            "setting1": "default_value",
            "setting2": 100,
            "enabled": True
        }

​

插件生命周期​

1. 加载阶段：插件被发现并导入
2. 初始化阶段：调用 __init__()和 register()方法
3. 配置加载：自动调用 load_config()加载配置
4. 运行阶段：处理命令和事件
5. 重载阶段：插件被重新加载时调用
6. 卸载阶段：插件被卸载时清理资源

​

可用接口和方法​

1. 核心方法（必须实现）​

get_handlers() -> Dict[str, Callable]​

返回插件提供的命令处理器字典。

def get_handlers(self):
    return {
        "weather": self.handle_weather,
        "timer": self.handle_timer
    }

​

get_commands_config() -> Dict​

返回要添加到机器人命令配置中的命令定义。

def get_commands_config(self):
    return {
        "天气": {
            "description": "查询天气信息",
            "handler": "weather",
            "permission": 1,
            "fuzzy_match": False,
            "group": "实用工具"
        }
    }

​

register()​

插件注册时调用，用于执行初始化操作。

def register(self):
    self.logger.info(f"插件 {self.plugin_name} 初始化完成")
    # 初始化操作

​

2. 配置管理方法​

get_default_config() -> Dict​

返回插件的默认配置。

def get_default_config(self):
    return {
        "api_key": "",
        "refresh_interval": 300,
        "enabled": True
    }

​

save_config()​

保存插件配置到文件。

def update_config(self, new_config):
    self.config.update(new_config)
    self.save_config()

​

访问机器人实例（bot）​

插件可以通过 self.bot访问机器人实例的所有功能：​

1. 配置信息​

# 机器人主配置
bot_config = self.bot.bot_config

# 游戏服务器配置
game_config = self.bot.game_server_config

# 权限配置
permissions = self.bot.permissions

​

2. 发送消息​

# 发送群消息
await self.bot.send_group_message(group_id, "消息内容")

# 发送艾特消息
await self.bot.send_at_message(group_id, user_id, "消息内容")

​

3. 权限检查​

# 获取用户权限级别
user_permission = self.bot.get_user_permission(user_id)

# 权限级别常量
from JCbotMain import PERMISSION_NONE, PERMISSION_MEMBER, PERMISSION_ADMIN, PERMISSION_SUPER_ADMIN

if user_permission >= PERMISSION_ADMIN:
    # 管理员操作

​

4. 游戏服务器交互​

# 执行游戏命令
result = await self.bot.execute_command("say Hello World")

# 获取在线玩家
players_info = await self.bot.fetch_players_info()

​

5. 工具函数​

# 格式化时间
duration = self.bot.format_duration(3600)  # 返回 "1小时"

# 获取系统状态
cpu_usage = self.bot.get_cpu_usage()
memory_usage = self.bot.get_memory_usage()

​

命令处理器示例​

async def handle_your_command(self, group_id: int, user_id: int, command: str, param: str, options: Dict = None) -> str:
    """
    处理命令的方法
    """
    
    # 检查权限
    user_permission = self.bot.get_user_permission(user_id)
    if user_permission < PERMISSION_ADMIN:
        return "权限不足，需要管理员权限"
    
    # 处理参数
    if not param:
        return "请输入参数"
    
    try:
        # 执行操作
        result = await self.do_something(param)
        return f"操作成功: {result}"
        
    except Exception as e:
        self.logger.error(f"处理命令时出错: {e}")
        return f"操作失败: {e}"

​

高级功能​

1. 定时任务​

def register(self):
    # 创建定时任务
    self.schedule_task = asyncio.create_task(self.periodic_task())

async def periodic_task(self):
    while True:
        try:
            await self.check_updates()
            await asyncio.sleep(300)
        except Exception as e:
            self.logger.error(f"定时任务出错: {e}")
            await asyncio.sleep(60)

​

2. Web API 调用​

async def call_api(self, endpoint, params=None):
    """调用API的辅助方法"""
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(f"https://api.example.com/{endpoint}", 
                                 params=params, timeout=10) as response:
                if response.status == 200:
                    return await response.json()
                return None
    except Exception as e:
        self.logger.error(f"API调用失败: {e}")
        return None

​

依赖管理​

在 requirements.txt中声明依赖：

requests>=2.25.0
beautifulsoup4>=4.9.0
python-dateutil>=2.8.0

​

完整插件示例​

# -*- coding: utf-8 -*-
from typing import Dict, List, Optional, Set, Tuple, Callable, Any, Union
import os
import json
import aiohttp
import asyncio
from datetime import datetime

from JCbotMain import JCbotPlugin

class WeatherPlugin(JCbotPlugin):
    def __init__(self, bot_instance):
        super().__init__(bot_instance)
        self.plugin_name = "Weather"
        self.version = "1.0.0"
        self.description = "天气预报插件"
        
    def register(self):
        self.logger.info(f"天气预报插件已加载")
        
    def get_handlers(self):
        return {"weather": self.handle_weather}
        
    def get_commands_config(self):
        return {
            "天气": {
                "description": "查询天气预报",
                "handler": "weather",
                "permission": 1,
                "fuzzy_match": False,
                "group": "实用工具"
            }
        }
        
    def get_default_config(self):
        return {
            "api_key": "",
            "default_city": "北京",
            "refresh_interval": 3600
        }
        
    async def handle_weather(self, group_id: int, user_id: int, command: str, 
                           param: str, options: Dict = None) -> str:
        if not self.config.get("api_key"):
            return "请先设置天气API密钥"
            
        city = param.strip() if param else self.config.get("default_city", "北京")
        if not city:
            return "请输入城市名称"
            
        try:
            weather_data = await self.get_weather_data(city)
            response = self.format_weather_response(weather_data, city)
            return response
            
        except Exception as e:
            self.logger.error(f"获取天气信息失败: {e}")
            return f"获取天气信息失败: {e}"
            
    async def get_weather_data(self, city: str) -> Dict:
        api_key = self.config.get("api_key")
        url = "http://api.openweathermap.org/data/2.5/weather"
        
        params = {
            "q": city,
            "appid": api_key,
            "units": "metric",
            "lang": "zh_cn"
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params, timeout=10) as response:
                if response.status == 200:
                    return await response.json()
                raise Exception(f"API返回错误: {response.status}")
                    
    def format_weather_response(self, data: Dict, city: str) -> str:
        if data.get("cod") != 200:
            return f"无法获取 {city} 的天气信息"
            
        main = data.get("main", {})
        weather = data.get("weather", [{}])[0]
        
        return (
            f"{city}天气预报:\n"
            f"天气: {weather.get('description', '未知')}\n"
            f"温度: {main.get('temp', '未知')}°C\n"
            f"湿度: {main.get('humidity', '未知')}%"
        )

​

调试和测试​

1. 日志调试​

# 在代码中添加详细的日志
self.logger.debug(f"收到命令: {command}, 参数: {param}")
self.logger.info(f"用户 {user_id} 使用了功能")

​

2. 错误处理​

async def handle_command(self, group_id, user_id, command, param, options):
    try:
        # 主要逻辑
        
    except ConnectionError as e:
        self.logger.error(f"网络连接错误: {e}")
        return "网络连接失败，请稍后重试"
        
    except ValueError as e:
        self.logger.warning(f"参数错误: {e}")
        return "参数格式错误，请检查输入"
        
    except Exception as e:
        self.logger.exception(f"未预期的错误: {e}")
        return "系统错误，请联系管理员"

​

发布和分享​

1. 确保插件有清晰的文档
2. 提供示例配置和使用方法
3. 测试插件在不同环境下的兼容性
4. 可以提交到插件仓库共享给其他用户

​

注意事项​

注意事项	说明
性能考虑	避免阻塞操作，使用异步IO
错误处理	妥善处理所有可能的异常
资源清理	在插件卸载时释放资源
配置备份	重要配置应该有备份机制
权限控制	遵循最小权限原则

通过本文档，你应该能够全面了解如何为 JCbot 开发功能丰富的插件。如有任何问题，可以参考现有插件的实现或查阅相关API文档。​