feat(sleep): 为睡眠系统添加随机时间偏移功能

- 新增每日睡眠和起床时间随机偏移量配置选项
- 实现缓存机制确保同一天内使用相同的偏移量
- 重构睡眠时间检查逻辑以支持动态时间偏移
- 更新相关配置类和插件清单格式
This commit is contained in:
雅诺狐
2025-09-07 08:20:39 +08:00
parent 653599b7e7
commit cb994a4e17
6 changed files with 55 additions and 13 deletions

View File

@@ -1,5 +1,6 @@
from datetime import datetime, time
from datetime import datetime, time, timedelta
from typing import Optional
import random
from src.common.logger import get_logger
from src.config.config import global_config
@@ -10,15 +11,37 @@ logger = get_logger("time_checker")
class TimeChecker:
def __init__(self, schedule_source):
self.schedule_source = schedule_source
# 缓存当天的偏移量,确保一天内使用相同的偏移量
self._daily_sleep_offset = None
self._daily_wake_offset = None
self._offset_date = None
def _get_daily_offsets(self):
"""获取当天的睡眠和起床时间偏移量,每天生成一次"""
today = datetime.now().date()
# 如果是新的一天,重新生成偏移量
if self._offset_date != today:
sleep_offset_range = global_config.sleep_system.sleep_time_offset_minutes
wake_offset_range = global_config.sleep_system.wake_up_time_offset_minutes
# 生成 ±offset_range 范围内的随机偏移量
self._daily_sleep_offset = random.randint(-sleep_offset_range, sleep_offset_range)
self._daily_wake_offset = random.randint(-wake_offset_range, wake_offset_range)
self._offset_date = today
logger.debug(f"生成新的每日偏移量 - 睡觉时间偏移: {self._daily_sleep_offset}分钟, 起床时间偏移: {self._daily_wake_offset}分钟")
return self._daily_sleep_offset, self._daily_wake_offset
def is_in_theoretical_sleep_time(self, now_time: time) -> tuple[bool, Optional[str]]:
if global_config.sleep_system.sleep_by_schedule:
if self.schedule_source.get_today_schedule():
return self._is_in_schedule_sleep_time(now_time)
else:
return self._is_in_fixed_sleep_time(now_time)
return self._is_in_sleep_time(now_time)
else:
return self._is_in_fixed_sleep_time(now_time)
return self._is_in_sleep_time(now_time)
def _is_in_schedule_sleep_time(self, now_time: time) -> tuple[bool, Optional[str]]:
"""检查当前时间是否落在日程表的任何一个睡眠活动中"""
@@ -49,20 +72,33 @@ class TimeChecker:
continue
return False, None
def _is_in_fixed_sleep_time(self, now_time: time) -> tuple[bool, Optional[str]]:
"""检查当前时间是否在固定的睡眠时间内"""
def _is_in_sleep_time(self, now_time: time) -> tuple[bool, Optional[str]]:
"""检查当前时间是否在固定的睡眠时间内(应用偏移量)"""
try:
start_time_str = global_config.sleep_system.fixed_sleep_time
end_time_str = global_config.sleep_system.fixed_wake_up_time
start_time = datetime.strptime(start_time_str, "%H:%M").time()
end_time = datetime.strptime(end_time_str, "%H:%M").time()
# 获取当天的偏移量
sleep_offset, wake_offset = self._get_daily_offsets()
# 解析基础时间
base_start_time = datetime.strptime(start_time_str, "%H:%M")
base_end_time = datetime.strptime(end_time_str, "%H:%M")
# 应用偏移量
actual_start_time = (base_start_time + timedelta(minutes=sleep_offset)).time()
actual_end_time = (base_end_time + timedelta(minutes=wake_offset)).time()
logger.debug(f"固定睡眠时间检查 - 基础时间: {start_time_str}-{end_time_str}, "
f"偏移后时间: {actual_start_time.strftime('%H:%M')}-{actual_end_time.strftime('%H:%M')}, "
f"当前时间: {now_time.strftime('%H:%M')}")
if start_time <= end_time:
if start_time <= now_time < end_time:
return True, "固定睡眠时间"
if actual_start_time <= actual_end_time:
if actual_start_time <= now_time < actual_end_time:
return True, f"固定睡眠时间(偏移后: {actual_start_time.strftime('%H:%M')}-{actual_end_time.strftime('%H:%M')})"
else:
if now_time >= start_time or now_time < end_time:
return True, "固定睡眠时间"
if now_time >= actual_start_time or now_time < actual_end_time:
return True, f"固定睡眠时间(偏移后: {actual_start_time.strftime('%H:%M')}-{actual_end_time.strftime('%H:%M')})"
except ValueError as e:
logger.error(f"固定的睡眠时间格式不正确,请使用 HH:MM 格式: {e}")
return False, None

View File

@@ -283,6 +283,7 @@ class DefaultReplyer:
return False, None, None
from src.plugin_system.core.event_manager import event_manager
# 触发 POST_LLM 事件(请求 LLM 之前)
if not from_plugin:
result = await event_manager.trigger_event(
EventType.POST_LLM, plugin_name="SYSTEM", prompt=prompt, stream_id=stream_id
@@ -304,6 +305,7 @@ class DefaultReplyer:
"model": model_name,
"tool_calls": tool_call,
}
# 触发 AFTER_LLM 事件
if not from_plugin:
result = await event_manager.trigger_event(