diff --git a/src/chat/emoji_system/emoji_manager.py b/src/chat/emoji_system/emoji_manager.py index b1a359d3c..6a9437757 100644 --- a/src/chat/emoji_system/emoji_manager.py +++ b/src/chat/emoji_system/emoji_manager.py @@ -1107,28 +1107,22 @@ class EmojiManager: if emoji_base64 is None: # 再次检查读取 logger.error(f"[注册失败] 无法读取图片以生成描述: {filename}") return False - task = asyncio.create_task(self.build_emoji_description(emoji_base64)) - def after_built_description(fut: asyncio.Future): - if fut.cancelled(): - logger.error(f"[注册失败] 描述生成任务被取消: {filename}") - elif fut.exception(): - logger.error(f"[注册失败] 描述生成任务出错 ({filename}): {fut.exception()}") - else: - description, emotions = fut.result() + # 等待描述生成完成 + description, emotions = await self.build_emoji_description(emoji_base64) - if not description: # 检查描述是否成功生成或审核通过 - logger.warning(f"[注册失败] 未能生成有效描述或审核未通过: {filename}") - # 删除未能生成描述的文件 - try: - os.remove(file_full_path) - logger.info(f"[清理] 删除描述生成失败的文件: {filename}") - except Exception as e: - logger.error(f"[错误] 删除描述生成失败文件时出错: {e!s}") - return False - new_emoji.description = description - new_emoji.emotion = emotions - task.add_done_callback(after_built_description) + if not description: # 检查描述是否成功生成或审核通过 + logger.warning(f"[注册失败] 未能生成有效描述或审核未通过: {filename}") + # 删除未能生成描述的文件 + try: + os.remove(file_full_path) + logger.info(f"[清理] 删除描述生成失败的文件: {filename}") + except Exception as e: + logger.error(f"[错误] 删除描述生成失败文件时出错: {e!s}") + return False + + new_emoji.description = description + new_emoji.emotion = emotions except Exception as build_desc_error: logger.error(f"[注册失败] 生成描述/情感时出错 ({filename}): {build_desc_error}") # 同样考虑删除文件 diff --git a/src/chat/message_receive/message_handler.py b/src/chat/message_receive/message_handler.py index a7238c2fd..29d78e2b3 100644 --- a/src/chat/message_receive/message_handler.py +++ b/src/chat/message_receive/message_handler.py @@ -134,7 +134,7 @@ class MessageHandler: predicate=_is_adapter_response, handler=self._handle_adapter_response_route, name="adapter_response_handler", - message_type="adapter_response", + priority=100 ) # 注册 notice 消息处理器（处理通知消息，如戳一戳、禁言等） @@ -152,7 +152,7 @@ class MessageHandler: predicate=_is_notice_message, handler=self._handle_notice_message, name="notice_message_handler", - message_type="notice", + priority=90 ) # 注册默认消息处理器（处理所有其他消息） @@ -160,6 +160,7 @@ class MessageHandler: predicate=lambda _: True, # 匹配所有消息 handler=self._handle_normal_message, name="default_message_handler", + priority=50 ) logger.info("MessageHandler 已向 MessageRuntime 注册处理器和钩子") @@ -314,7 +315,7 @@ class MessageHandler: # 触发 notice 事件（可供插件监听） await event_manager.trigger_event( EventType.ON_NOTICE_RECEIVED, - permission_group="USER", + permission_group="SYSTEM", message=message, notice_type=notice_type, chat_stream=chat, diff --git a/src/common/database/core/session.py b/src/common/database/core/session.py index b033088f9..9cd701fc2 100644 --- a/src/common/database/core/session.py +++ b/src/common/database/core/session.py @@ -126,6 +126,12 @@ async def get_db_session_direct() -> AsyncGenerator[AsyncSession, None]: 用于特殊场景，如需要完全独立的连接时。 一般情况下应使用 get_db_session()。 + 事务管理说明： + - 正常退出时自动提交事务 + - 发生异常时自动回滚事务 + - 如果用户代码已手动调用 commit/rollback，再次调用是安全的 + - 适用于所有数据库类型（SQLite, MySQL, PostgreSQL） + Yields: AsyncSession: SQLAlchemy异步会话对象 """ @@ -139,8 +145,16 @@ async def get_db_session_direct() -> AsyncGenerator[AsyncSession, None]: await _apply_session_settings(session, global_config.database.database_type) yield session + + # 正常退出时提交事务 + # 这对所有数据库都很重要，因为 SQLAlchemy 默认不是 autocommit 模式 + # 检查事务是否活动，避免在已回滚的事务上提交 + if session.is_active: + await session.commit() except Exception: - await session.rollback() + # 检查是否需要回滚（事务是否活动） + if session.is_active: + await session.rollback() raise finally: await session.close() diff --git a/src/common/database/optimization/batch_scheduler.py b/src/common/database/optimization/batch_scheduler.py index 821d0b1af..f2d2591fb 100644 --- a/src/common/database/optimization/batch_scheduler.py +++ b/src/common/database/optimization/batch_scheduler.py @@ -17,7 +17,7 @@ from typing import Any, TypeVar from sqlalchemy import delete, insert, select, update -from src.common.database.core.session import get_db_session +from src.common.database.core.session import get_db_session_direct from src.common.logger import get_logger from src.common.memory_utils import estimate_size_smart @@ -330,7 +330,7 @@ class AdaptiveBatchScheduler: operations: list[BatchOperation], ) -> None: """批量执行查询操作""" - async with get_db_session() as session: + async with get_db_session_direct() as session: for op in operations: try: # 构建查询 @@ -371,7 +371,7 @@ class AdaptiveBatchScheduler: operations: list[BatchOperation], ) -> None: """批量执行插入操作""" - async with get_db_session() as session: + async with get_db_session_direct() as session: try: # 收集数据，并过滤掉 id=None 的情况（让数据库自动生成） all_data = [] @@ -387,7 +387,7 @@ class AdaptiveBatchScheduler: # 批量插入 stmt = insert(operations[0].model_class).values(all_data) await session.execute(stmt) - await session.commit() + # 注意：commit 由 get_db_session_direct 上下文管理器自动处理 # 设置结果 for op in operations: @@ -402,20 +402,21 @@ class AdaptiveBatchScheduler: except Exception as e: logger.error(f"批量插入失败: {e}") - await session.rollback() + # 注意：rollback 由 get_db_session_direct 上下文管理器自动处理 for op in operations: if op.future and not op.future.done(): op.future.set_exception(e) + raise # 重新抛出异常以触发 rollback async def _execute_update_batch( self, operations: list[BatchOperation], ) -> None: """批量执行更新操作""" - async with get_db_session() as session: + async with get_db_session_direct() as session: results = [] try: - # 🔧 修复：收集所有操作后一次性commit，而不是循环中多次commit + # 🔧 收集所有操作后一次性commit，而不是循环中多次commit for op in operations: # 构建更新语句 stmt = update(op.model_class) @@ -430,8 +431,7 @@ class AdaptiveBatchScheduler: result = await session.execute(stmt) results.append((op, result.rowcount)) - # 所有操作成功后，一次性commit - await session.commit() + # 注意：commit 由 get_db_session_direct 上下文管理器自动处理 # 设置所有操作的结果 for op, rowcount in results: @@ -446,21 +446,22 @@ class AdaptiveBatchScheduler: except Exception as e: logger.error(f"批量更新失败: {e}") - await session.rollback() + # 注意：rollback 由 get_db_session_direct 上下文管理器自动处理 # 所有操作都失败 for op in operations: if op.future and not op.future.done(): op.future.set_exception(e) + raise # 重新抛出异常以触发 rollback async def _execute_delete_batch( self, operations: list[BatchOperation], ) -> None: """批量执行删除操作""" - async with get_db_session() as session: + async with get_db_session_direct() as session: results = [] try: - # 🔧 修复：收集所有操作后一次性commit，而不是循环中多次commit + # 🔧 收集所有操作后一次性commit，而不是循环中多次commit for op in operations: # 构建删除语句 stmt = delete(op.model_class) @@ -472,8 +473,7 @@ class AdaptiveBatchScheduler: result = await session.execute(stmt) results.append((op, result.rowcount)) - # 所有操作成功后，一次性commit - await session.commit() + # 注意：commit 由 get_db_session_direct 上下文管理器自动处理 # 设置所有操作的结果 for op, rowcount in results: @@ -488,11 +488,12 @@ class AdaptiveBatchScheduler: except Exception as e: logger.error(f"批量删除失败: {e}") - await session.rollback() + # 注意：rollback 由 get_db_session_direct 上下文管理器自动处理 # 所有操作都失败 for op in operations: if op.future and not op.future.done(): op.future.set_exception(e) + raise # 重新抛出异常以触发 rollback async def _adjust_parameters(self) -> None: """根据性能自适应调整参数""" diff --git a/src/common/database/optimization/connection_pool.py b/src/common/database/optimization/connection_pool.py index 4030bc061..ed7d3e5ef 100644 --- a/src/common/database/optimization/connection_pool.py +++ b/src/common/database/optimization/connection_pool.py @@ -123,6 +123,12 @@ class ConnectionPoolManager: """ 获取数据库会话的透明包装器 如果有可用连接则复用，否则创建新连接 + + 事务管理说明： + - 正常退出时自动提交事务 + - 发生异常时自动回滚事务 + - 如果用户代码已手动调用 commit/rollback，再次调用是安全的（空操作） + - 支持所有数据库类型：SQLite、MySQL、PostgreSQL """ connection_info = None @@ -151,21 +157,30 @@ class ConnectionPoolManager: yield connection_info.session - # 🔧 修复：正常退出时提交事务 - # 这对SQLite至关重要，因为SQLite没有autocommit + # 🔧 正常退出时提交事务 + # 这对所有数据库（SQLite、MySQL、PostgreSQL）都很重要 + # 因为 SQLAlchemy 默认使用事务模式，不会自动提交 + # 注意：如果用户代码已调用 commit()，这里的 commit() 是安全的空操作 if connection_info and connection_info.session: try: - await connection_info.session.commit() + # 检查事务是否处于活动状态，避免在已回滚的事务上提交 + if connection_info.session.is_active: + await connection_info.session.commit() except Exception as commit_error: logger.warning(f"提交事务时出错: {commit_error}") - await connection_info.session.rollback() + try: + await connection_info.session.rollback() + except Exception: + pass # 忽略回滚错误，因为事务可能已经结束 raise except Exception: # 发生错误时回滚连接 if connection_info and connection_info.session: try: - await connection_info.session.rollback() + # 检查是否需要回滚（事务是否活动） + if connection_info.session.is_active: + await connection_info.session.rollback() except Exception as rollback_error: logger.warning(f"回滚连接时出错: {rollback_error}") raise diff --git a/src/plugin_system/core/event_manager.py b/src/plugin_system/core/event_manager.py index f9e4de553..93f63642b 100644 --- a/src/plugin_system/core/event_manager.py +++ b/src/plugin_system/core/event_manager.py @@ -433,6 +433,7 @@ class EventManager: EventType.AFTER_LLM, EventType.POST_SEND, EventType.AFTER_SEND, + EventType.ON_NOTICE_RECEIVED ] for event_name in default_events: