pigway-frame-flow/pigway_frame_flow/drivers/kokonna/widgets.py

"""Read-only official widget RPC; separate from the device OpenAPI credential."""

import json
import time
from pathlib import Path

import requests

from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import Widget, WidgetState


class WidgetReader:
    def __init__(self, token="", robot_id=None, mock=False):
        self._token = token
        self._robot_id = robot_id
        self.mock = mock
        self._cached = None
        self._cached_at = 0.0

    def read(self) -> WidgetState:
        if self._cached is not None and time.monotonic() - self._cached_at < 60:
            return self._cached
        snapshot = json.loads(
            Path(__file__).with_name("widget_catalog.json").read_text()
        )
        catalog = [
            Widget(**{k: row[k] for k in ("id", "name", "tags", "arg_keys")})
            for row in snapshot["catalog"]
        ]
        if self.mock:
            return WidgetState(
                source="mock",
                configured=True,
                catalog=catalog,
                current=catalog[0],
                current_verified=True,
            )
        if not self._token or self._robot_id is None:
            return WidgetState(
                source="snapshot",
                configured=False,
                catalog=catalog,
                observed_at=snapshot["observed_at"],
                status="AUTH_REQUIRED",
            )
        try:
            response = requests.post(
                "https://robot.galaxyguide.cn/api/rpc",
                headers={"Authorization": f"Bearer {self._token}"},
                json={
                    "jsonrpc": "2.0",
                    "method": "appFrame.getFrameCovers",
                    "params": {"robotId": self._robot_id},
                    "id": int(time.time() * 1000),
                },
                timeout=15,
                allow_redirects=False,
            )
            if response.status_code in (401, 403):
                raise FlowError(E.UNAUTHORIZED, "Official widget session expired", 401)
            response.raise_for_status()
            payload = response.json()
            if payload.get("error"):
                raise FlowError(
                    E.CLOUD_UNAVAILABLE, "Official widget query rejected", 502
                )
            data = payload["result"]
            rows = data["rows"]
            catalog = [
                Widget(
                    id=row["id"],
                    name=row["name"],
                    tags=row.get("tags") or [],
                    arg_keys=list((row.get("args") or {}).keys())
                    if isinstance(row.get("args"), dict)
                    else [],
                )
                for row in rows
            ]
            selected = data.get("frameCoverRobot") or {}
            cover = selected.get("frameCover") or {}
            current_id = cover.get("id", selected.get("coverId"))
            current = next((row for row in catalog if row.id == current_id), None)
            robot = data.get("robot") or {}
            no_widget = (
                "frameCoverRobot" in data
                and data["frameCoverRobot"] is None
                and "frameCoverRobotId" in robot
                and robot["frameCoverRobotId"] is None
            )
            result = WidgetState(
                source="live",
                configured=True,
                catalog=catalog,
                current=current,
                current_verified=current is not None or no_widget,
                status="OK"
                if current
                else "NO_WIDGET"
                if no_widget
                else "CURRENT_UNVERIFIED",
            )
        except FlowError:
            raise
        except (
            requests.RequestException,
            ValueError,
            KeyError,
            TypeError,
            AttributeError,
        ):
            raise FlowError(
                E.CLOUD_UNAVAILABLE, "Official widget query unavailable", 502
            ) from None
        self._cached, self._cached_at = result, time.monotonic()
        return result