"""查询编排：目标解析 → 取数 → 加工 → 渲染，返回给 Trigger 层发送的结果。 只产出结果对象（文本/图片字节），不直接发消息；"正在查询…"等进度提示通过 `notify` 回调交给 Trigger 层发送，本层保持与 nonebot 消息 API 解耦。 """ from __future__ import annotations import asyncio import time from dataclasses import dataclass from collections.abc import Awaitable, Callable from nonebot import logger from ..repository import Bf2042SettingsManager, UserManager from ..utils.args import ( BF2042_ERROR_HINT, BF_GAME_NAMES, parse_bf2042_args, ) from ..utils.render_bf2042 import build_bf2042_career_card, build_bf2042_list_card from ..utils.render_ea import build_bf6_stats_card, build_stats_card from ..utils.text import convert_to_hours from .fetchers.bf2042 import ( bf2042_list_title, get_bf2042_gamemodes, get_bf2042_info, get_bf2042_list, ) from .fetchers.gametools import ( fetch_all_stats, get_bf6_rank, get_player_info_by_name, player_display_name, ) from .processing import ( get_best_classes, get_best_weapon_and_best_vehicle, get_vehicle_destroyed, stats_calculator, ) Notify = Callable[[str], Awaitable[None]] _PLATFORM = "pc" @dataclass class QueryReply: """一次查询的最终回复：文本 +（可选）PNG 图片字节。""" text: str image: bytes | None = None async def run_query( *, cmd: str, input_text: str, qq_id: str, notify: Notify ) -> QueryReply: """`/bft` 及全部别名共用的查询入口。 cmd: 实际输入的游戏别名（bf3/bf4/bf1/bfv/bf2042/bf6）。 input_text: 命令参数（EAID 或 BF2042 子命令），可为空。 qq_id: 触发者 QQ，用于读绑定与取头像。 notify: 进度提示回调（"正在查询…"）。 """ started = time.time() if cmd not in BF_GAME_NAMES: return QueryReply( "指令异常：请使用 bf3 / bf4 / bf1 / bfv / bf2042 / bf6（可带 EAID）" ) if cmd == "bf2042": reply = await _run_bf2042(input_text, qq_id, notify) else: reply = await _run_ea(cmd, input_text, qq_id, notify) if reply.image is not None: reply.text = f"本次查询耗时：{round(time.time() - started, 3)}s" return reply def _resolve_bound_record(qq_id: str) -> QueryReply | dict: """无参数查询时取绑定记录；未绑定返回给用户的提示。""" content = UserManager().get_user_by_qq(qq_id) if not content: return QueryReply( "未检测到id，也未检测到绑定记录，请使用 绑定 EAID 进行绑定操作" ) return content async def _run_ea(cmd: str, input_text: str, qq_id: str, notify: Notify) -> QueryReply: """EA 系（BF3/4/1/5/6）：gametools 取数 + 卡片渲染。""" if not input_text: content = _resolve_bound_record(qq_id) if isinstance(content, QueryReply): return content player = content["ea_player_name"] player_id = content["ea_player_id"] user_id = content["ea_user_id"] else: content = await get_player_info_by_name(input_text, _PLATFORM) if "errors" in content: return QueryReply("未找到玩家，请检查是否拼写正确") player_id = content["personaId"] user_id = content["userId"] player = player_display_name(content) await notify(f"正在查询 {player} 的 {BF_GAME_NAMES[cmd]} 数据，请耐心等待") try: rank = 0 if cmd == "bf6": play_stat, rank = await asyncio.gather( fetch_all_stats(cmd, player, player_id, user_id, _PLATFORM), get_bf6_rank(player, player_id, user_id, _PLATFORM), ) logger.info(f"等级：{rank}") else: play_stat = await fetch_all_stats( cmd, player, player_id, user_id, _PLATFORM ) if "errors" in play_stat: logger.warning(play_stat["errors"][0]) return QueryReply(str(play_stat["errors"][0])) img = await _render_ea_card(cmd, play_stat, rank, player, qq_id) if isinstance(img, QueryReply): return img return QueryReply("", image=img) except Exception as e: # noqa: BLE001 - 兜底给用户可见提示，明细写日志 logger.exception(f"异常：{e}") return QueryReply("查询异常，请稍后再试") async def _render_ea_card(cmd: str, play_stat, rank, player, qq_id): """渲染卡片；返回 PNG 字节，或对用户的错误提示（QueryReply）。""" 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: return QueryReply("查询接口异常，请等待一段时间后查询") 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}") img = await build_bf6_stats_card( game="bf6", qq_id=qq_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=play_stat["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"] 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( cmd, player, pid, kd, kpm, spm, acc, head_shots, play_stat["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, qq_id=qq_id, ) if img is None: return QueryReply("图片渲染失败，请稍后再试") return img.getvalue() async def _run_bf2042(input_text: str, qq_id: str, notify: Notify) -> QueryReply: """BF2042 独立流程：tracker.gg 浏览器通道。 与其它战地不同，BF2042 不经过 EA 接口，所以玩家名要么来自指令参数， 要么来自绑定记录，不需要 personaId/userId。 指令形式: /bf2042 [EAID] 生涯卡 /bf2042 武器|载具|装备 [EAID] 榜单卡 /bf2042 模式 [EAID] 各模式明细 """ target = parse_bf2042_args(input_text) player = target.player if not player: content = _resolve_bound_record(qq_id) if isinstance(content, QueryReply): return content player = content["ea_player_name"] if target.view == "list": what = f"{bf2042_list_title(target.kind)}榜" elif target.view == "gamemode": what = "各模式数据" else: what = "数据" await notify(f"正在查询 {player} 的 {BF_GAME_NAMES['bf2042']}{what}，请耐心等待") try: if target.view == "list": result = await get_bf2042_list(player, _PLATFORM, target.kind) elif target.view == "gamemode": result = await get_bf2042_gamemodes(player, _PLATFORM) else: result = await get_bf2042_info(player, _PLATFORM) except Exception as e: # noqa: BLE001 - 兜底,别让异常静默 logger.exception(f"BF2042 查询异常：{e}") return QueryReply("BF2042 查询异常，请稍后再试") if not result.ok: logger.warning(f"BF2042 查询失败({result.reason})：{result.detail}") hint = BF2042_ERROR_HINT.get(result.reason, "查询失败，请稍后再试") return QueryReply(f"BF2042 {hint}") info = result.data if isinstance(result.data, dict) else {} if target.view == "career": # 新版生涯卡:旧版布局 + tracker.gg 数据映射;背景按支持者特权选 from .processing import build_bf2042_career_card as build_card_slots try: support = Bf2042SettingsManager().is_support(qq_id) except Exception: # noqa: BLE001 - 装饰性功能,失败退回公共背景 support = False card = build_card_slots( info.get("_profile") or {}, info.get("_weapons") or [], info.get("_vehicles") or [], info.get("_soldiers") or [], ) card["_qq_id"] = qq_id card["_support"] = support img = build_bf2042_career_card(card) else: img = build_bf2042_list_card(info) return QueryReply("", image=img.getvalue())