import random from datetime import datetime, timedelta from io import BytesIO from services.log import logger from services.db_context import db from models.sigin_group_user import SignGroupUser from models.group_member_info import GroupInfoUser from models.bag_user import BagUser from configs.config import MAX_SIGN_GOLD from utils.img_utils import CreateImg import aiohttp from asyncio.exceptions import TimeoutError import math import asyncio async def group_user_check_in(user_qq: int, group: int) -> str: '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.date() == present.date(): return _handle_already_checked_in(user) return await _handle_check_in(user_qq, group, present) # ok def _handle_already_checked_in(user: SignGroupUser) -> str: return f'已经签到过啦~ 好感度：{user.impression:.2f}' async def _handle_check_in(user_qq: int, group: int, present: datetime) -> str: user = await SignGroupUser.ensure(user_qq, group, for_update=True) impression_added = random.random() present = present + timedelta(hours=8) 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 new_impression = user.impression + impression_added message = random.choice(( '谢谢，你是个好人！', '对了，来喝杯茶吗？', )) await user.update( checkin_count=user.checkin_count + 1, checkin_time_last=present, impression=new_impression, add_probability=0, specify_probability=0, ).apply() # glod = await random_glod(user_qq, group, specify_probability) if user.impression < 1: impression = 1 else: impression = user.impression gold = random.randint(1, 100) imgold = random.randint(1, int(impression)) if imgold > MAX_SIGN_GOLD: imgold = MAX_SIGN_GOLD await BagUser.add_gold(user_qq, group, gold + imgold) 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: {new_impression:.2f} (+{impression_added * 2:.2f}).获取金币：{gold+imgold}') return f'{message} 好感度：{new_impression:.2f} (+{impression_added/2:.2f}×2)！！！\n获取金币：{gold} \n好感度额外获得金币：{imgold}' else: logger.info(f'(USER {user.user_qq}, GROUP {user.belonging_group})' f' CHECKED IN successfully. score: {new_impression:.2f} (+{impression_added:.2f}).获取金币：{gold+imgold}') return f'{message} 好感度：{new_impression:.2f} (+{impression_added:.2f})\n获取金币：{gold} \n好感度额外获得金币：{imgold}' async def group_user_check(user_qq: int, group: int) -> str: # 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 '好感度：{:.2f}\n金币：{}\n历史签到数：{}\n上次签到日期：{}'.format( user.impression, gold, user.checkin_count, user.checkin_time_last.date() if user.checkin_time_last != datetime.min else '从未', ) async def group_impression_rank(group: int) -> str: result = "\t好感度排行榜\t\n" user_qq_list, impression_list, _ = await SignGroupUser.query_impression_all(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.select_member_info(user_qq, group)).user_name except Exception as e: 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.query_impression_all(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.select_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('真寻的好感度总榜') W.text((int((1740-font_w)/2), int((260-font_h)/2)), '真寻的好感度总榜') return W.pic2bs4()