feat(sleep): 引入休眠阶段消息唤醒机制与愤怒情绪模式

- 扩展日程休眠逻辑,支持在睡眠时段通过@/私聊唤醒
- 新增 WakeUpManager 持续追踪唤醒阈值并触发愤怒状态
- MoodManager 中追加“被吵醒”情绪标识与愤怒提示模板
- HeartFChat 主循环集成唤醒处理管线,兼容原逻辑
- 配置中心开放阈值、衰减、持续时间等可调参数

此举允许机器人即便在“睡觉”时也能被重要消息强行“闹醒”,并以生气口吻做出回应,提升拟人化交互体验。
This commit is contained in:
minecraft1024a
2025-08-21 16:19:34 +08:00
parent 74ec33580a
commit 4f7836ae0c
9 changed files with 609 additions and 11 deletions

293
tests/test_wakeup_system.py Normal file
View File

@@ -0,0 +1,293 @@
import pytest
import asyncio
import time
from unittest.mock import Mock, patch, MagicMock
import sys
import os
# 添加项目根目录到Python路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from src.chat.chat_loop.wakeup_manager import WakeUpManager
from src.chat.chat_loop.hfc_context import HfcContext
from src.config.official_configs import WakeUpSystemConfig
class TestWakeUpManager:
"""唤醒度管理器测试类"""
@pytest.fixture
def mock_context(self):
"""创建模拟的HFC上下文"""
context = Mock(spec=HfcContext)
context.stream_id = "test_chat_123"
context.log_prefix = "[TEST]"
context.running = True
return context
@pytest.fixture
def wakeup_config(self):
"""创建测试用的唤醒度配置"""
return WakeUpSystemConfig(
enable=True,
wakeup_threshold=15.0,
private_message_increment=3.0,
group_mention_increment=2.0,
decay_rate=0.2,
decay_interval=30.0,
angry_duration=300.0 # 5分钟
)
@pytest.fixture
def wakeup_manager(self, mock_context, wakeup_config):
"""创建唤醒度管理器实例"""
with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
mock_global_config.wakeup_system = wakeup_config
manager = WakeUpManager(mock_context)
return manager
def test_initialization(self, wakeup_manager, wakeup_config):
"""测试初始化"""
assert wakeup_manager.wakeup_value == 0.0
assert wakeup_manager.is_angry == False
assert wakeup_manager.wakeup_threshold == wakeup_config.wakeup_threshold
assert wakeup_manager.private_message_increment == wakeup_config.private_message_increment
assert wakeup_manager.group_mention_increment == wakeup_config.group_mention_increment
assert wakeup_manager.decay_rate == wakeup_config.decay_rate
assert wakeup_manager.decay_interval == wakeup_config.decay_interval
assert wakeup_manager.angry_duration == wakeup_config.angry_duration
assert wakeup_manager.enabled == wakeup_config.enable
@patch('src.manager.schedule_manager.schedule_manager')
@patch('src.mood.mood_manager.mood_manager')
def test_private_message_wakeup_accumulation(self, mock_mood_manager, mock_schedule_manager, wakeup_manager):
"""测试私聊消息唤醒度累积"""
# 模拟休眠状态
mock_schedule_manager.is_sleeping.return_value = True
# 发送5条私聊消息 (5 * 3.0 = 15.0,达到阈值)
for i in range(4):
result = wakeup_manager.add_wakeup_value(is_private_chat=True)
assert result == False # 前4条消息不应该触发唤醒
assert wakeup_manager.wakeup_value == (i + 1) * 3.0
# 第5条消息应该触发唤醒
result = wakeup_manager.add_wakeup_value(is_private_chat=True)
assert result == True
assert wakeup_manager.is_angry == True
assert wakeup_manager.wakeup_value == 0.0 # 唤醒后重置
# 验证情绪管理器被调用
mock_mood_manager.set_angry_from_wakeup.assert_called_once_with("test_chat_123")
@patch('src.manager.schedule_manager.schedule_manager')
@patch('src.mood.mood_manager.mood_manager')
def test_group_mention_wakeup_accumulation(self, mock_mood_manager, mock_schedule_manager, wakeup_manager):
"""测试群聊艾特消息唤醒度累积"""
# 模拟休眠状态
mock_schedule_manager.is_sleeping.return_value = True
# 发送7条群聊艾特消息 (7 * 2.0 = 14.0,未达到阈值)
for i in range(7):
result = wakeup_manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
assert result == False
assert wakeup_manager.wakeup_value == (i + 1) * 2.0
# 第8条消息应该触发唤醒 (8 * 2.0 = 16.0超过阈值15.0)
result = wakeup_manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
assert result == True
assert wakeup_manager.is_angry == True
assert wakeup_manager.wakeup_value == 0.0
# 验证情绪管理器被调用
mock_mood_manager.set_angry_from_wakeup.assert_called_once_with("test_chat_123")
@patch('src.manager.schedule_manager.schedule_manager')
def test_group_message_without_mention(self, mock_schedule_manager, wakeup_manager):
"""测试群聊未艾特消息不增加唤醒度"""
# 模拟休眠状态
mock_schedule_manager.is_sleeping.return_value = True
# 发送群聊消息但未被艾特
result = wakeup_manager.add_wakeup_value(is_private_chat=False, is_mentioned=False)
assert result == False
assert wakeup_manager.wakeup_value == 0.0 # 不应该增加
@patch('src.manager.schedule_manager.schedule_manager')
def test_no_accumulation_when_not_sleeping(self, mock_schedule_manager, wakeup_manager):
"""测试非休眠状态下不累积唤醒度"""
# 模拟非休眠状态
mock_schedule_manager.is_sleeping.return_value = False
# 发送私聊消息
result = wakeup_manager.add_wakeup_value(is_private_chat=True)
assert result == False
assert wakeup_manager.wakeup_value == 0.0 # 不应该增加
def test_disabled_system(self, mock_context):
"""测试系统禁用时的行为"""
disabled_config = WakeUpSystemConfig(enable=False)
with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
mock_global_config.wakeup_system = disabled_config
manager = WakeUpManager(mock_context)
with patch('src.manager.schedule_manager.schedule_manager') as mock_schedule_manager:
mock_schedule_manager.is_sleeping.return_value = True
# 即使发送消息也不应该累积唤醒度
result = manager.add_wakeup_value(is_private_chat=True)
assert result == False
assert manager.wakeup_value == 0.0
@patch('src.mood.mood_manager.mood_manager')
def test_angry_state_expiration(self, mock_mood_manager, wakeup_manager):
"""测试愤怒状态过期"""
# 手动设置愤怒状态
wakeup_manager.is_angry = True
wakeup_manager.angry_start_time = time.time() - 400 # 400秒前开始愤怒超过300秒持续时间
# 检查愤怒状态应该已过期
is_angry = wakeup_manager.is_in_angry_state()
assert is_angry == False
assert wakeup_manager.is_angry == False
# 验证情绪管理器被调用清除愤怒状态
mock_mood_manager.clear_angry_from_wakeup.assert_called_once_with("test_chat_123")
def test_angry_prompt_addition(self, wakeup_manager):
"""测试愤怒状态提示词"""
# 非愤怒状态
prompt = wakeup_manager.get_angry_prompt_addition()
assert prompt == ""
# 愤怒状态
wakeup_manager.is_angry = True
wakeup_manager.angry_start_time = time.time()
prompt = wakeup_manager.get_angry_prompt_addition()
assert "吵醒" in prompt and "生气" in prompt
def test_status_info(self, wakeup_manager):
"""测试状态信息获取"""
# 设置一些状态
wakeup_manager.wakeup_value = 10.5
wakeup_manager.is_angry = True
wakeup_manager.angry_start_time = time.time()
status = wakeup_manager.get_status_info()
assert status["wakeup_value"] == 10.5
assert status["wakeup_threshold"] == 15.0
assert status["is_angry"] == True
assert status["angry_remaining_time"] > 0
@pytest.mark.asyncio
async def test_decay_loop(self, wakeup_manager):
"""测试衰减循环"""
# 设置初始唤醒度
wakeup_manager.wakeup_value = 5.0
# 模拟一次衰减
with patch('asyncio.sleep') as mock_sleep:
# 创建一个会立即停止的衰减循环
wakeup_manager.context.running = False
# 手动调用衰减逻辑
if wakeup_manager.wakeup_value > 0:
old_value = wakeup_manager.wakeup_value
wakeup_manager.wakeup_value = max(0, wakeup_manager.wakeup_value - wakeup_manager.decay_rate)
assert wakeup_manager.wakeup_value == 4.8 # 5.0 - 0.2 = 4.8
@pytest.mark.asyncio
@patch('src.mood.mood_manager.mood_manager')
async def test_angry_state_expiration_in_decay_loop(self, mock_mood_manager, wakeup_manager):
"""测试衰减循环中愤怒状态过期"""
# 设置过期的愤怒状态
wakeup_manager.is_angry = True
wakeup_manager.angry_start_time = time.time() - 400 # 400秒前
# 手动调用衰减循环中的愤怒状态检查逻辑
current_time = time.time()
if wakeup_manager.is_angry and current_time - wakeup_manager.angry_start_time >= wakeup_manager.angry_duration:
wakeup_manager.is_angry = False
mock_mood_manager.clear_angry_from_wakeup(wakeup_manager.context.stream_id)
assert wakeup_manager.is_angry == False
mock_mood_manager.clear_angry_from_wakeup.assert_called_once_with("test_chat_123")
@pytest.mark.asyncio
async def test_start_stop_lifecycle(self, wakeup_manager):
"""测试启动和停止生命周期"""
# 测试启动
await wakeup_manager.start()
assert wakeup_manager._decay_task is not None
assert not wakeup_manager._decay_task.done()
# 测试停止
await wakeup_manager.stop()
assert wakeup_manager._decay_task.cancelled()
@pytest.mark.asyncio
async def test_disabled_system_start(self, mock_context):
"""测试禁用系统的启动行为"""
disabled_config = WakeUpSystemConfig(enable=False)
with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
mock_global_config.wakeup_system = disabled_config
manager = WakeUpManager(mock_context)
await manager.start()
assert manager._decay_task is None # 禁用时不应该创建衰减任务
class TestWakeUpSystemIntegration:
"""唤醒度系统集成测试"""
@patch('src.manager.schedule_manager.schedule_manager')
@patch('src.mood.mood_manager.mood_manager')
def test_mixed_message_types(self, mock_mood_manager, mock_schedule_manager):
"""测试混合消息类型的唤醒度累积"""
mock_schedule_manager.is_sleeping.return_value = True
# 创建配置和管理器
config = WakeUpSystemConfig(
enable=True,
wakeup_threshold=10.0, # 降低阈值便于测试
private_message_increment=3.0,
group_mention_increment=2.0,
decay_rate=0.2,
decay_interval=30.0,
angry_duration=300.0
)
context = Mock(spec=HfcContext)
context.stream_id = "test_mixed"
context.log_prefix = "[MIXED]"
context.running = True
with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
mock_global_config.wakeup_system = config
manager = WakeUpManager(context)
# 发送2条私聊消息 (2 * 3.0 = 6.0)
manager.add_wakeup_value(is_private_chat=True)
manager.add_wakeup_value(is_private_chat=True)
assert manager.wakeup_value == 6.0
# 发送2条群聊艾特消息 (2 * 2.0 = 4.0, 总计10.0)
manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
assert manager.wakeup_value == 8.0
# 最后一条消息触发唤醒
result = manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
assert result == True
assert manager.is_angry == True
assert manager.wakeup_value == 0.0
mock_mood_manager.set_angry_from_wakeup.assert_called_once_with("test_mixed")
if __name__ == "__main__":
# 运行测试
pytest.main([__file__, "-v"])