From eb02157b94d4b85504cb97e08de0aad77b1996ea Mon Sep 17 00:00:00 2001 From: HibiKier <45528451+HibiKier@users.noreply.github.com> Date: Sun, 27 Feb 2022 17:30:26 +0800 Subject: [PATCH] Add files via upload --- plugins/draw_card/__init__.py | 256 +++++++++++ plugins/draw_card/count_manager.py | 169 ++++++++ plugins/draw_card/handles/azur_handle.py | 317 ++++++++++++++ plugins/draw_card/handles/base_handle.py | 298 +++++++++++++ plugins/draw_card/handles/fgo_handle.py | 221 ++++++++++ plugins/draw_card/handles/genshin_handle.py | 429 +++++++++++++++++++ plugins/draw_card/handles/guardian_handle.py | 400 +++++++++++++++++ plugins/draw_card/handles/onmyoji_handle.py | 179 ++++++++ plugins/draw_card/handles/pcr_handle.py | 147 +++++++ plugins/draw_card/handles/pretty_handle.py | 426 ++++++++++++++++++ plugins/draw_card/handles/prts_handle.py | 292 +++++++++++++ 11 files changed, 3134 insertions(+) create mode 100644 plugins/draw_card/__init__.py create mode 100644 plugins/draw_card/count_manager.py create mode 100644 plugins/draw_card/handles/azur_handle.py create mode 100644 plugins/draw_card/handles/base_handle.py create mode 100644 plugins/draw_card/handles/fgo_handle.py create mode 100644 plugins/draw_card/handles/genshin_handle.py create mode 100644 plugins/draw_card/handles/guardian_handle.py create mode 100644 plugins/draw_card/handles/onmyoji_handle.py create mode 100644 plugins/draw_card/handles/pcr_handle.py create mode 100644 plugins/draw_card/handles/pretty_handle.py create mode 100644 plugins/draw_card/handles/prts_handle.py diff --git a/plugins/draw_card/__init__.py b/plugins/draw_card/__init__.py new file mode 100644 index 00000000..22bae797 --- /dev/null +++ b/plugins/draw_card/__init__.py @@ -0,0 +1,256 @@ +import asyncio +import traceback +from cn2an import cn2an +from dataclasses import dataclass +from typing import Optional, Set, Tuple + +import nonebot +from nonebot import on_regex, on_keyword +from nonebot.adapters.onebot.v11 import MessageEvent +from nonebot.log import logger +from nonebot.matcher import Matcher +from nonebot.params import RegexGroup +from nonebot.permission import SUPERUSER +from nonebot.typing import T_Handler +from nonebot_plugin_apscheduler import scheduler + +from .handles.base_handle import BaseHandle +from .handles.azur_handle import AzurHandle +from .handles.fgo_handle import FgoHandle +from .handles.genshin_handle import GenshinHandle +from .handles.guardian_handle import GuardianHandle +from .handles.onmyoji_handle import OnmyojiHandle +from .handles.pcr_handle import PcrHandle +from .handles.pretty_handle import PrettyHandle +from .handles.prts_handle import PrtsHandle + +from .config import draw_config + + +__zx_plugin_name__ = "游戏抽卡" +__plugin_usage__ = """ +usage： + 模拟赛马娘，原神，明日方舟，坎公骑冠剑，公主连结(国/台)，碧蓝航线，FGO，阴阳师进行抽卡 + 指令： + 原神[1-180]抽: 原神常驻池 + 原神角色[1-180]抽: 原神角色UP池子 + 原神武器[1-180]抽: 原神武器UP池子 + 重置原神抽卡: 清空当前卡池的抽卡次数[即从0开始计算UP概率] + 方舟[1-300]抽: 方舟卡池，当有当期UP时指向UP池 + 赛马娘[1-200]抽: 赛马娘卡池，当有当期UP时指向UP池 + 坎公骑冠剑[1-300]抽: 坎公骑冠剑卡池，当有当期UP时指向UP池 + pcr/公主连接[1-300]抽: 公主连接卡池 + 碧蓝航线/碧蓝[重型/轻型/特型/活动][1-300]抽: 碧蓝航线重型/轻型/特型/活动卡池 + fgo[1-300]抽: fgo卡池 + 阴阳师[1-300]抽: 阴阳师卡池 + * 以上指令可以通过 XX一井 来指定最大抽取数量 * + * 示例：原神一井 * +""".strip() +__plugin_superuser_usage__ = """ +usage： + 卡池方面的更新 + 指令： + 更新方舟信息 + 重载方舟卡池 + 更新原神信息 + 重载原神卡池 + 更新赛马娘信息 + 重载赛马娘卡池 + 更新坎公骑冠剑信息 + 更新碧蓝航线信息 + 更新fgo信息 + 更新阴阳师信息 +""".strip() +__plugin_des__ = "就算是模拟抽卡也不能改变自己是个非酋" +__plugin_cmd__ = [ + "原神[1-180]抽", + "原神角色[1-180]抽", + "原神武器[1-180]抽", + "重置原神抽卡", + "方舟[1-300]抽", + "赛马娘[1-200]抽", + "坎公骑冠剑[1-300]抽", + "pcr/公主连接[1-300]抽", + "fgo[1-300]抽", + "阴阳师[1-300]抽", + "更新方舟信息 [_superuser]", + "重载方舟卡池 [_superuser]", + "更新原神信息 [_superuser]", + "重载原神卡池 [_superuser]", + "更新赛马娘信息 [_superuser]", + "重载赛马娘卡池 [_superuser]", + "更新坎公骑冠剑信息 [_superuser]", + "更新碧蓝航线信息 [_superuser]", + "更新fgo信息 [_superuser]", + "更新阴阳师信息 [_superuser]", +] +__plugin_type__ = ("抽卡相关", 1) +__plugin_version__ = 0.1 +__plugin_author__ = "HibiKier" +__plugin_settings__ = { + "level": 5, + "default_status": True, + "limit_superuser": False, + "cmd": ["游戏抽卡", "抽卡"], +} + + +@dataclass +class Game: + keywords: Set[str] + handle: BaseHandle + flag: bool + max_count: int = 300 # 一次最大抽卡数 + reload_time: Optional[int] = None # 重载UP池时间（小时） + + +games = ( + Game({"azur", "碧蓝", "碧蓝航线"}, AzurHandle(), draw_config.AZUR_FLAG), + Game({"fgo", "命运冠位指定"}, FgoHandle(), draw_config.FGO_FLAG), + Game( + {"genshin", "原神"}, + GenshinHandle(), + draw_config.GENSHIN_FLAG, + max_count=180, + reload_time=18, + ), + Game( + {"guardian", "坎公骑冠剑"}, + GuardianHandle(), + draw_config.GUARDIAN_FLAG, + reload_time=4, + ), + Game({"onmyoji", "阴阳师"}, OnmyojiHandle(), draw_config.ONMYOJI_FLAG), + Game({"pcr", "公主连结", "公主连接", "公主链接", "公主焊接"}, PcrHandle(), draw_config.PCR_FLAG), + Game( + {"pretty", "马娘", "赛马娘"}, + PrettyHandle(), + draw_config.PRETTY_FLAG, + max_count=200, + reload_time=4, + ), + Game({"prts", "方舟", "明日方舟"}, PrtsHandle(), draw_config.PRTS_FLAG, reload_time=4), +) + + +def create_matchers(): + def draw_handler(game: Game) -> T_Handler: + async def handler( + matcher: Matcher, event: MessageEvent, args: Tuple[str, ...] = RegexGroup() + ): + pool_name, num, unit = args + if num == "单": + num = 1 + else: + try: + num = int(cn2an(num, mode="smart")) + except ValueError: + await matcher.finish("必！须！是！数！字！") + if unit == "井": + num *= game.max_count + if num < 1: + await matcher.finish("虚空抽卡？？？") + elif num > game.max_count: + await matcher.finish("一井都满不足不了你嘛！快爬开！") + pool_name = ( + pool_name.replace("池", "") + .replace("武器", "arms") + .replace("角色", "char") + .replace("卡牌", "card") + .replace("卡", "card") + ) + try: + res = await game.handle.draw(num, pool_name=pool_name, user_id=event.user_id) + except: + logger.warning(traceback.format_exc()) + await matcher.finish("出错了...") + await matcher.finish(res, at_sender=True) + + return handler + + def update_handler(game: Game) -> T_Handler: + async def handler(matcher: Matcher): + await game.handle.update_info() + await matcher.finish("更新完成！") + + return handler + + def reload_handler(game: Game) -> T_Handler: + async def handler(matcher: Matcher): + res = await game.handle.reload_pool() + if res: + await matcher.finish(res) + + return handler + + def reset_handler(game: Game) -> T_Handler: + async def handler(matcher: Matcher, event: MessageEvent): + if game.handle.reset_count(event.user_id): + await matcher.finish("重置成功！") + + return handler + + def scheduled_job(game: Game) -> T_Handler: + async def handler(): + await game.handle.reload_pool() + + return handler + + for game in games: + pool_pattern = r"([^\s单0-9零一二三四五六七八九百十]{0,3})" + num_pattern = r"(单|[0-9零一二三四五六七八九百十]{1,3})" + unit_pattern = r"([抽|井|连])" + draw_regex = r".*?(?:{})\s*{}\s*{}\s*{}".format( + "|".join(game.keywords), pool_pattern, num_pattern, unit_pattern + ) + update_keywords = {f"更新{keyword}信息" for keyword in game.keywords} + reload_keywords = {f"重载{keyword}卡池" for keyword in game.keywords} + reset_keywords = {f"重置{keyword}抽卡" for keyword in game.keywords} + if game.flag: + on_regex(draw_regex, priority=5, block=True).append_handler( + draw_handler(game) + ) + on_keyword( + update_keywords, permission=SUPERUSER, priority=1, block=True + ).append_handler(update_handler(game)) + on_keyword(reload_keywords, priority=1, block=True).append_handler( + reload_handler(game) + ) + on_keyword(reset_keywords, priority=1, block=True).append_handler( + reset_handler(game) + ) + if game.reload_time: + scheduler.add_job( + scheduled_job(game), trigger="cron", hour=game.reload_time, minute=1 + ) + + +create_matchers() + + +# 更新资源 +@scheduler.scheduled_job( + "cron", + hour=4, + minute=1, +) +async def _(): + tasks = [] + for game in games: + if game.flag: + tasks.append(asyncio.ensure_future(game.handle.update_info())) + await asyncio.gather(*tasks) + + +driver = nonebot.get_driver() + + +@driver.on_startup +async def _(): + tasks = [] + for game in games: + if game.flag: + game.handle.init_data() + if not game.handle.data_exists(): + tasks.append(asyncio.ensure_future(game.handle.update_info())) + await asyncio.gather(*tasks) diff --git a/plugins/draw_card/count_manager.py b/plugins/draw_card/count_manager.py new file mode 100644 index 00000000..58acbaf3 --- /dev/null +++ b/plugins/draw_card/count_manager.py @@ -0,0 +1,169 @@ +from typing import Optional +from pydantic import BaseModel +import time + + +class BaseUserCount(BaseModel): + count: int = 1 # 当前抽卡次数 + time_: int = time.time() # 抽卡时间，当超过一定时间时将重置抽卡次数 + timeout: int = 60 # 超时时间60秒 + + +class DrawCountManager: + """ + 抽卡统计保底 + """ + + def __init__( + self, game_draw_count_rule: tuple, star2name: tuple, max_draw_count: int + ): + """ + 初始化保底统计 + + 例如：DrawCountManager((10, 90, 180), ("4", "5", "5")) + + 抽卡保底需要的次数和返回的对应名称，例如星级等 + + :param game_draw_count_rule：抽卡规则 + :param star2name：星级对应的名称 + :param max_draw_count：最大累计抽卡次数，当下次单次抽卡超过该次数时将会清空数据 + + """ + # 只有保底 + self._data = {} + self._guarantee_tuple = game_draw_count_rule + self._star2name = star2name + self._max_draw_count = max_draw_count + + def increase(self, key: int, value: int = 1): + """ + 用户抽卡次数加1 + """ + if self._data.get(key) is None: + self._data[key] = BaseUserCount() + else: + self._data[key].count += 1 + + def get_max_guarantee(self): + """ + 获取最大保底抽卡次数 + """ + return self._guarantee_tuple[-1] + + def get_user_count(self, key: int) -> int: + """ + 获取当前抽卡次数 + """ + return self._data[key].count + + def update_time(self, key: int): + """ + 更新抽卡时间 + """ + self._data[key].time_ = time.time() + + def reset(self, key: int): + """ + 清空记录 + """ + del self._data[key] + + +class GenshinCountManager(DrawCountManager): + class UserCount(BaseUserCount): + five_index: int = 0 # 获取五星时的抽卡次数 + four_index: int = 0 # 获取四星时的抽卡次数 + is_up: bool = False # 下次五星是否必定为up + + def increase(self, key: int, value: int = 1): + """ + 用户抽卡次数加1 + """ + if self._data.get(key) is None: + self._data[key] = self.UserCount() + else: + self._data[key].count += 1 + + def set_is_up(self, key: int, value: bool): + """ + 设置下次是否必定up + """ + self._data[key].is_up = value + + def is_up(self, key: int) -> bool: + """ + 判断该次保底是否必定为up + """ + return self._data[key].is_up + + def get_user_five_index(self, key: int) -> int: + """ + 获取用户上次获取五星的次数 + """ + return self._data[key].five_index + + def get_user_four_index(self, key: int) -> int: + """ + 获取用户上次获取四星的次数 + """ + return self._data[key].four_index + + def mark_five_index(self, key: int): + """ + 标记用户该次次数为五星 + """ + self._data[key].five_index = self._data[key].count + + def mark_four_index(self, key: int): + """ + 标记用户该次次数为四星 + """ + self._data[key].four_index = self._data[key].count + + def check_timeout(self, key: int): + """ + 检查用户距离上次抽卡是否超时 + """ + if key in self._data.keys() and self._is_timeout(key): + del self._data[key] + + def check_count(self, key: int, count: int): + """ + 检查用户该次抽卡次数累计是否超过最大限制次数 + """ + if ( + key in self._data.keys() + and self._data[key].count + count > self._max_draw_count + ): + del self._data[key] + + def _is_timeout(self, key: int) -> bool: + return time.time() - self._data[key].time_ > self._data[key].timeout + + def get_user_guarantee_count(self, key: int) -> int: + return ( + self.get_max_guarantee() + - ( + ( + self._data[key].count % self.get_max_guarantee() + if self._data[key].count > 0 + else 0 + ) + - self._data[key].five_index + ) + ) % self.get_max_guarantee() or self.get_max_guarantee() + + def check(self, key: int) -> Optional[int]: + """ + 是否保底 + """ + # print(self._data) + user: GenshinCountManager.UserCount = self._data[key] + if user.count - user.five_index == 90: + user.five_index = user.count + return 5 + if user.count - user.four_index == 10: + user.four_index = user.count + return 4 + return None + diff --git a/plugins/draw_card/handles/azur_handle.py b/plugins/draw_card/handles/azur_handle.py new file mode 100644 index 00000000..9434077b --- /dev/null +++ b/plugins/draw_card/handles/azur_handle.py @@ -0,0 +1,317 @@ +import random +import dateparser +from lxml import etree +from typing import List, Optional +from urllib.parse import unquote +from pydantic import ValidationError +from nonebot.log import logger +from nonebot.adapters.onebot.v11 import Message +from utils.message_builder import image + +from .base_handle import BaseHandle, BaseData, UpEvent as _UpEvent, UpChar as _UpChar +from ..config import draw_config +from ..util import remove_prohibited_str, cn2py +from utils.image_utils import BuildImage +import asyncio + +try: + import ujson as json +except ModuleNotFoundError: + import json + + +class AzurChar(BaseData): + type_: str # 舰娘类型 + + @property + def star_str(self) -> str: + return ["白", "蓝", "紫", "金"][self.star - 1] + + +class UpChar(_UpChar): + type_: str # 舰娘类型 + + +class UpEvent(_UpEvent): + up_char: List[UpChar] # up对象 + + +class AzurHandle(BaseHandle[AzurChar]): + def __init__(self): + super().__init__("azur", "碧蓝航线") + self.max_star = 4 + self.config = draw_config.azur + self.ALL_CHAR: List[AzurChar] = [] + self.UP_EVENT: Optional[UpEvent] = None + + def get_card(self, pool_name: str, **kwargs) -> AzurChar: + if pool_name == "轻型": + type_ = ["驱逐", "轻巡", "维修"] + elif pool_name == "重型": + type_ = ["重巡", "战列", "战巡", "重炮"] + else: + type_ = ["维修", "潜艇", "重巡", "轻航", "航母"] + up_pool_flag = pool_name == "活动" + # Up + up_ship = [x for x in self.UP_EVENT.up_char if x.zoom > 0] + # print(up_ship) + acquire_char = None + if up_ship and up_pool_flag: + up_zoom = [(0, up_ship[0].zoom / 100)] + # 初始化概率 + cur_ = up_ship[0].zoom / 100 + for i in range(len(up_ship)): + try: + up_zoom.append((cur_, cur_ + up_ship[i+1].zoom / 100)) + cur_ += up_ship[i+1].zoom / 100 + except IndexError: + pass + rand = random.random() + # 抽取up + for i, zoom in enumerate(up_zoom): + if zoom[0] <= rand <= zoom[1]: + try: + acquire_char = [x for x in self.ALL_CHAR if x.name == up_ship[i].name][0] + except IndexError: + pass + # 没有up或者未抽取到up + if not acquire_char: + star = self.get_star( + [4, 3, 2, 1], + [ + self.config.AZUR_FOUR_P, + self.config.AZUR_THREE_P, + self.config.AZUR_TWO_P, + self.config.AZUR_ONE_P, + ], + ) + acquire_char = random.choice([ + x + for x in self.ALL_CHAR + if x.star == star and x.type_ in type_ and not x.limited + ]) + return acquire_char + + # async def draw(self, count: int, **kwargs) -> Message: + # return await asyncio.get_event_loop().run_in_executor(None, self._draw, count) + + async def draw(self, count: int, **kwargs) -> Message: + index2card = self.get_cards(count, **kwargs) + cards = [card[0] for card in index2card] + up_list = [x.name for x in self.UP_EVENT.up_char] if self.UP_EVENT.up_char else [] + result = self.format_result(index2card, **{**kwargs, "up_list": up_list}) + return image(b64=self.generate_img(cards).pic2bs4()) + result + + def generate_card_img(self, card: AzurChar) -> BuildImage: + sep_w = 5 + sep_t = 5 + sep_b = 20 + w = 100 + h = 100 + bg = BuildImage(w + sep_w * 2, h + sep_t + sep_b, font="msyh.ttf") + frame_path = self.img_path / f"{card.star}_star.png" + frame = BuildImage(w, h, background=frame_path) + img_path = self.img_path / f"{cn2py(card.name)}.png" + img = BuildImage(w, h, background=img_path) + # 加圆角 + img.circle_corner(6) + bg.paste(img, (sep_w, sep_t), alpha=True) + bg.paste(frame, (sep_w, sep_t), alpha=True) + bg.circle_corner(6) + # 加名字 + text = card.name[:6] + "..." if len(card.name) > 7 else card.name + text_w, text_h = bg.getsize(text) + bg.text( + (sep_w + (w - text_w) / 2, h + sep_t + (sep_b - text_h) / 2), + text, + fill=["#808080", "#3b8bff", "#8000ff", "#c90", "#ee494c"][card.star - 1], + ) + return bg + + def _init_data(self): + self.ALL_CHAR = [ + AzurChar( + name=value["名称"], + star=int(value["星级"]), + limited="可以建造" not in value["获取途径"], + type_=value["类型"], + ) + for value in self.load_data().values() + ] + self.load_up_char() + + def load_up_char(self): + try: + data = self.load_data(f"draw_card_up/{self.game_name}_up_char.json") + self.UP_EVENT = UpEvent.parse_obj(data.get("char", {})) + except ValidationError: + logger.warning(f"{self.game_name}_up_char 解析出错") + + def dump_up_char(self): + if self.UP_EVENT: + data = {"char": json.loads(self.UP_EVENT.json())} + self.dump_data(data, f"draw_card_up/{self.game_name}_up_char.json") + + async def _update_info(self): + info = {} + # 更新图鉴 + url = "https://wiki.biligame.com/blhx/舰娘图鉴" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + contents = dom.xpath( + "//div[@class='resp-tabs-container']/div[@class='resp-tab-content']" + ) + for index, content in enumerate(contents): + char_list = content.xpath("./table/tbody/tr[2]/td/div/div/div/div") + for char in char_list: + try: + name = char.xpath("./a/@title")[0] + frame = char.xpath("./div/a/img/@alt")[0] + avatar = char.xpath("./a/img/@srcset")[0] + except IndexError: + continue + member_dict = { + "名称": remove_prohibited_str(name), + "头像": unquote(str(avatar).split(" ")[-2]), + "星级": self.parse_star(frame), + "类型": self.parse_type(index), + } + info[member_dict["名称"]] = member_dict + # 更新额外信息 + for key in info.keys(): + url = f"https://wiki.biligame.com/blhx/{key}" + result = await self.get_url(url) + if not result: + info[key]["获取途径"] = [] + logger.warning(f"{self.game_name_cn} 获取额外信息错误 {key}") + continue + try: + dom = etree.HTML(result, etree.HTMLParser()) + time = dom.xpath( + "//table[@class='wikitable sv-general']/tbody[1]/tr[4]/td[2]//text()" + )[0] + sources = [] + if "无法建造" in time: + sources.append("无法建造") + elif "活动已关闭" in time: + sources.append("活动限定") + else: + sources.append("可以建造") + info[key]["获取途径"] = sources + except IndexError: + info[key]["获取途径"] = [] + logger.warning(f"{self.game_name_cn} 获取额外信息错误 {key}") + self.dump_data(info) + logger.info(f"{self.game_name_cn} 更新成功") + # 下载头像 + for value in info.values(): + await self.download_img(value["头像"], value["名称"]) + # 下载头像框 + idx = 1 + BLHX_URL = "https://patchwiki.biligame.com/images/blhx" + for url in [ + "/1/15/pxho13xsnkyb546tftvh49etzdh74cf.png", + "/a/a9/k8t7nx6c8pan5vyr8z21txp45jxeo66.png", + "/a/a5/5whkzvt200zwhhx0h0iz9qo1kldnidj.png", + "/a/a2/ptog1j220x5q02hytpwc8al7f229qk9.png", + "/6/6d/qqv5oy3xs40d3055cco6bsm0j4k4gzk.png", + ]: + await self.download_img(BLHX_URL + url, f"{idx}_star") + idx += 1 + await self.update_up_char() + + @staticmethod + def parse_star(star: str) -> int: + if star in ["舰娘头像外框普通.png", "舰娘头像外框白色.png"]: + return 1 + elif star in ["舰娘头像外框稀有.png", "舰娘头像外框蓝色.png"]: + return 2 + elif star in ["舰娘头像外框精锐.png", "舰娘头像外框紫色.png"]: + return 3 + elif star in ["舰娘头像外框超稀有.png", "舰娘头像外框金色.png"]: + return 4 + elif star in ["舰娘头像外框海上传奇.png", "舰娘头像外框彩色.png"]: + return 5 + elif star in [ + "舰娘头像外框最高方案.png", + "舰娘头像外框决战方案.png", + "舰娘头像外框超稀有META.png", + "舰娘头像外框精锐META.png", + ]: + return 6 + else: + return 6 + + @staticmethod + def parse_type(index: int) -> str: + azur_types = [ + "驱逐", + "轻巡", + "重巡", + "超巡", + "战巡", + "战列", + "航母", + "航站", + "轻航", + "重炮", + "维修", + "潜艇", + "运输", + ] + try: + return azur_types[index] + except IndexError: + return azur_types[0] + + async def update_up_char(self): + url = "https://wiki.biligame.com/blhx/游戏活动表" + result = await self.get_url(url) + if not result: + logger.warning(f"{self.game_name_cn}获取活动表出错") + return + try: + dom = etree.HTML(result, etree.HTMLParser()) + dd = dom.xpath("//div[@class='timeline2']/dl/dd/a")[0] + url = "https://wiki.biligame.com" + dd.xpath("./@href")[0] + title = dd.xpath("string(.)") + result = await self.get_url(url) + if not result: + logger.warning(f"{self.game_name_cn}获取活动页面出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + timer = dom.xpath("//span[@class='eventTimer']")[0] + start_time = dateparser.parse(timer.xpath("./@data-start")[0]) + end_time = dateparser.parse(timer.xpath("./@data-end")[0]) + ships = dom.xpath("//table[@class='shipinfo']") + up_chars = [] + for ship in ships: + name = ship.xpath("./tbody/tr/td[2]/p/a/@title")[0] + type_ = ship.xpath("./tbody/tr/td[2]/p/small/text()")[0] # 舰船类型 + try: + p = float(str(ship.xpath(".//sup/text()")[0]).strip("%")) + except IndexError: + p = 0 + star = self.parse_star( + ship.xpath("./tbody/tr/td[1]/div/div/div/a/img/@alt")[0] + ) + up_chars.append(UpChar(name=name, star=star, limited=False, zoom=p, type_=type_)) + self.UP_EVENT = UpEvent( + title=title, + pool_img="", + start_time=start_time, + end_time=end_time, + up_char=up_chars, + ) + self.dump_up_char() + except Exception as e: + logger.warning(f"{self.game_name_cn}UP更新出错 {type(e)}：{e}") + + async def _reload_pool(self) -> Optional[Message]: + await self.update_up_char() + self.load_up_char() + if self.UP_EVENT: + return Message(f"重载成功！\n当前活动：{self.UP_EVENT.title}") diff --git a/plugins/draw_card/handles/base_handle.py b/plugins/draw_card/handles/base_handle.py new file mode 100644 index 00000000..9e019884 --- /dev/null +++ b/plugins/draw_card/handles/base_handle.py @@ -0,0 +1,298 @@ +import math +import random +import aiohttp +import asyncio +import aiofiles +from PIL import Image +from datetime import datetime +from pydantic import BaseModel, Extra +from asyncio.exceptions import TimeoutError +from typing import Dict, List, Optional, TypeVar, Generic, Tuple +from nonebot.adapters.onebot.v11 import Message +from configs.path_config import IMAGE_PATH +from utils.message_builder import image +from nonebot.log import logger +import asyncio + +try: + import ujson as json +except ModuleNotFoundError: + import json + +from utils.image_utils import BuildImage +from ..config import DRAW_PATH, draw_config +from ..util import cn2py, circled_number + + +class BaseData(BaseModel, extra=Extra.ignore): + name: str # 名字 + star: int # 星级 + limited: bool # 限定 + + def __eq__(self, other: "BaseData"): + return self.name == other.name + + def __hash__(self): + return hash(self.name) + + @property + def star_str(self) -> str: + return "".join(["★" for _ in range(self.star)]) + + +class UpChar(BaseData): + zoom: float # up提升倍率 + + +class UpEvent(BaseModel): + title: str # up池标题 + pool_img: str # up池封面 + start_time: Optional[datetime] # 开始时间 + end_time: Optional[datetime] # 结束时间 + up_char: List[UpChar] # up对象 + + +TC = TypeVar("TC", bound="BaseData") + + +class BaseHandle(Generic[TC]): + def __init__(self, game_name: str, game_name_cn: str, game_card_color: str = "#ffffff"): + self.game_name = game_name + self.game_name_cn = game_name_cn + self.max_star = 1 # 最大星级 + self.data_path = DRAW_PATH + self.img_path = IMAGE_PATH / f"draw_card/{self.game_name}" + self.up_path = DRAW_PATH / "draw_card_up" + self.img_path.mkdir(parents=True, exist_ok=True) + self.up_path.mkdir(parents=True, exist_ok=True) + self.data_files: List[str] = [f"{self.game_name}.json"] + self.game_card_color: str = game_card_color + + 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, **kwargs) + cards = [card[0] for card in index2card] + result = self.format_result(index2card) + return image(b64=self.generate_img(cards).pic2bs4()) + result + + # 抽取卡池 + def get_card(self, **kwargs) -> TC: + raise NotImplementedError + + def get_cards(self, count: int, **kwargs) -> List[Tuple[TC, int]]: + return [(self.get_card(**kwargs), i) for i in range(count)] + + # 获取星级 + @staticmethod + def get_star(star_list: List[int], probability_list: List[float]) -> int: + return random.choices(star_list, weights=probability_list, k=1)[0] + + def format_result(self, index2card: List[Tuple[TC, int]], **kwargs) -> str: + card_list = [card[0] for card in index2card] + results = [ + self.format_star_result(card_list, **kwargs), + self.format_max_star(index2card, **kwargs), + self.format_max_card(card_list, **kwargs), + ] + results = [rst for rst in results if rst] + return "\n".join(results) + + def format_star_result(self, card_list: List[TC], **kwargs) -> str: + star_dict: Dict[str, int] = {} # 记录星级及其次数 + + card_list_sorted = sorted(card_list, key=lambda c: c.star, reverse=True) + for card in card_list_sorted: + try: + star_dict[card.star_str] += 1 + except KeyError: + star_dict[card.star_str] = 1 + + rst = "" + for star_str, count in star_dict.items(): + rst += f"[{star_str}×{count}] " + return rst.strip() + + def format_max_star( + self, card_list: List[Tuple[TC, int]], up_list: List[str] = [], **kwargs + ) -> str: + up_list = up_list or kwargs.get("up_list", []) + rst = "" + for card, index in card_list: + if card.star == self.max_star: + if card.name in up_list: + rst += f"第 {index} 抽获取UP {card.name}\n" + else: + rst += f"第 {index} 抽获取 {card.name}\n" + return rst.strip() + + def format_max_card(self, card_list: List[TC], **kwargs) -> str: + card_dict: Dict[TC, int] = {} # 记录卡牌抽取次数 + + for card in card_list: + try: + card_dict[card] += 1 + except KeyError: + card_dict[card] = 1 + + max_count = max(card_dict.values()) + max_card = list(card_dict.keys())[list(card_dict.values()).index(max_count)] + if max_count <= 1: + return "" + return f"抽取到最多的是{max_card.name}，共抽取了{max_count}次" + + def generate_img( + self, + cards: List[TC], + num_per_line: int = 5, + max_per_line: Tuple[int, int] = (40, 10), + ) -> BuildImage: + """ + 生成统计图片 + :param cards: 卡牌列表 + :param num_per_line: 单行角色显示数量 + :param max_per_line: 当card_list超过一定数值时，更改单行数量 + """ + if len(cards) > max_per_line[0]: + num_per_line = max_per_line[1] + if len(cards) > 90: + card_dict: Dict[TC, int] = {} # 记录卡牌抽取次数 + for card in cards: + try: + card_dict[card] += 1 + except KeyError: + card_dict[card] = 1 + card_list = list(card_dict) + num_list = list(card_dict.values()) + else: + card_list = cards + num_list = [1] * len(cards) + + card_imgs: List[BuildImage] = [] + for card, num in zip(card_list, num_list): + card_img = self.generate_card_img(card) + + # 数量 > 1 时加数字上标 + if num > 1: + label = circled_number(num) + label_w = int(min(card_img.w, card_img.h) / 7) + label = label.resize( + ( + int(label_w * label.width / label.height), + label_w, + ), + Image.ANTIALIAS, + ) + card_img.paste(label, alpha=True) + + card_imgs.append(card_img) + + img_w = card_imgs[0].w + img_h = card_imgs[0].h + if len(card_imgs) < num_per_line: + w = img_w * len(card_imgs) + else: + w = img_w * num_per_line + h = img_h * math.ceil(len(card_imgs) / num_per_line) + img = BuildImage(w, h, img_w, img_h, color=self.game_card_color) + for card_img in card_imgs: + img.paste(card_img) + return img + + def generate_card_img(self, card: TC) -> BuildImage: + img = str(self.img_path / f"{cn2py(card.name)}.png") + return BuildImage(100, 100, background=img) + + def load_data(self, filename: str = "") -> dict: + if not filename: + filename = f"{self.game_name}.json" + filepath = self.data_path / filename + if not filepath.exists(): + return {} + with filepath.open("r", encoding="utf8") as f: + return json.load(f) + + def dump_data(self, data: dict, filename: str = ""): + if not filename: + filename = f"{self.game_name}.json" + filepath = self.data_path / filename + with filepath.open("w", encoding="utf8") as f: + json.dump(data, f, ensure_ascii=False, indent=4) + + def data_exists(self) -> bool: + for file in self.data_files: + if not (self.data_path / file).exists(): + return False + return True + + def _init_data(self): + raise NotImplementedError + + def init_data(self): + try: + self._init_data() + except Exception as e: + logger.warning(f"{self.game_name_cn} 导入角色数据错误：{type(e)}：{e}") + + async def _update_info(self): + raise NotImplementedError + + def client(self) -> aiohttp.ClientSession: + headers = { + "User-Agent": '"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; TencentTraveler 4.0)"' + } + return aiohttp.ClientSession(headers=headers) + + async def update_info(self): + try: + async with asyncio.Semaphore(draw_config.SEMAPHORE): + async with self.client() as session: + self.session = session + await self._update_info() + except Exception as e: + logger.warning(f"{self.game_name_cn} 更新数据错误：{type(e)}：{e}") + self.init_data() + + async def get_url(self, url: str) -> str: + result = "" + retry = 5 + for i in range(retry): + try: + async with self.session.get(url, timeout=10) as response: + result = await response.text() + break + except TimeoutError: + logger.warning(f"访问 {url} 超时, 重试 {i + 1}/{retry}") + await asyncio.sleep(1) + return result + + async def download_img(self, url: str, name: str) -> bool: + img_path = self.img_path / f"{cn2py(name)}.png" + if img_path.exists(): + return True + try: + async with self.session.get(url, timeout=10) as response: + async with aiofiles.open(str(img_path), "wb") as f: + await f.write(await response.read()) + return True + except TimeoutError: + logger.warning(f"下载 {self.game_name_cn} 图片超时，名称：{name}，url：{url}") + return False + except: + logger.warning(f"下载 {self.game_name_cn} 链接错误，名称：{name}，url：{url}") + return False + + async def _reload_pool(self) -> Optional[Message]: + return None + + async def reload_pool(self) -> Optional[Message]: + try: + async with self.client() as session: + self.session = session + return await self._reload_pool() + except Exception as e: + logger.warning(f"{self.game_name_cn} 重载UP池错误：{type(e)}：{e}") + + def reset_count(self, user_id: int) -> bool: + return False diff --git a/plugins/draw_card/handles/fgo_handle.py b/plugins/draw_card/handles/fgo_handle.py new file mode 100644 index 00000000..a491b906 --- /dev/null +++ b/plugins/draw_card/handles/fgo_handle.py @@ -0,0 +1,221 @@ +import random +from lxml import etree +from typing import List, Tuple +from PIL import ImageDraw +from nonebot.log import logger + +try: + import ujson as json +except ModuleNotFoundError: + import json + +from .base_handle import BaseHandle, BaseData +from ..config import draw_config +from ..util import remove_prohibited_str, cn2py, load_font +from utils.image_utils import BuildImage + + +class FgoData(BaseData): + pass + + +class FgoChar(FgoData): + pass + + +class FgoCard(FgoData): + pass + + +class FgoHandle(BaseHandle[FgoData]): + def __init__(self): + super().__init__("fgo", "命运-冠位指定") + self.data_files.append("fgo_card.json") + self.max_star = 5 + self.config = draw_config.fgo + self.ALL_CHAR: List[FgoChar] = [] + self.ALL_CARD: List[FgoCard] = [] + + def get_card(self, mode: int = 1) -> FgoData: + if mode == 1: + star = self.get_star( + [8, 7, 6, 5, 4, 3], + [ + self.config.FGO_SERVANT_FIVE_P, + self.config.FGO_SERVANT_FOUR_P, + self.config.FGO_SERVANT_THREE_P, + self.config.FGO_CARD_FIVE_P, + self.config.FGO_CARD_FOUR_P, + self.config.FGO_CARD_THREE_P, + ], + ) + elif mode == 2: + star = self.get_star( + [5, 4], [self.config.FGO_CARD_FIVE_P, self.config.FGO_CARD_FOUR_P] + ) + else: + star = self.get_star( + [8, 7, 6], + [ + self.config.FGO_SERVANT_FIVE_P, + self.config.FGO_SERVANT_FOUR_P, + self.config.FGO_SERVANT_THREE_P, + ], + ) + if star > 5: + star -= 3 + chars = [x for x in self.ALL_CHAR if x.star == star and not x.limited] + else: + chars = [x for x in self.ALL_CARD if x.star == star and not x.limited] + return random.choice(chars) + + def get_cards(self, count: int, **kwargs) -> List[Tuple[FgoData, int]]: + card_list = [] # 获取所有角色 + servant_count = 0 # 保底计算 + card_count = 0 # 保底计算 + for i in range(count): + servant_count += 1 + card_count += 1 + if card_count == 9: # 四星卡片保底 + mode = 2 + elif servant_count == 10: # 三星从者保底 + mode = 3 + else: # 普通抽 + mode = 1 + card = self.get_card(mode) + if isinstance(card, FgoCard) and card.star > self.max_star - 2: + card_count = 0 + if isinstance(card, FgoChar): + servant_count = 0 + card_list.append((card, i + 1)) + return card_list + + def generate_card_img(self, card: FgoData) -> BuildImage: + sep_w = 5 + sep_t = 5 + sep_b = 20 + w = 128 + h = 140 + bg = BuildImage(w + sep_w * 2, h + sep_t + sep_b) + img_path = str(self.img_path / f"{cn2py(card.name)}.png") + img = BuildImage(w, h, background=img_path) + bg.paste(img, (sep_w, sep_t), alpha=True) + # 加名字 + text = card.name[:6] + "..." if len(card.name) > 7 else card.name + font = load_font(fontsize=16) + text_w, text_h = font.getsize(text) + draw = ImageDraw.Draw(bg.markImg) + draw.text( + (sep_w + (w - text_w) / 2, h + sep_t + (sep_b - text_h) / 2), + text, + font=font, + fill="gray", + ) + return bg + + def _init_data(self): + self.ALL_CHAR = [ + FgoChar( + name=value["名称"], + star=int(value["星级"]), + limited=True + if not ("圣晶石召唤" in value["入手方式"] or "圣晶石召唤（Story卡池）" in value["入手方式"]) + else False, + ) + for value in self.load_data().values() + ] + self.ALL_CARD = [ + FgoCard(name=value["名称"], star=int(value["星级"]), limited=False) + for value in self.load_data("fgo_card.json").values() + ] + + async def _update_info(self): + # fgo.json + fgo_info = {} + for i in range(500): + url = f"http://fgo.vgtime.com/servant/ajax?card=&wd=&ids=&sort=12777&o=desc&pn={i}" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} page {i} 出错") + continue + fgo_data = json.loads(result) + if int(fgo_data["nums"]) <= 0: + break + for x in fgo_data["data"]: + name = remove_prohibited_str(x["name"]) + member_dict = { + "id": x["id"], + "card_id": x["charid"], + "头像": x["icon"], + "名称": remove_prohibited_str(x["name"]), + "职阶": x["classes"], + "星级": int(x["star"]), + "hp": x["lvmax4hp"], + "atk": x["lvmax4atk"], + "card_quick": x["cardquick"], + "card_arts": x["cardarts"], + "card_buster": x["cardbuster"], + "宝具": x["tprop"], + } + fgo_info[name] = member_dict + # 更新额外信息 + for key in fgo_info.keys(): + url = f'http://fgo.vgtime.com/servant/{fgo_info[key]["id"]}' + result = await self.get_url(url) + if not result: + fgo_info[key]["入手方式"] = ["圣晶石召唤"] + logger.warning(f"{self.game_name_cn} 获取额外信息错误 {key}") + continue + try: + dom = etree.HTML(result, etree.HTMLParser()) + obtain = dom.xpath( + "//table[contains(string(.),'入手方式')]/tr[8]/td[3]/text()" + )[0] + obtain = str(obtain).strip() + if "限时活动免费获取 活动结束后无法获得" in obtain: + obtain = ["活动获取"] + elif "非限时UP无法获得" in obtain: + obtain = ["限时召唤"] + else: + if "&" in obtain: + obtain = obtain.split("&") + else: + obtain = obtain.split(" ") + obtain = [s.strip() for s in obtain if s.strip()] + fgo_info[key]["入手方式"] = obtain + except IndexError: + fgo_info[key]["入手方式"] = ["圣晶石召唤"] + logger.warning(f"{self.game_name_cn} 获取额外信息错误 {key}") + self.dump_data(fgo_info) + logger.info(f"{self.game_name_cn} 更新成功") + # fgo_card.json + fgo_card_info = {} + for i in range(500): + url = f"http://fgo.vgtime.com/equipment/ajax?wd=&ids=&sort=12958&o=desc&pn={i}" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn}卡牌 page {i} 出错") + continue + fgo_data = json.loads(result) + if int(fgo_data["nums"]) <= 0: + break + for x in fgo_data["data"]: + name = remove_prohibited_str(x["name"]) + member_dict = { + "id": x["id"], + "card_id": x["equipid"], + "头像": x["icon"], + "名称": name, + "星级": int(x["star"]), + "hp": x["lvmax_hp"], + "atk": x["lvmax_atk"], + "skill_e": str(x["skill_e"]).split("
")[:-1], + } + fgo_card_info[name] = member_dict + self.dump_data(fgo_card_info, "fgo_card.json") + logger.info(f"{self.game_name_cn} 卡牌更新成功") + # 下载头像 + for value in fgo_info.values(): + await self.download_img(value["头像"], value["名称"]) + for value in fgo_card_info.values(): + await self.download_img(value["头像"], value["名称"]) diff --git a/plugins/draw_card/handles/genshin_handle.py b/plugins/draw_card/handles/genshin_handle.py new file mode 100644 index 00000000..94016527 --- /dev/null +++ b/plugins/draw_card/handles/genshin_handle.py @@ -0,0 +1,429 @@ +import random +import dateparser +from lxml import etree +from PIL import Image, ImageDraw +from urllib.parse import unquote +from typing import List, Optional, Tuple +from pydantic import ValidationError +from datetime import datetime, timedelta +from nonebot.adapters.onebot.v11 import Message +from utils.message_builder import image +from nonebot.log import logger +import asyncio + +try: + import ujson as json +except ModuleNotFoundError: + import json + +from .base_handle import BaseHandle, BaseData, UpChar, UpEvent +from ..config import draw_config +from ..count_manager import GenshinCountManager +from ..util import remove_prohibited_str, cn2py, load_font +from utils.image_utils import BuildImage + + +class GenshinData(BaseData): + pass + + +class GenshinChar(GenshinData): + pass + + +class GenshinArms(GenshinData): + pass + + +class GenshinHandle(BaseHandle[GenshinData]): + def __init__(self): + super().__init__("genshin", "原神", "#ebebeb") + self.data_files.append("genshin_arms.json") + self.max_star = 5 + self.config = draw_config.genshin + + self.ALL_CHAR: List[GenshinData] = [] + self.ALL_ARMS: List[GenshinData] = [] + self.UP_CHAR: Optional[UpEvent] = None + self.UP_ARMS: Optional[UpEvent] = None + + self.count_manager = GenshinCountManager((10, 90), ("4", "5"), 180) + + # 抽取卡池 + def get_card( + self, pool_name: str, mode: int = 1, add: float = 0.0, is_up: bool = False + ): + """ + mode 1：普通抽 2：四星保底 3：五星保底 + """ + if mode == 1: + star = self.get_star( + [5, 4, 3], + [ + self.config.GENSHIN_FIVE_P + add, + self.config.GENSHIN_FOUR_P, + self.config.GENSHIN_THREE_P, + ], + ) + elif mode == 2: + star = self.get_star( + [5, 4], + [self.config.GENSHIN_G_FIVE_P + add, self.config.GENSHIN_G_FOUR_P], + ) + else: + star = 5 + + if pool_name == "char": + up_event = self.UP_CHAR + all_list = self.ALL_CHAR + [ + x for x in self.ALL_ARMS if x.star == star and x.star < 5 + ] + elif pool_name == "arms": + up_event = self.UP_ARMS + all_list = self.ALL_ARMS + [ + x for x in self.ALL_CHAR if x.star == star and x.star < 5 + ] + else: + up_event = None + all_list = self.ALL_ARMS + self.ALL_CHAR + + acquire_char = None + # 是否UP + if up_event and star > 3: + # 获取up角色列表 + up_list = [x.name for x in up_event.up_char if x.star == star] + # 成功获取up角色 + if random.random() < 0.5 or is_up: + up_name = random.choice(up_list) + try: + acquire_char = [x for x in all_list if x.name == up_name][0] + except IndexError: + pass + if not acquire_char: + chars = [x for x in all_list if x.star == star and not x.limited] + acquire_char = random.choice(chars) + return acquire_char + + def get_cards( + self, count: int, user_id: int, pool_name: str + ) -> List[Tuple[GenshinData, int]]: + card_list = [] # 获取角色列表 + add = 0.0 + count_manager = self.count_manager + count_manager.check_timeout(user_id) # 检查上次抽卡次数是否超时 + count_manager.check_count(user_id, count) # 检查次数累计 + pool = self.UP_CHAR if pool_name == "char" else self.UP_ARMS + for i in range(count): + count_manager.increase(user_id) + star = count_manager.check(user_id) # 是否有四星或五星保底 + if ( + count_manager.get_user_count(user_id) + - count_manager.get_user_five_index(user_id) + ) % count_manager.get_max_guarantee() >= 72: + add += draw_config.genshin.I72_ADD + if star: + if star == 4: + card = self.get_card(pool_name, 2, add=add) + else: + card = self.get_card( + pool_name, 3, add, count_manager.is_up(user_id) + ) + else: + card = self.get_card(pool_name, 1, add, count_manager.is_up(user_id)) + # print(f"{count_manager.get_user_count(user_id)}：", + # count_manager.get_user_five_index(user_id), star, card.star, add) + # 四星角色 + if card.star == 4: + count_manager.mark_four_index(user_id) + # 五星角色 + elif card.star == self.max_star: + add = 0 + count_manager.mark_five_index(user_id) # 记录五星保底 + count_manager.mark_four_index(user_id) # 记录四星保底 + if pool and card.name in [ + x.name for x in pool.up_char if x.star == self.max_star + ]: + count_manager.set_is_up(user_id, True) + else: + count_manager.set_is_up(user_id, False) + card_list.append((card, count_manager.get_user_count(user_id))) + count_manager.update_time(user_id) + return card_list + + def generate_card_img(self, card: GenshinData) -> BuildImage: + sep_w = 10 + sep_h = 5 + frame_w = 112 + frame_h = 132 + img_w = 106 + img_h = 106 + bg = BuildImage(frame_w + sep_w * 2, frame_h + sep_h * 2, color="#EBEBEB") + frame_path = str(self.img_path / "avatar_frame.png") + frame = Image.open(frame_path) + # 加名字 + text = card.name + font = load_font(fontsize=14) + text_w, text_h = font.getsize(text) + draw = ImageDraw.Draw(frame) + draw.text( + ((frame_w - text_w) / 2, frame_h - 15 - text_h / 2), + text, + font=font, + fill="gray", + ) + img_path = str(self.img_path / f"{cn2py(card.name)}.png") + img = BuildImage(img_w, img_h, background=img_path) + if isinstance(card, GenshinArms): + # 武器卡背景不是透明的，切去上方两个圆弧 + r = 12 + circle = Image.new("L", (r * 2, r * 2), 0) + alpha = Image.new("L", img.size, 255) + alpha.paste(circle, (-r - 3, -r - 3)) # 左上角 + alpha.paste(circle, (img_h - r + 3, -r - 3)) # 右上角 + img.markImg.putalpha(alpha) + star_path = str(self.img_path / f"{card.star}_star.png") + star = Image.open(star_path) + bg.paste(frame, (sep_w, sep_h), alpha=True) + bg.paste(img, (sep_w + 3, sep_h + 3), alpha=True) + bg.paste(star, (sep_w + int((frame_w - star.width) / 2), sep_h - 6), alpha=True) + return bg + + def format_pool_info(self, pool_name: str) -> str: + info = "" + up_event = None + if pool_name == "char": + up_event = self.UP_CHAR + elif pool_name == "arms": + up_event = self.UP_ARMS + if up_event: + star5_list = [x.name for x in up_event.up_char if x.star == 5] + star4_list = [x.name for x in up_event.up_char if x.star == 4] + if star5_list: + info += f"五星UP：{' '.join(star5_list)}\n" + if star4_list: + info += f"四星UP：{' '.join(star4_list)}\n" + info = f"当前up池：{up_event.title}\n{info}" + return info.strip() + + async def draw(self, count: int, user_id: int, pool_name: str = "", **kwargs) -> Message: + return await asyncio.get_event_loop().run_in_executor(None, self._draw, count, user_id, pool_name) + + def _draw(self, count: int, user_id: int, pool_name: str = "", **kwargs) -> Message: + index2cards = self.get_cards(count, user_id, pool_name) + cards = [card[0] for card in index2cards] + up_event = None + if pool_name == "char": + up_event = self.UP_CHAR + elif pool_name == "arms": + up_event = self.UP_ARMS + up_list = [x.name for x in up_event.up_char] if up_event else [] + result = self.format_star_result(cards) + result += ( + "\n" + max_star_str + if (max_star_str := self.format_max_star(index2cards, up_list=up_list)) + else "" + ) + result += f"\n距离保底发还剩 {self.count_manager.get_user_guarantee_count(user_id)} 抽" + # result += "\n【五星：0.6%，四星：5.1%，第72抽开始五星概率每抽加0.585%】" + pool_info = self.format_pool_info(pool_name) + img = self.generate_img(cards) + bk = BuildImage(img.w, img.h + 50, font_size=20, color="#ebebeb") + bk.paste(img) + bk.text((0, img.h + 10), "【五星：0.6%，四星：5.1%，第72抽开始五星概率每抽加0.585%】") + return pool_info + image(b64=bk.pic2bs4()) + result + + def _init_data(self): + self.ALL_CHAR = [ + GenshinChar( + name=value["名称"], + star=int(value["星级"]), + limited=value["常驻/限定"] == "限定UP", + ) + for key, value in self.load_data().items() + if "旅行者" not in key + ] + self.ALL_ARMS = [ + GenshinArms( + name=value["名称"], + star=int(value["星级"]), + limited="祈愿" not in value["获取途径"], + ) + for value in self.load_data("genshin_arms.json").values() + ] + self.load_up_char() + + def load_up_char(self): + try: + data = self.load_data(f"draw_card_up/{self.game_name}_up_char.json") + self.UP_CHAR = UpEvent.parse_obj(data.get("char", {})) + self.UP_ARMS = UpEvent.parse_obj(data.get("arms", {})) + except ValidationError: + logger.warning(f"{self.game_name}_up_char 解析出错") + + def dump_up_char(self): + if self.UP_CHAR and self.UP_ARMS: + data = { + "char": json.loads(self.UP_CHAR.json()), + "arms": json.loads(self.UP_ARMS.json()), + } + self.dump_data(data, f"draw_card_up/{self.game_name}_up_char.json") + + async def _update_info(self): + # genshin.json + char_info = {} + url = "https://wiki.biligame.com/ys/角色筛选" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + else: + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath("//table[@id='CardSelectTr']/tbody/tr") + for char in char_list: + try: + name = char.xpath("./td[1]/a/@title")[0] + avatar = char.xpath("./td[1]/a/img/@srcset")[0] + star = char.xpath("./td[3]/text()")[0] + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar).split(" ")[-2]), + "名称": remove_prohibited_str(name), + "星级": int(str(star).strip()[:1]), + } + char_info[member_dict["名称"]] = member_dict + # 更新额外信息 + for key in char_info.keys(): + result = await self.get_url(f"https://wiki.biligame.com/ys/{key}") + if not result: + char_info[key]["常驻/限定"] = "未知" + logger.warning(f"{self.game_name_cn} 获取额外信息错误 {key}") + continue + try: + dom = etree.HTML(result, etree.HTMLParser()) + limit = dom.xpath( + "//table[contains(string(.),'常驻/限定')]/tbody/tr[6]/td/text()" + )[0] + char_info[key]["常驻/限定"] = str(limit).strip() + except IndexError: + char_info[key]["常驻/限定"] = "未知" + logger.warning(f"{self.game_name_cn} 获取额外信息错误 {key}") + self.dump_data(char_info) + logger.info(f"{self.game_name_cn} 更新成功") + # genshin_arms.json + arms_info = {} + url = "https://wiki.biligame.com/ys/武器图鉴" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + else: + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath("//table[@id='CardSelectTr']/tbody/tr") + for char in char_list: + try: + name = char.xpath("./td[1]/a/@title")[0] + avatar = char.xpath("./td[1]/a/img/@srcset")[0] + star = char.xpath("./td[4]/img/@alt")[0] + sources = str(char.xpath("./td[5]/text()")[0]).split(",") + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar).split(" ")[-2]), + "名称": remove_prohibited_str(name), + "星级": int(str(star).strip()[:1]), + "获取途径": [s.strip() for s in sources if s.strip()], + } + arms_info[member_dict["名称"]] = member_dict + self.dump_data(arms_info, "genshin_arms.json") + logger.info(f"{self.game_name_cn} 武器更新成功") + # 下载头像 + for value in char_info.values(): + await self.download_img(value["头像"], value["名称"]) + for value in arms_info.values(): + await self.download_img(value["头像"], value["名称"]) + # 下载星星 + idx = 1 + YS_URL = "https://patchwiki.biligame.com/images/ys" + for url in [ + "/1/13/7xzg7tgf8dsr2hjpmdbm5gn9wvzt2on.png", + "/b/bc/sd2ige6d7lvj7ugfumue3yjg8gyi0d1.png", + "/e/ec/l3mnhy56pyailhn3v7r873htf2nofau.png", + "/9/9c/sklp02ffk3aqszzvh8k1c3139s0awpd.png", + "/c/c7/qu6xcndgj6t14oxvv7yz2warcukqv1m.png", + ]: + await self.download_img(YS_URL + url, f"{idx}_star") + idx += 1 + # 下载头像框 + await self.download_img( + YS_URL + "/2/2e/opbcst4xbtcq0i4lwerucmosawn29ti.png", f"avatar_frame" + ) + await self.update_up_char() + + async def update_up_char(self): + url = "https://wiki.biligame.com/ys/祈愿" + result = await self.get_url(url) + if not result: + logger.warning(f"{self.game_name_cn}获取祈愿页面出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + tables = dom.xpath( + "//div[@class='mw-parser-output']/div[@class='row']/div/table[@class='wikitable']/tbody" + ) + if not tables or len(tables) < 2: + logger.warning(f"{self.game_name_cn}获取活动祈愿出错") + return + try: + for index, table in enumerate(tables): + title = table.xpath("./tr[1]/th/img/@title")[0] + title = str(title).split("」")[0] + "」" if "」" in title else title + pool_img = str(table.xpath("./tr[1]/th/img/@srcset")[0]).split(" ")[-2] + time = table.xpath("./tr[2]/td/text()")[0] + star5_list = table.xpath("./tr[3]/td/a/@title") + star4_list = table.xpath("./tr[4]/td/a/@title") + start, end = str(time).split("~") + start_time = dateparser.parse(start) + end_time = dateparser.parse(end) + if not start_time and end_time: + start_time = end_time - timedelta(days=20) + if start_time and end_time and start_time <= datetime.now() <= end_time: + up_event = UpEvent( + title=title, + pool_img=pool_img, + start_time=start_time, + end_time=end_time, + up_char=[ + UpChar(name=name, star=5, limited=False, zoom=50) + for name in star5_list + ] + + [ + UpChar(name=name, star=4, limited=False, zoom=50) + for name in star4_list + ], + ) + if index == 0: + self.UP_CHAR = up_event + elif index == 1: + self.UP_ARMS = up_event + if self.UP_CHAR and self.UP_ARMS: + self.dump_up_char() + logger.info( + f"成功获取{self.game_name_cn}当前up信息...当前up池: {self.UP_CHAR.title} & {self.UP_ARMS.title}" + ) + except Exception as e: + logger.warning(f"{self.game_name_cn}UP更新出错 {type(e)}：{e}") + + def reset_count(self, user_id: int) -> bool: + self.count_manager.reset(user_id) + return True + + async def _reload_pool(self) -> Optional[Message]: + await self.update_up_char() + self.load_up_char() + if self.UP_CHAR and self.UP_ARMS: + return Message( + Message.template("重载成功！\n当前UP池子：{} & {}{:image}{:image}").format( + self.UP_CHAR.title, + self.UP_ARMS.title, + self.UP_CHAR.pool_img, + self.UP_ARMS.pool_img, + ) + ) diff --git a/plugins/draw_card/handles/guardian_handle.py b/plugins/draw_card/handles/guardian_handle.py new file mode 100644 index 00000000..9c6766e0 --- /dev/null +++ b/plugins/draw_card/handles/guardian_handle.py @@ -0,0 +1,400 @@ +import re +import random +import dateparser +from lxml import etree +from PIL import ImageDraw +from datetime import datetime +from urllib.parse import unquote +from typing import List, Optional, Tuple +from pydantic import ValidationError +from nonebot.adapters.onebot.v11 import Message +from utils.message_builder import image +from nonebot.log import logger +import asyncio + +try: + import ujson as json +except ModuleNotFoundError: + import json + +from .base_handle import BaseHandle, BaseData, UpChar, UpEvent +from ..config import draw_config +from ..util import remove_prohibited_str, cn2py, load_font +from utils.image_utils import BuildImage + + +class GuardianData(BaseData): + pass + + +class GuardianChar(GuardianData): + pass + + +class GuardianArms(GuardianData): + pass + + +class GuardianHandle(BaseHandle[GuardianData]): + def __init__(self): + super().__init__("guardian", "坎公骑冠剑") + self.data_files.append("guardian_arms.json") + self.config = draw_config.guardian + + self.ALL_CHAR: List[GuardianChar] = [] + self.ALL_ARMS: List[GuardianArms] = [] + self.UP_CHAR: Optional[UpEvent] = None + self.UP_ARMS: Optional[UpEvent] = None + + def get_card(self, pool_name: str, mode: int = 1) -> GuardianData: + if pool_name == "char": + if mode == 1: + star = self.get_star( + [3, 2, 1], + [ + self.config.GUARDIAN_THREE_CHAR_P, + self.config.GUARDIAN_TWO_CHAR_P, + self.config.GUARDIAN_ONE_CHAR_P, + ], + ) + else: + star = self.get_star( + [3, 2], + [ + self.config.GUARDIAN_THREE_CHAR_P, + self.config.GUARDIAN_TWO_CHAR_P, + ], + ) + up_event = self.UP_CHAR + self.max_star = 3 + all_data = self.ALL_CHAR + else: + if mode == 1: + star = self.get_star( + [5, 4, 3, 2], + [ + self.config.GUARDIAN_FIVE_ARMS_P, + self.config.GUARDIAN_FOUR_ARMS_P, + self.config.GUARDIAN_THREE_ARMS_P, + self.config.GUARDIAN_TWO_ARMS_P, + ], + ) + else: + star = self.get_star( + [5, 4], + [ + self.config.GUARDIAN_FIVE_ARMS_P, + self.config.GUARDIAN_FOUR_ARMS_P, + ], + ) + up_event = self.UP_ARMS + self.max_star = 5 + all_data = self.ALL_ARMS + + acquire_char = None + # 是否UP + if up_event and star == self.max_star and pool_name: + # 获取up角色列表 + up_list = [x.name for x in up_event.up_char if x.star == star] + # 成功获取up角色 + if random.random() < 0.5: + up_name = random.choice(up_list) + try: + acquire_char = [x for x in all_data if x.name == up_name][0] + except IndexError: + pass + if not acquire_char: + chars = [x for x in all_data if x.star == star and not x.limited] + acquire_char = random.choice(chars) + return acquire_char + + def get_cards(self, count: int, pool_name: str) -> List[Tuple[GuardianData, int]]: + card_list = [] + card_count = 0 # 保底计算 + for i in range(count): + card_count += 1 + # 十连保底 + if card_count == 10: + card = self.get_card(pool_name, 2) + card_count = 0 + else: + card = self.get_card(pool_name, 1) + if card.star > self.max_star - 2: + card_count = 0 + card_list.append((card, i + 1)) + return card_list + + def format_pool_info(self, pool_name: str) -> str: + info = "" + up_event = self.UP_CHAR if pool_name == "char" else self.UP_ARMS + if up_event: + if pool_name == "char": + up_list = [x.name for x in up_event.up_char if x.star == 3] + info += f'三星UP：{" ".join(up_list)}\n' + else: + up_list = [x.name for x in up_event.up_char if x.star == 5] + info += f'五星UP：{" ".join(up_list)}\n' + info = f"当前up池：{up_event.title}\n{info}" + return info.strip() + + async def draw(self, count: int, pool_name: str, **kwargs) -> Message: + return await asyncio.get_event_loop().run_in_executor(None, self._draw, count, pool_name) + + def _draw(self, count: int, pool_name: str, **kwargs) -> Message: + index2card = self.get_cards(count, pool_name) + cards = [card[0] for card in index2card] + up_event = self.UP_CHAR if pool_name == "char" else self.UP_ARMS + up_list = [x.name for x in up_event.up_char] if up_event else [] + result = self.format_result(index2card, up_list=up_list) + pool_info = self.format_pool_info(pool_name) + return pool_info + image(b64=self.generate_img(cards).pic2bs4()) + result + + def generate_card_img(self, card: GuardianData) -> BuildImage: + sep_w = 1 + sep_h = 1 + block_w = 170 + block_h = 90 + img_w = 90 + img_h = 90 + if isinstance(card, GuardianChar): + block_color = "#2e2923" + font_color = "#e2ccad" + star_w = 90 + star_h = 30 + star_name = f"{card.star}_star.png" + frame_path = "" + else: + block_color = "#EEE4D5" + font_color = "#A65400" + star_w = 45 + star_h = 45 + star_name = f"{card.star}_star_rank.png" + frame_path = str(self.img_path / "avatar_frame.png") + bg = BuildImage(block_w + sep_w * 2, block_h + sep_h * 2, color="#F6F4ED") + block = BuildImage(block_w, block_h, color=block_color) + star_path = str(self.img_path / star_name) + star = BuildImage(star_w, star_h, background=star_path) + img_path = str(self.img_path / f"{cn2py(card.name)}.png") + img = BuildImage(img_w, img_h, background=img_path) + block.paste(img, (0, 0), alpha=True) + if frame_path: + frame = BuildImage(img_w, img_h, background=frame_path) + block.paste(frame, (0, 0), alpha=True) + block.paste( + star, + (int((block_w + img_w - star_w) / 2), block_h - star_h - 30), + alpha=True, + ) + # 加名字 + text = card.name[:4] + "..." if len(card.name) > 5 else card.name + font = load_font(fontsize=14) + text_w, _ = font.getsize(text) + draw = ImageDraw.Draw(block.markImg) + draw.text( + ((block_w + img_w - text_w) / 2, 55), + text, + font=font, + fill=font_color, + ) + bg.paste(block, (sep_w, sep_h)) + return bg + + def _init_data(self): + self.ALL_CHAR = [ + GuardianChar(name=value["名称"], star=int(value["星级"]), limited=False) + for value in self.load_data().values() + ] + self.ALL_ARMS = [ + GuardianArms(name=value["名称"], star=int(value["星级"]), limited=False) + for value in self.load_data("guardian_arms.json").values() + ] + self.load_up_char() + + def load_up_char(self): + try: + data = self.load_data(f"draw_card_up/{self.game_name}_up_char.json") + self.UP_CHAR = UpEvent.parse_obj(data.get("char", {})) + self.UP_ARMS = UpEvent.parse_obj(data.get("arms", {})) + except ValidationError: + logger.warning(f"{self.game_name}_up_char 解析出错") + + def dump_up_char(self): + if self.UP_CHAR and self.UP_ARMS: + data = { + "char": json.loads(self.UP_CHAR.json()), + "arms": json.loads(self.UP_ARMS.json()), + } + self.dump_data(data, f"draw_card_up/{self.game_name}_up_char.json") + + async def _update_info(self): + # guardian.json + guardian_info = {} + url = "https://wiki.biligame.com/gt/英雄筛选表" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + else: + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath("//table[@id='CardSelectTr']/tbody/tr") + for char in char_list: + try: + name = char.xpath("./td[1]/a/@title")[0] + avatar = char.xpath("./td[1]/a/img/@src")[0] + star = char.xpath("./td[1]/span/img/@alt")[0] + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar)), + "名称": remove_prohibited_str(name), + "星级": int(str(star).split(" ")[0].replace("Rank", "")), + } + guardian_info[member_dict["名称"]] = member_dict + self.dump_data(guardian_info) + logger.info(f"{self.game_name_cn} 更新成功") + # guardian_arms.json + guardian_arms_info = {} + url = "https://wiki.biligame.com/gt/武器" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 武器出错") + else: + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath( + "//div[@class='resp-tabs-container']/div[1]/div/table[2]/tbody/tr" + ) + for char in char_list: + try: + name = char.xpath("./td[2]/a/@title")[0] + avatar = char.xpath("./td[1]/div/div/div/a/img/@src")[0] + star = char.xpath("./td[3]/text()")[0] + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar)), + "名称": remove_prohibited_str(name), + "星级": int(str(star).strip()), + } + guardian_arms_info[member_dict["名称"]] = member_dict + self.dump_data(guardian_arms_info, "guardian_arms.json") + logger.info(f"{self.game_name_cn} 武器更新成功") + url = "https://wiki.biligame.com/gt/盾牌" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 盾牌出错") + else: + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath( + "//div[@class='resp-tabs-container']/div[2]/div/table[1]/tbody/tr" + ) + for char in char_list: + try: + name = char.xpath("./td[2]/a/@title")[0] + avatar = char.xpath("./td[1]/div/div/div/a/img/@src")[0] + star = char.xpath("./td[3]/text()")[0] + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar)), + "名称": remove_prohibited_str(name), + "星级": int(str(star).strip()), + } + guardian_arms_info[member_dict["名称"]] = member_dict + self.dump_data(guardian_arms_info, "guardian_arms.json") + logger.info(f"{self.game_name_cn} 盾牌更新成功") + # 下载头像 + for value in guardian_info.values(): + await self.download_img(value["头像"], value["名称"]) + for value in guardian_arms_info.values(): + await self.download_img(value["头像"], value["名称"]) + # 下载星星 + idx = 1 + GT_URL = "https://patchwiki.biligame.com/images/gt" + for url in [ + "/4/4b/ardr3bi2yf95u4zomm263tc1vke6i3i.png", + "/5/55/6vow7lh76gzus6b2g9cfn325d1sugca.png", + "/b/b9/du8egrd2vyewg0cuyra9t8jh0srl0ds.png", + ]: + await self.download_img(GT_URL + url, f"{idx}_star") + idx += 1 + # 另一种星星 + idx = 1 + for url in [ + "/6/66/4e2tfa9kvhfcbikzlyei76i9crva145.png", + "/1/10/r9ihsuvycgvsseyneqz4xs22t53026m.png", + "/7/7a/o0k86ru9k915y04azc26hilxead7xp1.png", + "/c/c9/rxz99asysz0rg391j3b02ta09mnpa7v.png", + "/2/2a/sfxz0ucv1s6ewxveycz9mnmrqs2rw60.png", + ]: + await self.download_img(GT_URL + url, f"{idx}_star_rank") + idx += 1 + # 头像框 + await self.download_img( + GT_URL + "/8/8e/ogbqslbhuykjhnc8trtoa0p0nhfzohs.png", f"avatar_frame" + ) + await self.update_up_char() + + async def update_up_char(self): + url = "https://wiki.biligame.com/gt/首页" + result = await self.get_url(url) + if not result: + logger.warning(f"{self.game_name_cn}获取公告出错") + return + try: + dom = etree.HTML(result, etree.HTMLParser()) + announcement = dom.xpath( + "//div[@class='mw-parser-output']/div/div[3]/div[2]/div/div[2]/div[3]" + )[0] + title = announcement.xpath("./font/p/b/text()")[0] + match = re.search(r"从(.*?)开始.*?至(.*?)结束", title) + if not match: + logger.warning(f"{self.game_name_cn}找不到UP时间") + return + start, end = match.groups() + start_time = dateparser.parse(start.replace("月", "/").replace("日", "")) + end_time = dateparser.parse(end.replace("月", "/").replace("日", "")) + if not (start_time and end_time) or not ( + start_time <= datetime.now() <= end_time + ): + return + divs = announcement.xpath("./font/div") + char_index = 0 + arms_index = 0 + for index, div in enumerate(divs): + if div.xpath("string(.)") == "角色": + char_index = index + elif div.xpath("string(.)") == "武器": + arms_index = index + chars = divs[char_index + 1 : arms_index] + arms = divs[arms_index + 1 :] + up_chars = [] + up_arms = [] + for char in chars: + name = char.xpath("./p/a/@title")[0] + up_chars.append(UpChar(name=name, star=3, limited=False, zoom=0)) + for arm in arms: + name = arm.xpath("./p/a/@title")[0] + up_arms.append(UpChar(name=name, star=5, limited=False, zoom=0)) + self.UP_CHAR = UpEvent( + title=title, + pool_img="", + start_time=start_time, + end_time=end_time, + up_char=up_chars, + ) + self.UP_ARMS = UpEvent( + title=title, + pool_img="", + start_time=start_time, + end_time=end_time, + up_char=up_arms, + ) + self.dump_up_char() + logger.info(f"成功获取{self.game_name_cn}当前up信息...当前up池: {title}") + except Exception as e: + logger.warning(f"{self.game_name_cn}UP更新出错 {type(e)}：{e}") + + async def _reload_pool(self) -> Optional[Message]: + await self.update_up_char() + self.load_up_char() + if self.UP_CHAR and self.UP_ARMS: + return Message(f"重载成功！\n当前UP池子：{self.UP_CHAR.title}") diff --git a/plugins/draw_card/handles/onmyoji_handle.py b/plugins/draw_card/handles/onmyoji_handle.py new file mode 100644 index 00000000..5797e830 --- /dev/null +++ b/plugins/draw_card/handles/onmyoji_handle.py @@ -0,0 +1,179 @@ +import random +from lxml import etree +from typing import List, Tuple +from nonebot.log import logger +from PIL import Image, ImageDraw +from PIL.Image import Image as IMG + +try: + import ujson as json +except ModuleNotFoundError: + import json + +from .base_handle import BaseHandle, BaseData +from ..config import draw_config +from ..util import remove_prohibited_str, cn2py, load_font +from utils.image_utils import BuildImage + + +class OnmyojiChar(BaseData): + @property + def star_str(self) -> str: + return ["N", "R", "SR", "SSR", "SP"][self.star - 1] + + +class OnmyojiHandle(BaseHandle[OnmyojiChar]): + def __init__(self): + super().__init__("onmyoji", "阴阳师") + self.max_star = 5 + self.config = draw_config.onmyoji + self.ALL_CHAR: List[OnmyojiChar] = [] + + def get_card(self, **kwargs) -> OnmyojiChar: + star = self.get_star( + [5, 4, 3, 2], + [ + self.config.ONMYOJI_SP, + self.config.ONMYOJI_SSR, + self.config.ONMYOJI_SR, + self.config.ONMYOJI_R, + ], + ) + chars = [x for x in self.ALL_CHAR if x.star == star and not x.limited] + return random.choice(chars) + + def format_max_star(self, card_list: List[Tuple[OnmyojiChar, int]]) -> str: + rst = "" + for card, index in card_list: + if card.star == self.max_star: + rst += f"第 {index} 抽获取SP {card.name}\n" + elif card.star == self.max_star - 1: + rst += f"第 {index} 抽获取SSR {card.name}\n" + return rst.strip() + + @staticmethod + def star_label(star: int) -> IMG: + text, color1, color2 = [ + ("N", "#7E7E82", "#F5F6F7"), + ("R", "#014FA8", "#37C6FD"), + ("SR", "#6E0AA4", "#E94EFD"), + ("SSR", "#E5511D", "#FAF905"), + ("SP", "#FA1F2D", "#FFBBAF"), + ][star - 1] + w = 200 + h = 110 + # 制作渐变色图片 + base = Image.new("RGBA", (w, h), color1) + top = Image.new("RGBA", (w, h), color2) + mask = Image.new("L", (w, h)) + mask_data = [] + for y in range(h): + mask_data.extend([int(255 * (y / h))] * w) + mask.putdata(mask_data) + base.paste(top, (0, 0), mask) + # 透明图层 + font = load_font("gorga.otf", 100) + alpha = Image.new("L", (w, h)) + draw = ImageDraw.Draw(alpha) + draw.text((20, -30), text, fill="white", font=font) + base.putalpha(alpha) + # stroke + bg = Image.new("RGBA", (w, h)) + draw = ImageDraw.Draw(bg) + draw.text( + (20, -30), + text, + font=font, + fill="gray", + stroke_width=3, + stroke_fill="gray", + ) + bg.paste(base, (0, 0), base) + return bg + + def generate_img(self, card_list: List[OnmyojiChar]) -> BuildImage: + return super().generate_img(card_list, num_per_line=10) + + def generate_card_img(self, card: OnmyojiChar) -> BuildImage: + bg = BuildImage(73, 240, color="#F1EFE9") + img_path = str(self.img_path / f"{cn2py(card.name)}_mark_btn.png") + img = BuildImage(0, 0, background=img_path) + img = Image.open(img_path).convert("RGBA") + label = self.star_label(card.star).resize((60, 33), Image.ANTIALIAS) + bg.paste(img, (0, 0), alpha=True) + bg.paste(label, (0, 135), alpha=True) + font = load_font("msyh.ttf", 16) + draw = ImageDraw.Draw(bg.markImg) + text = "\n".join([t for t in card.name[:4]]) + _, text_h = font.getsize_multiline(text, spacing=0) + draw.text( + (40, 150 + (90 - text_h) / 2), text, font=font, fill="gray", spacing=0 + ) + return bg + + def _init_data(self): + self.ALL_CHAR = [ + OnmyojiChar( + name=value["名称"], + star=["N", "R", "SR", "SSR", "SP"].index(value["星级"]) + 1, + limited=True + if key + in [ + "奴良陆生", + "卖药郎", + "鬼灯", + "阿香", + "蜜桃&芥子", + "犬夜叉", + "杀生丸", + "桔梗", + "朽木露琪亚", + "黑崎一护", + "灶门祢豆子", + "灶门炭治郎", + ] + else False, + ) + for key, value in self.load_data().items() + ] + + async def _update_info(self): + info = {} + url = "https://yys.res.netease.com/pc/zt/20161108171335/js/app/all_shishen.json?v74=" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + return + data = json.loads(result) + for x in data: + name = remove_prohibited_str(x["name"]) + member_dict = { + "id": x["id"], + "名称": name, + "星级": x["level"], + } + info[name] = member_dict + # logger.info(f"{name} is update...") + # 更新头像 + for key in info.keys(): + url = f'https://yys.163.com/shishen/{info[key]["id"]}.html' + result = await self.get_url(url) + if not result: + info[key]["头像"] = "" + continue + try: + dom = etree.HTML(result, etree.HTMLParser()) + avatar = dom.xpath("//div[@class='pic_wrap']/img/@src")[0] + avatar = "https:" + avatar + info[key]["头像"] = avatar + except IndexError: + info[key]["头像"] = "" + logger.warning(f"{self.game_name_cn} 获取头像错误 {key}") + self.dump_data(info) + logger.info(f"{self.game_name_cn} 更新成功") + # 下载头像 + for value in info.values(): + await self.download_img(value["头像"], value["名称"]) + # 下载书签形式的头像 + url = f"https://yys.res.netease.com/pc/zt/20161108171335/data/mark_btn/{value['id']}.png" + await self.download_img(url, value["名称"] + "_mark_btn") diff --git a/plugins/draw_card/handles/pcr_handle.py b/plugins/draw_card/handles/pcr_handle.py new file mode 100644 index 00000000..d82ee27a --- /dev/null +++ b/plugins/draw_card/handles/pcr_handle.py @@ -0,0 +1,147 @@ +import random +from lxml import etree +from typing import List, Tuple +from PIL import ImageDraw +from urllib.parse import unquote +from nonebot.log import logger + +from .base_handle import BaseHandle, BaseData +from ..config import draw_config +from ..util import remove_prohibited_str, cn2py, load_font +from utils.image_utils import BuildImage + + +class PcrChar(BaseData): + pass + + +class PcrHandle(BaseHandle[PcrChar]): + def __init__(self): + super().__init__("pcr", "公主连结") + self.max_star = 3 + self.config = draw_config.pcr + self.ALL_CHAR: List[PcrChar] = [] + + def get_card(self, mode: int = 1) -> PcrChar: + if mode == 2: + star = self.get_star( + [3, 2], [self.config.PCR_G_THREE_P, self.config.PCR_G_TWO_P] + ) + else: + star = self.get_star( + [3, 2, 1], + [self.config.PCR_THREE_P, self.config.PCR_TWO_P, self.config.PCR_ONE_P], + ) + chars = [x for x in self.ALL_CHAR if x.star == star and not x.limited] + return random.choice(chars) + + def get_cards(self, count: int, **kwargs) -> List[Tuple[PcrChar, int]]: + card_list = [] + card_count = 0 # 保底计算 + for i in range(count): + card_count += 1 + # 十连保底 + if card_count == 10: + card = self.get_card(2) + card_count = 0 + else: + card = self.get_card(1) + if card.star > self.max_star - 2: + card_count = 0 + card_list.append((card, i + 1)) + return card_list + + def generate_card_img(self, card: PcrChar) -> BuildImage: + sep_w = 5 + sep_h = 5 + star_h = 15 + img_w = 90 + img_h = 90 + font_h = 20 + bg = BuildImage(img_w + sep_w * 2, img_h + font_h + sep_h * 2, color="#EFF2F5") + star_path = str(self.img_path / "star.png") + star = BuildImage(star_h, star_h, background=star_path) + img_path = str(self.img_path / f"{cn2py(card.name)}.png") + img = BuildImage(img_w, img_h, background=img_path) + bg.paste(img, (sep_w, sep_h), alpha=True) + for i in range(card.star): + bg.paste(star, (sep_w + img_w - star_h * (i + 1), sep_h), alpha=True) + # 加名字 + text = card.name[:5] + "..." if len(card.name) > 6 else card.name + font = load_font(fontsize=14) + text_w, text_h = font.getsize(text) + draw = ImageDraw.Draw(bg.markImg) + draw.text( + (sep_w + (img_w - text_w) / 2, sep_h + img_h + (font_h - text_h) / 2), + text, + font=font, + fill="gray", + ) + return bg + + def _init_data(self): + self.ALL_CHAR = [ + PcrChar( + name=value["名称"], + star=int(value["星级"]), + limited=True if "（" in key else False, + ) + for key, value in self.load_data().items() + ] + + async def _update_info(self): + info = {} + if draw_config.PCR_TAI: + url = "https://wiki.biligame.com/pcr/角色图鉴" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath( + "//div[@class='resp-tab-content']/div[@class='unit-icon']" + ) + for char in char_list: + try: + name = char.xpath("./a/@title")[0] + avatar = char.xpath("./a/img/@srcset")[0] + star = len(char.xpath("./div[1]/img")) + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar).split(" ")[-2]), + "名称": remove_prohibited_str(name), + "星级": star, + } + info[member_dict["名称"]] = member_dict + else: + url = "https://wiki.biligame.com/pcr/角色筛选表" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath("//table[@id='CardSelectTr']/tbody/tr") + for char in char_list: + try: + name = char.xpath("./td[1]/a/@title")[0] + avatar = char.xpath("./td[1]/a/img/@srcset")[0] + star = char.xpath("./td[4]/text()")[0] + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar).split(" ")[-2]), + "名称": remove_prohibited_str(name), + "星级": int(str(star).strip()), + } + info[member_dict["名称"]] = member_dict + self.dump_data(info) + logger.info(f"{self.game_name_cn} 更新成功") + # 下载头像 + for value in info.values(): + await self.download_img(value["头像"], value["名称"]) + # 下载星星 + await self.download_img( + "https://patchwiki.biligame.com/images/pcr/0/02/s75ys2ecqhu2xbdw1wf1v9ccscnvi5g.png", + "star", + ) diff --git a/plugins/draw_card/handles/pretty_handle.py b/plugins/draw_card/handles/pretty_handle.py new file mode 100644 index 00000000..da56bd64 --- /dev/null +++ b/plugins/draw_card/handles/pretty_handle.py @@ -0,0 +1,426 @@ +import re +import random +import dateparser +from lxml import etree +from PIL import ImageDraw +from bs4 import BeautifulSoup +from datetime import datetime +from urllib.parse import unquote +from typing import List, Optional, Tuple +from pydantic import ValidationError +from nonebot.adapters.onebot.v11 import Message +from utils.message_builder import image +from nonebot.log import logger +import asyncio + +try: + import ujson as json +except ModuleNotFoundError: + import json + +from .base_handle import BaseHandle, BaseData, UpChar, UpEvent +from ..config import draw_config +from ..util import remove_prohibited_str, cn2py, load_font +from utils.image_utils import BuildImage + + +class PrettyData(BaseData): + pass + + +class PrettyChar(PrettyData): + pass + + +class PrettyCard(PrettyData): + @property + def star_str(self) -> str: + return ["R", "SR", "SSR"][self.star - 1] + + +class PrettyHandle(BaseHandle[PrettyData]): + def __init__(self): + super().__init__("pretty", "赛马娘", "#eff2f5") + self.data_files.append("pretty_card.json") + self.max_star = 3 + self.config = draw_config.pretty + + self.ALL_CHAR: List[PrettyChar] = [] + self.ALL_CARD: List[PrettyCard] = [] + self.UP_CHAR: Optional[UpEvent] = None + self.UP_CARD: Optional[UpEvent] = None + + def get_card(self, pool_name: str, mode: int = 1) -> PrettyData: + if mode == 1: + star = self.get_star( + [3, 2, 1], + [ + self.config.PRETTY_THREE_P, + self.config.PRETTY_TWO_P, + self.config.PRETTY_ONE_P, + ], + ) + else: + star = self.get_star( + [3, 2], [self.config.PRETTY_THREE_P, self.config.PRETTY_TWO_P] + ) + up_pool = None + if pool_name == "char": + up_pool = self.UP_CHAR + all_list = self.ALL_CHAR + else: + up_pool = self.UP_CARD + all_list = self.ALL_CARD + + all_char = [x for x in all_list if x.star == star and not x.limited] + acquire_char = None + # 有UP池子 + if up_pool and star in [x.star for x in up_pool.up_char]: + up_list = [x.name for x in up_pool.up_char if x.star == star] + # 抽到UP + if random.random() < 1 / len(all_char) * (0.7 / 0.1385): + up_name = random.choice(up_list) + try: + acquire_char = [x for x in all_list if x.name == up_name][0] + except IndexError: + pass + if not acquire_char: + acquire_char = random.choice(all_char) + return acquire_char + + def get_cards(self, count: int, pool_name: str) -> List[Tuple[PrettyData, int]]: + card_list = [] + card_count = 0 # 保底计算 + for i in range(count): + card_count += 1 + # 十连保底 + if card_count == 10: + card = self.get_card(pool_name, 2) + card_count = 0 + else: + card = self.get_card(pool_name, 1) + if card.star > self.max_star - 2: + card_count = 0 + card_list.append((card, i + 1)) + return card_list + + def format_pool_info(self, pool_name: str) -> str: + info = "" + up_event = self.UP_CHAR if pool_name == "char" else self.UP_CARD + if up_event: + star3_list = [x.name for x in up_event.up_char if x.star == 3] + star2_list = [x.name for x in up_event.up_char if x.star == 2] + star1_list = [x.name for x in up_event.up_char if x.star == 1] + if star3_list: + if pool_name == "char": + info += f'三星UP：{" ".join(star3_list)}\n' + else: + info += f'SSR UP：{" ".join(star3_list)}\n' + if star2_list: + if pool_name == "char": + info += f'二星UP：{" ".join(star2_list)}\n' + else: + info += f'SR UP：{" ".join(star2_list)}\n' + if star1_list: + if pool_name == "char": + info += f'一星UP：{" ".join(star1_list)}\n' + else: + info += f'R UP：{" ".join(star1_list)}\n' + info = f"当前up池：{up_event.title}\n{info}" + return info.strip() + + async def draw(self, count: int, pool_name: str, **kwargs) -> Message: + return await asyncio.get_event_loop().run_in_executor(None, self._draw, count, pool_name) + + def _draw(self, count: int, pool_name: str, **kwargs) -> Message: + pool_name = "char" if not pool_name else pool_name + index2card = self.get_cards(count, pool_name) + cards = [card[0] for card in index2card] + up_event = self.UP_CHAR if pool_name == "char" else self.UP_CARD + up_list = [x.name for x in up_event.up_char] if up_event else [] + result = self.format_result(index2card, up_list=up_list) + pool_info = self.format_pool_info(pool_name) + return pool_info + image(b64=self.generate_img(cards).pic2bs4()) + result + + def generate_card_img(self, card: PrettyData) -> BuildImage: + if isinstance(card, PrettyChar): + star_h = 30 + img_w = 200 + img_h = 219 + font_h = 50 + bg = BuildImage(img_w, img_h + font_h, color="#EFF2F5") + star_path = str(self.img_path / "star.png") + star = BuildImage(star_h, star_h, background=star_path) + img_path = str(self.img_path / f"{cn2py(card.name)}.png") + img = BuildImage(img_w, img_h, background=img_path) + star_w = star_h * card.star + for i in range(card.star): + bg.paste(star, (int((img_w - star_w) / 2) + star_h * i, 0), alpha=True) + bg.paste(img, (0, 0), alpha=True) + # 加名字 + text = card.name[:5] + "..." if len(card.name) > 6 else card.name + font = load_font(fontsize=30) + text_w, _ = font.getsize(text) + draw = ImageDraw.Draw(bg.markImg) + draw.text( + ((img_w - text_w) / 2, img_h), + text, + font=font, + fill="gray", + ) + return bg + else: + sep_w = 10 + img_w = 200 + img_h = 267 + font_h = 75 + bg = BuildImage(img_w + sep_w * 2, img_h + font_h, color="#EFF2F5") + label_path = str(self.img_path / f"{card.star}_label.png") + label = BuildImage(40, 40, background=label_path) + img_path = str(self.img_path / f"{cn2py(card.name)}.png") + img = BuildImage(img_w, img_h, background=img_path) + bg.paste(img, (sep_w, 0), alpha=True) + bg.paste(label, (30, 3), alpha=True) + # 加名字 + text = "" + texts = [] + font = load_font(fontsize=25) + for t in card.name: + if font.getsize(text + t)[0] > 190: + texts.append(text) + text = "" + if len(texts) >= 2: + texts[-1] += "..." + break + else: + text += t + if text: + texts.append(text) + text = "\n".join(texts) + text_w, _ = font.getsize_multiline(text) + draw = ImageDraw.Draw(bg.markImg) + draw.text( + ((img_w - text_w) / 2, img_h), + text, + font=font, + align="center", + fill="gray", + ) + return bg + + def _init_data(self): + self.ALL_CHAR = [ + PrettyChar( + name=value["名称"], + star=int(value["初始星级"]), + limited=False, + ) + for value in self.load_data().values() + ] + self.ALL_CARD = [ + PrettyCard( + name=value["中文名"], + star=["R", "SR", "SSR"].index(value["稀有度"]) + 1, + limited=True if "卡池" not in value["获取方式"] else False, + ) + for value in self.load_data("pretty_card.json").values() + ] + self.load_up_char() + + def load_up_char(self): + try: + data = self.load_data(f"draw_card_up/{self.game_name}_up_char.json") + self.UP_CHAR = UpEvent.parse_obj(data.get("char", {})) + self.UP_CARD = UpEvent.parse_obj(data.get("card", {})) + except ValidationError: + logger.warning(f"{self.game_name}_up_char 解析出错") + + def dump_up_char(self): + if self.UP_CHAR and self.UP_CARD: + data = { + "char": json.loads(self.UP_CHAR.json()), + "card": json.loads(self.UP_CARD.json()), + } + self.dump_data(data, f"draw_card_up/{self.game_name}_up_char.json") + + async def _update_info(self): + # pretty.json + pretty_info = {} + url = "https://wiki.biligame.com/umamusume/赛马娘图鉴" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + else: + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath("//table[@id='CardSelectTr']/tbody/tr") + for char in char_list: + try: + name = char.xpath("./td[1]/a/@title")[0] + avatar = char.xpath("./td[1]/a/img/@srcset")[0] + star = len(char.xpath("./td[3]/img")) + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar).split(" ")[-2]), + "名称": remove_prohibited_str(name), + "初始星级": star, + } + pretty_info[member_dict["名称"]] = member_dict + self.dump_data(pretty_info) + logger.info(f"{self.game_name_cn} 更新成功") + # pretty_card.json + pretty_card_info = {} + url = "https://wiki.biligame.com/umamusume/支援卡图鉴" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 卡牌出错") + else: + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath("//table[@id='CardSelectTr']/tbody/tr") + for char in char_list: + try: + name = char.xpath("./td[1]/div/a/@title")[0] + name_cn = char.xpath("./td[3]/a/text()")[0] + avatar = char.xpath("./td[1]/div/a/img/@srcset")[0] + star = str(char.xpath("./td[5]/text()")[0]).strip() + sources = str(char.xpath("./td[7]/text()")[0]).strip() + except IndexError: + continue + member_dict = { + "头像": unquote(str(avatar).split(" ")[-2]), + "名称": remove_prohibited_str(name), + "中文名": remove_prohibited_str(name_cn), + "稀有度": star, + "获取方式": [sources] if sources else [], + } + pretty_card_info[member_dict["中文名"]] = member_dict + self.dump_data(pretty_card_info, "pretty_card.json") + logger.info(f"{self.game_name_cn} 卡牌更新成功") + # 下载头像 + for value in pretty_info.values(): + await self.download_img(value["头像"], value["名称"]) + for value in pretty_card_info.values(): + await self.download_img(value["头像"], value["中文名"]) + # 下载星星 + PRETTY_URL = "https://patchwiki.biligame.com/images/umamusume" + await self.download_img( + PRETTY_URL + "/1/13/e1hwjz4vmhtvk8wlyb7c0x3ld1s2ata.png", "star" + ) + # 下载稀有度标志 + idx = 1 + for url in [ + "/f/f7/afqs7h4snmvovsrlifq5ib8vlpu2wvk.png", + "/3/3b/d1jmpwrsk4irkes1gdvoos4ic6rmuht.png", + "/0/06/q23szwkbtd7pfkqrk3wcjlxxt9z595o.png", + ]: + await self.download_img(PRETTY_URL + url, f"{idx}_label") + idx += 1 + await self.update_up_char() + + async def update_up_char(self): + announcement_url = "https://wiki.biligame.com/umamusume/公告" + result = await self.get_url(announcement_url) + if not result: + logger.warning(f"{self.game_name_cn}获取公告出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + announcements = dom.xpath("//div[@id='mw-content-text']/div/div/span/a") + title = "" + url = "" + for announcement in announcements: + try: + title = announcement.xpath("./@title")[0] + url = "https://wiki.biligame.com/" + announcement.xpath("./@href")[0] + if re.match(r".*?\d{8}$", title) or re.match( + r"^\d{1,2}月\d{1,2}日.*?", title + ): + break + except IndexError: + continue + if not title: + logger.warning(f"{self.game_name_cn}未找到新UP公告") + return + result = await self.get_url(url) + if not result: + logger.warning(f"{self.game_name_cn}获取UP公告出错") + return + try: + start_time = None + end_time = None + char_img = "" + card_img = "" + up_chars = [] + up_cards = [] + soup = BeautifulSoup(result, "lxml") + heads = soup.find_all("span", {"class": "mw-headline"}) + for head in heads: + if "时间" in head.text: + time = head.find_next("p").text.split("\n")[0] + if "～" in time: + start, end = time.split("～") + start_time = dateparser.parse(start) + end_time = dateparser.parse(end) + elif "赛马娘" in head.text: + char_img = head.find_next("a", {"class": "image"}).find("img")[ + "src" + ] + lines = str(head.find_next("p").text).split("\n") + chars = [ + line + for line in lines + if "★" in line and "（" in line and "）" in line + ] + for char in chars: + star = char.count("★") + name = re.split(r"[（）]", char)[-2].strip() + up_chars.append( + UpChar(name=name, star=star, limited=False, zoom=70) + ) + elif "支援卡" in head.text: + card_img = head.find_next("a", {"class": "image"}).find("img")[ + "src" + ] + lines = str(head.find_next("p").text).split("\n") + cards = [ + line + for line in lines + if "R" in line and "（" in line and "）" in line + ] + for card in cards: + star = 3 if "SSR" in card else 2 if "SR" in card else 1 + name = re.split(r"[（）]", card)[-2].strip() + up_cards.append( + UpChar(name=name, star=star, limited=False, zoom=70) + ) + if start_time and end_time: + if start_time <= datetime.now() <= end_time: + self.UP_CHAR = UpEvent( + title=title, + pool_img=char_img, + start_time=start_time, + end_time=end_time, + up_char=up_chars, + ) + self.UP_CARD = UpEvent( + title=title, + pool_img=card_img, + start_time=start_time, + end_time=end_time, + up_char=up_cards, + ) + self.dump_up_char() + logger.info(f"成功获取{self.game_name_cn}当前up信息...当前up池: {title}") + except Exception as e: + logger.warning(f"{self.game_name_cn}UP更新出错 {type(e)}：{e}") + + async def _reload_pool(self) -> Optional[Message]: + await self.update_up_char() + self.load_up_char() + if self.UP_CHAR and self.UP_CARD: + return Message( + Message.template("重载成功！\n当前UP池子：{}{:image}{:image}").format( + self.UP_CHAR.title, + self.UP_CHAR.pool_img, + self.UP_CARD.pool_img, + ) + ) diff --git a/plugins/draw_card/handles/prts_handle.py b/plugins/draw_card/handles/prts_handle.py new file mode 100644 index 00000000..728b0a57 --- /dev/null +++ b/plugins/draw_card/handles/prts_handle.py @@ -0,0 +1,292 @@ +import re +import random +import dateparser +from lxml import etree +from PIL import ImageDraw +from bs4 import BeautifulSoup +from datetime import datetime +from urllib.parse import unquote +from typing import List, Optional, Tuple +from pydantic import ValidationError +from nonebot.adapters.onebot.v11 import Message +from utils.message_builder import image +from nonebot.log import logger +import asyncio + +try: + import ujson as json +except ModuleNotFoundError: + import json + +from .base_handle import BaseHandle, BaseData, UpChar, UpEvent +from ..config import draw_config +from ..util import remove_prohibited_str, cn2py, load_font +from utils.image_utils import BuildImage + + +class Operator(BaseData): + recruit_only: bool # 公招限定 + event_only: bool # 活动获得干员 + # special_only: bool # 升变/异格干员 + + +class PrtsHandle(BaseHandle[Operator]): + def __init__(self): + super().__init__("prts", "明日方舟", "#eff2f5") + self.max_star = 6 + self.config = draw_config.prts + + self.ALL_OPERATOR: List[Operator] = [] + self.UP_EVENT: Optional[UpEvent] = None + + def get_card(self, add: float) -> Operator: + star = self.get_star( + [6, 5, 4, 3], + [ + self.config.PRTS_SIX_P + add, + self.config.PRTS_FIVE_P, + self.config.PRTS_FOUR_P, + self.config.PRTS_THREE_P, + ], + ) + + all_operators = [ + x + for x in self.ALL_OPERATOR + if x.star == star and not any([x.limited, x.event_only, x.recruit_only]) + ] + acquire_operator = None + + if self.UP_EVENT: + up_operators = [x for x in self.UP_EVENT.up_char if x.star == star] + # UPs + try: + zooms = [x.zoom for x in up_operators] + zoom_sum = sum(zooms) + if random.random() < zoom_sum: + up_name = random.choices(up_operators, weights=zooms, k=1)[0].name + acquire_operator = [ + x for x in self.ALL_OPERATOR if x.name == up_name + ][0] + except IndexError: + pass + if not acquire_operator: + acquire_operator = random.choice(all_operators) + return acquire_operator + + def get_cards(self, count: int) -> List[Tuple[Operator, int]]: + card_list = [] # 获取所有角色 + add = 0.0 + count_idx = 0 + for i in range(count): + count_idx += 1 + card = self.get_card(add) + if card.star == self.max_star: + add = 0.0 + count_idx = 0 + elif count_idx > 50: + add += 0.02 + card_list.append((card, i + 1)) + return card_list + + def format_pool_info(self) -> str: + info = "" + if self.UP_EVENT: + star6_list = [x.name for x in self.UP_EVENT.up_char if x.star == 6] + star5_list = [x.name for x in self.UP_EVENT.up_char if x.star == 5] + star4_list = [x.name for x in self.UP_EVENT.up_char if x.star == 4] + if star6_list: + info += f"六星UP：{' '.join(star6_list)}\n" + if star5_list: + info += f"五星UP：{' '.join(star5_list)}\n" + if star4_list: + info += f"四星UP：{' '.join(star4_list)}\n" + info = f"当前up池: {self.UP_EVENT.title}\n{info}" + return info.strip() + + 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)] + 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() + return pool_info + image(b64=self.generate_img(cards).pic2bs4()) + result + + def generate_card_img(self, card: Operator) -> BuildImage: + sep_w = 5 + sep_h = 5 + star_h = 15 + img_w = 120 + img_h = 120 + font_h = 20 + bg = BuildImage(img_w + sep_w * 2, img_h + font_h + sep_h * 2, color="#EFF2F5") + star_path = str(self.img_path / "star.png") + star = BuildImage(star_h, star_h, background=star_path) + img_path = str(self.img_path / f"{cn2py(card.name)}.png") + img = BuildImage(img_w, img_h, background=img_path) + bg.paste(img, (sep_w, sep_h), alpha=True) + for i in range(card.star): + bg.paste(star, (sep_w + img_w - 5 - star_h * (i + 1), sep_h), alpha=True) + # 加名字 + text = card.name[:7] + "..." if len(card.name) > 8 else card.name + font = load_font(fontsize=16) + text_w, text_h = font.getsize(text) + draw = ImageDraw.Draw(bg.markImg) + draw.text( + (sep_w + (img_w - text_w) / 2, sep_h + img_h + (font_h - text_h) / 2), + text, + font=font, + fill="gray", + ) + return bg + + def _init_data(self): + self.ALL_OPERATOR = [ + Operator( + name=value["名称"], + star=int(value["星级"]), + limited="干员寻访" not in value["获取途径"], + recruit_only=True + if "干员寻访" not in value["获取途径"] and "公开招募" in value["获取途径"] + else False, + event_only=True if "活动获取" in value["获取途径"] else False, + ) + for key, value in self.load_data().items() + if "阿米娅" not in key + ] + self.load_up_char() + + def load_up_char(self): + try: + data = self.load_data(f"draw_card_up/{self.game_name}_up_char.json") + self.UP_EVENT = UpEvent.parse_obj(data.get("char", {})) + except ValidationError: + logger.warning(f"{self.game_name}_up_char 解析出错") + + def dump_up_char(self): + if self.UP_EVENT: + data = {"char": json.loads(self.UP_EVENT.json())} + self.dump_data(data, f"draw_card_up/{self.game_name}_up_char.json") + + async def _update_info(self): + info = {} + url = "https://wiki.biligame.com/arknights/干员数据表" + result = await self.get_url(url) + if not result: + logger.warning(f"更新 {self.game_name_cn} 出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + char_list = dom.xpath("//table[@id='CardSelectTr']/tbody/tr") + for char in char_list: + try: + 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") + 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], + } + info[member_dict["名称"]] = member_dict + self.dump_data(info) + logger.info(f"{self.game_name_cn} 更新成功") + # 下载头像 + for value in info.values(): + await self.download_img(value["头像"], value["名称"]) + # 下载星星 + await self.download_img( + "https://patchwiki.biligame.com/images/pcr/0/02/s75ys2ecqhu2xbdw1wf1v9ccscnvi5g.png", + "star", + ) + await self.update_up_char() + + async def update_up_char(self): + announcement_url = "https://ak.hypergryph.com/news.html" + result = await self.get_url(announcement_url) + if not result: + logger.warning(f"{self.game_name_cn}获取公告出错") + return + dom = etree.HTML(result, etree.HTMLParser()) + activity_urls = dom.xpath( + "//ol[@class='articleList' and @data-category-key='ACTIVITY']/li/a/@href" + ) + start_time = None + end_time = None + up_chars = [] + pool_img = "" + title = "" + for activity_url in activity_urls: + activity_url = f"https://ak.hypergryph.com{activity_url}" + result = await self.get_url(activity_url) + if not result: + logger.warning(f"{self.game_name_cn}获取公告 {activity_url} 出错") + continue + soup = BeautifulSoup(result, "lxml") + contents = soup.find_all("p") + for index, content in enumerate(contents): + if re.search("(.*)(寻访|复刻).*?开启", content.text): + title = content.text + if "【" in title and "】" in title: + title = re.split(r"[【】]", title)[1] + lines = [str(contents[index + i + 1].text) for i in range(5)] + time = "" + chars: List[str] = [] + for line in lines: + match = re.search( + r"(\d{1,2}月\d{1,2}日.*?-.*?\d{1,2}月\d{1,2}日.*?$)", line + ) + if match: + time = match.group(1) + if "★" in line: + chars.append(line) + if not time: + continue + start, end = time.replace("月", "/").replace("日", "").split("-")[:2] + start_time = dateparser.parse(start) + end_time = dateparser.parse(end) + pool_img = content.find_previous("img")["src"] + for char in chars: + star = char.split("（")[0].count("★") + name = re.split(r"[：（]", char)[1] + names = name.split("/") if "/" in name else [name] + names = [name.replace("[限定]", "").strip() for name in names] + if "权值" in char: + match = re.search(r"（在.*?以.*?(\d+).*?倍权值.*?）", char) + else: + match = re.search(r"（占.*?的.*?(\d+).*?%）", char) + zoom = 1 + if match: + zoom = float(match.group(1)) + zoom = zoom / 100 if zoom > 10 else zoom + for name in names: + up_chars.append( + UpChar(name=name, star=star, limited=False, zoom=zoom) + ) + break + if title and start_time and end_time: + if start_time <= datetime.now() <= end_time: + self.UP_EVENT = UpEvent( + title=title, + pool_img=pool_img, + start_time=start_time, + end_time=end_time, + up_char=up_chars, + ) + self.dump_up_char() + logger.info(f"成功获取{self.game_name_cn}当前up信息...当前up池: {title}") + break + + async def _reload_pool(self) -> Optional[Message]: + await self.update_up_char() + self.load_up_char() + if self.UP_EVENT: + return f"重载成功！\n当前UP池子：{self.UP_EVENT.title}" + image( + self.UP_EVENT.pool_img + )