diff --git a/main.py b/main.py index cf0ac4d..431f881 100644 --- a/main.py +++ b/main.py @@ -1,9 +1,11 @@ from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult +from astrbot.core.message.message_event_result import MessageChain from astrbot.api.star import Context, Star, register from astrbot.api import logger -from astrbot.api.provider import ProviderRequest +from astrbot.api.provider import (ProviderRequest,LLMResponse ) from astrbot.core.agent.message import TextPart -@register("helloworld", "YourName", "一个简单的 Hello World 插件", "1.0.0") +from tool.pattern import Pattern +@register("wait_msg", "Wbucket", "一个简单的 llm等待回复插件", "0.0.1") class MyPlugin(Star): def __init__(self, context: Context): super().__init__(context) @@ -13,7 +15,7 @@ class MyPlugin(Star): # 注册指令的装饰器。指令名为 helloworld。注册成功后,发送 `/helloworld` 就会触发这个指令,并回复 `你好, {user_name}!` @filter.on_llm_request() - async def helloworld(self, event: AstrMessageEvent,req : ProviderRequest): + async def content_inject(self, event: AstrMessageEvent,req : ProviderRequest): """这是一个 hello world 指令""" # 这是 handler 的描述,将会被解析方便用户了解插件内容。建议填写。 # logger.info("系统提示词:"+req.system_prompt) req.extra_user_content_parts.append(TextPart( @@ -39,5 +41,12 @@ class MyPlugin(Star): # async def awa(self,event:AstrMessageEvent): # yield event.plain_result(event.get_message_str()) + @filter.on_llm_response() + async def pattern_extract(self,event:AstrMessageEvent,req:LLMResponse): + if not req.role == 'assistant': + return + llm_res = req.completion_text + dump = Pattern(llm_res) + await event.send(MessageChain().message("awa定时时间消息:{0}过滤后:{1}".format(dump.count,dump.string))) async def terminate(self): """可选择实现异步的插件销毁方法,当插件被卸载/停用时会调用。""" diff --git a/tool/pattern.py b/tool/pattern.py new file mode 100644 index 0000000..25a698a --- /dev/null +++ b/tool/pattern.py @@ -0,0 +1,48 @@ +import re + +class Pattern: + _pattern = r'((?!wt).)*' + + _MAX_SEC = 1800 + _MIN_SEC = 10 + # 定义最大与最小的范围,固定输入的规范 + def __init__(self,o_str:str) -> None: + res = self.extract(o_str) + self.count:int|None = res['second'] + self.string:str = res["content"] + + + def sec_range(self,i)->int: + if i > Pattern._MAX_SEC: + return Pattern._MAX_SEC + if i < Pattern._MIN_SEC : + return Pattern._MIN_SEC + return i + + + def load(self,pattern_str:str)->int|None: + content = pattern_str.strip().removeprefix("").removesuffix("").strip() + print(content) + try: + num = int(content) + return self.sec_range(num) + except: + return None + + def extract(self,o_str:str): + # 多次匹配防止重复模板 + timerSec :None|int = None + res=re.finditer(Pattern._pattern,o_str) + for i in res : + # 对结果和源字符串处理 + o_str = o_str.replace(i.group(),"",1) + num = self.load(i.group()) + if num and timerSec: + timerSec = num + #消除不完整/非法模板 + o_str= o_str.replace("","") + o_str=o_str.replace("","") + return { + "content":o_str, + "second":timerSec + }