✨ 接入AkashaSystem (#493)

GenshinUID/genshinuid_enka/to_data.py

@@ -13,6 +13,7 @@ from gsuid_core.utils.api.enka.request import get_enka_info
 from gsuid_core.utils.api.minigg.request import get_weapon_info
 
 from .mono.Character import Character
+from ..utils.api.cv.request import _CvApi
 from .draw_normal import get_artifact_score_data
 from ..utils.resource.RESOURCE_PATH import PLAYER_PATH
 from ..utils.ambr_to_minigg import convert_ambr_to_weapon
@@ -409,9 +410,37 @@ async def enka_to_dict(
                 json.dumps(char_data, indent=4, ensure_ascii=False)
             )
 
+    if 0:
+        threading.Thread(
+            target=lambda: asyncio.run(_restore_cv_data(uid, now)),
+            daemon=True,
+        ).start()
+
     return char_dict_list
 
 
+async def _restore_cv_data(uid: str, now: str):
+    cv_api = _CvApi()
+    data = await cv_api.get_rank_data(uid)
+    path = PLAYER_PATH / str(uid) / 'rank.json'
+    if path.exists():
+        async with aiofiles.open(path, 'r', encoding='UTF-8') as file:
+            rank_data = json.loads(await file.read())
+    else:
+        rank_data = {}
+    if not isinstance(data, int):
+        for i in data['data']:
+            rank_data[i['characterId']] = {
+                'calculations': i['calculations'],
+                'time': now,
+            }
+        async with aiofiles.open(path, 'w', encoding='UTF-8') as file:
+            await file.write(
+                json.dumps(rank_data, indent=4, ensure_ascii=False)
+            )
+    await cv_api.close()
+
+
 async def enka_to_data(
     uid: str, enka_data: Optional[EnkaData] = None
 ) -> Union[dict, str]:

GenshinUID/utils/api/cv/api.py

@@ -0,0 +1,3 @@
+MAIN_API = 'https://akasha.cv/api/filters/accounts/'
+RANK_API = 'https://akasha.cv/api/getCalculationsForUser/{}'
+DATA_API = 'https://akasha.cv/api/user/{}'

GenshinUID/utils/api/cv/request.py

@@ -0,0 +1,69 @@
+from urllib.parse import unquote
+from typing import Any, Dict, Union, Literal, Optional
+
+from gsuid_core.logger import logger
+from aiohttp import TCPConnector, ClientSession, ContentTypeError
+
+from .api import DATA_API, MAIN_API, RANK_API
+
+
+class _CvApi:
+    ssl_verify = True
+    _HEADER = {}
+
+    def __init__(self):
+        self.session = ClientSession(
+            connector=TCPConnector(verify_ssl=self.ssl_verify)
+        )
+
+    async def get_session_id(self) -> str:
+        async with self.session.get(MAIN_API) as resp:
+            cookies = resp.cookies
+            cookies_dict = dict(cookies)
+            sid = cookies_dict.get(
+                "connect.sid",
+                {"connect.sid": "NVybrjSdSZISA0JRuKFoZIndoCfDWdA2"},
+            ).get("connect.sid", "NVybrjSdSZISA0JRuKFoZIndoCfDWdA2")
+            sid = unquote(str(sid))
+            sessionID = sid.split(".")[0].split(":")[-1]
+            return sessionID
+
+    async def get_base_data(self, uid: str) -> Union[Dict, int]:
+        sessionID = await self.get_session_id()
+        return await self._cv_request(
+            DATA_API.format(uid), 'GET', self._HEADER, {'sessionID': sessionID}
+        )
+
+    async def get_rank_data(self, uid: str) -> Union[Dict, int]:
+        await self.get_base_data(uid)
+        return await self._cv_request(
+            RANK_API.format(uid), 'GET', self._HEADER
+        )
+
+    async def close(self):
+        # 调用session对象的close方法关闭会话
+        await self.session.close()
+
+    async def _cv_request(
+        self,
+        url: str,
+        method: Literal['GET', 'POST'] = 'GET',
+        header: Dict[str, Any] = _HEADER,
+        params: Optional[Dict[str, Any]] = None,
+        data: Optional[Dict[str, Any]] = None,
+    ) -> Union[Dict, int]:
+        async with self.session.request(
+            method,
+            url=url,
+            headers=header,
+            params=params,
+            json=data,
+            timeout=300,
+        ) as resp:
+            try:
+                raw_data = await resp.json()
+            except ContentTypeError:
+                _raw_data = await resp.text()
+                raw_data = {'retcode': -999, 'data': _raw_data}
+            logger.debug(raw_data)
+            return raw_data