|
|
import asyncio |
|
|
from base64 import urlsafe_b64decode |
|
|
from datetime import datetime, timezone |
|
|
from functools import lru_cache |
|
|
from typing import TYPE_CHECKING, Any, Literal, Optional |
|
|
|
|
|
from pydantic import BaseModel, Field |
|
|
|
|
|
from hibiapi.api.bika.constants import BikaConstants |
|
|
from hibiapi.utils.net import BaseNetClient |
|
|
|
|
|
if TYPE_CHECKING: |
|
|
from .api import BikaEndpoints |
|
|
|
|
|
|
|
|
class BikaLogin(BaseModel): |
|
|
email: str |
|
|
password: str |
|
|
|
|
|
|
|
|
class JWTHeader(BaseModel): |
|
|
alg: str |
|
|
typ: Literal["JWT"] |
|
|
|
|
|
|
|
|
class JWTBody(BaseModel): |
|
|
id: str = Field(alias="_id") |
|
|
iat: datetime |
|
|
exp: datetime |
|
|
|
|
|
|
|
|
@lru_cache(maxsize=4) |
|
|
def load_jwt(token: str): |
|
|
def b64pad(data: str): |
|
|
return data + "=" * (-len(data) % 4) |
|
|
|
|
|
head, body, _ = token.split(".") |
|
|
head_data = JWTHeader.parse_raw(urlsafe_b64decode(b64pad(head))) |
|
|
body_data = JWTBody.parse_raw(urlsafe_b64decode(b64pad(body))) |
|
|
return head_data, body_data |
|
|
|
|
|
|
|
|
class NetRequest(BaseNetClient): |
|
|
_token: Optional[str] = None |
|
|
|
|
|
def __init__(self): |
|
|
super().__init__( |
|
|
headers=BikaConstants.DEFAULT_HEADERS.copy(), |
|
|
proxies=BikaConstants.CONFIG["proxy"].as_dict(), |
|
|
) |
|
|
self.auth_lock = asyncio.Lock() |
|
|
|
|
|
@property |
|
|
def token(self) -> Optional[str]: |
|
|
if self._token is None: |
|
|
return None |
|
|
_, body = load_jwt(self._token) |
|
|
return None if body.exp < datetime.now(timezone.utc) else self._token |
|
|
|
|
|
async def login(self, endpoint: "BikaEndpoints"): |
|
|
login_data = BikaConstants.CONFIG["account"].get(BikaLogin) |
|
|
login_result: dict[str, Any] = await endpoint.request( |
|
|
"auth/sign-in", |
|
|
body=login_data.dict(), |
|
|
no_token=True, |
|
|
) |
|
|
assert login_result["code"] == 200, login_result["message"] |
|
|
if not ( |
|
|
isinstance(login_data := login_result.get("data"), dict) |
|
|
and "token" in login_data |
|
|
): |
|
|
raise ValueError("failed to read Bika account token.") |
|
|
self._token = login_data["token"] |
|
|
|
