HeXi/hexi/plugins/nonebot_plugin_bf_bot/__init__.py

from nonebot import on_command, require
from nonebot.adapters import Message, Event, Bot
from nonebot.log import logger
from nonebot.matcher import Matcher
from nonebot.params import CommandArg
from nonebot.plugin import PluginMetadata
from nonebot.rule import to_me
import time
from .user_data.data_utils import UserManager

require("nonebot_plugin_alconna")
from nonebot_plugin_alconna import UniMessage
from .data import *
from .data_utils import *
from .image_builder import *
from .text_utils import *

__plugin_meta__ = PluginMetadata(
    name="BF查询",
    description="战地3，4，1，5，2042，6",
    usage="""
        bf3: /bf3 EAID (查询BF3数据)
        bf4: /bf4 EAID (查询BF4数据)
        bf1: /bf1 EAID (查询BF5数据)
        bfv: /bfv EAID (查询BF6数据)
        bf2042: /bf2042 EAID (查询BF2042数据)
        bf6: /bf6 EAID (查询BF6数据)
        绑定: /绑定 EAID (绑定你的QQ与EAID)
        解绑: /解绑 (解除你的QQ与当前绑定的EAID)
        修改绑定: /修改绑定 EAID (修改你的QQ与当前绑定的EAID)
        """.strip(),
    type="application",
    extra={

    },
)

query = on_command("bft", rule=to_me(), aliases={"bf3", "bf4", "bfv", "bf1", "bf2042", "bf6"}, block=True)
bind = on_command("bfbind", rule=to_me(), aliases={"绑定"}, block=True)
unbind = on_command("unbind", rule=to_me(), aliases={"解绑"}, block=True)
update_bind = on_command("update_bind", rule=to_me(), aliases={"修改绑定"}, block=True)

bf_dict = {
    "bf3": "战地3",
    "bf4": "战地4",
    "bf1": "战地1",
    "bfv": "战地5",
    "bf2042": "战地2042",
    "bf6": "战地6",
}


@query.handle()
async def handle_function(event: Event, matcher: Matcher, msg: Message = CommandArg()):
    start_time = time.time()
    usermanager = UserManager()
    cmd = matcher.state["_prefix"]["command"][0]
    game = cmd
    input_text = msg.extract_plain_text().strip()
    logger.info(f"{type(input_text)}, {repr(input_text)}")
    if input_text is None or input_text == "":
        content = usermanager.get_user_by_qq(event.get_user_id())
        if not content:
            await UniMessage.at(user_id=event.get_user_id()).text(
                "未检测到id，也未检测到绑定记录，请使用 绑定 EAID 进行绑定操作").send()
            return
        player_id = content['ea_player_id']
        user_id = content['ea_user_id']
        player = content['ea_player_name']
        platform = "pc"
    else:
        content = await get_player_info_by_name(input_text, "pc")
        if 'errors' in content:
            await UniMessage.at(user_id=event.get_user_id()).text("未找到玩家，请检查是否拼写正确").send()
            return
        player_id = content['personaId']
        user_id = content['userId']
        player = content.get('personaName') or content.get('displayName') or content.get('username', 'error')
        if 'errors' in player:
            await UniMessage.text(f"请检查id是否正确！").send()
            return
        platform = "pc"
    await UniMessage.text(f"正在查询 {player} 的 {bf_dict[cmd]} 数据，请耐心等待").send()
    play_stat = ""
    try:
        if cmd == "bf3":
            play_stat = await get_data_bf3(player, player_id, user_id, platform)
        elif cmd == "bf4":
            play_stat = await get_data_bf4(player, player_id, user_id, platform)
        elif cmd == "bf1":
            play_stat = await get_data_bf1(player, player_id, user_id, platform)
        elif cmd == "bfv":
            play_stat = await get_data_bfv(player, player_id, user_id, platform)
        elif cmd == "bf2042":
            await UniMessage.at(event.get_user_id()).text("正在开发中，敬请期待！").send()
        elif cmd == "bf6":
            play_stat, rank = await asyncio.gather(
                get_data_bf6(player, player_id, user_id, platform),
                get_bf6_rank(player)
            )
            logger.info(f"等级：{rank}")
        else:
            await UniMessage.at(event.get_user_id()).text("指令异常").send()
            return
        if "errors" in play_stat:
            logger.warning(play_stat['errors'][0])
            msg = play_stat['errors'][0]
            await UniMessage.text(f"{msg}").send()
            return
        else:
            # logger.info(f"结果{play_stat}")
            # player = play_stat['userName']
            weapon, vehicle = await get_best_weapon_and_best_vehicle(play_stat)
            pid = play_stat['userId']
            kd = play_stat['killDeath']
            kpm = play_stat['killsPerMinute']
            acc = play_stat['accuracy']
            # 战地3 特化
            if cmd == 'bf3':
                head_shots = play_stat['headShots']
            else:
                head_shots = play_stat['headshots']
            time_play = convert_to_hours(play_stat['timePlayed'])
            logger.info(f"游玩时长监测：{time_play}")
            if time_play == 0:
                await UniMessage.at(event.get_user_id()).text("查询接口异常，请等待一段时间后查询").send()
                return
            kills = int(play_stat['kills'])
            kill_assists = int(play_stat['killAssists'])
            revives = int(play_stat['revives'])
            wins = int(play_stat['wins'])
            loses = int(play_stat['loses'])
            best_weapon = weapon
            best_vehicle = vehicle
            best_class = play_stat['bestClass']
            destroyed = await get_vehicle_destroyed(play_stat["vehicles"])
            if cmd == 'bf6':
                best_class = await get_best_classes(play_stat)
                captured = play_stat['objective']['captured']
                score = sum(item.get("score", 0) for item in play_stat['classes'])
                # 计算4兵种游玩时长
                seconds = sum(item.get("secondsPlayed", 0) for item in play_stat['classes'])
                kpm = round(kills / (int(seconds) / 60), 2)
                logger.info(f"击杀数：{kills}，游玩分钟数：{seconds / 60}，计算kpm值：{kpm}")
                spm = int(score / (int(seconds) / 60))
                logger.info(f"计算spm值：{spm}")
                repairs = play_stat['repairs']
                img = await build_bf6_stats_card(game='bf6',
                                                 qq_id=event.get_user_id(),
                                                 player=player,
                                                 pid=pid,
                                                 rank=rank,
                                                 kd=kd,
                                                 kpm=kpm,
                                                 spm=spm,
                                                 acc=acc,
                                                 head_shots=head_shots,
                                                 time_play=time_play,
                                                 kills=kills,
                                                 kill_assists=kill_assists,
                                                 revives=revives,
                                                 repairs=repairs,
                                                 captured=captured,
                                                 score=score,
                                                 wins=wins,
                                                 loses=loses,
                                                 destroyed=destroyed,
                                                 best_weapon=best_weapon,
                                                 best_vehicle=best_vehicle,
                                                 best_class=best_class)
            else:
                spm = play_stat['scorePerMinute']
                rank = play_stat['rank']
                longest_head_shot = play_stat['longestHeadShot']
                highest_ill_streak = play_stat['highestKillStreak']
                stat_data, level_designation = await stats_calculator(play_stat, cmd)
                img = await build_stats_card(game, player, pid, kd, kpm, spm, acc, head_shots, rank, time_play, kills,
                                             kill_assists, revives, wins, loses, destroyed, best_weapon, best_vehicle,
                                             best_class,
                                             longest_head_shot, highest_ill_streak, stat_data, level_designation)
            end_time = time.time()
            duration = round(end_time - start_time, 3)
            await UniMessage.image(raw=img.getvalue()).at(user_id=event.get_user_id()).text(
                f"本次查询耗时：{duration}s").send()
    except Exception as e:
        logger.exception(f"异常：{e}")


@bind.handle()
async def bind_user(event: Event, matcher: Matcher, msg: Message = CommandArg()):
    user_manager = UserManager()
    player_name = msg.extract_plain_text()
    qq_id = event.get_user_id()

    # 执行已有数据查询操作
    user_record = user_manager.get_user_by_qq(qq_id)
    if user_record:
        user_info = await get_player_info_by_ea_id(user_record['ea_player_id'], user_record['ea_user_id'], 'pc')
        await (UniMessage.at(user_id=event.get_user_id())
               .text(f"您已经将账号绑定到: {user_info['personaName']}，如果你需要更改绑定请发送 修改绑定 EAID ").send())
        return
    else:
        user_info_json = await get_player_info_by_name(player_name, "pc")
        if 'errors' in user_info_json:
            await UniMessage.at(user_id=event.get_user_id()).text("未找到玩家，请检查是否拼写正确").send()
        player_id = user_info_json['personaId']
        user_id = user_info_json['userId']
        player_name = user_info_json.get('personaName') or user_info_json.get('displayName') or user_info_json.get(
            'username', 'error')
        avatar_url = user_info_json['avatar']
        logger.info(f"QQ号：{qq_id}，EA双ID：{player_id}-{user_id}")

        # 检测ea账号 是否被绑定至其他qq
        bind_record = user_manager.get_user_by_ea_id(user_id)
        if bind_record:
            await (UniMessage.at(user_id=event.get_user_id())
                   .text(
                f"您的EA账号 {bind_record['ea_player_name']} 已经绑定到: {bind_record['qq_id']}，请您确认绑定的EAID").send())
            return
        else:
            # 执行绑定操作
            record = user_manager.add_user(qq_id, player_name, player_id, user_id, "[]", "[]")
            logger.info(f"插入数据返回结果：{record}")
            if record > 0:
                await UniMessage.at(user_id=event.get_user_id()).text(f"您已经成功绑定至: {player_name}").send()


@unbind.handle()
async def bind_user(event: Event, matcher: Matcher, msg: Message = CommandArg()):
    user_manager = UserManager()
    qq_id = event.get_user_id()

    # 执行已有数据查询操作
    user_record = user_manager.get_user_by_qq(qq_id)
    if user_record:
        user_info = await get_player_info_by_ea_id(user_record['ea_player_id'], user_record['ea_user_id'], 'pc')
        rows = user_manager.delete_user_by_qq(qq_id)
        if rows > 0:
            await (UniMessage.at(user_id=event.get_user_id())
                   .text(
                f"您已经与: {user_info['personaName']} 成功解除绑定，如果你需要重新绑定请发送 修改绑定 EAID ").send())
            return
    else:
        await UniMessage.at(user_id=event.get_user_id()).text(f"未查询到绑定记录").send()


@update_bind.handle()
async def bind_user(event: Event, matcher: Matcher, msg: Message = CommandArg()):
    user_manager = UserManager()
    player_name = msg.extract_plain_text()
    qq_id = event.get_user_id()

    # 执行已有数据查询操作
    user_record = user_manager.get_user_by_qq(qq_id)
    if user_record:
        user_info_json = await get_player_info_by_name(player_name, "pc")
        if 'errors' in user_info_json:
            await UniMessage.at(user_id=event.get_user_id()).text("未找到玩家，请检查是否拼写正确").send()
        player_id = user_info_json['personaId']
        user_id = user_info_json['userId']
        player_name = user_info_json.get('personaName') or user_info_json.get('displayName') or user_info_json.get(
            'username', 'error')
        avatar_url = user_info_json['avatar']
        logger.info(f"QQ号：{qq_id}，EA双ID：{player_id}-{user_id}")

        # 执行修改绑定操作
        record = user_manager.update_user(qq_id, player_name, player_id, user_id, "[]", "[]")
        logger.info(f"修改数据返回结果：{record}")
        if record > 0 and record == 1:
            await UniMessage.at(user_id=event.get_user_id()).text(
                f"您已经成功从：{user_record['ea_player_name']}修改绑定至: {player_name}").send()
    else:
        await UniMessage.at(user_id=event.get_user_id()).text(
            f"未查询到绑定记录: {player_name}，如果需要绑定，请@机器人并发送  绑定 {player_name} ").send()