better:日志系统现已可以动态更新

2025-03-26 22:42:19 +08:00
parent ee4a2f6e72
commit 572bffc273
11 changed files with 296 additions and 428 deletions
--- a/src/plugins/schedule/schedule_generator.py
+++ b/src/plugins/schedule/schedule_generator.py
@@ -1,159 +1,149 @@
 import datetime
-import json
-import re
-from typing import Dict, Union
-
-from nonebot import get_driver
-
+import os
+import sys
+from typing import Dict
+import asyncio
 # 添加项目根目录到 Python 路径
+root_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
+sys.path.append(root_path)

-from src.plugins.chat.config import global_config
-from ...common.database import db  # 使用正确的导入语法
-from ..models.utils_model import LLM_request
-from src.common.logger import get_module_logger
+from src.common.database import db # noqa: E402
+from src.common.logger import get_module_logger, SCHEDULE_STYLE_CONFIG, LogConfig # noqa: E402
+from src.plugins.models.utils_model import LLM_request # noqa: E402
+from src.plugins.chat.config import global_config # noqa: E402

-logger = get_module_logger("scheduler")

-driver = get_driver()
-config = driver.config
+schedule_config = LogConfig(
+    # 使用海马体专用样式
+    console_format=SCHEDULE_STYLE_CONFIG["console_format"],
+    file_format=SCHEDULE_STYLE_CONFIG["file_format"],
+)   
+logger = get_module_logger("scheduler", config=schedule_config)


 class ScheduleGenerator:
-    enable_output: bool = True
+    # enable_output: bool = True

-    def __init__(self):
-        # 根据global_config.llm_normal这一字典配置指定模型
-        # self.llm_scheduler = LLMModel(model = global_config.llm_normal,temperature=0.9)
-        self.llm_scheduler = LLM_request(model=global_config.llm_normal, temperature=0.9, request_type="scheduler")
+    def __init__(self, ):
+        # 使用离线LLM模型
+        self.llm_scheduler_all = LLM_request(
+            model= global_config.llm_reasoning, temperature=0.9, max_tokens=2048,request_type="schedule")
+        self.llm_scheduler_doing = LLM_request(
+            model= global_config.llm_normal, temperature=0.9, max_tokens=2048,request_type="schedule")
+        
        self.today_schedule_text = ""
-        self.today_schedule = {}
-        self.tomorrow_schedule_text = ""
-        self.tomorrow_schedule = {}
+        self.today_done_list = []
+
        self.yesterday_schedule_text = ""
-        self.yesterday_schedule = {}
+        self.yesterday_done_list = []

-    async def initialize(self):
+        self.name = ""
+        self.personality = ""
+        self.behavior = ""
+
+        self.start_time = datetime.datetime.now()
+        
+        self.schedule_doing_update_interval = 60 #最好大于60
+    
+    def initialize(self,name: str = "bot_name", personality: str = "你是一个爱国爱党的新时代青年", behavior: str = "你非常外向，喜欢尝试新事物和人交流",interval: int = 60):
+        """初始化日程系统"""
+        self.name = name
+        self.behavior = behavior
+        self.schedule_doing_update_interval = interval
+        
+        for pers in personality:
+            self.personality += pers + "\n"
+        
+
+    async def mai_schedule_start(self):
+        """启动日程系统，每5分钟执行一次move_doing，并在日期变化时重新检查日程"""
+        try:
+            logger.info(f"日程系统启动/刷新时间: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}")
+            # 初始化日程
+            await self.check_and_create_today_schedule()
+            self.print_schedule()
+            
+            while True:
+                print(self.get_current_num_task(1, True))
+                
+                current_time = datetime.datetime.now()
+                
+                # 检查是否需要重新生成日程（日期变化）
+                if current_time.date() != self.start_time.date():
+                    logger.info("检测到日期变化，重新生成日程")
+                    self.start_time = current_time
+                    await self.check_and_create_today_schedule()
+                    self.print_schedule()
+                
+                # 执行当前活动
+                current_activity = await self.move_doing(mind_thinking="")
+                logger.info(f"当前活动: {current_activity}")
+                
+                await asyncio.sleep(self.schedule_doing_update_interval) 
+                
+        except Exception as e:
+            logger.error(f"日程系统运行时出错: {str(e)}")
+            logger.exception("详细错误信息：")
+
+    async def check_and_create_today_schedule(self):
+        """检查昨天的日程，并确保今天有日程安排
+        
+        Returns:
+            tuple: (today_schedule_text, today_schedule) 今天的日程文本和解析后的日程字典
+        """
        today = datetime.datetime.now()
-        tomorrow = datetime.datetime.now() + datetime.timedelta(days=1)
-        yesterday = datetime.datetime.now() - datetime.timedelta(days=1)
+        yesterday = today - datetime.timedelta(days=1)
+        
+        # 先检查昨天的日程
+        self.yesterday_schedule_text, self.yesterday_done_list = self.load_schedule_from_db(yesterday)
+        if self.yesterday_schedule_text:
+            logger.debug(f"已加载{yesterday.strftime('%Y-%m-%d')}的日程")
+        
+        # 检查今天的日程
+        self.today_schedule_text, self.today_done_list = self.load_schedule_from_db(today)
+        if not self.today_done_list:
+            self.today_done_list = []
+        if not self.today_schedule_text:
+            logger.info(f"{today.strftime('%Y-%m-%d')}的日程不存在，准备生成新的日程")
+            self.today_schedule_text = await self.generate_daily_schedule(target_date=today)

-        self.today_schedule_text, self.today_schedule = await self.generate_daily_schedule(target_date=today)
-        self.tomorrow_schedule_text, self.tomorrow_schedule = await self.generate_daily_schedule(
-            target_date=tomorrow, read_only=True
-        )
-        self.yesterday_schedule_text, self.yesterday_schedule = await self.generate_daily_schedule(
-            target_date=yesterday, read_only=True
-        )
-
-    async def generate_daily_schedule(
-        self, target_date: datetime.datetime = None, read_only: bool = False
-    ) -> Dict[str, str]:
+        self.save_today_schedule_to_db()
+    
+    def construct_daytime_prompt(self, target_date: datetime.datetime):
        date_str = target_date.strftime("%Y-%m-%d")
        weekday = target_date.strftime("%A")

-        schedule_text = str
-
-        existing_schedule = db.schedule.find_one({"date": date_str})
-        if existing_schedule:
-            if self.enable_output:
-                logger.debug(f"{date_str}的日程已存在:")
-            schedule_text = existing_schedule["schedule"]
-            # print(self.schedule_text)
-
-        elif not read_only:
-            logger.debug(f"{date_str}的日程不存在，准备生成新的日程。")
-            prompt = (
-                f"""我是{global_config.BOT_NICKNAME}，{global_config.PROMPT_SCHEDULE_GEN}，请为我生成{date_str}（{weekday}）的日程安排，包括："""
-                + """
-            1. 早上的学习和工作安排
-            2. 下午的活动和任务
-            3. 晚上的计划和休息时间
-            请按照时间顺序列出具体时间点和对应的活动，用一个时间点而不是时间段来表示时间，用JSON格式返回日程表，
-            仅返回内容，不要返回注释，不要添加任何markdown或代码块样式，时间采用24小时制，
-            格式为{"时间": "活动","时间": "活动",...}。"""
-            )
-
-            try:
-                schedule_text, _, _ = await self.llm_scheduler.generate_response(prompt)
-                db.schedule.insert_one({"date": date_str, "schedule": schedule_text})
-                self.enable_output = True
-            except Exception as e:
-                logger.error(f"生成日程失败: {str(e)}")
-                schedule_text = "生成日程时出错了"
-            # print(self.schedule_text)
+        prompt =  f"你是{self.name}，{self.personality}，{self.behavior}"
+        prompt += f"你昨天的日程是：{self.yesterday_schedule_text}\n"
+        prompt += f"请为你生成{date_str}（{weekday}）的日程安排，结合你的个人特点和行为习惯\n"
+        prompt += "推测你的日程安排，包括你一天都在做什么，有什么发现和思考，具体一些，详细一些，需要1500字以上，精确到每半个小时，记得写明时间\n"
+        prompt += "直接返回你的日程，不要输出其他内容："
+        return prompt
+    
+    def construct_doing_prompt(self,time: datetime.datetime,mind_thinking: str = ""):
+        now_time = time.strftime("%H:%M")
+        if self.today_done_list:
+            previous_doings = self.get_current_num_task(10, True)
+            # print(previous_doings)
        else:
-            if self.enable_output:
-                logger.debug(f"{date_str}的日程不存在。")
-            schedule_text = "忘了"
-
-            return schedule_text, None
-
-        schedule_form = self._parse_schedule(schedule_text)
-        return schedule_text, schedule_form
-
-    def _parse_schedule(self, schedule_text: str) -> Union[bool, Dict[str, str]]:
-        """解析日程文本，转换为时间和活动的字典"""
-        try:
-            reg = r"\{(.|\r|\n)+\}"
-            matched = re.search(reg, schedule_text)[0]
-            schedule_dict = json.loads(matched)
-            self._check_schedule_validity(schedule_dict)
-            return schedule_dict
-        except json.JSONDecodeError:
-            logger.exception("解析日程失败: {}".format(schedule_text))
-            return False
-        except ValueError as e:
-            logger.exception(f"解析日程失败: {str(e)}")
-            return False
-        except Exception as e:
-            logger.exception(f"解析日程发生错误:{str(e)}")
-            return False
-
-    def _check_schedule_validity(self, schedule_dict: Dict[str, str]):
-        """检查日程是否合法"""
-        if not schedule_dict:
-            return
-        for time_str in schedule_dict.keys():
-            try:
-                self._parse_time(time_str)
-            except ValueError:
-                raise ValueError("日程时间格式不正确") from None
-
-    def _parse_time(self, time_str: str) -> str:
-        """解析时间字符串，转换为时间"""
-        return datetime.datetime.strptime(time_str, "%H:%M")
-
-    def get_current_task(self) -> str:
-        """获取当前时间应该进行的任务"""
-        current_time = datetime.datetime.now().strftime("%H:%M")
-
-        # 找到最接近当前时间的任务
-        closest_time = None
-        min_diff = float("inf")
-
-        # 检查今天的日程
-        if not self.today_schedule:
-            return "摸鱼"
-        for time_str in self.today_schedule.keys():
-            diff = abs(self._time_diff(current_time, time_str))
-            if closest_time is None or diff < min_diff:
-                closest_time = time_str
-                min_diff = diff
-
-        # 检查昨天的日程中的晚间任务
-        if self.yesterday_schedule:
-            for time_str in self.yesterday_schedule.keys():
-                if time_str >= "20:00":  # 只考虑晚上8点之后的任务
-                    # 计算与昨天这个时间点的差异（需要加24小时）
-                    diff = abs(self._time_diff(current_time, time_str))
-                    if diff < min_diff:
-                        closest_time = time_str
-                        min_diff = diff
-                        return closest_time, self.yesterday_schedule[closest_time]
-
-        if closest_time:
-            return closest_time, self.today_schedule[closest_time]
-        return "摸鱼"
+            previous_doings = "你没做什么事情"
+            
+        
+        prompt = f"你是{self.name}，{self.personality}，{self.behavior}"
+        prompt += f"你今天的日程是：{self.today_schedule_text}\n"
+        if mind_thinking:
+            prompt += f"你脑子里在想：{mind_thinking}\n"
+        prompt += f"你之前做了的事情是：{previous_doings}，从之前到现在已经过去了{self.schedule_doing_update_interval/60}分钟了\n"
+        prompt += f"现在是{now_time}，结合你的个人特点和行为习惯,"
+        prompt += "推测你现在做什么，具体一些，详细一些\n"
+        prompt += "直接返回你在做的事情，不要输出其他内容："
+        return prompt
+    
+    async def generate_daily_schedule(
+        self, target_date: datetime.datetime = None,) -> Dict[str, str]:
+        daytime_prompt = self.construct_daytime_prompt(target_date)
+        daytime_response,_ = await self.llm_scheduler_all.generate_response_async(daytime_prompt)
+        return daytime_response

    def _time_diff(self, time1: str, time2: str) -> int:
        """计算两个时间字符串之间的分钟差"""
@@ -174,14 +164,127 @@ class ScheduleGenerator:

    def print_schedule(self):
        """打印完整的日程安排"""
-        if not self._parse_schedule(self.today_schedule_text):
-            logger.warning("今日日程有误，将在两小时后重新生成")
+        if not self.today_schedule_text:
+            logger.warning("今日日程有误，将在下次运行时重新生成")
            db.schedule.delete_one({"date": datetime.datetime.now().strftime("%Y-%m-%d")})
        else:
            logger.info("=== 今日日程安排 ===")
-            for time_str, activity in self.today_schedule.items():
-                logger.info(f"时间[{time_str}]: 活动[{activity}]")
+            logger.info(self.today_schedule_text)
            logger.info("==================")
            self.enable_output = False
+
+    async def update_today_done_list(self):
+        # 更新数据库中的 today_done_list
+        today_str = datetime.datetime.now().strftime("%Y-%m-%d")
+        existing_schedule = db.schedule.find_one({"date": today_str})
+        
+        if existing_schedule:
+            # 更新数据库中的 today_done_list
+            db.schedule.update_one(
+                {"date": today_str},
+                {"$set": {"today_done_list": self.today_done_list}}
+            )
+            logger.debug(f"已更新{today_str}的已完成活动列表")
+        else:
+            logger.warning(f"未找到{today_str}的日程记录")
+
+    async def move_doing(self,mind_thinking: str = ""):
+        current_time = datetime.datetime.now()
+        doing_prompt = self.construct_doing_prompt(current_time,mind_thinking)
+        doing_response,_ = await self.llm_scheduler_doing.generate_response_async(doing_prompt)
+        self.today_done_list.append((current_time,doing_response))
+
+        await self.update_today_done_list()
+        
+        return doing_response
+
+    async def get_task_from_time_to_time(self, start_time: str, end_time: str):
+        """获取指定时间范围内的任务列表
+        
+        Args:
+            start_time (str): 开始时间，格式为"HH:MM"
+            end_time (str): 结束时间，格式为"HH:MM"
+            
+        Returns:
+            list: 时间范围内的任务列表
+        """
+        result = []
+        for task in self.today_done_list:
+            task_time = task[0]  # 获取任务的时间戳
+            task_time_str = task_time.strftime("%H:%M")
+            
+            # 检查任务时间是否在指定范围内
+            if self._time_diff(start_time, task_time_str) >= 0 and self._time_diff(task_time_str, end_time) >= 0:
+                result.append(task)
+        
+        return result
+
+    def get_current_num_task(self, num=1, time_info = False):
+        """获取最新加入的指定数量的日程
+        
+        Args:
+            num (int): 需要获取的日程数量，默认为1
+            
+        Returns:
+            list: 最新加入的日程列表
+        """
+        if not self.today_done_list:
+            return []
+            
+        # 确保num不超过列表长度
+        num = min(num, len(self.today_done_list))
+        pre_doings = ""
+        for doing in self.today_done_list[-num:]:
+            
+            if time_info:
+                time_str = doing[0].strftime("%H:%M")
+                pre_doings += time_str + "时，" + doing[1] + "\n"
+            else:
+                pre_doings += doing[1] + "\n"
+
+        # 返回最新的num条日程
+        return pre_doings
+
+    def save_today_schedule_to_db(self):
+        """保存日程到数据库，同时初始化 today_done_list"""
+        date_str = datetime.datetime.now().strftime("%Y-%m-%d")
+        schedule_data = {
+            "date": date_str,
+            "schedule": self.today_schedule_text,
+            "today_done_list": self.today_done_list if hasattr(self, 'today_done_list') else []
+        }
+        # 使用 upsert 操作，如果存在则更新，不存在则插入
+        db.schedule.update_one(
+            {"date": date_str},
+            {"$set": schedule_data},
+            upsert=True
+        )
+        logger.debug(f"已保存{date_str}的日程到数据库")
+
+    def load_schedule_from_db(self, date: datetime.datetime):
+        """从数据库加载日程，同时加载 today_done_list"""
+        date_str = date.strftime("%Y-%m-%d")
+        existing_schedule = db.schedule.find_one({"date": date_str})
+        
+        if existing_schedule:
+            schedule_text = existing_schedule["schedule"]
+            return schedule_text, existing_schedule.get("today_done_list", [])
+        else:
+            logger.debug(f"{date_str}的日程不存在")
+            return None, None
+
+async def main():
+    # 使用示例
+    scheduler = ScheduleGenerator()
+    scheduler.initialize(name="麦麦", personality="你叫麦麦，你19岁，是一个大二的女大学生，你有一头黑色短发，你会刷贴吧，你现在在学习心理学", behavior="你比较内向，一般熬夜比较晚，然后第二天早上10点起床吃早午饭",interval=60)
+    await scheduler.mai_schedule_start()
+    
+
+
 # 当作为组件导入时使用的实例
 bot_schedule = ScheduleGenerator()
+
+if __name__ == "__main__":
+    import asyncio
+    # 当直接运行此文件时执行
+    asyncio.run(main())