Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-09-28 16:20:56 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'HibiKier:main' into main

Browse Source
This commit is contained in:
AkashiCoin
2022-05-25 01:01:11 +08:00
committed by GitHub
parent 8d3f101a68 07407749fe
commit 310945d9cd
42 changed files with 395 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+29 -1
View File
@@ -238,6 +238,34 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
## 更新
### 2022/5/24
* fix: 修正了签到时日期时间的显示不补齐零的bug （符合日期时间表示法 ISO 8601）[@pull/600](https://github.com/HibiKier/zhenxun_bot/pull/600)
* 更新"微博热搜"接口 [@pull/579](https://github.com/HibiKier/zhenxun_bot/pull/579)
* refactor&fix(manager): modify argument [@pull/576](https://github.com/HibiKier/zhenxun_bot/pull/576)
* 修复复读不能复读图片的问题 [@pull/573](https://github.com/HibiKier/zhenxun_bot/pull/573)
* 修复抽卡插件：方舟抽卡的抽取和显示问题 [@pull/581](https://github.com/HibiKier/zhenxun_bot/pull/581)
### 2022/5/23 \[v0.1.5.6]
* 修复"清除已删除插件数据" [@pull/545](https://github.com/HibiKier/zhenxun_bot/pull/545)
* 修复有置顶的up主B站动态获取失败 [@pull/552](https://github.com/HibiKier/zhenxun_bot/pull/552)
* 添加pixiv搜图多关键词支持;修复p站搜图数量参数问题 [@pull/441](https://github.com/HibiKier/zhenxun_bot/pull/441)
* 修复开箱更新价格错误传参
* 修复pix无法正确查询uid
* 新增色图插件添加配置项ALLOW_GROUP_R18：允许群聊中使用色图r
* 新增PIX插件添加配置项ALLOW_GROUP_SETU：允许非超级用户使用-s参数
* 新增PIX插件添加配置项ALLOW_GROUP_R18：允许非超级用户使用-r参数
### 2022/5/22 \[v0.1.5.4]
* 使用action自动更新poetry.lock [@pull/515](https://github.com/HibiKier/zhenxun_bot/pull/515)
* fix(bilibili_sub): card is None and timeout [@pull/516](https://github.com/HibiKier/zhenxun_bot/pull/516)
* 修复了epic有时获取新免费游戏消息时获取不到图片
* 修复好感度满时签到出错（虽然是不可能满的
* 修复原神资源图标下载路径错误
* 修复自动更新群组可能失败
### 2022/5/21
* 修复搜番无结果时报错无正确反馈
@@ -279,7 +307,7 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
* fix: 更改p搜api，解决p搜无法使用的问题 [@pull/155](https://github.com/HibiKier/zhenxun_bot/pull/155)
### 2022/5/9
### 2022/5/9 \[v0.1.5.3]
* 替换了疫情API
* 修复了私聊.ban/.unban出错
__version__
+1 -1
View File
@@ -1 +1 @@
__version__: v0.1.5.3
__version__: v0.1.5.6
basic_plugins/chat_history/chat_message.py
+3 -2
View File
@@ -2,6 +2,7 @@ from configs.config import Config
from models.chat_history import ChatHistory
from nonebot import on_message
from nonebot.adapters.onebot.v11 import GroupMessageEvent, MessageEvent
from utils.utils import get_message_text
from ._rule import rule
@@ -22,10 +23,10 @@ chat_history = on_message(rule=rule, priority=1, block=False)
async def _(event: MessageEvent):
if isinstance(event, GroupMessageEvent):
await ChatHistory.add_chat_msg(
event.user_id, event.group_id, str(event.get_message())
event.user_id, event.group_id, str(event.get_message()), get_message_text(event.json())
)
else:
await ChatHistory.add_chat_msg(event.user_id, None, str(event.get_message()))
await ChatHistory.add_chat_msg(event.user_id, None, str(event.get_message()), get_message_text(event.json()))
# @test.handle()
basic_plugins/init_plugin_config/init_none_plugin_count_manager.py
+29 -19
View File
@@ -20,26 +20,36 @@ def init_none_plugin_count_manager():
"""
清除已删除插件数据
"""
modules = [x.module for x in get_matchers()]
for module in none_plugin_count_manager.keys():
if module not in modules:
none_plugin_count_manager.add_count(module)
else:
none_plugin_count_manager.reset(module)
if none_plugin_count_manager.check(module):
modules = [x.plugin_name for x in get_matchers()]
plugins_manager_list = list(plugins_manager.keys())
for module in plugins_manager_list:
if module not in modules or none_plugin_count_manager.check(module):
try:
plugin_name = plugins_manager.get(module)["plugin_name"]
except (AttributeError, KeyError):
plugin_name = ""
try:
plugins2settings_manager.delete(module)
plugins2count_manager.delete(module)
plugins2cd_manager.delete(module)
plugins2block_manager.delete(module)
plugins_manager.delete(module)
resources_manager.remove_resource(module)
logger.info(f"{module}:{plugin_name} 插件疑似已删除，清除对应插件数据...")
except Exception as e:
logger.error(
f"{module}:{plugin_name} 插件疑似已删除，清除对应插件数据失败...{type(e)}：{e}"
)
if none_plugin_count_manager.check(module):
try:
plugins2settings_manager.delete(module)
plugins2settings_manager.save()
plugins2count_manager.delete(module)
plugins2count_manager.save()
plugins2cd_manager.delete(module)
plugins2cd_manager.save()
plugins2block_manager.delete(module)
plugins2block_manager.save()
plugins_manager.delete(module)
plugins_manager.save()
resources_manager.remove_resource(module)
none_plugin_count_manager.delete(module)
logger.info(f"{module}:{plugin_name} 插件疑似已删除，清除对应插件数据...")
except Exception as e:
logger.exception(
f"{module}:{plugin_name} 插件疑似已删除，清除对应插件数据失败...{type(e)}：{e}")
else:
none_plugin_count_manager.add_count(module)
logger.info(
f"{module}:{plugin_name} 插件疑似已删除，加载{none_plugin_count_manager._max_count}次失败后将清除对应插件数据，当前次数：{none_plugin_count_manager._data[module]}")
else:
none_plugin_count_manager.reset(module)
none_plugin_count_manager.save()
basic_plugins/init_plugin_config/init_plugins_limit.py
+3 -3
View File
@@ -28,7 +28,7 @@ def init_plugins_cd_limit(data_path):
_module = _plugin.module
plugin_cd_limit = _module.__getattribute__("__plugin_cd_limit__")
plugins2cd_manager.add_cd_limit(
matcher.plugin_name, data_dict=plugin_cd_limit
matcher.plugin_name, **plugin_cd_limit
)
except AttributeError:
pass
@@ -76,7 +76,7 @@ def init_plugins_block_limit(data_path):
_module = _plugin.module
plugin_block_limit = _module.__getattribute__("__plugin_block_limit__")
plugins2block_manager.add_block_limit(
matcher.plugin_name, data_dict=plugin_block_limit
matcher.plugin_name, **plugin_block_limit
)
except AttributeError:
pass
@@ -124,7 +124,7 @@ def init_plugins_count_limit(data_path):
_module = _plugin.module
plugin_count_limit = _module.__getattribute__("__plugin_count_limit__")
plugins2count_manager.add_count_limit(
matcher.plugin_name, data_dict=plugin_count_limit
matcher.plugin_name, **plugin_count_limit
)
except AttributeError:
pass
basic_plugins/scripts.py
+4
View File
@@ -108,6 +108,10 @@ async def _():
"ALTER TABLE genshin ADD bind_group Integer;",
"genshin"
), # 新增原神群号绑定字段
(
"ALTER TABLE chat_history ADD plain_text Text;",
"chat_history"
), # 新增纯文本
]
for sql in sql_str:
try:
basic_plugins/super_cmd/manager_group.py
+1 -1
View File
@@ -92,7 +92,7 @@ async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
level = 0
if not msg:
await add_group_level.finish("用法：修改群权限 [group] [level]")
msg = msg.split(" ")
msg = msg.split()
if len(msg) < 2:
await add_group_level.finish("参数不完全..[group] [level]")
if is_number(msg[0]) and is_number(msg[1]):
basic_plugins/super_cmd/update_friend_group_info.py
+1
View File
@@ -45,6 +45,7 @@ async def _():
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
1
):
num += 1
logger.info(f"自动更新群组 {g} 信息成功")
models/chat_history.py
+3 -3
View File
@@ -11,12 +11,13 @@ class ChatHistory(db.Model):
user_qq = db.Column(db.BigInteger(), nullable=False)
group_id = db.Column(db.BigInteger())
text = db.Column(db.Text())
plain_text = db.Column(db.Text())
create_time = db.Column(db.DateTime(timezone=True), nullable=False)
@classmethod
async def add_chat_msg(cls, user_qq: int, group_id: Optional[int], text: str):
async def add_chat_msg(cls, user_qq: int, group_id: Optional[int], text: str, plain_text: str):
await cls.create(
user_qq=user_qq, group_id=group_id, text=text, create_time=datetime.now()
user_qq=user_qq, group_id=group_id, text=text, plain_text=plain_text, create_time=datetime.now()
)
@classmethod
@@ -97,7 +98,6 @@ class ChatHistory(db.Model):
if date_scope:
sql += f"AND create_time BETWEEN '{date_scope[0]}' AND '{date_scope[1]}' "
sql += f"GROUP BY user_qq ORDER BY sum {order if order and order.upper() != 'DES' else ''} LIMIT {limit}"
print(sql)
return await db.all(db.text(sql))
@classmethod
plugins/ai/__init__.py
+1 -1
View File
@@ -37,7 +37,7 @@ Config.add_plugin_config(
"alapi", "ALAPI_TOKEN", None, help_="在 https://admin.alapi.cn/user/login 登录后获取token"
)
ai = on_message(rule=to_me(), priority=8)
ai = on_message(rule=to_me(), priority=998)
@ai.handle()
plugins/alapi/_data_source.py
-25
View File
@@ -1,7 +1,3 @@
from nonebot.adapters.onebot.v11 import MessageSegment
from utils.image_utils import BuildImage
from utils.message_builder import image
from configs.path_config import IMAGE_PATH
from typing import Optional, Tuple, Union
from configs.config import Config
from utils.http_utils import AsyncHttpx
@@ -28,24 +24,3 @@ async def get_data(url: str, params: Optional[dict] = None) -> Tuple[Union[dict,
return f'发生了错误...code：{data["code"]}', 999
except TimeoutError:
return "超时了....", 998
def gen_wbtop_pic(data: dict) -> MessageSegment:
"""
生成微博热搜图片
:param data: 微博热搜数据
"""
bk = BuildImage(700, 32 * 50 + 280, 700, 32, color="#797979")
wbtop_bk = BuildImage(700, 280, background=f"{IMAGE_PATH}/other/webtop.png")
bk.paste(wbtop_bk)
text_bk = BuildImage(700, 32 * 50, 700, 32, color="#797979")
for i, data in enumerate(data):
title = f"{i+1}. {data['hot_word']}"
hot = data["hot_word_num"]
img = BuildImage(700, 30, font_size=20)
w, h = img.getsize(title)
img.text((10, int((30 - h) / 2)), title)
img.text((580, int((30 - h) / 2)), hot)
text_bk.paste(img)
bk.paste(text_bk, (0, 280))
return image(b64=bk.pic2bs4())
plugins/bilibili_sub/data_source.py
+4 -4
View File
@@ -324,21 +324,21 @@ async def get_user_dynamic(
wait_until="networkidle",
timeout=10000,
)
await page.set_viewport_size({"width": 2560, "height": 1080})
await page.set_viewport_size({"width": 2560, "height": 1080, "timeout": 10000*20}) # timeout: 200s
# 删除置顶
await page.evaluate(
"""
xs = document.getElementsByClassName('bili-dyn-item__tag');
for (x of xs) {
x.parentNode.remove();
x.parentNode.parentNode.remove();
}
"""
)
card = await page.query_selector(".bili-dyn-list__item")
card = page.locator(".bili-dyn-list__item").first
await card.wait_for()
# 截图并保存
await card.screenshot(
path=dynamic_path / f"{local_user.sub_id}_{dynamic_upload_time}.jpg",
timeout=100000,
)
except Exception as e:
logger.error(f"B站订阅：获取用户动态 发送错误 {type(e)}：{e}")
plugins/bilibili_sub/utils.py
-1
View File
@@ -38,7 +38,6 @@ async def create_live_des_image(uid: int, title: str, cover: str, tags: str, des
ava = BuildImage(100, 100, background=BytesIO(await get_pic(face)))
ava.circle()
cover = BuildImage(470, 265, background=BytesIO(await get_pic(cover)))
print()
def _create_live_des_image(
plugins/black_word/__init__.py
-1
View File
@@ -241,7 +241,6 @@ async def _(event: MessageEvent, arg: Message = CommandArg()):
uid = int(msg[0])
id_ = int(msg[1])
punish_level = int(msg[2])
print(uid, id_, punish_level)
rst = await set_user_punish(uid, id_, punish_level)
await set_punish.send(rst)
logger.info(
plugins/black_word/model.py
-1
View File
@@ -68,7 +68,6 @@ class BlackWord(db.Model):
user = await query.where(cls.black_word == black_word).order_by(cls.id.desc()).gino.first()
elif id_:
user_list = await query.gino.all()
print(len(user_list))
if len(user_list) == 0 or (id_ < 0 or id_ > len(user_list)):
return False
user = user_list[id_]
plugins/black_word/utils.py
-2
View File
@@ -88,9 +88,7 @@ class BlackWordManager:
:param group_id: 群号
:param message: 消息
"""
print(user_id, group_id, message)
if data := self._check(message):
print(data)
if data[0]:
await _add_user_black_word(
user_id, group_id, data[0], message, int(data[1])
plugins/bt/data_source.py
+4 -4
View File
@@ -3,10 +3,10 @@ from configs.config import Config
from bs4 import BeautifulSoup
import platform
if platform.system() == "Windows":
import asyncio
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# if platform.system() == "Windows":
# import asyncio
#
# asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
url = "http://www.eclzz.mobi"
plugins/check_zhenxun_update/data_source.py
+3 -3
View File
@@ -14,9 +14,9 @@ import tarfile
import shutil
import os
if str(platform.system()).lower() == "windows":
policy = asyncio.WindowsSelectorEventLoopPolicy()
asyncio.set_event_loop_policy(policy)
# if str(platform.system()).lower() == "windows":
# policy = asyncio.WindowsSelectorEventLoopPolicy()
# asyncio.set_event_loop_policy(policy)
driver = nonebot.get_driver()
plugins/draw_card/handles/prts_handle.py
+12 -4
View File
@@ -53,7 +53,7 @@ class PrtsHandle(BaseHandle[Operator]):
all_operators = [
x
for x in self.ALL_OPERATOR
if x.star == star and not any([x.limited, x.event_only, x.recruit_only])
if x.star == star and not any([x.limited, x.recruit_only, x.event_only])
]
acquire_operator = None
@@ -106,10 +106,13 @@ class PrtsHandle(BaseHandle[Operator]):
async def draw(self, count: int, **kwargs) -> Message:
return await asyncio.get_event_loop().run_in_executor(None, self._draw, count)
def _draw(self, count: int, **kwargs) -> Message:
index2card = self.get_cards(count)
cards = [card[0] for card in self.get_cards(count)]
"""这里cards修复了抽卡图文不符的bug"""
cards = [card[0] for card in index2card]
up_list = [x.name for x in self.UP_EVENT.up_char] if self.UP_EVENT else []
result = self.format_result(index2card, up_list=up_list)
pool_info = self.format_pool_info()
@@ -185,14 +188,15 @@ class PrtsHandle(BaseHandle[Operator]):
avatar = char.xpath("./td[1]/div/div/div/a/img/@srcset")[0]
name = char.xpath("./td[2]/a/text()")[0]
star = char.xpath("./td[5]/text()")[0]
sources = str(char.xpath("./td[8]/text()")[0]).split("\n")
"""这里sources修好了干员获取标签有问题的bug，如三星只能抽到卡缇就是这个原因"""
sources = [_.strip('\n') for _ in char.xpath("./td[8]/text()")]
except IndexError:
continue
member_dict = {
"头像": unquote(str(avatar).split(" ")[-2]),
"名称": remove_prohibited_str(str(name).strip()),
"星级": int(str(star).strip()),
"获取途径": [s for s in sources if s],
"获取途径": sources,
}
info[member_dict["名称"]] = member_dict
self.dump_data(info)
@@ -245,6 +249,10 @@ class PrtsHandle(BaseHandle[Operator]):
if match:
time = match.group(1)
if "★" in line:
"""这里修复了某些池子六星名称显示错误的问题(如奔崖号角)"""
if line[0] != '★':
idx = line.find('★')
line = line[idx:]
chars.append(line)
if not time:
continue
plugins/epic/data_source.py
+24 -37
View File
@@ -1,58 +1,34 @@
from httpx import AsyncClient
from datetime import datetime
from nonebot.log import logger
from nonebot.adapters.onebot.v11 import Bot
from configs.config import NICKNAME
from utils.http_utils import AsyncHttpx
# 获取所有 Epic Game Store 促销游戏
# 方法参考：RSSHub /epicgames 路由
# https://github.com/DIYgod/RSSHub/blob/master/lib/routes/epicgames/index.js
async def get_epic_game():
# 现在没用 graphql 辣
"""prv_graphql Code
epic_url = "https://www.epicgames.com/store/backend/graphql-proxy"
headers = {
"Referer": "https://www.epicgames.com/store/zh-CN/",
"Content-Type": "application/json; charset=utf-8",
}
data = {
"query": "query searchStoreQuery($allowCountries: String, $category: String, $count: Int, $country: String!, $keywords: String, $locale: String, $namespace: String, $sortBy: String, $sortDir: String, $start: Int, $tag: String, $withPrice: Boolean = false, $withPromotions: Boolean = false) {\n Catalog {\n searchStore(allowCountries: $allowCountries, category: $category, count: $count, country: $country, keywords: $keywords, locale: $locale, namespace: $namespace, sortBy: $sortBy, sortDir: $sortDir, start: $start, tag: $tag) {\n elements {\n title\n id\n namespace\n description\n effectiveDate\n keyImages {\n type\n url\n }\n seller {\n id\n name\n }\n productSlug\n urlSlug\n url\n items {\n id\n namespace\n }\n customAttributes {\n key\n value\n }\n categories {\n path\n }\n price(country: $country) @include(if: $withPrice) {\n totalPrice {\n discountPrice\n originalPrice\n voucherDiscount\n discount\n currencyCode\n currencyInfo {\n decimals\n }\n fmtPrice(locale: $locale) {\n originalPrice\n discountPrice\n intermediatePrice\n }\n }\n lineOffers {\n appliedRules {\n id\n endDate\n discountSetting {\n discountType\n }\n }\n }\n }\n promotions(category: $category) @include(if: $withPromotions) {\n promotionalOffers {\n promotionalOffers {\n startDate\n endDate\n discountSetting {\n discountType\n discountPercentage\n }\n }\n }\n upcomingPromotionalOffers {\n promotionalOffers {\n startDate\n endDate\n discountSetting {\n discountType\n discountPercentage\n }\n }\n }\n }\n }\n paging {\n count\n total\n }\n }\n }\n}\n",
"variables": {
"allowCountries": "CN",
"category": "freegames",
"count": 1000,
"country": "CN",
"locale": "zh-CN",
"sortBy": "effectiveDate",
"sortDir": "asc",
"withPrice": True,
"withPromotions": True,
},
}
"""
epic_url = "https://store-site-backend-static-ipv4.ak.epicgames.com/freeGamesPromotions?locale=zh-CN&country=CN&allowCountries=CN"
headers = {
"Referer": "https://www.epicgames.com/store/zh-CN/",
"Content-Type": "application/json; charset=utf-8",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.80 Safari/537.36",
}
async with AsyncClient(headers=headers) as client:
try:
res = await client.get(epic_url, timeout=10.0)
res_json = res.json()
games = res_json["data"]["Catalog"]["searchStore"]["elements"]
return games
except Exception as e:
logger.error(str(e))
return None
try:
res = await AsyncHttpx.get(epic_url, headers=headers, timeout=10)
res_json = res.json()
games = res_json["data"]["Catalog"]["searchStore"]["elements"]
return games
except Exception as e:
logger.error(f"Epic 访问接口错误 {type(e)}：{e}")
return None
# 获取 Epic Game Store 免费游戏信息
# 处理免费游戏的信息方法借鉴 pip 包 epicstore_api 示例
# https://github.com/SD4RK/epicstore_api/blob/master/examples/free_games_example.py
async def get_epic_free(bot: Bot, Type_Event: str):
async def get_epic_free(bot: Bot, type_event: str):
games = await get_epic_game()
if not games:
return "Epic 可能又抽风啦，请稍后再试（", 404
@@ -81,7 +57,7 @@ async def get_epic_free(bot: Bot, Type_Event: str):
end_date = datetime.fromisoformat(end_date_iso).strftime(
"%b.%d %H:%M"
)
if Type_Event == "Group":
if type_event == "Group":
_message = "\n由 {} 公司发行的游戏 {} ({}) 在 UTC 时间 {} 即将推出免费游玩，预计截至 {}。".format(
game_corp, game_name, game_price, start_date, end_date
)
@@ -101,8 +77,19 @@ async def get_epic_free(bot: Bot, Type_Event: str):
msg_list.append(msg)
else:
for image in game["keyImages"]:
if image["type"] == "Thumbnail":
if (
image.get("url")
and not game_thumbnail
and image["type"]
in [
"Thumbnail",
"VaultOpened",
"DieselStoreFrontWide",
"OfferImageWide",
]
):
game_thumbnail = image["url"]
break
for pair in game["customAttributes"]:
if pair["key"] == "developerName":
game_dev = pair["value"]
@@ -127,7 +114,7 @@ async def get_epic_free(bot: Bot, Type_Event: str):
game_url = "https://www.epicgames.com/store/zh-CN/p/{}".format(
game_url_part
)
if Type_Event == "Group":
if type_event == "Group":
_message = "[CQ:image,file={}]\n\nFREE now :: {} ({})\n{}\n此游戏由 {} 开发、{} 发行，将在 UTC 时间 {} 结束免费游玩，戳链接速度加入你的游戏库吧~\n{}\n".format(
game_thumbnail,
game_name,
plugins/fudu.py
+7 -6
View File
@@ -106,20 +106,21 @@ async def _(event: GroupMessageEvent):
"fudu", "FUDU_PROBABILITY"
) and not _fudu_list.is_repeater(event.group_id):
if random.random() < 0.2:
await fudu.finish("[[_task|fudu]]打断施法！")
if msg.endswith("打断施法！"):
await fudu.finish("[[_task|fudu]]打断" + msg)
else:
await fudu.finish("[[_task|fudu]]打断施法！")
_fudu_list.set_repeater(event.group_id)
if img and msg:
rst = msg + image(f"compare_{event.group_id}_img.jpg", "temp")
rst = msg + image(TEMP_PATH / f"compare_{event.group_id}_img.jpg")
elif img:
rst = image(f"compare_{event.group_id}_img.jpg", "temp")
rst = image(TEMP_PATH / f"compare_{event.group_id}_img.jpg")
elif msg:
rst = msg
else:
rst = ""
if rst:
if rst.endswith("打断施法！"):
rst = "打断" + rst
await fudu.send("[[_task|fudu]]" + rst)
await fudu.finish("[[_task|fudu]]" + rst)
async def get_fudu_img_hash(url, group_id):
plugins/genshin/query_resource_points/query_resource.py
+8 -8
View File
@@ -1,3 +1,4 @@
from pathlib import Path
from typing import Tuple, Optional, List
from configs.path_config import IMAGE_PATH, TEXT_PATH
from utils.message_builder import image
@@ -136,7 +137,7 @@ async def download_resource_data(semaphore: Semaphore):
asyncio.ensure_future(
download_image(
img_url,
f"{icon_path}/{id_}.png",
icon_path / f"{id_}.png",
semaphore,
True,
)
@@ -178,7 +179,7 @@ async def download_map_init(semaphore: Semaphore, flag: bool = False):
map_url = _map_data["url"]
await download_image(
map_url,
f"{map_path}/{idx}.png",
map_path / f"{idx}.png",
semaphore,
force_flag=flag,
)
@@ -227,15 +228,14 @@ async def download_resource_type():
except (TimeoutError, ConnectTimeout):
logger.warning("下载原神资源类型数据超时....")
except Exception as e:
logger.error(f"载原神资源类型数据超时 {type(e)}：{e}")
logger.error(f"载原神资源类型数据错误 {type(e)}：{e}")
# 初始化资源图标
def gen_icon(icon: str):
def gen_icon(icon: Path):
A = BuildImage(0, 0, background=f"{icon_path}/box.png")
B = BuildImage(0, 0, background=f"{icon_path}/box_alpha.png")
icon_ = icon_path / f"{icon}"
icon_img = BuildImage(115, 115, background=icon_)
icon_img = BuildImage(115, 115, background=icon)
icon_img.circle()
B.paste(icon_img, (17, 10), True)
B.paste(A, alpha=True)
@@ -246,14 +246,14 @@ def gen_icon(icon: str):
# 下载图片
async def download_image(
img_url: str,
path: str,
path: Path,
semaphore: Semaphore,
gen_flag: bool = False,
force_flag: bool = False,
):
async with semaphore:
try:
if not os.path.exists(path) or not is_valid or force_flag:
if not path.exists() or not is_valid(path) or force_flag:
if await AsyncHttpx.download_file(img_url, path, timeout=10):
logger.info(f"下载原神资源图标：{img_url}")
if gen_flag:
plugins/open_cases/__init__.py
+4 -4
View File
@@ -211,16 +211,16 @@ update_price = on_command("更新开箱价格", priority=1, permission=SUPERUSER
@update_price.handle()
async def _( event: MessageEvent):
await update_price.send(await util_get_buff_price(str(event.get_message())))
async def _(event: MessageEvent, arg: Message = CommandArg()):
await update_price.send(await util_get_buff_price(arg.extract_plain_text().strip()))
update_img = on_command("更新开箱图片", priority=1, permission=SUPERUSER, block=True)
@update_img.handle()
async def _(event: MessageEvent):
await update_img.send(await util_get_buff_img(str(event.get_message())))
async def _(event: MessageEvent, arg: Message = CommandArg()):
await update_img.send(await util_get_buff_img(str(arg.extract_plain_text().strip())))
# 重置开箱
plugins/pid_search.py
+5 -3
View File
@@ -1,7 +1,9 @@
from nonebot import on_command
from nonebot.adapters.onebot.v11 import Bot, MessageEvent, Message, GroupMessageEvent
from nonebot.typing import T_State
from utils.utils import is_number
from configs.config import Config
from utils.utils import is_number, change_pixiv_image_links
from utils.message_builder import image
from services.log import logger
from asyncio.exceptions import TimeoutError
@@ -31,8 +33,6 @@ __plugin_settings__ = {
pid_search = on_command("p搜", aliases={"pixiv搜", "P搜"}, priority=5, block=True)
url = "https://api.obfs.dev/api/pixiv/"
@pid_search.handle()
async def _h(event: MessageEvent, state: T_State, arg: Message = CommandArg()):
@@ -50,6 +50,7 @@ headers = {
@pid_search.got("pid", prompt="需要查询的图片PID是？")
async def _g(event: MessageEvent, state: T_State, pid: str = Arg("pid")):
url = Config.get_config("hibiapi", "HIBIAPI") + "/api/pixiv/"
if pid in ["取消", "算了"]:
await pid_search.finish("已取消操作...")
if not is_number(pid):
@@ -84,6 +85,7 @@ async def _g(event: MessageEvent, state: T_State, pid: str = Arg("pid")):
for image_url in data["meta_pages"]:
image_list.append(image_url["image_urls"]["original"])
for i, img_url in enumerate(image_list):
img_url = change_pixiv_image_links(img_url)
if not await AsyncHttpx.download_file(
img_url,
IMAGE_PATH / "temp" / f"pid_search_{event.user_id}_{i}.png",
plugins/pix_gallery/_data_source.py
+3 -3
View File
@@ -20,9 +20,9 @@ try:
except ModuleNotFoundError:
import json
if str(platform.system()).lower() == "windows":
policy = asyncio.WindowsSelectorEventLoopPolicy()
asyncio.set_event_loop_policy(policy)
# if str(platform.system()).lower() == "windows":
# policy = asyncio.WindowsSelectorEventLoopPolicy()
# asyncio.set_event_loop_policy(policy)
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6;"
plugins/pix_gallery/_model/omega_pixiv_illusts.py
+8 -9
View File
@@ -139,15 +139,14 @@ class OmegaPixivIllusts(db.Model):
data = await cls.select('pid').gino.all()
return [x[0] for x in data]
@classmethod
async def test(cls, nsfw_tag: int = 1):
if nsfw_tag is not None:
query = cls.query.where(cls.nsfw_tag == nsfw_tag)
else:
query = cls.query
query = query.where((cls.width - cls.height) < 50)
for x in await query.gino.all():
print(x.pid)
# async def test(cls, nsfw_tag: int = 1):
# if nsfw_tag is not None:
# query = cls.query.where(cls.nsfw_tag == nsfw_tag)
# else:
# query = cls.query
# query = query.where((cls.width - cls.height) < 50)
# for x in await query.gino.all():
# print(x.pid)
plugins/pix_gallery/pix.py
+11 -5
View File
@@ -18,6 +18,7 @@ usage：
查看 pix 好康图库
指令：
pix ?*[tags]: 通过 tag 获取相似图片，不含tag时随机抽取
pid [uid]: 通过uid获取图片
pix pid[pid]: 查看图库中指定pid图片
""".strip()
__plugin_superuser_usage__ = """
@@ -49,7 +50,9 @@ __plugin_configs__ = {
"value": None,
"help": "单次发送的图片数量达到指定值时转发为合并消息",
"default_value": None,
}
},
"ALLOW_GROUP_SETU": {"value": False, "help": "允许非超级用户使用-s参数", "default_value": False},
"ALLOW_GROUP_R18": {"value": False, "help": "允许非超级用户使用-r参数", "default_value": False},
}
@@ -80,10 +83,13 @@ async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
nsfw_tag = 2
else:
nsfw_tag = 0
if nsfw_tag != 0 and str(event.user_id) not in bot.config.superusers:
await pix.finish("你不能看这些噢，这些都是是留给管理员看的...")
if n := len(x) == 1 and is_number(x[0]):
num = int(x[-1])
if str(event.user_id) not in bot.config.superusers:
if (nsfw_tag == 1 and not Config.get_config("pix", "ALLOW_GROUP_SETU")) or (
nsfw_tag == 2 and not Config.get_config("pix", "ALLOW_GROUP_R18")
):
await pix.finish("你不能看这些噢，这些都是是留给管理员看的...")
if n := len(x) == 1 and is_number(x[0]) and int(x[0]) < 100:
num = int(x[0])
keyword = ""
elif n > 1:
if is_number(x[-1]):
plugins/pixiv_rank_search/__init__.py
+4 -2
View File
@@ -44,6 +44,8 @@ usage：
搜图 樱岛麻衣
搜图 樱岛麻衣 5
搜图 樱岛麻衣 5 r18
搜图 樱岛麻衣#1000users 5
【多个关键词用#分割】
【默认为 热度排序】
【注意空格！！】【在线搜索会较慢】【数量可能不符？可能该页数量不够，也可能被R-18屏蔽】
""".strip()
@@ -161,8 +163,8 @@ async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
info_list = None
num = 10
page = 1
if (n := len(msg)) == 1:
keyword = msg[0]
if (n := len(msg)) > 0:
keyword = msg[0].replace("#"," ")
if n > 1:
if not is_number(msg[1]):
await pixiv_keyword.finish("图片数量必须是数字！", at_sender=True)
plugins/pixiv_rank_search/data_source.py
+4 -4
View File
@@ -8,10 +8,10 @@ from services.log import logger
from pathlib import Path
import platform
if platform.system() == "Windows":
import asyncio
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# if platform.system() == "Windows":
# import asyncio
#
# asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
headers = {
plugins/send_setu_/send_setu/__init__.py
+11 -4
View File
@@ -97,6 +97,7 @@ __plugin_configs__ = {
},
"TIMEOUT": {"value": 10, "help": "色图下载超时限制(秒)", "default_value": 10},
"SHOW_INFO": {"value": True, "help": "是否显示色图的基本信息，如PID等", "default_value": True},
"ALLOW_GROUP_R18": {"value": False, "help": "在群聊中启用R18权限", "default_value": False},
}
Config.add_plugin_config("pixiv", "PIXIV_NGINX_URL", "i.pixiv.re", help_="Pixiv反向代理")
@@ -148,10 +149,16 @@ async def _(
if cmd[0] == "色图r" and isinstance(event, PrivateMessageEvent):
r18 = 1
num = 10
elif cmd[0] == "色图r" and isinstance(event, GroupMessageEvent):
await setu.finish(
random.choice(["这种不好意思的东西怎么可能给这么多人看啦", "羞羞脸！给我滚出克私聊！", "变态变态变态变态大变态！"])
)
elif (
cmd[0] == "色图r"
and isinstance(event, GroupMessageEvent)
):
if not Config.get_config("send_setu", "ALLOW_GROUP_R18"):
await setu.finish(
random.choice(["这种不好意思的东西怎么可能给这么多人看啦", "羞羞脸！给我滚出克私聊！", "变态变态变态变态大变态！"])
)
else:
r18 = 1
# 有 数字 的话先尝试本地色图id
if msg and is_number(msg):
setu_list, code = await get_setu_list(int(msg), r18=r18)
plugins/sign_in/utils.py
+17 -21
View File
@@ -24,6 +24,8 @@ from io import BytesIO
import asyncio
import random
import nonebot
import time
import locale
import os
@@ -55,13 +57,10 @@ async def get_card(
date = datetime.now().date()
_type = "view" if is_card_view else "sign"
card_file = (
Path(SIGN_TODAY_CARD_PATH)
/ f"{user_id}_{user.group_id}_{_type}_{date}.png"
Path(SIGN_TODAY_CARD_PATH) / f"{user_id}_{user.group_id}_{_type}_{date}.png"
)
if card_file.exists():
return image(
f"{user_id}_{user.group_id}_{_type}_{date}.png", "sign/today_card"
)
return image(f"{user_id}_{user.group_id}_{_type}_{date}.png", "sign/today_card")
else:
if add_impression == -1:
card_file = (
@@ -75,9 +74,7 @@ async def get_card(
)
is_card_view = True
ava = BytesIO(await get_user_avatar(user_id))
uid = await GroupInfoUser.get_group_member_uid(
user.user_qq, user.group_id
)
uid = await GroupInfoUser.get_group_member_uid(user.user_qq, user.group_id)
impression_list = None
if is_card_view:
_, impression_list, _ = await SignGroupUser.get_all_impression(
@@ -128,9 +125,13 @@ def _generate_card(
level, next_impression, previous_impression = get_level_and_next_impression(
user.impression
)
interpolation = next_impression - user.impression
if level == "9":
level = "8"
interpolation = 0
info_img.text((0, 0), f"· 好感度等级：{level} [{lik2relation[level]}]")
info_img.text((0, 20), f"· {NICKNAME}对你的态度：{level2attitude[level]}")
info_img.text((0, 40), f"· 距离升级还差 {next_impression - user.impression:.2f} 好感度")
info_img.text((0, 40), f"· 距离升级还差 {interpolation:.2f} 好感度")
bar_bk = BuildImage(220, 20, background=SIGN_RESOURCE_PATH / "bar_white.png")
bar = BuildImage(220, 20, background=SIGN_RESOURCE_PATH / "bar.png")
@@ -240,7 +241,9 @@ def _generate_card(
f"上次签到日期：{'从未' if user.checkin_time_last == datetime.min else user.checkin_time_last.date()}",
)
today_data.text((0, 25), f"总金币：{gold}")
default_setu_prob = Config.get_config("send_setu", "INITIAL_SETU_PROBABILITY") * 100
default_setu_prob = (
Config.get_config("send_setu", "INITIAL_SETU_PROBABILITY") * 100
)
today_data.text(
(0, 50),
f"色图概率：{(default_setu_prob + user.impression if user.impression < 100 else 100):.2f}%",
@@ -259,15 +262,12 @@ def _generate_card(
today_data.text((0, 25), f"金币 + {gold}")
_type = "sign"
current_date = datetime.now()
week = current_date.isoweekday()
current_datetime_str = current_date.strftime("%Y-%m-%d %a %H:%M:%S")
data = current_date.date()
hour = current_date.hour
minute = current_date.minute
second = current_date.second
data_img = BuildImage(
0,
0,
plain_text=f"时间：{data} {weekdays[week]} {hour}:{minute}:{second}",
plain_text=f"时间：{current_datetime_str}",
color=(255, 255, 255, 0),
font_size=20,
)
@@ -286,12 +286,8 @@ def _generate_card(
bk.paste(today_sign_text_img, (550, 180), True)
bk.paste(today_data, (580, 220), True)
bk.paste(watermark, (15, 400), True)
bk.save(
SIGN_TODAY_CARD_PATH / f"{user_id}_{user.group_id}_{_type}_{data}.png"
)
return image(
f"{user_id}_{user.group_id}_{_type}_{data}.png", "sign/today_card"
)
bk.save(SIGN_TODAY_CARD_PATH / f"{user_id}_{user.group_id}_{_type}_{data}.png")
return image(f"{user_id}_{user.group_id}_{_type}_{data}.png", "sign/today_card")
def generate_progress_bar_pic():
plugins/update_gocqhttp/data_source.py
+2 -2
View File
@@ -5,8 +5,8 @@ import asyncio
import platform
import os
if platform.system() == "Windows":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# if platform.system() == "Windows":
# asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
url = "https://github.com/Mrs4s/go-cqhttp/releases"
plugins/alapi/wbtop.py → plugins/wbtop/__init__.py
Executable → Regular
+78 -70
View File
@@ -1,70 +1,78 @@
from nonebot import on_command
from nonebot.adapters.onebot.v11 import MessageEvent, GroupMessageEvent, Message
from nonebot.params import CommandArg
from services.log import logger
from ._data_source import get_data, gen_wbtop_pic
from utils.utils import is_number
from configs.path_config import IMAGE_PATH
from utils.http_utils import AsyncPlaywright
import asyncio
__zx_plugin_name__ = "微博热搜"
__plugin_usage__ = """
usage：
在QQ上吃个瓜
指令：
微博热搜：发送实时热搜
微博热搜 [id]：截图该热搜页面
示例：微博热搜 5
""".strip()
__plugin_des__ = "刚买完瓜，在吃瓜现场"
__plugin_cmd__ = ["微博热搜", "微博热搜 [id]"]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["微博热搜"],
}
wbtop = on_command("wbtop", aliases={"微博热搜"}, priority=5, block=True)
wbtop_url = "https://v2.alapi.cn/api/new/wbtop"
wbtop_data = []
@wbtop.handle()
async def _(event: MessageEvent, arg: Message = CommandArg()):
global wbtop_data
msg = arg.extract_plain_text().strip()
if not wbtop_data or not msg:
data, code = await get_data(wbtop_url)
if code != 200:
await wbtop.finish(data, at_sender=True)
wbtop_data = data["data"]
if not msg:
img = await asyncio.get_event_loop().run_in_executor(
None, gen_wbtop_pic, wbtop_data
)
await wbtop.send(img)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 查询微博热搜"
)
if is_number(msg) and 0 < int(msg) <= 50:
url = wbtop_data[int(msg) - 1]["url"]
await wbtop.send("开始截取数据...")
img = await AsyncPlaywright.screenshot(
url,
f"{IMAGE_PATH}/temp/wbtop_{event.user_id}.png",
"#pl_feed_main",
wait_time=12
)
if img:
await wbtop.send(img)
else:
await wbtop.send("发生了一些错误.....")
from nonebot import on_command
from nonebot.adapters.onebot.v11 import MessageEvent, GroupMessageEvent, Message
from nonebot.params import CommandArg
from services.log import logger
from .data_source import gen_wbtop_pic,get_wbtop
from utils.utils import is_number
from configs.path_config import IMAGE_PATH
from utils.http_utils import AsyncPlaywright
import asyncio
import datetime
__zx_plugin_name__ = "微博热搜"
__plugin_usage__ = """
usage：
在QQ上吃个瓜
指令：
微博热搜：发送实时热搜
微博热搜 [id]：截图该热搜页面
示例：微博热搜 5
""".strip()
__plugin_des__ = "刚买完瓜，在吃瓜现场"
__plugin_cmd__ = ["微博热搜", "微博热搜 [id]"]
__plugin_version__ = 0.2
__plugin_author__ = "HibiKier & yajiwa"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["微博热搜"],
}
wbtop = on_command("wbtop", aliases={"微博热搜"}, priority=5, block=True)
wbtop_url = "https://weibo.com/ajax/side/hotSearch"
wbtop_data = []
@wbtop.handle()
async def _(event: MessageEvent, arg: Message = CommandArg()):
global wbtop_data
msg = arg.extract_plain_text().strip()
if not wbtop_data or not msg:
if wbtop_data:
now_time = datetime.datetime.now()
if now_time > wbtop_data["time"] + datetime.timedelta(minutes=5):
data, code = await get_wbtop(wbtop_url)
if code != 200:
await wbtop.finish(data, at_sender=True)
wbtop_data = data
else:
data, code = await get_wbtop(wbtop_url)
if code != 200:
await wbtop.finish(data, at_sender=True)
wbtop_data = data
if not msg:
img = await asyncio.get_event_loop().run_in_executor(
None, gen_wbtop_pic, wbtop_data["data"]
)
await wbtop.send(img)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 查询微博热搜"
)
if is_number(msg) and 0 < int(msg) <= 50:
url = wbtop_data["data"][int(msg) - 1]["url"]
await wbtop.send("开始截取数据...")
img = await AsyncPlaywright.screenshot(
url,
f"{IMAGE_PATH}/temp/wbtop_{event.user_id}.png",
"#pl_feed_main",
wait_time=12
)
if img:
await wbtop.send(img)
else:
await wbtop.send("发生了一些错误.....")
plugins/wbtop/data_source.py
+62
View File
@@ -0,0 +1,62 @@
from nonebot.adapters.onebot.v11 import MessageSegment
from utils.image_utils import BuildImage
from utils.message_builder import image
from configs.path_config import IMAGE_PATH
from typing import Tuple, Union
from utils.http_utils import AsyncHttpx
import datetime
async def get_wbtop(url: str) -> Tuple[Union[dict, str], int]:
"""
:param url: 请求链接
"""
n = 0
while True:
try:
data = []
get_response = (await AsyncHttpx.get(url, timeout=20))
if get_response.status_code == 200:
data_json = get_response.json()['data']['realtime']
for data_item in data_json:
# 如果是广告，则不添加
if 'is_ad' in data_item:
continue
dic = {
'hot_word': data_item['note'],
'hot_word_num': str(data_item['num']),
'url': 'https://s.weibo.com/weibo?q=%23' + data_item['word'] + '%23',
}
data.append(dic)
if not data:
return "没有搜索到...", 997
return {'data': data, 'time': datetime.datetime.now()}, 200
else:
if n > 2:
return f'获取失败,请十分钟后再试', 999
else:
n += 1
continue
except TimeoutError:
return "超时了....", 998
def gen_wbtop_pic(data: dict) -> MessageSegment:
"""
生成微博热搜图片
:param data: 微博热搜数据
"""
bk = BuildImage(700, 32 * 50 + 280, 700, 32, color="#797979")
wbtop_bk = BuildImage(700, 280, background=f"{IMAGE_PATH}/other/webtop.png")
bk.paste(wbtop_bk)
text_bk = BuildImage(700, 32 * 50, 700, 32, color="#797979")
for i, data in enumerate(data):
title = f"{i + 1}. {data['hot_word']}"
hot = data["hot_word_num"]
img = BuildImage(700, 30, font_size=20)
w, h = img.getsize(title)
img.text((10, int((30 - h) / 2)), title)
img.text((580, int((30 - h) / 2)), hot)
text_bk.paste(img)
bk.paste(text_bk, (0, 280))
return image(b64=bk.pic2bs4())
plugins/word_clouds/data_source.py
+19 -11
View File
@@ -12,6 +12,7 @@ import numpy as np
import matplotlib.pyplot as plt
from io import BytesIO
from configs.path_config import IMAGE_PATH, FONT_PATH
from services import logger
from utils.http_utils import AsyncHttpx
from models.chat_history import ChatHistory
from configs.config import Config
@@ -19,10 +20,11 @@ from configs.config import Config
async def pre_precess(msg: List[str], config) -> str:
return await asyncio.get_event_loop().run_in_executor(
None, _pre_precess, msg,config)
None, _pre_precess, msg, config
)
def _pre_precess(msg: List[str],config) -> str:
def _pre_precess(msg: List[str], config) -> str:
"""对消息进行预处理"""
# 过滤掉命令
command_start = tuple([i for i in config.command_start if i])
@@ -46,7 +48,6 @@ def _pre_precess(msg: List[str],config) -> str:
return msg
async def draw_word_cloud(messages, config):
wordcloud_dir = IMAGE_PATH / "wordcloud"
wordcloud_dir.mkdir(exist_ok=True, parents=True)
@@ -57,19 +58,21 @@ async def draw_word_cloud(messages, config):
url = "https://ghproxy.com/https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/resources/image/wordcloud/default.png"
try:
await AsyncHttpx.download_file(url, zx_logo_path)
except:
except Exception as e:
logger.error(f"词云图片资源下载发生错误 {type(e)}：{e}")
return False
if not wordcloud_ttf.exists():
ttf_url = 'https://ghproxy.com/https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/resources/font/STKAITI.TTF'
ttf_url = "https://ghproxy.com/https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/resources/font/STKAITI.TTF"
try:
await AsyncHttpx.download_file(ttf_url, wordcloud_ttf)
except:
except Exception as e:
logger.error(f"词云字体资源下载发生错误 {type(e)}：{e}")
return False
topK = min(int(len(messages)), 100000)
read_name = jieba.analyse.extract_tags(await pre_precess(messages, config), topK=topK,
withWeight=True,
allowPOS=())
read_name = jieba.analyse.extract_tags(
await pre_precess(messages, config), topK=topK, withWeight=True, allowPOS=()
)
name = []
value = []
for t in read_name:
@@ -79,10 +82,11 @@ async def draw_word_cloud(messages, config):
name[i] = str(name[i])
dic = dict(zip(name, value))
if Config.get_config("word_clouds", "WORD_CLOUDS_TEMPLATE") == 1:
def random_pic(base_path: str) -> str:
path_dir = os.listdir(base_path)
path = random.sample(path_dir, 1)[0]
return (str(base_path) + "/" + str(path))
return str(base_path) + "/" + str(path)
mask = np.array(IMG.open(random_pic(wordcloud_dir)))
wc = WordCloud(
@@ -113,7 +117,11 @@ async def draw_word_cloud(messages, config):
async def get_list_msg(user_id, group_id, days):
messages_list = await ChatHistory()._get_msg(uid=user_id, gid=group_id, type_="group", days=days).gino.all()
messages_list = (
await ChatHistory()
._get_msg(uid=user_id, gid=group_id, type_="group", days=days)
.gino.all()
)
if messages_list:
messages = [i.text for i in messages_list]
return messages
utils/image_utils.py
+1 -1
View File
@@ -121,7 +121,7 @@ def fig2b64(plt_: plt) -> str:
return "base64://" + base64_str
def is_valid(file: str) -> bool:
def is_valid(file: Union[str, Path]) -> bool:
"""
说明：
判断图片是否损坏
utils/manager/none_plugin_count_manager.py
+3 -7
View File
@@ -4,7 +4,6 @@ from pathlib import Path
class NonePluginCountManager(StaticData):
"""
插件加载容忍管理器，当连续 max_count 次插件加载，视为删除插件，清楚数据
"""
@@ -15,6 +14,8 @@ class NonePluginCountManager(StaticData):
:param max_count: 容忍最大次数
"""
super().__init__(file)
if not self._data:
self._data = {}
self._max_count = max_count
def add_count(self, module: str, count: int = 1):
@@ -42,10 +43,5 @@ class NonePluginCountManager(StaticData):
:param module: 模块
"""
if module in self._data.keys():
return self._data.keys() > self._max_count
return self._data[module] >= self._max_count
return False
utils/manager/plugins2block_manager.py
+6 -13
View File
@@ -21,18 +21,17 @@ class Plugins2blockManager(StaticData):
with open(file, "r", encoding="utf8") as f:
self._data = yaml.load(f)
if "PluginBlockLimit" in self._data.keys():
self._data = (
self._data["PluginBlockLimit"] if self._data["PluginBlockLimit"] else {}
)
self._data = self._data["PluginBlockLimit"] or {}
def add_block_limit(
self,
plugin: str,
*,
status: Optional[bool] = True,
check_type: Optional[str] = "all",
limit_type: Optional[str] = "user",
rst: Optional[str] = None,
data_dict: Optional[dict] = None,
**kwargs # 用于接收额外实参
):
"""
添加插件调用 block 限制
@@ -41,16 +40,10 @@ class Plugins2blockManager(StaticData):
:param check_type: 检查类型 'private'/'group'/'all'，限制私聊/群聊/全部
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
:param data_dict: 封装好的字典数据
"""
if data_dict:
status = data_dict.get("status")
check_type = data_dict.get("check_type")
limit_type = data_dict.get("limit_type")
rst = data_dict.get("rst")
status = status if status is not None else True
check_type = check_type if check_type else "all"
limit_type = limit_type if limit_type else "user"
status = status or True
check_type = check_type or "all"
limit_type = limit_type or "user"
if check_type not in ["all", "group", "private"]:
raise ValueError(
f"{plugin} 添加block限制错误，‘check_type‘ 必须为 'private'/'group'/'all'"
utils/manager/plugins2cd_manager.py
+5 -12
View File
@@ -34,7 +34,7 @@ class Plugins2cdManager(StaticData):
check_type: Optional[str] = "all",
limit_type: Optional[str] = "user",
rst: Optional[str] = None,
data_dict: Optional[dict] = None,
**kwargs # 用于接收额外实参
):
"""
添加插件调用 cd 限制
@@ -44,18 +44,11 @@ class Plugins2cdManager(StaticData):
:param check_type: 检查类型 'private'/'group'/'all'，限制私聊/群聊/全部
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
:param data_dict: 封装好的字典数据
"""
if data_dict:
cd = data_dict.get("cd")
status = data_dict.get("status")
check_type = data_dict.get("check_type")
limit_type = data_dict.get("limit_type")
rst = data_dict.get("rst")
cd = cd if cd is not None else 5
status = status if status is not None else True
check_type = check_type if check_type else "all"
limit_type = limit_type if limit_type else "user"
cd = cd or 5
status = status or True
check_type = check_type or "all"
limit_type = limit_type or "user"
if check_type not in ["all", "group", "private"]:
raise ValueError(
f"{plugin} 添加cd限制错误，‘check_type‘ 必须为 'private'/'group'/'all'"
utils/manager/plugins2count_manager.py
+4 -10
View File
@@ -33,7 +33,7 @@ class Plugins2countManager(StaticData):
status: Optional[bool] = True,
limit_type: Optional[str] = "user",
rst: Optional[str] = None,
data_dict: Optional[dict] = None,
**kwargs # 用于接收额外实参
):
"""
添加插件调用 次数 限制
@@ -42,16 +42,10 @@ class Plugins2countManager(StaticData):
:param status: 默认开关状态
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
:param data_dict: 封装好的字典数据
"""
if data_dict:
max_count = data_dict.get("max_count")
status = data_dict.get("status")
limit_type = data_dict.get("limit_type")
rst = data_dict.get("rst")
status = status if status is not None else True
limit_type = limit_type if limit_type else "user"
max_count = max_count if max_count is not None else 5
max_count = max_count or 5
status = status or True
limit_type = limit_type or "user"
if limit_type not in ["user", "group"]:
raise ValueError(f"{plugin} 添加count限制错误，‘limit_type‘ 必须为 'user'/'group'")
self._data[plugin] = {
utils/manager/plugins2settings_manager.py
+8 -17
View File
@@ -47,26 +47,17 @@ class Plugins2settingsManager(StaticData):
:param plugin_type: 插件类型
:param cost_gold: 需要消费的金币
"""
if kwargs:
level = kwargs.get("level") if kwargs.get("level") is not None else 5
default_status = (
kwargs.get("default_status")
if kwargs.get("default_status") is not None
else True
)
limit_superuser = (
kwargs.get("limit_superuser")
if kwargs.get("limit_superuser") is not None
else False
)
cmd = kwargs.get("cmd") if kwargs.get("cmd") is not None else []
cost_gold = cost_gold if kwargs.get("cost_gold") else 0
level = level or 5
cmd = cmd or []
cost_gold = cost_gold or 0
self._data[plugin] = {
"level": level if level is not None else 5,
"default_status": default_status if default_status is not None else True,
"limit_superuser": limit_superuser
if limit_superuser is not None
else False,
"limit_superuser": (
limit_superuser
if limit_superuser is not None
else False
),
"cmd": cmd,
"plugin_type": list(
plugin_type if plugin_type is not None else ("normal",)
utils/utils.py
+3 -1
View File
@@ -1,7 +1,7 @@
from datetime import datetime
from collections import defaultdict
from nonebot import require
from configs.config import SYSTEM_PROXY
from configs.config import SYSTEM_PROXY, Config
from typing import List, Union, Optional, Type, Any
from nonebot.adapters.onebot.v11 import Bot, Message
from nonebot.matcher import matchers, Matcher
@@ -389,6 +389,8 @@ def change_pixiv_image_links(
url = img_sp[0]
img_type = img_sp[1]
url = url.replace("original", "master") + f"_master1200.{img_type}"
if not nginx_url:
nginx_url = Config.get_config("pixiv", "PIXIV_NGINX_URL")
if nginx_url:
url = (
url.replace("i.pximg.net", nginx_url)