[Plugin] astrbot_plugin_wfbot

[Youzini-afk]

插件信息

{
  "name": "astrbot_plugin_wfbot",
  "display_name": "WF游戏相关数据查询订阅",
  "desc": "Warframe 数据层（世界状态/PublicExport/多镜像/market），支持缓存、清理与定时刷新",
  "author": "Youzini",
  "repo": "https://github.com/Youzini-afk/astrbot_plugin_wfbot",
  "tags": [
    "游戏",
    "warframe"
  ],
  "social_link": "https://github.com/Youzini-afk"
}

插件检查清单

• 我的插件经过完整的测试
• 我的插件不包含恶意代码
• 我已阅读并同意遵守该项目的 行为准则。

[astrpluginreviewer]

🤖 AI代码审核报告 for astrbot_plugin_wfbot

您好！我已经对你提交的插件代码进行了初步自动化审核，作为初步参考:

代码审查报告

总体评价

这是一个功能丰富的 Warframe 数据查询与订阅插件，代码结构清晰，异步处理得当。整体质量较高，但存在一些需要改进的细节问题。

---

main.py

问题清单

1. 【严重】缺少插件注册装饰器或明确的类标识

位置: 类定义 class WarframeDatasourcePlugin(Star)
问题: 代码中使用了 @register 装饰器，但根据审查要求，在 v3.5.20 版本之后该装饰器已废弃。虽然代码可以工作，但建议遵循新版本规范，移除装饰器，让 AstrBot 自动识别继承自 Star 的类。
建议:

# 移除 @register 装饰器
class WarframeDatasourcePlugin(Star):
    def __init__(self, context: Context, config: AstrBotConfig):
        super().__init__(context)

2. 【中等】复杂方法职责过多

位置: __init__ 方法（约 150 行）
问题: 构造函数过于庞大，包含配置解析、缓存初始化、权限映射、订阅存储等多项职责，违反单一职责原则。
建议: 将配置解析、权限初始化等逻辑提取为独立的私有方法：

def __init__(self, context: Context, config: AstrBotConfig):
    super().__init__(context)
    self.config = config
    
    # 提取配置解析
    self._init_config()
    self._init_data_sources()
    self._init_permissions()
    self._init_subscriptions()
    self._maybe_start_background()

def _init_config(self):
    # 配置解析逻辑
    ...

def _init_permissions(self):
    # 权限映射逻辑
    ...

3. 【中等】潜在的数据竞态条件

位置: _on_worldstate_updated 方法
问题: 方法中存在多次异步操作和锁操作，但在构建 topic_payload 和 topic_ids 时，如果 ws2 在并发情况下被修改，可能导致数据不一致。
建议: 在关键数据处理前进行深拷贝：

import copy
ws2_snapshot = copy.deepcopy(ws2)
# 使用 ws2_snapshot 进行后续处理

4. 【轻微】硬编码的魔法数值

位置: 多处（如 limit: int = 10, limit: int = 8）
问题: 代码中多次出现硬编码的限制数量，不利于维护。
建议: 定义为类常量或配置项：

class WarframeDatasourcePlugin(Star):
    DEFAULT_FISSURE_LIMIT = 10
    DEFAULT_ALERT_LIMIT = 8
    
    def _format_fissures_filtered(self, ..., limit: int | None = None):
        limit = limit or self.DEFAULT_FISSURE_LIMIT

5. 【轻微】异常处理过于宽泛

位置: 多处 except Exception: 块
问题: 使用通用的 Exception 捕获可能掩盖真实错误，例如：

try:
    val = await getter("wf_push_enabled", True)
    self._push_enabled = bool(val)
except Exception:
    logger.debug("load kv wf_push_enabled failed", exc_info=True)

建议: 捕获更具体的异常类型，或至少记录异常详情以便调试。

6. 【轻微】方法名称不够清晰

位置: _normalize_sub_topic_args
问题: 方法名未明确表达其核心功能（解析并规范化订阅主题）。
建议: 重命名为 _parse_and_normalize_subscription_topic。

7. 【建议】缺少类型提示

位置: 部分方法参数和返回值
问题: 虽然代码使用了类型提示，但部分复杂方法的返回值类型不够明确，例如 _topic_text 返回 tuple[str, str] 可读性不强。
建议: 使用 TypedDict 或 NamedTuple 增强可读性：

from typing import NamedTuple

class TopicTextResult(NamedTuple):
    title: str
    message: str

async def _topic_text(self, topic: str, ws: dict) -> TopicTextResult:
    ...
    return TopicTextResult(title=title, message=msg)

8. 【建议】数据持久化路径处理可以优化

位置: _resolve_plugin_data_dir 方法
问题: 虽然已正确使用 StarTools.get_data_dir()，但异常处理后的 fallback 路径 Path(__file__).resolve().parent / ".plugin_data" 可能导致数据分散。
建议: 在异常情况下也应尽量使用标准路径，或至少记录警告日志。

---

wf_cache.py

问题清单

1. 【建议】缺少异常处理和日志记录

位置: write_bytes 和 write_json 方法
问题: 写入操作可能因磁盘空间不足或权限问题失败，但未捕获异常。
建议: 添加异常处理和日志记录：

async def write_bytes(self, data: bytes, *parts: str) -> Path:
    p = self.path(*parts)
    try:
        await asyncio.to_thread(p.parent.mkdir, parents=True, exist_ok=True)
        async with aiofiles.open(p, "wb") as f:
            await f.write(data)
        return p
    except OSError as e:
        logger.error(f"Failed to write bytes to {p}: {e}")
        raise

---

wf_config.py

无明显问题

该文件使用 dataclass 定义配置，结构清晰，符合规范。

---

wf_datasource.py

问题清单

1. 【轻微】配置修改方式不够优雅

位置: create 方法中的配置更新
问题: 使用 {**cfg.__dict__, ...} 修改 frozen dataclass 不够直观。
建议: 使用 dataclasses.replace：

from dataclasses import replace

cfg = replace(cfg, data_dir=Path(StarTools.get_data_dir(plugin_id)) / "warframe")

---

wf_events.py

无明显问题

事件总线实现简洁高效，符合异步最佳实践。

---

wf_format.py

问题清单

1. 【轻微】硬编码的字典映射

位置: MISSION_TYPE_ZH, FACTION_ZH 等
问题: 大量硬编码映射散布在模块中，维护成本高。
建议: 考虑从配置文件或 JSON 加载这些映射，或至少集中管理。

2. 【建议】函数职责过于单一导致重复

位置: _translate_mission_like, _translate_boss, _translate_faction
问题: 这些函数逻辑高度相似，可以抽象为通用翻译函数。
建议:

def _translate_from_mapping(value: Any, mapping: dict[str, str], fallback_handler: Callable | None = None) -> str:
    if not isinstance(value, str) or not value:
        return "-"
    if value in mapping:
        return mapping[value]
    if fallback_handler:
        return fallback_handler(value)
    return value

---

wf_http.py

问题清单

1. 【中等】Session 管理可能导致资源泄漏

位置: _session_for_url 方法
问题: 在某些异常情况下，Session 可能未正确关闭。虽然有 aclose 方法，但如果插件异常退出，可能导致泄漏。
建议: 使用上下文管理器或确保在插件卸载时调用 aclose。

2. 【轻微】重试逻辑缺少日志

位置: get 方法的重试循环
问题: 重试时未记录日志，不利于排查网络问题。
建议:

except (aiohttp.ClientError, asyncio.TimeoutError) as e:
    logger.debug(f"HTTP request failed (attempt {attempt}/{1 + self._retries}): {e}")
    if self._max_request_time_seconds is not None and ...

---

wf_i18n.py

无明显问题

简洁的繁简转换实现，使用 lru_cache 优化性能。

---

wf_manager.py

问题清单

1. 【中等】日志导入符合规范

位置: from astrbot.api import logger
说明: ✅ 正确使用框架提供的 logger。

2. 【建议】清理逻辑可以异步化

位置: cleanup 方法
问题: 虽然提供了 cleanup_async，但实际清理操作仍是同步的，可能阻塞事件循环。
建议: 将文件删除等 I/O 操作包装为异步任务池。

---

wf_render.py

问题清单

1. 【轻微】缺少 PIL 依赖检查

位置: render_text_to_png_bytes 函数
问题: 直接导入 PIL，如果未安装会抛出 ImportError。
建议: 在插件初始化时检查依赖，或提供友好的错误提示。

2. 【建议】字体查找逻辑复杂

位置: _pick_fonts 函数
问题: 字体查找逻辑冗长，可以提取为独立函数。
建议: 封装为 _find_system_font(candidates: list[str], size: int)。

---

wf_source_*.py 系列文件

通用问题

1. 【建议】缺少 API 限流保护

位置: wf_source_market.py, wf_source_worldstate.py
问题: 对外部 API 的请求未实现限流，可能触发 API 速率限制。
建议: 使用 asyncio.Semaphore 或 aiolimiter 库控制并发请求数。

2. 【轻微】日志记录不足

位置: 多个文件中的异常处理
问题: 部分网络请求失败未记录详细日志。
建议: 增加 logger.warning 或 logger.error 记录。

---

wf_subscriptions.py

问题清单

1. 【中等】文件写入缺少原子性保证

位置: save 方法
问题: 直接覆盖写入可能在写入过程中崩溃导致数据损坏。
建议: 使用临时文件+重命名的原子写入方式：

async def save(self, data: dict) -> None:
    temp_path = self._path.with_suffix('.tmp')
    async with aiofiles.open(temp_path, "w", encoding="utf-8") as f:
        await f.write(json.dumps(data, ensure_ascii=False, indent=2))
    await asyncio.to_thread(temp_path.replace, self._path)

---

代码评审团

Linus Torvalds（直率、技术导向）

"你的代码写得还行，至少没用那些垃圾的 ORM 框架。但我看到你在 main.py 的 __init__ 里塞了 150 行代码？这是什么鬼！你是打算把整个插件都塞进构造函数里吗？拆分它，让每个函数只做一件事，这不是火箭科学！

还有，你的异常处理太宽泛了。except Exception 就像是在说'我不在乎发生了什么，反正继续跑'。这在生产环境中是找死的做法。至少捕获具体的异常类型，或者记录完整的堆栈跟踪，别让未来的维护者（可能就是你自己）骂街。

不过，你用了 asyncio 处理 I/O，这点做得不错。但记得，Session 管理要严格，别让资源泄漏成为常态。"

---

Grace Hopper（先驱、务实主义者）

"年轻人，你的代码结构很好，异步处理也很到位。但我注意到你在订阅存储时直接覆盖写文件，这在我那个年代就知道是危险的——磁带机可不会等你写完再断电！

建议使用临时文件+原子重命名，这是经过时间检验的最佳实践。另外，你的配置解析逻辑分散在多个地方，如果将来需要支持新的配置项，维护成本会很高。考虑使用配置类或验证器（如 Pydantic）来集中管理。

最后，我看到你在处理用户输入时做了规范化，这很好。但别忘了，安全性永远是第一位的。虽然你已经做了路径过滤（如 _sanitize_export_filename），但要确保所有外部输入都经过严格验证。"

---

Tony Stark（创新、未来导向）

"伙计，你这个插件功能齐全，但我觉得可以更酷一点。你有没有想过引入机器学习来预测用户最关心的任务？比如分析历史订阅数据，自动推荐'你可能感兴趣的裂隙'。

另外，你的图片渲染功能很棒，但如果能支持主题切换（暗色/亮色模式），甚至让用户自定义背景图，那就更炫了。我的 JARVIS 可以做到这些，你也可以。

还有，考虑一下性能监控。你可以在关键方法中嵌入计时器，收集响应时间数据，甚至搭建一个简易的仪表板。这样不仅能帮助你优化代码，还能在出问题时快速定位瓶颈。

最后，你的代码注释还不错，但我建议增加一些设计决策的注释。比如为什么选择这种缓存策略，为什么用这个 TTL 值。未来的你（或者接手的人）会感谢现在的你。"

---

Disclaimer: 以上评审内容由 AI 自动生成,所涉及人物形象与现实无关，不代表真实人物观点。如果给出的建议无关痛痒请忽略。

---

🔍 审核摘要

统计信息

• 仓库文件总数: 19 个 Python 文件
• 已选择审核文件: 15 / 19
  已发送至 AI 审核的文件清单

main.py
__init__.py
wf_cache.py
wf_config.py
wf_datasource.py
wf_events.py
wf_format.py
wf_http.py
wf_i18n.py
wf_manager.py
wf_render.py
wf_source_market.py
wf_source_mirrors.py
wf_source_public_export.py
wf_source_worldstate.py

注意：由于项目规模或Token限制，本次仅审核了部分优先文件。以上报告内容由 AI 直接生成。

---

此报告由AI自动生成，旨在提供初步反馈和改进建议，不能完全替代人工审核。最终决策以社区维护者的人工审核为准。目前自动审核（仓库地址）处于试验阶段，如遇问题请向维护者反馈。