pigway-frame-flow/pigway_frame_flow/integrations/kokonna_official/rpc.py

"""Strict account RPC allowlist. Never falls back to a device key or retries."""

import time

import requests

from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.integrations.kokonna_official.capabilities import READ_OPERATIONS


class OfficialRPC:
    def __init__(self, token: str = ""):
        self._token = token

    @property
    def authorized(self) -> bool:
        return bool(self._token)

    def _call(self, method: str, params: dict, *, allow_write=False) -> dict:
        if method == "appFrame.setFrameCover" and not allow_write:
            raise FlowError(
                E.UNSUPPORTED_CAPABILITY, "Use the explicit widget operation", 422
            )
        if method not in READ_OPERATIONS and method not in (
            "app.login",
            "appFrame.setFrameCover",
        ):
            raise FlowError(
                E.UNSUPPORTED_CAPABILITY, "Official operation is not enabled", 422
            )
        if method != "app.login" and not self._token:
            raise FlowError(
                E.CONFIG_REQUIRED, "Official account authorization required", 503
            )
        try:
            response = requests.post(
                "https://robot.galaxyguide.cn/api/rpc",
                headers={"Authorization": "Bearer " + self._token}
                if method != "app.login"
                else {},
                json={
                    "jsonrpc": "2.0",
                    "method": method,
                    "params": params,
                    "id": int(time.time() * 1000),
                },
                timeout=15,
                allow_redirects=False,
            )
            if response.status_code in (401, 403):
                self.disconnect()
                raise FlowError(
                    E.OFFICIAL_AUTH_REQUIRED,
                    "Official account authentication required",
                    401,
                )
            payload = response.json()
            error = payload.get("error")
            if isinstance(error, dict) and error.get("message") == "请重新登陆":
                self.disconnect()
                raise FlowError(
                    E.OFFICIAL_AUTH_REQUIRED, "Official account session expired", 401
                )
            response.raise_for_status()
            if error or not isinstance(payload.get("result"), dict):
                raise ValueError()
            return payload["result"]
        except FlowError:
            raise
        except (requests.RequestException, ValueError, AttributeError):
            if method == "appFrame.setFrameCover":
                raise FlowError(
                    E.WRITE_RESULT_UNKNOWN, "Official widget write result unknown", 409
                ) from None
            raise FlowError(
                E.CLOUD_UNAVAILABLE, "Official service query failed", 502
            ) from None

    def login(self, username: str, password: str) -> None:
        if not username or not password:
            raise FlowError(E.INVALID_REQUEST, "Official account credentials required")
        result = self._call("app.login", {"username": username, "password": password})
        token = result.get("token")
        if not isinstance(token, str) or not token:
            raise FlowError(E.UNAUTHORIZED, "Official account login failed", 401)
        self._token = token

    def disconnect(self) -> None:
        self._token = ""

    def devices(self) -> dict:
        return self._call("app.getRobots", {})

    def device(self, robot_id: int) -> dict:
        return self._call("appFrame.getFrame", {"id": robot_id})

    def widgets(self, robot_id: int) -> dict:
        return self._call("appFrame.getFrameCovers", {"robotId": robot_id})

    def set_widget(self, robot_id, cover_id, args):
        return self._call(
            "appFrame.setFrameCover",
            {"robotId": robot_id, "coverId": cover_id, "args": args},
            allow_write=True,
        )