82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
from astrbot.api.event import AstrMessageEvent
|
|
from astrbot.api.provider import LLMResponse
|
|
from astrbot.api import logger
|
|
from astrbot.core.message.message_event_result import MessageChain,Plain
|
|
from ..tool.pattern import Pattern
|
|
from astrbot.api.star import Star
|
|
import asyncio
|
|
|
|
class TimerService:
|
|
def __init__(self,star:Star) -> None:
|
|
self.star = star
|
|
self.tasks:list[TimerTask] = []
|
|
pass
|
|
|
|
# 解析模板并进行定时
|
|
def parse(self,event:AstrMessageEvent):
|
|
res = event.get_result()
|
|
if not res:
|
|
logger.info("[wait_msg]:解析失败：没拿到消息链")
|
|
return
|
|
|
|
Second:int|None = None
|
|
logger.info("[wait_msg]:开始解析模板")
|
|
|
|
for index,i in enumerate(res.chain) :
|
|
logger.info(f"获得消息链类型:{i.type}内容:{i}")
|
|
if not isinstance(i,Plain) or not i.type == "Plain":
|
|
continue
|
|
dump = Pattern(i.text)
|
|
i.text = dump.string
|
|
Second = dump.count
|
|
# 改变信息
|
|
res.chain[index] = i
|
|
|
|
event.set_result(res)
|
|
if not Second is None:
|
|
self.timer(event,Second)
|
|
return
|
|
|
|
|
|
# 定时:
|
|
def timer(self,event:AstrMessageEvent,Sec:int):
|
|
logger.debug(f"进行一个{Sec}秒的对{event.get_sender_id}({event.get_sender_name})的等")
|
|
task = asyncio.create_task(self.timer_task(event,Sec))
|
|
self.tasks.append(TimerTask(event=event,task=task))
|
|
|
|
# 等待,唤醒任务本体
|
|
async def timer_task(self,event:AstrMessageEvent,second:int):
|
|
await asyncio.sleep(second)
|
|
await self.response_time(event)
|
|
|
|
# 调用llm
|
|
async def response_time(self,event:AstrMessageEvent):
|
|
try:
|
|
umo = event.unified_msg_origin
|
|
conv_mnger = self.star.context.conversation_manager
|
|
provider =await self.star.context.get_current_chat_provider_id(umo)
|
|
conv_id = await conv_mnger.get_curr_conversation_id(umo)
|
|
if conv_id is None:
|
|
logger.error(f"conversation_id获取失败!!umo:{umo}")
|
|
return False
|
|
conversation = await conv_mnger.get_conversation(umo,conv_id)
|
|
|
|
except:
|
|
logger.error("消息管理失败!")
|
|
return False
|
|
logger.debug(f"获取上下文:{conversation}")
|
|
req = await self.star.context.llm_generate(
|
|
chat_provider_id=provider,
|
|
prompt="请输入你的上下文,这里是测试环境，你的输出将会被打印至控制台",
|
|
# contexts=conversation
|
|
|
|
)
|
|
logger.debug(f"llm结果:{req.completion_text }")
|
|
|
|
class TimerTask:
|
|
def __init__(self,event:AstrMessageEvent,task:asyncio.Task) -> None:
|
|
self.event = event
|
|
self.task = task
|
|
pass
|
|
|