-
Notifications
You must be signed in to change notification settings - Fork 33
/
autoreplysticker.py
269 lines (244 loc) · 9.13 KB
/
autoreplysticker.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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
""" Pagermaid Plugins AutoReplySticker """
# ______ _
# | ___ \ | |
# | |_/ /__ _ __ | |_ __ _ ___ ___ _ __ ___
# | __/ _ \ '_ \| __/ _` |/ __/ _ \ '_ \ / _ \
# | | | __/ | | | || (_| | (_| __/ | | | __/
# \_| \___|_| |_|\__\__,_|\___\___|_| |_|\___|
#
from os import mkdir
from os.path import exists
from asyncio import sleep
from random import randint
import yaml
from telethon.errors.rpcerrorlist import StickersetInvalidError
from telethon.tl.custom.message import Message
from telethon.tl.functions.messages import GetAllStickersRequest
from telethon.tl.functions.messages import GetStickerSetRequest
from telethon.tl.types import InputStickerSetID
from pagermaid import log, version
from pagermaid.listener import listener
from pagermaid.utils import alias_command
async def ars_check(message: Message) -> None:
try:
config = yaml.load(open(r"./plugins/autoreplysticker/config.yml"), Loader=yaml.FullLoader)
except FileNotFoundError:
await message.edit("自动回复贴纸的相关设置不存在。\n请使用 `-ars help` 查看设置方法")
return
_sticker_id = config['sticker_id']
_sticker_hash = config['sticker_hash']
_num = config['num']
_time = config['time']
_white = config['whitelist']
_noti = await message.reply(
'您当前的设置为:\n'
f'sticker_id: {_sticker_id}\n'
f'sticker_hash: {_sticker_hash}\n'
f'time: {_time}\n'
f'num: {_num}\n'
f'白名单群组id: {_white}\n'
'\n'
'本消息15秒后自动删除')
await message.delete()
await sleep(15)
await _noti.delete()
async def ars_getall(message: Message) -> None:
sticker_sets = await message.client(GetAllStickersRequest(0))
sticker_pack_list = []
for sticker_set in sticker_sets.sets:
if len(sticker_pack_list) < 10:
text = "我发现了一个Sticker Pack，名为\n" + sticker_set.title + "\n" + "ID为： `" + \
str(sticker_set.id) + "` \n" + "Hash为： `" + str(sticker_set.access_hash) + \
"` \n" + "共有" + str(sticker_set.count) + "张"
sticker_pack_list.extend([text])
else:
sticker_pack_list_old = sticker_pack_list
send_text = '\n\n'.join(sticker_pack_list_old)
await message.client.send_message(message.chat_id, send_text)
sticker_pack_list = []
await sleep(2)
sendtext = '\n\n'.join(sticker_pack_list)
try:
await message.client.send_message(message.chat_id, sendtext)
await message.delete()
except ValueError:
await message.client.send_message(message.chat_id, '您还没有添加贴纸包。')
await message.delete()
async def ars_help(message: Message) -> None:
await message.reply(
'欢迎使用自动回复贴纸\n'
'设置方法为\n'
'先使用 `-ars getall` 获取贴纸包的id和hash\n'
'之后使用 `-ars set` 贴纸包id 贴纸包hash 自动删除时间 第i张贴纸 第j张贴纸 ...\n'
'比如 `-ars set 000 001 10 0 1 2 3` 的意义为\n'
'设置贴纸包id为000, hash为001, 自动回复10秒后删除, 随机从第0, 1, 2, 3张贴纸中选择一张自动回复\n\n'
'如果您想要对某个群或某个人禁用自动回复,请在该群中回复`-ars w` 或使用`-ars w <数字>` 进行设置. 该数字可通过-id命令查询'
'如有使用问题,请前往 [这里](https://t.me/PagerMaid_Modify) 请求帮助')
await message.delete()
async def ars_whitelist(message: Message) -> None:
chat_id = str(message.chat_id)
try:
config = yaml.load(open(r"./plugins/autoreplysticker/config.yml"), Loader=yaml.FullLoader)
except FileNotFoundError:
await message.edit("自动回复贴纸的相关设置不存在。\n请使用 `-ars help` 查看设置方法")
return
try:
_white = config['whitelist']
except:
white_list = ['0']
set_state('whitelist', white_list)
_white = config['whitelist']
if len(message.parameter) == 1:
white_id = chat_id
else:
try:
white_id = str(message.parameter[1])
except:
_noti = await message.edit('您输入的不是一个合理的数字')
await sleep(5)
await _noti.delete()
return
_white.append(white_id)
try:
_white.remove('0')
except:
pass
_white = list(set(_white))
set_state('whitelist', _white)
_noti = await message.edit('OK')
await sleep(5)
await _noti.delete()
def set_state(name: str, state: list) -> None:
file_name = "./plugins/autoreplysticker/config.yml"
if exists(file_name):
with open(file_name) as f:
doc = yaml.safe_load(f)
doc[name] = state
with open(file_name, 'w') as f:
yaml.safe_dump(doc, f, default_flow_style=False)
else:
dc = {name: state}
with open(file_name, 'w', encoding='utf-8') as f:
yaml.dump(dc, f)
def get_name(sender: Message.sender) -> str:
"""
get_name(Message.sender)
"""
username = sender.username
first_name = sender.first_name
last_name = sender.last_name
_id = sender.id
if username == None:
if last_name == None:
name = f'[{first_name}](tg://user?id={_id})'
else:
name = f'[{first_name}{last_name}](tg://user?id={_id})'
else:
name = f'@{username}'
return name
def process_link(chatid: int, msgid: int) -> str:
"""
process_link(chat_id, message_id)
return https://t.me/c/chat_id/message_id
"""
if chatid < 0:
if chatid < -1000000000000:
chatid *= -1
chatid -= 1000000000000
else:
chatid *= -1
link = f'https://t.me/c/{chatid}/{msgid}'
return link
@listener(is_plugin=True, outgoing=True, command=alias_command("ars"))
async def ars(context):
if not exists('./plugins/autoreplysticker'):
mkdir('./plugins/autoreplysticker')
if len(context.parameter) == 0:
await ars_help(context)
return
if context.parameter[0] == 'set':
if len(context.parameter) < 5:
await context.reply('请正确输入 `-ars set` <sticker_id> <sticker_hash> <time> <num>')
await context.delete()
return
try:
set_state('sticker_id', context.parameter[1])
set_state('sticker_hash', context.parameter[2])
set_state('time', context.parameter[3])
num_list = []
for i in range(4, len(context.parameter)):
num_list.append(context.parameter[i])
set_state('num', num_list)
white_list = ['0']
set_state('whitelist', white_list)
except:
await context.reply('设置失败,请手动设置`./plugins/autoreplysticker/config.yml`')
return
_noti = await context.reply('设置成功')
await sleep(10)
await _noti.delete()
await ars_check(context)
elif context.parameter[0] == 'check':
await ars_check(context)
elif context.parameter[0] == 'getall':
await ars_getall(context)
elif context.parameter[0] == 'help':
await ars_help(context)
elif context.parameter[0] == 'w':
await ars_whitelist(context)
@listener(incoming=True, ignore_edited=True)
async def process_message(context):
reply_user_id = 0
link = process_link(context.chat_id, context.id)
me = await context.client.get_me()
try:
reply = await context.get_reply_message()
except ValueError:
return
try:
reply_user_id = reply.sender.id
if context.sticker:
return
if context.chat_id > 0:
return
except:
pass
try:
config = yaml.load(open(r"./plugins/autoreplysticker/config.yml"), Loader=yaml.FullLoader)
_sticker_id = int(config['sticker_id'])
_sticker_hash = int(config['sticker_hash'])
_num = config['num']
_time = int(config['time'])
_whitelist = config['whitelist']
except:
return
try:
if str(context.chat_id) in _whitelist:
return
if str(context.sender.id) in _whitelist:
return
if context.sender.bot:
return
except:
pass
if reply and reply_user_id == me.id:
try:
stickers = await context.client(
GetStickerSetRequest(
stickerset=InputStickerSetID(
id=_sticker_id, access_hash=_sticker_hash)))
except StickersetInvalidError:
await log('配置错误。')
return
try:
i = randint(0, len(_num) - 1)
sticker = await context.client.send_file(
context.chat_id,
stickers.documents[int(_num[i])],
reply_to=context.id)
await sleep(_time)
await sticker.delete()
await log(
f'#被回复\n在 [{context.chat.title}]({process_link(context.chat_id, context.id)})\n获得了 {get_name(context.sender)} 的回复')
except:
pass