Spaces:
Runtime error
Runtime error
File size: 2,041 Bytes
0a1b571 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
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"]
|