diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..43ae0e2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.py[cod] diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..2ec5af6 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,74 @@ +# 更新日志 (CHANGELOG) + +## [v2.3.1]（当前版本） + +- 移除项目内测试文件与测试相关开发依赖/说明 + +## [v2.3.0] + +- 新段位：对齐 OverFast API 4.13 `CompetitiveDivision`，新增翡翠 `emerald`（白金与钻石之间）、王者 `ultimate`（最高段位）；`champion` 改显示为王者并保留兼容旧数据，`top500` 保留 +- 自建服务：新增 `api_base_url` 配置项（默认官方 `https://overfast-api.tekrop.fr`），插件所有请求经该地址构造，支持自建 OverFast 实例；非法地址自动回退官方 + +## [v2.2.0] + +- 重构：`main.py` 拆分为 `constants.py`/`utils.py`/`bindings.py`，旧导入路径保留兼容 +- 性能：API 客户端默认超时 30s→15s，进程级 GET 缓存（玩家 120s/榜单 600s/英雄 3600s），429/503 按 `retry_after` 自动重试 +- 并发：图片/头像下载复用单 session + 6 并发上限，不再每图新建 session +- 泄漏：`image_renderer` 新增 `cleanup_old_images()`（每小时，`_save_image` 自动触发），`owherostats` 图片模式避免 54 行超长文本 +- 健壮：玩家 ID 归一化+长度/字符校验+URL quote；图片开关默认与 schema 一致为开启；移除 `COMPETITIVE_ICONS` 重复；日志统一 AstrBot logger +- 杂项：修正 README 仓库地址，删除重复 `logo.png.png` + +## [v2.1.0] + +- 英雄映射新增第 53 位英雄 D.Mon（重装，第 4 赛季）与第 54 位英雄血律 Doctrine（支援，第 5 赛季），现可通过 /owhero D.Mon、/owhero 血律 等查询 +- 英雄中文映射补全至 54 名英雄（与 OverFast API HeroKey 枚举对齐） + +## [v2.0.0] + +- 删除 /owsearch 与 /owheroes 指令（功能被 /owsummary 与 /owhero 覆盖） +- /owherostats 展示全部英雄胜率（不再截断前 15）；/owhero 同步查询并展示该英雄的全服胜率与选取率 +- 多账号绑定：每个 QQ 号可绑定多个 OW 账号（配置 max_binds_per_user，默认 3），KV 结构升级并自动迁移旧数据 +- 新增 /owbinds 查看绑定列表、/owdefault 设置默认查询账号；/owunbind 支持指定账号解绑 +- 新增查询结果图片渲染功能（owsummary/owstats/owcareer/owme），渲染为 OW 风格卡片 +- 绑定账号时支持选择平台（PC端/主机端）：/owbind <玩家ID> [pc|主机] +- 英雄中文映射补全至 52 名英雄（与 API HeroKey 枚举对齐），新增骇灾、弗蕾娅、无漾、斩仇、安燃、金驭、埃姆雷、瑞稀、飞天猫、西拉 +- API 报错结构化解析（含状态码、错误详情、retry_after），完整报错记录到控制台 debug 日志 +- 插件会自动将 # 替换为 - 再查询 + +## [v1.3.0] + +- 重构插件 +- 重构命令 +- 缓存更改为KV缓存 +- 添加简单配置界面 +- 计划下一版本进行战绩图片渲染 + +## [v1.2.1] + +- 缓存降级增强 +- 更改描述页 +- 增加英雄查询（独立查询） +- 可以查询单个英雄战绩，可以查询休闲和竞技两个模式 +- 绑定提示优化 +- 添加主机平台战绩查询 +- resp 异常修复 +- 错误提示细化 + +## [v1.1.1] + +- 找不到方法解决常玩英雄的两百报错问题，直接删去常玩英雄项目查询 +- 职责显示更改为中文 +- 竞技段位未定级也显示 + +## [v1.1.0] + +- 修复API域名失效问题 +- 增强错误处理 +- 添加详细统计信息 +- 将API请求分为五段整合后再发出减少超时可能 +- 增加申请管控防止过量请求 + +## [v1.0.0] + +- 修复API域名失效问题 +- 添加自动重试机制 diff --git a/LICENSE b/LICENSE index c90b403..656ca01 100644 --- a/LICENSE +++ b/LICENSE @@ -1,5 +1,26 @@ MIT License +<<<<<<< HEAD +Copyright (c) 2026 TZYCeng + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +======= Copyright (c) 2026 y0lo Permission is hereby granted, free of charge, to any person obtaining a copy of this software and @@ -16,3 +37,4 @@ LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE A EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +>>>>>>> 558bcd8abb48f869ac425cbb85e248b450ddc36e diff --git a/README.md b/README.md index 5b12237..4fff2e4 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,74 @@ +<<<<<<< HEAD +# AstrBOT 守望先锋国际服查询插件 +作者其实是个小白来的，部分由ai创建 +# 注意,部分使用steam启动的玩家无法被查询 + +🎮 **守望先锋·归来国际服查询插件** + +[![Version](https://img.shields.io/badge/version-v2.3.1-blue.svg)](https://github.com/TZYCeng/astrbot_plugin_owcx) +[![Python](https://img.shields.io/badge/python-3.8+-green.svg)](https://www.python.org/) +[![AstrBot](https://img.shields.io/badge/AstrBot-4.9+-orange.svg)](https://github.com/AstrBotDevs/AstrBot) +[![License](https://img.shields.io/badge/license-MIT-yellow.svg)](https://opensource.org/licenses/MIT) + + +## 📦 安装方法 +### 方法一：通过AstrBot插件市场 +1. 打开AstrBot WebUI +2. 进入插件管理 +3. 搜索 "astrbot_plugin_owcx" +4. 点击安装 +### 方法二：手动安装 +1. 下载插件文件到AstrBot的插件目录 +```bash +cd AstrBot/data/plugins +git clone https://github.com/TZYCeng/astrbot_plugin_owcx.git +``` +2. 安装依赖 +```bash +pip install -r requirements.txt +``` +3. 重启AstrBot + +## 使用方法 +### owsummary +查询玩家摘要信息（头像、竞技段位等）。 用法: /owsummary [玩家ID] 示例: /owsummary TeKrop#2217 或 /owsummary 同志有程#3156 说明: 玩家ID支持直接输入 #，插件会自动处理；省略 ID 则查询绑定的账号（使用绑定平台） +### owstats +查询玩家统计概览（胜率、KDA等）。 用法: /owstats [玩家ID] [游戏模式] 游戏模式: 快速、竞技（默认） 示例: /owstats TeKrop#2217 竞技 说明: 省略 ID 则查询绑定的账号 +### owcareer +查询玩家生涯统计（按英雄详细数据）。 用法: /owcareer [玩家ID] <游戏模式> [英雄名] 游戏模式: 快速、竞技 英雄名: 可选，支持中文（如 源氏、安娜）或英文（如 genji, ana） 示例: /owcareer TeKrop#2217 竞技 /owcareer TeKrop#2217 竞技 源氏 /owcareer 竞技 源氏 (已绑定ID后) +### owhero +查询英雄详细信息（含该英雄在当前配置地区/平台/模式下的全服胜率与选取率）。 用法: /owhero <英雄名> 英雄名支持中文（如 源氏、安娜）或英文（如 genji, ana） 示例: /owhero 源氏, /owhero ana +### owherostats +查询全服英雄胜率/选取率排行榜（按地区服务器统计，展示全部英雄）。 用法: /owherostats [角色] [地区] 角色: 坦克、输出、支援（可选） 地区: 亚服(asia)、美服(americas)、欧服(europe)，不填则使用配置中的默认地区 平台与游戏模式使用配置文件中的默认值 示例: /owherostats /owherostats 输出 /owherostats 支援 欧服 +### owbind +绑定 Overwatch ID 到当前 QQ 号，可选择平台（PC端/主机端），每个 QQ 号可绑定多个账号（数量上限见配置 max_binds_per_user，默认 3 个），新绑定的账号自动成为默认查询账号。 用法: /owbind <玩家ID> [平台] 平台: pc(电脑端)、console(主机端)，不填则使用配置中的默认平台 示例: /owbind TeKrop#2217 或 /owbind TeKrop#2217 主机 说明: 旧版本绑定数据自动迁移，无需手动处理 +### owunbind +解绑当前 QQ 号绑定的 Overwatch ID。 用法: /owunbind [玩家ID] 说明: 不填玩家ID时解绑当前默认查询账号；解绑默认账号后自动切换到剩余账号 +### owbinds +查看当前 QQ 号绑定的所有 Overwatch 账号（含平台与默认标记）。 用法: /owbinds +### owdefault +设置默认查询的 Overwatch 账号，快捷指令省略 ID 时将查询该账号。 用法: /owdefault <玩家ID> 示例: /owdefault TeKrop#2217 +### owme +快捷查询默认绑定账号的摘要信息。 用法: /owme 说明: 需要先使用 /owbind 绑定，多账号时查询 /owdefault 设置的账号 + + +## ⚙️ 配置说明 +在 AstrBot WebUI 的插件配置中可设置以下选项： +- **默认查询的游戏模式**: 竞技比赛 / 快速游戏 +- **默认查询的平台**: pc / console（未绑定平台或绑定时未选择平台时使用） +- **默认查询的地区服务器** (`default_region`): asia(亚服) / americas(美服) / europe(欧服)，用于 /owherostats 等按地区统计的查询 +- **每QQ号绑定上限** (`max_binds_per_user`): 每个 QQ 号最多可绑定的 Overwatch 账号数量，默认 3 +- **图片渲染开关** (`enable_image_render`): 开启后（默认开启），/owsummary、/owstats、/owcareer、/owme、/owherostats 的查询结果会渲染为卡片图片；渲染失败自动回退文字；临时图片每小时自动清理，Docker 内如无中文字体请关闭或挂载 `fonts/font.ttf` +- **报错展示开关** (`show_api_error`): 开启后，查询出错时机器人会把 API 返回的具体报错（含建议重试等待时间）回复给查询者；无论开关与否，完整报错与堆栈都会记录在控制台 debug 日志中 +- **API 服务地址** (`api_base_url`): OverFast API 地址，默认官方 `https://overfast-api.tekrop.fr`；自建服务（如 Docker 部署开源 OverFast）填入自建地址即可，非法值自动回退官方 +- 图片渲染依赖 Pillow：`pip install Pillow>=9.0.0`（已在 requirements.txt 中） +- 如系统缺少中文字体导致渲染文字异常，可在插件目录下新建 `fonts` 文件夹并放入任意中文字体文件（命名为 `font.ttf` 或 `font.ttc`） + +## 📕 更新日志 + +版本更新记录已迁移至 [CHANGELOG.md](CHANGELOG.md)。 +======= # overwatch_bot -an agent to fetch and analysis overwatch datastics \ No newline at end of file +an agent to fetch and analysis overwatch datastics +>>>>>>> 558bcd8abb48f869ac425cbb85e248b450ddc36e diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..da7991f --- /dev/null +++ b/__init__.py @@ -0,0 +1,5 @@ +"""OW战绩查询插件 - AstrBot Plugin for Overwatch Stats Query.""" + +from .main import OverwatchStatsPlugin + +__all__ = ["OverwatchStatsPlugin"] diff --git a/_conf_schema.json b/_conf_schema.json new file mode 100644 index 0000000..57596a2 --- /dev/null +++ b/_conf_schema.json @@ -0,0 +1,40 @@ +{ + "default_gamemode": { + "description": "默认查询的游戏模式", + "type": "string", + "options": ["竞技比赛", "快速游戏"], + "default": "竞技比赛" + }, + "default_platform": { + "description": "默认查询的平台（未绑定或绑定时未选择平台时使用）", + "type": "string", + "options": ["pc", "console"], + "default": "pc" + }, + "default_region": { + "description": "默认查询的地区（用于英雄胜率/选取率排行榜等查询）", + "type": "string", + "options": ["asia", "americas", "europe"], + "default": "asia" + }, + "max_binds_per_user": { + "description": "每个 QQ 号最多可绑定的 Overwatch 账号数量", + "type": "int", + "default": 3 + }, + "enable_image_render": { + "description": "是否启用查询结果图片渲染（渲染失败时自动回退为文字）", + "type": "bool", + "default": true + }, + "show_api_error": { + "description": "是否在查询出错时把 API 返回的具体报错回复给查询者", + "type": "bool", + "default": false + }, + "api_base_url": { + "description": "OverFast API 服务地址（官方或自建，如 https://overfast-api.tekrop.fr）", + "type": "string", + "default": "https://overfast-api.tekrop.fr" + } +} diff --git a/api_client.py b/api_client.py new file mode 100644 index 0000000..de36dc8 --- /dev/null +++ b/api_client.py @@ -0,0 +1,357 @@ +""" +提供对 OverFast API (https://overfast-api.tekrop.fr) 的所有异步 HTTP 调用封装， +用于查询 Overwatch 玩家战绩、英雄信息等数据。 +""" + +import asyncio +import time +from typing import Any +from urllib.parse import quote + +import aiohttp + +try: # 统一使用 AstrBot 日志；独立单测时回退标准 logging + from astrbot.api import logger +except Exception: # pragma: no cover + import logging + + logger = logging.getLogger(__name__) + + +class OverFastAPIError(ValueError): + """OverFast API 请求错误。 + + 携带 HTTP 状态码、服务端返回的错误详情以及重试等待时间， + 继承 ValueError 以兼容既有的错误捕获逻辑。 + + Attributes: + status_code: HTTP 状态码。 + detail: 服务端返回的原始错误文本。 + retry_after: 服务端建议的重试等待秒数（可能为 None）。 + """ + + def __init__( + self, + message: str, + status_code: int | None = None, + detail: str = "", + retry_after: int | None = None, + ) -> None: + super().__init__(message) + self.status_code = status_code + self.detail = detail + self.retry_after = retry_after + + +# 进程级 GET 缓存：key -> (expire_ts, data)，降低限流风险 +_CACHE: dict[str, tuple[float, Any]] = {} + + +def _cache_get(key: str) -> Any | None: + item = _CACHE.get(key) + if not item: + return None + expire, data = item + if expire < time.monotonic(): + _CACHE.pop(key, None) + return None + return data + + +def _cache_set(key: str, data: Any, ttl: float) -> None: + if ttl <= 0: + return + # 简单容量保护 + if len(_CACHE) > 512: + _CACHE.clear() + _CACHE[key] = (time.monotonic() + ttl, data) + + +def clear_cache() -> None: + _CACHE.clear() + + +class OverFastAPIClient: + """OverFast API 异步客户端。 + + 使用示例: + async with OverFastAPIClient() as client: + result = await client.search_players("TeKrop") + summary = await client.get_player_summary("TeKrop-2217") + """ + + def __init__( + self, + base_url: str = "https://overfast-api.tekrop.fr", + timeout: float = 15.0, + ) -> None: + """初始化 OverFast API 客户端。 + + Args: + base_url: API 基础 URL，官方实例或自建实例（如 Docker 自部署）。 + 插件侧通过 `api_base_url` 配置项传入。 + timeout: 请求超时时间（秒），默认 15 秒（聊天场景不宜过长）。 + """ + self.base_url = base_url.rstrip("/") + self.timeout = aiohttp.ClientTimeout(total=timeout) + self._session: aiohttp.ClientSession | None = None + + @property + def session(self) -> aiohttp.ClientSession: + """获取当前 HTTP session。 + + Returns: + aiohttp.ClientSession 实例。 + + Raises: + RuntimeError: 如果 session 未初始化（未使用 async with）。 + """ + if self._session is None or self._session.closed: + raise RuntimeError( + "HTTP session 未初始化。请使用 `async with OverFastAPIClient() as client:`" + ) + return self._session + + async def __aenter__(self) -> "OverFastAPIClient": + """异步上下文管理器入口，创建 HTTP session。""" + self._session = aiohttp.ClientSession(timeout=self.timeout) + return self + + async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + """异步上下文管理器出口，确保 session 被关闭。""" + if self._session and not self._session.closed: + await self._session.close() + self._session = None + + @staticmethod + def _quote_segment(seg: str) -> str: + return quote(seg, safe="-") + + async def _request( + self, + endpoint: str, + params: dict | None = None, + *, + use_cache: bool = False, + cache_ttl: float = 300.0, + max_retries: int = 2, + ) -> dict | list: + """发送 GET 请求到 OverFast API（含缓存/限流重试）。 + + Args: + endpoint: API 端点路径（不含 base_url），如 `/players`。 + params: 查询参数字典，值为 None 的键会被自动过滤。 + use_cache: 是否启用进程级 GET 缓存。 + cache_ttl: 缓存秒数。 + max_retries: 429/503 时的最大重试次数。 + + Raises: + OverFastAPIError: 当返回 HTTP 4xx/5xx 错误时。 + """ + filtered_params = {k: v for k, v in (params or {}).items() if v is not None} + url = f"{self.base_url}/{endpoint.lstrip('/')}" + logger.debug(f"OverFast API 请求: GET {url} params={filtered_params}") + + cache_key = "" + if use_cache: + sorted_items = sorted((k, str(v)) for k, v in filtered_params.items()) + cache_key = f"GET {url} {sorted_items}" + cached = _cache_get(cache_key) + if cached is not None: + logger.debug(f"OverFast API 缓存命中: {url}") + return cached + + last_err: OverFastAPIError | None = None + for attempt in range(max_retries + 1): + async with self.session.get(url, params=filtered_params) as response: + if response.status >= 400: + error_text = "" + retry_after: int | None = None + try: + error_json = await response.json() + if isinstance(error_json, dict): + error_text = str(error_json.get("error", "")) + ra = error_json.get("retry_after") + if isinstance(ra, (int, float)): + retry_after = int(ra) + if not error_text and "detail" in error_json: + error_text = str(error_json["detail"]) + if not error_text: + error_text = str(error_json) + else: + error_text = str(error_json) + except Exception: + error_text = await response.text() or f"HTTP {response.status}" + + logger.debug( + f"OverFast API 错误: GET {url} params={filtered_params} " + f"-> HTTP {response.status}, error={error_text}, retry_after={retry_after}" + ) + + if response.status == 404: + raise OverFastAPIError( + f"未找到请求的资源: {error_text}", + status_code=404, detail=error_text, retry_after=retry_after, + ) + if response.status in (429, 503) and attempt < max_retries: + wait = retry_after if retry_after else (2 ** attempt) + wait = min(max(wait, 1), 10) + logger.debug(f"OverFast API 限流，{wait}s 后重试 ({attempt + 1}/{max_retries})") + await asyncio.sleep(wait) + last_err = OverFastAPIError( + f"请求过于频繁: {error_text}", + status_code=response.status, detail=error_text, + retry_after=retry_after, + ) + continue + if response.status == 429: + raise OverFastAPIError( + f"请求过于频繁，请稍后重试: {error_text}", + status_code=429, detail=error_text, retry_after=retry_after, + ) + if response.status == 503: + raise OverFastAPIError( + f"服务暂时不可用（可能被限流）: {error_text}", + status_code=503, detail=error_text, retry_after=retry_after, + ) + raise OverFastAPIError( + f"API 错误 (HTTP {response.status}): {error_text}", + status_code=response.status, detail=error_text, retry_after=retry_after, + ) + + try: + data = await response.json() + except aiohttp.ContentTypeError as e: + raw_text = await response.text() + raise ValueError( + f"无法解析 API 响应为 JSON: {e}. 原始响应: {raw_text[:500]}" + ) from e + + logger.debug(f"OverFast API 响应: {type(data).__name__}") + if use_cache: + _cache_set(cache_key, data, cache_ttl) + return data + + assert last_err is not None + raise last_err + + async def search_players( + self, + name: str, + order_by: str = "name:asc", + offset: int = 0, + limit: int = 20, + ) -> dict: + """搜索玩家。""" + return await self._request( # type: ignore[return-value] + "/players", + params={"name": name, "order_by": order_by, "offset": offset, "limit": limit}, + use_cache=True, + cache_ttl=120, + ) + + async def get_player_summary(self, player_id: str) -> dict: + """获取玩家摘要信息。""" + pid = self._quote_segment(player_id) + return await self._request(f"/players/{pid}/summary", use_cache=True, cache_ttl=120) # type: ignore[return-value] + + async def get_player_full(self, player_id: str) -> dict: + """获取玩家完整数据（摘要 + 统计数据）。""" + pid = self._quote_segment(player_id) + return await self._request(f"/players/{pid}", use_cache=True, cache_ttl=120) # type: ignore[return-value] + + async def get_player_stats_summary( + self, + player_id: str, + gamemode: str | None = None, + platform: str | None = None, + ) -> dict: + """获取玩家统计摘要。""" + pid = self._quote_segment(player_id) + return await self._request( # type: ignore[return-value] + f"/players/{pid}/stats/summary", + params={"gamemode": gamemode, "platform": platform}, + use_cache=True, + cache_ttl=120, + ) + + async def get_player_career_stats( + self, + player_id: str, + gamemode: str, + platform: str | None = None, + hero: str | None = None, + ) -> dict: + """获取玩家生涯统计（按英雄分类的详细数据）。""" + pid = self._quote_segment(player_id) + return await self._request( # type: ignore[return-value] + f"/players/{pid}/stats/career", + params={"gamemode": gamemode, "platform": platform, "hero": hero}, + use_cache=True, + cache_ttl=120, + ) + + async def get_heroes_stats( + self, + platform: str, + gamemode: str, + region: str, + role: str | None = None, + map_key: str | None = None, + competitive_division: str | None = None, + order_by: str = "hero:asc", + ) -> list: + """获取英雄统计数据（全服英雄选取率/胜率排行榜）。 + + competitive_division 可选 bronze/silver/gold/platinum/emerald/diamond/master/grandmaster + （API 4.13 段位；ultimate 为玩家段位新值，榜单筛选暂不支持）。 + """ + return await self._request( # type: ignore[return-value] + "/heroes/stats", + params={ + "platform": platform, + "gamemode": gamemode, + "region": region, + "role": role, + "map": map_key, + "competitive_division": competitive_division, + "order_by": order_by, + }, + use_cache=True, + cache_ttl=600, + ) + + async def get_hero_info(self, hero_key: str) -> dict: + """获取英雄详细信息。""" + hk = self._quote_segment(hero_key.strip().lower()) + return await self._request(f"/heroes/{hk}", use_cache=True, cache_ttl=3600) # type: ignore[return-value] + + async def list_heroes( + self, + role: str | None = None, + gamemode: str | None = None, + ) -> list: + """获取英雄列表。""" + return await self._request( # type: ignore[return-value] + "/heroes", + params={"role": role, "gamemode": gamemode}, + use_cache=True, + cache_ttl=3600, + ) + + async def list_roles(self) -> list: + """获取所有角色类型。""" + return await self._request("/roles", use_cache=True, cache_ttl=3600) # type: ignore[return-value] + + async def list_gamemodes(self) -> list: + """获取所有游戏模式。""" + return await self._request("/gamemodes", use_cache=True, cache_ttl=3600) # type: ignore[return-value] + + async def list_maps(self, gamemode: str | None = None) -> list: + """获取所有地图。""" + return await self._request( # type: ignore[return-value] + "/maps", + params={"gamemode": gamemode}, + use_cache=True, + cache_ttl=3600, + ) diff --git a/bindings.py b/bindings.py new file mode 100644 index 0000000..b8bf76f --- /dev/null +++ b/bindings.py @@ -0,0 +1,159 @@ +"""KV 多账号绑定存储，含旧版自动迁移。""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from astrbot.api.event import AstrMessageEvent + +_BIND_KEY_PREFIX = "bind_" + + +def bind_key(qq_id: str) -> str: + return f"{_BIND_KEY_PREFIX}{qq_id}" + + +def normalize_loaded(raw, default_platform: str) -> dict | None: + """将 KV 原始值归一化为新版结构；无法识别返回 None；需迁移的旧版返回新结构。""" + empty: dict = {"accounts": [], "default": None} + if raw is None or raw == "": + return empty + if isinstance(raw, dict) and "accounts" in raw: + accounts = [] + for item in raw.get("accounts") or []: + if not isinstance(item, dict): + continue + pid = item.get("player_id") + platform = item.get("platform") + if platform not in ("pc", "console"): + platform = default_platform + if pid: + accounts.append({"player_id": pid, "platform": platform}) + default = raw.get("default") + if not any(a["player_id"] == default for a in accounts): + default = accounts[0]["player_id"] if accounts else None + return {"accounts": accounts, "default": default} + if isinstance(raw, dict) and raw.get("player_id"): + platform = raw.get("platform") + if platform not in ("pc", "console"): + platform = default_platform + return { + "accounts": [{"player_id": raw["player_id"], "platform": platform}], + "default": raw["player_id"], + } + if isinstance(raw, str) and raw: + return { + "accounts": [{"player_id": raw, "platform": default_platform}], + "default": raw, + } + return None + + +class BindingStore: + """依赖 Star 的 get/put/delete_kv_data，需在插件类中混入使用。""" + + _default_platform_fallback = "pc" + + async def _load_bindings(self, event: "AstrMessageEvent") -> dict: + from astrbot.api import logger as _logger + + empty: dict = {"accounts": [], "default": None} + qq_id = event.get_sender_id() + if not qq_id: + return empty + try: + raw = await self.get_kv_data(bind_key(qq_id), None) # type: ignore[attr-defined] + except Exception as e: + _logger.debug(f"读取绑定信息失败: {e}") + return empty + + default_platform = getattr(self, "default_platform", self._default_platform_fallback) + result = normalize_loaded(raw, default_platform) + if result is None: + return empty + # 旧版结构需回写迁移 + if not (isinstance(raw, dict) and "accounts" in raw): + try: + await self.put_kv_data(bind_key(qq_id), result) # type: ignore[attr-defined] + except Exception as e: + _logger.debug(f"迁移绑定数据失败: {e}") + return result + + async def _save_bindings(self, event: "AstrMessageEvent", data: dict) -> bool: + from astrbot.api import logger as _logger + + qq_id = event.get_sender_id() + if not qq_id: + return False + try: + await self.put_kv_data(bind_key(qq_id), data) # type: ignore[attr-defined] + return True + except Exception as e: + _logger.error(f"保存绑定信息失败: {e}") + return False + + async def _get_binding(self, event: "AstrMessageEvent") -> tuple[str | None, str | None]: + data = await self._load_bindings(event) + accounts = data.get("accounts", []) + if not accounts: + return None, None + default = data.get("default") + for acc in accounts: + if acc.get("player_id") == default: + return acc["player_id"], acc.get("platform") + first = accounts[0] + return first.get("player_id"), first.get("platform") + + async def _add_binding( + self, event: "AstrMessageEvent", player_id: str, platform: str = "pc" + ) -> tuple[str, dict | None]: + if platform not in ("pc", "console"): + platform = getattr(self, "default_platform", "pc") + data = await self._load_bindings(event) + accounts = data["accounts"] + for acc in accounts: + if acc["player_id"].lower() == player_id.lower(): + acc["player_id"] = player_id + acc["platform"] = platform + data["default"] = player_id + if await self._save_bindings(event, data): + return "updated", data + return "error", None + max_binds = max(int(getattr(self, "max_binds_per_user", 3) or 3), 1) + if len(accounts) >= max_binds: + return "limit", None + accounts.append({"player_id": player_id, "platform": platform}) + data["default"] = player_id + if await self._save_bindings(event, data): + return "added", data + return "error", None + + async def _remove_binding( + self, event: "AstrMessageEvent", player_id: str | None = None + ) -> tuple[str, str | None]: + from astrbot.api import logger as _logger + + data = await self._load_bindings(event) + accounts = data["accounts"] + if not accounts: + return "empty", None + target = player_id or data.get("default") or accounts[0]["player_id"] + idx = next( + (i for i, a in enumerate(accounts) if a["player_id"].lower() == target.lower()), + None, + ) + if idx is None: + return "not_found", None + removed = accounts.pop(idx) + if not accounts: + try: + await self.delete_kv_data(bind_key(event.get_sender_id())) # type: ignore[attr-defined] + except Exception as e: + _logger.error(f"删除绑定信息失败: {e}") + return "error", None + return "removed", removed["player_id"] + if data.get("default") == removed["player_id"]: + data["default"] = accounts[0]["player_id"] + if await self._save_bindings(event, data): + return "removed", removed["player_id"] + return "error", None diff --git a/constants.py b/constants.py new file mode 100644 index 0000000..7e24e33 --- /dev/null +++ b/constants.py @@ -0,0 +1,149 @@ +"""共享常量：段位/角色/模式/平台/地区/英雄映射。""" + +RANK_MAPPING = { + "bronze": ("青铜", ""), + "silver": ("白银", ""), + "gold": ("黄金", ""), + "platinum": ("铂金", ""), + "emerald": ("翡翠", ""), + "diamond": ("钻石", ""), + "master": ("大师", ""), + "grandmaster": ("宗师", ""), + # API 4.13 起最高段位改名为 ultimate；保留 champion 兼容旧数据/旧自建实例 + "champion": ("王者", ""), + "ultimate": ("王者", ""), + "top500": ("五百强", ""), +} + +ROLE_MAPPING = { + "tank": ("坦克", ""), + "damage": ("输出", ""), + "support": ("支援", ""), +} + +GAMEMODE_MAPPING = { + "quickplay": "快速游戏", + "competitive": "竞技比赛", +} + +GAMEMODE_REVERSE_MAPPING = { + "快速": "quickplay", + "快速游戏": "quickplay", + "qp": "quickplay", + "quick": "quickplay", + "quickplay": "quickplay", + "竞技": "competitive", + "竞技模式": "competitive", + "竞技比赛": "competitive", + "排位": "competitive", + "排位赛": "competitive", + "comp": "competitive", + "competitive": "competitive", +} + +PLATFORM_MAPPING = { + "pc": "PC端", + "console": "主机端", +} + +PLATFORM_REVERSE_MAPPING = { + "pc": "pc", + "电脑": "pc", + "电脑端": "pc", + "端游": "pc", + "console": "console", + "主机": "console", + "主机端": "console", + "ps": "console", + "ps4": "console", + "ps5": "console", + "xbox": "console", + "switch": "console", + "ns": "console", +} + +REGION_MAPPING = { + "asia": "亚服", + "americas": "美服", + "europe": "欧服", +} + +REGION_REVERSE_MAPPING = { + "asia": "asia", + "亚服": "asia", + "亚洲": "asia", + "亚太": "asia", + "americas": "americas", + "america": "americas", + "美服": "americas", + "美洲": "americas", + "europe": "europe", + "eu": "europe", + "欧服": "europe", + "欧洲": "europe", +} + +HERO_NAME_MAPPING = { + "doomfist": "末日铁拳", + "dva": "D.Va", + "dmon": "D.Mon", + "domina": "金驭", + "hazard": "骇灾", + "junker-queen": "渣客女王", + "mauga": "毛加", + "orisa": "奥丽莎", + "ramattra": "拉玛刹", + "reinhardt": "莱因哈特", + "roadhog": "路霸", + "sigma": "西格玛", + "winston": "温斯顿", + "wrecking-ball": "破坏球", + "zarya": "查莉娅", + "anran": "安燃", + "ashe": "艾什", + "bastion": "堡垒", + "cassidy": "卡西迪", + "echo": "回声", + "emre": "埃姆雷", + "freja": "弗蕾娅", + "genji": "源氏", + "hanzo": "半藏", + "junkrat": "狂鼠", + "mei": "美", + "pharah": "法老之鹰", + "reaper": "死神", + "shion": "紫苑", + "sierra": "西拉", + "sojourn": "索杰恩", + "soldier-76": "士兵:76", + "sombra": "黑影", + "symmetra": "秩序之光", + "torbjorn": "托比昂", + "tracer": "猎空", + "vendetta": "斩仇", + "venture": "探奇", + "widowmaker": "黑百合", + "ana": "安娜", + "baptiste": "巴蒂斯特", + "brigitte": "布丽吉塔", + "doctrine": "血律", + "illari": "伊拉锐", + "jetpack-cat": "飞天猫", + "juno": "朱诺", + "kiriko": "雾子", + "lifeweaver": "生命之梭", + "lucio": "卢西奥", + "mercy": "天使", + "mizuki": "瑞稀", + "moira": "莫伊拉", + "wuyang": "无漾", + "zenyatta": "禅雅塔", +} + +HERO_NAME_REVERSE_MAPPING = {cn: en for en, cn in HERO_NAME_MAPPING.items()} + +ROLE_CN_TO_EN = { + "坦克": "tank", "tank": "tank", "重装": "tank", + "输出": "damage", "damage": "damage", + "支援": "support", "support": "support", "辅助": "support", +} diff --git a/docs/archive-2026-09-29.md b/docs/archive-2026-09-29.md new file mode 100644 index 0000000..ac8ceb1 --- /dev/null +++ b/docs/archive-2026-09-29.md @@ -0,0 +1,63 @@ +# 对话归档：OWCX 插件评估、重构与 v2.3.0 更新 + +- 归档日期：2026-09-29 +- 仓库：`https://github.com/TZYCeng/astrbot_plugin_owcx.git` +- 分支：`main` +- 版本跨度：v2.1.0 → v2.2.0 → v2.3.0 + +## 初评发现 + +- `main.py` 约 1904 行，常量、工具与绑定逻辑混杂。 +- 图片与头像下载为每次请求单独创建 aiohttp session；临时图片没有清理。 +- API 请求缺少缓存与 429/503 重试，默认超时 30 秒。 +- 玩家 ID 缺少校验及 URL 编码；重复定义竞技图标；日志实现不统一。 +- 图片开关默认值与配置 schema 不一致；README 仓库地址有误，存在重复 logo 文件。 + +## v2.2.0：重构与健壮性（`9f3deca`） + +- 将映射拆至 `constants.py`，纯逻辑拆至 `utils.py`，绑定存储与迁移拆至 `bindings.py`；`main.py` 缩减至约 700 行，并保留旧导入路径兼容。 +- API 超时改为 15 秒；增加进程级 GET 缓存（玩家 120 秒、榜单 600 秒、英雄 3600 秒）及 429/503 的 `retry_after` 重试（最多 2 次）；对 URL 参数进行 quote。 +- 图片/头像下载复用 session，并限制同时下载数为 6。 +- 新增旧临时图片清理逻辑及英雄统计卡片渲染；补充玩家 ID 校验；统一使用 AstrBot logger。 +- 图片渲染默认值与 schema 统一为开启；修正 README 仓库地址并删除重复 logo 文件。 + +## v2.3.0：新段位与 API 地址配置化（`c1c813e`） + +依据用户提供的 OverFast API 4.13.0 `openapi.json`： + +- 段位枚举新增 `emerald`（翡翠）及 `ultimate`（王者）。 +- `champion` 保留兼容旧数据并显示为王者；`top500` 保留。 +- 54 个 `HeroKey` 与现有英雄映射一致，无需改动。 +- `_conf_schema.json` 增加 `api_base_url`，默认 `https://overfast-api.tekrop.fr`，可填自建 OverFast 服务地址。 +- `normalize_api_base_url()` 去除首尾空白和尾斜杠；空值或非 HTTP(S) 地址回退官方地址。 +- 全部 API 客户端由插件的 `_api_client()` 按配置地址创建，启动日志显示当前 API 地址。 +- `metadata.yaml` 版本更新至 v2.3.0。 + +## 历史提交 + +| 提交 | 日期 | 内容 | +| --- | --- | --- | +| `9f3deca` | 2026-09-29 | v2.2.0 结构重构、缓存重试、下载复用与健壮性修正 | +| `c1c813e` | 2026-09-29 | v2.3.0 新段位与 API 地址配置化 | +| `fa5d65e` | 2026-09-29 | 开发辅助文件与渲染检查（本次已移除相关文件） | + +## 后续关注 + +- 在真实 AstrBot 环境验证 `/owsummary`、`/owstats`、`/owcareer`、`/owherostats` 及绑定指令。 +- 使用真实翡翠/王者账号确认文字与卡片显示。 +- 配置自建 OverFast 服务地址后确认请求走向；上游尚未支持 `ultimate` 榜单筛选时不传该筛选值。 +- 如上游公布不同正式中文译名，可再调整 `champion` 对应显示。 +- 核对新增英雄中文名是否与官方最新译名完全一致。 + +## 关键文件 + +| 文件 | 说明 | +| --- | --- | +| `main.py` | `OverwatchStatsPlugin` 主类与指令入口 | +| `api_client.py` | `OverFastAPIClient` / `OverFastAPIError`、缓存、重试与基础 URL | +| `constants.py` | 段位、角色、模式、平台、地区与英雄映射 | +| `utils.py` | 纯逻辑及 API 地址归一化 | +| `bindings.py` | KV 绑定存储与旧数据迁移 | +| `image_renderer.py` | 卡片渲染与临时图片清理 | +| `_conf_schema.json` | 插件配置 schema | +| `CHANGELOG.md` | 版本变更记录 | diff --git a/image_renderer.py b/image_renderer.py new file mode 100644 index 0000000..c720542 --- /dev/null +++ b/image_renderer.py @@ -0,0 +1,706 @@ +""" +查询结果图片渲染模块。 + +所有渲染函数返回生成的 PNG 文件路径，调用方负责发送 + +""" + +from __future__ import annotations + +import io +import os +import re +import tempfile +from typing import Optional, Sequence + +from PIL import Image, ImageDraw, ImageFont + +# 常见中文字体不含彩色 emoji 字形，直接绘制会变成方框， +# 因此渲染前统一移除 emoji，改用配色区分内容。 +_EMOJI_PATTERN = re.compile( + "[" + "\U0001F000-\U0001FAFF" # 各类 emoji 区块 + "\U00002600-\U000027BF" # 杂项符号与装饰符号 + "\U00002300-\U000023FF" # 杂项技术符号（⏱ 等） + "\U00002B00-\U00002BFF" # 箭头与星形（⭐ 等） + "\U0001F1E6-\U0001F1FF" # 区域指示符 + "]\ufe0f?", + flags=re.UNICODE, +) + + +def _clean_text(text: object) -> str: + """移除文本中的 emoji 并规整空白，避免字体缺字形显示为方框。""" + if text is None: + return "" + cleaned = _EMOJI_PATTERN.sub("", str(text)) + return re.sub(r"\s{2,}", " ", cleaned).strip() + +# ===== 配色方案（OW 风格深色主题）===== +BG_COLOR = (24, 26, 33) # 页面背景 +CARD_COLOR = (34, 37, 46) # 卡片背景 +CARD_COLOR_ALT = (40, 44, 55) # 次级卡片背景 +ACCENT_COLOR = (255, 153, 51) # OW 橙 +TEXT_MAIN = (240, 240, 245) # 主文本 +TEXT_SUB = (205, 210, 224) # 次级文本（提高对比度） +TEXT_DIM = (150, 156, 172) # 弱文本 +DIVIDER_COLOR = (60, 64, 78) # 分隔线 +ROLE_COLORS = { + "坦克": (96, 165, 250), # 蓝 + "输出": (248, 113, 113), # 红 + "支援": (74, 222, 128), # 绿 +} +# 赞赏等级配色（仿游戏内勋章色调） +ENDORSEMENT_COLORS = { + 1: (150, 152, 158), + 2: (92, 184, 92), + 3: (70, 160, 220), + 4: (170, 105, 220), + 5: (240, 165, 45), +} + +# ===== 字体查找 ===== +# 按优先级列出常见的中文字体路径，覆盖 Linux / Windows / macOS +_FONT_CANDIDATES = [ + # 插件自带字体目录（用户可自行放置字体文件） + os.path.join(os.path.dirname(os.path.abspath(__file__)), "fonts", "font.ttf"), + os.path.join(os.path.dirname(os.path.abspath(__file__)), "fonts", "font.ttc"), + # Linux 常见 + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc", + "/usr/share/fonts/noto-cjk/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc", + "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", + "/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf", + # Windows 常见 + "C:/Windows/Fonts/msyh.ttc", + "C:/Windows/Fonts/msyh.ttf", + "C:/Windows/Fonts/simhei.ttf", + "C:/Windows/Fonts/simsun.ttc", + # macOS 常见 + "/System/Library/Fonts/PingFang.ttc", + "/System/Library/Fonts/STHeiti Light.ttc", + "/System/Library/Fonts/Hiragino Sans GB.ttc", +] + +_font_cache: dict[int, ImageFont.FreeTypeFont] = {} + + +def _load_font(size: int) -> ImageFont.FreeTypeFont: + """加载支持中文的字体。 + + Args: + size: 字号。 + + Returns: + 字体对象。找不到中文字体时返回 Pillow 默认字体。 + """ + if size in _font_cache: + return _font_cache[size] + font = None + for path in _FONT_CANDIDATES: + if os.path.exists(path): + try: + font = ImageFont.truetype(path, size) + break + except Exception: + continue + if font is None: + font = ImageFont.load_default() + _font_cache[size] = font + return font + + +def _text_width(draw: ImageDraw.ImageDraw, text: str, font: ImageFont.FreeTypeFont) -> int: + """测量文本宽度。""" + if not text: + return 0 + bbox = draw.textbbox((0, 0), text, font=font) + return bbox[2] - bbox[0] + + +def _text_height(font: ImageFont.FreeTypeFont) -> int: + """获取字体行高。""" + bbox = font.getbbox("国Ag") + return bbox[3] - bbox[1] + + +def _fit_text(draw: ImageDraw.ImageDraw, text: object, font: ImageFont.FreeTypeFont, max_width: int) -> str: + """清理文本并截断到指定宽度内（超出部分以省略号结尾）。""" + text = _clean_text(text) + if _text_width(draw, text, font) <= max_width: + return text + ellipsis = "..." + while text and _text_width(draw, text + ellipsis, font) > max_width: + text = text[:-1] + return (text + ellipsis) if text else ellipsis + + +def _paste_avatar(card: Image.Image, avatar_bytes: bytes, x: int, y: int, size: int) -> None: + """将头像以圆形裁剪后粘贴到卡片上。 + + Args: + card: 目标卡片图像。 + avatar_bytes: 头像图片二进制数据。 + x, y: 粘贴位置左上角坐标。 + size: 头像边长（正方形）。 + """ + try: + avatar = Image.open(io.BytesIO(avatar_bytes)).convert("RGBA") + avatar = avatar.resize((size, size), Image.LANCZOS) + # 圆形遮罩 + mask = Image.new("L", (size, size), 0) + mask_draw = ImageDraw.Draw(mask) + mask_draw.ellipse((0, 0, size, size), fill=255) + card.paste(avatar, (x, y), mask) + # 描边 + border = ImageDraw.Draw(card) + border.ellipse((x, y, x + size, y + size), outline=ACCENT_COLOR, width=3) + except Exception: + # 头像处理失败时静默跳过，不影响整体渲染 + pass + + +def _rounded_rect( + draw: ImageDraw.ImageDraw, + xy: tuple[int, int, int, int], + radius: int, + fill: tuple[int, int, int], +) -> None: + """绘制圆角矩形。""" + draw.rounded_rectangle(xy, radius=radius, fill=fill) + + +def _save_image(img: Image.Image) -> str: + """将图像保存到临时文件并返回路径（顺带清理过期图片防堆积）。""" + try: + cleanup_old_images(max_age_seconds=3600) + except Exception: + pass + fd, path = tempfile.mkstemp(suffix=".png", prefix="ow_render_") + os.close(fd) + img.save(path, format="PNG") + return path + + +def cleanup_old_images(max_age_seconds: int = 3600) -> int: + """删除临时目录中超时的本插件渲染图，返回删除数量。""" + import time as _time + + tmpdir = tempfile.gettempdir() + now = _time.time() + removed = 0 + try: + for name in os.listdir(tmpdir): + if not (name.startswith("ow_render_") and name.endswith(".png")): + continue + full = os.path.join(tmpdir, name) + try: + if now - os.path.getmtime(full) > max_age_seconds: + os.remove(full) + removed += 1 + except Exception: + continue + except Exception: + return removed + return removed + + +# ===== 卡片渲染 ===== + +def _draw_endorsement_badge( + img: Image.Image, x: int, y: int, level: int, size: int = 34 +) -> int: + """绘制游戏内风格的赞赏等级六边形勋章，返回勋章宽度。 + + Args: + img: 目标图像。 + x, y: 勋章左上角坐标。 + level: 赞赏等级（0-5）。 + size: 勋章外接尺寸。 + + Returns: + 勋章占用宽度。 + """ + import math + + level = max(0, min(int(level or 0), 5)) + color = ENDORSEMENT_COLORS.get(level, ENDORSEMENT_COLORS[5]) + cx, cy = x + size // 2, y + size // 2 + r = size // 2 + + badge = Image.new("RGBA", (size, size), (0, 0, 0, 0)) + d = ImageDraw.Draw(badge) + points = [ + (cx + r * math.cos(math.radians(60 * i - 30)), + cy + r * math.sin(math.radians(60 * i - 30))) + for i in range(6) + ] + d.polygon(points, fill=color) + # 内圈 + r2 = r - 4 + points2 = [ + (cx + r2 * math.cos(math.radians(60 * i - 30)), + cy + r2 * math.sin(math.radians(60 * i - 30))) + for i in range(6) + ] + d.polygon(points2, fill=(28, 30, 38)) + img.paste(badge, (x, y), badge) + + # 等级数字 + font = _load_font(int(size * 0.62)) + text = str(level) + tw = _text_width(ImageDraw.Draw(img), text, font) + bbox = font.getbbox(text) + th = bbox[3] - bbox[1] + ImageDraw.Draw(img).text( + (cx - tw // 2, cy - th // 2 - bbox[1]), text, font=font, fill=TEXT_MAIN + ) + return size + + +def _draw_rank_section( + img: Image.Image, + y: int, + rank_rows: Sequence[dict], + width: int, + pad: int, +) -> int: + """绘制竞技段位区（坦克/输出/支援三职责，含段位图标），返回结束 y 坐标。 + + Args: + img: 目标图像。 + y: 起始 y 坐标。 + rank_rows: [{"role_text", "rank_text", "rank_icon_bytes"}, ...]。 + width: 卡片宽度。 + pad: 卡片内边距。 + + Returns: + 绘制结束后的 y 坐标。 + """ + font_body = _load_font(26) + draw = ImageDraw.Draw(img) + draw.text((pad + 6, y), "竞技段位", font=font_body, fill=ACCENT_COLOR) + y += _text_height(font_body) + 12 + for row in rank_rows: + role_text = _fit_text(draw, row.get("role_text", ""), font_body, 200) + rank_text = _fit_text(draw, row.get("rank_text", ""), font_body, width - pad - 320) + color = TEXT_MAIN + for role_cn, c in ROLE_COLORS.items(): + if role_cn in role_text: + color = c + break + draw.text((pad + 20, y + 4), role_text, font=font_body, fill=color) + icon_x = pad + 230 + icon_bytes = row.get("rank_icon_bytes") + if icon_bytes: + _paste_square_icon(img, icon_bytes, icon_x, y, 36, radius=18) + draw.text((icon_x + 46, y + 4), rank_text, font=font_body, fill=TEXT_MAIN) + y += 44 + return y + + +def _draw_heroes_section( + img: Image.Image, + y: int, + top_heroes: Sequence[tuple[str, str, Optional[bytes]]], + width: int, + pad: int, +) -> int: + """绘制常玩英雄区（最多 3 个，含英雄头像、名称与场次），返回结束 y 坐标。 + + Args: + img: 目标图像。 + y: 起始 y 坐标。 + top_heroes: [(英雄名, 场次/时长文本, 英雄头像数据), ...]。 + width: 卡片宽度。 + pad: 卡片内边距。 + + Returns: + 绘制结束后的 y 坐标。 + """ + font_body = _load_font(26) + font_hero = _load_font(22) + font_small = _load_font(20) + draw = ImageDraw.Draw(img) + + draw.line((pad, y, width - pad, y), fill=DIVIDER_COLOR, width=1) + y += 12 + draw.text((pad + 6, y), "常玩英雄", font=font_body, fill=ACCENT_COLOR) + y += _text_height(font_body) + 12 + slot_w = (width - pad * 2) // 3 + portrait_size = 76 + for idx, (hero_name, info_text, portrait_bytes) in enumerate(top_heroes[:3]): + cx = pad + idx * slot_w + if portrait_bytes: + _paste_square_icon(img, portrait_bytes, cx + 4, y, portrait_size, radius=14) + tx = cx + 4 + portrait_size + 12 + name_text = _fit_text(draw, hero_name, font_hero, slot_w - portrait_size - 20) + info_text_c = _fit_text(draw, info_text, font_small, slot_w - portrait_size - 20) + draw.text((tx, y + 12), name_text, font=font_hero, fill=TEXT_MAIN) + draw.text((tx, y + 12 + _text_height(font_hero) + 8), info_text_c, font=font_small, fill=TEXT_SUB) + return y + portrait_size + 16 + + +def _banner_mask(width: int, height: int, radius: int) -> Image.Image: + """生成仅上两角圆润的矩形遮罩（用于名片横幅）。""" + mask = Image.new("L", (width, height), 0) + d = ImageDraw.Draw(mask) + # 先画一个超出底部的圆角矩形，再补齐下半部分，使底部保持直角 + d.rounded_rectangle((0, 0, width, height + radius), radius=radius, fill=255) + d.rectangle((0, height - radius, width, height), fill=255) + return mask + + +def _paste_banner(card: Image.Image, banner_bytes: bytes, xy: tuple[int, int, int, int]) -> None: + """将名片横幅按游戏内效果缩放裁剪后粘贴到卡片指定区域。 + + 名片为宽幅图片，按区域宽度等比缩放并居中裁剪，上两角保持与卡片一致的圆角。 + + Args: + card: 目标卡片图像。 + banner_bytes: 名片图片二进制数据。 + xy: 目标区域 (x1, y1, x2, y2)。 + """ + try: + x1, y1, x2, y2 = xy + w, h = x2 - x1, y2 - y1 + banner = Image.open(io.BytesIO(banner_bytes)).convert("RGBA") + # 等比缩放至覆盖目标区域后居中裁剪 + scale = max(w / banner.width, h / banner.height) + new_size = (int(banner.width * scale + 0.5), int(banner.height * scale + 0.5)) + banner = banner.resize(new_size, Image.LANCZOS) + left = (banner.width - w) // 2 + top = (banner.height - h) // 2 + banner = banner.crop((left, top, left + w, top + h)) + # 底部压暗，保证横幅上的文字可读（alpha 混合而非替换） + overlay = Image.new("RGBA", (w, h), (0, 0, 0, 0)) + od = ImageDraw.Draw(overlay) + for i in range(h): + alpha = int(110 * (i / h) ** 1.5) + od.line((0, i, w, i), fill=(10, 12, 18, alpha)) + banner = Image.alpha_composite(banner, overlay) + mask = _banner_mask(w, h, 18) + card.paste(banner.convert("RGB"), (x1, y1), mask) + except Exception: + # 名片处理失败时静默跳过，不影响整体渲染 + pass + + +def _paste_square_icon(card: Image.Image, icon_bytes: bytes, x: int, y: int, size: int, radius: int = 8) -> None: + """将方形图标（英雄头像/段位图标）圆角粘贴到卡片上。""" + try: + icon = Image.open(io.BytesIO(icon_bytes)).convert("RGBA") + icon = icon.resize((size, size), Image.LANCZOS) + mask = Image.new("L", (size, size), 0) + mask_draw = ImageDraw.Draw(mask) + mask_draw.rounded_rectangle((0, 0, size, size), radius=radius, fill=255) + card.paste(icon, (x, y), mask) + except Exception: + pass + + +def render_summary_card( + username: str, + title: str, + endorsement_level: int, + rank_rows: Sequence[dict], + platform_label: str, + avatar_bytes: Optional[bytes] = None, + namecard_bytes: Optional[bytes] = None, + top_heroes: Sequence[tuple[str, str, Optional[bytes]]] = (), + footer_note: Optional[str] = None, +) -> str: + """渲染玩家摘要卡片（游戏内名片 + 头像组合效果）。 + + 名片作为顶部横幅，头像以圆形叠加在横幅左侧（还原游戏内个人资料页效果）； + 下方依次为竞技段位（含段位图标）与常玩英雄 TOP3（含英雄头像）。 + + Args: + username: 玩家名。 + title: 头衔（可为空）。 + endorsement_level: 赞赏等级。 + rank_rows: [{"role_text", "rank_text", "rank_icon_bytes"}, ...]。 + platform_label: 平台显示文本，如 "PC端"。 + avatar_bytes: 头像图片数据（可选）。 + namecard_bytes: 名片横幅图片数据（可选，纯文字模式不展示）。 + top_heroes: [(英雄名, 游戏时长文本, 英雄头像数据), ...]。 + footer_note: 底部备注（可选）。 + + Returns: + 生成的 PNG 文件路径。 + """ + W = 680 + PAD = 28 + font_title = _load_font(36) + font_sub = _load_font(23) + font_body = _load_font(26) + font_hero = _load_font(22) + font_small = _load_font(20) + + username = _fit_text(ImageDraw.Draw(Image.new("RGB", (1, 1))), username, font_title, W - PAD * 2 - 150) + title = _clean_text(title) + platform_label = _clean_text(platform_label) + footer_note = _clean_text(footer_note) + + banner_h = 150 if namecard_bytes else 0 + avatar_size = 108 if avatar_bytes else 0 + # 无名片时保留原有的文本头部高度 + header_h = banner_h if namecard_bytes else (max(avatar_size, 96) + 10) + rank_block_h = len(rank_rows) * 44 + 58 + heroes_block_h = (150 if top_heroes else 0) + footer_h = 34 if footer_note else 10 + H = PAD + header_h + rank_block_h + heroes_block_h + footer_h + 24 + + img = Image.new("RGB", (W, H), BG_COLOR) + draw = ImageDraw.Draw(img) + _rounded_rect(draw, (10, 10, W - 10, H - 10), 18, CARD_COLOR) + + # ===== 头部：名片横幅 + 头像（游戏内效果）===== + if namecard_bytes: + _paste_banner(img, namecard_bytes, (10, 10, W - 10, 10 + banner_h)) + else: + draw.rectangle((10, 10, W - 10, 16), fill=ACCENT_COLOR) + + text_x = PAD + 10 + text_y_center = 10 + (banner_h // 2 if namecard_bytes else header_h // 2) + if avatar_bytes: + av_x = PAD + 8 + av_y = text_y_center - avatar_size // 2 + _paste_avatar(img, avatar_bytes, av_x, av_y, avatar_size) + text_x = av_x + avatar_size + 18 + + # 用户名 / 头衔（垂直居中于头部区域） + lines_h = _text_height(font_title) + 10 + if title: + lines_h += _text_height(font_sub) + 6 + lines_h += 40 # 赞赏勋章行 + ty = text_y_center - lines_h // 2 + draw.text((text_x, ty), username, font=font_title, fill=TEXT_MAIN) + ty += _text_height(font_title) + 10 + if title: + draw.text((text_x, ty), title, font=font_sub, fill=ACCENT_COLOR) + ty += _text_height(font_sub) + 6 + + # 赞赏等级勋章 + 平台 + badge_size = 34 + _draw_endorsement_badge(img, text_x, ty, endorsement_level, badge_size) + label = _fit_text(draw, f"赞赏 {endorsement_level} 级 {platform_label}", font_sub, W - text_x - badge_size - 24) + draw.text((text_x + badge_size + 10, ty + (badge_size - _text_height(font_sub)) // 2), label, font=font_sub, fill=TEXT_SUB) + + y = 10 + header_h + 12 + draw.line((PAD, y, W - PAD, y), fill=DIVIDER_COLOR, width=2) + y += 14 + + # ===== 竞技段位（坦克/输出/支援三职责，含段位图标）===== + y = _draw_rank_section(img, y, rank_rows, W, PAD) + + # ===== 常玩英雄（含英雄头像与场次/时长）===== + if top_heroes: + y += 4 + y = _draw_heroes_section(img, y, top_heroes, W, PAD) + + if footer_note: + y += 2 + draw.line((PAD, y, W - PAD, y), fill=DIVIDER_COLOR, width=1) + y += 10 + draw.text((PAD + 6, y), _fit_text(draw, footer_note, font_small, W - PAD * 2 - 12), font=font_small, fill=TEXT_DIM) + + return _save_image(img) + + +def render_stats_card( + player_id: str, + gamemode_label: str, + platform_label: str, + stat_rows: Sequence[tuple[str, str]], + rank_rows: Sequence[dict] = (), + top_heroes: Sequence[tuple[str, str, Optional[bytes]]] = (), +) -> str: + """渲染玩家统计概览卡片。 + + Args: + player_id: 玩家 ID。 + gamemode_label: 游戏模式显示文本。 + platform_label: 平台显示文本。 + stat_rows: [(标签, 数值), ...]。 + rank_rows: 竞技模式时传入的三职责段位行（含段位图标），可为空。 + top_heroes: 常玩英雄 [(名称, 场次文本, 头像数据), ...]，可为空。 + + Returns: + 生成的 PNG 文件路径。 + """ + W = 720 + PAD = 28 + font_title = _load_font(30) + font_sub = _load_font(22) + font_body = _load_font(24) + + rows = [(_clean_text(label), _clean_text(value)) for label, value in stat_rows] + per_col = (len(rows) + 1) // 2 + grid_h = per_col * 54 + 10 + rank_h = (len(rank_rows) * 44 + 70) if rank_rows else 0 + heroes_h = 158 if top_heroes else 0 + H = PAD * 2 + 96 + rank_h + grid_h + heroes_h + 20 + + img = Image.new("RGB", (W, H), BG_COLOR) + draw = ImageDraw.Draw(img) + _rounded_rect(draw, (10, 10, W - 10, H - 10), 18, CARD_COLOR) + draw.rectangle((10, 10, W - 10, 16), fill=ACCENT_COLOR) + + y = PAD + 6 + draw.text((PAD + 6, y), "统计概览", font=font_title, fill=ACCENT_COLOR) + y += _text_height(font_title) + 10 + sub = _fit_text( + draw, + f"{player_id} | {_clean_text(gamemode_label)} | {_clean_text(platform_label)}", + font_sub, + W - PAD * 2 - 12, + ) + draw.text((PAD + 6, y), sub, font=font_sub, fill=TEXT_SUB) + y += _text_height(font_sub) + 14 + draw.line((PAD, y, W - PAD, y), fill=DIVIDER_COLOR, width=2) + y += 16 + + # ===== 竞技段位区（仅竞技模式）===== + if rank_rows: + y = _draw_rank_section(img, y, rank_rows, W, PAD) + draw.line((PAD, y, W - PAD, y), fill=DIVIDER_COLOR, width=1) + y += 14 + + # ===== 两列网格布局 ===== + col_w = (W - PAD * 2 - 20) // 2 + for idx, (label, value) in enumerate(rows): + col = idx // per_col + row = idx % per_col + cx = PAD + 6 + col * (col_w + 20) + cy = y + row * 54 + _rounded_rect(draw, (cx - 6, cy - 6, cx + col_w - 6, cy + 42), 10, CARD_COLOR_ALT) + label_text = _fit_text(draw, label, font_body, col_w - 20) + draw.text((cx + 8, cy), label_text, font=font_body, fill=TEXT_SUB) + lw = _text_width(draw, label_text, font_body) + value_text = _fit_text(draw, value, font_body, col_w - lw - 34) + draw.text((cx + 8 + lw + 12, cy), value_text, font=font_body, fill=TEXT_MAIN) + y += per_col * 54 + 6 + + # ===== 常玩英雄（含头像与场次）===== + if top_heroes: + y = _draw_heroes_section(img, y, top_heroes, W, PAD) + + return _save_image(img) + + +def render_career_card( + player_id: str, + gamemode_label: str, + platform_label: str, + hero_filter: Optional[str], + heroes: Sequence[tuple[str, Sequence[tuple[str, str]], Optional[bytes]]], + remaining: int = 0, +) -> str: + """渲染生涯统计卡片（每个英雄区块带英雄头像）。 + + Args: + player_id: 玩家 ID。 + gamemode_label: 游戏模式显示文本。 + platform_label: 平台显示文本。 + hero_filter: 英雄筛选显示文本（可选）。 + heroes: [(英雄名, [(标签, 数值), ...], 英雄头像数据), ...]。 + remaining: 未显示的剩余英雄数量。 + + Returns: + 生成的 PNG 文件路径。 + """ + W = 720 + PAD = 28 + font_title = _load_font(30) + font_sub = _load_font(22) + font_hero = _load_font(26) + font_body = _load_font(23) + + # 计算高度 + hero_blocks_h = 0 + for item in heroes: + rows = item[1] + hero_blocks_h += 44 + max(len(rows), 1) * 36 + 14 + extra_h = 40 if remaining > 0 else 0 + H = PAD * 2 + 96 + hero_blocks_h + extra_h + 16 + + img = Image.new("RGB", (W, H), BG_COLOR) + draw = ImageDraw.Draw(img) + _rounded_rect(draw, (10, 10, W - 10, H - 10), 18, CARD_COLOR) + draw.rectangle((10, 10, W - 10, 16), fill=ACCENT_COLOR) + + y = PAD + 6 + draw.text((PAD + 6, y), "生涯统计", font=font_title, fill=ACCENT_COLOR) + y += _text_height(font_title) + 10 + sub = f"{player_id} | {_clean_text(gamemode_label)} | {_clean_text(platform_label)}" + if hero_filter: + sub += f" | 英雄: {_clean_text(hero_filter)}" + draw.text((PAD + 6, y), _fit_text(draw, sub, font_sub, W - PAD * 2 - 12), font=font_sub, fill=TEXT_SUB) + y += _text_height(font_sub) + 14 + draw.line((PAD, y, W - PAD, y), fill=DIVIDER_COLOR, width=2) + y += 16 + + for item in heroes: + hero_name, rows = item[0], item[1] + portrait_bytes = item[2] if len(item) > 2 else None + _rounded_rect(draw, (PAD, y, W - PAD, y + 36 + max(len(rows), 1) * 36 + 8), 12, CARD_COLOR_ALT) + # 英雄头像（若有）+ 英雄名 + text_x = PAD + 14 + if portrait_bytes: + _paste_square_icon(img, portrait_bytes, PAD + 12, y + 2, 34, radius=10) + text_x = PAD + 12 + 34 + 10 + draw.text((text_x, y + 6), _fit_text(draw, hero_name, font_hero, W - text_x - PAD - 12), font=font_hero, fill=ACCENT_COLOR) + y += 44 + for label, value in rows: + label_text = _fit_text(draw, label, font_body, 180) + draw.text((PAD + 26, y), label_text, font=font_body, fill=TEXT_SUB) + lw = _text_width(draw, label_text, font_body) + value_text = _fit_text(draw, value, font_body, W - PAD * 2 - 26 - lw - 24) + draw.text((PAD + 26 + lw + 12, y), value_text, font=font_body, fill=TEXT_MAIN) + y += 36 + y += 14 + + if remaining > 0: + draw.text((PAD + 14, y), f"... 还有 {remaining} 个英雄的数据未显示", font=font_sub, fill=TEXT_DIM) + + return _save_image(img) + + +def render_herostats_card( + title: str, + subtitle: str, + rows: Sequence[tuple[str, str]], + footer_note: str | None = None, +) -> str: + """渲染全服英雄胜率/选取率榜（文本行列表，避免超长单消息）。""" + W = 720 + PAD = 28 + font_title = _load_font(30) + font_sub = _load_font(22) + font_body = _load_font(23) + + H = PAD * 2 + 96 + max(len(rows), 1) * 34 + (40 if footer_note else 0) + 16 + img = Image.new("RGB", (W, H), BG_COLOR) + draw = ImageDraw.Draw(img) + _rounded_rect(draw, (10, 10, W - 10, H - 10), 18, CARD_COLOR) + draw.rectangle((10, 10, W - 10, 16), fill=ACCENT_COLOR) + + y = PAD + 6 + draw.text((PAD + 6, y), _fit_text(draw, title, font_title, W - PAD * 2 - 12), font=font_title, fill=ACCENT_COLOR) + y += _text_height(font_title) + 10 + draw.text((PAD + 6, y), _fit_text(draw, subtitle, font_sub, W - PAD * 2 - 12), font=font_sub, fill=TEXT_SUB) + y += _text_height(font_sub) + 14 + draw.line((PAD, y, W - PAD, y), fill=DIVIDER_COLOR, width=2) + y += 14 + for name, value in rows: + _rounded_rect(draw, (PAD, y, W - PAD, y + 30), 10, CARD_COLOR_ALT) + draw.text((PAD + 14, y + 3), _fit_text(draw, name, font_body, 260), font=font_body, fill=TEXT_SUB) + draw.text((PAD + 300, y + 3), _fit_text(draw, value, font_body, W - PAD * 2 - 300 - 14), font=font_body, fill=TEXT_MAIN) + y += 34 + if footer_note: + y += 8 + draw.text((PAD + 14, y), _fit_text(draw, footer_note, font_sub, W - PAD * 2 - 28), font=font_sub, fill=TEXT_DIM) + return _save_image(img) diff --git a/logo.png b/logo.png new file mode 100644 index 0000000..d8f9a47 Binary files /dev/null and b/logo.png differ diff --git a/main.py b/main.py new file mode 100644 index 0000000..c06431b --- /dev/null +++ b/main.py @@ -0,0 +1,746 @@ +"""OW战绩查询插件（重构版：常量/工具/绑定已拆分）。""" + +import asyncio + +import aiohttp + +import astrbot.api.message_components as Comp +from astrbot.api import logger +from astrbot.api.all import AstrBotConfig +from astrbot.api.event import filter, AstrMessageEvent +from astrbot.api.star import Context, Star + +from .api_client import OverFastAPIClient +from .bindings import BindingStore +from .constants import ( + GAMEMODE_MAPPING, + GAMEMODE_REVERSE_MAPPING, + HERO_NAME_MAPPING, + HERO_NAME_REVERSE_MAPPING, + PLATFORM_MAPPING, + PLATFORM_REVERSE_MAPPING, + RANK_MAPPING, + REGION_MAPPING, + REGION_REVERSE_MAPPING, + ROLE_CN_TO_EN, + ROLE_MAPPING, +) +from .utils import ( + DEFAULT_API_BASE_URL, + build_rank_data as _build_rank_data, + extract_top_heroes as _extract_top_heroes, + format_number as _format_number, + format_time_played as _format_time_played, + format_time_short as _format_time_short, + get_hero_name_cn as _get_hero_name_cn, + get_platform_display as _get_platform_display, + get_rank_display as _get_rank_display, + get_region_display as _get_region_display, + get_role_display as _get_role_display, + normalize_api_base_url, + normalize_player_id as _normalize_player_id, + resolve_gamemode as _resolve_gamemode, + resolve_hero_name as _resolve_hero_name, + resolve_platform as _resolve_platform, + resolve_region as _resolve_region, + validate_player_id as _validate_player_id, +) + +# 向后兼容：保留旧导入路径 +__all__ = [ + "OverwatchStatsPlugin", + "RANK_MAPPING", + "ROLE_MAPPING", + "GAMEMODE_MAPPING", + "PLATFORM_MAPPING", + "REGION_MAPPING", + "HERO_NAME_MAPPING", +] + +try: + from . import image_renderer + + _RENDERER_AVAILABLE = True + try: + image_renderer.cleanup_old_images() + except Exception: + pass +except Exception: # pragma: no cover + image_renderer = None # type: ignore[assignment] + _RENDERER_AVAILABLE = False + +_DOWNLOAD_SEM = asyncio.Semaphore(6) + + +class OverwatchStatsPlugin(BindingStore, Star): + """OW战绩查询插件主类（基于 OverFast API）。""" + + def __init__(self, context: Context, config: AstrBotConfig | None = None) -> None: + super().__init__(context) + self.config = config or {} + raw_gamemode = self.config.get("default_gamemode", "competitive") + resolved = _resolve_gamemode(str(raw_gamemode)) + self.default_gamemode: str = resolved if resolved in ("quickplay", "competitive") else "competitive" + raw_platform = self.config.get("default_platform", "pc") + resolved_platform = _resolve_platform(str(raw_platform)) + self.default_platform: str = resolved_platform if resolved_platform in ("pc", "console") else "pc" + raw_region = self.config.get("default_region", "asia") + resolved_region = _resolve_region(str(raw_region)) + self.default_region: str = resolved_region if resolved_region in ("asia", "americas", "europe") else "asia" + # 与 _conf_schema.json 保持一致，默认开启图片渲染 + self.enable_image_render: bool = bool(self.config.get("enable_image_render", True)) + self.show_api_error: bool = bool(self.config.get("show_api_error", False)) + # OverFast 开源可自建，允许通过配置覆盖 API 地址 + self.api_base_url: str = normalize_api_base_url( + str(self.config.get("api_base_url", DEFAULT_API_BASE_URL) or DEFAULT_API_BASE_URL) + ) + try: + self.max_binds_per_user: int = max(int(self.config.get("max_binds_per_user", 3) or 3), 1) + except (TypeError, ValueError): + self.max_binds_per_user = 3 + if self.enable_image_render and not _RENDERER_AVAILABLE: + logger.warning("已开启图片渲染，但渲染模块加载失败（可能未安装 Pillow），将回退文字输出。pip install Pillow") + logger.info(f"OW战绩查询插件已加载（图片渲染: {'开启' if self.enable_image_render else '关闭'}，API: {self.api_base_url}）") + + def _api_client(self) -> OverFastAPIClient: + """按配置地址构造 API 客户端（支持自建 OverFast 实例）。""" + return OverFastAPIClient(base_url=self.api_base_url) + + def _effective_platform(self, bound_platform: str | None = None) -> str: + if bound_platform in ("pc", "console"): + return bound_platform + return self.default_platform + + def _api_error_reply(self, e: Exception, friendly: str) -> str: + logger.debug(f"API 错误详情: {type(e).__name__}: {e}", exc_info=True) + if self.show_api_error: + detail = getattr(e, "detail", "") or str(e) + retry_after = getattr(e, "retry_after", None) + reply = f"查询出错: {detail}" + if retry_after: + reply += f"\n建议等待 {retry_after} 秒后重试" + return reply + return friendly + + @staticmethod + async def _download_with_session(session: aiohttp.ClientSession, url: str) -> bytes | None: + if not url: + return None + try: + async with _DOWNLOAD_SEM: + async with session.get(url) as resp: + if resp.status == 200: + return await resp.read() + except Exception as e: + logger.debug(f"下载图片失败: {e}") + return None + + @classmethod + async def _download_many(cls, urls: list[str]) -> list[bytes | None]: + if not urls: + return [] + timeout = aiohttp.ClientTimeout(total=10) + async with aiohttp.ClientSession(timeout=timeout) as session: + return await asyncio.gather(*(cls._download_with_session(session, u) for u in urls)) + + async def _fetch_hero_portraits(self, hero_keys: list[str]) -> dict[str, bytes]: + if not hero_keys: + return {} + portrait_map: dict[str, str] = {} + try: + async with self._api_client() as client: + heroes_list = await client.list_heroes() + for h in heroes_list or []: + if isinstance(h, dict) and h.get("key") and h.get("portrait"): + portrait_map[h["key"]] = h["portrait"] + except Exception as e: + logger.debug(f"获取英雄头像列表失败: {e}") + return {} + urls = [portrait_map.get(k, "") for k in hero_keys] + results = await self._download_many(urls) + return {k: b for k, b in zip(hero_keys, results) if b} + + def _prepare_player_id(self, raw: str, from_binding: bool) -> str: + """绑定存储已是规范化形式直接用；外部输入归一化+校验。""" + if from_binding: + return raw + norm = _normalize_player_id(raw) + try: + _validate_player_id(norm) + except ValueError as e: + raise ValueError(str(e)) from e + return norm + + async def _send_summary_result( + self, event: AstrMessageEvent, full: dict, player_id: str, platform: str, + footer_note: str | None = None, extra_lines: list[str] | None = None, + ): + summary = full.get("summary") if isinstance(full.get("summary"), dict) else full + stats = full.get("stats") if isinstance(full.get("stats"), dict) else {} + username = summary.get("username", player_id) + avatar_url = summary.get("avatar") or "" + namecard_url = summary.get("namecard") or "" + endorsement = summary.get("endorsement", {}) + endorsement_level = endorsement.get("level", 0) if isinstance(endorsement, dict) else 0 + title = summary.get("title") or "" + shown_platform, rank_rows = _build_rank_data(summary, platform) + top_heroes = _extract_top_heroes(stats, platform) + + if self.enable_image_render and _RENDERER_AVAILABLE: + try: + urls = [avatar_url, namecard_url] + [r.get("rank_icon", "") for r in rank_rows] + results = await self._download_many(urls) + avatar_bytes, namecard_bytes = results[0], results[1] + render_rows = [ + {"role_text": r["role_text"], "rank_text": r["rank_text"], "rank_icon_bytes": b} + for r, b in zip(rank_rows, results[2:]) + ] or [{"role_text": "段位", "rank_text": "暂无竞技段位数据", "rank_icon_bytes": None}] + portraits = await self._fetch_hero_portraits([k for k, _ in top_heroes]) + render_heroes = [(_get_hero_name_cn(k), _format_time_short(v), portraits.get(k)) for k, v in top_heroes] + img_path = image_renderer.render_summary_card( + username=username, title=title, endorsement_level=endorsement_level, + rank_rows=render_rows, platform_label=_get_platform_display(shown_platform), + avatar_bytes=avatar_bytes, namecard_bytes=namecard_bytes, + top_heroes=render_heroes, footer_note=footer_note, + ) + yield event.chain_result([Comp.Image.fromFileSystem(img_path)]) + return + except Exception as e: + logger.warning(f"摘要图片渲染失败，回退文字: {e}") + logger.debug("堆栈:", exc_info=True) + + rank_lines = [f" {r['role_text']}: {r['rank_text']}" for r in rank_rows] or [" 暂无竞技段位数据"] + lines = [f"👤 {username}"] + if title: + lines.append(f"头衔: {title}") + lines.append(f"赞赏等级: {endorsement_level}") + lines.append(f"竞技段位 ({_get_platform_display(shown_platform)}):") + lines.extend(rank_lines) + if top_heroes: + lines.append("常玩英雄: " + "、".join(f"{_get_hero_name_cn(k)}({_format_time_played(v)})" for k, v in top_heroes)) + if extra_lines: + lines.extend(extra_lines) + result_text = "\n".join(lines) + if avatar_url: + yield event.chain_result([Comp.Image.fromURL(avatar_url), Comp.Plain(result_text)]) + else: + yield event.plain_result(result_text) + + @filter.command("owsummary") + async def player_summary(self, event: AstrMessageEvent, player_id: str = ""): + """查询玩家摘要信息。用法: /owsummary [玩家ID]""" + bound_platform: str | None = None + from_binding = False + if not player_id or not player_id.strip(): + bound_id, bound_platform = await self._get_binding(event) + if not bound_id: + yield event.plain_result("你还没有绑定 Overwatch ID。\n用法: /owsummary <玩家ID>\n或 /owbind <玩家ID> [平台] 后直接查询") + return + player_id = bound_id + from_binding = True + platform = self._effective_platform(bound_platform) + try: + player_id = self._prepare_player_id(player_id, from_binding) + except ValueError as e: + yield event.plain_result(f"玩家 ID 非法: {e}") + return + try: + async with self._api_client() as client: + full = await client.get_player_full(player_id) + except ValueError as e: + err = str(e).lower() + if "not found" in err or "未找到" in err or "404" in err: + yield event.plain_result(f"未找到玩家 `{player_id}`，请检查 ID。") + return + yield event.plain_result(self._api_error_reply(e, "获取玩家信息失败，请稍后重试。")) + return + except Exception: + logger.debug("获取玩家摘要错误堆栈:", exc_info=True) + yield event.plain_result("获取玩家信息时出错，请稍后重试。") + return + async for res in self._send_summary_result(event, full, player_id, platform): + yield res + + @filter.command("owstats") + async def player_stats(self, event: AstrMessageEvent, player_id: str = "", gamemode: str = ""): + """查询玩家统计概览。用法: /owstats [玩家ID] [快速|竞技]""" + bound_platform: str | None = None + from_binding = False + if not player_id or not player_id.strip(): + bound_id, bound_platform = await self._get_binding(event) + if not bound_id: + yield event.plain_result("你还没有绑定 Overwatch ID。用法: /owstats <玩家ID> [快速|竞技]") + return + player_id = bound_id + from_binding = True + platform = self._effective_platform(bound_platform) + try: + player_id = self._prepare_player_id(player_id, from_binding) + except ValueError as e: + yield event.plain_result(f"玩家 ID 非法: {e}") + return + gamemode = _resolve_gamemode(gamemode or self.default_gamemode) or "" + if gamemode not in ("quickplay", "competitive"): + yield event.plain_result("游戏模式无效。可选: 快速(quickplay)、竞技(competitive)") + return + try: + async with self._api_client() as client: + data = await client.get_player_stats_summary(player_id, gamemode=gamemode, platform=platform) + full: dict = {} + try: + full = await client.get_player_full(player_id) + except Exception as fe: + logger.debug(f"完整数据获取失败: {fe}") + except ValueError as e: + err = str(e).lower() + if "not found" in err or "404" in err: + yield event.plain_result(f"未找到玩家 `{player_id}` 或资料私密。") + return + yield event.plain_result(self._api_error_reply(e, "获取统计失败，请稍后重试。")) + return + except Exception: + logger.debug("获取统计错误堆栈:", exc_info=True) + yield event.plain_result("获取统计信息时出错，请稍后重试。") + return + + general = data.get("general", {}) if isinstance(data, dict) else {} + if not general: + yield event.plain_result(f"玩家 `{player_id}` | {GAMEMODE_MAPPING.get(gamemode, gamemode)}暂无统计数据。") + return + average = general.get("average", {}) if isinstance(general.get("average"), dict) else {} + games_played = general.get("games_played", 0) or 0 + games_won = general.get("games_won", 0) or 0 + games_lost = general.get("games_lost", 0) or 0 + winrate = general.get("winrate", 0) or 0 + kda = general.get("kda", 0) or 0 + eliminations_avg = average.get("eliminations", 0) or 0 + deaths_avg = average.get("deaths", 0) or 0 + damage_avg = average.get("damage", 0) or 0 + healing_avg = average.get("healing", 0) or 0 + time_played = general.get("time_played", 0) or 0 + + summary = full.get("summary") if isinstance(full.get("summary"), dict) else {} + stats_full = full.get("stats") if isinstance(full.get("stats"), dict) else {} + shown_platform, rank_data = _build_rank_data(summary, platform) if summary else (platform, []) + comp_rank_rows = rank_data if gamemode == "competitive" else [] + heroes_stats = data.get("heroes", {}) if isinstance(data.get("heroes"), dict) else {} + top_hero_rows = sorted( + ((k, int(v.get("games_played") or 0)) for k, v in heroes_stats.items() + if isinstance(v, dict) and (v.get("games_played") or 0) > 0), + key=lambda kv: kv[1], reverse=True, + )[:3] + if not top_hero_rows and stats_full: + top_hero_rows = [(k, 0) for k, _ in _extract_top_heroes(stats_full, platform)] + gamemode_label = GAMEMODE_MAPPING.get(gamemode, gamemode) + + if self.enable_image_render and _RENDERER_AVAILABLE: + try: + render_rank_rows = [] + if comp_rank_rows: + icons = await self._download_many([r.get("rank_icon", "") for r in comp_rank_rows]) + for r, b in zip(comp_rank_rows, icons): + render_rank_rows.append({"role_text": r["role_text"], "rank_text": r["rank_text"], "rank_icon_bytes": b}) + portraits = await self._fetch_hero_portraits([k for k, _ in top_hero_rows]) + render_heroes = [(_get_hero_name_cn(k), f"{_format_number(g)}场" if g else "常用英雄", portraits.get(k)) for k, g in top_hero_rows] + stat_rows = [ + ("场次", f"{_format_number(games_played)} ({_format_number(games_won)}胜/{_format_number(games_lost)}负)"), + ("胜率", f"{winrate}%"), + ("KDA", f"{kda:.2f}" if kda else "N/A"), + ("消灭", f"{_format_number(eliminations_avg)}/场" if eliminations_avg else "N/A"), + ("死亡", f"{_format_number(deaths_avg)}/场" if deaths_avg else "N/A"), + ("伤害", f"{_format_number(damage_avg)}/场" if damage_avg else "N/A"), + ("治疗", f"{_format_number(healing_avg)}/场" if healing_avg else "N/A"), + ("游戏时间", _format_time_played(time_played)), + ] + img_path = image_renderer.render_stats_card( + player_id=player_id, gamemode_label=gamemode_label, + platform_label=_get_platform_display(shown_platform), + stat_rows=stat_rows, rank_rows=render_rank_rows, top_heroes=render_heroes, + ) + yield event.chain_result([Comp.Image.fromFileSystem(img_path)]) + return + except Exception as e: + logger.warning(f"统计图片渲染失败，回退文字: {e}") + + lines = [f"{player_id} | {gamemode_label} | {_get_platform_display(shown_platform)}", "━━━━━━━━━━━━"] + if comp_rank_rows: + lines.append(f"竞技段位 ({_get_platform_display(shown_platform)}):") + lines.extend(f" {r['role_text']}: {r['rank_text']}" for r in comp_rank_rows) + lines.extend([ + f"场次: {_format_number(games_played)} ({_format_number(games_won)}胜/{_format_number(games_lost)}负)", + f"胜率: {winrate}%", + f"KDA: {kda:.2f}" if kda else "KDA: N/A", + f"消灭: {_format_number(eliminations_avg)}/场" if eliminations_avg else "消灭: N/A", + f"死亡: {_format_number(deaths_avg)}/场" if deaths_avg else "死亡: N/A", + f"伤害: {_format_number(damage_avg)}/场" if damage_avg else "伤害: N/A", + f"治疗: {_format_number(healing_avg)}/场" if healing_avg else "治疗: N/A", + f"游戏时间: {_format_time_played(time_played)}", + ]) + if top_hero_rows: + lines.append("常玩英雄: " + "、".join(f"{_get_hero_name_cn(k)}({_format_number(g)}场)" if g else _get_hero_name_cn(k) for k, g in top_hero_rows)) + yield event.plain_result("\n".join(lines)) + + @filter.command("owcareer") + async def player_career(self, event: AstrMessageEvent, player_id: str = "", gamemode: str = "", hero: str = ""): + """用法: /owcareer [玩家ID] <快速|竞技> [英雄]""" + bound_platform: str | None = None + from_binding = False + if not player_id or not player_id.strip(): + bound_id, bound_platform = await self._get_binding(event) + if not bound_id: + yield event.plain_result("用法: /owcareer <玩家ID> <游戏模式> [英雄名]") + return + player_id = bound_id + from_binding = True + elif not gamemode or not gamemode.strip(): + resolved = _resolve_gamemode(player_id) + if resolved in ("quickplay", "competitive"): + bound_id, bound_platform = await self._get_binding(event) + if bound_id: + gamemode = player_id + player_id = bound_id + from_binding = True + else: + yield event.plain_result("你还没有绑定 ID。/owbind <玩家ID> 后可 /owcareer 竞技 源氏") + return + platform = self._effective_platform(bound_platform) + resolved_mode = _resolve_gamemode(gamemode) + if not resolved_mode or resolved_mode not in ("quickplay", "competitive"): + yield event.plain_result("请输入有效模式。用法: /owcareer <玩家ID> <快速|竞技> [英雄]") + return + try: + player_id = self._prepare_player_id(player_id, from_binding) + except ValueError as e: + yield event.plain_result(f"玩家 ID 非法: {e}") + return + gamemode = resolved_mode + hero_key = _resolve_hero_name(hero) if hero else None + try: + async with self._api_client() as client: + data = await client.get_player_career_stats(player_id, gamemode=gamemode, platform=platform, hero=hero_key) + except ValueError as e: + err = str(e).lower() + if "not found" in err or "404" in err: + yield event.plain_result(f"未找到玩家 `{player_id}` 或资料私密。") + return + yield event.plain_result(self._api_error_reply(e, "获取生涯统计失败，请稍后重试。")) + return + except Exception: + logger.debug("生涯统计错误堆栈:", exc_info=True) + yield event.plain_result("获取生涯统计时出错，请稍后重试。") + return + if not data or not isinstance(data, dict): + yield event.plain_result(f"玩家 `{player_id}` 暂无生涯统计数据。") + return + hero_cards: list[tuple[str, str, list[tuple[str, str]]]] = [] + remaining = 0 + for hero_key_raw, categories in data.items(): + if not isinstance(categories, dict): + continue + hero_name = _get_hero_name_cn(hero_key_raw) + rows: list[tuple[str, str]] = [] + combat = categories.get("combat", {}) + if combat: + eliminations = combat.get("eliminations") + deaths = combat.get("deaths") + kd = "" + if eliminations and deaths and int(deaths) > 0: + kd = f" (K/D: {float(eliminations) / float(deaths):.2f})" + elif eliminations: + kd = " (K/D: ∞)" + if eliminations is not None: + rows.append(("消灭", f"{_format_number(eliminations)}{kd}")) + if deaths is not None: + rows.append(("死亡", _format_number(deaths))) + if combat.get("damage_done") is not None: + rows.append(("伤害", _format_number(combat.get("damage_done")))) + if combat.get("healing_done") is not None: + rows.append(("治疗", _format_number(combat.get("healing_done")))) + game_stats = categories.get("game", {}) + if game_stats: + gp, gw = game_stats.get("games_played"), game_stats.get("games_won") + if gp is not None and gw is not None: + gl = (gp or 0) - (gw or 0) + wr = (gw / gp * 100) if gp > 0 else 0 + rows.append(("场次", f"{_format_number(gp)} ({_format_number(gw)}胜/{_format_number(gl)}负, {wr:.1f}%)")) + if game_stats.get("time_played"): + rows.append(("时间", _format_time_played(game_stats.get("time_played")))) + hero_cards.append((hero_key_raw, hero_name, rows)) + if len(hero_cards) >= 8: + remaining = max(len(data) - 8, 0) + break + gamemode_label = GAMEMODE_MAPPING.get(gamemode, gamemode) + hero_filter_name = _get_hero_name_cn(hero_key) if hero_key else None + if self.enable_image_render and _RENDERER_AVAILABLE and hero_cards: + try: + portraits = await self._fetch_hero_portraits([k for k, _, _ in hero_cards]) + render_heroes = [(name, rows, portraits.get(key)) for key, name, rows in hero_cards] + img_path = image_renderer.render_career_card( + player_id=player_id, gamemode_label=gamemode_label, + platform_label=_get_platform_display(platform), + hero_filter=hero_filter_name, heroes=render_heroes, remaining=remaining, + ) + yield event.chain_result([Comp.Image.fromFileSystem(img_path)]) + return + except Exception as e: + logger.warning(f"生涯图片渲染失败，回退文字: {e}") + hero_filter = f" | 英雄: {hero_filter_name}" if hero_filter_name else "" + lines = [f"{player_id} | {gamemode_label} | {_get_platform_display(platform)}{hero_filter}", "━━━━━━━━━━━━"] + for _, hero_name, rows in hero_cards: + lines.append(f"\n{hero_name}") + lines.extend(f" {label}: {value}" for label, value in rows) + if remaining > 0: + lines.append(f"\n... 还有 {remaining} 个英雄未显示") + if not hero_cards: + lines.append("暂无生涯统计数据。") + yield event.plain_result("\n".join(lines)) + + @filter.command("owhero") + async def hero_info(self, event: AstrMessageEvent, hero_name: str = ""): + """用法: /owhero <英雄名>""" + if not hero_name or not hero_name.strip(): + yield event.plain_result("请输入英雄名称。用法: /owhero <英雄名>") + return + hero_key = _resolve_hero_name(hero_name) + hero_stat: dict | None = None + try: + async with self._api_client() as client: + data = await client.get_hero_info(hero_key) + try: + all_stats = await client.get_heroes_stats( + platform=self.default_platform, gamemode=self.default_gamemode, region=self.default_region) + for item in all_stats or []: + if isinstance(item, dict) and item.get("hero") == hero_key: + hero_stat = item + break + except Exception as se: + logger.debug(f"英雄胜率获取失败: {se}") + except ValueError as e: + err = str(e).lower() + if "not found" in err or "404" in err: + yield event.plain_result(f"未找到英雄 `{hero_key}`。") + return + yield event.plain_result(self._api_error_reply(e, "获取英雄信息失败，请稍后重试。")) + return + except Exception: + logger.debug("英雄信息错误堆栈:", exc_info=True) + yield event.plain_result("获取英雄信息时出错，请稍后重试。") + return + name = data.get("name", hero_key) + description = data.get("description", "") + role = data.get("role", "") + hp = data.get("hitpoints", {}) if isinstance(data.get("hitpoints"), dict) else {} + portrait = data.get("portrait", "") + abilities = data.get("abilities", []) + story_summary = data.get("story", {}).get("summary", "") if isinstance(data.get("story"), dict) else "" + lines = [f"{name}", f"角色: {_get_role_display(role) if role else '未知'}", + f"生命: {hp.get('health', 0)}{' | 护甲: ' + str(hp.get('armor')) if hp.get('armor') else ''}{' | 护盾: ' + str(hp.get('shields')) if hp.get('shields') else ''}"] + if hero_stat: + lines.append(f"全服({_get_region_display(self.default_region)}|{GAMEMODE_MAPPING.get(self.default_gamemode, self.default_gamemode)}): 胜率 {(hero_stat.get('winrate') or 0):.1f}% | 选取率 {(hero_stat.get('pickrate') or 0):.1f}%") + lines.append("━━━━━━━━━━━━") + if abilities: + lines.append("技能:") + for ability in abilities[:8]: + if isinstance(ability, dict): + desc = ability.get("description", "") + if len(desc) > 80: + desc = desc[:77] + "..." + lines.append(f" • {ability.get('name', '')}: {desc}") + if story_summary: + s = story_summary[:197] + "..." if len(story_summary) > 200 else story_summary + lines += ["", "背景故事:", f" {s}"] + result_text = "\n".join(lines) + if portrait: + yield event.chain_result([Comp.Image.fromURL(portrait), Comp.Plain(result_text)]) + else: + yield event.plain_result(result_text) + + @filter.command("owherostats") + async def heroes_stats(self, event: AstrMessageEvent, role: str = "", region: str = ""): + """用法: /owherostats [坦克|输出|支援] [亚服|美服|欧服]""" + role_filter: str | None = None + resolved_region: str | None = None + for arg in (role, region): + arg = (arg or "").strip() + if not arg: + continue + lowered = arg.lower() + if lowered in ROLE_CN_TO_EN or arg in ROLE_CN_TO_EN: + if role_filter is not None: + yield event.plain_result("角色参数重复。可选: 坦克、输出、支援") + return + role_filter = ROLE_CN_TO_EN.get(lowered) or ROLE_CN_TO_EN.get(arg) + elif _resolve_region(arg): + if resolved_region is not None: + yield event.plain_result("地区参数重复。可选: 亚服、美服、欧服") + return + resolved_region = _resolve_region(arg) + else: + yield event.plain_result(f"无法识别 `{arg}`。用法: /owherostats [角色] [地区]") + return + region_v = resolved_region or self.default_region + try: + async with self._api_client() as client: + stats = await client.get_heroes_stats( + platform=self.default_platform, gamemode=self.default_gamemode, + region=region_v, role=role_filter, order_by="winrate:desc") + except ValueError as e: + yield event.plain_result(self._api_error_reply(e, "获取英雄统计失败，请稍后重试。")) + return + except Exception: + logger.debug("英雄统计错误堆栈:", exc_info=True) + yield event.plain_result("获取英雄统计时出错，请稍后重试。") + return + if not stats or not isinstance(stats, list): + yield event.plain_result("暂无英雄统计数据。") + return + role_cn = ROLE_MAPPING.get(role_filter, (role_filter,))[0] if role_filter else "" + title = "英雄胜率排行" + subtitle = f"{_get_region_display(region_v)}{' | ' + role_cn if role_cn else ''} | {GAMEMODE_MAPPING.get(self.default_gamemode, self.default_gamemode)} | {_get_platform_display(self.default_platform)}" + rows: list[tuple[str, str]] = [] + text_lines = [f"{title} | {subtitle}", "━━━━━━━━━━━━"] + for idx, item in enumerate(stats, 1): + if not isinstance(item, dict): + continue + hn = _get_hero_name_cn(item.get("hero", "")) + wr = item.get("winrate", 0) or 0 + pr = item.get("pickrate", 0) or 0 + rows.append((f"{idx:>2}. {hn}", f"胜率 {wr:.1f}% | 选取率 {pr:.1f}%")) + text_lines.append(f"{idx:>2}. {hn:<6} 胜率 {wr:.1f}% | 选取率 {pr:.1f}%") + text_lines += ["", "可加参数筛选: /owherostats 输出 欧服"] + if self.enable_image_render and _RENDERER_AVAILABLE: + try: + img_path = image_renderer.render_herostats_card(title, subtitle, rows, "可加参数筛选: /owherostats 输出 欧服") + yield event.chain_result([Comp.Image.fromFileSystem(img_path)]) + return + except Exception as e: + logger.warning(f"榜单图片渲染失败，回退文字: {e}") + yield event.plain_result("\n".join(text_lines)) + + @filter.command("owbind") + async def bind_id(self, event: AstrMessageEvent, player_id: str = "", platform: str = ""): + """用法: /owbind <玩家ID> [pc|主机]""" + if not player_id or not player_id.strip(): + yield event.plain_result(f"请输入玩家 ID。用法: /owbind <玩家ID> [平台]\n最多绑定 {self.max_binds_per_user} 个，用 /owbinds 查看") + return + if platform and platform.strip(): + rp = _resolve_platform(platform) + if not rp: + yield event.plain_result("平台无效。可选: pc(电脑端)、console(主机端)") + return + platform_v = rp + else: + platform_v = self.default_platform + try: + player_id = self._prepare_player_id(player_id, False) + except ValueError as e: + yield event.plain_result(f"玩家 ID 非法: {e}") + return + user_name = event.get_sender_name() + current = await self._load_bindings(event) + if not any(a["player_id"].lower() == player_id.lower() for a in current["accounts"]) and len(current["accounts"]) >= self.max_binds_per_user: + yield event.plain_result(f"{user_name} 最多绑定 {self.max_binds_per_user} 个账号。/owunbind 解绑后再试。") + return + try: + async with self._api_client() as client: + await client.get_player_summary(player_id) + except ValueError as e: + err = str(e).lower() + if "not found" in err or "404" in err: + yield event.plain_result(f"未找到玩家 `{player_id}`，请检查 ID。") + return + except Exception: + pass + result, data = await self._add_binding(event, player_id, platform_v) + if result == "limit": + yield event.plain_result(f"最多绑定 {self.max_binds_per_user} 个账号。") + elif result in ("added", "updated"): + action = "绑定" if result == "added" else "更新绑定" + count = len(data["accounts"]) if data else 1 + yield event.plain_result( + f"{user_name} 已{action}: `{player_id}`\n平台: {_get_platform_display(platform_v)}（已设默认）\n" + f"已绑定 {count}/{self.max_binds_per_user}\n/owme 快捷查询，/owbinds 查看列表") + else: + yield event.plain_result("绑定失败，请稍后重试。") + + @filter.command("owunbind") + async def unbind_id(self, event: AstrMessageEvent, player_id: str = ""): + user_name = event.get_sender_name() + target = _normalize_player_id(player_id) if player_id and player_id.strip() else None + result, removed_id = await self._remove_binding(event, target) + if result == "empty": + yield event.plain_result(f"{user_name} 还没有绑定任何 ID。/owbind <玩家ID> 绑定。") + elif result == "not_found": + yield event.plain_result(f"你没有绑定 `{target}`。/owbinds 查看列表。") + elif result == "removed": + remaining = await self._load_bindings(event) + tip = f"剩余 {len(remaining['accounts'])} 个，默认 `{remaining['default']}`" if remaining["accounts"] else "需重新 /owbind 绑定" + yield event.plain_result(f"已解绑: `{removed_id}`\n{tip}") + else: + yield event.plain_result("解绑失败，请稍后重试。") + + @filter.command("owbinds") + async def list_bindings(self, event: AstrMessageEvent): + user_name = event.get_sender_name() + data = await self._load_bindings(event) + accounts = data["accounts"] + if not accounts: + yield event.plain_result(f"{user_name} 还没有绑定 ID。/owbind <玩家ID> 绑定。") + return + default = data.get("default") + lines = [f"{user_name} 的绑定列表 ({len(accounts)}/{self.max_binds_per_user}):", "━━━━━━━━━━━━"] + for idx, acc in enumerate(accounts, 1): + mark = " 默认" if acc["player_id"] == default else "" + lines.append(f"{idx}. `{acc['player_id']}` | {_get_platform_display(acc.get('platform', self.default_platform))}{mark}") + lines += ["", "/owdefault <玩家ID> 切换默认，/owunbind [玩家ID] 解绑"] + yield event.plain_result("\n".join(lines)) + + @filter.command("owdefault") + async def set_default_binding(self, event: AstrMessageEvent, player_id: str = ""): + user_name = event.get_sender_name() + if not player_id or not player_id.strip(): + yield event.plain_result("用法: /owdefault <玩家ID>") + return + target = _normalize_player_id(player_id) + data = await self._load_bindings(event) + if not data["accounts"]: + yield event.plain_result(f"{user_name} 还没有绑定 ID。") + return + matched = next((a for a in data["accounts"] if a["player_id"].lower() == target.lower()), None) + if not matched: + yield event.plain_result(f"你没有绑定 `{target}`。/owbinds 查看列表。") + return + data["default"] = matched["player_id"] + if await self._save_bindings(event, data): + yield event.plain_result(f"已设默认: `{matched['player_id']}`") + else: + yield event.plain_result("设置失败，请稍后重试。") + + @filter.command("owme") + async def quick_summary(self, event: AstrMessageEvent): + bound_id, bound_platform = await self._get_binding(event) + user_name = event.get_sender_name() + if not bound_id: + yield event.plain_result(f"{user_name} 还没有绑定 ID。/owbind <玩家ID> 绑定。") + return + platform = self._effective_platform(bound_platform) + try: + async with self._api_client() as client: + full = await client.get_player_full(bound_id) + except ValueError as e: + err = str(e).lower() + if "not found" in err or "未找到" in err or "404" in err: + yield event.plain_result(f"未找到绑定玩家 `{bound_id}`，可能改名/私密，请重绑。") + return + yield event.plain_result(self._api_error_reply(e, "获取玩家信息失败，请稍后重试。")) + return + except Exception: + logger.debug("owme 错误堆栈:", exc_info=True) + yield event.plain_result("获取玩家信息时出错，请稍后重试。") + return + async for res in self._send_summary_result( + event, full, bound_id, platform, + footer_note=f"快捷查询 | {bound_id} | {_get_platform_display(platform)}", + extra_lines=["", "/owstats [模式] 统计，/owcareer <模式> [英雄] 生涯，/owunbind 解绑"], + ): + yield res diff --git a/metadata.yaml b/metadata.yaml new file mode 100644 index 0000000..903c620 --- /dev/null +++ b/metadata.yaml @@ -0,0 +1,6 @@ +name: astrbot_plugin_owcx +display_name: 守望先锋查询 +desc: Overwatch 查询插件 - 基于 OverFast API 查询玩家战绩、英雄信息等 +version: v2.3.1 # 插件版本号。格式：v1.1.1 或者 v1.1 +author: TZYCeng # 作者 +repo: https://github.com/TZYCeng/astrbot_plugin_owcx # 插件的仓库地址 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..5336407 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +aiohttp>=3.8.0 +Pillow>=9.0.0 diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..96d271e --- /dev/null +++ b/utils.py @@ -0,0 +1,224 @@ +"""纯函数工具：归一化/解析/格式化/数据提取，可单测。""" +from __future__ import annotations + +import re +from urllib.parse import quote + +try: + from .constants import ( + GAMEMODE_REVERSE_MAPPING, + HERO_NAME_MAPPING, + HERO_NAME_REVERSE_MAPPING, + PLATFORM_MAPPING, + PLATFORM_REVERSE_MAPPING, + RANK_MAPPING, + REGION_MAPPING, + REGION_REVERSE_MAPPING, + ROLE_MAPPING, + ) +except ImportError: # 作为独立模块导入时回退 + from constants import ( + GAMEMODE_REVERSE_MAPPING, + HERO_NAME_MAPPING, + HERO_NAME_REVERSE_MAPPING, + PLATFORM_MAPPING, + PLATFORM_REVERSE_MAPPING, + RANK_MAPPING, + REGION_MAPPING, + REGION_REVERSE_MAPPING, + ROLE_MAPPING, + ) + +_PLAYER_ID_RE = re.compile(r"^[A-Za-z0-9\u4e00-\u9fa5_\- ]{3,32}$") +_TIER_ROMAN = ["", "I", "II", "III", "IV", "V"] + +DEFAULT_API_BASE_URL = "https://overfast-api.tekrop.fr" + + +def normalize_api_base_url(raw: str | None) -> str: + """归一化 API 地址：去空格/去尾斜杠，无效时回退官方地址。""" + url = (raw or "").strip().rstrip("/") + if not url or not url.lower().startswith(("http://", "https://")): + return DEFAULT_API_BASE_URL + return url + + +def normalize_player_id(player_id: str) -> str: + """# -> - 并 strip，不做合法性校验。""" + return player_id.replace("#", "-").strip() + + +def validate_player_id(player_id: str) -> str: + """归一化 + 合法性校验，非法时抛 ValueError。 + + 允许字母/数字/中文/_/-/空格，长度归一化后 3~32，且必须含 - 分隔符。 + 返回可直接拼 URL path 的 quote 结果。 + """ + norm = normalize_player_id(player_id) + if len(norm) < 3 or len(norm) > 32: + raise ValueError(f"玩家 ID 长度非法: {player_id!r}") + if "/" in player_id or "\\" in player_id or ".." in norm: + raise ValueError(f"玩家 ID 非法: {player_id!r}") + # BattleTag 归一化后应为 Name-XXXX 形式；宽松校验：至少含 - 或纯字母数字 + if not _PLAYER_ID_RE.match(norm): + raise ValueError(f"玩家 ID 含非法字符: {player_id!r}") + return norm + + +def quote_player_id(player_id: str) -> str: + """返回 URL-safe 的玩家 ID（已归一化+quote）。""" + return quote(validate_player_id(player_id), safe="-") + + +def get_rank_display(division: str | None, tier: int | None) -> str: + if not division: + return "未定位" + rank_info = RANK_MAPPING.get(division.lower(), (division, "")) + tier_str = _TIER_ROMAN[tier] if tier and 1 <= tier <= 5 else "" + name = rank_info[0] + return f"{name}{' ' + tier_str if tier_str else ''}".strip() + + +def build_rank_data(summary: dict, platform: str) -> tuple[str, list[dict]]: + comp = summary.get("competitive") + if not isinstance(comp, dict): + comp = summary.get("competitive_ranks") + comp = comp if isinstance(comp, dict) else {} + + shown_platform = platform + container = comp.get(platform) + if not isinstance(container, dict) or not container: + other_platform = "console" if platform == "pc" else "pc" + other = comp.get(other_platform) + if isinstance(other, dict) and other: + shown_platform = other_platform + container = other + else: + container = None + + rows: list[dict] = [] + if container: + for role_key in ["tank", "damage", "support"]: + role_cn = ROLE_MAPPING.get(role_key, (role_key, ""))[0] + role_text = f"{role_cn}" + role_data = container.get(role_key) + if isinstance(role_data, dict) and role_data.get("division"): + rows.append({ + "role_text": role_text, + "rank_text": get_rank_display(role_data.get("division"), role_data.get("tier")), + "rank_icon": role_data.get("rank_icon") or "", + }) + else: + rows.append({"role_text": role_text, "rank_text": "未定位", "rank_icon": ""}) + return shown_platform, rows + + +def extract_top_heroes(stats: dict, platform: str, count: int = 3) -> list[tuple[str, int]]: + if not isinstance(stats, dict): + return [] + for plat in (platform, "console" if platform == "pc" else "pc"): + plat_stats = stats.get(plat) + if not isinstance(plat_stats, dict): + continue + totals: dict[str, int] = {} + for gamemode in ("quickplay", "competitive"): + gm_stats = plat_stats.get(gamemode) + if not isinstance(gm_stats, dict): + continue + comparisons = gm_stats.get("heroes_comparisons") + if not isinstance(comparisons, dict): + continue + time_played = comparisons.get("time_played") + if not isinstance(time_played, dict): + continue + for item in time_played.get("values") or []: + if not isinstance(item, dict): + continue + hero = item.get("hero") + value = item.get("value") or 0 + if hero: + try: + totals[hero] = totals.get(hero, 0) + int(value) + except (TypeError, ValueError): + continue + if totals: + return sorted(totals.items(), key=lambda kv: kv[1], reverse=True)[:count] + return [] + + +def get_role_display(role_key: str) -> str: + return ROLE_MAPPING.get(role_key.lower(), (role_key, ""))[0] + + +def get_hero_name_cn(hero_key: str) -> str: + return HERO_NAME_MAPPING.get(hero_key.lower(), hero_key) + + +def resolve_hero_name(hero_input: str) -> str: + hero_input = hero_input.strip() + lowered = hero_input.lower() + if lowered in HERO_NAME_MAPPING: + return lowered + if hero_input in HERO_NAME_REVERSE_MAPPING: + return HERO_NAME_REVERSE_MAPPING[hero_input] + for cn_name, en_key in HERO_NAME_REVERSE_MAPPING.items(): + if lowered in cn_name.lower() or cn_name.lower() in lowered: + return en_key + return lowered + + +def resolve_platform(platform_input: str) -> str | None: + if not platform_input: + return None + return PLATFORM_REVERSE_MAPPING.get(platform_input.strip().lower()) + + +def get_platform_display(platform: str) -> str: + return PLATFORM_MAPPING.get(platform, platform) + + +def resolve_region(region_input: str) -> str | None: + if not region_input: + return None + return REGION_REVERSE_MAPPING.get(region_input.strip().lower()) + + +def get_region_display(region: str) -> str: + return REGION_MAPPING.get(region, region) + + +def resolve_gamemode(mode_input: str) -> str | None: + if not mode_input: + return None + return GAMEMODE_REVERSE_MAPPING.get(mode_input.strip().lower()) + + +def format_time_played(seconds: int | float | None) -> str: + if not seconds: + return "0小时" + total_minutes = int(seconds) // 60 + hours = total_minutes // 60 + minutes = total_minutes % 60 + if hours > 0 and minutes > 0: + return f"{hours}小时{minutes}分钟" + elif hours > 0: + return f"{hours}小时" + return f"{minutes}分钟" + + +def format_time_short(seconds: int | float | None) -> str: + if not seconds: + return "0分钟" + total_minutes = int(seconds) // 60 + hours = total_minutes / 60 + if hours >= 1: + return f"{hours:.1f}小时" + return f"{total_minutes}分钟" + + +def format_number(num: int | float | None) -> str: + if num is None: + return "0" + if isinstance(num, float): + return f"{num:,.1f}" if num != int(num) else f"{int(num):,}" + return f"{num:,}"