"""BF2042 战绩取数(tracker.gg,直连主通道 / 扩展桥兜底)。 取数调度全部走 ``tracker_edge.fetch_bf2042_segments``:主通道是 ``tracker_api`` 的无浏览器直连(curl_cffi + TRN Bearer 令牌),浏览器扩展桥仅作兜底;本模块 只做"拿数据 → 防御式解析 → 卡片字段",对通道无感知。 响应结构实测: - 一次取数会带来 5 个关键响应: - ``profile/{platform}/{name}`` → ``{"data": {platformInfo, segments:[overview, gamemode...]}}`` - ``.../segments/weapon`` → ``{"data": [段数组]}`` - ``.../segments/vehicle`` → ``{"data": [段数组]}`` - ``.../segments/gadget`` → 装备(榜单视图用) - ``.../segments/soldier`` → 兵种(卡片用) - overview 里没有 SPM / 命中率 / 分数,所以卡片改展示真实存在的 「每分钟伤害」「场均击杀」。 解析全部防御式:字段缺失退化成占位符,不抛异常。 """ import re from nonebot import logger from .tracker_edge import TrackerResult, fetch_bf2042_segments __all__ = [ "get_bf2042_info", "get_bf2042_list", "get_bf2042_gamemodes", "bf2042_list_title", ] _MISSING = "—" # 用户层用 "pc",tracker.gg 用平台 slug _PLATFORM_ALIASES = { "pc": "origin", "origin": "origin", "steam": "steam", "xbl": "xbl", "psn": "psn", } def _platform_slug(platform: str) -> str: return _PLATFORM_ALIASES.get((platform or "").lower(), "origin") # --------------------------------------------------------------------------- # # 取值工具:全部容忍缺失 # --------------------------------------------------------------------------- # def _stat(stats: dict, *keys) -> dict | None: """返回第一个命中的 stat。tracker 的 stat 形如 {"value":..,"displayValue":..}。""" for key in keys: entry = stats.get(key) if isinstance(entry, dict): return entry if entry is not None: return {"value": entry, "displayValue": entry} return None def _value(entry: dict | None): return entry.get("value") if entry else None def _display(entry: dict | None, default=_MISSING): """优先用 displayValue(带单位/百分比,免去单位猜测)。""" if not entry: return default display = entry.get("displayValue") if display is not None: return display value = entry.get("value") return default if value is None else value def _number(entry: dict | None, default=_MISSING): """数值型字段,取不到时给 0(计数类字段缺省即 0)。""" value = _value(entry) return default if value is None else value def _segment(segments: list[dict], seg_type: str) -> dict: for seg in segments or []: if seg.get("type") == seg_type: return seg return {} def _best_of(segments: list[dict], seg_type: str) -> dict: """从同类 segment 里挑击杀最高的那个(武器/载具)。""" candidates = [s for s in (segments or []) if s.get("type") == seg_type] if not candidates: return {} def _kills(seg: dict) -> float: return _value(_stat(seg.get("stats") or {}, "kills")) or 0 return max(candidates, key=_kills) def _name_of(seg: dict) -> str: meta = seg.get("metadata") or {} # metadata.name 常带尾随空格(如 "PKP-BP ") name = meta.get("displayName") or meta.get("name") or "" return name.strip() or _MISSING # --------------------------------------------------------------------------- # # 解析 # --------------------------------------------------------------------------- # def normalize_bf2042(profile: dict, extra_segments: list[dict] | None = None) -> dict: """把 tracker.gg profile 响应整理成文字卡需要的字段。 extra_segments: 由 /segments/weapon、/segments/vehicle 另取的段,拼进来一起挑最佳。 """ data = (profile or {}).get("data") or {} segments = list(data.get("segments") or []) + list(extra_segments or []) platform_info = data.get("platformInfo") or {} overview = (_segment(segments, "overview").get("stats")) or {} wins = _value(_stat(overview, "wins")) or 0 losses = _value(_stat(overview, "losses")) or 0 matches = _value(_stat(overview, "matches")) or (wins + losses) best_weapon_seg = _best_of(segments, "weapon") best_vehicle_seg = _best_of(segments, "vehicle") def _sub(seg: dict) -> dict: stats = seg.get("stats") or {} return { "名称": _name_of(seg), "类别": (seg.get("metadata") or {}).get("category") or _MISSING, "击杀": _number(_stat(stats, "kills")), "KPM": _value(_stat(stats, "killsPerMinute")) or _MISSING, "时长": _display(_stat(stats, "timePlayed")), "命中率": _display(_stat(stats, "shotsAccuracy")), "爆头率": _display(_stat(stats, "headshotPercentage")), "摧毁": _number(_stat(stats, "destroyed")), # 仅载具有 } return { "玩家名称": platform_info.get("platformUserHandle") or _MISSING, "游玩平台": platform_info.get("platformSlug") or _MISSING, "游戏等级": _display(_stat(overview, "level")), "游玩时长": _display(_stat(overview, "timePlayed")), "游玩场次": matches or _MISSING, "对局胜率": _display(_stat(overview, "wlPercentage")), "玩家K/D": _value(_stat(overview, "kdRatio")) or _MISSING, "玩家KPM": _value(_stat(overview, "killsPerMinute")) or _MISSING, # tracker 的 BF2042 overview 没有 SPM / 得分,用真实存在的这两项替代 "每分钟伤害": _display(_stat(overview, "dmgPerMin")), "场均击杀": _value(_stat(overview, "killsPerMatch")) or _MISSING, "击杀总计": _number(_stat(overview, "kills")), "游戏助攻": _number(_stat(overview, "assists")), "救援数量": _number(_stat(overview, "revives")), "爆头率": _display(_stat(overview, "headshotPercentage")), "最佳武器": _sub(best_weapon_seg), "最佳载具": _sub(best_vehicle_seg), } # --------------------------------------------------------------------------- # # 榜单 / 模式视图 # --------------------------------------------------------------------------- # def _raw(entry: dict | None): """原样取 value(比率类字段用,displayValue 偶尔缺失)。""" value = _value(entry) return _MISSING if value is None else value # tracker.gg 偶尔漏译,metadata.name 会是 gad_open_equip_rpgv2 这种原始键 _RAW_NAME_PREFIXES = ("gad_open_equip_", "gad_", "wep_", "veh_") def _pretty_name(name: str) -> str: if not name or name == _MISSING or not re.fullmatch(r"[a-z0-9_]+", name): return name for prefix in _RAW_NAME_PREFIXES: if name.startswith(prefix): name = name[len(prefix) :] break return name.replace("_", " ").upper() # kind -> 标题 / 段名 / 排序字段 / (列名, stat 键, 取法) _LIST_VIEWS: dict[str, dict] = { "weapon": { "标题": "武器", "段": "weapon", "排序": "kills", "列": ( ("击杀", "kills", _number), ("KPM", "killsPerMinute", _raw), ("命中率", "shotsAccuracy", _display), ("爆头率", "headshotPercentage", _display), ), }, "vehicle": { "标题": "载具", "段": "vehicle", "排序": "kills", "列": ( ("击杀", "kills", _number), ("KPM", "killsPerMinute", _raw), ("摧毁", "destroyed", _number), ("时长", "timePlayed", _display), ), }, "gadget": { "标题": "装备", "段": "gadget", "排序": "uses", # 装备大多 0 杀,使用次数才是主指标 "列": ( ("使用", "uses", _number), ("击杀", "kills", _number), ("部署", "deployments", _number), ("时长", "timePlayed", _display), ), }, } _GAMEMODE_COLUMNS = ( ("场次", "matches", _number), ("胜率", "wlPercentage", _display), ("击杀", "kills", _number), ("KPM", "killsPerMinute", _raw), ) def _rows_of(segments: list[dict], columns, sort_key: str, top: int) -> list[dict]: """按 sort_key 降序取前 top,摊平成「名称/标签/值列表」的渲染行。""" def rank(seg: dict) -> float: return _value(_stat(seg.get("stats") or {}, sort_key)) or 0 rows = [] for item in sorted(segments or [], key=rank, reverse=True)[:top]: stats = item.get("stats") or {} rows.append( { "名称": _pretty_name(_name_of(item)), "标签": (item.get("metadata") or {}).get("category") or "", "值": [str(fn(_stat(stats, key))) for _, key, fn in columns], } ) return rows def bf2042_list_title(kind: str) -> str: """榜单中文标题(武器/载具/装备),指令层拼提示语用。""" view = _LIST_VIEWS.get(kind) return view["标题"] if view else "" def normalize_bf2042_list(kind: str, segments: list[dict], top: int = 15) -> dict: """把 weapon/vehicle/gadget 段整理成榜单卡字段(按主指标排序,取前 top)。""" view = _LIST_VIEWS[kind] return { "标题": view["标题"], "列名": [name for name, _, _ in view["列"]], "行": _rows_of(segments, view["列"], view["排序"], top), } def normalize_bf2042_gamemodes(profile: dict, top: int = 15) -> dict: """把 profile 里的 gamemode 段整理成榜单卡字段(按游玩时长排序)。""" segments = ((profile or {}).get("data") or {}).get("segments") or [] modes = [s for s in segments if s.get("type") == "gamemode"] return { "标题": "模式", "列名": [name for name, _, _ in _GAMEMODE_COLUMNS], "行": _rows_of(modes, _GAMEMODE_COLUMNS, "timePlayed", top), } # --------------------------------------------------------------------------- # # 抓取(调度在 tracker_edge:直连 API 主通道 / 扩展桥兜底) # --------------------------------------------------------------------------- # def _handle(profile: dict | None, fallback: str) -> str: info = ((profile or {}).get("data") or {}).get("platformInfo") or {} return info.get("platformUserHandle") or fallback def _failure(reason: str, detail: str) -> TrackerResult: return TrackerResult(False, data=None, status=None, reason=reason, detail=detail) async def _load_profile(slug: str, name: str, segments: tuple[str, ...] = ()): """经调度层取数,返回 (profile 数据, 段字典, None)。 段字典键:weapon / vehicle / gadget / soldier。一次取数全部拉齐, 不做任何"按需补段"的额外请求。失败返回 (None, {}, 带 reason 结果)。 """ result = await fetch_bf2042_segments(name, slug) if result.get("error"): return ( None, {}, _failure( str(result["error"]), str(result.get("detail") or "tracker.gg 取数失败") ), ) profile = result.get("profile") if not isinstance(profile, dict) or not profile.get("data"): logger.warning(f"BF2042 取到的 profile 为空: {name}") return None, {}, _failure("not_found", "tracker.gg 未返回该玩家的数据") seg_lists = { "weapon": result.get("weapon") or [], "vehicle": result.get("vehicle") or [], "gadget": result.get("gadget") or [], "soldier": result.get("soldier") or [], } return profile, seg_lists, None async def get_bf2042_info(player: str, platform: str = "pc") -> TrackerResult: """查询 BF2042 生涯数据,返回 TrackerResult(data 为生涯卡字段 dict)。 data 里同时带上原始段(``_profile``/``_weapons``/``_vehicles``/``_soldiers``), 供新版卡片(``services.processing.build_bf2042_career_card``)构建完整槽位。 """ slug = _platform_slug(platform) name = (player or "").strip() if not name: return TrackerResult(False, reason="error", detail="玩家名为空") profile, seg_lists, err = await _load_profile(slug, name) if err is not None: return err extra = seg_lists.get("weapon", []) + seg_lists.get("vehicle", []) logger.info(f"BF2042 生涯取数成功: {name}(武器/载具段 {len(extra)} 条)") data = normalize_bf2042(profile or {}, extra) data["_profile"] = profile or {} data["_weapons"] = seg_lists.get("weapon") or [] data["_vehicles"] = seg_lists.get("vehicle") or [] data["_soldiers"] = seg_lists.get("soldier") or [] return TrackerResult(True, data=data, status=200, reason="ok") async def get_bf2042_list(player: str, platform: str, kind: str) -> TrackerResult: """查询武器 / 载具 / 装备榜,返回 TrackerResult(data 为榜单卡字段 dict)。 kind ∈ weapon / vehicle / gadget;数据全在页面一次加载的截获结果里。 """ slug = _platform_slug(platform) name = (player or "").strip() if not name: return TrackerResult(False, reason="error", detail="玩家名为空") view = _LIST_VIEWS.get(kind) if view is None: return TrackerResult(False, reason="error", detail=f"未知榜单类型 {kind}") segment = view["段"] profile, seg_lists, err = await _load_profile(slug, name) if err is not None: return err rows_source = seg_lists.get(segment, []) if not rows_source: # 段拉取失败或该玩家确实没有这类数据(直连通道四段全取,桥兜底仅三段) return TrackerResult( False, reason="error", detail=f"当前通道暂无「{view['标题']}」数据", ) data = normalize_bf2042_list(kind, rows_source) data["玩家名称"] = _handle(profile, name) logger.info(f"BF2042 {data['标题']}榜取数成功: {name}({len(data['行'])} 条)") return TrackerResult(True, data=data, status=200, reason="ok") async def get_bf2042_gamemodes(player: str, platform: str = "pc") -> TrackerResult: """查询各游戏模式明细(在 profile 响应的 gamemode 段里,不用额外请求)。""" slug = _platform_slug(platform) name = (player or "").strip() if not name: return TrackerResult(False, reason="error", detail="玩家名为空") profile, _, err = await _load_profile(slug, name) if err is not None: return err data = normalize_bf2042_gamemodes(profile or {}) data["玩家名称"] = _handle(profile, name) logger.info(f"BF2042 模式取数成功: {name}({len(data['行'])} 条)") return TrackerResult(True, data=data, status=200, reason="ok")