mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 14:22:55 +08:00
288 lines
11 KiB
Python
288 lines
11 KiB
Python
import contextlib
|
||
import random
|
||
import dateparser
|
||
from lxml import etree
|
||
from typing import List, Optional, Tuple
|
||
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
|
||
|
||
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] if self.UP_EVENT else []
|
||
)
|
||
# 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)):
|
||
with contextlib.suppress(IndexError):
|
||
up_zoom.append((cur_, cur_ + up_ship[i + 1].zoom / 100))
|
||
cur_ += up_ship[i + 1].zoom / 100
|
||
rand = random.random()
|
||
# 抽取up
|
||
for i, zoom in enumerate(up_zoom):
|
||
if zoom[0] <= rand <= zoom[1]:
|
||
with contextlib.suppress(IndexError):
|
||
acquire_char = [x for x in self.ALL_CHAR if x.name == up_ship[i].name][0]
|
||
# 没有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:
|
||
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='mw-body-content mw-content-ltr']/div[@class='mw-parser-output']"
|
||
)
|
||
for index, content in enumerate(contents):
|
||
char_list = content.xpath("./div[@id='CardSelectTr']/div")
|
||
for char in char_list:
|
||
try:
|
||
name = char.xpath("./div/a/@title")[0]
|
||
frame = char.xpath("./div/div/a/img/@alt")[0]
|
||
avatar = char.xpath("./div/a/img/@srcset")[0]
|
||
except IndexError:
|
||
continue
|
||
member_dict = {
|
||
"名称": remove_prohibited_str(name),
|
||
"头像": unquote(str(avatar).split(" ")[-2]),
|
||
"星级": self.parse_star(frame),
|
||
"类型": char.xpath("./@data-param1")[0].split(",")[1],
|
||
}
|
||
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
|
||
|
||
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, ValueError):
|
||
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}")
|