Astrbot Steam好友在线监控插件

一、先说结果：这次改了什么

这次调整没有去追求“更大更全”，而是先把几个基础点做稳：

修正了配置项和文案里的乱码问题
让 online_only 配置真正参与推送逻辑
复用 httpx.AsyncClient，避免每次轮询都新建连接
支持按批次请求 Steam API，避免 ID 多时一次请求太长
把图片渲染放到线程里，避免阻塞事件循环
图片使用唯一文件名保存，避免手动查询和自动推送互相覆盖
对状态文件采用更稳妥的写入方式，减少运行中损坏文件的风险

如果一句话概括，就是把“能跑”整理成了“可以长期跑”。

二、这个插件真正麻烦的地方，不在请求 API

一开始看这类插件，很容易觉得重点在“怎么调 Steam 接口”。实际上，Steam 用户摘要接口并不复杂，真正麻烦的是后面的状态管理。

因为插件不是查一次数据就结束，而是要持续轮询。只要是持续轮询，就一定要回答两个问题：

上一次看到的状态是什么
这一次和上一次相比，哪些变化值得推送

如果没有这层“状态对比”，插件就只能反复发送“某某在线”“某某在线”，消息会很快失去意义。

所以这次实现里，核心不是单次拉取，而是“保存上一次状态，再做差异判断”。

状态文件大致是这样的：

{
  "76561198000000000": {
    "personaname": "example",
    "personastate": 1,
    "gameextrainfo": "Dota 2",
    "offline_since": "",
    "ts": "2026-03-24T10:00:00"
  },
  "_last_sync": "2026-03-24T10:00:00"
}

这里面真正有用的是三个字段：

personastate：当前在线状态
gameextrainfo：当前游戏名称
offline_since：从什么时候开始离线

前两个决定“要不要推送变化”，最后一个决定“轮询频率要不要放慢”。

三、轮询本身不难，难的是别把轮询写得太重

后台任务的入口是 _poll_loop()。这个函数做的事情很简单：

读取配置中的 SteamID
调用 Steam API 拉取状态
对比本地状态
生成事件
如果有事件，就推送
再决定下一次 sleep 多久

这类循环看起来普通，但有一个很常见的问题：一不小心就会把所有逻辑都塞进去。等到后面想改一条推送规则时，会发现自己要在一个很长的函数里来回找分支。

所以这次我把它拆成了几个职责比较清楚的方法：

_fetch_players() 负责拉数据
_build_transition_events() 负责判断变化
_update_state_and_collect_events() 负责更新本地状态
_render_status_image() 负责异步触发图片生成

这样做有个很现实的好处：以后如果你想新增“游戏切到指定分类才提醒”这类需求，基本只需要改事件判定这一层，不用碰轮询主循环。

四、状态变化的判断，尽量写得直白一点

我比较倾向于把这类逻辑写得接近自然语言，而不是过度抽象。因为这块代码以后很可能需要经常改。

当前的实现思路大概是这样：

if previous_state == 0 and current_state != 0:
    events.append(f"{name}: 上线（{persona_text(current_state)}）")
    if online_only:
        return events

if online_only:
    return events

if previous_state != 0 and current_state == 0:
    events.append(f"{name}: 下线")
elif previous_state != current_state and previous_state != 0 and current_state != 0:
    events.append(
        f"{name}: 状态变更（{persona_text(previous_state)} -> {persona_text(current_state)}）"
    )

if current_state != 0:
    if not previous_game and current_game:
        events.append(f"{name}: 启动游戏《{current_game}》")
    elif previous_game and not current_game:
        events.append(f"{name}: 关闭游戏《{previous_game}》")
    elif previous_game and current_game and previous_game != current_game:
        events.append(f"{name}: 切换游戏《{previous_game}》 -> 《{current_game}》")

这段逻辑看起来不复杂，但解决了两个实际问题。

第一个问题是 online_only 终于真的生效了。很多插件都有这个配置项，但最后只是写在配置里，并没有进入真正的分支判断。结果就是配置看起来有，行为却不变。

第二个问题是事件粒度统一了。上线、下线、状态切换、游戏切换都在同一个方法里处理，后续读代码时比较容易找到完整规则。

我的经验是，这种“业务判断很多，但每条都不长”的逻辑，最好少用花哨写法。直接写分支，后面维护起来反而更轻松。

五、图片渲染这件事，最好别直接堵在协程里

插件里有一个很直观的功能：把当前状态渲染成一张图片。

这部分在功能上很加分，因为纯文本推送虽然省事，但视觉上不够集中。头像、状态、正在运行的游戏，放在一张图里会更清楚。

但图片渲染也有一个很典型的坑：PIL 是同步的，下载头像和游戏封面也同样是同步的。如果这些逻辑直接跑在协程里，后台轮询和手动命令都会变得迟钝。

所以这里我保留了同步渲染函数，但把入口放到了线程里：

async def _render_status_image(self, players: list[dict[str, Any]]) -> str:
    async with self._render_lock:
        return await asyncio.to_thread(self._build_status_image_sync, players)

这个写法不复杂，但很实用。

它背后的思路其实是：不是所有逻辑都必须“全异步化”，有时候把同步工作明确地丢到线程里，已经足够解决问题，而且代码也更稳。

这里还顺手加了一个 Lock。因为状态图有可能同时被两个地方触发：

后台轮询检测到事件时自动推送
用户执行 /sfm_status 时手动查询

如果没有这把锁，再加上文件名固定，就很容易出现一边刚生成、一边又被覆盖的情况。

六、文件写入这种小事，往往最容易被忽略

插件运行久了以后，真正让人头疼的往往不是“接口超时”，而是一些看起来不起眼的小问题，比如状态文件突然坏掉，或者目录里积了一堆旧图片。

所以这次顺手补了两个细节。

1. 状态文件用临时文件再替换

def _save_state(self):
    temp_file = self.state_file.with_suffix(".tmp")
    temp_file.write_text(
        json.dumps(self.state, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    temp_file.replace(self.state_file)

这不是很复杂的技巧，但在这类长期运行的小插件里很有价值。至少比“直接覆盖原文件”更稳一些。

2. 状态图不再复用固定文件名

以前如果总是输出成 steam_status_latest.png，短时间内连续生成图片时，互相覆盖几乎是必然的。现在改成了带时间戳和随机后缀的唯一文件名：

output_path = self.render_dir / (
    f"steam_status_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid4().hex[:8]}.png"
)

同时每次生成后只保留最近的一部分缓存，避免目录越来越大。

这类改动不显眼，但通常比多加一个新功能更影响体验。

七、和旧配置兼容，比想象中更重要

这次整理时，还有一个容易被忽视的问题：插件旧版本可能已经把部分推送目标写进了状态文件，而不是标准配置项。

如果新版本直接忽略这些旧数据，升级后用户就会感觉“怎么什么都没推送了”。这类问题最麻烦，因为代码不报错，但用户体验是断的。

所以初始化时我加了一步轻量迁移：

def _migrate_legacy_state(self):
    legacy_targets = self.state.get("_push_targets")
    if not legacy_targets:
        return

    if isinstance(legacy_targets, list):
        migrated_targets = _unique_items(str(item) for item in legacy_targets)
    else:
        migrated_targets = parse_ids(str(legacy_targets))

    if migrated_targets and not parse_ids(self.config.get("push_targets", "")):
        self.config["push_targets"] = ",".join(migrated_targets)
        self._save_config_safe()

    self.state.pop("_push_targets", None)
    self._save_state()

这段代码的技术含量不高，但它体现的是另一个开发经验：很多“维护成本”并不来自新功能，而是来自对旧数据的处理。

八、轮询频率不是固定值，体验会更好

如果把轮询时间写死成 60 秒，代码当然没问题，但使用体验未必最好。

举个很实际的场景：

大家都在线时，状态变化更频繁，轮询间隔应该短一点
大家已经离线了很久，再保持高频轮询意义不大

所以这里做了一个简单的动态策略：

if offline_minutes_max >= 30:
    return max(default_sec, 600)
if offline_minutes_max >= 10:
    return max(default_sec, 300)
return default_sec

这个算法并不复杂，但已经能起到不错的效果。它没有把逻辑做得很“聪明”，只是让插件在“无人变化”的阶段少做一些无意义的请求。

我的经验是，小插件不一定要追求复杂调度，先把简单策略落好，通常已经很够用了。

九、文档这件事，最好和代码一起改

这次顺手一起更新了：

README.md
_conf_schema.json
metadata.yaml

原因很简单。像这种 AstrBot 插件，用户最早接触到的不是代码，而是安装页、配置页和 README。

如果代码改了，但配置说明还是旧的，或者某个选项已经有效了，文档里还写得含糊不清，最后还是会回到“明明加了功能，但别人不知道怎么用”的状态。

这次我比较在意的一点，就是让 online_only 的描述和实际行为保持一致。这个细节不大，但它能明显减少误解。

十、这次开发里几个比较实用的经验

这里不写成“最佳实践”，只记录几个确实有帮助的点。

1. 先拆职责，再谈优化

如果轮询、事件判断、推送、图片渲染全挤在一个函数里，后面很难继续改。先把结构理顺，很多优化自然就有落点了。

2. 异步不是目的，不卡主流程才是目的

这类插件里，最怕的是把所有东西都写成协程，然后在协程里干了很多同步工作。与其表面上“全异步”，不如老老实实把阻塞逻辑放线程里。

3. 配置项一定要落到真实行为上

尤其是布尔配置，最容易出现“文档有、界面有、代码里没用上”的情况。只要是暴露给用户的配置，最好都能在核心流程里找到明确入口。

4. 小型插件更需要控制边界

插件越小，越不适合引入很重的抽象。像这次实现里，大部分方法都保持在一个比较短的长度，就是为了后续查问题时能快速定位。

十一、如果后面还要继续扩展，我会怎么做

当前版本已经适合继续往下迭代。如果要做下一步增强，我会优先考虑这些方向：

给头像和游戏封面增加带过期时间的缓存，而不只是进程内缓存
给推送目标增加权限或范围控制，比如限制只有管理员能绑定
给状态图增加可选主题，让样式更容易调整
给频繁切换游戏的情况做简单节流，避免短时间推送过多
给请求异常做更细一点的分类，比如网络错误、接口错误、配置错误分别提示

这些都不是必须马上做的功能，但都属于“真正用起来之后会自然想到”的优化方向。

十二、最后

这次整理之后，这个插件的状态更像一个可以长期维护的小项目，而不是一个只验证功能的脚本。

它没有用很重的架构，也没有刻意做复杂抽象，但在轮询、状态差异、图片渲染和兼容性这几个关键点上，已经有了比较清楚的边界。

如果你准备继续在这个插件上加功能，我比较建议先从事件规则或者图片样式下手。这两个方向改动直观，回报也比较明显，而且不容易破坏主流程。

如果只是想稳定使用当前版本，那么至少可以放心一点：它现在处理的，已经不只是“能查到 Steam 状态”，而是“能比较稳地把变化推出来”。

SZC's Blog