sansenhoshiandClaude Code
a31aab00e8
- 旧平铺模块（data/data_utils/database_op/image_builder*/get_bf6_data/ gametools_bf6/bf6_data/img_utils）拆分为 handlers（绑定/查询）、 services（fetchers/processing/query）、utils（渲染/参数）分层； 新增 repository.py / config.py / paths.py - 字体/图片/模板资源迁入 res/，新增 BF2042 卡面素材（bg/object_icon 等） - browser_ext/：BF2042 扩展捕获桥（WS 直连兜底通道）源码入库 - 新增 hexi/web/src/plugins/bf_bot 前端管理页 - 新增 tests：test_bf_bot_helpers / repository / tracker_api Co-Authored-By: Claude Code <noreply@anthropic.com>
392 lines
14 KiB
Python
392 lines
14 KiB
Python
"""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")
|