okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(python): separate capture, state, and value codecs448d7dc

main
branchesmainall refs →
5.3 KiB164 linesraw
1"""Bounded frame capture and reference construction. Page owns retention and lifetime."""
2
3from __future__ import annotations
4
5from dataclasses import dataclass
6from typing import Any
7
8from .errors import ProtocolError
9from .transport import CDPSession, JSON
10from .observation import Node, attributes_from_snapshot, ax_properties, walk_ax
11
12
13@dataclass
14class FrameSession:
15    cdp: CDPSession
16    target_id: str
17    root_frame: str = ""
18    generation: int = 0
19    ready: bool = False
20    error: str | None = None
21
22
23@dataclass(frozen=True)
24class Reference:
25    session_id: str
26    frame_id: str
27    generation: int
28    backend_id: int
29    role: str
30    name: str
31    attributes: dict[str, str]
32
33
34@dataclass(frozen=True)
35class CaptureLimits:
36    snapshot_id: str
37    max_nodes: int
38    max_text: int
39    max_depth: int
40
41
42@dataclass
43class CapturedFrames:
44    rows: list[Node]
45    references: dict[str, Reference]
46    warnings: list[JSON]
47
48
49async def capture_frames(
50    frames: list[dict[str, Any]],
51    sessions: dict[str, FrameSession],
52    limits: CaptureLimits,
53) -> CapturedFrames:
54    warnings: list[JSON] = []
55    dom_by_session = {}
56    refs: dict[str, Reference] = {}
57    rows: list[Node] = []
58    for frame in frames:
59        if len(rows) >= limits.max_nodes:
60            warnings.append(
61                {"kind": "frame_omitted_by_node_limit", "frame_id": frame["id"]}
62            )
63            continue
64        sid = frame["session_id"]
65        state = sessions[sid]
66        generation = state.generation
67        try:
68            if sid not in dom_by_session:
69                dom_by_session[sid] = attributes_from_snapshot(
70                    await state.cdp.send(
71                        "DOMSnapshot.captureSnapshot",
72                        {"computedStyles": []},
73                    )
74                )
75            raw = (
76                await state.cdp.send(
77                    "Accessibility.getFullAXTree",
78                    {
79                        "frameId": frame["id"],
80                        "depth": limits.max_depth,
81                    },
82                )
83            )["nodes"]
84        except ProtocolError as exc:
85            warnings.append(
86                {
87                    "kind": "frame_unavailable",
88                    "frame_id": frame["id"],
89                    "error": exc.to_dict(),
90                }
91            )
92            continue
93        attributes = dom_by_session[sid]
94        nodes = walk_ax(raw)
95        raw_ids = {n["nodeId"] for n in raw}
96        if any(c not in raw_ids for n in raw for c in n.get("childIds", [])):
97            warnings.append({"kind": "depth_or_tree_omission", "frame_id": frame["id"]})
98        ids: dict[str, str] = {}
99        for ax, parent, depth in nodes:
100            if len(rows) >= limits.max_nodes:
101                warnings.append({"kind": "node_limit", "frame_id": frame["id"]})
102                break
103            row_id = f"{limits.snapshot_id}:{len(rows) + 1}"
104            ids[ax["nodeId"]] = row_id
105            role = str(ax.get("role", {}).get("value", ""))
106            name = str(ax.get("name", {}).get("value", ""))
107            backend = ax.get("backendDOMNodeId")
108            attrs = attributes.get(backend, {})
109            actionable = bool(
110                backend
111                and backend in attributes
112                and role not in {"StaticText", "RootWebArea", "InlineTextBox"}
113            )
114            ref = row_id if actionable else None
115            row: Node = {
116                "id": row_id,
117                "ref": ref,
118                "role": "text" if role == "StaticText" else role,
119                "name": name[: limits.max_text],
120                "frame_id": frame["id"],
121                "parent": ids.get(parent),
122                "depth": depth,
123            }
124            if len(name) > limits.max_text:
125                row["name_truncated"] = True
126                warnings.append({"kind": "text_truncated", "node": row_id})
127            props = ax_properties(ax)
128            for key in (
129                "disabled",
130                "checked",
131                "selected",
132                "expanded",
133                "required",
134                "readonly",
135                "focused",
136                "level",
137            ):
138                if key in props:
139                    row[key] = props[key]
140            if attrs:
141                row["attributes"] = dict(attrs)
142                if "href" in attrs:
143                    row["href"] = attrs["href"]
144            if "value" in ax:
145                value = ax["value"].get("value")
146                row["value"] = (
147                    "[redacted]" if attrs.get("type") == "password" else value
148                )
149                if (
150                    isinstance(row["value"], str)
151                    and len(row["value"]) > limits.max_text
152                ):
153                    row["value"] = row["value"][: limits.max_text]
154                    warnings.append({"kind": "value_truncated", "node": row_id})
155            if ref and backend is not None:
156                refs[ref] = Reference(
157                    sid, frame["id"], generation, backend, role, name, dict(attrs)
158                )
159            rows.append(row)
160        if state.generation != generation:
161            warnings.append(
162                {"kind": "document_changed_during_capture", "frame_id": frame["id"]}
163            )
164    return CapturedFrames(rows, refs, warnings)