skill-template/tools/ui_inspector/models/state.py

"""UIState model + stable business fingerprint."""

from __future__ import annotations

import hashlib
import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional

SCHEMA_STATE = "openclaw.ui_inspector.state.v1"

# Keys ignored when hashing business identity.
_FINGERPRINT_EXCLUDE_TOP = {
    "captured_at",
    "id",
    "previous_state_id",
    "tree_path",
    "index_path",
    "locatable_path",
    "screenshot_path",
    "duplicate_of",
    "observation_count",
    "change_reason",
}

_SESSION_KEYS = {
    "pid",
    "hwnd",
    "process_id",
    "thread_id",
    "elapsed_ms",
    "dump_ms",
    "warmup_ms",
    "timings",
    "screenshot",
    "screen_path",
}


def _now_iso() -> str:
    return datetime.now().astimezone().isoformat()


def _new_id(prefix: str = "st") -> str:
    return f"{prefix}_{uuid.uuid4().hex[:12]}"


def _strip_session(obj: Any) -> Any:
    if isinstance(obj, dict):
        out: dict[str, Any] = {}
        for k, v in obj.items():
            lk = str(k).lower()
            if lk in _SESSION_KEYS or lk.endswith("_ms") or lk.endswith("_path"):
                continue
            if lk in {"left", "top", "right", "bottom", "x", "y", "width", "height"}:
                # Absolute coordinates are not primary identity.
                continue
            if lk == "rect" or lk == "bounds" or lk == "bounding_rectangle":
                continue
            out[str(k)] = _strip_session(v)
        return out
    if isinstance(obj, list):
        return [_strip_session(x) for x in obj]
    return obj


def _locatable_signature(rows: list[dict[str, Any]], *, limit: int = 120) -> list[dict[str, Any]]:
    """Compact visible-business signature for fingerprinting."""
    sig: list[dict[str, Any]] = []
    for r in rows[:limit]:
        if not isinstance(r, dict):
            continue
        entry = {
            "control_type": r.get("control_type") or r.get("role") or "",
            "name": r.get("name") or "",
            "value": r.get("value") or r.get("text") or "",
            "automation_id": r.get("automation_id") or r.get("test_id") or "",
            "class_name": r.get("class_name") or "",
        }
        # Drop empty noise rows
        if not any(str(v).strip() for v in entry.values()):
            continue
        sig.append(entry)
    # Stable order
    sig.sort(
        key=lambda e: (
            str(e.get("control_type") or ""),
            str(e.get("automation_id") or ""),
            str(e.get("name") or ""),
            str(e.get("value") or ""),
        )
    )
    return sig


def compute_state_fingerprint(
    *,
    surface: str,
    driver: str = "",
    url: str = "",
    page_title: str = "",
    window: Optional[dict[str, Any]] = None,
    locatable: Optional[list[dict[str, Any]]] = None,
    app: Optional[dict[str, Any]] = None,
    extra_business: Optional[dict[str, Any]] = None,
) -> str:
    """
    SHA-256 of stable business identity.

    Excludes timestamps, pid/hwnd, screenshot paths, absolute coords, timings.
    """
    win = dict(window or {})
    app_d = dict(app or {})
    payload = {
        "surface": surface or "",
        "driver": driver or "",
        "url": url or win.get("url") or app_d.get("url") or "",
        "page_title": page_title or "",
        "window": {
            "title": win.get("title") or app_d.get("title") or "",
            "class_name": win.get("class_name") or app_d.get("class_name") or "",
            "process_name": win.get("process_name") or app_d.get("process_name") or "",
        },
        "locatable": _locatable_signature(list(locatable or [])),
        "extra": _strip_session(extra_business or {}),
    }
    raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


@dataclass
class UIState:
    schema: str = SCHEMA_STATE
    id: str = field(default_factory=lambda: _new_id("st"))
    surface: str = "windows"
    driver: str = ""
    captured_at: str = field(default_factory=_now_iso)
    fingerprint: str = ""
    previous_state_id: Optional[str] = None
    change_reason: str = ""
    app: dict[str, Any] = field(default_factory=dict)
    window: dict[str, Any] = field(default_factory=dict)
    url: str = ""
    page_title: str = ""
    tree_path: str = ""
    index_path: str = ""
    locatable_path: str = ""
    screenshot_path: str = ""
    stats: dict[str, Any] = field(default_factory=dict)
    selection: dict[str, Any] = field(default_factory=dict)
    duplicate_of: Optional[str] = None
    observation_count: int = 1
    may_contain_sensitive_data: bool = False
    # In-memory locatable used for fingerprint/diff (not always persisted as field).
    locatable: list[dict[str, Any]] = field(default_factory=list, repr=False)

    def ensure_fingerprint(self) -> str:
        if self.fingerprint:
            return self.fingerprint
        self.fingerprint = compute_state_fingerprint(
            surface=self.surface,
            driver=self.driver,
            url=self.url,
            page_title=self.page_title,
            window=self.window,
            locatable=self.locatable,
            app=self.app,
        )
        return self.fingerprint

    def bump_observation(self) -> None:
        self.observation_count = int(self.observation_count or 0) + 1
        self.captured_at = _now_iso()