sansenhoshi/HeXi
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(steam-info): 资料页状态/最近游玩/ban 描述增强，时长图渲染与定时任务迁移

Browse Source 
- steam.py / draw.py / handlers 扩充：个人资料页状态构建、最近游玩、
  游戏头像与中文游戏名查询、ban 描述
- 定时播报迁入 hexi.core.scheduler
- 新增 tests/test_steam_profile_parse.py（资料 HTML/XML 解析）

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
sansenhoshiandClaude Code committed 2026-10-13 17:53:22 +08:00
1 parent 0ac1958932
commit 6af60a7d55
6 files changed
+1298 -125
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

hexi/plugins/nonebot_plugin_steam_info/__init__.py
+7 -2
View File
@@ -16,7 +16,7 @@ require("nonebot_plugin_localstore")
require("nonebot_plugin_apscheduler")
from nonebot_plugin_alconna.uniseg import get_target as uni_get_target
import nonebot_plugin_localstore as store
from nonebot_plugin_apscheduler import scheduler
from hexi.core.scheduler import scheduled_job
from nonebot_plugin_alconna import Text, Image, UniMessage, Target
from nonebot.adapters.onebot.v11 import Bot
from .config import Config
@@ -29,13 +29,18 @@ from .steam import (
get_steam_users_info,
init_api_key_manager,
get_play_time as fetch_play_time,
get_recently_played,
fetch_game_header_image,
get_game_cn_names,
localize_game_name,
migrate_game_names_json,
update_steamdb_apps,
build_profile_status,
describe_ban,
_is_full_sync_done,
_query_game_names,
_query_game_icon_hashes,
_query_game_name_ch_by_en,
_upsert_game_names,
_upsert_game_icon_hashes,
)
@@ -464,7 +469,7 @@ async def update_steam_info():
_broadcast_lock = asyncio.Lock()
@scheduler.scheduled_job(
@scheduled_job(
"interval", minutes=config.steam_request_interval / 60, id="update_steam_info"
)
async def fetch_and_broadcast_steam_info():
hexi/plugins/nonebot_plugin_steam_info/draw.py
+357 -95
View File
@@ -2,12 +2,12 @@ import numpy as np
from functools import lru_cache
from io import BytesIO
from pathlib import Path
from typing import List, Dict, Tuple
from typing import List, Dict, Tuple, Optional
from colorsys import rgb_to_hsv, hsv_to_rgb
from PIL import Image, ImageDraw, ImageFont, ImageFilter, ImageEnhance
from .utils import hex_to_rgb
from .models import DrawPlayerStatusData, Achievements
from .models import DrawPlayerStatusData, Achievements, ProfileCardExtra
WIDTH = 400
@@ -824,6 +824,7 @@ def draw_game_info(
completed_achievement_number: int,
total_achievement_number: int,
achievement_color: Tuple[int, int, int],
two_week_time: Optional[str] = None,
) -> Image.Image:
bg = Image.new("RGBA", (880, 110 + 64 + 10), (0, 0, 0, 110))
header = header.resize((229, 86), Image.BICUBIC)
@@ -831,11 +832,27 @@ def draw_game_info(
draw = ImageDraw.Draw(bg)
# 近两周时长(可选):右上角绿色小字,游戏名超宽时先截断避让
two_week_width = 0
if two_week_time:
font_two_week = get_font(font_light_path, 20)
two_week_width = int(draw.textlength(two_week_time, font=font_two_week))
draw.text(
(bg.width - two_week_width - 10, 22),
two_week_time,
font=font_two_week,
fill=hex_to_rgb("91c257"),
)
# 画游戏名
name_font = get_font(font_regular_path, 26)
max_name_width = bg.width - 260 - 20 - two_week_width - (30 if two_week_width else 0)
if draw.textlength(game_name, font=name_font) > max_name_width:
game_name = _truncate_with_ellipsis(draw, game_name, name_font, max_name_width)
draw.text(
(260, 10),
game_name,
font=get_font(font_regular_path, 26),
font=name_font,
fill=(255, 255, 255),
)
@@ -880,7 +897,7 @@ def draw_game_info(
(int(x), 20),
f"{completed_achievement_number} / {total_achievement_number}",
font=font,
fill=(130, 130, 130),
fill=(185, 185, 185),
)
x += (
font.getlength(f"{completed_achievement_number} / {total_achievement_number}")
@@ -913,168 +930,413 @@ def draw_game_info(
return bg
# 信息卡尺寸/排版常量(调整版式优先动这里)
CARD_WIDTH = 1200 # 卡片条宽度,背景按此居中裁切
CARD_MARGIN = 20 # 左右安全边距
LEFT_W = 880 # 左栏(游戏动态)宽度
RIGHT_X = CARD_MARGIN + LEFT_W + CARD_MARGIN # 920:右栏起点
RIGHT_W = CARD_WIDTH - RIGHT_X - CARD_MARGIN # 260:右栏宽度
AVATAR_SIZE = 160
AVATAR_X = 30 # 头像框约 1.36×头像,需留出外溢空间
HEADER_TEXT_X = AVATAR_X + AVATAR_SIZE + 30 # 220:头部文字统一左对齐点
GAME_GAP = 16 # 游戏卡片间距
COUNT_ROW_MAX_STEP = 44 # 右栏计数条目间距上限(面板矮时自动收紧)
AVATAR_FRAME_SCALE = 1.36 # 头像框相对头像的缩放(对齐 Steam 内孔比例)
# 状态行配色,key 为 ProfileCardExtra.status_kind
status_row_colors = {
"in-game": hex_to_rgb("91c257"),
"online": hex_to_rgb("6dcef5"),
"away": hex_to_rgb("e0a948"),
"offline": hex_to_rgb("969697"),
"private": hex_to_rgb("969697"),
}
# 风控标记配色,key 为 ProfileCardExtra.ban_kind
ban_tag_colors = {
"vac": hex_to_rgb("c4453c"),
"trade": hex_to_rgb("c98a2e"),
"limited": hex_to_rgb("67665c"),
}
# 右栏状态区标题,key 为 ProfileCardExtra.status_kind
status_titles = {
"in-game": "当前正在游戏",
"online": "当前在线",
"away": "当前在线",
"offline": "当前离线",
"private": "当前状态",
}
def _draw_kv_parts(
draw: ImageDraw.ImageDraw,
x: int,
y: int,
parts: List[Tuple[str, tuple]],
font,
) -> int:
"""按 (文本, 颜色) 依次绘制,返回右边缘 x(间距由文本自带的空格控制)"""
for text, color in parts:
draw.text((x, y), text, font=font, fill=color)
x += int(draw.textlength(text, font=font))
return x
def _draw_status_row(
draw: ImageDraw.ImageDraw,
x: int,
y: int,
text: str,
kind: str = "online",
second_line: str = "",
) -> None:
"""状态行:彩色圆点 + 文案(游戏中/在线/离线/资料私密),过长可折第二行"""
color = status_row_colors.get(kind, status_row_colors["online"])
font = get_font(font_regular_path, 17)
draw.ellipse((x, y + 4, x + 11, y + 15), fill=color)
draw.text((x + 19, y), text, font=font, fill=color)
if second_line:
draw.text((x + 19, y + 22), second_line, font=font, fill=color)
def _split_two_lines(
draw: ImageDraw.ImageDraw, text: str, font, max_width: int
) -> List[str]:
"""超宽文本折成最多两行:优先在最后一个放得下的分隔符处断,否则省略号截断"""
if draw.textlength(text, font=font) <= max_width:
return [text]
break_at = None
for idx, char in enumerate(text):
if char in ":：" or char == " ":
if draw.textlength(text[: idx + 1], font=font) <= max_width:
break_at = idx + 1
if break_at is None:
return [_truncate_with_ellipsis(draw, text, font, max_width)]
second = text[break_at:].strip()
if draw.textlength(second, font=font) > max_width:
second = _truncate_with_ellipsis(draw, second, font, max_width)
return [text[:break_at].strip(), second]
def _draw_level_badge(
draw: ImageDraw.ImageDraw, x: int, center_y: int, level: int
) -> int:
"""Steam 风格等级圆牌,返回右边缘 x"""
diameter = 34
top = center_y - diameter // 2
draw.ellipse(
(x, top, x + diameter, top + diameter),
fill=hex_to_rgb("1b2838"),
outline=hex_to_rgb("66c0f4"),
width=2,
)
font = get_font(font_bold_path, 18)
text = str(level)
text_width = draw.textlength(text, font=font)
draw.text(
(x + (diameter - text_width) / 2, top + (diameter - font.size) / 2 - 3),
text,
font=font,
fill=(255, 255, 255),
)
return x + diameter
def _draw_ban_tag(
draw: ImageDraw.ImageDraw, x: int, y: int, text: str, kind: str
) -> int:
"""风控标记圆角标签,返回右边缘 x"""
font = get_font(font_regular_path, 13)
padding_x, padding_y = 9, 4
width = draw.textlength(text, font=font) + padding_x * 2
height = font.size + padding_y * 2
draw.rounded_rectangle(
(x, y, x + width, y + height),
radius=height // 2,
fill=ban_tag_colors.get(kind, ban_tag_colors["limited"]),
)
draw.text(
(x + padding_x, y + padding_y - 1),
text,
font=font,
fill=(255, 255, 255),
)
return int(x + width)
def draw_player_status(
player_bg: Image.Image,
player_avatar: Image.Image,
player_name: str,
player_id: str,
player_description: str,
player_last_two_weeks_time: str, # e.g. 10.2 小时
player_last_two_weeks_time: str, # e.g. 10.2 小时(过去 2 周)
player_games: List[DrawPlayerStatusData],
extra: Optional[ProfileCardExtra] = None,
avatar_frame: Optional[bytes] = None,
):
"""信息卡:通栏头部 + 左栏「游戏动态」+ 右栏「当前在线/计数」(参考 Steam 官方资料页)"""
if isinstance(player_bg, bytes):
player_bg = Image.open(BytesIO(player_bg))
if isinstance(player_avatar, bytes):
player_avatar = Image.open(BytesIO(player_avatar))
extra = extra or {}
bg = recolor_image(
player_bg.crop(
(
(player_bg.width - 960) // 2,
(player_bg.width - CARD_WIDTH) // 2,
0,
(player_bg.width + 960) // 2,
(player_bg.width + CARD_WIDTH) // 2,
player_bg.height,
)
),
10,
10,
)
# 调暗背景
enhancer = ImageEnhance.Brightness(bg)
bg = enhancer.enhance(0.7)
# bg.size = (960, 1020)
player_avatar = player_avatar.resize((200, 200))
bg.paste(player_avatar, (40, 40))
bg = ImageEnhance.Brightness(bg).enhance(0.7)
draw = ImageDraw.Draw(bg)
# 画头像外框
draw.rectangle((40, 40, 240, 240), outline=(83, 164, 196), width=3)
# ---------- 头部:头像(+头像框)/ 昵称+等级+风控 / 简介 / 好友代码·注册时间 ----------
avatar = player_avatar.convert("RGB").resize(
(AVATAR_SIZE, AVATAR_SIZE), Image.BICUBIC
)
bg.paste(avatar_with_border(avatar), (AVATAR_X - 2, AVATAR_X - 2))
if avatar_frame:
try:
frame = Image.open(BytesIO(avatar_frame)).convert("RGBA")
frame_size = int(AVATAR_SIZE * AVATAR_FRAME_SCALE)
frame = frame.resize((frame_size, frame_size), Image.LANCZOS)
frame_pos = AVATAR_X + (AVATAR_SIZE - frame_size) // 2
bg.paste(frame, (frame_pos, frame_pos), frame)
except Exception:
pass # 头像框只是装饰,画不出来不影响整卡
# 画昵称
# 好友代码/注册时间的标签色:比正文灰稍亮一档(#969696 在深色底上偏暗)
text_gray = hex_to_rgb("b4b4b4")
name_font = get_font(font_light_path, 38)
reserved = 34 + 10 if extra.get("level") is not None else 0
if extra.get("ban_text"):
ban_font = get_font(font_regular_path, 13)
reserved += int(draw.textlength(extra["ban_text"], font=ban_font)) + 18 + 10
max_name_width = CARD_WIDTH - CARD_MARGIN - HEADER_TEXT_X - reserved
if draw.textlength(player_name, font=name_font) > max_name_width:
player_name = _truncate_with_ellipsis(
draw, player_name, name_font, max_name_width
)
name_y = AVATAR_X + 6
draw.text(
(280, 48),
player_name,
font=get_font(font_light_path, 40),
fill=(255, 255, 255),
(HEADER_TEXT_X, name_y), player_name, font=name_font, fill=(255, 255, 255)
)
# 画ID
draw.text(
(280, 100),
f"好友代码: {player_id}",
font=get_font(font_regular_path, 19),
fill=(191, 191, 191),
)
# 等级圆牌 / 风控标记紧随昵称
cursor = HEADER_TEXT_X + int(draw.textlength(player_name, font=name_font)) + 12
if extra.get("level") is not None:
cursor = _draw_level_badge(draw, cursor, name_y + 21, extra["level"]) + 10
if extra.get("ban_text"):
_draw_ban_tag(
draw, cursor, name_y + 10, extra["ban_text"], extra.get("ban_kind") or "limited"
)
# 画简介
# 简介(最多 3 行)
desc_font = get_font(font_light_path, 20)
desc_top = name_y + 60
max_desc_width = CARD_WIDTH - CARD_MARGIN - HEADER_TEXT_X
line_width = 0
offset = 0
line = ""
for idx, char in enumerate(player_description):
line += char
line_width += get_font(font_light_path, 22).getlength(char)
if line_width > 640 or idx == len(player_description) - 1 or char == "\n":
line_width += desc_font.getlength(char)
if (
line_width > max_desc_width
or idx == len(player_description) - 1
or char == "\n"
):
draw.text(
(280, 132 + offset),
(HEADER_TEXT_X, desc_top + offset),
line,
font=get_font(font_light_path, 22),
fill=(255, 255, 255),
font=desc_font,
fill=(235, 235, 235),
)
line = ""
offset += 25
line_width = 0
if offset >= 25 * 4:
offset += 27
if offset >= 27 * 3:
break
# 画游戏
# 好友代码 / 注册时间
meta_font = get_font(font_regular_path, 16)
meta_y = desc_top + offset + 14
meta_parts = [("好友代码: ", text_gray), (player_id, (255, 255, 255))]
if extra.get("member_since"):
meta_parts += [
(" 注册时间: ", text_gray),
(extra["member_since"], (255, 255, 255)),
]
_draw_kv_parts(draw, HEADER_TEXT_X, meta_y, meta_parts, meta_font)
header_h = max(AVATAR_X + AVATAR_SIZE + 26, meta_y + 22) + 14
draw.rectangle(
(CARD_MARGIN, header_h - 1, CARD_WIDTH - CARD_MARGIN, header_h),
fill=hex_to_rgb("333439"),
)
# ---------- 左栏:游戏动态 ----------
body_top = header_h + 16
brightest_color, darkest_color = get_brightest_and_darkest_color(player_bg)
brightest_color = tuple(map(lambda x: x - 30 if x >= 30 else 0, brightest_color))
darkest_color = tuple(
map(lambda x: x + 30 if x <= 255 - 30 else 255, darkest_color)
)
brightest_color = (brightest_color[0], brightest_color[1], brightest_color[2], 128)
brightest_color = random_color_offset(brightest_color, 20)
darkest_color = (darkest_color[0], darkest_color[1], darkest_color[2], 128)
darkest_color = random_color_offset(darkest_color, 20)
# 画游戏信息
hsv_achievement_color = rgb_to_hsv(*brightest_color[:3])
# 成就条底色:取背景亮色的深色变体(压暗一些,保证白色文字和数字读得清)
hsv_achievement_color = rgb_to_hsv(*brightest_color)
achievement_color = tuple(
map(
int,
hsv_to_rgb(
hsv_achievement_color[0],
hsv_achievement_color[1] * 0.85,
hsv_achievement_color[2] * 0.6,
hsv_achievement_color[1] * 0.8,
hsv_achievement_color[2] * 0.35,
),
)
)
game_images: List[Image.Image] = []
for idx, game in enumerate(player_games):
game_image = Image.open(BytesIO(game["game_header"]))
game_info = draw_game_info(
game_image,
game["game_name"],
game["game_time"],
game["last_play_time"],
game["achievements"],
game["completed_achievement_number"],
game["total_achievement_number"],
achievement_color,
cards: List[Image.Image] = []
for game in player_games:
cards.append(
draw_game_info(
Image.open(BytesIO(game["game_header"])),
game["game_name"],
game["game_time"],
game["last_play_time"],
game["achievements"],
game["completed_achievement_number"],
game["total_achievement_number"],
achievement_color,
two_week_time=game.get("two_week_time"),
)
)
game_images.append(game_info)
# 画半透明黑色背景
bg_game = Image.new(
"RGBA", (920, 106 + sum([game_image.height + 26 for game_image in game_images]))
# 背景太矮时先向下补一段(拉伸底部横条),避免内容整体被裁
min_height = body_top + 58 + 110 + CARD_MARGIN
if bg.height < min_height:
strip = bg.crop((0, bg.height - 8, CARD_WIDTH, bg.height)).resize(
(CARD_WIDTH, min_height - bg.height), Image.BICUBIC
)
canvas = Image.new("RGB", (CARD_WIDTH, min_height))
canvas.paste(bg, (0, 0))
canvas.paste(strip, (0, bg.height))
bg = canvas
draw = ImageDraw.Draw(bg)
# 按可用高度筛出画得下的游戏卡片(至少画第一条)
y = body_top + 58
max_bottom = bg.height - CARD_MARGIN
shown_cards: List[Image.Image] = []
for card in cards:
if shown_cards and y + card.height > max_bottom:
break
shown_cards.append(card)
y += card.height + GAME_GAP
left_bottom = (y - GAME_GAP + 12) if shown_cards else (body_top + 66)
# 左右两栏等高的半透明底板
left_panel = Image.new("RGBA", (LEFT_W, left_bottom - body_top), (0, 0, 0, 120))
bg.paste(left_panel, (CARD_MARGIN, body_top), left_panel)
right_panel = Image.new("RGBA", (RIGHT_W, left_bottom - body_top), (0, 0, 0, 120))
bg.paste(right_panel, (RIGHT_X, body_top), right_panel)
# 左栏标题条(渐变)+ 近两周总时长
gradient = create_gradient_image(
(LEFT_W, 50),
(brightest_color[0], brightest_color[1], brightest_color[2], 128),
(darkest_color[0], darkest_color[1], darkest_color[2], 128),
)
draw_game = ImageDraw.Draw(bg_game)
draw_game.rectangle(
(
0,
0,
920,
bg_game.height,
),
fill=(0, 0, 0, 120),
)
bg.paste(bg_game, (20, 272), bg_game)
# 画渐变条
gradient = create_gradient_image((920, 50), brightest_color, darkest_color)
bg.paste(gradient, (20, 272), gradient)
# 画渐变条的文字：最新动态，最近游戏
bg.paste(gradient, (CARD_MARGIN, body_top), gradient)
draw.text(
(34, 279),
"最新动态",
(CARD_MARGIN + 14, body_top + 7),
"游戏动态",
font=get_font(font_light_path, 26),
fill=(255, 255, 255),
)
if player_last_two_weeks_time is not None:
width = get_font(font_light_path, 26).getlength(
player_last_two_weeks_time
)
if player_last_two_weeks_time:
total_font = get_font(font_light_path, 22)
total_width = total_font.getlength(player_last_two_weeks_time)
draw.text(
(960 - width - 34, 279),
(CARD_MARGIN + LEFT_W - total_width - 14, body_top + 9),
player_last_two_weeks_time,
font=get_font(font_light_path, 26),
font=total_font,
fill=(255, 255, 255),
)
y = body_top + 58
for card in shown_cards:
bg.paste(card, (CARD_MARGIN, y), card.convert("RGBA"))
y += card.height + GAME_GAP
y = 350
for idx, game_image in enumerate(game_images):
bg.paste(
game_image,
((920 - game_image.width) // 2 + 20, y),
game_image.convert("RGBA"),
# ---------- 右栏:当前在线 + 计数 ----------
status_title = status_titles.get(
extra.get("status_kind") or "online", "当前在线"
)
draw.text(
(RIGHT_X + 20, body_top + 14),
status_title,
font=get_font(font_light_path, 20),
fill=hex_to_rgb("c5d6d4"),
)
status_extra_h = 0
if extra.get("status_text"):
status_font = get_font(font_regular_path, 17)
status_lines = _split_two_lines(
draw, extra["status_text"], status_font, RIGHT_W - 60
)
y += game_image.height + 26
status_extra_h = 22 * (len(status_lines) - 1)
_draw_status_row(
draw,
RIGHT_X + 20,
body_top + 46,
status_lines[0],
extra.get("status_kind") or "online",
status_lines[1] if len(status_lines) > 1 else "",
)
divider_y = body_top + 76 + status_extra_h
draw.rectangle(
(RIGHT_X + 20, divider_y, RIGHT_X + RIGHT_W - 20, divider_y + 1),
fill=hex_to_rgb("333439"),
)
player_bg.paste(bg, ((player_bg.width - 960) // 2, 0), bg.convert("RGBA"))
count_rows = list((extra.get("counts") or {}).items())
panel_h = left_bottom - body_top - status_extra_h
step = COUNT_ROW_MAX_STEP
if count_rows:
step = min(COUNT_ROW_MAX_STEP, max(26, (panel_h - 114) // len(count_rows)))
row_font = get_font(font_regular_path, 15)
ry = divider_y + 14
for label, value in count_rows:
if ry + 20 > left_bottom - 12: # 面板放不下就不画,保证不越界
break
draw.text((RIGHT_X + 20, ry), label, font=row_font, fill=hex_to_rgb("8f989b"))
value_text = str(value)
draw.text(
(RIGHT_X + RIGHT_W - 20 - row_font.getlength(value_text), ry),
value_text,
font=row_font,
fill=(255, 255, 255),
)
ry += step
# 内容高于背景时(极端情况),背景向下拉伸补足后再贴回
if bg.height > player_bg.height:
strip = player_bg.crop(
(0, player_bg.height - 8, player_bg.width, player_bg.height)
).resize((player_bg.width, bg.height - player_bg.height), Image.BICUBIC)
canvas = Image.new("RGB", (player_bg.width, bg.height))
canvas.paste(player_bg, (0, 0))
canvas.paste(strip, (0, player_bg.height))
player_bg = canvas
player_bg.paste(bg, ((player_bg.width - CARD_WIDTH) // 2, 0), bg.convert("RGBA"))
return player_bg
hexi/plugins/nonebot_plugin_steam_info/handlers/steam.py
+117 -17
View File
@@ -34,20 +34,31 @@ from .. import (
get_steam_users_info,
get_target,
get_user_data,
get_recently_played,
fetch_game_header_image,
image_to_bytes,
key_select,
localize_game_name,
simplize_steam_player_data,
build_profile_status,
describe_ban,
_fill_missing_icon_hashes,
_query_game_names,
_query_game_name_ch_by_en,
_upsert_game_icon_hashes,
update_steamdb_apps,
)
# 信息卡「游戏动态」最多展示的条数(按近两周时长排序,列表本身最多 8 条)
MAX_ACTIVITY_GAMES = 4
help = on_command("steamhelp", aliases={"steam帮助"}, priority=10)
bind = on_command("steambind", aliases={"绑定steam"}, priority=10)
unbind = on_command("steamunbind", aliases={"解绑steam"}, priority=10)
info = on_command("steaminfo", aliases={"steam信息"}, priority=10)
# steaminfo 会额外调一次 Steam API(近两周时长),按群冷却防止被刷配额
info = cooldown(30, scope="group", hint="群里刚有人查过,{secs} 秒后再试")(
on_command("steaminfo", aliases={"steam信息"}, priority=10)
)
check = cooldown(30, scope="group", hint="群里刚有人查过,{secs} 秒后再试")(
on_command("steamcheck", aliases={"查看steam", "查steam", "谁在玩游戏"}, priority=10)
)
@@ -128,29 +139,118 @@ async def info_handle(bot: Bot, event: Event, target: Target = Depends(get_targe
steam_id = user_data["steam_id"]
steam_friend_code = str(int(steam_id) - STEAM_ID_OFFSET)
player_data = await get_user_data(steam_id, cache_path, config.proxy)
names = await get_game_cn_names(
[g.get("appid") for g in player_data["game_data"] if g.get("appid")],
db_path,
# 资料页抓取与近两周列表 API 并行;后者是可选增强,无配额时自动跳过
player_data, recent_games = await asyncio.gather(
get_user_data(steam_id, cache_path, config.proxy),
get_recently_played(steam_id, config.proxy),
)
for g in player_data["game_data"]:
g["game_name"] = localize_game_name(g["game_name"], g.get("appid"), names)
draw_data = [
recent_games = recent_games or []
# 游戏动态以最近两周列表为准(资料页只有最近几款),资料页解析的
# 成就/最后运行日期按 appid 合并进来;列表没有的游戏用资料页顺序补齐
page_games = {
g["appid"]: g for g in player_data["game_data"] if g.get("appid")
}
rows = list(recent_games)
known_appids = {r["appid"] for r in rows}
rows += [
{
"game_header": game["game_image"],
"game_name": game["game_name"],
"game_time": f"{game['play_time']} 小时",
"last_play_time": game["last_played"],
"achievements": game["achievements"],
"completed_achievement_number": game.get("completed_achievement_number"),
"total_achievement_number": game.get("total_achievement_number"),
"appid": appid,
"name": game["game_name"],
"two_week_hours": None,
"total_hours": None,
}
for game in player_data["game_data"]
for appid, game in page_games.items()
if appid not in known_appids
]
rows = rows[:MAX_ACTIVITY_GAMES]
# 缺封面的行(不在资料页里)从公开 CDN 并发补,带缓存
missing_appids = [r["appid"] for r in rows if r["appid"] not in page_games]
cdn_headers = {}
if missing_appids:
fetched = await asyncio.gather(
*(
fetch_game_header_image(appid, cache_path, config.proxy)
for appid in missing_appids
)
)
cdn_headers = dict(zip(missing_appids, fetched))
names = await get_game_cn_names([r["appid"] for r in rows], db_path)
draw_data = []
for row in rows:
page = page_games.get(row["appid"], {})
header = page.get("game_image") or cdn_headers.get(row["appid"])
if header is None:
logger.warning(f"游戏 {row['appid']} 封面获取失败,本条跳过")
continue
if row["total_hours"] is not None:
game_time = f"{row['total_hours']:.1f}"
else:
game_time = str(page.get("play_time") or "")
draw_data.append(
{
"game_header": header,
"game_name": localize_game_name(
page.get("game_name") or row["name"], row["appid"], names
),
"game_time": f"{game_time} 小时",
"last_play_time": page.get("last_played") or "",
"achievements": page.get("achievements") or [],
"completed_achievement_number": page.get(
"completed_achievement_number"
),
"total_achievement_number": page.get("total_achievement_number"),
"two_week_time": (
f"近两周 {row['two_week_hours']:.1f} 小时"
if row["two_week_hours"]
else None
),
}
)
# 近两周合计:API 列表求和最准;没有列表(资料不公开/无配额)时退回页面文案
if recent_games:
total_hours = sum(r["two_week_hours"] for r in recent_games)
two_week_total = (
f"{total_hours:.1f} 小时(过去 2 周)" if total_hours > 0 else "近两周未游玩"
)
else:
two_week_total = player_data["recent_2_week_play_time"]
# 状态行/风控标记:资料页「当前正在游戏」优先,其次资料 XML;
# XML 兜底路径的英文游戏名用映射表反查中文
status = build_profile_status(
player_data["online_state"],
player_data["state_message"],
player_data["privacy_state"],
player_data["current_game"],
)
if status.get("game_name_en"):
cn_name = _query_game_name_ch_by_en(db_path, status["game_name_en"])
if cn_name:
status["text"] = cn_name
ban = describe_ban(
player_data["vac_banned"],
player_data["trade_ban_state"],
player_data["limited_account"],
)
extra = {
"level": player_data["level"],
"counts": player_data["counts"],
"member_since": player_data["member_since"],
"status_text": status["text"],
"status_kind": status["kind"],
"ban_text": ban["text"] if ban else None,
"ban_kind": ban["kind"] if ban else None,
}
image = draw_player_status(
player_data["background"], player_data["avatar"], player_data["player_name"],
str(steam_friend_code), player_data["description"],
player_data["recent_2_week_play_time"], draw_data,
two_week_total, draw_data,
extra=extra,
avatar_frame=player_data.get("avatar_frame"),
)
await info.finish(await UniMessage(Image(raw=image_to_bytes(image))).export(bot))
hexi/plugins/nonebot_plugin_steam_info/models.py
+32 -1
View File
@@ -1,4 +1,4 @@
from typing import TypedDict, List
from typing import TypedDict, List, Optional, Dict
class Player(TypedDict):
@@ -57,9 +57,38 @@ class PlayerData(TypedDict):
player_name: str
background: bytes
avatar: bytes
avatar_frame: Optional[bytes] # 头像框(社区物品)PNG,无则 None
description: str
recent_2_week_play_time: str
game_data: List[GameData]
# 以下字段来自主页 HTML 解析 + 社区 ?xml=1(不占 API 配额),缺失时为默认值
level: Optional[int]
counts: Dict[str, int]
member_since: Optional[str]
online_state: Optional[str]
state_message: Optional[str]
current_game: Optional[str] # 资料页「当前正在游戏」的游戏名(已本地化)
privacy_state: Optional[str]
vac_banned: bool
trade_ban_state: Optional[str]
limited_account: bool
hours_2wk: Optional[float]
class ProfileCardExtra(TypedDict, total=False):
"""信息卡头部增强字段(缺失的键不绘制对应元素)
status_kind / ban_kind 决定配色,取值见 draw.py 的 status_row_colors /
ban_tag_colors;status_kind 为 private 时表示资料私密降级提示。
"""
level: Optional[int]
counts: Dict[str, int]
member_since: Optional[str]
status_text: Optional[str]
status_kind: Optional[str]
ban_text: Optional[str]
ban_kind: Optional[str]
class DrawPlayerStatusData(TypedDict):
@@ -70,6 +99,7 @@ class DrawPlayerStatusData(TypedDict):
achievements: List[Achievements]
completed_achievement_number: int
total_achievement_number: int
two_week_time: Optional[str] # e.g. 近两周 3.2 小时,缺失为 None
__all__ = [
@@ -78,5 +108,6 @@ __all__ = [
"PlayerSummariesResponse",
"ProcessedPlayer",
"PlayerSummariesProcessedResponse",
"ProfileCardExtra",
"DrawPlayerStatusData",
]
hexi/plugins/nonebot_plugin_steam_info/steam.py
+395 -10
View File
@@ -8,7 +8,9 @@ from nonebot.log import logger
from typing import List, Optional, Dict
from datetime import datetime, timezone
import asyncio
import hashlib
import json
from xml.etree import ElementTree
from hexi.core.rate_limit import acquire
from .config import Config
@@ -338,6 +340,249 @@ async def _fetch(
return default
# 交易封禁 XML 取值 -> 展示文案(未知取值原样展示)
TRADE_BAN_TEXT = {
"Trade Banned": "交易封禁",
"Probation": "交易限制",
}
# 默认在线状态文案 -> 中文(自定义状态原样展示)
STATE_MESSAGE_TEXT = {
"Online": "在线",
"Away": "离开",
}
# 计数行展示顺序(页面本身顺序不稳定,这里固定;未列出的标签按页面顺序排后面)
COUNT_LABEL_ORDER = ("游戏", "评测", "徽章", "好友", "截图", "创意工坊项目", "指南")
def _url_digest(url: str) -> str:
"""URL → 短摘要,用于生成缓存文件名"""
return hashlib.md5(url.encode()).hexdigest()[:10]
def parse_profile_background_url(html: str) -> Optional[str]:
"""解析资料页背景图 URL(纯函数,便于单测)
普通背景在 CSS 的 background-image 里;动画背景没有 CSS 图,
但 <video poster="..."> 上带着 1920×1080 的静态首帧,同样可用。
"""
candidates = [
m.group(1).strip()
for m in re.finditer(
r"background-image:\s*url\(\s*['\"]?(.*?)['\"]?\s*\)", html
)
]
# 背景图都来自社区物品(/items/);页面里其他 url() 不一定是背景,优先挑这类
for url in candidates:
if "/items/" in url and not url.lower().endswith((".webm", ".mp4")):
return url
for url in candidates:
if not url.lower().endswith((".webm", ".mp4")):
return url
poster = re.search(r"<video[^>]*\bposter=\"([^\"]+)\"", html)
if poster:
return poster.group(1)
return None
def parse_avatar_frame_url(html: str) -> Optional[str]:
"""解析头像框(社区物品)图片 URL,优先 prefers-reduced-motion 的静态版本"""
soup = BeautifulSoup(html, "html.parser")
frame = soup.select_one("div.profile_avatar_frame")
if frame is None:
return None
source = frame.select_one('source[media*="reduced-motion"]')
if source and source.get("srcset"):
return source["srcset"].split()[0]
img = frame.select_one("img")
if img is None:
return None
if img.get("src"):
return img["src"]
srcset = img.get("srcset") or ""
return srcset.split()[0] if srcset else None
def _parse_member_since(value: Optional[str]) -> Optional[str]:
"""注册时间归一化为 YYYY-MM-DD;无法识别时原样返回"""
if not value:
return None
for fmt in ("%B %d, %Y", "%Y年%m月%d日"):
try:
return datetime.strptime(value, fmt).strftime("%Y-%m-%d")
except ValueError:
continue
return value
def parse_profile_xml(text: str) -> dict:
"""解析社区资料 XML(纯函数,便于单测)
私密资料只返回极少节点,缺失字段按 None/False 处理;解析失败返回空 dict。
"""
try:
root = ElementTree.fromstring(text)
except ElementTree.ParseError:
return {}
def _text(tag: str) -> Optional[str]:
node = root.find(tag)
if node is None or node.text is None:
return None
return node.text.strip() or None
def _flag(tag: str) -> bool:
return _text(tag) == "1"
hours = _text("hoursPlayed2Wk")
try:
hours_2wk = float(hours) if hours else None
except ValueError:
hours_2wk = None
trade_ban = _text("tradeBanState")
return {
"member_since": _parse_member_since(_text("memberSince")),
"online_state": _text("onlineState"),
"state_message": _text("stateMessage"),
"vac_banned": _flag("vacBanned"),
"trade_ban_state": None if trade_ban in (None, "None") else trade_ban,
"limited_account": _flag("isLimitedAccount"),
"hours_2wk": hours_2wk,
"privacy_state": _text("privacyState"),
}
def parse_profile_extras(html: str) -> dict:
"""从主页 HTML 解析等级与计数行(纯函数,便于单测)"""
soup = BeautifulSoup(html, "html.parser")
level = None
level_node = soup.select_one("div.persona_level span.friendPlayerLevelNum")
if level_node and level_node.text.strip().isdigit():
level = int(level_node.text.strip())
counts: Dict[str, int] = {}
for link in soup.select("div.profile_count_link"):
label_node = link.select_one(".count_link_label")
total_node = link.select_one(".profile_count_link_total")
if label_node is None or total_node is None:
continue
label = label_node.text.strip()
# 空值多为 &nbsp; 占位(如未公开的库存),没有数字就不展示
value = total_node.text.replace("\xa0", "").replace(",", "").strip()
if not value.isdigit():
continue
counts[label] = int(value)
# 按展示顺序固定排列,未列出的标签按页面顺序排后面
ordered: Dict[str, int] = {
label: counts[label] for label in COUNT_LABEL_ORDER if label in counts
}
ordered.update({k: v for k, v in counts.items() if k not in ordered})
return {"level": level, "counts": ordered}
def parse_in_game_name(html: str) -> Optional[str]:
"""解析资料页「当前正在游戏」的游戏名(纯函数,便于单测)
这块比社区 XML 的 onlineState 更新更及时,且随 Accept-Language 本地化。
"""
soup = BeautifulSoup(html, "html.parser")
node = soup.select_one("div.profile_in_game_name")
if node is None:
return None
return node.get_text(strip=True) or None
def _extract_in_game_name(state_message: Optional[str]) -> str:
"""从 XML stateMessage 提取游戏名
实测两种写法都有:"In-Game: Team Fortress 2" / "In-Game<br/>VRChat"。
"""
text = re.sub(r"<[^>]+>", " ", state_message or "").strip()
if text.lower().startswith("in-game"):
text = text[len("In-Game"):].lstrip(":： ").strip()
return text or "未知游戏"
def build_profile_status(
online_state: Optional[str],
state_message: Optional[str],
privacy_state: Optional[str],
current_game: Optional[str] = None,
) -> dict:
"""整理卡片状态行文案(kind 决定配色)
优先级:资料页「当前正在游戏」> XML onlineState/stateMessage。
返回 dict:text / kind;中文名缺失时带 game_name_en 供调用方本地化。
"""
if privacy_state and privacy_state != "public":
return {"text": "资料私密,部分信息不可见", "kind": "private"}
if current_game:
# 资料页文案已按 zh-CN 本地化,直接用;卡片会配「当前正在游戏」标题
return {"text": current_game, "kind": "in-game"}
if online_state == "in-game":
name_en = _extract_in_game_name(state_message)
return {
"text": name_en,
"kind": "in-game",
"game_name_en": name_en,
}
if online_state == "online":
# 默认状态文案是英文(Online/Away),换成中文;自定义状态原样展示
text = STATE_MESSAGE_TEXT.get(state_message or "", state_message or "在线")
return {"text": text, "kind": "online"}
if online_state == "offline":
return {"text": "离线", "kind": "offline"}
return {"text": state_message or "", "kind": "online"}
def describe_ban(
vac_banned: bool,
trade_ban_state: Optional[str],
limited_account: bool,
) -> Optional[dict]:
"""风控标记(优先级 VAC > 交易封禁 > 受限账号),无异常返回 None"""
if vac_banned:
return {"text": "VAC 封禁", "kind": "vac"}
if trade_ban_state:
return {
"text": TRADE_BAN_TEXT.get(trade_ban_state, trade_ban_state),
"kind": "trade",
}
if limited_account:
return {"text": "受限账号", "kind": "limited"}
return None
async def fetch_profile_xml(steam_id, proxy: str = None) -> dict:
"""抓取社区资料 XML(memberSince/在线状态/封禁/近两周时长),不占 API 配额
与主页同属社区页面请求;失败或私密资料时字段缺失,调用方按空值降级。
"""
url = f"https://steamcommunity.com/profiles/{steam_id}/?xml=1"
try:
async with httpx.AsyncClient(
proxy=proxy,
timeout=httpx.Timeout(20.0),
follow_redirects=True,
headers={"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"},
) as client:
response = await client.get(url)
if response.status_code != 200:
logger.warning(
f"获取 Steam 资料 XML 失败: HTTP {response.status_code}"
)
return {}
return parse_profile_xml(response.text)
except Exception as exc:
# 增强项失败不影响主卡片,任何异常都降级为空结果
logger.warning(f"获取 Steam 资料 XML 失败: {exc}")
return {}
async def get_user_data(
steam_id: int, cache_path: Path, proxy: str = None
) -> PlayerData:
@@ -355,15 +600,30 @@ async def get_user_data(
"description": "No information given.",
"background": default_background,
"avatar": default_avatar,
"avatar_frame": None,
"player_name": "Unknown",
"recent_2_week_play_time": None,
"game_data": [],
# 资料 XML(社区接口,不占 API 配额)解锁的字段,缺失保持默认值
"level": None,
"counts": {},
"member_since": None,
"online_state": None,
"state_message": None,
"current_game": None,
"privacy_state": None,
"vac_banned": False,
"trade_ban_state": None,
"limited_account": False,
"hours_2wk": None,
}
local_time = datetime.now(timezone.utc).astimezone()
utc_offset_minutes = int(local_time.utcoffset().total_seconds() / 60)
timezone_cookie_value = f"{utc_offset_minutes},0"
# XML 与主页并行抓取,省一个往返;失败只记日志,字段留空
xml_task = asyncio.create_task(fetch_profile_xml(steam_id, proxy))
try:
async with httpx.AsyncClient(
proxy=proxy,
@@ -384,8 +644,14 @@ async def get_user_data(
response.raise_for_status()
except httpx.RequestError as exc:
logger.error(f"Failed to get user data: {exc}")
await xml_task # 主页失败也要回收任务,避免悬挂
return result
result.update(await xml_task)
result.update(parse_profile_extras(html))
# 资料页的「当前正在游戏」比 XML 更新更及时(且已本地化)
result["current_game"] = parse_in_game_name(html)
# player name
player_name = re.search(r"<title>Steam 社区 :: (.*?)</title>", html)
if player_name:
@@ -407,12 +673,15 @@ async def get_user_data(
# remove xml
result["description"] = re.sub(r"<.*?>", "", result["description"])
# background
background_url = re.search(r"background-image: url\( \'(.*?)\' \)", html)
# background:普通背景走 CSS,动画背景用 <video> 的静态首帧;按 URL 缓存
background_url = parse_profile_background_url(html)
if background_url:
background_url = background_url.group(1)
suffix = Path(background_url.split("?")[0]).suffix or ".jpg"
result["background"] = await _fetch(
background_url, default_background, proxy=proxy
background_url,
default_background,
cache_file=cache_path / f"bg_{_url_digest(background_url)}{suffix}",
proxy=proxy,
)
# avatar
@@ -427,6 +696,16 @@ async def get_user_data(
avatar_url, default_avatar, cache_file=avatar_file, proxy=proxy
)
# avatar frame(社区物品,取静态版本);拿不到就不画
frame_url = parse_avatar_frame_url(html)
if frame_url:
result["avatar_frame"] = await _fetch(
frame_url,
b"",
cache_file=cache_path / f"frame_{_url_digest(frame_url)}.png",
proxy=proxy,
)
# recent 2 week play time
# \t<div class="recentgame_quicklinks recentgame_recentplaytime">\r\n\t\t\t\t\t\t\t\t\t<div>15.5 小时（过去 2 周）</div>
play_time_text = re.search(
@@ -437,6 +716,14 @@ async def get_user_data(
play_time_text = play_time_text.group(1)
result["recent_2_week_play_time"] = play_time_text
# 页面没写近两周时长时才用 XML 的 hoursPlayed2Wk 兜底
# (部分账号 XML 恒为 0.0,与页面/API 不符,不能优先采信)
if result["recent_2_week_play_time"] is None and result["hours_2wk"] is not None:
hours_2wk = result["hours_2wk"]
result["recent_2_week_play_time"] = (
"近两周未游玩" if hours_2wk <= 0 else f"{hours_2wk:g} 小时(过去 2 周)"
)
# game data — 先收集信息，再并行下载图片
soup = BeautifulSoup(html, "html.parser")
recent_games = soup.find_all("div", class_="recent_game")
@@ -447,15 +734,22 @@ async def get_user_data(
game_info = {}
game_info["game_name"] = game.find("div", class_="game_name").text.strip()
game_info["game_image_url"] = game.find("img", class_="game_capsule")["src"]
game_info_split = game_info["game_image_url"].split("/")
# 封面 URL 形如 .../steam/apps/{appid}/header.jpg,提取 appid 用于本地化
appid_str = game_info_split[-2]
game_info["appid"] = int(appid_str) if appid_str.isdigit() else None
# 封面 URL 有两种形态(新版多一段 hash):
# .../steam/apps/{appid}/header.jpg
# .../store_item_assets/steam/apps/{appid}/{hash}/capsule_xxx.jpg?t=...
# 取 /apps/ 后的数字段,别用 [-2](新版会取到 hash 导致 appid=None)
appid_match = re.search(r"/apps/(\d+)/", game_info["game_image_url"])
game_info["appid"] = int(appid_match.group(1)) if appid_match else None
if game_info["appid"] is not None:
header_name = f"header_{game_info['appid']}.jpg"
else:
url_digest = hashlib.md5(game_info["game_image_url"].encode()).hexdigest()
header_name = f"header_{url_digest[:10]}.jpg"
download_tasks.append((
game_info, "game_image",
game_info["game_image_url"], default_header_image,
cache_path / f"header_{game_info_split[-2]}.jpg", proxy,
cache_path / header_name, proxy,
))
play_time_text = game.find("div", class_="game_info_details").text.strip()
@@ -581,6 +875,31 @@ def _query_game_names(db_path: Optional[Path], appids: list[int]) -> dict[str, d
}
def _query_game_name_ch_by_en(db_path: Optional[Path], name_en: str) -> Optional[str]:
"""按英文名反查中文名(仅本地映射表,用于 XML 状态里的英文游戏名)
name_en 默认没有索引,首次调用时顺带建,避免每次全表扫描。
"""
if isinstance(db_path, str):
db_path = Path(db_path)
if not name_en or db_path is None:
return None
try:
with closing(_get_game_db_conn(db_path)) as conn:
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_game_names_name_en "
"ON game_names(name_en)"
)
row = conn.execute(
"SELECT name_ch FROM game_names "
"WHERE name_en = ? AND name_ch != '' LIMIT 1",
(name_en,),
).fetchone()
return row[0] if row else None
except sqlite3.Error:
return None
def _query_game_icon_hashes(db_path: Optional[Path], appids: list[int]) -> dict[int, str]:
"""按 appid 批量查询 clienticon hash,返回 {appid: hash}"""
if isinstance(db_path, str):
@@ -1008,11 +1327,77 @@ async def get_play_time(steam_id: str) -> list[dict]:
return game_list
async def get_recently_played(
steam_id: str,
proxy: str = None,
wait: bool = True,
timeout: float = 10,
) -> Optional[list[dict]]:
"""获取最近两周玩过的游戏(含两周/总时长),顺序即 Steam 返回的近两周时长降序
返回 [{"appid", "name", "two_week_hours", "total_hours"}];
无 API 配额返回 None(调用方跳过该增强项);接口失败/资料不公开返回 []。
"""
if not await acquire_steam_quota(wait=wait, timeout=timeout):
logger.warning("Steam API 配额已满,近两周时长查询被跳过")
return None
try:
manager = get_api_key_manager()
except RuntimeError:
logger.error("API key manager 未初始化,无法获取近两周时长")
return []
api_key = await manager.get_next_key()
url = (
"https://api.steampowered.com/IPlayerService/GetRecentlyPlayedGames/v1/"
f"?key={api_key}&steamid={steam_id}"
)
try:
async with httpx.AsyncClient(
proxy=proxy, timeout=httpx.Timeout(30.0)
) as client:
response = await client.get(url)
except Exception as e:
logger.error(f"获取近两周时长失败: {e}")
return []
if response.status_code != 200:
reason = (response.text or "").strip().replace("\n", " ")[:200]
logger.warning(
f"近两周时长 API 失败,状态码: {response.status_code} | {reason}"
)
return []
games = response.json().get("response", {}).get("games") or []
return [
{
"appid": int(g["appid"]),
"name": g.get("name") or "Unknown",
"two_week_hours": round(g.get("playtime_2weeks", 0) / 60, 1),
"total_hours": round(g.get("playtime_forever", 0) / 60, 1),
}
for g in games
if g.get("appid")
]
async def fetch_game_header_image(
appid: int, cache_path: Path, proxy: str = None
) -> Optional[bytes]:
"""按 appid 从公开 CDN 取 header.jpg(资料页没给封面的游戏用),带磁盘缓存"""
url = f"https://cdn.cloudflare.steamstatic.com/steam/apps/{appid}/header.jpg"
data = await _fetch(
url, b"", cache_file=cache_path / f"header_{appid}.jpg", proxy=proxy
)
return data or None
if __name__ == "__main__":
from nonebot.log import logger
import asyncio
data = asyncio.run(get_user_data(76561199135038179, None))
data = asyncio.run(get_user_data(76561198402219863, None))
with open("bg.jpg", "wb") as f:
f.write(data["background"])
tests/test_steam_profile_parse.py
+390
View File
@@ -0,0 +1,390 @@
"""Steam 信息卡解析层 (steam.py 纯函数) 单元测试
steam.py 只依赖 httpx/bs4/nonebot.log + 相对导入的 config/models，不碰
NoneBot 运行时；仿照 test_mc_servers.py 造一棵一次性包骨架（`__path__` 指向
插件目录），相对导入就能自动落到真实文件上，且不触发插件包 __init__.py。
测的是资料页 HTML 与社区 ?xml=1 的解析、状态/风控文案整理，全离线。
样本取自真实抓取结果（76561199135038179 公开资料）裁剪而成。
"""
import importlib.util
import sys
import types
from pathlib import Path
import pytest
_PLUGIN_DIR = (
Path(__file__).resolve().parents[1]
/ "hexi"
/ "plugins"
/ "nonebot_plugin_steam_info"
)
_PKG = "steam_parse_testpkg"
PROFILE_XML = """<?xml version="1.0" encoding="UTF-8" standalone="yes"?><profile>
<steamID64>76561199135038179</steamID64>
<steamID><![CDATA[zhaomaoniu]]></steamID>
<onlineState>online</onlineState>
<stateMessage><![CDATA[Online]]></stateMessage>
<privacyState>public</privacyState>
<vacBanned>0</vacBanned>
<tradeBanState>None</tradeBanState>
<isLimitedAccount>0</isLimitedAccount>
<memberSince>January 29, 2021</memberSince>
<hoursPlayed2Wk>0.0</hoursPlayed2Wk>
</profile>"""
PROFILE_XML_IN_GAME = """<profile>
<onlineState>in-game</onlineState>
<stateMessage><![CDATA[In-Game: Team Fortress 2]]></stateMessage>
<privacyState>public</privacyState>
<memberSince>January 29, 2021</memberSince>
</profile>"""
PROFILE_XML_PRIVATE = """<profile>
<steamID64>76561199135038179</steamID64>
<privacyState>private</privacyState>
</profile>"""
PROFILE_XML_BANNED = """<profile>
<onlineState>offline</onlineState>
<privacyState>public</privacyState>
<vacBanned>1</vacBanned>
<tradeBanState>Trade Banned</tradeBanState>
<isLimitedAccount>1</isLimitedAccount>
<hoursPlayed2Wk>7.5</hoursPlayed2Wk>
</profile>"""
PROFILE_HTML = """
<div class="profile_header_badgeinfo">
<div class="persona_name persona_level">
<div class="friendPlayerLevel lvl_0"><span class="friendPlayerLevelNum">
7
</span></div> 级
</div>
</div>
<div class="profile_item_links">
<div class="profile_count_link ellipsis">
<a href="https://steamcommunity.com/profiles/76561199135038179/games/">
<span class="count_link_label">游戏</span>&nbsp;
<span class="profile_count_link_total"> 1,234 </span>
</a>
</div>
<div class="profile_count_link ellipsis">
<a href="https://steamcommunity.com/profiles/76561199135038179/inventory/">
<span class="count_link_label">库存</span>&nbsp;
<span class="profile_count_link_total"> &nbsp; </span>
</a>
</div>
<div class="profile_count_link ellipsis">
<a href="https://steamcommunity.com/profiles/76561199135038179/recommended/">
<span class="count_link_label">评测</span>&nbsp;
<span class="profile_count_link_total"> 2 </span>
</a>
</div>
<div class="profile_count_link ellipsis">
<a href="https://steamcommunity.com/profiles/76561199135038179/badges/">
<span class="count_link_label">徽章</span>&nbsp;
<span class="profile_count_link_total"> 7 </span>
</a>
</div>
<div class="profile_count_link ellipsis">
<a href="https://steamcommunity.com/profiles/76561199135038179/friends/">
<span class="count_link_label">好友</span>&nbsp;
<span class="profile_count_link_total"> 12 </span>
</a>
</div>
</div>
"""
def _load_by_path(name: str, path: Path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
_STUB_MODULES = ("hexi", "hexi.core", "hexi.core.rate_limit")
def _install_rate_limit_stub() -> dict:
"""steam.py 顶层 `from hexi.core.rate_limit import acquire`，而 hexi.core 包
__init__ 会触发 NoneBot 运行时初始化；纯函数测试用不到限流，塞一个 no-op
acquire 并屏蔽包 __init__（沿用 test_rate_limit.py 的隔离思路）。返回现场用于还原。
"""
saved = {name: sys.modules.get(name) for name in _STUB_MODULES}
hexi_dir = _PLUGIN_DIR.parents[1]
hexi = types.ModuleType("hexi")
hexi.__path__ = [str(hexi_dir)] # type: ignore[attr-defined]
core = types.ModuleType("hexi.core")
core.__path__ = [str(hexi_dir / "core")] # type: ignore[attr-defined]
rate_limit = types.ModuleType("hexi.core.rate_limit")
async def _acquire(*_args, **_kwargs) -> bool:
return True
rate_limit.acquire = _acquire # type: ignore[attr-defined]
sys.modules["hexi"] = hexi
sys.modules["hexi.core"] = core
sys.modules["hexi.core.rate_limit"] = rate_limit
return saved
def _restore_modules(saved: dict) -> None:
for name, module in saved.items():
if module is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = module
@pytest.fixture(scope="module")
def steam():
"""一次性包骨架 + 真实 steam.py（不触发插件 __init__.py）"""
pkg = types.ModuleType(_PKG)
pkg.__path__ = [str(_PLUGIN_DIR)] # type: ignore[attr-defined]
sys.modules[_PKG] = pkg
saved = _install_rate_limit_stub()
try:
yield _load_by_path(f"{_PKG}.steam", _PLUGIN_DIR / "steam.py")
finally:
_restore_modules(saved)
for name in list(sys.modules):
if name == _PKG or name.startswith(f"{_PKG}."):
sys.modules.pop(name, None)
def test_parse_profile_xml_full(steam):
data = steam.parse_profile_xml(PROFILE_XML)
assert data["member_since"] == "2021-01-29"
assert data["online_state"] == "online"
assert data["state_message"] == "Online"
assert data["vac_banned"] is False
assert data["trade_ban_state"] is None
assert data["limited_account"] is False
assert data["hours_2wk"] == 0.0
assert data["privacy_state"] == "public"
def test_parse_profile_xml_in_game(steam):
data = steam.parse_profile_xml(PROFILE_XML_IN_GAME)
assert data["online_state"] == "in-game"
assert data["state_message"] == "In-Game: Team Fortress 2"
def test_parse_profile_xml_private_profile(steam):
data = steam.parse_profile_xml(PROFILE_XML_PRIVATE)
assert data["privacy_state"] == "private"
assert data["member_since"] is None
assert data["online_state"] is None
assert data["hours_2wk"] is None
assert data["vac_banned"] is False
def test_parse_profile_xml_bans_and_hours(steam):
data = steam.parse_profile_xml(PROFILE_XML_BANNED)
assert data["vac_banned"] is True
assert data["trade_ban_state"] == "Trade Banned"
assert data["limited_account"] is True
assert data["hours_2wk"] == 7.5
def test_parse_profile_xml_invalid_text(steam):
assert steam.parse_profile_xml("not xml at all") == {}
def test_member_since_localized_fallback(steam):
"""本地化/未知格式：能识别的归一化，认不出就原样保留"""
xml_zh = "<profile><memberSince>2021年1月29日</memberSince></profile>"
assert steam.parse_profile_xml(xml_zh)["member_since"] == "2021-01-29"
xml_odd = "<profile><memberSince>29/01/2021</memberSince></profile>"
assert steam.parse_profile_xml(xml_odd)["member_since"] == "29/01/2021"
def test_parse_profile_extras(steam):
data = steam.parse_profile_extras(PROFILE_HTML)
assert data["level"] == 7
# 库存是空占位被跳过；千位逗号被去掉
assert data["counts"] == {"游戏": 1234, "评测": 2, "徽章": 7, "好友": 12}
def test_parse_profile_extras_missing(steam):
data = steam.parse_profile_extras("<html><body>无资料</body></html>")
assert data["level"] is None
assert data["counts"] == {}
IN_GAME_HTML = """
<div class="responsive_status_info">
<div class="profile_in_game persona in-game">
<div class="profile_in_game_header">当前正在游戏</div>
<div class="profile_in_game_name"> Subnautica 2：异星水域 </div>
<div class="profile_in_game_joingame"><a href="steam://joinlobby/1962700/xyz/">加入游戏</a></div>
</div>
</div>
"""
def test_build_profile_status_in_game(steam):
# XML 两种写法都要认:"In-Game: X" 和 "In-Game<br/>X"
status = steam.build_profile_status(
"in-game", "In-Game: Team Fortress 2", "public"
)
assert status["text"] == "Team Fortress 2"
assert status["kind"] == "in-game"
assert status["game_name_en"] == "Team Fortress 2"
status = steam.build_profile_status("in-game", "In-Game<br/>VRChat", "public")
assert status["text"] == "VRChat"
assert status["game_name_en"] == "VRChat"
def test_parse_in_game_name(steam):
assert steam.parse_in_game_name(IN_GAME_HTML) == "Subnautica 2：异星水域"
assert steam.parse_in_game_name(PROFILE_HTML) is None
def test_build_profile_status_prefers_page_in_game(steam):
"""资料页「当前正在游戏」优先于 XML(XML 会滞后,且资料页已本地化)"""
status = steam.build_profile_status(
"online", "Online", "public", current_game="Subnautica 2：异星水域"
)
assert status == {"text": "Subnautica 2：异星水域", "kind": "in-game"}
def test_build_profile_status_online_and_offline(steam):
# 默认英文状态换中文;自定义状态原样保留
assert steam.build_profile_status("online", "Online", "public") == {
"text": "在线",
"kind": "online",
}
assert steam.build_profile_status("online", "摸鱼中", "public") == {
"text": "摸鱼中",
"kind": "online",
}
assert steam.build_profile_status("offline", None, "public") == {
"text": "离线",
"kind": "offline",
}
def test_build_profile_status_private_overrides(steam):
status = steam.build_profile_status("offline", None, "private")
assert status == {"text": "资料私密,部分信息不可见", "kind": "private"}
BACKDROP_CSS = """
<style>
.profile_header_bg { background-image: url( 'https://shared.fastly.steamstatic.com/community_assets/images/items/2458530/abc.jpg' ); }
</style>
"""
BACKDROP_CSS_TIGHT = """
<style>
body { background-image: url("https://cdn.cloudflare.steamstatic.com/steamcommunity/public/images/items/123/def.png"); }
</style>
"""
BACKDROP_CSS_WITH_DECOY = """
<style>
.btn { background-image: url('https://example.com/logo.png'); }
.bg { background-image: url('https://shared.fastly.steamstatic.com/community_assets/images/items/999/real.jpg'); }
</style>
"""
BACKDROP_ANIMATED = """
<div class="profile_animated_background">
<video playsinline autoplay muted loop poster="https://shared.fastly.steamstatic.com/community_assets/images/items/1173010/aaa.jpg">
<source src="https://shared.fastly.steamstatic.com/community_assets/images/items/1173010/bbb.webm" type="video/webm">
<source src="https://shared.fastly.steamstatic.com/community_assets/images/items/1173010/ccc.mp4" type="video/mp4">
</video>
</div>
"""
AVATAR_FRAME = """
<div class="playerAvatar profile_header_size online">
<div class="playerAvatarAutoSizeInner">
<div class="profile_avatar_frame">
<picture>
<source media="(prefers-reduced-motion: reduce)" srcset="https://shared.fastly.steamstatic.com/community_assets/images/items/1263950/static.png"></source>
<source srcset="https://shared.fastly.steamstatic.com/community_assets/images/items/1263950/animated.png"></source>
<img src="https://shared.fastly.steamstatic.com/community_assets/images/items/1263950/animated.png">
</picture>
</div>
<picture>
<img srcset="https://avatars.fastly.steamstatic.com/aaa_full.jpg">
</picture>
</div>
</div>
"""
AVATAR_FRAME_IMG_ONLY = """
<div class="playerAvatar">
<div class="profile_avatar_frame">
<img src="https://shared.fastly.steamstatic.com/community_assets/images/items/1/only.png">
</div>
</div>
"""
def test_background_css_variants(steam):
assert steam.parse_profile_background_url(BACKDROP_CSS) == (
"https://shared.fastly.steamstatic.com/community_assets/images/items/2458530/abc.jpg"
)
assert steam.parse_profile_background_url(BACKDROP_CSS_TIGHT) == (
"https://cdn.cloudflare.steamstatic.com/steamcommunity/public/images/items/123/def.png"
)
def test_background_prefers_community_item(steam):
"""页面里其他 url() 不一定是背景,优先挑 /items/ 的社区物品图"""
assert steam.parse_profile_background_url(BACKDROP_CSS_WITH_DECOY) == (
"https://shared.fastly.steamstatic.com/community_assets/images/items/999/real.jpg"
)
def test_background_animated_falls_back_to_poster(steam):
"""动画背景没有 CSS 图,取 <video poster> 的静态首帧"""
assert steam.parse_profile_background_url(BACKDROP_ANIMATED) == (
"https://shared.fastly.steamstatic.com/community_assets/images/items/1173010/aaa.jpg"
)
assert steam.parse_profile_background_url("<html><body>无背景</body></html>") is None
def test_avatar_frame_prefers_static_source(steam):
assert steam.parse_avatar_frame_url(AVATAR_FRAME) == (
"https://shared.fastly.steamstatic.com/community_assets/images/items/1263950/static.png"
)
def test_avatar_frame_img_fallback(steam):
assert steam.parse_avatar_frame_url(AVATAR_FRAME_IMG_ONLY) == (
"https://shared.fastly.steamstatic.com/community_assets/images/items/1/only.png"
)
# 没有头像框(普通头像的 <picture>)不该误判
assert steam.parse_avatar_frame_url(PROFILE_HTML) is None
def test_describe_ban_priority(steam):
assert steam.describe_ban(True, "Trade Banned", True) == {
"text": "VAC 封禁",
"kind": "vac",
}
assert steam.describe_ban(False, "Trade Banned", False) == {
"text": "交易封禁",
"kind": "trade",
}
assert steam.describe_ban(False, "Probation", False) == {
"text": "交易限制",
"kind": "trade",
}
assert steam.describe_ban(False, None, True) == {
"text": "受限账号",
"kind": "limited",
}
assert steam.describe_ban(False, None, False) is None