generated from LlmKira/llmbot_plugin_bilisearch
-
Notifications
You must be signed in to change notification settings - Fork 0
/
__init__.py
176 lines (152 loc) · 5.58 KB
/
__init__.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
# -*- coding: utf-8 -*-
# @Time : 2023/10/22 下午11:30
# @Author : sudoskys
# @File : __init__.py.py
# @Software: PyCharm
__plugin_name__ = "reply_emoji"
__openapi_version__ = "20231017"
import os
import pathlib
import random
from typing import Optional
import emoji
from llmkira.sdk.func_calling import verify_openapi_version
verify_openapi_version(__plugin_name__, __openapi_version__)
from loguru import logger
from pydantic import validator, BaseModel, Field
from llmkira.schema import RawMessage
from llmkira.sdk.endpoint.openai import Function
from llmkira.sdk.func_calling import PluginMetadata, BaseTool
from llmkira.sdk.func_calling.schema import FuncPair
from llmkira.task import Task, TaskHeader
from llmbot_plugin_sticker.event import StickerEvent
dir_path = os.path.split(os.path.realpath(__file__))[0]
_pack = pathlib.Path(dir_path + '/sticker.zip')
if not _pack.exists():
raise ValueError("sticker.zip not found")
_cache = pathlib.Path.home() / ".cache" / "sticker"
if not _cache.exists():
_cache.mkdir(parents=False, exist_ok=True)
# 解压到 cache
import zipfile
print("Unzip sticker")
with zipfile.ZipFile(_pack, "r") as zip_ref:
zip_ref.extractall(_cache)
sticker_event = StickerEvent(sticker_dir=_cache)
sticker = Function(name=__plugin_name__,
description=f"Respond emotionally(emoji_sticker) to messages")
sticker.add_property(
property_name="select_emoji",
property_description=f"EMOJI ONLY IN {sticker_event.prompt()}",
property_type="string",
required=True
)
class Sticker(BaseModel):
select_emoji: Optional[str] = Field(default=None, description=f"EMOJI ONLY IN {sticker_event.prompt()}")
class Config:
extra = "allow"
@validator("select_emoji")
def delay_validator(cls, v):
if not v:
raise ValueError("没想好要发什么表情呢")
return v
class StickerTool(BaseTool):
"""
搜索工具
"""
silent: bool = True
function: Function = sticker
keywords: list = ["贴纸", "表情", "emoji", "sticker"]
require_auth: bool = False
repeatable = True
def pre_check(self):
return True
def func_message(self, message_text):
"""
如果合格则返回message,否则返回None,表示不处理
"""
for i in self.keywords:
if i in message_text:
return self.function
# 正则匹配
if self.pattern:
match = self.pattern.match(message_text)
if match:
return self.function
# 加入随机因子
if len(message_text) < 50:
if random.randint(0, 100) < 50:
return self.function
return None
async def failed(self, platform, task, receiver, reason):
try:
_meta = task.task_meta.reply_notify(
plugin_name=__plugin_name__,
callback=task.task_meta.callback,
)
await Task(queue=platform).send_task(
task=TaskHeader(
sender=task.sender,
receiver=receiver,
task_meta=_meta,
message=[
RawMessage(
user_id=receiver.user_id,
chat_id=receiver.chat_id,
text=f"刚刚想发贴纸发不出来,因为系统说{reason} ..."
)
]
)
)
except Exception as e:
logger.error(e)
async def callback(self, sign: str, task: TaskHeader):
return None
async def run(self, task: TaskHeader, receiver: TaskHeader.Location, arg, **kwargs):
"""
处理message,返回message
"""
try:
_set = Sticker.parse_obj(arg)
logger.debug("Plugin: {} run with arg: {}", __plugin_name__, arg)
_sticker, _sticker_path = sticker_event.get_sticker(_set.select_emoji)
if not _sticker_path:
raise ValueError(f"找不着表情")
_meta = task.task_meta.reply_message(
plugin_name=__plugin_name__,
callback=TaskHeader.Meta.Callback(
role="function",
name=__plugin_name__
)
)
file = await RawMessage.upload_file(name=f"{emoji.demojize(_sticker)}.webp",
data=_sticker_path.read_bytes())
await Task(queue=receiver.platform).send_task(
task=TaskHeader(
sender=task.sender, # 继承发送者
receiver=receiver, # 因为可能有转发,所以可以单配
task_meta=_meta,
message=[
RawMessage(
user_id=receiver.user_id,
chat_id=receiver.chat_id,
text=f"🍖 Sticker: {_sticker}",
just_file=True,
file=[file]
)
]
)
)
except Exception as e:
logger.exception(e)
await self.failed(platform=receiver.platform, task=task, receiver=receiver, reason=str(e))
__plugin_meta__ = PluginMetadata(
name=__plugin_name__,
description="send sticker when chat....",
usage="just wait sometime....",
openapi_version=__openapi_version__,
function={
FuncPair(function=sticker, tool=StickerTool)
},
homepage="https://github.com/LlmKira"
)