liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

加入bot，自动注入内容，在项目启动完成之后，给每个插件注入bot

Browse Source
This commit is contained in:
liuwei
2025-05-20 15:10:26 +08:00
parent 322297a69c
commit da89eea4f1
6 changed files with 69 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
main.py
View File

@@ -110,7 +110,7 @@ def jobs(robot: Robot):
# ✅ 每 3 小时登录验证
@async_job.at_times(["14:43"])
async def login_check_job():
robot.login_twice_auto_auth()
await robot.login_twice_auto_auth()
@async_job.at_times(["11:15"])
async def update_image_cache_job():

5
plugin_common/message_plugin_interface.py
View File

@@ -1,5 +1,6 @@
from typing import Dict, Any, Tuple, Optional, List
from plugin_common.plugin_interface import PluginInterface
from wechat_ipad import WechatAPIClient
class MessagePluginInterface(PluginInterface):
@@ -15,6 +16,10 @@ class MessagePluginInterface(PluginInterface):
"""支持的命令列表"""
return []
# 需要完成jobs 的业务，所以完成bot注入
def set_bot(self, bot: WechatAPIClient) -> None:
self.bot: WechatAPIClient = bot
def can_process(self, message: Dict[str, Any]) -> bool:
"""
检查插件是否可以处理该消息

51
plugin_common/plugin_interface.py
View File

@@ -8,55 +8,55 @@ from typing import Dict, Any, List, Optional, Tuple
class PluginStatus(Enum):
"""插件状态枚举"""
UNLOADED = 0 # 未加载
LOADED = 1 # 已加载但未启动
RUNNING = 2 # 运行中
STOPPED = 3 # 已停止
ERROR = 4 # 错误状态
UNLOADED = 0 # 未加载
LOADED = 1 # 已加载但未启动
RUNNING = 2 # 运行中
STOPPED = 3 # 已停止
ERROR = 4 # 错误状态
class PluginInterface(ABC):
"""插件基础接口，所有插件必须实现此接口"""
@property
@abstractmethod
def name(self) -> str:
"""插件名称"""
pass
@property
@abstractmethod
def version(self) -> str:
"""插件版本"""
pass
@property
@abstractmethod
def description(self) -> str:
"""插件描述"""
pass
@property
@abstractmethod
def author(self) -> str:
"""插件作者"""
pass
@property
def dependencies(self) -> List[str]:
"""插件依赖，返回依赖的其他插件名称列表"""
return []
@property
def status(self) -> PluginStatus:
"""获取插件当前状态"""
return self._status
@status.setter
def status(self, value: PluginStatus):
"""设置插件状态"""
self._status = value
def __init__(self):
"""初始化插件"""
self._status = PluginStatus.UNLOADED
@@ -64,7 +64,8 @@ class PluginInterface(ABC):
self._plugin_path = ""
# 初始化日志记录器
self.LOG = logger
self.bot = None
def load_config(self) -> bool:
"""
从插件目录下的config.toml加载配置
@@ -87,7 +88,7 @@ class PluginInterface(ABC):
except Exception as e:
self.LOG.info(f"加载插件配置失败: {e}")
return False
def set_plugin_path(self, path: str) -> None:
"""
设置插件路径
@@ -96,7 +97,7 @@ class PluginInterface(ABC):
path: 插件路径
"""
self._plugin_path = path
def get_plugin_path(self) -> str:
"""
获取插件路径
@@ -105,7 +106,7 @@ class PluginInterface(ABC):
插件路径
"""
return self._plugin_path
@abstractmethod
def initialize(self, context: Dict[str, Any]) -> bool:
"""
@@ -118,7 +119,7 @@ class PluginInterface(ABC):
初始化是否成功
"""
pass
@abstractmethod
def start(self) -> bool:
"""
@@ -128,7 +129,7 @@ class PluginInterface(ABC):
启动是否成功
"""
pass
@abstractmethod
def stop(self) -> bool:
"""
@@ -138,7 +139,7 @@ class PluginInterface(ABC):
停止是否成功
"""
pass
def configure(self, config: Dict[str, Any]) -> bool:
"""
配置插件
@@ -151,7 +152,7 @@ class PluginInterface(ABC):
"""
self._config.update(config)
return True
def get_config(self) -> Dict[str, Any]:
"""
获取插件配置
@@ -160,7 +161,7 @@ class PluginInterface(ABC):
插件配置
"""
return self._config
def cleanup(self) -> bool:
"""
清理插件资源，在卸载前调用
@@ -169,10 +170,10 @@ class PluginInterface(ABC):
清理是否成功
"""
return True
def get_config_path(self) -> str:
return os.path.join(self._plugin_path, 'config.toml')
def can_process(self, data: Any) -> bool:
"""检查插件是否可以处理给定的数据
@@ -199,4 +200,4 @@ class PluginInterface(ABC):
Returns:
(是否已处理, 处理结果)
"""
raise NotImplementedError("子类必须实现此方法")
raise NotImplementedError("子类必须实现此方法")

42
plugin_common/plugin_manager.py
View File

@@ -11,14 +11,15 @@ from plugin_common.message_plugin_interface import MessagePluginInterface
from plugin_common.scheduled_plugin_interface import ScheduledPluginInterface
from plugin_common.plugin_registry import PluginRegistry
from plugin_common.event_system import EventSystem, EventType
from wechat_ipad import WechatAPIClient
class PluginManager:
"""插件管理器，负责加载、卸载、启动、停止插件"""
# 单例实例
_instance = None
@classmethod
def get_instance(cls, plugin_dir=None):
"""获取单例实例
@@ -32,14 +33,13 @@ class PluginManager:
if cls._instance is None:
cls._instance = cls(plugin_dir=plugin_dir or "plugins")
return cls._instance
def __new__(cls, *args, **kwargs):
"""实现单例模式"""
if cls._instance is None:
cls._instance = super(PluginManager, cls).__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self, plugin_dir: str = "plugins"):
"""
@@ -154,7 +154,7 @@ class PluginManager:
self.LOG.warning(f"PluginManager：加载失败的插件模块: {failed_modules}")
self.LOG.info(f"PluginManager：当前已加载的插件实例: {list(self.plugins.keys())}")
self.LOG.info(f"PluginManager：最终的模块映射关系: {self.module_to_display}")
return self.plugins
def _get_module_name_from_plugin(self, plugin: PluginInterface) -> Optional[str]:
@@ -171,7 +171,7 @@ class PluginManager:
# 获取完整模块路径
full_module = plugin.__class__.__module__
module_parts = full_module.split('.')
# 处理不同的模块路径情况
if len(module_parts) >= 2 and module_parts[0] == 'plugins':
# 对于目录插件，模块名在第二个位置
@@ -270,10 +270,10 @@ class PluginManager:
# 获取显示名称
display_name = plugin.name
# 存储插件实例
self.plugins[display_name] = plugin
# 添加模块名到显示名的映射
self.module_to_display[module_name] = display_name
# self.LOG.info(f"PluginManager：添加模块映射 {module_name} -> {display_name}")
@@ -317,7 +317,7 @@ class PluginManager:
# 获取显示名称
display_name = plugin.name
# 存储插件实例
self.plugins[display_name] = plugin
@@ -403,7 +403,7 @@ class PluginManager:
# 记录原插件状态和模块名
was_running = plugin.status == PluginStatus.RUNNING
module_name = self._get_module_name_from_plugin(plugin)
if not module_name:
self.LOG.error(f"无法获取插件 {display_name} 的模块名，重载失败")
return None
@@ -546,21 +546,21 @@ class PluginManager:
self.module_to_display[module_name] = display_name
self.LOG.info(f"PluginManager：添加缺失的模块映射 {module_name} -> {display_name}")
return display_name, plugin
# 不区分大小写比较
if module_name and module_name.lower() == name.lower():
if module_name not in self.module_to_display:
self.module_to_display[module_name] = display_name
return display_name, plugin
# 检查名称是否包含在模块名中（不区分大小写）
if module_name and name.lower() in module_name.lower():
return display_name, plugin
# 检查模块名是否包含在名称中（不区分大小写）
if module_name and module_name.lower() in name.lower():
return display_name, plugin
# 检查名称是否包含在显示名称中（不区分大小写）
if name.lower() in display_name.lower():
return display_name, plugin
@@ -570,5 +570,15 @@ class PluginManager:
# 记录未找到插件的详细信息，帮助调试
self.LOG.warning(f"未找到插件: {name}，当前已加载插件: {list(self.plugins.keys())}")
self.LOG.warning(f"模块映射: {self.module_to_display}")
return None, None
return None, None
def inject_bot(self, bot: WechatAPIClient):
self.system_context["bot"] = bot
for name, plugin in self.plugins.items():
if hasattr(plugin, "set_bot"):
try:
plugin.set_bot(bot)
self.LOG.info(f"已成功注入 bot 到插件 {name}")
except Exception as e:
self.LOG.error(f"注入 bot 到插件 {name} 失败: {e}")

2
plugins/game_task/main.py
View File

@@ -46,7 +46,6 @@ class GameTaskPlugin(MessagePluginInterface):
def __init__(self):
super().__init__()
self.LOG = logger
self.bot: WechatAPIClient = None
async_job.at_times(["14:43"])(self.run_random_task_assignment)
def initialize(self, context: Dict[str, Any]) -> bool:
@@ -117,7 +116,6 @@ class GameTaskPlugin(MessagePluginInterface):
gbm: GroupBotManager = message.get("gbm")
all_contacts = message.get("all_contacts", {})
self.bot: WechatAPIClient = message.get("bot")
self.LOG.debug(f"插件执行： {self.name}：{content}")
# 检查权限

20
robot.py
View File

@@ -214,7 +214,7 @@ class Robot:
self.LOG.error(f"获取新消息失败 {e}")
if "用户可能退出" in str(e):
self.LOG.error(f"用户可能退出: {e}")
self.login_twice_auto_auth()
await self.login_twice_auto_auth()
await asyncio.sleep(5)
continue
@@ -296,7 +296,8 @@ class Robot:
self.nickname = self.ipad_bot.nickname
self.alias = self.ipad_bot.alias
self.phone = self.ipad_bot.phone
# 注入加载完成的bot
self.plugin_manager.inject_bot(self.ipad_bot)
self.LOG.info(
f"wechat_ipad登录账号信息: wxid: {self.wxid} 昵称: {self.nickname} 微信号: {self.alias} 手机号: {self.phone}")
break
@@ -315,7 +316,7 @@ class Robot:
self.LOG.error(f"wechat_ipad heartbeat: {e}")
if "用户可能退出" in str(e):
self.LOG.error(f"用户可能退出: {e}")
self.login_twice_auto_auth()
await self.login_twice_auto_auth()
await asyncio.sleep(60)
async def _heartbeat_task_long(self):
@@ -332,7 +333,7 @@ class Robot:
self.LOG.error(f"wechat_ipad heartbeat long: {e}")
if "用户可能退出" in str(e):
self.LOG.error(f"用户可能退出: {e}")
self.login_twice_auto_auth()
await self.login_twice_auto_auth()
await asyncio.sleep(120)
async def _process_ipad_message(self, message: WxMessage):
@@ -412,9 +413,9 @@ class Robot:
self.gbm.load_local_cache()
await self.ipad_bot.send_text_message("filehelper", "已更新")
elif content.clean_content == "PDF":
self.generate_sehuatang_pdf()
await self.generate_sehuatang_pdf()
elif content.clean_content == "排行榜":
self.generate_and_send_ranking()
await self.generate_and_send_ranking()
if is_group:
self.LOG.debug(f"入库和记录群消息: {message}")
@@ -614,10 +615,10 @@ class Robot:
self.contacts_db.delete_chatroom_all_info(group_id)
self.LOG.info("联系人信息刷新完成")
def login_twice_auto_auth(self) -> None:
async def login_twice_auto_auth(self) -> None:
try:
self.LOG.info(f"定时进行二次登录动作")
resp = self.ipad_bot.twice_auto_auth()
resp = await self.ipad_bot.twice_auto_auth()
if resp:
self.LOG.info(f"定时二次登录成功！")
if self.ipad_running:
@@ -632,7 +633,6 @@ class Robot:
# ============================================== 业务内容==========================================================
async def news_baidu_report_auto(self) -> None:
try:
news = News().get_baidu_news()
@@ -648,6 +648,7 @@ class Robot:
self.LOG.error(f"newsEnReport error：{e}")
# 使用装饰器标记定时任务 星期五 10:30 执行
async def send_epic_free_games(self):
try:
if is_friday():
@@ -657,6 +658,7 @@ class Robot:
self.LOG.error(f"sendEpicFreeGames error：{e}")
# 使用装饰器标记定时任务
async def message_count_to_db(self):
try:
self.message_storage.write_to_db()