Files
skill-template/tools/ui_inspector/artifacts.py
chendelian 46bc573777
All checks were successful
技能自动化发布 / release (push) Successful in 10s
Release v1.0.66: table default_sort_json for data management
2026-08-02 12:17:55 +08:00

695 lines
20 KiB
Python

"""Flow-oriented capture artifacts: one flow document, many steps."""
from __future__ import annotations
import json
import re
from datetime import datetime
from pathlib import Path
from typing import Any, Optional
try:
from . import __version__
except ImportError:
try:
import importlib.util
from pathlib import Path
_spec = importlib.util.spec_from_file_location(
"ui_inspector_pkg", Path(__file__).resolve().parent / "__init__.py"
)
_mod = importlib.util.module_from_spec(_spec) # type: ignore[arg-type]
assert _spec and _spec.loader
_spec.loader.exec_module(_mod)
__version__ = getattr(_mod, "__version__", "0.8.0")
except Exception:
__version__ = "0.8.0"
def _local_ts() -> str:
now = datetime.now().astimezone()
return now.strftime("%Y%m%dT%H%M%S%z")
def safe_slug(text: str, *, fallback: str = "session") -> str:
text = (text or "").strip()
if not text:
return fallback
text = re.sub(r"\s+", "-", text)
text = re.sub(r"[^\w\u4e00-\u9fff\-]+", "", text, flags=re.UNICODE)
text = text.strip("-_") or fallback
return text[:80]
def _clean_ui_text(text: str) -> str:
"""Strip WeCom-style rich markup and collapse whitespace."""
s = (text or "").strip()
if not s:
return ""
s = re.sub(r"<[^>]+>", "", s)
s = re.sub(r"\s+", " ", s).strip()
return s
# Name/AutomationId keys that usually mark "what page am I on"
_PAGE_HINT_KEYS: dict[str, int] = {
"panel_title_label": 100,
"title": 95,
"ww_btn_text": 70,
"main_chattext": 45,
}
_LAYOUT_NOISE = {
"bg",
"navigation_bar",
"outer_layout",
"content_container",
"title_layout",
"closebtn",
"VerticalLayoutUI",
"HorizontalLayoutUI",
"wbtn_container",
"search_layout",
"search_layout_border",
"inner_search_layout",
"ControlUI",
"LabelUI",
"item",
"externals",
"status_layout",
"search_button_region",
"search_icon",
"ww_btn_icon",
"wbadge_label",
"lasttime",
"groupbuddydescription",
"bottom_text",
"chattext",
"pane",
"Pane",
}
_CHROME_TEXT = {
"关闭",
"最小化",
"最大化",
"还原",
"取消",
"确定",
"关闭窗口",
}
# Left-rail / chrome values that appear on every WeCom page — not page identity
_GLOBAL_NAV_TEXT = {
"消息",
"邮件",
"文档",
"日程",
"待办",
"会议",
"智能文档",
"通讯录",
"微盘",
"高级功能",
"工作台",
"收藏",
"智能总结",
"回到最新位置",
}
def _row_key(r: dict[str, Any]) -> str:
return _clean_ui_text(str(r.get("name") or r.get("automation_id") or ""))
def _looks_like_search_dialog(
window_title: str, window_class: str, rows: list[dict[str, Any]]
) -> bool:
t = window_title or ""
c = window_class or ""
if "Search" in c or "搜索" in t or "添加客户" in t or "添加联系人" in t:
return True
for r in rows:
if _row_key(r) == "title":
v = _clean_ui_text(str(r.get("value") or ""))
if "添加" in v or "搜索" in v:
return True
return False
def _short_entity(text: str, *, limit: int = 16) -> str:
"""Clean contact/chat title: strip markup leftovers and @微信 suffix."""
s = _clean_ui_text(text)
if not s:
return ""
s = re.sub(r"@微信\s*$", "", s)
s = re.sub(r"@WeCom\s*$", "", s, flags=re.I)
s = s.strip(" -_|")
if len(s) > limit:
s = s[:limit].rstrip("-_")
return s
def _raw_value(r: dict[str, Any]) -> str:
return str(r.get("value") or "")
def _is_selected_buddy(raw_value: str) -> bool:
# WeCom marks the active conversation with semi-transparent white color markup.
return "#99ffffff" in (raw_value or "").lower()
def derive_step_label(
*,
window_title: str = "",
window_class: str = "",
locatable: Optional[list[dict[str, Any]]] = None,
max_hints: int = 2,
) -> str:
"""
Build a human-readable step label from window + page-specific UIA hints.
Prefers: search box text, page title, selected chat, compose draft, action buttons.
Avoids bare「会话列表」when a concrete chat/search/input signal exists.
"""
rows = list(locatable or [])
base = _clean_ui_text(window_title)
if not base:
for r in rows:
if (r.get("control_type") or "") == "WindowControl":
base = _clean_ui_text(str(r.get("name") or ""))
if base:
break
if not base and window_class:
base = window_class
has_panel_title = any(
_row_key(r) == "panel_title_label" and _clean_ui_text(str(r.get("value") or ""))
for r in rows
)
has_groupbuddy = any(_row_key(r) == "groupbuddyname" for r in rows)
has_search_result = any(_row_key(r) == "search_result" for r in rows)
has_searchedit = any(_row_key(r) in {"searchedit", "search_edit"} for r in rows)
search_dialog = _looks_like_search_dialog(window_title, window_class, rows)
scored: list[tuple[int, str]] = []
searchedit_value = ""
selected_buddy = ""
first_buddy = ""
for r in rows:
key = _row_key(r)
if not key or key in _LAYOUT_NOISE:
continue
raw = _raw_value(r)
value = _clean_ui_text(raw)
name_disp = _clean_ui_text(str(r.get("name") or ""))
ctype = str(r.get("control_type") or "")
if key in {"searchedit", "search_edit"} and value:
searchedit_value = _short_entity(value, limit=20)
# Main-window search AND add-friend dialog both matter.
scored.append((98, f"搜索{searchedit_value}"))
continue
if key == "send_rich_edit" and value:
draft = _short_entity(value, limit=12)
if draft:
scored.append((86, f"输入{draft}"))
continue
if key == "groupbuddyname" and value:
entity = _short_entity(value, limit=14)
if not entity:
continue
if _is_selected_buddy(raw) and not selected_buddy:
selected_buddy = entity
scored.append((92, entity))
elif not first_buddy and not r.get("is_offscreen"):
first_buddy = entity
continue
pri = _PAGE_HINT_KEYS.get(key)
if pri is not None and value:
if value in _CHROME_TEXT or value in _GLOBAL_NAV_TEXT:
continue
if key == "title" and value == base:
continue
if key == "main_chattext" and (has_panel_title or has_groupbuddy):
continue
entity = _short_entity(value, limit=16)
if not entity:
continue
scored.append((pri, entity))
continue
if ctype in {
"ButtonControl",
"EditControl",
"TabItemControl",
"TextControl",
"HyperlinkControl",
"MenuItemControl",
"CheckBoxControl",
"RadioButtonControl",
"ComboBoxControl",
"ListItemControl",
}:
label = _short_entity(value or name_disp, limit=16)
if (
not label
or label in _CHROME_TEXT
or label in _LAYOUT_NOISE
or label in _GLOBAL_NAV_TEXT
):
continue
scored.append((40, label))
# Conversation context when no selected buddy / search text yet
if (
has_groupbuddy
and not selected_buddy
and first_buddy
and not has_panel_title
and not searchedit_value
):
scored.append((78, first_buddy))
strong = any(p >= 90 for p, _ in scored)
if search_dialog and has_searchedit and not searchedit_value and not strong:
scored.append((75, "搜索框"))
elif search_dialog and has_search_result and searchedit_value and not strong:
scored.append((50, "搜索结果"))
elif (
has_groupbuddy
and not has_panel_title
and not strong
and not first_buddy
and not selected_buddy
):
scored.append((60, "会话列表"))
scored.sort(key=lambda x: (-x[0], x[1]))
hints: list[str] = []
seen: set[str] = set()
for _pri, text in scored:
if not text or text == base or text in seen:
continue
if text in _CHROME_TEXT or text in _GLOBAL_NAV_TEXT:
continue
compact = (
text[2:] if text.startswith(("搜索", "输入")) and len(text) > 2 else text
)
if compact and any(
compact == s or s.endswith(compact) or compact in s for s in seen
):
continue
seen.add(text)
if compact:
seen.add(compact)
hints.append(text)
if len(hints) >= max_hints:
break
parts = [p for p in [base, *hints] if p]
if not parts:
parts = ["step"]
return "_".join(parts)
def _label_slug_key(label: str) -> str:
return safe_slug(label, fallback="step")
def _unique_step_dirname(steps_root: Path, idx: int, label: str) -> str:
"""NN_label, with -2/-3 suffix if the folder already exists."""
slug = _label_slug_key(label)
base = f"{idx:02d}_{slug}"
if not (steps_root / base).exists():
return base
n = 2
while (steps_root / f"{base}-{n}").exists():
n += 1
return f"{base}-{n}"
def _existing_label_slugs(flow: dict[str, Any]) -> set[str]:
out: set[str] = set()
for s in flow.get("steps") or []:
lab = s.get("label") or ""
name = str(s.get("name") or "")
if lab:
out.add(_label_slug_key(str(lab)))
m = re.match(r"^\d{2}_(.+?)(?:-\d+)?$", name)
if m:
out.add(m.group(1))
return out
def resolve_step_label(
flow: dict[str, Any],
*,
window_title: str = "",
window_class: str = "",
locatable: Optional[list[dict[str, Any]]] = None,
) -> str:
"""Derive label; if it collides with earlier steps, add more hints."""
used = _existing_label_slugs(flow)
label = derive_step_label(
window_title=window_title,
window_class=window_class,
locatable=locatable,
max_hints=2,
)
if _label_slug_key(label) not in used:
return label
label3 = derive_step_label(
window_title=window_title,
window_class=window_class,
locatable=locatable,
max_hints=3,
)
if _label_slug_key(label3) not in used:
return label3
# Last resort: keep richest label; folder uniquifier adds -2
return label3
def default_captures_root(tool_dir: Optional[Path] = None) -> Path:
base = tool_dir or Path(__file__).resolve().parent
return base / "_captures"
def write_json(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(data, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as fh:
for row in rows:
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
def read_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def build_meta(
*,
window: dict[str, Any],
scene: str,
skill_root: Optional[str] = None,
extra: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
now = datetime.now().astimezone()
meta = {
"schema": "openclaw.ui_inspector.capture.v1",
"surface": "windows",
"captured_at": now.isoformat(),
"scene": scene,
"skill_root": skill_root,
"app": {
"title": window.get("title"),
"class_name": window.get("class_name"),
"process_name": window.get("process_name"),
"pid": window.get("pid"),
"hwnd": window.get("hwnd"),
},
"environment": {
"os": "windows",
"tz": now.tzname(),
},
"tool": {
"name": "openclaw-ui-inspector",
"version": __version__,
"driver": "uiautomation",
},
}
if extra:
meta["extra"] = extra
return meta
def noteworthy_from_index(index: list[dict[str, Any]], *, limit: int = 80) -> list[dict[str, Any]]:
rows = [
r
for r in index
if (r.get("name") or r.get("value") or r.get("automation_id"))
]
rows.sort(
key=lambda r: (
0 if r.get("value") else 1,
0 if r.get("name") else 1,
int(r.get("depth") or 0) if isinstance(r.get("depth"), int) and r.get("depth") >= 0 else 99,
)
)
return rows[:limit]
def capture_window_png(hwnd: int, dest: Path) -> bool:
try:
from PIL import ImageGrab
except ImportError:
return False
try:
import ctypes
from ctypes import wintypes
user32 = ctypes.windll.user32
rect = wintypes.RECT()
if not user32.GetWindowRect(int(hwnd), ctypes.byref(rect)):
return False
bbox = (int(rect.left), int(rect.top), int(rect.right), int(rect.bottom))
if bbox[2] <= bbox[0] or bbox[3] <= bbox[1]:
return False
img = ImageGrab.grab(bbox=bbox, all_screens=True)
dest.parent.mkdir(parents=True, exist_ok=True)
img.save(str(dest))
return True
except Exception:
try:
img = ImageGrab.grab(all_screens=True)
dest.parent.mkdir(parents=True, exist_ok=True)
img.save(str(dest))
return True
except Exception:
return False
# --- Flow document API -------------------------------------------------
def create_flow(
captures_root: Path,
*,
flow_name: str = "desktop-flow",
) -> Path:
"""开始流程:新建一份流程文档目录。"""
captures_root.mkdir(parents=True, exist_ok=True)
slug = safe_slug(flow_name, fallback="flow")
flow_dir = captures_root / f"{_local_ts()}_{slug}"
flow_dir.mkdir(parents=True, exist_ok=False)
(flow_dir / "steps").mkdir(parents=True, exist_ok=True)
now = datetime.now().astimezone().isoformat()
flow = {
"schema": "openclaw.ui_inspector.flow.v1",
"flow_name": flow_name,
"status": "open",
"created_at": now,
"updated_at": now,
"ended_at": None,
"step_count": 0,
"steps": [],
"tool": {
"name": "openclaw-ui-inspector",
"version": __version__,
},
}
write_json(flow_dir / "flow.json", flow)
write_flow_summary(flow_dir, flow)
return flow_dir
def load_flow(flow_dir: Path) -> dict[str, Any]:
return read_json(flow_dir / "flow.json")
def save_flow(flow_dir: Path, flow: dict[str, Any]) -> None:
flow["updated_at"] = datetime.now().astimezone().isoformat()
write_json(flow_dir / "flow.json", flow)
write_flow_summary(flow_dir, flow)
def next_step_index(flow: dict[str, Any]) -> int:
return int(flow.get("step_count") or 0) + 1
def add_step_dir(
flow_dir: Path,
flow: dict[str, Any],
*,
window_title: str = "",
window_class: str = "",
locatable: Optional[list[dict[str, Any]]] = None,
label: Optional[str] = None,
) -> tuple[Path, int, str]:
"""Allocate steps/NN_<page-fingerprint>/ under an open flow."""
if flow.get("status") != "open":
raise RuntimeError("当前流程已结束,请先「开始流程」。")
idx = next_step_index(flow)
if not label:
label = resolve_step_label(
flow,
window_title=window_title,
window_class=window_class,
locatable=locatable,
)
steps_root = flow_dir / "steps"
steps_root.mkdir(parents=True, exist_ok=True)
step_name = _unique_step_dirname(steps_root, idx, label)
step_dir = steps_root / step_name
step_dir.mkdir(parents=True, exist_ok=False)
return step_dir, idx, step_name
def register_step(
flow_dir: Path,
flow: dict[str, Any],
*,
step_index: int,
step_name: str,
window: dict[str, Any],
stats: dict[str, Any],
label: Optional[str] = None,
) -> dict[str, Any]:
entry = {
"index": step_index,
"name": step_name,
"label": label or step_name,
"captured_at": datetime.now().astimezone().isoformat(),
"window_title": window.get("title"),
"window_class": window.get("class_name"),
"pid": window.get("pid"),
"hwnd": window.get("hwnd"),
"findall": stats.get("findall_count"),
"locatable": stats.get("locatable_count"),
"kept": stats.get("nodes_kept"),
"uia_sparse": stats.get("uia_sparse"),
"path": f"steps/{step_name}",
}
steps = list(flow.get("steps") or [])
steps.append(entry)
flow["steps"] = steps
flow["step_count"] = len(steps)
save_flow(flow_dir, flow)
return entry
def end_flow(flow_dir: Path, flow: dict[str, Any]) -> dict[str, Any]:
if flow.get("status") == "ended":
return flow
flow["status"] = "ended"
flow["ended_at"] = datetime.now().astimezone().isoformat()
save_flow(flow_dir, flow)
return flow
def write_flow_summary(flow_dir: Path, flow: dict[str, Any]) -> None:
lines = [
f"# 流程:{flow.get('flow_name')}",
"",
f"- status: `{flow.get('status')}`",
f"- created_at: `{flow.get('created_at')}`",
f"- updated_at: `{flow.get('updated_at')}`",
f"- ended_at: `{flow.get('ended_at')}`",
f"- step_count: `{flow.get('step_count')}`",
"",
"## 步骤",
"",
]
steps = flow.get("steps") or []
if not steps:
lines.append("(尚无拾取步骤。点「拾取当前页」追加。)")
lines.append("")
else:
for s in steps:
lines.append(f"### {s.get('index')}. `{s.get('name')}`")
lines.append("")
if s.get("label"):
lines.append(f"- label: `{s.get('label')}`")
lines.append(
f"- window: `{s.get('window_title')}` / `{s.get('window_class')}`"
)
lines.append(
f"- stats: findall=`{s.get('findall')}` locatable=`{s.get('locatable')}` "
f"kept=`{s.get('kept')}` sparse=`{s.get('uia_sparse')}`"
)
lines.append(f"- path: `{s.get('path')}/`")
lines.append("")
lines.extend(
[
"## 说明",
"",
"- 一个流程对应本目录一份文档(`flow.json` + `SUMMARY.md` + `steps/`)。",
"- 每拾取一页,在 `steps/` 下新增一步;给 AI 时优先读本 `SUMMARY.md` 与各步 `locatable.json`。",
"- Name ≠ 界面中文时请看 Value。",
"",
]
)
(flow_dir / "SUMMARY.md").write_text("\n".join(lines), encoding="utf-8")
def write_step_summary(
step_dir: Path,
*,
meta: dict[str, Any],
stats: Optional[dict[str, Any]] = None,
noteworthy: Optional[list[dict[str, Any]]] = None,
) -> None:
app = meta.get("app") or {}
lines = [
"# Step Summary",
"",
f"- captured_at: `{meta.get('captured_at')}`",
f"- window: `{app.get('title')}` / `{app.get('class_name')}` pid=`{app.get('pid')}`",
"",
]
if stats:
lines.extend(
[
"## Stats",
"",
f"- kept: `{stats.get('nodes_kept')}`",
f"- findall: `{stats.get('findall_count')}`",
f"- locatable: `{stats.get('locatable_count')}`",
f"- uia_sparse: `{stats.get('uia_sparse')}`",
"",
]
)
if noteworthy:
lines.extend(["## Notable", ""])
for item in noteworthy[:40]:
lines.append(
f"- `{item.get('control_type')}` Name=`{item.get('name')}` "
f"Value=`{item.get('value')}`"
)
lines.append("")
lines.extend(
[
"## Files",
"",
"- `meta.json` / `tree.json` / `index.jsonl` / `locatable.json` / `screen.png`",
"",
]
)
(step_dir / "SUMMARY.md").write_text("\n".join(lines), encoding="utf-8")
# Back-compat aliases used by older smoke scripts
def create_session_dir(captures_root: Path, *, scene: str = "desktop") -> Path:
return create_flow(captures_root, flow_name=scene)