diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 32656c5..c7c5380 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -7,7 +7,7 @@ on: workflow_dispatch: inputs: release-tag: - description: 'Release Tag (v1.x.x)' + description: 'Release Tag (v2.x.x)' required: true jobs: @@ -20,7 +20,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v4 with: - python-version: '3.11' + python-version: '3.8' - name: Install dependencies run: | diff --git a/.run/BuildDocServe.run.xml b/.run/BuildDocServe.run.xml new file mode 100644 index 0000000..bd561ac --- /dev/null +++ b/.run/BuildDocServe.run.xml @@ -0,0 +1,17 @@ + + + + \ No newline at end of file diff --git a/docs/api/interface.md b/docs/api/interface.md new file mode 100644 index 0000000..c48a3c4 --- /dev/null +++ b/docs/api/interface.md @@ -0,0 +1,346 @@ +## Client + +_mhyy.Client()_ + +> 米哈云游客户端。 + +`versions`: dict + +> 所有游戏类型的版本号字典,若只想获取指定游戏类型的版本号,请使用 get_client_version() 方法。 + +`get_wallet_data`(self, user: [User](#user)) + +> 获取指定用户的钱包数据。 +> +> **形参:** +> +> - user ([**User**](#User)): 发起请求的用户。 +> +> **返回值:** [`WalletData`](#walletdata) - 该用户的钱包数据。 + +`get_notifications`(self, user: [User](#user), *, status: Optional[[NotificationStatus](#notificationstatus)] = None, +type_: Optional[[NotificationType](#notificationtype)] = None, is_sort: Optional[bool] = True) + +> 获取指定用户的通知信息。 +> +> **形参:** +> +> - user ([**User**](#User)): 发起请求的用户。 +> - status (**Optional[[NotificationStatus](#notificationstatus)]**): 筛选指定的通知状态。 +> - type_ (**Optional[[NotificationType](#notificationtype)]**): 筛选指定的通知种类。 +> - Optional (**Optional[bool]**): 是否排序。 +> +> **返回值:** `List[`[`Notification`](#notification)`]` - 一个列表,包含了指定用户的通知信息。 + +`get_client_version`(self, game_type: [GameType](#gametype)) + +> 获取指定游戏类型的版本号,若想获取字典类型的所有版本号,请使用 versions 属性。 +> +> **形参:** +> +> - game_type ([GameType](#gametype)): 游戏类型。 +> +> **返回值:** `str` - 该游戏类型的版本号。 + +## GameType + +_class mhyy.GameType_ + +> 游戏类型。 + +`GenshinImpact` = 0 + +> 云·原神。 + +`StarRail` = 1 + +> 云·星穹铁道。 + +## Notification + +> 通知类。 + +_class mhyy.Notification_ +(id_: str, status: [NotificationStatus](#notificationstatus), type_: +[NotificationType](#notificationtype), priority: int, source: str, desc: str, msg: str, created_at: str) + +!!! Note + + 这里不提供构造方法,因为根本没有导出。 + +`id`: str + +> 通知 ID。 + +`status`: [NotificationStatus](#notificationstatus) + +> 通知状态。 + +`type`: [NotificationType](#notificationtype) + +> 通知种类。 + +`priority`: int + +> 作用未知,根据名称推测是通知的优先级。 + +`source`: str + +> 作用未知。 + +`desc`: str + +> 作用未知,根据名称推测是通知的描述。 + +`msg`: str + +> 一个字符串,包含了 json 文本格式的该通知的内容。 + +`create_at`: str + +> 一个字符串,是秒级的时间戳 (10位),描述了该通知何时被创建。 + +`from_data_dict`(cls, data: dict) + +> 从特定的数据结构生成 Notification。 +> +> **形参:** +> +> - data (**dict**): 消息数据。 +> +> **返回值:** [`Notification`](#notification) - 包装后的消息数据。 + +## NotificationStatus + +_class mhyy.NotificationStatus_ + +> 通知状态。 + +`Read` = 'NotificationStatusRead' + +> 已读。 + +`Unread` = 'NotificationStatusUnread' + +> 未读。 + +`Undefined` = 'NotificationStatusUndefined' + +> 未定义。 + +`get_status_by_name`(cls, status: str) + +> 从字符串获取枚举成员。 +> +> **形参:** +> +> - status (**str**): 成员字符串。 +> +> **返回值:** [`NotificationStatus`](#notificationstatus) - 枚举成员。 + +## NotificationType + +_class mhyy.NotificationType_ + +> 通知种类。 + +`Popup` = 'NotificationTypePopup' + +> 弹窗通知。 + +`Undefined` = 'NotificationTypeUndefined' + +> 未定义。 + +`get_type_by_name`(cls, type_: str) + +> 从字符串获取枚举成员。 +> +> **形参:** +> +> - status (**str**): 成员字符串。 +> +> **返回值:** [`NotificationType`](#notificationtype) - 枚举成员。 + +## User + +_class mhyy.User(combo_token: str, sys_version: str, device_id: str, device_name: str, device_model: str, +*, game_type: Optional[[GameType](#gametype)] = None, client_type: Optional[[UserClientType](#userclienttype)] = +UserClientType.Android, channel: Optional[UserChannel] = UserChannel.Official +)_ + +> 用户类。 +> +> 形参: +> +> - **combo_token** (str): 对应 headers 中的 x-rpc-combo_token。 +> - **sys_version** (str): 对应 headers 中的 x-rpc-sys_version。 +> - **device_id** (str): 对应 headers 中的 x-rpc-device_id。 +> - **device_name** (str): 对应 headers 中的 x-rpc-device_name。 +> - **device_model** (str): 对应 headers 中的 x-rpc-device_model。 +> - **game_type** (Optional[[GameType](#gametype)]): 游戏类型,若为空则将会从 combo_token 中自动识别。 +> - **client_type** (Optional[[UserClientType](#userclienttype)]): 用户的客户端种类。 +> - **channel** (Optional[[UserChannel](#userchannel)]): 用户的游戏渠道。 + +!!! Note + + 这里不提供成员的文档说明,因为他是形参的 Getter。 + +`combo_token`: str + +`sys_version`: str + +`device_id`: str + +`device_name`: str + +`device_model`: str + +`client_type`: str + +`game_type`: [GameType](#gametype) + +`channel`: [UserChannel](#userchannel) + +`get_user_headers`(self) + +> 获取该用户的 headers。 +> +> **返回值:** dict - 字典格式的该用户的 headers。 + +## UserChannel + +_class mhyy.UserChannel_ + +> 游戏渠道。 + +!!! Note + + 目前 mhyy.py 仅支持官方服务器的操作。 + +`Official` = 0 + +> 官方服。 + +## UserClientType + +_class mhyy.UserClientType_ + +> 客户端类型。 + +!!! Note + + 目前 mhyy.py 仅支持模拟安卓设备操作。 + +`Android` = 2 + +> 安卓。 + +## WalletData + +_class mhyy.WalletData_ + +!!! Info + + 事实上,mhyy.WalletData包含五个子类,分别是 + `mhyy.CoinData`、`mhyy.FreeTimeData`、`mhyy.StatusData`、 + `mhyy.StatData`、`mhyy.PlayCardData`。 + + 但这些成员子类都没有导出且`mhyy.WalletData`与他的成员子类都是 `dataclass`。 + 没有构造函数,仅供读取数据。 + + 所以在本文档中不会描述其五个子类,仅当作成员描写。 + +### `coin` + +> 用户的 原点 / 星云币 时长数据。 + +`coin.coin_num`: int + +> 原点 / 星云币 数。 + +`coin.free_coin_num`: int + +> 免费 原点 / 星云币 数。 + +`coin.coin_limit`: int + +> 原点 / 星云币 的数量上限。 + +`coin.exchange`: int + +> 与原点时长的汇率。通常来说,10 原点 / 星云币 = 1 游戏时长。 + +### `free_time` + +> 用户的免费时长数据。 + +`free_time.send_freetime`: int + +> 若该用户是本日第一次登录,那么此处的值为每日赠送的免费时长 (min)。反之,此处恒为 0。 + +`free_time.free_time`: int + +> 总免费时长 (min)。 + +`free_time.free_time_limit`: int + +> 免费时长上限 (min)。 + +`free_time.over_freetime`: int + +> 该用户是本日第一次登录且赠送的时长有一部分超出了免费时长上限,那么此处的值为溢出的免费时长 (min)。 + +### `status` + +> 未知数据 + +!!! Warning + + 这是一条未知的数据,具体内容待补充。 + +`status.status`: int + +`status.msg`: str + +`status.total_time_status`: int + +`status.status_new`: int + +### `stat` + +> 未知数据 + +!!! Warning + + 这是一条未知的数据,具体内容待补充。 + +`vip_point`: str + +### `play_card` + +> 用户的畅玩卡数据 + +`play_card.expire`: str + +> 畅玩卡过期时间(格式未知) + +!!! Warning + + 这是一条未知的数据,请谨慎使用。 + +`play_card.msg`: str + +> 畅玩卡信息 + +`play_card.short_msg`: str + +> 畅玩卡短信息 / 状态 + +`play_card.play_card_limit`: str + +> 未知 + +!!! Warning + + 这是一条未知的数据,具体内容待补充。 \ No newline at end of file diff --git a/docs/index.md b/docs/index.md index 35fccfd..c06608c 100644 --- a/docs/index.md +++ b/docs/index.md @@ -23,11 +23,10 @@ $ pip install mhyy.py ``` -像这样修改你的 headers 后进行签到操作吧 : +像这样创建用户并填写 headers 后进行签到操作吧 : ```pycon ->>> from mhyy import Client, User ->>> client = Client() +>>> from mhyy import User, Client >>> user = User( ... combo_token="x-rpc-combo_token", ... sys_version="x-rpc-sys_version", @@ -35,13 +34,53 @@ $ pip install mhyy.py ... device_name="x-rpc-device_name", ... device_model="x-rpc-device_model" ... ) ->>> r = client.get_wallet(user) +>>> client = Client() +>>> r = client.get_wallet_data(user) >>> r -WalletData({'coin': {'coin_num': '1160', 'free_coin_num': '0', 'coin_limit': '150000', 'exchange': '10'}, ...) ->>> r.is_signin +WalletData(coin=CoinData(coin_num=0, free_coin_num=0, coin_limit=200000, ...)) +>>> r.free_time.free_time +600 +>>> r.free_time.send_freetime +15 +>>> r.is_sign_in() True ->>> r.free_time.sent -datetime.time(0, 15) ->>> r.free_time.total -datetime.time(6, 10) +``` + +是的,mhyy.py 支持 云·原神、云·星穹铁道。你无需填写某一位用户属于什么游戏,mhyy.py 将会自动识别并进行操作。 + +在上述操作中,你成功完成了从**定义用户**到**获取钱包信息**再到**判断是否进行了签到操作**的过程。 + +关于如何获取 headers,本章不做叙述。在未来可能会更新相关内容。 + +## 特性 + +得益于 [httpx](https://www.python-httpx.org/) 的优秀,mhyy.py 有足够的稳定性。 + +- 支持 云·原神、云·星穹铁道。 +- 支持 多种用户公用同一个客户端。 +- 简易的 API 封装。 + +## 文档 + +如果你想要继续学习 mhyy.py 的有关知识,请转到 [快速入门](usage/quick_start.md)。 + +[API 参考](api/interface.md) 提供了mhyy.py API 参考。 + +## 依赖 + +mhyy.py 依赖于以下优秀的开源库。 + +- [`httpx`](https://github.com/encode/httpx/) mhyy.py 的底层实现。 +- [`dataclasses-json`](https://github.com/lidatong/dataclasses-json) 用于封装数据类。 + +## 安装方式 + +!!! Note + + mhyy.py 需要 python 3.8 及以上运行环境! + +使用 pip 安装 mhyy.py : + +```shell +$ pip install mhyy.py ``` diff --git a/docs/usage/quick_start.md b/docs/usage/quick_start.md index e69de29..1cc2eef 100644 --- a/docs/usage/quick_start.md +++ b/docs/usage/quick_start.md @@ -0,0 +1,88 @@ +首先,我们假设你已经安装好了 mhyy.py,如果你还没有,请参见 [安装方式](../index.md#_4) 小节。 + +让我们先导入 mhyy.py。 + +```pycon +>>> import mhyy +``` + +## 定义一个用户 + +在进行这一步之前,你要通过 **抓包** 等方式获取用户的 headers 信息。 + +若你已经拥有了你的 headers,像这样定义一个用户吧: + +```pycon +>>> user = mhyy.User( +... combo_token="x-rpc-combo_token", +... sys_version="x-rpc-sys_version", +... device_id="x-rpc-device_id", +... device_name="x-rpc-device_name", +... device_model="x-rpc-device_model" +... ) +``` + +至此,你已经完成了定义用户的操作。 + +## 定义一个客户端并进行客户端操作 + +这一步你并不需要准备任何东西,你只需要像这样: + +```pycon +>>> client = mhyy.Client() +``` + +至此,你已经完成了客户端的定义。 + +!!! Tip + + 在进行客户端操作前,你需要先了解一下: + + 在 mhyy.py 中不同种游戏是共同使用一个客户端的,那意味着你可以使用同一个客户端对不同游戏种类的不同用户进行操作。 + + 游戏种类将由 mhyy.py 根据 `combo_token` 自动判断或由你自己填写。 + + 参见 [User](../api/interface.md#user)、[GameType](../api/interface.md#gametype) + + +### 获取钱包信息 / 签到 + +这可能是使用 mhyy.py 最多的场景了,你只需要像这样: + +```pycon +>>> wallet = client.get_wallet_data(user=user) +``` + +就可以将这个用户的钱包信息储存到 `wallet` 变量中。 + +这个操作适用于获取该用户的获取钱包信息或者白嫖每天赠送的 15 分钟免费时长。 + +!!! Note + + 只有每天第一次进行 获取钱包信息 / 签到 操作才能视为签到操作 + + 你可以使用这个代码来判断: + + [`wallet.is_sign_in()`](../api/interface.md#walletdata) 将会返回本次操作是否为签到操作。 + +具体返回的内容请参考开发人员接口中的 [WalletData](../api/interface.md#walletdata) 小结。 + +### 获取通知列表 + +这个操作用于获取指定用户的通知信息,像这样: + +```pycon +>>> notifications = client.get_notifications(user=user) +``` + +就可以获取该用户的所有通知。 + +如果你只想获得**未读**通知,你需要修改一下你的代码: + +```pycon +>>> notifications = client.get_notifications(user=user, status=mhyy.NotificationStatus.Unread) +``` + +即可。 + +至此,你已经学会了 mhyy.py 的核心用法,具体客户端操作请参考开发人员接口中的 [Client](../api/interface.md#client) 小节。 \ No newline at end of file diff --git a/mkdocs.yml b/mkdocs.yml index 3e71260..26bff8a 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -19,11 +19,18 @@ repo_url: https://github.com/GuangChen2333/mhyy.py # Navigation nav: - 简介: index.md -# - 使用: -# - 快速开始: usage/quick_start.md + - 使用: + - 快速入门: usage/quick_start.md + - API 参考: + - 开发人员接口: api/interface.md + +# Plugin +plugins: + - tags # noinspection YAMLSchemaValidation markdown_extensions: + - admonition - pymdownx.highlight: use_pygments: true - pymdownx.superfences \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 5569d19..aa11237 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,20 +4,21 @@ build-backend = "setuptools.build_meta" [project] name = "mhyy.py" -version = "1.0.3" +version = "2.0.0" authors = [ { name = "GuangChen2333", email = "guangchenworks@outlook.com" }, ] -description = "Python API for miHoYo Cloud Gaming (Genshin Impact Cloud)" +description = "Python API for miHoYo Cloud Gaming (Genshin Impact Cloud, Honkai:StarRail Cloud)" readme = "README.md" -requires-python = ">=3.7" +requires-python = ">=3.8" classifiers = [ "Programming Language :: Python :: 3", "License :: OSI Approved :: BSD License", "Operating System :: OS Independent", ] dependencies = [ - "httpx" + "httpx", + "dataclasses-json" ] [project.urls] diff --git a/requirements.txt b/requirements.txt index 7e606b5..4ddda55 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ # Dependencies httpx==0.26.0 +dataclasses-json==0.6.4 # Document -mkdocs-material==9.5.3 +mkdocs-material==9.5.7 diff --git a/src/mhyy/__init__.py b/src/mhyy/__init__.py index 5520cba..49d0247 100644 --- a/src/mhyy/__init__.py +++ b/src/mhyy/__init__.py @@ -1,7 +1,10 @@ -# Entities +# Client. from ._client import Client +from ._types import GameType + +# User. from ._user import User +from ._types import UserClientType, UserChannel -# Enums -from ._user import UserType, UserChannel, UserCGGameBiz, UserOpBiz, UserCps -from ._notification import NotificationType, NotificationStatus +# Notification. +from ._types import NotificationType, NotificationStatus diff --git a/src/mhyy/_api.py b/src/mhyy/_api.py index a376111..9cc6cf2 100644 --- a/src/mhyy/_api.py +++ b/src/mhyy/_api.py @@ -1,11 +1,83 @@ -API_CLOUDGAME = "https://api-cloudgame.mihoyo.com" -SDK_STATIC = "https://sdk-static.mihoyo.com" +from ._types import GameType, UserChannel -class APICloudGame: - WALLET = API_CLOUDGAME + "/hk4e_cg_cn/wallet/wallet/get" - NOTIFICATION = API_CLOUDGAME + "/hk4e_cg_cn/gamer/api/listNotifications" +class API: + @staticmethod + def get_launcher_key(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "eYd89JmJ", + GameType.StarRail: "6KcVuOkbcqjJomjZ" + }[game_type] + @staticmethod + def get_launcher_id(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "18", + GameType.StarRail: "33" + }[game_type] -class APIStatic: - VERSION = SDK_STATIC + "/hk4e_cn/mdk/launcher/api/resource?key=eYd89JmJ&launcher_id=18" + @staticmethod + def get_game_version_url(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "https://sdk-static.mihoyo.com/hk4e_cn/mdk/launcher/api/resource", + GameType.StarRail: "https://api-launcher.mihoyo.com/hkrpg_cn/mdk/launcher/api/resource" + }[game_type] + + @staticmethod + def get_app_id(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "1953439974", + GameType.StarRail: "1953445976" + }[game_type] + + @staticmethod + def get_vendor_id(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "1", + GameType.StarRail: "2" + }[game_type] + + @staticmethod + def get_cg_game_biz(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "hk4e_cn", + GameType.StarRail: "hkrpg_cn" + }[game_type] + + @staticmethod + def get_op_biz(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "clgm_cn", + GameType.StarRail: "clgm_hkrpg-cn" + }[game_type] + + @staticmethod + def get_cps(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "cyydmihoyo", + GameType.StarRail: "gw_An_C" + }[game_type] + + @staticmethod + def get_channel_id(user_channel: UserChannel, game_type: GameType) -> str: + channels = { + UserChannel.Official: { + GameType.GenshinImpact: "cyydmihoyo", + GameType.StarRail: "gw_An_C" + } + } + return channels[user_channel][game_type] + + @staticmethod + def get_wallet_data_url(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "https://api-cloudgame.mihoyo.com/hk4e_cg_cn/wallet/wallet/get", + GameType.StarRail: "https://cg-hkrpg-api.mihoyo.com/hkrpg_cn/cg/wallet/wallet/get" + }[game_type] + + @staticmethod + def get_notifications_url(game_type: GameType) -> str: + return { + GameType.GenshinImpact: "https://api-cloudgame.mihoyo.com/hk4e_cg_cn/gamer/api/listNotifications", + GameType.StarRail: "https://cg-hkrpg-api.mihoyo.com/hkrpg_cn/cg/gamer/api/listNotifications" + }[game_type] diff --git a/src/mhyy/_client.py b/src/mhyy/_client.py index b8068e7..7ed6b16 100644 --- a/src/mhyy/_client.py +++ b/src/mhyy/_client.py @@ -1,19 +1,15 @@ -import enum -import typing -import httpx - +from enum import Enum from typing import Optional, List - -from ._api import APIStatic, APICloudGame -from ._wallet import WalletData +from ._types import GameType, NotificationType, NotificationStatus +from ._api import API from ._user import User -from ._exception import WebRequestError, APIError -from ._notification import Notification, NotificationStatus, NotificationType - -T = typing.TypeVar("T", bound="Client") +from ._exceptions import WebRequestError, APIRequestError +from ._wallet import WalletData +from ._notification import Notification +import httpx -class ClientStatus(enum.Enum): +class ClientStatus(Enum): # UNOPENED: # The client has been instantiated, but has not been used to send a web request, # or been opened by entering the context of a `with` block. @@ -27,19 +23,20 @@ class ClientStatus(enum.Enum): class Client: - def __init__(self: T): - self._client = httpx.Client() - self._status = ClientStatus.UNOPENED - version_rep = self._client.get(APIStatic.VERSION) - self._version = version_rep.json()["data"]["game"]["latest"]["version"] - - def __repr__(self) -> str: - return f"Client()" - - def __str__(self) -> str: - return f"Client[version: {self.version}, is_closed: {self.is_closed}, status: {self.status.name}]" + """ + 米哈云游客户端。 + """ + + def __init__(self): + self._client: httpx.Client = httpx.Client() + self._status: ClientStatus = ClientStatus.UNOPENED + # The version will be updated on the first request from the corresponding game. + self._versions: dict = { + GameType.GenshinImpact: None, + GameType.StarRail: None + } - def __enter__(self: T) -> T: + def __enter__(self): if self._status != ClientStatus.UNOPENED: msg = { ClientStatus.OPENED: "Cannot open a client instance more than once.", @@ -54,68 +51,163 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None: self._status = ClientStatus.CLOSED self._client.close() - def close(self) -> None: - self._status = ClientStatus.CLOSED - self._client.close() + def _update_version(self, game_type: GameType) -> None: + """ + 更新客户端版本。 - @property - def is_closed(self) -> bool: - return self._status == ClientStatus.CLOSED + Args: + game_type (GameType): 要更新的客户端类型。 + """ + version_url = API.get_game_version_url(game_type) - @property - def status(self) -> ClientStatus: - return self._status + resp = self._client.get(version_url, params={ + "key": API.get_launcher_key(game_type), + "launcher_id": API.get_launcher_id(game_type) + }).json() - @property - def version(self) -> str: - return self._version + self._versions[game_type] = resp["data"]["game"]["latest"]["version"] + + def _get_common_headers(self, game_type: GameType) -> dict: + """ + 获取指定游戏类型的 headers 常量。 + + Args: + game_type (GameType): 需要获取 headers 的游戏类型。 - def _get_common_headers(self) -> dict: + Returns: + 一个字典,包含了指定游戏的 headers。 + """ return { - "x-rpc-app_version": self._version, - "x-rpc-app_id": "1953439974", - "x-rpc-vendor_id": "1", - "Referer": "https://app.mihoyo.com" + "x-rpc-app_version": self._versions[game_type], + "x-rpc-app_id": API.get_app_id(game_type), + "x-rpc-vendor_id": API.get_vendor_id(game_type), + "x-rpc-cg_game_biz": API.get_cg_game_biz(game_type), + "x-rpc-op_biz": API.get_op_biz(game_type), + "x-rpc-cps": API.get_cps(game_type) } - def _web_get(self, user: User, url: str, *, params: Optional[dict] = None) -> httpx.Response: + def _user_web_get(self, user: User, url: str, params: Optional[dict] = None) -> httpx.Response: + """ + 附带用户 headers 的 get 类型请求。 + + Args: + user (User): 发起请求的用户。 + url (str): 目标 URL。 + params (Optional[dict]): 可选的参数。 + + Returns: + httpx 库的 Response 类。 + """ + + # Check the client state. if self._status == ClientStatus.CLOSED: raise RuntimeError("Cannot send a request, as the client has been closed.") - self._status = ClientStatus.OPENED + # Check version. + if self._versions[user.game_type] is None: + self._update_version(user.game_type) + + # Update client state. + if self._status != ClientStatus.OPENED: + self._status = ClientStatus.OPENED + + # Get the special common headers of the game. + headers: dict = self._get_common_headers(user.game_type) + + user_headers: dict = user.get_user_headers() + user_headers["x-rpc-channel"] = API.get_channel_id(user.channel, user.game_type) + + headers.update(user_headers) - headers = self._get_common_headers() - headers.update(user.header) resp = self._client.get(url, headers=headers, params=params) + if resp.status_code != 200: - raise WebRequestError(f"Status code: {resp.status_code}") + raise WebRequestError( + f"An error occurred in the network request, status code: {resp.status_code}", + resp.status_code + ) + return resp - def get_wallet(self, user: User) -> WalletData: - resp = self._web_get(user, APICloudGame.WALLET) - resp_data = resp.json() - if resp_data["retcode"] != 0: - raise APIError(f"Retcode: {resp_data['retcode']}, Message: {resp_data['message']}") - return WalletData(resp_data["data"]) + def get_wallet_data(self, user: User) -> WalletData: + """ + 获取指定用户的钱包数据。 + + Args: + user (User): 发起请求的用户。 + + Returns: + 该用户的钱包数据。 + """ + r = self._user_web_get(user, API.get_wallet_data_url(user.game_type)).json() + if r['retcode'] != 0: + raise APIRequestError(r['message'], r['retcode']) + return WalletData.from_dict(r['data']) def get_notifications( self, user: User, *, status: Optional[NotificationStatus] = None, - ntype: Optional[NotificationType] = None + type_: Optional[NotificationType] = None, + is_sort: Optional[bool] = True ) -> List[Notification]: - params = {} - if status is not None: + """ + 获取指定用户的通知信息。 + + Args: + user (User): 发起请求的用户。 + status (Optional[NotificationStatus]): 筛选指定的通知状态。 + type_ (Optional[NotificationType]): 筛选指定的通知种类。 + is_sort (Optional[bool]): 是否排序。 + + Returns: + 一个列表,包含了指定用户的指定通知。 + """ + + # Build params. + + params = { + "is_sort": is_sort + } + + # Optional params. + if status: params["status"] = status.value - if ntype is not None: - params["type"] = ntype.value + if type_: + params["type"] = type_.value + + r = self._user_web_get( + user, + API.get_notifications_url(user.game_type), + params=params + ).json() + + if r['retcode'] != 0: + raise APIRequestError(r['message'], r['retcode']) - resp = self._web_get(user, APICloudGame.NOTIFICATION, params=params) - resp_data = resp.json() - if resp_data["retcode"] != 0: - raise APIError(f"Retcode: {resp_data['retcode']}, Message: {resp_data['message']}") notifications = [] - for data in resp_data["data"]["list"]: - notifications.append(Notification(data)) + + for notification_data in r['data']['list']: + notifications.append(Notification.from_data_dict(notification_data)) + return notifications + + def get_client_version(self, game_type: GameType) -> str: + """ + 获取指定游戏类型的版本号,若想获取字典类型的所有版本号,请使用 versions 属性。 + + Args: + game_type (GameType): 游戏类型。 + + Returns: + 该游戏类型的版本号。 + """ + return self._versions[game_type] + + @property + def versions(self) -> dict: + """ + 所有游戏类型的版本号字典,若只想获取指定游戏类型的版本号,请使用 get_client_version() 方法。 + """ + return self._versions diff --git a/src/mhyy/_exception.py b/src/mhyy/_exception.py deleted file mode 100644 index fd99b4a..0000000 --- a/src/mhyy/_exception.py +++ /dev/null @@ -1,13 +0,0 @@ -class WebRequestError(RuntimeError): - def __init__(self, message): - self.message = message - - -class APIError(RuntimeError): - def __init__(self, message): - self.message = message - - -class UndefinedNameWarning(RuntimeWarning): - def __init__(self, message): - self.message = message diff --git a/src/mhyy/_exceptions.py b/src/mhyy/_exceptions.py new file mode 100644 index 0000000..1bae38e --- /dev/null +++ b/src/mhyy/_exceptions.py @@ -0,0 +1,35 @@ +class WebRequestError(RuntimeError): + def __init__(self, message: str, status_code: int): + self._message = message + self._status_code = status_code + + @property + def message(self): + return self._message + + @property + def status_code(self): + return self._status_code + + +class APIRequestError(RuntimeError): + def __init__(self, message: str, ret_code: int): + self._message = message + self._ret_code = ret_code + + @property + def message(self): + return self._message + + @property + def ret_code(self): + return self._ret_code + + +class ComboTokenInvalidError(RuntimeError): + def __init__(self, message: str): + self._message = message + + @property + def message(self): + return self._message diff --git a/src/mhyy/_notification.py b/src/mhyy/_notification.py index 8680ddc..985285e 100644 --- a/src/mhyy/_notification.py +++ b/src/mhyy/_notification.py @@ -1,71 +1,136 @@ -import warnings -import datetime -import enum +from typing import TypeVar +from ._types import NotificationType, NotificationStatus -from ._exception import UndefinedNameWarning - -class NotificationStatus(enum.Enum): - UNREAD = "NotificationStatusUnread" - READ = "NotificationStatusRead" - UNDEFINED = "NotificationStatusUndefined" - - -class NotificationType(enum.Enum): - POPUP = "NotificationTypePopup" - UNDEFINED = "NotificationTypeUndefined" +T = TypeVar("T", bound="Notification") class Notification: - def __init__(self, data: dict): - self._data = data - - if data["status"] not in [status.value for status in NotificationStatus]: - self._status = NotificationStatus.UNDEFINED - warnings.warn(f'The name {data["status"]} is undefined.', UndefinedNameWarning) - else: - self._status = NotificationStatus(data["status"]) - - if data["type"] not in [ntype.value for ntype in NotificationType]: - self._type = NotificationType.UNDEFINED - warnings.warn(f'The name {data["type"]} is undefined.', UndefinedNameWarning) - else: - self._type = NotificationType(data["type"]) + """ + 通知类。 + """ + def __init__( + self, + id_: str, + status: NotificationStatus, + type_: NotificationType, + priority: int, + source: str, + desc: str, + msg: str, + created_at: str + ): + self._id = id_ + self._status = status + self._type = type_ + self._priority = priority + self._source = source + self._desc = desc + self._msg = msg + self._created_at = created_at def __repr__(self) -> str: - return f"Notification({self._data})" - - def __str__(self) -> str: - return f"Notification<{self.msg}>" + return (f"Notification(id={self._id}, status={self._status}, type={self._type}, priority={self._priority}, " + f"source={self._source}, desc={self._desc}, msg={self.msg}, created_at={self._created_at})") + + @classmethod + def from_data_dict(cls, data: dict) -> T: + """ + 从特定的数据结构生成 Notification。 + + Args: + data (dict): 消息数据。 + + Returns: + 包装后的消息数据。 + """ + return cls( + id_=data['id'], + status=NotificationStatus.get_status_by_name(data['status']), + type_=NotificationType.get_type_by_name(data['type']), + priority=data['priority'], + source=data['source'], + desc=data['desc'], + msg=data['msg'], + created_at=data['created_at'] + ) @property def id(self) -> str: - return self._data["id"] + """ + 通知 ID。 + + Returns: + 该通知的 ID。 + """ + return self._id @property def status(self) -> NotificationStatus: + """ + 通知状态。 + + Returns: + 该通知的通知状态。 + """ return self._status @property def type(self) -> NotificationType: + """ + 通知种类。 + + Returns: + 该通知的通知种类。 + """ return self._type @property def priority(self) -> int: - return self._data["priority"] + """ + 作用未知,根据名称推测是通知的优先级。 + + Returns: + 该通知的优先级。 + """ + return self._priority @property def source(self) -> str: - return self._data["source"] + """ + 作用未知。 + + Returns: + 一个字符串,作用未知。 + """ + return self._source @property def desc(self) -> str: - return self._data["desc"] + """ + 作用未知,根据名称推测是通知的描述。 + + Returns: + 该通知的描述。 + """ + return self._desc @property def msg(self) -> str: - return self._data["msg"] + """ + 通知内容。 + + Returns: + 一个字符串,包含了 json 文本格式的该通知的内容。 + """ + return self._msg @property - def created_at(self) -> datetime.datetime: - return datetime.datetime.fromtimestamp(int(self._data["created_at"])) + def create_at(self): + """ + 通知创建时间。 + + Returns: + 一个字符串,是秒级的时间戳 (10位),描述了该通知何时被创建。 + """ + return self._created_at diff --git a/src/mhyy/_types.py b/src/mhyy/_types.py new file mode 100644 index 0000000..5cd9265 --- /dev/null +++ b/src/mhyy/_types.py @@ -0,0 +1,100 @@ +from enum import Enum +from typing import TypeVar + + +class GameType(Enum): + """ + 游戏类型。 + + Attributes: + GenshinImpact: 云·原神。 + StarRail: 云·星穹铁道。 + """ + GenshinImpact = 0 + StarRail = 1 + + +class UserClientType(Enum): + """ + 客户端类型。 + + Attributes: + Android: 安卓。 + """ + Android = 2 + + +class UserChannel(Enum): + """ + 游戏渠道。 + + Attributes: + Official: 官方服。 + """ + Official = 0 + + +T = TypeVar("T", bound="NotificationStatus") + + +class NotificationStatus(Enum): + """ + 通知状态。 + + Attributes: + Read: 已读。 + Unread: 未读。 + Undefined: 未定义。 + """ + Read = "NotificationStatusRead" + Unread = "NotificationStatusUnread" + + Undefined = "NotificationStatusUndefined" + + @classmethod + def get_status_by_name(cls, status: str) -> T: + """ + 从字符串获取枚举成员。 + + Args: + status (str): 成员字符串。 + + Returns: + 枚举成员。 + """ + for member in cls: + if member.value == status: + return member + return cls.Undefined + + +V = TypeVar("V", bound="NotificationType") + + +class NotificationType(Enum): + """ + 通知种类。 + + Attributes: + Popup: 弹窗通知。 + Undefined: 未定义。 + """ + Popup = "NotificationTypePopup" + + Undefined = "NotificationTypeUndefined" + + @classmethod + def get_type_by_name(cls, type_: str) -> V: + """ + 从字符串获取枚举成员。 + + Args: + type_ (str): 成员字符串。 + + Returns: + 枚举成员。 + """ + for member in cls: + if member.value == type_: + return member + return cls.Undefined diff --git a/src/mhyy/_user.py b/src/mhyy/_user.py index de17085..88fe5d6 100644 --- a/src/mhyy/_user.py +++ b/src/mhyy/_user.py @@ -1,27 +1,13 @@ -import enum - - -class UserType(enum.Enum): - AndroidUser = 2 - - -class UserChannel(enum.Enum): - Mihoyo = "cyydmihoyo" - - -class UserCGGameBiz(enum.Enum): - CN = "hk4e_cn" - - -class UserOpBiz(enum.Enum): - CN = "clgm_cn" - - -class UserCps(enum.Enum): - Mihoyo = "cyydmihoyo" +import warnings +from typing import Optional +from ._types import UserChannel, UserClientType, GameType +from ._exceptions import ComboTokenInvalidError class User: + """ + 用户类。 + """ def __init__( self, combo_token: str, @@ -30,38 +16,74 @@ def __init__( device_name: str, device_model: str, *, - user_type: UserType = UserType.AndroidUser, - channel: UserChannel = UserChannel.Mihoyo, - cg_game_biz: UserCGGameBiz = UserCGGameBiz.CN, - op_biz: UserOpBiz = UserOpBiz.CN, - cps: UserCps = UserCps.Mihoyo, - language: str = "zh-cn" + game_type: Optional[GameType] = None, + client_type: Optional[UserClientType] = UserClientType.Android, + channel: Optional[UserChannel] = UserChannel.Official ): + """ + 创建一个用户。 + + Args: + combo_token (str): 对应 headers 中的 x-rpc-combo_token。 + sys_version (str): 对应 headers 中的 x-rpc-sys_version。 + device_id (str): 对应 headers 中的 x-rpc-device_id。 + device_name (str): 对应 headers 中的 x-rpc-device_name。 + device_model (str): 对应 headers 中的 x-rpc-device_model。 + game_type (Optional[GameType]): 游戏类型,若为空则将会从 combo_token 中自动识别。 + client_type (Optional[UserClientType]): 用户的客户端种类。 + channel (Optional[UserChannel]): 用户的游戏渠道。 + """ self._combo_token = combo_token self._sys_version = sys_version self._device_id = device_id self._device_name = device_name self._device_model = device_model - self._user_type = user_type + self._client_type = client_type + self._game_type = game_type self._channel = channel - self._cg_game_biz = cg_game_biz - self._op_biz = op_biz - self._cps = cps - self._language = language - - def __str__(self) -> str: - return (f"User[combo_token: {self._combo_token}, sys_version: {self._sys_version}, " - f"device_id: {self._device_id}, device_name: {self._device_name}, " - f"device_model: {self._device_model}, user_type: {self._user_type.name}, " - f"channel: {self._channel.name}, cg_game_biz: {self._cg_game_biz.name}, " - f"op_biz: {self._op_biz.name}, cps: {self._cps}, language: {self._language}]") - - def __repr__(self) -> str: - return (f"User(combo_token={self._combo_token}, sys_version={self._sys_version}, device_id={self._device_id}, " - f"device_name={self._device_name}, device_model={self._device_model}, " - f"user_type={self._user_type}, channel={self._channel}, " - f"cg_game_biz={self._cg_game_biz}, op_biz={self._op_biz}, " - f"cps={self._cps}, language={self._language})") + + # Automatic detection of the game type. + + try: + bi = self._combo_token.split(";bi=")[1] + except IndexError: + raise ComboTokenInvalidError( + "An error occurred in the automatic detection of the game type, " + "the 'bi' segment was not found in combo token." + ) + + detected_game_type = { + "hk4e_cn": GameType.GenshinImpact, + "hkrpg_cn": GameType.StarRail + }[bi] + + if self._game_type is None: + self._game_type = detected_game_type + else: + if self._game_type != detected_game_type: + warnings.warn( + "The program detected a difference between the GameType you entered and the GameType it detected. " + "This time, it will use your input as the standard. So the data may be incorrect.”" + "Please pay attention to the GameType." + ) + + def get_user_headers(self) -> dict: + """ + 获取该用户的 headers。 + + Returns: + 字典格式的该用户的 header。 + """ + return { + "x-rpc-combo_token": self._combo_token, + "x-rpc-sys_version": self._sys_version, + "x-rpc-device_id": self._device_id, + "x-rpc-device_name": self._device_name, + "x-rpc-device_model": self._device_model, + # Client type in headers must be string. + "x-rpc-client_type": str(self._client_type.value), + "x-rpc-channel": self._client_type + } @property def combo_token(self) -> str: @@ -81,48 +103,16 @@ def device_name(self) -> str: @property def device_model(self) -> str: - return self.device_model - - @property - def user_type(self) -> UserType: - return self._user_type - - @property - def channel(self) -> UserChannel: - return self._channel - - @property - def cps(self) -> UserCps: - return self._cps - - @property - def cg_game_biz(self) -> UserCGGameBiz: - return self.cg_game_biz + return self._device_model @property - def language(self) -> str: - return self._language + def client_type(self) -> UserClientType: + return self._client_type @property - def op_biz(self) -> UserOpBiz: - return self._op_biz + def game_type(self) -> GameType: + return self._game_type @property - def header(self) -> dict: - """ - 用户层的请求头 - :return: 发出 Web 请求时的 `Headers` - """ - return { - "x-rpc-combo_token": self._combo_token, - "x-rpc-client_type": str(self._user_type.value), - "x-rpc-sys_version": self._sys_version, - "x-rpc-channel": self._channel.value, - "x-rpc-cps": self._cps.value, - "x-rpc-cg_game_biz": self._cg_game_biz.value, - "x-rpc-device_id": self._device_id, - "x-rpc-device_name": self._device_name, - "x-rpc-device_model": self._device_model, - "x-rpc-language": self._language, - "x-rpc-op_biz": self._op_biz.value - } + def channel(self) -> UserChannel: + return self._channel diff --git a/src/mhyy/_wallet.py b/src/mhyy/_wallet.py index 3dc6563..657ddf2 100644 --- a/src/mhyy/_wallet.py +++ b/src/mhyy/_wallet.py @@ -1,171 +1,106 @@ -import datetime +from dataclasses import dataclass +from dataclasses_json import dataclass_json -class WalletData: +@dataclass_json +@dataclass(frozen=True) +class CoinData: """ - 用户的钱包数据 + 用户的 原点 / 星云币 时长数据。 + + Attributes: + coin_num (int): 原点 / 星云币 数。 + free_coin_num (int): 免费 原点 / 星云币 数。 + coin_limit (int): 原点 / 星云币 的数量上限。 + exchange (int): 与原点时长的汇率。通常来说,10 原点 / 星云币 = 1 游戏时长。 """ + coin_num: int + free_coin_num: int + coin_limit: int + exchange: int - class _Coin: - """ - 米云币 - """ - def __init__(self, data: dict): - self._data = data - - def __repr__(self) -> str: - return f"WalletData._Coin({self._data})" - - def __str__(self) -> str: - return f"WalletData.Coin[coin: {self.coin}, free: {self.free}, limit: {self.limit}]" - - @property - def coin(self) -> int: - """ - 米云币数量 - """ - return int(self._data["coin_num"]) - - @property - def free(self) -> int: - """ - 免费米云币数量 - """ - return int(self._data["free_coin_num"]) - - @property - def limit(self) -> int: - """ - 米云币上限 - """ - return int(self._data["coin_limit"]) - - class _FreeTime: - """ - 免费时长 - """ +@dataclass_json +@dataclass(frozen=True) +class FreeTimeData: + """ + 用户的免费时长数据。 - def __init__(self, data: dict): - self._data = data - - def __repr__(self) -> str: - return f"WalletData._Freetime({self._data})" - - def __str__(self) -> str: - return (f"WalletData.Freetime[" - f"sent: {self.sent}, total: {self.total}, " - f"limit: {self.limit}, overflow: {self.overflow}]") - - @property - def sent(self) -> datetime.time: - """ - 每日登陆赠送的时长 - :return: 如果是首次请求(即签到动作),则返回赠送的时长,否则为 `0` - """ - send_free_time = int(self._data["send_freetime"]) - return datetime.time(send_free_time // 60, send_free_time % 60) - - @property - def total(self) -> datetime.time: - """ - 总共的免费时长 - """ - free_time = int(self._data["free_time"]) - return datetime.time(free_time // 60, free_time % 60) - - @property - def limit(self) -> datetime.time: - """ - 免费时长上限 - """ - limit = int(self._data["free_time_limit"]) - return datetime.time(limit // 60, limit % 60) - - @property - def overflow(self) -> datetime.time: - """ - 溢出的时长 - :return: 如果是首次请求(即签到动作), 则返回本次签到溢出的时长 - """ - over = int(self._data["over_freetime"]) - return datetime.time(over // 60, over % 60) - - class _PlayCard: - """ - 畅玩卡 - """ + Attributes: + send_freetime (int): 若该用户是本日第一次登录,那么此处的值为每日赠送的免费时长 (min)。反之,此处恒为 0。 + free_time (int): 总免费时长 (min)。 + free_time_limit (int): 免费时长上限 (min)。 + over_freetime (int): 该用户是本日第一次登录且赠送的时长有一部分超出了免费时长上限,那么此处的值为溢出的免费时长 (min)。 + """ + send_freetime: int + free_time: int + free_time_limit: int + over_freetime: int - def __init__(self, data: dict): - self._data = data - - def __repr__(self) -> str: - return f"WalletData._PlayCard(f{self._data})" - - def __str__(self) -> str: - return f"WalletData.PlayCard[expire: {self.expire}, message: {self.message}, status: {self.status}]" - - @property - def expire(self) -> str: - """ - 过期时间 - :return: 字符串形式的过期时间 - """ - return self._data["expire"] - - @property - def message(self) -> str: - """ - 点击开通畅玩卡上面的信息 - :return: 提示信息 - """ - return self._data["msg"] - - @property - def status(self) -> str: - """ - 状态信息 - """ - return self._data["short_msg"] - - def __init__(self, data: dict): - self._data = data - self._coin = self._Coin(data["coin"]) - self._free_time = self._FreeTime(data["free_time"]) - self._play_card = self._PlayCard(data["play_card"]) - - def __repr__(self) -> str: - return f"WalletData({self._data})" - - def __str__(self) -> str: - return f"WalletData[Coin: {self.coin}, FreeTime: {self.free_time}, PlayCard: {self.play_card}]" - - @property - def coin(self): - """ - 米云币 - """ - return self._coin - @property - def free_time(self): - """ - 免费时长 - """ - return self._free_time +@dataclass_json +@dataclass(frozen=True) +class StatusData: + """ + 未知数据 + """ + status: int + msg: str + total_time_status: int + status_new: int - @property - def play_card(self): - """ - 畅玩卡 - """ - return self._play_card - @property - def is_signin(self) -> bool: +@dataclass_json +@dataclass(frozen=True) +class StatData: + """ + 未知数据 + """ + vip_point: str + + +@dataclass_json +@dataclass(frozen=True) +class PlayCardData: + """ + 用户的畅玩卡数据 + + Attributes: + expire (str): 畅玩卡过期时间 + msg (str): 畅玩卡信息 + short_msg (str): 畅玩卡短信息 / 状态 + play_card_limit (str): 未知 + """ + expire: str + msg: str + short_msg: str + play_card_limit: str + + +@dataclass_json +@dataclass(frozen=True) +class WalletData: + """ + 用户的钱包数据。 + + Attributes: + coin (CoinData): 原点时长数据。 + free_time (FreeTimeData): 免费时长数据。 + status (StatusData): 未知。 + stat (StatData): 未知。 + play_card (PlayCardData): 畅玩卡数据。 + """ + coin: CoinData + free_time: FreeTimeData + status: StatusData + stat: StatData + play_card: PlayCardData + + def is_sign_in(self) -> bool: """ - 是否是每日签到动作 - :return: 是则为真 + 判断本次行为是否为签到操作 (本日第一次登录)。 + + Returns: + 若为 True,则本次行为是签到操作。 """ - # 这里的逻辑是如果给的免费时间为 0 就不是第一次上线了 - return self.free_time.sent != datetime.time(0, 0) + return self.free_time.send_freetime != 0