diff --git a/CHANGELOG.md b/CHANGELOG.md deleted file mode 100644 index 158b0ab..0000000 --- a/CHANGELOG.md +++ /dev/null @@ -1,25 +0,0 @@ -# 更新日志 - -## 0701-1140 - -1. 新增推送总开关配置项，支持选择开启的推送事件 -2. 添加推送相关数据库字段与操作方法 -3. 实现开始/结束游戏推送逻辑与开关控制 -4. 优化推送检测逻辑，减少无效渲染 -5. 更新帮助文档与更新日志 -6. 替换旧版背景图片资源 - -## 0701-1011 - -### 新增 -- 帮助菜单 - -## 0630-2223 - -### 新增 -- 结束游戏绘图 - -### 修改 -- 开始游戏绘图用户名上调 5px -- 开始游戏背景图改为渐变色 -- 优化绑定解绑提示 \ No newline at end of file diff --git a/README.md b/README.md index e5fdd62..5a375a5 100644 --- a/README.md +++ b/README.md @@ -35,9 +35,8 @@ ## 丨命令列表 -### steam帮助 -Steam帮助菜单 - +### steam帮助(图片更新可能不及时) +Steam帮助菜单 ### 命令说明 #### 绑定账号 @@ -68,6 +67,7 @@ | `steam游戏成就123456` | 查看appid123456的成就详情 | | `steam开启成就推送` | 开启自己的成就推送功能 | | `steam关闭成就推送` | 关闭自己的成就推送功能 | +| `steam玩什么` | 从自己的游戏库随机挑3个游戏 | #### 商店相关 | 命令 | 说明 | @@ -76,11 +76,16 @@ | `steam取消订阅降价123456` | 取消订阅商店游戏降价信息 | | `steam订阅降价查看` | 查看订阅的商店游戏降价信息 | +#### 社交相关 +| 命令 | 说明 | +|------|:------:| +| `@xx steam加好友123456` | 获取被 @ 用户的好友码 | +| `steam群排行` | 查看群游玩时长排行榜 | + #### 其他服务 | 命令 | 说明 | |------|:------:| | `steam帮助` | 呼出本插件帮助菜单 | -| `@xx steam加好友` | 获取被 @ 用户的好友码 | | `清除全部缓存` | 清除全部缓存 | ## 效果图 @@ -89,14 +94,18 @@ ### steam成就推送 - + ### steam开始游戏 / 结束游戏推送 - + ### steam游戏墙 +## steam玩什么 + + + ## 计划功能 @@ -104,10 +113,10 @@ - [ ] 玩家上下线状态推送 - [ ] 带游玩时长的游戏库存图片 - [x] 游戏降价 / 打折信息订阅推送 -- [ ] steam玩什么 (随机抽游戏) +- [x] steam玩什么 (随机抽游戏) - [x] 游戏名 / 成就名本地化名字获取 - [ ] 总结推送功能 (定时总结周期内游戏情况而不每次状态变化都推送) -- [ ] 群游玩时长排行榜 +- [x] 群游玩时长排行榜 - [x] 隐藏 steamid / 好友码 @@ -117,7 +126,7 @@ - 本项目仅供学习使用，请勿用于商业用途 - [GPL-3.0 License](LICENSE) - +- 部分功能没有写绘图，以后再说 ## 致谢 diff --git a/SteamUID/SteamHelp/help.json b/SteamUID/SteamHelp/help.json index 4af67e3..94199d6 100644 --- a/SteamUID/SteamHelp/help.json +++ b/SteamUID/SteamHelp/help.json @@ -115,6 +115,14 @@ "need_ck": false, "need_sk": false, "need_admin": false + }, + { + "name": "玩什么", + "desc": "从自己的游戏库随机挑3个游戏", + "eg": "玩什么", + "need_ck": false, + "need_sk": false, + "need_admin": false } ] }, @@ -147,21 +155,34 @@ } ] }, - "其他服务": { - "desc": "一些杂项服务", + "社交相关": { + "desc": "加好友和排行榜功能", "data": [ { - "name": "帮助", - "desc": "呼出本插件帮助菜单", - "eg": "帮助", + "name": "加好友", + "desc": "获取被 @ 用户的好友码", + "eg": "加好友", "need_ck": false, "need_sk": false, "need_admin": false }, { - "name": "@xx 加好友", - "desc": "获取被 @ 用户的好友码", - "eg": "加好友", + "name": "群排行", + "desc": "查看群游玩时长排行榜", + "eg": "群排行", + "need_ck": false, + "need_sk": false, + "need_admin": false + } + ] + }, + "其他服务": { + "desc": "一些杂项服务", + "data": [ + { + "name": "帮助", + "desc": "呼出本插件帮助菜单", + "eg": "帮助", "need_ck": false, "need_sk": false, "need_admin": false diff --git a/SteamUID/SteamLibarary/__init__.py b/SteamUID/SteamLibarary/__init__.py index b42ff74..1f12a28 100644 --- a/SteamUID/SteamLibarary/__init__.py +++ b/SteamUID/SteamLibarary/__init__.py @@ -6,7 +6,7 @@ from ..utils.exceptions import SteamError, SteamValidationError from ..utils.target import resolve_target_steamid64 -from .library_service import build_library_wall +from .library_service import build_library_wall, build_random_pick library_SV = SV("steam库存相关") @@ -26,3 +26,19 @@ async def get_steamlibrary_image(bot: Bot, ev: Event): except Exception as e: logger.exception(f"[steam库存] 未知错误: {e!r}") await bot.send(f"发生未知错误: {e}") + +@library_SV.on_command(("玩什么",)) +async def get_my_steamlibrary_image(bot: Bot, ev: Event): + try: + steamid64 = await resolve_target_steamid64(ev, ev.text.strip()) + if not steamid64: + raise SteamValidationError("请先绑定 steam 账号") + + await bot.send("正在从游戏库中随机挑选......") + img_bytes = await build_random_pick(steamid64) + await bot.send(MessageSegment.image(img_bytes)) + except SteamError as e: + await bot.send(str(e)) + except Exception as e: + logger.exception(f"[steam库存] 未知错误: {e!r}") + await bot.send(f"发生未知错误: {e}") \ No newline at end of file diff --git a/SteamUID/SteamLibarary/library_service.py b/SteamUID/SteamLibarary/library_service.py index ebb4e31..ee84214 100644 --- a/SteamUID/SteamLibarary/library_service.py +++ b/SteamUID/SteamLibarary/library_service.py @@ -1,3 +1,4 @@ +import random from io import BytesIO from gsuid_core.data_store import get_res_path @@ -7,6 +8,7 @@ from ..SteamConfig.interface import SteamAPI from ..utils.api import get_steamlibrary_by_steamid64 from ..utils.exceptions import SteamConfigError, SteamValidationError +from ..utils.PIL.draw import draw_what_to_play from ..utils.PIL.steam_wall import build_wall from ..utils.utils import batch_download_images @@ -50,3 +52,44 @@ async def build_library_wall(steamid64: str) -> bytes: buf = BytesIO() wall.save(buf, format="JPEG", quality=pic_quality, subsampling=0) return buf.getvalue() + + +async def build_random_pick(steamid64: str) -> bytes: + """从用户 Steam 游戏库中随机选取 3 款游戏，生成推荐图片。""" + api_key = SteamConfig.get_config("SteamWebAPIKey").data + if not api_key: + raise SteamConfigError("请先配置 steam web api key") + + library = await get_steamlibrary_by_steamid64(api_key, steamid64) + games = library.get("games") + if games is None: + raise SteamValidationError("获取 steam 游戏库列表失败") + if not games: + raise SteamValidationError("该 steam 账号暂无游戏库存") + pick_count = min(3, len(games)) + picks = random.sample(games, pick_count) + + for game in picks: + appid = str(game.get("appid", "")) + name = game.get("name", "未知游戏") + playtime = ( + game.get("playtime_forever", 0) or + game.get("playtime_windows_forever", 0) + + game.get("playtime_mac_forever", 0) + + game.get("playtime_linux_forever", 0) + + game.get("playtime_deck_forever", 0) + ) + cover_url = SteamAPI.GetGameCoverImageURL(appid, variant='library_600x900') + game_data.append({ + "appid": appid, + "name": name, + "playtime": playtime, + "cover_url": cover_url, + }) }) + + img = await draw_what_to_play(game_data) + img = img.convert("RGB") + + buf = BytesIO() + img.save(buf, format="JPEG", quality=pic_quality, subsampling=0) + return buf.getvalue() diff --git a/SteamUID/SteamPoll/poll_service.py b/SteamUID/SteamPoll/poll_service.py index 204e6be..f13866d 100644 --- a/SteamUID/SteamPoll/poll_service.py +++ b/SteamUID/SteamPoll/poll_service.py @@ -1,4 +1,5 @@ import json +import time from gsuid_core.logger import logger from gsuid_core.segment import MessageSegment @@ -18,6 +19,7 @@ SteamBind, SteamArchivementInfo, SteamPriceInfo, + SteamPlayRecord, ) from ..utils.PIL.draw import draw_game_status_photo, draw_archivements_photo from ..utils.steam_status import ( @@ -173,13 +175,32 @@ async def _dispatch_to_subs(subs, send_msg, push_column, steamid64) -> None: async def flush_status_updates(update_list) -> None: - """将有变化的状态数据写回数据库""" + """将有变化的状态数据写回状态轮询数据库""" for steamid64, info in update_list: await SteamIDInfo.upsert_steamuserinfo( steamid64, json.dumps(info, ensure_ascii=False) ) - +async def update_game_record(push_list) -> None: + """写入游戏记录数据库""" + now = int(time.time()) + for info, old_info in push_list: + steamid64 = info.get("steamid") + if not steamid64: + continue + old_appid = old_info.get("gameid", "") + new_appid = info.get("gameid", "") + # 结束旧游戏（覆盖"切换游戏"与"退出游戏"两种场景） + if old_appid: + ret = await SteamPlayRecord.upsert_record(steamid64, old_appid, end_ts=now) + if ret != 0: + logger.warning( + f"[SteamPoll] 结束游玩记录失败: steamid={steamid64} appid={old_appid}" + ) + # 开始新游戏 + if new_appid: + await SteamPlayRecord.upsert_record(steamid64, new_appid, now) + async def poll_and_push_game_status() -> None: """游戏状态轮询主入口：拉取状态、检测变化、推送、更新基线、落盘。""" try: @@ -198,6 +219,8 @@ async def poll_and_push_game_status() -> None: await process_game_status_push(push_list, game_info_map) await update_achievement_baselines(push_list) await flush_status_updates(update_list) + await update_game_record(push_list) + except Exception as error: logger.warning(f"[SteamPoll] 游戏状态轮询失败: {error!r}") diff --git a/SteamUID/SteamRanking/__init__.py b/SteamUID/SteamRanking/__init__.py new file mode 100644 index 0000000..46b64d5 --- /dev/null +++ b/SteamUID/SteamRanking/__init__.py @@ -0,0 +1,54 @@ +from gsuid_core.sv import SV +from gsuid_core.bot import Bot +from gsuid_core.models import Event +from gsuid_core.logger import logger +from gsuid_core.segment import MessageSegment + +from ..utils.exceptions import SteamError, SteamValidationError +# from ..utils.target import resolve_target_steamid64 +from ..utils.utils import time_convert_s +from .ranking_service import get_group_ranking_list + +ranking_sv = SV("steam排名服务") + +@ranking_sv.on_command(("群排行", "群排名")) +async def group_ranking(bot: Bot, ev: Event): + """按用户游戏时长从高到低取5位返回""" + try: + if not ev.group_id: + raise SteamValidationError("请在群聊中使用此功能") + + ranking_list = await get_group_ranking_list(ev.group_id) + top5 = ranking_list[:5] + + if not top5: + await bot.send("本群暂无游戏时长排行数据") + return + + text = f"本群游戏时长排行：\n" + for i, item in enumerate(top5, 1): + text += f"{i}. {item['user_id']} ({time_convert_s(item['total_duration'])})\n" + + await bot.send(text) + + except SteamError as e: + await bot.send(str(e)) + except Exception as e: + logger.exception(f"[SteamRanking - 群排行] 未知错误: {e!r}") + await bot.send("发生未知错误，请联系管理员查看控制台") + +""" +============鸽了============= + +@ranking_sv.on_command(("我的排行", "我的排名")) +async def my_ranking(bot: Bot, ev: Event): + 获取用户在群中的排名，应为总排行 + ... + 用户所在排行位置 + +@ranking_sv.on_command(("游戏排行", "游戏排行")) +async def game_ranking(bot: Bot, ev: Event): + 获取用户在某appid在某群中的用户游玩时长排行 + try: + if not ev.group_id: + raise SteamValidationError("请在群聊中使用此功能") + steamid64 = await resolve_target_steamid64(ev) +""" diff --git a/SteamUID/SteamRanking/ranking_service.py b/SteamUID/SteamRanking/ranking_service.py new file mode 100644 index 0000000..1cb3402 --- /dev/null +++ b/SteamUID/SteamRanking/ranking_service.py @@ -0,0 +1,65 @@ +from ..utils.exceptions import SteamError +from ..utils.database.models import SteamBind, SteamPlayRecord + + +async def get_group_ranking_list(group_id: str) -> list[dict]: + """获取群排名列表 + + 流程： + 1. 从 SteamBind 取群内所有绑定 → 构建 steamid64 → user_id 映射 + 2. 收集所有 steamid64，批量查询已结束的游玩记录 + 3. 计算每条记录时长 (end_ts - start_ts)，按 user_id 累加 + 4. 按总时长降序排序 + + Returns: + list[dict]: 每项包含: + - user_id (str): 平台用户ID + - total_duration (int): 总游戏时长（秒） + - steamid64s (list[str]): 该用户绑定的所有SteamID + """ + # 第1步：取群绑定列表，构建映射 + binds = await SteamBind.get_binds_by_group(group_id) + if not binds: + return [] + + steamid_to_user: dict[str, str] = {} + user_steamids: dict[str, list[str]] = {} + all_steamids: list[str] = [] + + for bind in binds: + sid = bind.steamid64 + uid = bind.user_id + steamid_to_user[sid] = uid + all_steamids.append(sid) + if uid not in user_steamids: + user_steamids[uid] = [] + if sid not in user_steamids[uid]: + user_steamids[uid].append(sid) + + # 第2步：批量查询已结束的游玩记录 + # with_session 重试耗尽后会隐式返回 None（区别于空列表的正常业务结果） + records = await SteamPlayRecord.get_records_by_steamids(all_steamids) + if records is None: + raise SteamError("查询游玩记录失败，请稍后重试") + + # 第3步：计算时长并按 user_id 累加 + user_durations: dict[str, int] = {} + for record in records: + uid = steamid_to_user.get(record.steamid64) + if uid is None: + continue + duration = record.end_ts - record.start_ts # type: ignore + user_durations[uid] = user_durations.get(uid, 0) + duration + + # 第4步：构建结果列表并按总时长降序排序 + ranking_list = [ + { + "user_id": uid, + "total_duration": duration, + "steamid64s": user_steamids.get(uid, []), + } + for uid, duration in user_durations.items() + ] + ranking_list.sort(key=lambda x: x["total_duration"], reverse=True) + + return ranking_list diff --git a/SteamUID/utils/PIL/draw.py b/SteamUID/utils/PIL/draw.py index 5c5381a..530afe8 100644 --- a/SteamUID/utils/PIL/draw.py +++ b/SteamUID/utils/PIL/draw.py @@ -611,3 +611,194 @@ def _section_h(items: list[dict]) -> int: cur_y += _BIND_SECTION_GAP return canvas + + +# ==================== 玩什么推荐卡片 ==================== + +_PLAY_BG_TOP = (0x17, 0x1a, 0x21) +_PLAY_BG_BOTTOM = (0x1b, 0x28, 0x38) +_PLAY_CARD_BG: tuple[int, int, int, int] = (0x2a, 0x47, 0x5e, 255) +_PLAY_CARD_BG_ALT: tuple[int, int, int, int] = (0x1e, 0x36, 0x4a, 255) +_PLAY_WHITE: tuple[int, int, int, int] = (0xff, 0xff, 0xff, 255) +_PLAY_GRAY: tuple[int, int, int, int] = (0x8f, 0x98, 0xa0, 255) +_PLAY_BLUE: tuple[int, int, int, int] = (0x66, 0xc0, 0xf4, 255) +_PLAY_PLACEHOLDER: tuple[int, int, int, int] = (0x3a, 0x4f, 0x5e, 255) +_PLAY_DIVIDER: tuple[int, int, int, int] = (0x3d, 0x5c, 0x78, 255) + +_PLAY_SCALE = 2.0 +_PLAY_LOGICAL_W = 600 + + +def _play_A(v: float) -> int: + """逻辑像素 -> 实际像素""" + return round(v * _PLAY_SCALE) + + +def _play_fmt_playtime(minutes: int) -> str: + """格式化游玩时长""" + if minutes <= 0: + return "未游玩" + if minutes < 60: + return f"{minutes} 分钟" + hours = minutes / 60 + if hours < 100: + return f"{hours:.1f} 小时" + return f"{round(hours)} 小时" + + +async def _play_load_cover( + cover_url: str, appid: str, target_w: int, target_h: int +) -> Image.Image: + """下载游戏封面并缩放到目标尺寸；失败返回纯色占位图。""" + if not cover_url: + return Image.new("RGBA", (target_w, target_h), _PLAY_PLACEHOLDER) + try: + cache_path = cache_path_for_url(cover_url, appid) + img = await _load_or_download(cover_url, cache_path) + img = img.resize((target_w, target_h), Image.Resampling.LANCZOS) + except Exception: + return Image.new("RGBA", (target_w, target_h), _PLAY_PLACEHOLDER) + return img + +def _play_paste_rounded_top( + canvas: Image.Image, + img: Image.Image, + pos: tuple[int, int], + radius: int, +) -> None: + """将 img 粘贴到 canvas，仅圆角化顶部两角，底部保持直角。""" + w, h = img.size + mask = Image.new("L", (w, h), 0) + md = ImageDraw.Draw(mask) + md.rounded_rectangle((0, 0, w - 1, h - 1), radius=radius, fill=255) + md.rectangle((0, radius, w, h), fill=255) + canvas.paste(img, pos, mask) + + +async def draw_what_to_play(picks: list[dict]) -> Image.Image: + """ + 绘制"今天玩什么"推荐卡片。 + + 参数: + picks: [{"appid": str, "name": str, "playtime": int, "cover_url": str}, ...] + 长度 1-3 + + 返回: + PIL.Image.Image —— RGBA 画布 + """ + A = _play_A + + # —— 布局常量（逻辑像素）—— + SIDE_PAD = 20 + CARD_GAP = 10 + CARD_W = 180 + CARD_RADIUS = 10 + COVER_PAD = 6 + COVER_RADIUS = 6 + TOP_PAD = 24 + BOTTOM_PAD = 24 + TITLE_H = 40 + SUBTITLE_H = 20 + TITLE_GAP = 8 + SECTION_GAP = 20 + NAME_TOP_PAD = 10 + NAME_H = 22 + NAME_PLAY_GAP = 4 + PLAYTIME_H = 16 + NAME_BOT_PAD = 4 + + # 派生尺寸 + cover_w = CARD_W - 2 * COVER_PAD # 168 + cover_h = round(cover_w * 1.5) # 252 (2:3) + name_area_h = NAME_TOP_PAD + NAME_H + NAME_PLAY_GAP + PLAYTIME_H + NAME_BOT_PAD # 56 + card_h = COVER_PAD + cover_h + name_area_h # 314 + canvas_h = TOP_PAD + TITLE_H + TITLE_GAP + SUBTITLE_H + SECTION_GAP + card_h + BOTTOM_PAD + + W_act = A(_PLAY_LOGICAL_W) + H_act = A(canvas_h) + + # —— 创建画布 + 渐变背景 —— + canvas = Image.new("RGBA", (W_act, H_act)) + draw_vertical_gradient(canvas, W_act, H_act, _PLAY_BG_TOP, _PLAY_BG_BOTTOM) + draw = ImageDraw.Draw(canvas) + + # —— 字体 —— + font_title = core_font(A(30)) + font_subtitle = core_font(A(14)) + font_name = core_font(A(16)) + font_playtime = core_font(A(12)) + + # —— 标题 —— + title_text = "今天玩什么" + title_x = (W_act - draw.textlength(title_text, font_title)) / 2 + title_y = text_y_for_center(A(TOP_PAD) + A(TITLE_H) / 2, font_title, title_text) + draw.text((title_x, title_y), title_text, font=font_title, fill=_PLAY_BLUE) + + # —— 副标题 —— + subtitle_text = "从你的游戏库中随机推荐" + sub_x = (W_act - draw.textlength(subtitle_text, font_subtitle)) / 2 + sub_y = text_y_for_center( + A(TOP_PAD + TITLE_H + TITLE_GAP) + A(SUBTITLE_H) / 2, + font_subtitle, subtitle_text, + ) + draw.text((sub_x, sub_y), subtitle_text, font=font_subtitle, fill=_PLAY_GRAY) + + # —— 分割线 —— + divider_y = A(TOP_PAD + TITLE_H + TITLE_GAP + SUBTITLE_H + SECTION_GAP // 2) + draw.line( + [(A(SIDE_PAD), divider_y), (A(_PLAY_LOGICAL_W - SIDE_PAD), divider_y)], + fill=_PLAY_DIVIDER, width=A(1), + ) + + # —— 并发下载封面 —— + cover_target_w = A(cover_w) + cover_target_h = A(cover_h) + covers = await asyncio.gather( + *[_play_load_cover( + g["cover_url"], g["appid"], cover_target_w, cover_target_h + ) for g in picks] + ) + + # —— 绘制卡片 —— + cards_top = A(TOP_PAD + TITLE_H + TITLE_GAP + SUBTITLE_H + SECTION_GAP) + + for i, game in enumerate(picks): + cx = A(SIDE_PAD) + A(i * (CARD_W + CARD_GAP)) + cy = cards_top + cw = A(CARD_W) + ch = A(card_h) + + # 卡片背景（圆角矩形） + card_bg = _PLAY_CARD_BG if i % 2 == 0 else _PLAY_CARD_BG_ALT + draw.rounded_rectangle( + (cx, cy, cx + cw, cy + ch), radius=A(CARD_RADIUS), fill=card_bg, + ) + + # 封面图（顶部圆角） + cover = covers[i] + cover_x = cx + A(COVER_PAD) + cover_y = cy + A(COVER_PAD) + _play_paste_rounded_top(canvas, cover, (cover_x, cover_y), A(COVER_RADIUS)) + + # 游戏名 + name_text = _truncate_to_width( + game["name"], font_name, cw - 2 * A(COVER_PAD), + ) + name_center_x = cx + cw // 2 + name_cy = cover_y + cover_target_h + A(NAME_TOP_PAD) + A(NAME_H) / 2 + name_y = text_y_for_center(name_cy, font_name, name_text) + draw.text( + (_center_text_x(name_center_x, name_text, font_name), name_y), + name_text, font=font_name, fill=_PLAY_WHITE, + ) + + # 游玩时长 + playtime_text = _play_fmt_playtime(game["playtime"]) + playtime_cy = name_cy + A(NAME_H) / 2 + A(NAME_PLAY_GAP) + A(PLAYTIME_H) / 2 + playtime_y = text_y_for_center(playtime_cy, font_playtime, playtime_text) + draw.text( + (_center_text_x(name_center_x, playtime_text, font_playtime), playtime_y), + playtime_text, font=font_playtime, fill=_PLAY_GRAY, + ) + + return canvas diff --git a/SteamUID/utils/database/__init__.py b/SteamUID/utils/database/__init__.py index 6c38e71..488c048 100644 --- a/SteamUID/utils/database/__init__.py +++ b/SteamUID/utils/database/__init__.py @@ -1,4 +1,4 @@ -from .models import SteamBind, SteamIDInfo, SteamArchivementInfo, SteamPriceInfo +from .models import SteamBind, SteamIDInfo, SteamArchivementInfo, SteamPriceInfo, SteamPlayRecord from .models_cache import SteamApiCache, SteamArchivementCache # 内部缓存表，不注册到控制台 __all__ = [ @@ -6,6 +6,7 @@ "SteamIDInfo", "SteamArchivementInfo", "SteamPriceInfo", + "SteamPlayRecord", "SteamApiCache", "SteamArchivementCache", ] diff --git a/SteamUID/utils/database/admin.py b/SteamUID/utils/database/admin.py index a462729..9987590 100644 --- a/SteamUID/utils/database/admin.py +++ b/SteamUID/utils/database/admin.py @@ -1,6 +1,6 @@ from gsuid_core.webconsole.mount_app import PageSchema, GsAdminModel, site -from .models import SteamIDInfo, SteamBind, SteamArchivementInfo, SteamPriceInfo +from .models import SteamIDInfo, SteamBind, SteamArchivementInfo, SteamPriceInfo, SteamPlayRecord @@ -44,3 +44,13 @@ class SteamPriceInfoAdmin(GsAdminModel): ) # type: ignore model = SteamPriceInfo + +@site.register_admin +class SteamPlayRecordAdmin(GsAdminModel): + pk_name = "id" + page_schema = PageSchema( + label="Steam游玩记录管理", + icon="fa fa-database", + ) # type: ignore + + model = SteamPlayRecord diff --git a/SteamUID/utils/database/models.py b/SteamUID/utils/database/models.py index 9a3179b..3468aec 100644 --- a/SteamUID/utils/database/models.py +++ b/SteamUID/utils/database/models.py @@ -12,6 +12,7 @@ T_SteamIDInfo = TypeVar("T_SteamIDInfo", bound="SteamIDInfo") T_SteamArchivementInfo = TypeVar("T_SteamArchivementInfo", bound="SteamArchivementInfo") T_SteamPriceInfo = TypeVar("T_SteamPriceInfo", bound="SteamPriceInfo") +T_SteamPlayRecord = TypeVar("T_SteamPlayRecord", bound="SteamPlayRecord") class SteamIDInfo(BaseIDModel, table=True): __table_args__: Dict[str, Any] = {"extend_existing": True} @@ -381,6 +382,18 @@ async def get_binds_by_user( result = await session.execute(stmt) return list(result.scalars().all()) + @classmethod + @with_session + async def get_binds_by_group( + cls: Type[T_SteamBind], + session: AsyncSession, + group_id: str, + ) -> list["SteamBind"]: + """按群查所有绑定（用于排行榜：群 → steamid列表 → user_id映射）""" + stmt = select(cls).where(cls.group_id == group_id) + result = await session.execute(stmt) + return list(result.scalars().all()) + @classmethod @with_session async def delete_bind( @@ -581,4 +594,124 @@ async def get_main_id( return result.scalars().first() +class SteamPlayRecord(BaseIDModel, table=True): + """Steam游戏游玩记录表（用于游戏排行榜及衍生功能）""" + __table_args__: Dict[str, Any] = {"extend_existing": True} + + steamid64: str = Field(default=None, index=True, title="SteamID64") + appid: str = Field(default=None, index=True, title="游戏AppID") + start_ts: int = Field(default=None, title="开始游戏时间(Unix时间戳)") + end_ts: Optional[int] = Field(default=None, title="结束游戏时间(Unix时间戳,NULL=进行中)") + + @classmethod + @with_session + async def upsert_record( + cls: Type[T_SteamPlayRecord], + session: AsyncSession, + steamid64: str, + appid: str, + start_ts: Optional[int] = None, + end_ts: Optional[int] = None, + ) -> int: + """ + 写入游玩记录（upsert）。 + - 不传 end_ts（开始游戏）：新增一条记录，start_ts 必填。 + - 传了 end_ts（结束游戏）：按 steamid64 + appid + end_ts IS NULL + 定位进行中的记录并写入 end_ts，start_ts 此时忽略。 + 一个玩家同一时间在同一个 appid 只会有一条 end_ts 为 NULL 的记录， + 因此该定位方式不会产生歧义。 + 返回 0 表示成功，-1 表示结束游戏时未找到进行中的记录。 + """ + if end_ts is None: + session.add( + cls( + steamid64=steamid64, # type: ignore + appid=appid, # type: ignore + start_ts=start_ts, # type: ignore + end_ts=None, # type: ignore + ) + ) + return 0 + + # 结束游戏：查找进行中的记录 + stmt = ( + select(cls) + .where( + cls.steamid64 == steamid64, + cls.appid == appid, + cls.end_ts.is_(None), # type: ignore + ) + .order_by(cls.id.desc()) # type: ignore + ) + result = await session.execute(stmt) + existing = result.scalars().first() + if existing is None: + return -1 + existing.end_ts = end_ts + session.add(existing) + return 0 + + @classmethod + @with_session + async def delete_record( + cls: Type[T_SteamPlayRecord], + session: AsyncSession, + record_id: int, + ) -> int: + """ + 按主键删除一条游玩记录。 + 0: 成功 + -1: 未找到匹配记录 + """ + stmt = delete(cls).where(cls.id == record_id) # type: ignore + result = await session.execute(stmt) + if result.rowcount and result.rowcount > 0: # type: ignore + return 0 + return -1 + + @classmethod + @with_session + async def get_records( + cls: Type[T_SteamPlayRecord], + session: AsyncSession, + steamid64: Optional[str] = None, + appid: Optional[str] = None, + end_before: Optional[int] = None, + end_after: Optional[int] = None, + ) -> list["SteamPlayRecord"]: + """ + 按条件查询游玩记录列表。所有参数可选，传入的条件以 AND 组合。 + steamid64 / appid 精确匹配；end_before / end_after 对 end_ts 做范围过滤（含边界）。 + end_ts 为 NULL 的记录不会出现在按时间范围过滤的结果中。 + """ + stmt = select(cls) + if steamid64 is not None: + stmt = stmt.where(cls.steamid64 == steamid64) + if appid is not None: + stmt = stmt.where(cls.appid == appid) + if end_before is not None: + stmt = stmt.where(cls.end_ts <= end_before) # type: ignore + if end_after is not None: + stmt = stmt.where(cls.end_ts >= end_after) # type: ignore + result = await session.execute(stmt) + return list(result.scalars().all()) + + @classmethod + @with_session + async def get_records_by_steamids( + cls: Type[T_SteamPlayRecord], + session: AsyncSession, + steamid64s: list[str], + ) -> list["SteamPlayRecord"]: + """按 steamid64 列表批量查询已结束的游玩记录（end_ts 为 NULL 的进行中记录不返回）""" + if not steamid64s: + return [] + stmt = select(cls).where( + cls.steamid64.in_(steamid64s), # type: ignore + cls.end_ts.is_not(None), # type: ignore + ) + result = await session.execute(stmt) + return list(result.scalars().all()) + + from . import admin # 注册到管理员 diff --git a/SteamUID/utils/utils.py b/SteamUID/utils/utils.py index 0406f2b..219b324 100644 --- a/SteamUID/utils/utils.py +++ b/SteamUID/utils/utils.py @@ -76,6 +76,27 @@ def HideStr(text: str) -> str: return text[0] + "*" * (len(text) - 3) + text[-2:] +def time_convert_s(seconds: int) -> str: + """将秒数转换为人类可读的时长，如 1天2小时30分45秒""" + if seconds < 0: + seconds = 0 + days = seconds // 86400 + hours = (seconds % 86400) // 3600 + minutes = (seconds % 3600) // 60 + secs = seconds % 60 + + parts = [] + if days > 0: + parts.append(f"{days}天") + if hours > 0: + parts.append(f"{hours}小时") + if minutes > 0: + parts.append(f"{minutes}分") + if secs > 0 or not parts: + parts.append(f"{secs}秒") + return "".join(parts) + + def maybe_hide_steamid(text: str) -> str: """根据 HideSteamID 配置决定是否对 steamid / 好友码套用 HideStr""" from ..SteamConfig import SteamConfig