mirror of
https://github.com/PaiGramTeam/SIMNet.git
synced 2024-11-16 03:55:28 +00:00
✨ Add transaction client
This commit is contained in:
parent
40d55ea0ee
commit
a92a7f84f8
104
simnet/client/components/transaction.py
Normal file
104
simnet/client/components/transaction.py
Normal file
@ -0,0 +1,104 @@
|
||||
from typing import Dict, Any, Optional, List, Union
|
||||
from urllib import parse
|
||||
|
||||
from simnet.client.base import BaseClient
|
||||
from simnet.client.routes import YSULOG_URL
|
||||
from simnet.models.genshin.transaction import BaseTransaction, TransactionKind, ItemTransaction, Transaction
|
||||
from simnet.utils.lang import create_short_lang_code
|
||||
|
||||
|
||||
class TransactionClient(BaseClient):
|
||||
"""Transaction component."""
|
||||
|
||||
async def request_transaction(
|
||||
self,
|
||||
endpoint: str,
|
||||
authkey: str,
|
||||
*,
|
||||
method: str = "GET",
|
||||
lang: Optional[str] = None,
|
||||
params: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Make a request towards the transaction log endpoint.
|
||||
|
||||
Args:
|
||||
endpoint (str): The endpoint to make the request to.
|
||||
authkey (str): The authkey to use for the request.
|
||||
method (str, optional): The HTTP method to use. Defaults to "GET".
|
||||
lang (str, optional): The language to use for the request. Defaults to None.
|
||||
params (Dict[str, Any], optional): The parameters to use for the request. Defaults to None.
|
||||
"""
|
||||
params = dict(params or {})
|
||||
|
||||
base_url = YSULOG_URL.get_url(self.region)
|
||||
url = base_url / endpoint
|
||||
|
||||
params["authkey_ver"] = 1
|
||||
params["sign_type"] = 2
|
||||
params["authkey"] = parse.unquote(authkey)
|
||||
params["lang"] = create_short_lang_code(lang or self.lang)
|
||||
|
||||
return await self.request_lab(url, method=method, params=params)
|
||||
|
||||
async def _get_transaction_page(
|
||||
self,
|
||||
end_id: int,
|
||||
kind: str,
|
||||
authkey: str,
|
||||
*,
|
||||
lang: Optional[str] = None,
|
||||
) -> List[BaseTransaction]:
|
||||
"""Get a single page of transactions.
|
||||
|
||||
Args:
|
||||
end_id (int): The ID of the last transaction to get.
|
||||
kind (str): The kind of transaction to get.
|
||||
authkey (str): The authkey to use for the request.
|
||||
lang (str, optional): The language to use for the request. Defaults to None.
|
||||
"""
|
||||
kind = TransactionKind(kind)
|
||||
endpoint = "get" + kind.value.capitalize() + "Log"
|
||||
|
||||
data = await self.request_transaction(
|
||||
endpoint,
|
||||
lang=lang,
|
||||
authkey=authkey,
|
||||
params=dict(end_id=end_id, size=20),
|
||||
)
|
||||
|
||||
transactions: List[BaseTransaction] = []
|
||||
for trans in data["list"]:
|
||||
model = ItemTransaction if "name" in trans else Transaction
|
||||
transactions.append(model(**trans, kind=kind, lang=lang or self.lang))
|
||||
|
||||
return transactions
|
||||
|
||||
async def transaction_log(
|
||||
self,
|
||||
authkey: str,
|
||||
kind: Optional[Union[str, List[str]]] = None,
|
||||
*,
|
||||
limit: Optional[int] = None,
|
||||
lang: Optional[str] = None,
|
||||
end_id: int = 0,
|
||||
) -> List[BaseTransaction]:
|
||||
"""Get the transaction log of a user.
|
||||
|
||||
Arg:
|
||||
authkey (str): The authkey to use for the request.
|
||||
kind (Union[str, List[str]], optional): The kind of transaction to get. Defaults to None.
|
||||
limit (int, optional): The maximum number of transactions to get. Defaults to None.
|
||||
lang (str, optional): The language to use for the request. Defaults to None.
|
||||
end_id (int, optional): The ID of the last transaction to get. Defaults to 0.
|
||||
"""
|
||||
kinds = kind or ["primogem", "crystal", "resin", "artifact", "weapon"]
|
||||
|
||||
if isinstance(kinds, str):
|
||||
kinds = [kinds]
|
||||
|
||||
iterators: List[BaseTransaction] = []
|
||||
for value in kinds:
|
||||
iterator = await self._get_transaction_page(end_id, value, lang=lang, authkey=authkey)
|
||||
iterators.extend(iterator)
|
||||
|
||||
return iterators[: min(len(iterators), limit)] if limit else iterators
|
@ -5,6 +5,7 @@ from simnet.client.components.chronicle.genshin import GenshinBattleChronicleCli
|
||||
from simnet.client.components.daily import DailyRewardClient
|
||||
from simnet.client.components.diary.genshin import GenshinDiaryClient
|
||||
from simnet.client.components.lab import LabClient
|
||||
from simnet.client.components.transaction import TransactionClient
|
||||
from simnet.client.components.wish.genshin import GenshinWishClient
|
||||
from simnet.utils.enum_ import Game
|
||||
|
||||
@ -18,7 +19,8 @@ class GenshinClient(
|
||||
AuthClient,
|
||||
DailyRewardClient,
|
||||
LabClient,
|
||||
TransactionClient,
|
||||
):
|
||||
"""A simple http client for StarRail endpoints."""
|
||||
"""A simple http client for Genshin endpoints."""
|
||||
|
||||
game: Optional[Game] = Game.GENSHIN
|
||||
|
@ -5,13 +5,20 @@ from simnet.client.components.chronicle.genshin import GenshinBattleChronicleCli
|
||||
from simnet.client.components.daily import DailyRewardClient
|
||||
from simnet.client.components.diary.genshin import GenshinDiaryClient
|
||||
from simnet.client.components.lab import LabClient
|
||||
from simnet.client.components.transaction import TransactionClient
|
||||
from simnet.client.components.wish.genshin import GenshinWishClient
|
||||
from simnet.utils.enum_ import Region
|
||||
from simnet.utils.types import CookieTypes, HeaderTypes, TimeoutTypes
|
||||
|
||||
|
||||
class GenshinClient(
|
||||
GenshinBattleChronicleClient, GenshinWishClient, GenshinDiaryClient, AuthClient, DailyRewardClient, LabClient
|
||||
GenshinBattleChronicleClient,
|
||||
GenshinWishClient,
|
||||
GenshinDiaryClient,
|
||||
AuthClient,
|
||||
DailyRewardClient,
|
||||
LabClient,
|
||||
TransactionClient,
|
||||
):
|
||||
def __init__(
|
||||
self,
|
||||
|
@ -27,6 +27,7 @@ __all__ = (
|
||||
"INFO_LEDGER_URL",
|
||||
"HK4E_URL",
|
||||
"CODE_URL",
|
||||
"YSULOG_URL",
|
||||
)
|
||||
|
||||
|
||||
@ -255,6 +256,10 @@ INFO_LEDGER_URL = GameRoute(
|
||||
),
|
||||
)
|
||||
|
||||
YSULOG_URL = InternationalRoute(
|
||||
overseas="https://hk4e-api-os.hoyoverse.com/ysulog/api/",
|
||||
chinese="https://hk4e-api.mihoyo.com/ysulog/api/",
|
||||
)
|
||||
|
||||
HK4E_URL = Route("https://sg-hk4e-api.hoyoverse.com/common/hk4e_global/")
|
||||
|
||||
|
54
simnet/models/genshin/transaction.py
Normal file
54
simnet/models/genshin/transaction.py
Normal file
@ -0,0 +1,54 @@
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
from simnet.models.base import APIModel
|
||||
|
||||
__all__ = ("BaseTransaction", "ItemTransaction", "Transaction", "TransactionKind")
|
||||
|
||||
|
||||
class TransactionKind(str, Enum):
|
||||
"""Possible kind of transaction.
|
||||
|
||||
Attributes:
|
||||
PRIMOGEM: Primogem currency.
|
||||
CRYSTAL: Genesis crystal currency.
|
||||
RESIN: Resin currency.
|
||||
ARTIFACT: Artifact items from domains.
|
||||
WEAPON: Weapon items from domains and wishes.
|
||||
"""
|
||||
|
||||
PRIMOGEM = "primogem"
|
||||
CRYSTAL = "crystal"
|
||||
RESIN = "resin"
|
||||
ARTIFACT = "artifact"
|
||||
WEAPON = "weapon"
|
||||
|
||||
|
||||
class BaseTransaction(APIModel):
|
||||
"""Genshin transaction."""
|
||||
|
||||
kind: TransactionKind
|
||||
|
||||
id: int
|
||||
uid: int
|
||||
time: datetime
|
||||
amount: int = Field(alias="add_num")
|
||||
reason_id: int = Field(alias="reason")
|
||||
|
||||
|
||||
class Transaction(BaseTransaction):
|
||||
"""Genshin transaction of currency."""
|
||||
|
||||
kind: Literal[TransactionKind.PRIMOGEM, TransactionKind.CRYSTAL, TransactionKind.RESIN]
|
||||
|
||||
|
||||
class ItemTransaction(BaseTransaction):
|
||||
"""Genshin transaction of artifacts or weapons."""
|
||||
|
||||
kind: Literal[TransactionKind.ARTIFACT, TransactionKind.WEAPON]
|
||||
|
||||
name: str
|
||||
rarity: int = Field(alias="rank")
|
Loading…
Reference in New Issue
Block a user