TeamPGM/PagerMaid_Plugins_Pyro
2
0
Fork 0
mirror of https://github.com/TeamPGM/PagerMaid_Plugins_Pyro.git synced 2024-11-22 09:41:54 +00:00
Code Issues Actions Projects Releases Wiki Activity

'Refactored by Sourcery' (#8)

Browse Source 
Co-authored-by: Sourcery AI <>
This commit is contained in:
sourcery-ai[bot] 2022-06-14 15:54:55 +08:00 committed by GitHub
parent 33ca7c85c8
commit 1360fe05e2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 194 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
abstract/main.py
View File

@ -299,8 +299,7 @@ def text_to_emoji(text):
# 分词拼音检索
if word_py in emoji_py.keys():
text_with_emoji += emoji_py[word_py]
else:
if len(word) > 0: # if the two characters or more
elif len(word) > 0: # if the two characters or more
# 单字检索
for character in word:
if character in emoji.keys():
@ -308,10 +307,12 @@ def text_to_emoji(text):
else:
# 单字拼音检索
character_py = pinyin.get(character, format="strip")
if character_py in emoji_py.keys():
text_with_emoji += emoji_py[character_py]
else:
text_with_emoji += character
text_with_emoji += (
emoji_py[character_py]
if character_py in emoji_py.keys()
else character
)
else: # 只有一个汉字，前面已经检测过字和拼音都不在抽象词典中，直接加词
text_with_emoji += word.strip()
except Exception as e:

4
base64/main.py
View File

@ -12,9 +12,7 @@ async def b64e(_: Client, message: Message):
if not msg:
return await message.edit("`出错了呜呜呜 ~ 无效的参数。`")
result = b64encode(msg.encode("utf-8")).decode("utf-8")
if result:
if result := b64encode(msg.encode("utf-8")).decode("utf-8"):
await message.edit(f"`{result}`")

5
bin/main.py
View File

@ -17,7 +17,7 @@ async def card(_: Client, message: Message):
await message.edit("出错了呜呜呜 ~ 无效的参数。")
return
try:
r = requests.get("https://lookup.binlist.net/" + card_bin)
r = requests.get(f"https://lookup.binlist.net/{card_bin}")
except:
await message.edit("出错了呜呜呜 ~ 无法访问到binlist。")
return
@ -34,8 +34,7 @@ async def card(_: Client, message: Message):
await message.edit("出错了呜呜呜 ~ 无效的参数。")
return
msg_out = []
msg_out.extend(["BIN：" + card_bin])
msg_out = [f"BIN：{card_bin}"]
try:
msg_out.extend(["卡品牌：" + bin_json['scheme']])
except (KeyError, TypeError):

2
diss/main.py
View File

@ -5,7 +5,7 @@ from pagermaid.utils import Message, client, edit_delete
@listener(command="diss", description="儒雅随和版祖安语录。")
async def diss(_: Client, message: Message):
for i in range(5): # 最多尝试5次
for _ in range(5):
req = await client.get("https://api.oddfar.com/yl/q.php?c=1009&encode=text")
if req.status_code == 200:
return await message.edit(req.text)

20
eat/main.py
View File

@ -62,12 +62,12 @@ async def eat_it(context, user, base, mask, photo, number, layer=0):
isContinue = len(numberPosition) > 2 and layer == 0
if isContinue:
await context._client.download_media(
user.photo.big_file_id,
f"plugins{sep}eat{sep}" + str(user.id) + ".jpg",
user.photo.big_file_id, f"plugins{sep}eat{sep}{str(user.id)}.jpg"
)
try:
markImg = Image.open(f"plugins{sep}eat{sep}" + str(user.id) + ".jpg")
maskImg = Image.open(f"plugins{sep}eat{sep}mask" + str(numberPosition[2]) + ".png")
markImg = Image.open(f"plugins{sep}eat{sep}{str(user.id)}.jpg")
maskImg = Image.open(f"plugins{sep}eat{sep}mask{str(numberPosition[2])}.png")
except:
await context.edit(f"图片模版加载出错，请检查并更新配置：mask{str(numberPosition[2])}.png")
return base
@ -82,13 +82,11 @@ async def eat_it(context, user, base, mask, photo, number, layer=0):
async def updateConfig(context):
configFileRemoteUrl = sqlite.get(configFileRemoteUrlKey, "")
if configFileRemoteUrl:
if downloadFileFromUrl(configFileRemoteUrl, configFilePath) != 0:
if configFileRemoteUrl := sqlite.get(configFileRemoteUrlKey, ""):
if downloadFileFromUrl(configFileRemoteUrl, configFilePath) == 0:
return await loadConfigFile(context, True)
sqlite[configFileRemoteUrlKey] = configFileRemoteUrl
return -1
else:
return await loadConfigFile(context, True)
return 0
@ -183,9 +181,9 @@ async def downloadFileByIds(ids, context):
failSet.add(fileName)
await context.edit(f"下载文件异常，url：{file_url}")
notifyStr = "更新模版完成"
if len(sucSet) > 0:
if sucSet:
notifyStr = f'{notifyStr}\n成功模版如下：{"，".join(sucSet)}'
if len(failSet) > 0:
if failSet:
notifyStr = f'{notifyStr}\n失败模版如下：{"，".join(failSet)}'
await context.edit(notifyStr)
except:

6
everyday_en/main.py
View File

@ -33,13 +33,11 @@ async def get_everyday_en() -> None:
everyday_en_cache_time = date.today()
safe_remove(f"data{sep}everyday_en.jpg")
safe_remove(f"data{sep}everyday_en.mp3")
url = everyday_en_data_cache.get("fenxiang_img", "")
if url:
if url := everyday_en_data_cache.get("fenxiang_img", ""):
resp = await client.get(url, follow_redirects=True)
with open(f"data{sep}everyday_en.jpg", "wb") as f:
f.write(resp.content)
url = everyday_en_data_cache.get("tts", "")
if url:
if url := everyday_en_data_cache.get("tts", ""):
resp = await client.get(url, follow_redirects=True)
with open(f"data{sep}everyday_en.mp3", "wb") as f:
f.write(resp.content)

57
pmcaptcha/main.py
View File

@ -106,7 +106,10 @@ async def process_pm_captcha(client: Client, message: Message):
await message.reply('对方已设置禁止私聊，您已被封禁\n\nYou are not allowed to send private messages to me and been banned')
await do_action_and_read(client, cid, data)
return
if not captcha_success.check_id(cid) and sqlite.get("pmcaptcha." + str(cid)) is None:
if (
not captcha_success.check_id(cid)
and sqlite.get(f"pmcaptcha.{str(cid)}") is None
):
await client.read_chat_history(cid)
if data.get("blacklist", False) and message.text is not None:
for i in data.get("blacklist", "").split(","):
@ -124,7 +127,7 @@ async def process_pm_captcha(client: Client, message: Message):
key1 = random.randint(1, 10)
key2 = random.randint(1, 10)
await asyncio.sleep(random.randint(0, 100) / 1000)
sqlite['pmcaptcha.' + str(cid)] = str(key1 + key2)
sqlite[f'pmcaptcha.{str(cid)}'] = str(key1 + key2)
msg = await message.reply(
'已启用私聊验证。请发送 \"' + str(key1) + '+' + str(key2) + '\" 的答案(阿拉伯数字)来与我私聊\n请在' + str(wait) +
'秒内完成验证。您只有一次验证机会\n\nHuman verification is enabled.Please send the answer of this question \"' +
@ -132,14 +135,14 @@ async def process_pm_captcha(client: Client, message: Message):
' seconds to complete the verification.')
await asyncio.sleep(wait)
await msg.safe_delete() # noqa
if sqlite.get('pmcaptcha.' + str(cid)) is not None:
del sqlite['pmcaptcha.' + str(cid)]
if sqlite.get(f'pmcaptcha.{str(cid)}') is not None:
del sqlite[f'pmcaptcha.{str(cid)}']
await message.reply('验证超时,您已被封禁\n\nVerification timeout.You have been banned.')
await do_action_and_read(client, cid, data)
elif sqlite.get("pmcaptcha." + str(cid)):
if message.text == sqlite.get("pmcaptcha." + str(cid)):
elif sqlite.get(f"pmcaptcha.{str(cid)}"):
if message.text == sqlite.get(f"pmcaptcha.{str(cid)}"):
await message.safe_delete()
del sqlite['pmcaptcha.' + str(cid)]
del sqlite[f'pmcaptcha.{str(cid)}']
captcha_success.add_id(cid)
try:
await client.invoke(UpdateNotifySettings(peer=InputNotifyPeer(peer=await client.resolve_peer(cid)),
@ -153,7 +156,7 @@ async def process_pm_captcha(client: Client, message: Message):
data['pass'] = data.get('pass', 0) + 1
sqlite['pmcaptcha'] = data
else:
del sqlite['pmcaptcha.' + str(cid)]
del sqlite[f'pmcaptcha.{str(cid)}']
await message.reply('验证错误，您已被封禁\n\nVerification failed.You have been banned.')
await do_action_and_read(client, cid, data)
@ -169,10 +172,7 @@ async def pm_captcha(client: Client, message: Message):
await message.edit('请在私聊时使用此命令，或添加参数执行')
await asyncio.sleep(3)
await message.safe_delete()
if captcha_success.check_id(message.chat.id):
text = "已验证用户"
else:
text = "未验证/验证中用户"
text = "已验证用户" if captcha_success.check_id(message.chat.id) else "未验证/验证中用户"
await message.edit(text)
elif len(message.parameter) == 1:
if message.parameter[0] == "bl":
@ -191,13 +191,17 @@ async def pm_captcha(client: Client, message: Message):
return await message.safe_delete()
await message.edit(pm_captcha_help_msg)
elif message.parameter[0] == 'disablepm':
if data.get('disable', False):
status = '开启'
else:
status = '关闭'
await message.edit('当前禁止私聊状态: 已' + status +
'\n如需修改 请使用 ,pmcaptcha disablepm true/false' +
'\n此功能会放行联系人和白名单(已通过验证)用户')
status = '开启' if data.get('disable', False) else '关闭'
await message.edit(
(
(
f'当前禁止私聊状态: 已{status}'
+ '\n如需修改 请使用 ,pmcaptcha disablepm true/false'
)
+ '\n此功能会放行联系人和白名单(已通过验证)用户'
)
)
elif message.parameter[0] == 'stats':
await message.edit('自上次重置起，已进行验证 ' + str(data.get('pass', 0) + data.get('banned', 0)) +
' 次\n其中，通过验证 ' + str(data.get('pass', 0)) + ' 次，拦截 ' + str(data.get('banned', 0)) + ' 次')
@ -206,19 +210,18 @@ async def pm_captcha(client: Client, message: Message):
await asyncio.sleep(3)
await message.safe_delete()
elif message.parameter[0] == 'add':
await message.edit('已将id ' + cid_ + ' 添加至白名单')
await message.edit(f'已将id {cid_} 添加至白名单')
captcha_success.add_id(message.chat.id)
elif message.parameter[0] == 'del':
if captcha_success.del_id(message.chat.id):
await message.edit('已删除id ' + cid_ + ' 的验证记录')
await message.edit(f'已删除id {cid_} 的验证记录')
else:
await message.edit('记录不存在')
else:
await message.edit('参数错误')
else:
if message.parameter[0] == 'add':
elif message.parameter[0] == 'add':
if message.parameter[1].isnumeric():
await message.edit('已将id ' + message.parameter[1] + ' 添加至白名单')
await message.edit(f'已将id {message.parameter[1]} 添加至白名单')
captcha_success.add_id(int(message.parameter[1]))
await client.unarchive_chats(chat_ids=int(message.parameter[1]))
else:
@ -226,7 +229,7 @@ async def pm_captcha(client: Client, message: Message):
elif message.parameter[0] == 'del':
if message.parameter[1].isnumeric():
if captcha_success.del_id(int(message.parameter[1])):
await message.edit('已删除id ' + message.parameter[1] + ' 的验证记录')
await message.edit(f'已删除id {message.parameter[1]} 的验证记录')
else:
await message.edit('记录不存在')
else:
@ -261,9 +264,9 @@ async def pm_captcha(client: Client, message: Message):
elif message.parameter[0] == 'check':
if message.parameter[1].isnumeric():
if captcha_success.check_id(int(message.parameter[1])):
await message.edit('id ' + message.parameter[1] + ' 已验证')
await message.edit(f'id {message.parameter[1]} 已验证')
else:
await message.edit('id ' + message.parameter[1] + ' 未验证')
await message.edit(f'id {message.parameter[1]} 未验证')
else:
await message.edit('未知用户/无效id')
elif message.parameter[0] == 'disablepm':

6
resou/main.py
View File

@ -15,7 +15,7 @@ async def zhrs(_: Client, message: Message):
await message.edit("出错了呜呜呜 ~ API 数据解析失败。")
return
res = '知乎实时热搜榜：\n'
for i in range(0, 10):
for i in range(10):
res += f'\n{i + 1}.「<a href={data["list"][i]["url"]}>{data["list"][i]["query"]}</a>」'
await message.edit(res)
else:
@ -33,7 +33,7 @@ async def wbrs(_: Client, message: Message):
await message.edit("出错了呜呜呜 ~ API 数据解析失败。")
return
res = '微博实时热搜榜：\n'
for i in range(0, 10):
for i in range(10):
res += f'\n{i + 1}.「<a href={data["list"][i]["url"]}>{data["list"][i]["name"]}</a>」 ' \
f'热度：{data["list"][i]["hot"]}'
await message.edit(res)
@ -52,7 +52,7 @@ async def dyrs(_: Client, message: Message):
await message.edit("出错了呜呜呜 ~ API 数据解析失败。")
return
res = '抖音实时热搜榜：\n'
for i in range(0, 10):
for i in range(10):
res += f'\n{i + 1}.「{data["list"][i]["name"]}」 热度：{data["list"][i]["hot"]}'
await message.edit(res)
else:

24
tel/main.py
View File

@ -17,14 +17,26 @@ async def tel(_: Client, context: Message):
except ValueError:
await context.edit("出错了呜呜呜 ~ 无效的参数。")
return
req = get("https://tenapi.cn/tel?tel=" + message)
req = get(f"https://tenapi.cn/tel?tel={message}")
if req.status_code == 200:
data = json.loads(req.text)
if not 'msg' in data:
res = '电话号码：' + str(data['tel']) + '\n' + str(data['local']) + '\n' + str(data['duan']) + '\n' + str(
data['type']) + '\n' + str(data['yys']) + '\n' + str(data['bz'])
else:
res = data['msg']
res = (
data['msg']
if 'msg' in data
else '电话号码：'
+ str(data['tel'])
+ '\n'
+ str(data['local'])
+ '\n'
+ str(data['duan'])
+ '\n'
+ str(data['type'])
+ '\n'
+ str(data['yys'])
+ '\n'
+ str(data['bz'])
)
await context.edit(res)
else:
await context.edit("出错了呜呜呜 ~ 无法访问到 API 服务器 。")

21
telegram_update.py
View File

@ -3,12 +3,25 @@ from httpx import get, post
token = str(sys.argv[1])
main = json.loads(get("https://api.github.com/repos/TeamPGM/PagerMaid_Plugins_Pyro/commits/v2").content)
text = "#更新日志 #pyro #" + main['commit']['author']['name'].replace('_', '') + \
' \n\n🔨 [' + main['sha'][0:7] + '](https://github.com/TeamPGM/PagerMaid_Plugins_Pyro/commit/' + \
main['sha'] + '): ' + main['commit']['message']
text = (
(
(
(
"#更新日志 #pyro #"
+ main['commit']['author']['name'].replace('_', '')
+ ' \n\n🔨 ['
+ main['sha'][:7]
)
+ '](https://github.com/TeamPGM/PagerMaid_Plugins_Pyro/commit/'
)
+ main['sha']
)
+ '): '
) + main['commit']['message']
push_content = {'chat_id': '-1001441461877', 'disable_web_page_preview': 'True', 'parse_mode': 'markdown',
'text': text}
url = 'https://api.telegram.org/bot' + token + '/sendMessage'
url = f'https://api.telegram.org/bot{token}/sendMessage'
try:
main_req = post(url, data=push_content)
except:

2
weather/main.py
View File

@ -48,7 +48,7 @@ async def weather(_: Client, message: Message):
"=zh_cn&q=" + message.arguments)
if req.status_code == 200:
data = req.json()
cityName = "{}, {}".format(data["name"], data["sys"]["country"])
cityName = f'{data["name"]}, {data["sys"]["country"]}'
timeZoneShift = data["timezone"]
temp_Max = round(data["main"]["temp_max"], 2)
temp_Min = round(data["main"]["temp_min"], 2)

5
whois/main.py
View File

@ -11,7 +11,10 @@ async def whois(_: Client, context: Message):
except ValueError:
await context.edit("出错了呜呜呜 ~ 无效的参数。")
return
req = await client.get("https://namebeta.com/api/search/check?query=" + message)
req = await client.get(
f"https://namebeta.com/api/search/check?query={message}"
)
if req.status_code == 200:
try:
data = req.json()["whois"]["whois"].split("For more information")[0].rstrip()

24
xtao-some/main.py
View File

@ -15,9 +15,9 @@ async def guess(_: Client, message: Message):
text = {'text': text}
guess_json = (await client.post("https://lab.magiconch.com/api/nbnhhsh/guess", json=text)).json()
if len(guess_json) != 0:
guess_res = []
if not len(guess_json) == 0:
for num in range(0, len(guess_json)):
for num in range(len(guess_json)):
guess_res1 = json.loads(json.dumps(guess_json[num]))
guess_res1_name = guess_res1['name']
try:
@ -27,7 +27,7 @@ async def guess(_: Client, message: Message):
guess_res1_ans = ", ".join(guess_res1['inputting'])
except:
guess_res1_ans = "尚未录入"
guess_res.extend(["词组：" + guess_res1_name + "\n释义：" + guess_res1_ans])
guess_res.extend([f"词组：{guess_res1_name}" + "\n释义：" + guess_res1_ans])
await message.edit("\n\n".join(guess_res))
else:
await message.edit("没有匹配到拼音首字母缩写")
@ -42,19 +42,29 @@ async def wiki(_: Client, message: Message):
return await message.edit("请先输入一个关键词。")
message = await message.edit("获取中 . . .")
try:
req = await client.get("https://zh.wikipedia.org/w/api.php?action=query&list=search&format=json&formatversion=2&srsearch=" + text)
req = await client.get(
f"https://zh.wikipedia.org/w/api.php?action=query&list=search&format=json&formatversion=2&srsearch={text}"
)
wiki_json = json.loads(req.content.decode("utf-8"))
except:
return await message.edit("出错了呜呜呜 ~ 无法访问到维基百科。")
try:
if not len(wiki_json['query']['search']) == 0:
if len(wiki_json['query']['search']) != 0:
wiki_title = wiki_json['query']['search'][0]['title']
wiki_content = wiki_json['query']['search'][0]['snippet'].replace('<span class=\"searchmatch\">',
'**').replace(
'</span>', '**')
wiki_time = wiki_json['query']['search'][0]['timestamp'].replace('T', ' ').replace('Z', ' ')
text = '词条： [' + wiki_title + '](https://zh.wikipedia.org/zh-cn/' + wiki_title + ')\n\n' + \
wiki_content + '...\n\n此词条最后修订于 ' + wiki_time
text = (
(
f'词条： [{wiki_title}](https://zh.wikipedia.org/zh-cn/{wiki_title}'
+ ')\n\n'
+ wiki_content
)
+ '...\n\n此词条最后修订于 '
) + wiki_time
else:
text = "没有匹配到相关词条"
except KeyError: