mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 14:22:55 +08:00
214 lines
8.6 KiB
Python
214 lines
8.6 KiB
Python
from datetime import datetime, timedelta
|
|
from models.sign_group_user import SignGroupUser
|
|
from models.group_member_info import GroupInfoUser
|
|
from models.bag_user import BagUser
|
|
from configs.config import NICKNAME
|
|
from nonebot.adapters.cqhttp import MessageSegment
|
|
from asyncio.exceptions import TimeoutError
|
|
from utils.image_utils import CreateImg
|
|
from services.db_context import db
|
|
from .utils import get_card, SIGN_TODAY_CARD_PATH
|
|
from services.log import logger
|
|
from .random_event import random_event
|
|
from io import BytesIO
|
|
import random
|
|
import aiohttp
|
|
import math
|
|
import asyncio
|
|
import os
|
|
import secrets
|
|
|
|
|
|
async def group_user_check_in(
|
|
nickname: str, user_qq: int, group: int
|
|
) -> MessageSegment:
|
|
"Returns string describing the result of checking in"
|
|
present = datetime.now()
|
|
async with db.transaction():
|
|
# 取得相应用户
|
|
user = await SignGroupUser.ensure(user_qq, group, for_update=True)
|
|
# 如果同一天签到过，特殊处理
|
|
if (
|
|
user.checkin_time_last + timedelta(hours=8)
|
|
).date() >= present.date() or f"{user}_{group}_sign_{datetime.now().date()}" in os.listdir(
|
|
SIGN_TODAY_CARD_PATH
|
|
):
|
|
gold = await BagUser.get_gold(user_qq, group)
|
|
return await get_card(user, nickname, -1, gold, "")
|
|
return await _handle_check_in(nickname, user_qq, group, present) # ok
|
|
|
|
|
|
async def _handle_check_in(
|
|
nickname: str, user_qq: int, group: int, present: datetime
|
|
) -> MessageSegment:
|
|
user = await SignGroupUser.ensure(user_qq, group, for_update=True)
|
|
impression_added = (secrets.randbelow(90)+1)/100
|
|
critx2 = random.random()
|
|
add_probability = user.add_probability
|
|
specify_probability = user.specify_probability
|
|
if critx2 + add_probability > 0.97:
|
|
impression_added *= 2
|
|
elif critx2 < specify_probability:
|
|
impression_added *= 2
|
|
await SignGroupUser.sign(user, impression_added, present)
|
|
gold = random.randint(1, 100)
|
|
gift, gift_type = random_event(user.impression)
|
|
if gift_type == "gold":
|
|
await BagUser.add_gold(user_qq, group, gold + gift)
|
|
gift = f"额外金币 + {gift}"
|
|
else:
|
|
await BagUser.add_props(user_qq, group, gift)
|
|
gift += ' + 1'
|
|
if critx2 + add_probability > 0.97 or critx2 < specify_probability:
|
|
logger.info(
|
|
f"(USER {user.user_qq}, GROUP {user.belonging_group})"
|
|
f" CHECKED IN successfully. score: {user.impression:.2f} "
|
|
f"(+{impression_added * 2:.2f}).获取金币：{gold + gift if gift == 'gold' else gold}"
|
|
)
|
|
return await get_card(user, nickname, impression_added, gold, gift, True)
|
|
else:
|
|
logger.info(
|
|
f"(USER {user.user_qq}, GROUP {user.belonging_group})"
|
|
f" CHECKED IN successfully. score: {user.impression:.2f} "
|
|
f"(+{impression_added:.2f}).获取金币：{gold + gift if gift == 'gold' else gold}"
|
|
)
|
|
return await get_card(user, nickname, impression_added, gold, gift)
|
|
|
|
|
|
async def group_user_check(nickname: str, user_qq: int, group: int) -> MessageSegment:
|
|
# heuristic: if users find they have never checked in they are probable to check in
|
|
user = await SignGroupUser.ensure(user_qq, group)
|
|
gold = await BagUser.get_gold(user_qq, group)
|
|
return await get_card(user, nickname, None, gold, "", is_card_view=True)
|
|
|
|
|
|
async def group_impression_rank(group: int) -> str:
|
|
result = "\t好感度排行榜\t\n"
|
|
user_qq_list, impression_list, _ = await SignGroupUser.get_all_impression(group)
|
|
_count = 11
|
|
if user_qq_list and impression_list:
|
|
for i in range(1, len(user_qq_list)):
|
|
if len(user_qq_list) == 0 or len(impression_list) == 0 or i == _count:
|
|
break
|
|
impression = max(impression_list)
|
|
index = impression_list.index(impression)
|
|
user_qq = user_qq_list[index]
|
|
try:
|
|
user_name = (
|
|
await GroupInfoUser.get_member_info(user_qq, group)
|
|
).user_name
|
|
except AttributeError:
|
|
logger.info(f"USER {user_qq}, GROUP {group} 不在群内")
|
|
_count += 1
|
|
impression_list.remove(impression)
|
|
user_qq_list.remove(user_qq)
|
|
continue
|
|
result += f"{i - _count + 11}. {user_name}: {impression:.2f}\n"
|
|
impression_list.remove(impression)
|
|
user_qq_list.remove(user_qq)
|
|
return result[:-1]
|
|
|
|
|
|
async def random_gold(user_id, group_id, impression):
|
|
if impression < 1:
|
|
impression = 1
|
|
gold = random.randint(1, 100) + random.randint(1, int(impression))
|
|
if await BagUser.add_gold(user_id, group_id, gold):
|
|
return gold
|
|
else:
|
|
return 0
|
|
|
|
|
|
# 签到总榜
|
|
async def impression_rank(group_id: int, data: dict):
|
|
user_qq_list, impression_list, group_list = await SignGroupUser.get_all_impression(
|
|
group_id
|
|
)
|
|
users, impressions, groups = [], [], []
|
|
num = 0
|
|
for i in range(105 if len(user_qq_list) > 105 else len(user_qq_list)):
|
|
impression = max(impression_list)
|
|
index = impression_list.index(impression)
|
|
user = user_qq_list[index]
|
|
group = group_list[index]
|
|
user_qq_list.pop(index)
|
|
impression_list.pop(index)
|
|
group_list.pop(index)
|
|
if user not in users and impression < 100000:
|
|
if user not in data["0"]:
|
|
users.append(user)
|
|
impressions.append(impression)
|
|
groups.append(group)
|
|
else:
|
|
num += 1
|
|
for i in range(num):
|
|
impression = max(impression_list)
|
|
index = impression_list.index(impression)
|
|
user = user_qq_list[index]
|
|
group = group_list[index]
|
|
user_qq_list.pop(index)
|
|
impression_list.pop(index)
|
|
group_list.pop(index)
|
|
if user not in users and impression < 100000:
|
|
users.append(user)
|
|
impressions.append(impression)
|
|
groups.append(group)
|
|
return (await asyncio.gather(*[_pst(users, impressions, groups)]))[0]
|
|
|
|
|
|
async def _pst(users: list, impressions: list, groups: list):
|
|
lens = len(users)
|
|
count = math.ceil(lens / 33)
|
|
width = 10
|
|
idx = 0
|
|
A = CreateImg(1740, 3300, color="#FFE4C4")
|
|
async with aiohttp.ClientSession() as session:
|
|
for _ in range(count):
|
|
col_img = CreateImg(550, 3300, 550, 100, color="#FFE4C4")
|
|
for _ in range(33 if int(lens / 33) >= 1 else lens % 33 - 1):
|
|
idx += 1
|
|
if idx > 100:
|
|
break
|
|
impression = max(impressions)
|
|
index = impressions.index(impression)
|
|
user = users[index]
|
|
group = groups[index]
|
|
impressions.pop(index)
|
|
users.pop(index)
|
|
groups.pop(index)
|
|
try:
|
|
user_name = (
|
|
await GroupInfoUser.get_member_info(user, group)
|
|
).user_name
|
|
except AttributeError:
|
|
user_name = f"我名字呢？"
|
|
user_name = user_name if len(user_name) < 11 else user_name[:10] + "..."
|
|
impression = (
|
|
str(impression)[:4] if len(str(impression)) > 4 else impression
|
|
)
|
|
try:
|
|
async with session.get(
|
|
f"http://q1.qlogo.cn/g?b=qq&nk={user}&s=160", timeout=5
|
|
) as response:
|
|
ava = CreateImg(
|
|
50, 50, background=BytesIO(await response.read())
|
|
)
|
|
except TimeoutError:
|
|
ava = CreateImg(50, 50, color="white")
|
|
ava.circle()
|
|
bk = CreateImg(550, 100, color="#FFE4C4", font_size=30)
|
|
font_w, font_h = bk.getsize(f"{idx}")
|
|
bk.text((5, int((100 - font_h) / 2)), f"{idx}.")
|
|
bk.paste(ava, (55, int((100 - 50) / 2)), True)
|
|
bk.text((120, int((100 - font_h) / 2)), f"{user_name}")
|
|
bk.text((460, int((100 - font_h) / 2)), f"[{impression}]")
|
|
col_img.paste(bk)
|
|
A.paste(col_img, (width, 0))
|
|
lens -= 33
|
|
width += 580
|
|
W = CreateImg(1740, 3700, color="#FFE4C4", font_size=130)
|
|
W.paste(A, (0, 260))
|
|
font_w, font_h = W.getsize(f"{NICKNAME}的好感度总榜")
|
|
W.text((int((1740 - font_w) / 2), int((260 - font_h) / 2)), f"{NICKNAME}的好感度总榜")
|
|
return W.pic2bs4()
|