hibiapi/api/bika/net.py

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"]