mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 06:12:53 +08:00
修复签到好感度进度条错误
This commit is contained in:
parent
09d251de21
commit
379a689ebc
@ -334,6 +334,7 @@ PS: **ARM平台** 请使用全量版 同时 **如果你的机器 RAM < 1G 可能
|
||||
### 2023/3/18
|
||||
|
||||
* 修复色图重复发送相同图片
|
||||
* 修复签到好感度进度条错误
|
||||
|
||||
### 2023/3/12 \[v0.1.6.7]
|
||||
|
||||
|
||||
@ -26,7 +26,6 @@ from .config import (
|
||||
level2attitude,
|
||||
lik2level,
|
||||
lik2relation,
|
||||
weekdays,
|
||||
)
|
||||
|
||||
driver: Driver = nonebot.get_driver()
|
||||
@ -48,7 +47,7 @@ async def init_image():
|
||||
|
||||
|
||||
async def get_card(
|
||||
user: "SignGroupUser",
|
||||
user: SignGroupUser,
|
||||
nickname: str,
|
||||
add_impression: Optional[float],
|
||||
gold: Optional[int],
|
||||
@ -146,19 +145,13 @@ def _generate_card(
|
||||
|
||||
bar_bk = BuildImage(220, 20, background=SIGN_RESOURCE_PATH / "bar_white.png")
|
||||
bar = BuildImage(220, 20, background=SIGN_RESOURCE_PATH / "bar.png")
|
||||
ratio = 1 - (next_impression - user.impression) / (
|
||||
next_impression - previous_impression
|
||||
)
|
||||
bar.resize(w=int(bar.w * ratio), h=bar.h)
|
||||
bar_bk.paste(
|
||||
bar,
|
||||
(
|
||||
-int(
|
||||
220
|
||||
* (
|
||||
(next_impression - user.impression)
|
||||
/ (next_impression - previous_impression)
|
||||
)
|
||||
),
|
||||
0,
|
||||
),
|
||||
True,
|
||||
alpha=True,
|
||||
)
|
||||
font_size = 30
|
||||
if "好感度双倍加持卡" in gift:
|
||||
|
||||
@ -54,8 +54,8 @@ class AsyncHttpx:
|
||||
"""
|
||||
if not headers:
|
||||
headers = get_user_agent()
|
||||
proxy = proxy if proxy else cls.proxy if use_proxy else None
|
||||
async with httpx.AsyncClient(proxies=proxy, verify=verify) as client:
|
||||
proxy_ = proxy if proxy else cls.proxy if use_proxy else None
|
||||
async with httpx.AsyncClient(proxies=proxy_, verify=verify) as client:
|
||||
return await client.get(
|
||||
url,
|
||||
params=params,
|
||||
@ -75,8 +75,8 @@ class AsyncHttpx:
|
||||
files: Any = None,
|
||||
verify: bool = True,
|
||||
use_proxy: bool = True,
|
||||
proxy: Dict[str, str] = None,
|
||||
json: Optional[Dict[str, Union[Any]]] = None,
|
||||
proxy: Optional[Dict[str, str]] = None,
|
||||
json: Optional[Dict[str, Any]] = None,
|
||||
params: Optional[Dict[str, str]] = None,
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
cookies: Optional[Dict[str, str]] = None,
|
||||
@ -101,8 +101,8 @@ class AsyncHttpx:
|
||||
"""
|
||||
if not headers:
|
||||
headers = get_user_agent()
|
||||
proxy = proxy if proxy else cls.proxy if use_proxy else None
|
||||
async with httpx.AsyncClient(proxies=proxy, verify=verify) as client:
|
||||
proxy_ = proxy if proxy else cls.proxy if use_proxy else None
|
||||
async with httpx.AsyncClient(proxies=proxy_, verify=verify) as client:
|
||||
return await client.post(
|
||||
url,
|
||||
content=content,
|
||||
@ -125,7 +125,7 @@ class AsyncHttpx:
|
||||
params: Optional[Dict[str, str]] = None,
|
||||
verify: bool = True,
|
||||
use_proxy: bool = True,
|
||||
proxy: Dict[str, str] = None,
|
||||
proxy: Optional[Dict[str, str]] = None,
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
cookies: Optional[Dict[str, str]] = None,
|
||||
timeout: Optional[int] = 30,
|
||||
@ -175,10 +175,10 @@ class AsyncHttpx:
|
||||
else:
|
||||
if not headers:
|
||||
headers = get_user_agent()
|
||||
proxy = proxy if proxy else cls.proxy if use_proxy else None
|
||||
proxy_ = proxy if proxy else cls.proxy if use_proxy else None
|
||||
try:
|
||||
async with httpx.AsyncClient(
|
||||
proxies=proxy, verify=verify
|
||||
proxies=proxy_, verify=verify
|
||||
) as client:
|
||||
async with client.stream(
|
||||
"GET",
|
||||
@ -230,7 +230,7 @@ class AsyncHttpx:
|
||||
limit_async_number: Optional[int] = None,
|
||||
params: Optional[Dict[str, str]] = None,
|
||||
use_proxy: bool = True,
|
||||
proxy: Dict[str, str] = None,
|
||||
proxy: Optional[Dict[str, str]] = None,
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
cookies: Optional[Dict[str, str]] = None,
|
||||
timeout: Optional[int] = 30,
|
||||
|
||||
@ -265,6 +265,7 @@ class BuildImage:
|
||||
pos: Optional[Tuple[int, int]] = None,
|
||||
alpha: bool = False,
|
||||
center_type: Optional[Literal["center", "by_height", "by_width"]] = None,
|
||||
allow_negative: bool = False,
|
||||
):
|
||||
"""
|
||||
说明:
|
||||
@ -274,8 +275,9 @@ class BuildImage:
|
||||
:param pos: 贴图位置(左上角)
|
||||
:param alpha: 图片背景是否为透明
|
||||
:param center_type: 居中类型,可能的值 center: 完全居中,by_width: 水平居中,by_height: 垂直居中
|
||||
:param allow_negative: 允许使用负数作为坐标且不超出图片范围,从右侧开始计算
|
||||
"""
|
||||
await self.loop.run_in_executor(None, self.paste, img, pos, alpha, center_type)
|
||||
await self.loop.run_in_executor(None, self.paste, img, pos, alpha, center_type, allow_negative)
|
||||
|
||||
def paste(
|
||||
self,
|
||||
@ -283,6 +285,7 @@ class BuildImage:
|
||||
pos: Optional[Tuple[int, int]] = None,
|
||||
alpha: bool = False,
|
||||
center_type: Optional[Literal["center", "by_height", "by_width"]] = None,
|
||||
allow_negative: bool = False,
|
||||
):
|
||||
"""
|
||||
说明:
|
||||
@ -292,6 +295,7 @@ class BuildImage:
|
||||
:param pos: 贴图位置(左上角)
|
||||
:param alpha: 图片背景是否为透明
|
||||
:param center_type: 居中类型,可能的值 center: 完全居中,by_width: 水平居中,by_height: 垂直居中
|
||||
:param allow_negative: 允许使用负数作为坐标且不超出图片范围,从右侧开始计算
|
||||
"""
|
||||
if center_type:
|
||||
if center_type not in ["center", "by_height", "by_width"]:
|
||||
@ -311,7 +315,7 @@ class BuildImage:
|
||||
width = pos[0]
|
||||
height = int((self.h - img.h) / 2)
|
||||
pos = (width, height)
|
||||
if pos:
|
||||
if pos and allow_negative:
|
||||
if pos[0] < 0:
|
||||
pos = (self.w + pos[0], pos[1])
|
||||
if pos[1] < 0:
|
||||
|
||||
@ -1,5 +1,10 @@
|
||||
from typing import Tuple, Union, Dict
|
||||
from nonebot.adapters.onebot.v11 import MessageEvent, PrivateMessageEvent, GroupMessageEvent
|
||||
from typing import Dict, Optional, Tuple, Union
|
||||
|
||||
from nonebot.adapters.onebot.v11 import (
|
||||
GroupMessageEvent,
|
||||
MessageEvent,
|
||||
PrivateMessageEvent,
|
||||
)
|
||||
|
||||
|
||||
class WithdrawMessageManager:
|
||||
@ -27,7 +32,10 @@ class WithdrawMessageManager:
|
||||
self.data.remove(message_data)
|
||||
|
||||
def withdraw_message(
|
||||
self, event: MessageEvent, id_: Union[int, Dict[str, int]], conditions: Tuple[int, int]
|
||||
self,
|
||||
event: MessageEvent,
|
||||
id_: Union[int, Dict[str, int]],
|
||||
conditions: Optional[Tuple[int, int]],
|
||||
):
|
||||
"""
|
||||
便捷判断消息撤回
|
||||
@ -35,7 +43,7 @@ class WithdrawMessageManager:
|
||||
:param id_: 消息id 或 send 返回的字典
|
||||
:param conditions: 判断条件
|
||||
"""
|
||||
if conditions[0]:
|
||||
if conditions and conditions[0]:
|
||||
if (
|
||||
(conditions[1] == 0 and isinstance(event, PrivateMessageEvent))
|
||||
or (conditions[1] == 1 and isinstance(event, GroupMessageEvent))
|
||||
|
||||
@ -563,3 +563,67 @@ async def broadcast_group(
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Bot: {_bot.self_id} 获取群聊列表失败", command=log_cmd, e=e)
|
||||
|
||||
|
||||
async def broadcast_superuser(
|
||||
message: Union[str, Message, MessageSegment],
|
||||
bot: Optional[Union[Bot, List[Bot]]] = None,
|
||||
bot_id: Optional[Union[str, Set[str]]] = None,
|
||||
ignore_superuser: Optional[Set[int]] = None,
|
||||
check_func: Optional[Callable[[int], bool]] = None,
|
||||
log_cmd: Optional[str] = None,
|
||||
):
|
||||
"""获取所有Bot或指定Bot对象广播超级用户
|
||||
|
||||
Args:
|
||||
message (Any): 广播消息内容
|
||||
bot (Optional[Bot], optional): 指定bot对象. Defaults to None.
|
||||
bot_id (Optional[str], optional): 指定bot id. Defaults to None.
|
||||
ignore_superuser (Optional[List[int]], optional): 忽略的超级用户id. Defaults to None.
|
||||
check_func (Optional[Callable[[int], bool]], optional): 发送前对群聊检测方法,判断是否发送. Defaults to None.
|
||||
log_cmd (Optional[str], optional): 日志标记. Defaults to None.
|
||||
"""
|
||||
if not message:
|
||||
raise ValueError("超级用户广播消息不能为空")
|
||||
bot_dict = nonebot.get_bots()
|
||||
bot_list: List[Bot] = []
|
||||
if bot:
|
||||
if isinstance(bot, list):
|
||||
bot_list = bot
|
||||
else:
|
||||
bot_list.append(bot)
|
||||
elif bot_id:
|
||||
_bot_id_list = bot_id
|
||||
if isinstance(bot_id, str):
|
||||
_bot_id_list = [bot_id]
|
||||
for id_ in _bot_id_list:
|
||||
if bot_id in bot_dict:
|
||||
bot_list.append(bot_dict[bot_id])
|
||||
else:
|
||||
logger.warning(f"Bot:{id_} 对象未连接或不存在")
|
||||
else:
|
||||
bot_list = list(bot_dict.values())
|
||||
_used_user = []
|
||||
for _bot in bot_list:
|
||||
try:
|
||||
for user_id in _bot.config.superusers:
|
||||
try:
|
||||
if (
|
||||
ignore_superuser and int(user_id) in ignore_superuser
|
||||
) or user_id in _used_user:
|
||||
continue
|
||||
if check_func and not check_func(int(user_id)):
|
||||
continue
|
||||
_used_user.append(user_id)
|
||||
await _bot.send_private_message(
|
||||
user_id=int(user_id), message=message
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"广播超级用户发消息失败: {message}",
|
||||
command=log_cmd,
|
||||
user_id=user_id,
|
||||
e=e,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Bot: {_bot.self_id} 获取群聊列表失败", command=log_cmd, e=e)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user