ok

core/jx3_data.py

@@ -1,5 +1,5 @@
 import json
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Dict, Any, Optional, List, Union
 from collections import defaultdict
 
@@ -34,8 +34,32 @@ TONG_RANK_NAMES = {
     "本周浩气五十强",
 }
 
+ACHIEVEMENT_CHOICES = [
+    (0, None, "资历总览"),
+    (1, "1", "杂闻总览"),
+    (2, "2", "武学总览"),
+    (3, "3", "修为总览"),
+    (4, "4", "装备总览"),
+    (5, "5", "技艺总览"),
+    (6, "6", "阅读总览"),
+    (7, "7", "任务总览"),
+    (8, "8", "足迹总览"),
+    (9, "9", "战斗总览"),
+    (10, "10", "声望总览"),
+    (11, "11", "秘境总览"),
+    (12, "12", "帮会总览"),
+    (13, "13", "阵营总览"),
+    (14, "15", "节日总览"),
+    (15, "16", "活动总览"),
+    (16, "17", "风雨江湖路总览"),
+    (17, "40", "家园总览"),
+    (18, "41", "剑侠录总览"),
+]
+
+ACHIEVEMENT_CHOICE_MAP = {index: (menu_id, title) for index, menu_id, title in ACHIEVEMENT_CHOICES}
+
 class JX3Service:
-    def __init__(self, api_data_path: str, config: AstrBotConfig, sqlite: AsyncSQLiteDB):
+    def __init__(self, api_data_path: str, config: AstrBotConfig, sqlite: AsyncSQLiteDB, cache_sqlite: Optional[AsyncSQLiteDB] = None):
         # 加载API配置文件
         try:
             with open(api_data_path, 'r', encoding='utf-8') as f:
@@ -49,6 +73,7 @@ class JX3Service:
         self._config = config
         # 引用sqlite
         self._sql_db = sqlite
+        self._cache_db = cache_sqlite or sqlite
 
         # 获取配置中的 Token
         self.token = self._config.get("jx3api_token", "")
@@ -2386,6 +2411,244 @@ class JX3Service:
         return return_data
 
 
+    async def _load_achievement_cache(self, key: str) -> tuple[Optional[Dict[str, Any]], bool]:
+        """读取资历基础数据缓存，返回数据和是否已过期"""
+        try:
+            row = await self._cache_db.select_one("achievement_cache", "key=?", (key,))
+        except Exception as e:
+            logger.error(f"读取资历缓存失败: {e}")
+            return None, True
+
+        if not row:
+            return None, True
+
+        try:
+            payload = json.loads(row.get("content", "{}"))
+            updated_at = datetime.strptime(row.get("updated_at", ""), "%Y-%m-%d %H:%M:%S")
+            expired = datetime.now() - updated_at > timedelta(days=30)
+            return payload, expired
+        except Exception as e:
+            logger.error(f"解析资历缓存失败: {e}")
+            return None, True
+
+
+    async def _save_achievement_cache(self, key: str, payload: Dict[str, Any]):
+        """写入资历基础数据缓存"""
+        try:
+            await self._cache_db.execute(
+                """
+                INSERT INTO achievement_cache (key, content, updated_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(key) DO UPDATE SET
+                    content=excluded.content,
+                    updated_at=excluded.updated_at
+                """,
+                (
+                    key,
+                    json.dumps(payload, ensure_ascii=False),
+                    datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                ),
+            )
+        except Exception as e:
+            logger.error(f"写入资历缓存失败: {e}")
+
+
+    async def _get_achievement_base_data(self, cache_key: str, config_key: str) -> Optional[Dict[str, Any]]:
+        """获取资历菜单或点数数据，优先使用未过期缓存"""
+        cached, expired = await self._load_achievement_cache(cache_key)
+        if cached and not expired:
+            return cached
+
+        data: Optional[Dict[str, Any]] = await self._base_request(config_key, "GET")
+        if data and isinstance(data, dict):
+            await self._save_achievement_cache(cache_key, data)
+            return data
+
+        if cached:
+            logger.warning(f"资历基础数据接口失败，使用旧缓存: {cache_key}")
+            return cached
+
+        return None
+
+
+    def _flatten_achievement_ids(self, values: Any) -> list[int]:
+        """展开菜单中的单个资历 ID 和数组资历 ID"""
+        result = []
+
+        if isinstance(values, list):
+            for item in values:
+                result.extend(self._flatten_achievement_ids(item))
+            return result
+
+        try:
+            result.append(int(values))
+        except (TypeError, ValueError):
+            pass
+
+        return result
+
+
+    def _category_achievement_ids(self, category: Dict[str, Any]) -> list[int]:
+        """提取分类及其子分类包含的资历 ID"""
+        ids = self._flatten_achievement_ids(category.get("achievements", []))
+        for child in category.get("children", []) or []:
+            if isinstance(child, dict):
+                ids.extend(self._flatten_achievement_ids(child.get("achievements", [])))
+        return ids
+
+
+    def _build_achievement_progress(
+        self,
+        name: str,
+        achievement_ids: list[int],
+        point_map: Dict[str, Any],
+        completed_ids: set[int],
+    ) -> Dict[str, Any]:
+        """按资历点数计算完成进度"""
+        unique_ids = set(achievement_ids)
+        total_points = 0
+        completed_points = 0
+        completed_count = 0
+
+        for achievement_id in unique_ids:
+            try:
+                point = int(point_map.get(str(achievement_id), 0) or 0)
+            except (TypeError, ValueError):
+                point = 0
+
+            total_points += point
+            if achievement_id in completed_ids:
+                completed_count += 1
+                completed_points += point
+
+        percent = round(completed_points / total_points * 100, 2) if total_points else 0
+
+        return {
+            "name": name,
+            "total_points": total_points,
+            "completed_points": completed_points,
+            "percent": percent,
+            "percent_text": f"{percent:.2f}%",
+            "achievement_count": len(unique_ids),
+            "completed_count": completed_count,
+        }
+
+
+    async def zili(self, name: str, server: str, choice: int) -> Dict[str, Any]:
+        """角色资历"""
+        return_data = self._init_return_data()
+
+        if choice not in ACHIEVEMENT_CHOICE_MAP:
+            return_data["msg"] = "无效序号，结束会话"
+            return return_data
+
+        role_params = {"server": server, "name": name, "token": self.token}
+        role_data: Optional[Dict[str, Any]] = await self._base_request(
+            "jx3_jueshexinxi", "GET", params=role_params
+        )
+        if not role_data or not isinstance(role_data, dict):
+            return_data["msg"] = "未查询到角色"
+            return return_data
+
+        global_id = role_data.get("globalId")
+        if not global_id:
+            return_data["msg"] = "无法获取角色全区 ID"
+            return return_data
+
+        achievement_data = await self._api.get(
+            "https://next2.jx3box.com/api/next2/user-achievements",
+            params={"jx3id": global_id},
+            out_key="data",
+        )
+        if not achievement_data or not isinstance(achievement_data, dict):
+            return_data["msg"] = "未查询到资历数据"
+            return return_data
+
+        completed_ids = set()
+        achievements = achievement_data.get("achievements") or ""
+        if isinstance(achievements, str):
+            for item in achievements.split(","):
+                try:
+                    completed_ids.add(int(item.strip()))
+                except ValueError:
+                    continue
+        elif isinstance(achievements, list):
+            completed_ids = set(self._flatten_achievement_ids(achievements))
+
+        menu_payload = await self._get_achievement_base_data(
+            "achievement_menus",
+            "jx3box_achievement_menus",
+        )
+        point_payload = await self._get_achievement_base_data(
+            "achievement_points",
+            "jx3box_achievement_points",
+        )
+        if not menu_payload or not point_payload:
+            return_data["msg"] = "基础资历数据获取失败"
+            return return_data
+
+        menus = menu_payload.get("menus", {})
+        points = point_payload.get("points", {})
+        if not isinstance(menus, dict) or not isinstance(points, dict):
+            return_data["msg"] = "基础资历数据格式异常"
+            return return_data
+
+        selected_menu_id, selected_title = ACHIEVEMENT_CHOICE_MAP[choice]
+        items = []
+
+        if choice == 0:
+            total_ids = []
+            for _, menu_id, title in ACHIEVEMENT_CHOICES[1:]:
+                category = menus.get(str(menu_id), {})
+                if not isinstance(category, dict):
+                    continue
+                category_ids = self._category_achievement_ids(category)
+                total_ids.extend(category_ids)
+                items.append(
+                    self._build_achievement_progress(title, category_ids, points, completed_ids)
+                )
+            summary = self._build_achievement_progress(selected_title, total_ids, points, completed_ids)
+        else:
+            category = menus.get(str(selected_menu_id), {})
+            if not isinstance(category, dict) or not category:
+                return_data["msg"] = "未找到该资历分类"
+                return return_data
+
+            category_ids = self._category_achievement_ids(category)
+            summary = self._build_achievement_progress(selected_title, category_ids, points, completed_ids)
+            for child in category.get("children", []) or []:
+                if not isinstance(child, dict):
+                    continue
+                child_ids = self._flatten_achievement_ids(child.get("achievements", []))
+                items.append(
+                    self._build_achievement_progress(child.get("name", "未命名"), child_ids, points, completed_ids)
+                )
+
+        try:
+            return_data["temp"] = await load_template("zili.html")
+        except FileNotFoundError as e:
+            logger.error(f"加载模板失败: {e}")
+            return_data["msg"] = "系统错误：模板文件不存在"
+            return return_data
+
+        return_data["data"] = {
+            "title": selected_title,
+            "summary": summary,
+            "items": items,
+            "role_name": role_data.get("roleName", name),
+            "server": role_data.get("serverName", server),
+            "zone": role_data.get("zoneName", ""),
+            "force_name": role_data.get("forceName", "无"),
+            "camp_name": role_data.get("campName", "无"),
+            "tong_name": role_data.get("tongName", "无"),
+            "updated_at": achievement_data.get("updated_at", ""),
+            "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        }
+        return_data["code"] = 200
+
+        return return_data
+
+
     async def jueshe(self,name: str, server: str) -> Dict[str, Any]:
         """角色"""
         return_data = self._init_return_data()
@@ -2408,7 +2671,7 @@ class JX3Service:
                 f"服务器：{data.get('zoneName', '无')}·{data.get('serverName', '无')}\n"
                 f"名称：{data.get('roleName', '无')}\n"
                 f"角色ID：{data.get('roleId', '无')}\n"
-                f"全区ID：{data.get('globalRoleId', '无')}\n"
+                f"全区ID：{data.get('globalId', '无')}\n"
                 f"职业：{data.get('forceName', '无')}·{data.get('bodyName', '无')}\n"
                 f"帮会：{data.get('tongName', '无')}\n"
                 f"阵营：{data.get('campName', '无')}\n"