diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ea2708..5ccb0aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ - 小节正文写用户能看懂的变化（新能力、修复、注意事项），避免堆砌内部实现细节 - 打 tag 前**必须**为即将发布的版本新增一节；找不到匹配小节时，本次发布不会更新 changelog 字段 +## 1.0.66 + +- 数据管理支持表级默认排序：`_jiangchang_tables.default_sort_json`（多字段 JSON）；宿主优先采用，未配置则仍按 updated_at → created_at → 主键 +- 更新 SCHEMA.md 与 `upsert_table_metadata`；旧库幂等补列 + ## 1.0.65 - `replaces` create/update 可复用表列 lookup：字段名对齐列名，或使用 `bind.columnMapping` / 字段 `column` diff --git a/SKILL.md b/SKILL.md index 87b02a6..99dbda4 100644 --- a/SKILL.md +++ b/SKILL.md @@ -1,7 +1,7 @@ --- name: 技能开发模板（通用业务版） description: "OpenClaw 通用业务技能开发模板，供复制后定制新业务 skill。定制步骤见 development/DEVELOPMENT.md。" -version: 1.0.65 +version: 1.0.66 author: 深圳匠厂科技有限公司 metadata: openclaw: diff --git a/references/SCHEMA.md b/references/SCHEMA.md index 7054f39..083a877 100644 --- a/references/SCHEMA.md +++ b/references/SCHEMA.md @@ -133,6 +133,38 @@ metadata: | `sort_order` | 否 | 兼容保留字段，不用于替代物理字段顺序 | | `visible` | 是 | `1` 在宿主数据管理中展示 / `0` 隐藏 | | `readonly` | 是 | `1` 整张表只读；`0` 允许宿主根据字段权限执行写操作 | +| `default_sort_json` | 否 | 数据管理**默认行排序**（见下节）；空则宿主用内置默认 | + +#### 表级默认排序 `default_sort_json`（宿主支持） + +技能可声明该表在数据管理中的默认排序；宿主优先采用，未声明或声明无效时回退： + +`updated_at DESC` → `created_at DESC` → 主键 DESC。 + +JSON 数组（可多字段，按顺序）： + +```json +[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}] +``` + +规则： + +- `column` 必须是该表真实列名；敏感列会被宿主忽略 +- `direction` 仅 `asc` / `desc`（缺省按 `desc`） +- 用户在 UI 上手动改排序后，以用户选择为准；声明只影响未选择时的默认 +- 通过 `upsert_table_metadata(..., default_sort_json=...)` 写入；旧库由 `init_display_metadata_tables` 幂等 `ALTER` 补列 + +示例（消息时间线）： + +```python +upsert_table_metadata( + cursor, + table_name="messages", + display_name="消息", + readonly=0, + default_sort_json='[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}]', +) +``` 禁止： diff --git a/scripts/db/display_metadata.py b/scripts/db/display_metadata.py index 5326431..674a334 100644 --- a/scripts/db/display_metadata.py +++ b/scripts/db/display_metadata.py @@ -24,7 +24,8 @@ CREATE TABLE IF NOT EXISTS _jiangchang_tables ( description TEXT, sort_order INTEGER, visible INTEGER NOT NULL DEFAULT 1, - readonly INTEGER NOT NULL DEFAULT 0 + readonly INTEGER NOT NULL DEFAULT 0, + default_sort_json TEXT ) """ @@ -82,23 +83,50 @@ TASK_LOGS_COLUMN_METADATA: tuple[dict[str, object], ...] = ( ) +def _ensure_jiangchang_tables_columns(cursor) -> None: + """旧库补列：CREATE IF NOT EXISTS 不会给已有表加新字段。""" + cursor.execute("PRAGMA table_info('_jiangchang_tables')") + cols = {str(row[1]) for row in cursor.fetchall()} + if "default_sort_json" not in cols: + cursor.execute("ALTER TABLE _jiangchang_tables ADD COLUMN default_sort_json TEXT") + + def init_display_metadata_tables(cursor) -> None: cursor.execute(_JIANGCHANG_TABLES_DDL) cursor.execute(_JIANGCHANG_COLUMNS_DDL) + _ensure_jiangchang_tables_columns(cursor) -def upsert_table_metadata(cursor, *, table_name: str, display_name: str, description: str | None = None, visible: int = 1, readonly: int = 0) -> None: +def upsert_table_metadata( + cursor, + *, + table_name: str, + display_name: str, + description: str | None = None, + visible: int = 1, + readonly: int = 0, + default_sort_json: str | None = None, +) -> None: + """写入表级元数据。 + + default_sort_json: 宿主数据管理默认排序，JSON 数组，例如 + '[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}]' + 为空则宿主回退到 updated_at → created_at → 主键。 + """ cursor.execute( """ - INSERT INTO _jiangchang_tables (table_name, display_name, description, visible, readonly) - VALUES (?, ?, ?, ?, ?) + INSERT INTO _jiangchang_tables ( + table_name, display_name, description, visible, readonly, default_sort_json + ) + VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(table_name) DO UPDATE SET display_name = excluded.display_name, description = excluded.description, visible = excluded.visible, - readonly = excluded.readonly + readonly = excluded.readonly, + default_sort_json = excluded.default_sort_json """, - (table_name, display_name, description, visible, readonly), + (table_name, display_name, description, visible, readonly, default_sort_json), ) diff --git a/scripts/util/constants.py b/scripts/util/constants.py index e2efce1..611bf5c 100644 --- a/scripts/util/constants.py +++ b/scripts/util/constants.py @@ -1,6 +1,6 @@ """技能标识、版本与平台公共库约束（复制后请修改 slug/version/logger）。""" SKILL_SLUG = "your-skill-slug" -SKILL_VERSION = "1.0.61" +SKILL_VERSION = "1.0.66" LOG_LOGGER_NAME = "openclaw.skill.your_skill_slug" PLATFORM_KIT_MIN_VERSION = "1.2.2" diff --git a/tools/README.md b/tools/README.md index 7bd6a66..5a807a0 100644 --- a/tools/README.md +++ b/tools/README.md @@ -1,5 +1,14 @@ # skill-template 工具 +## 桌面 UIA 拾取（开发期） + +```powershell +cd D:\OpenClaw\client-commons\skill-template +python tools/ui_inspector/run.py --scene demo +``` + +说明见 [`ui_inspector/README.md`](ui_inspector/README.md)。企微暖树依赖本机 `jiangchang-platform-kit>=1.2.5`（与技能同调 `ensure_ui_ready`）。 + ## 业务仓也可以直接 clone（来源 A） 若项目经理已在**公司自建 Gitea**（[https://git.jc2009.com/](https://git.jc2009.com/)，基于开源 Gitea 自建，**不是** Gitea 官网）开好业务技能仓库并授权，技术人员应直接： diff --git a/tools/ui_inspector/.gitignore b/tools/ui_inspector/.gitignore new file mode 100644 index 0000000..0e4970c --- /dev/null +++ b/tools/ui_inspector/.gitignore @@ -0,0 +1,5 @@ +# 仅本工具目录使用；不修改 skill-template 根 .gitignore +_captures/ +__pycache__/ +*.pyc +.pytest_cache/ diff --git a/tools/ui_inspector/README.md b/tools/ui_inspector/README.md new file mode 100644 index 0000000..2955324 --- /dev/null +++ b/tools/ui_inspector/README.md @@ -0,0 +1,49 @@ +# OpenClaw UI Inspector（桌面试验版 v0.8.1） + +> **有焦点的窗口 = 采集目标**（与大小无关）。 +> 企微暖树：**只调用** `jiangchang_desktop_wake.ensure_ui_ready`（与技能同路径），不复制注入逻辑。 +> 主窗 → profile `wecom`；搜索浮层 `SearchResultWindow2` → profile `wecom_search`（需 kit≥1.2.6）。 + +## 启动 + +```powershell +cd D:\OpenClaw\client-commons\skill-template +# 建议使用已安装 jiangchang-platform-kit>=1.2.5 的 Python（与宿主共享 venv 一致） +python tools/ui_inspector/run.py --scene demo +``` + +依赖： + +```powershell +pip install -r tools/ui_inspector/requirements.txt +# 企微自动暖树（强烈建议） +pip install "jiangchang-platform-kit>=1.2.5" # 或指向你们 Gitea/本地 index +``` + +## 规则 + +1. 焦点在谁就采谁（排除 Inspector 自身 hwnd） +2. **自动拾取默认勾选**；点「开始流程」后先 **5 秒倒数**，请在此期间点击目标窗 +3. Inspector **保持置顶**便于看进度；置顶 ≠ 焦点，不因此改采本面板 +4. **不强制最大化**目标窗，保持操作者当时的大小/位置（仅最小化时会还原） +5. 不会因为窗口小 / FindAll 少而改采别的软件 +6. **企微自动暖树（无感）**：识别到 `WeWorkWindow` 时，在准备/拾取前静默调用 + `ensure_ui_ready("wecom")`；已暖则快速返回。结果写入每步 `meta.json` → `dump_stats.kit_wake` + +## 空壳说明 + +- **已装 kit**：冷企微应被 n2-wake 暖到 `findall` 较大、`provider_cold=false`。若仍冷，看状态栏 `kit_wake: FAILED …`。 +- **未装 kit**：行为与旧版相同，可能 `findall≈2`、`provider_cold=true`；状态栏会提示未安装 platform-kit。 + +## 验收建议（企微） + +1. 打开并登录桌面企业微信 +2. 启动 Inspector →「开始流程」→ 倒数内点到企微主窗 +3. 看详情区 / 第一步 `meta.json`：`kit_wake.warm=true`，`findall` 通常 ≫ 80 +4. 再搜联系人、开会话，确认 `groupbuddyname` 的 Value 能采到 + +## 冒烟（不弹 GUI） + +```powershell +python tools/ui_inspector/smoke_check.py +``` diff --git a/tools/ui_inspector/__init__.py b/tools/ui_inspector/__init__.py new file mode 100644 index 0000000..3aa7dfd --- /dev/null +++ b/tools/ui_inspector/__init__.py @@ -0,0 +1,3 @@ +"""Desktop UI Inspector — development-only UIA capture tool (additive, disposable).""" + +__version__ = "0.8.1" diff --git a/tools/ui_inspector/__main__.py b/tools/ui_inspector/__main__.py new file mode 100644 index 0000000..1716458 --- /dev/null +++ b/tools/ui_inspector/__main__.py @@ -0,0 +1,12 @@ +"""python -m tools.ui_inspector (from skill-template root, if tools is a package) + +Prefer: python tools/ui_inspector/run.py +""" + +from __future__ import annotations + +import runpy +from pathlib import Path + +if __name__ == "__main__": + runpy.run_path(str(Path(__file__).resolve().parent / "app.py"), run_name="__main__") diff --git a/tools/ui_inspector/app.py b/tools/ui_inspector/app.py new file mode 100644 index 0000000..440bde6 --- /dev/null +++ b/tools/ui_inspector/app.py @@ -0,0 +1,784 @@ +"""Tk panel v0.8.1: 开始流程 → 手动拾取 / 自动串行循环 → 结束流程。 + +企微主窗 / 搜索浮层在识别/拾取前静默调用 kit ensure（wecom / wecom_search）。 +""" + +from __future__ import annotations + +import argparse +import os +import subprocess +import sys +import threading +from pathlib import Path +import tkinter as tk +from tkinter import messagebox, simpledialog, ttk + +_HERE = Path(__file__).resolve().parent +if str(_HERE) not in sys.path: + sys.path.insert(0, str(_HERE)) + +import artifacts as art # noqa: E402 +import desktop_uia as uia # noqa: E402 + +try: + from . import __version__ +except Exception: + try: + import importlib.util + + _spec = importlib.util.spec_from_file_location( + "ui_inspector_pkg", _HERE / "__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.1") + except Exception: + __version__ = "0.8.1" + +# Auto loop: after each dump finishes, wait this long, then dump again. +_AUTO_GAP_MS = 2000 +# After frontmost target is resolved, short delay before first dump. +_AUTO_START_DELAY_MS = 400 +# After「开始流程」with auto-pick: countdown so user can focus the target app. +_START_COUNTDOWN_SEC = 5 +_FRONTMOST_WAIT_SEC = 15.0 + + +class InspectorApp: + def __init__( + self, + *, + captures_root: Path, + scene: str = "desktop", + max_depth: int = 16, + max_nodes: int = 8000, + ) -> None: + self.captures_root = captures_root + self.scene = scene + self.max_depth = max_depth + self.max_nodes = max_nodes + + self.flow_dir: Path | None = None + self.flow: dict | None = None + self.last_window: dict | None = None + self.sticky_pid: int | None = None + self._warmed_pids: set[int] = set() + self._busy = False + self._auto_loop_active = False + self._pending_end = False + self._auto_after_id: str | None = None + self._countdown_after_id: str | None = None + self._countdown_left = 0 + + self.root = tk.Tk() + self.root.title(f"OpenClaw UI Inspector v{__version__} (desktop)") + self.root.geometry("840x660") + self.root.minsize(720, 540) + self.root.attributes("-topmost", True) + + self.flow_name_var = tk.StringVar(value=self.scene) + self.auto_pick_var = tk.BooleanVar(value=True) + self.flow_var = tk.StringVar(value="流程：未开始（请先点「开始流程」）") + self.target_var = tk.StringVar(value="尚未拾取") + self.status_var = tk.StringVar( + value="用法：开始流程 →（默认自动拾取）5 秒倒数 → 串行 dump。" + "企微将自动 ensure_ui_ready（需 platform-kit≥1.2.5）；选窗看焦点，不强制放大。" + ) + + self._build_ui() + self.root.protocol("WM_DELETE_WINDOW", self._on_close) + + def _build_ui(self) -> None: + pad = {"padx": 8, "pady": 4} + + top = ttk.Frame(self.root) + top.pack(fill=tk.X, **pad) + ttk.Label(top, text="流程名").pack(side=tk.LEFT) + ttk.Entry(top, textvariable=self.flow_name_var, width=28).pack(side=tk.LEFT, padx=4) + ttk.Button(top, text="打开输出目录", command=self.on_open_out).pack(side=tk.LEFT, padx=8) + + btns = ttk.Frame(self.root) + btns.pack(fill=tk.X, **pad) + self.start_btn = ttk.Button(btns, text="1. 开始流程", command=self.on_start_flow) + self.start_btn.pack(side=tk.LEFT, padx=2) + self.pick_btn = ttk.Button(btns, text="2. 拾取当前页", command=self.on_pick_page) + self.pick_btn.pack(side=tk.LEFT, padx=2) + self.end_btn = ttk.Button(btns, text="3. 结束流程", command=self.on_end_flow) + self.end_btn.pack(side=tk.LEFT, padx=2) + self.auto_chk = ttk.Checkbutton( + btns, + text="自动拾取（串行·步间2秒）", + variable=self.auto_pick_var, + command=self.on_auto_toggle, + ) + self.auto_chk.pack(side=tk.LEFT, padx=12) + + info = ttk.LabelFrame(self.root, text="状态") + info.pack(fill=tk.X, **pad) + ttk.Label(info, textvariable=self.flow_var, wraplength=800, justify=tk.LEFT).pack( + anchor=tk.W, padx=6, pady=2 + ) + ttk.Label(info, textvariable=self.target_var, wraplength=800, justify=tk.LEFT).pack( + anchor=tk.W, padx=6, pady=2 + ) + ttk.Label(info, textvariable=self.status_var, wraplength=800, justify=tk.LEFT).pack( + anchor=tk.W, padx=6, pady=2 + ) + + tree_box = ttk.LabelFrame(self.root, text="本步拾取结果") + tree_box.pack(fill=tk.BOTH, expand=True, **pad) + scroll = ttk.Scrollbar(tree_box) + scroll.pack(side=tk.RIGHT, fill=tk.Y) + self.detail = tk.Text( + tree_box, + height=24, + wrap=tk.WORD, + yscrollcommand=scroll.set, + ) + self.detail.pack(fill=tk.BOTH, expand=True, padx=6, pady=6) + scroll.config(command=self.detail.yview) + self.detail.insert( + tk.END, + "1) 填写流程名，点「开始流程」——新建一份流程文档\n" + "2) 默认已勾选「自动拾取」：先 5 秒倒数（请点目标窗到最前端），\n" + " 然后串行：dump → 保存 → 等 2 秒 → 再 dump … 直到「结束流程」\n" + " 本窗口保持置顶方便看进度；选目标看焦点并排除自身\n" + " 不强制最大化目标窗（保持你当时的大小/位置）\n" + "3) 也可取消自动，改为手动点「拾取当前页」\n" + "4) 点「结束流程」——退出循环并关闭文档\n\n" + "完整内容在 _captures/<流程>/flow.json、SUMMARY.md、steps/\n", + ) + self.detail.configure(state=tk.DISABLED) + + tip = ( + "一个流程 = 一份文档；多次拾取 = 多个 steps。" + "自动拾取严格串行：上一拍完全结束才会开始下一拍。" + "给 AI 时优先丢 SUMMARY.md + 各步 locatable.json。" + ) + ttk.Label(self.root, text=tip, wraplength=800, foreground="#444").pack( + fill=tk.X, padx=8, pady=6 + ) + self._refresh_button_states() + + def _flow_is_open(self) -> bool: + return bool(self.flow_dir and self.flow and self.flow.get("status") == "open") + + def _in_start_countdown(self) -> bool: + return self._countdown_after_id is not None or self._countdown_left > 0 + + def _refresh_button_states(self) -> None: + open_flow = self._flow_is_open() + counting = self._in_start_countdown() + # Manual pick only when open, idle, and not in auto loop / countdown + can_manual = ( + open_flow + and not self._busy + and not self._auto_loop_active + and not counting + ) + self.pick_btn.configure(state=tk.NORMAL if can_manual else tk.DISABLED) + # End must stay clickable during a long dump so the loop can be stopped + self.end_btn.configure(state=tk.NORMAL if open_flow else tk.DISABLED) + self.start_btn.configure( + state=tk.DISABLED if (self._busy or counting) else tk.NORMAL + ) + # Don't flip checkbox mid-dump in a confusing way; allow toggle to stop loop + self.auto_chk.configure(state=tk.NORMAL) + + def _set_status(self, text: str) -> None: + self.status_var.set(text) + + def _set_detail(self, text: str) -> None: + self.detail.configure(state=tk.NORMAL) + self.detail.delete("1.0", tk.END) + self.detail.insert(tk.END, text) + self.detail.see("1.0") + self.detail.configure(state=tk.DISABLED) + + def _show_picking_placeholder(self, *, auto: bool) -> None: + n = int((self.flow or {}).get("step_count") or 0) + 1 + mode = "自动" if auto else "手动" + self.target_var.set(f"{mode}拾取中…（即将写入第 {n} 步）") + self._set_status(f"正在{mode}拾取：已清空本框上次结果，请稍候…") + self._set_detail( + "（已清空上次本步结果）\n\n" + f"正在读取控件树，完成后写入当前流程第 {n} 步…\n" + ) + + def _exclude_hwnds(self) -> set[int]: + try: + return {int(self.root.winfo_id())} + except Exception: + return set() + + def _update_flow_label(self) -> None: + if not self.flow_dir or not self.flow: + self.flow_var.set("流程：未开始（请先点「开始流程」）") + return + if self._auto_loop_active: + auto = "LOOP" + elif self.auto_pick_var.get() and self._flow_is_open(): + auto = "armed" + else: + auto = "off" + self.flow_var.set( + f"流程：{self.flow.get('flow_name')} | status={self.flow.get('status')} | " + f"steps={self.flow.get('step_count')} | auto={auto} | {self.flow_dir}" + ) + + def _cancel_auto_timer(self) -> None: + if self._auto_after_id is not None: + try: + self.root.after_cancel(self._auto_after_id) + except Exception: + pass + self._auto_after_id = None + + def _cancel_countdown(self) -> None: + if self._countdown_after_id is not None: + try: + self.root.after_cancel(self._countdown_after_id) + except Exception: + pass + self._countdown_after_id = None + self._countdown_left = 0 + + def _stop_auto_loop(self) -> None: + self._auto_loop_active = False + self._cancel_auto_timer() + self._cancel_countdown() + + def _start_auto_loop(self, *, first_delay_ms: int = _AUTO_START_DELAY_MS) -> None: + self._auto_loop_active = True + self._pending_end = False + self._update_flow_label() + self._refresh_button_states() + self._schedule_auto_next(first_delay_ms) + + def _set_inspector_topmost(self, enabled: bool) -> None: + try: + self.root.attributes("-topmost", bool(enabled)) + except Exception: + pass + + def _ensure_inspector_visible(self) -> None: + """Keep panel on-screen for progress; do not force-focus (avoids stealing FG).""" + try: + if str(self.root.state()) == "iconic": + self.root.deiconify() + except Exception: + pass + + def _arm_auto_with_countdown(self, *, seconds: int = _START_COUNTDOWN_SEC) -> None: + """Visible countdown, then resolve frontmost and enter auto loop.""" + if not self._flow_is_open(): + return + # Stop prior loop/timers without clearing the countdown we are about to arm. + self._auto_loop_active = False + self._cancel_auto_timer() + self._cancel_countdown() + self._ensure_inspector_visible() + # Stay topmost for progress; capture still uses focus + exclude self. + self._set_inspector_topmost(True) + self._countdown_left = max(1, int(seconds)) + self._update_flow_label() + self._set_status( + f"倒计时 {self._countdown_left} 秒：请把要采集的窗口点到最前端" + "（本面板置顶看进度；不强制放大目标窗）。" + ) + self._set_detail( + f"流程：{self.flow_dir}\n\n" + f"- {self._countdown_left} 秒内请点击目标窗口（给它焦点）\n" + f"- 本 Inspector 保持置顶，方便看进度；选窗排除自身\n" + f"- 倒计时结束后：识别最前端 → dump → 等 2 秒 → 再 dump …\n" + f"- 目标窗保持你当前的大小/位置（不强制最大化）\n" + f"- 可点「结束流程」或取消「自动拾取」中止\n" + ) + self._refresh_button_states() + self._countdown_after_id = self.root.after(1000, self._countdown_tick) + + def _countdown_tick(self) -> None: + self._countdown_after_id = None + if not self._flow_is_open(): + self._countdown_left = 0 + self._refresh_button_states() + return + if not self.auto_pick_var.get(): + self._countdown_left = 0 + self._refresh_button_states() + self._set_status("已取消自动拾取倒计时；可手动点「拾取当前页」。") + return + self._countdown_left -= 1 + if self._countdown_left > 0: + self._set_status( + f"倒计时 {self._countdown_left} 秒：请把要采集的窗口点到最前端…" + ) + self._refresh_button_states() + self._countdown_after_id = self.root.after(1000, self._countdown_tick) + return + self._countdown_left = 0 + self._refresh_button_states() + self._set_status("倒计时结束，正在识别最前端窗口…") + self._begin_auto_with_probe() + + def _begin_auto_with_probe(self) -> None: + """Resolve frontmost target (not Inspector), warmup best-effort, then loop.""" + if self._busy: + return + self._busy = True + self._auto_loop_active = False + self._refresh_button_states() + # Keep topmost for progress; target = focused window (Inspector hwnd excluded). + self._set_inspector_topmost(True) + self._ensure_inspector_visible() + self._set_status( + "自动拾取：正在识别最前端窗口（焦点窗=采集目标，已排除本面板）…" + ) + self._set_detail( + "识别最前端窗口中…\n\n" + "- 规则：谁有焦点就采谁（与窗口大小无关）\n" + "- 本面板可置顶；选目标时排除 Inspector 自身\n" + "- 不强制最大化目标窗\n" + "- 若是企微：自动调用 kit ensure_ui_ready（与技能同路径）\n" + "- FindAll 多少只作提示，不会因此改采别的软件\n" + ) + + def work() -> None: + try: + uia.ensure_com_sta() + report = uia.prepare_frontmost_for_capture( + exclude_hwnds=self._exclude_hwnds(), + timeout_sec=_FRONTMOST_WAIT_SEC, + ) + self.root.after(0, lambda r=report: self._on_frontmost_ready(r)) + except Exception as exc: + err = exc + self.root.after(0, lambda e=err: self._on_frontmost_fail(e)) + + threading.Thread(target=work, daemon=True).start() + + def _on_frontmost_ready(self, report: dict) -> None: + self._busy = False + self._refresh_button_states() + win = report.get("window") or {} + if win.get("pid"): + self.sticky_pid = int(win["pid"]) + self._warmed_pids.add(int(win["pid"])) + import kit_wake + + kit = report.get("kit_wake") or {} + kit_line = kit_wake.format_kit_wake_line(kit) + sparse_note = "" + if report.get("provider_cold") or report.get("uia_sparse"): + if kit.get("reason") == "kit_not_installed": + sparse_note = ( + f"（仍偏稀 FindAll={report.get('findall')}：未安装 platform-kit，" + "企微无法走 n2-wake）" + ) + elif kit.get("applicable") and not kit.get("ok"): + sparse_note = ( + f"（kit 暖树未成功 FindAll={report.get('findall')} " + f"code={kit.get('error_code')}；仍按此窗采集）" + ) + else: + sparse_note = ( + f"（UIA 仍偏稀 FindAll={report.get('findall')}；仍按此前端窗采集）" + ) + self._set_status( + f"已锁定最前端：{win.get('title')!r} / {win.get('class_name')!r} " + f"findall={report.get('findall')} — 开始自动串行拾取。{sparse_note}" + ) + self._set_detail( + f"采集目标（最前端）:\n" + f" title={win.get('title')!r}\n" + f" class={win.get('class_name')!r}\n" + f" hwnd={win.get('hwnd')} pid={win.get('pid')}\n" + f" findall={report.get('findall')} sparse={report.get('uia_sparse')} " + f"cold={report.get('provider_cold')}\n" + f" {kit_line}\n\n" + "即将进入自动循环（步间 2 秒）…\n" + ) + self._set_inspector_topmost(True) + self._ensure_inspector_visible() + self._start_auto_loop(first_delay_ms=_AUTO_START_DELAY_MS) + + def _on_frontmost_fail(self, exc: BaseException) -> None: + self._busy = False + self._auto_loop_active = False + # Keep auto checked (default on); user can retry or end flow. + self._set_inspector_topmost(True) + self._ensure_inspector_visible() + self._refresh_button_states() + self._set_status(f"未能锁定最前端目标: {exc}") + messagebox.showerror("UI Inspector", f"未能锁定最前端目标:\n{exc}") + + def _schedule_auto_next(self, delay_ms: int) -> None: + self._cancel_auto_timer() + if not self._auto_loop_active or not self._flow_is_open(): + return + self._auto_after_id = self.root.after(delay_ms, self._auto_loop_tick) + + def _auto_loop_tick(self) -> None: + self._auto_after_id = None + if not self._auto_loop_active or not self._flow_is_open(): + return + if self._busy: + # Should not happen in serial design; re-check soon. + self._schedule_auto_next(200) + return + self.on_pick_page(auto=True) + + def _continue_auto_loop_after_pick(self) -> None: + if self._pending_end: + self._pending_end = False + self._stop_auto_loop() + self._finish_end_flow() + return + if self._auto_loop_active and self._flow_is_open(): + self._set_status( + f"自动循环：第 {self.flow.get('step_count') if self.flow else '?'} 步已处理；" + f"{_AUTO_GAP_MS // 1000} 秒后开始下一拍。点「结束流程」退出循环。" + ) + self._schedule_auto_next(_AUTO_GAP_MS) + + def on_auto_toggle(self) -> None: + if self.auto_pick_var.get(): + if ( + self._flow_is_open() + and not self._auto_loop_active + and not self._busy + and not self._in_start_countdown() + ): + self._arm_auto_with_countdown() + else: + self._update_flow_label() + else: + was = self._auto_loop_active or self._in_start_countdown() + self._stop_auto_loop() + self._update_flow_label() + self._refresh_button_states() + if was and self._flow_is_open(): + self._set_status("已停止自动循环/倒计时；仍可手动点「拾取当前页」。") + elif self._flow_is_open(): + self._set_status("自动拾取已关闭；可手动点「拾取当前页」。") + + def on_open_out(self) -> None: + path = self.flow_dir if self.flow_dir else self.captures_root + path.mkdir(parents=True, exist_ok=True) + try: + os.startfile(str(path)) # type: ignore[attr-defined] + except Exception: + subprocess.Popen(["explorer", str(path)]) + + def on_start_flow(self) -> None: + if self._busy: + return + if self.flow and self.flow.get("status") == "open": + if not messagebox.askyesno( + "UI Inspector", + "当前已有进行中的流程。\n是否结束旧流程并开始新流程？", + ): + return + self._stop_auto_loop() + self._pending_end = False + try: + art.end_flow(self.flow_dir, self.flow) # type: ignore[arg-type] + except Exception: + pass + + name = (self.flow_name_var.get() or "").strip() + if not name: + name = simpledialog.askstring("流程名", "请输入流程名称：", initialvalue="desktop-flow") + if not name: + return + self.flow_name_var.set(name) + + try: + self.flow_dir = art.create_flow(self.captures_root, flow_name=name) + self.flow = art.load_flow(self.flow_dir) + self._pending_end = False + self._warmed_pids.clear() + self._update_flow_label() + self._refresh_button_states() + if self.auto_pick_var.get(): + # No modal / no minimize: keep panel visible for countdown + progress. + self._arm_auto_with_countdown(seconds=_START_COUNTDOWN_SEC) + else: + self._stop_auto_loop() + self._ensure_inspector_visible() + self._set_status( + f"已开始流程：{self.flow_dir.name}。请翻到目标页后点「拾取当前页」。" + ) + self._set_detail( + f"流程已创建：{self.flow_dir}\n\n" + f"- flow.json / SUMMARY.md / steps/\n" + f"- 手动点「拾取当前页」追加一步；或勾选「自动拾取」进入串行循环。\n" + ) + except Exception as exc: + messagebox.showerror("UI Inspector", str(exc)) + + def _finish_end_flow(self) -> None: + if not self.flow_dir or not self.flow: + return + if self.flow.get("status") != "open": + self._refresh_button_states() + return + try: + self.flow = art.end_flow(self.flow_dir, self.flow) + self._update_flow_label() + self._refresh_button_states() + n = self.flow.get("step_count") or 0 + self._set_status(f"流程已结束，共 {n} 步。可再点「开始流程」开新文档。") + self._set_detail( + f"流程已结束：{self.flow_dir}\n\n" + f"- 共 {n} 步\n" + f"- 请打开 SUMMARY.md 查看总览\n" + ) + messagebox.showinfo("UI Inspector", f"流程已结束\n共 {n} 步\n{self.flow_dir}") + except Exception as exc: + messagebox.showerror("UI Inspector", str(exc)) + + def on_end_flow(self) -> None: + if not self.flow_dir or not self.flow: + messagebox.showwarning("UI Inspector", "没有进行中的流程。") + return + if self.flow.get("status") != "open": + messagebox.showinfo("UI Inspector", "流程已经结束。") + return + + self._stop_auto_loop() + # Keep「自动拾取」勾选状态（默认开），方便下一轮直接开始。 + + if self._busy: + self._pending_end = True + self._set_status("已请求结束：等待当前这一拍 dump/保存完成…") + self._refresh_button_states() + return + + self._pending_end = False + self._finish_end_flow() + + def on_pick_page(self, *, auto: bool = False) -> None: + if self._busy: + return + if not self._flow_is_open(): + if not auto: + messagebox.showwarning("UI Inspector", "请先点「开始流程」。") + return + if auto and not self._auto_loop_active: + return + + self._busy = True + self._refresh_button_states() + self._show_picking_placeholder(auto=auto) + self.root.update_idletasks() + + def work() -> None: + try: + uia.ensure_com_sta() + # Keep panel topmost for progress; selection excludes Inspector hwnd. + self.root.after(0, lambda: self._set_inspector_topmost(True)) + flow = art.load_flow(self.flow_dir) # type: ignore[arg-type] + if flow.get("status") != "open": + raise RuntimeError("流程已结束，取消本步拾取。") + info, cands = uia.find_capture_target( + exclude_hwnds=self._exclude_hwnds(), + prefer_pid=None if auto else self.sticky_pid, + ) + # Prefer light after PID already warmed; dump_window_tree escalates + # to full poke if locatable still sparse (still no force maximize). + warmup_mode = ( + "light" if int(info.pid) in self._warmed_pids else "full" + ) + tree, index, stats, locatable = uia.dump_window_tree( + info.hwnd, + max_depth=self.max_depth, + max_nodes=self.max_nodes, + only_visible=False, + warmup_mode=warmup_mode, + merge_same_pid_if_sparse=True, + ) + self._warmed_pids.add(int(info.pid)) + window = info.to_dict() + label = art.resolve_step_label( + flow, + window_title=info.title or "", + window_class=info.class_name or "", + locatable=locatable, + ) + step_dir, step_idx, step_name = art.add_step_dir( + self.flow_dir, # type: ignore[arg-type] + flow, + window_title=info.title or "", + window_class=info.class_name or "", + locatable=locatable, + label=label, + ) + meta = art.build_meta( + window=window, + scene=flow.get("flow_name") or "flow", + extra={ + "flow_dir": str(self.flow_dir), + "step_index": step_idx, + "step_name": step_name, + "step_label": label, + "auto_pick": auto, + "dump_stats": stats, + "selection": { + "strategy": "z-order+last-active-popup", + "sticky_pid": self.sticky_pid, + "candidate_count": len(cands), + }, + }, + ) + art.write_json(step_dir / "meta.json", meta) + art.write_json(step_dir / "tree.json", tree) + art.write_jsonl(step_dir / "index.jsonl", index) + art.write_json(step_dir / "locatable.json", locatable) + png_ok = art.capture_window_png(info.hwnd, step_dir / "screen.png") + noteworthy = art.noteworthy_from_index(index) + art.write_step_summary( + step_dir, meta=meta, stats=stats, noteworthy=noteworthy + ) + art.register_step( + self.flow_dir, # type: ignore[arg-type] + flow, + step_index=step_idx, + step_name=step_name, + window=window, + stats=stats, + label=label, + ) + flow = art.load_flow(self.flow_dir) # type: ignore[arg-type] + + loc_text = uia.format_locatable_text(locatable) + tree_text = uia.format_tree_text(tree) + import kit_wake + + warm = stats.get("warmup") or {} + kit = stats.get("kit_wake") or {} + kit_line = kit_wake.format_kit_wake_line(kit) + sparse_note = "" + if stats.get("uia_sparse"): + sparse_note = "\n⚠ uia_sparse=true：可定位候选偏少，请结合 screen.png。\n" + mode = "自动" if auto else "手动" + header = ( + f"流程步骤: #{step_idx} `{step_name}`（{mode}）\n" + f"窗口: title={info.title!r} class={info.class_name!r} " + f"pid={info.pid} hwnd={info.hwnd}\n" + f"kit: {kit_line}\n" + f"软poke: findall {warm.get('findall_before')} → {warm.get('findall_after')} " + f"woke={warm.get('woke')}\n" + f"统计: kept={stats.get('nodes_kept')} findall={stats.get('findall_count')} " + f"locatable={stats.get('locatable_count')} png={png_ok} " + f"sparse={stats.get('uia_sparse')} cold={stats.get('provider_cold')}\n" + f"已写入: {step_dir}\n" + f"流程文档: {self.flow_dir}\n" + f"{sparse_note}" + f"{'=' * 60}\n" + ) + body = header + "\n" + loc_text + "\n\n" + ("=" * 60) + "\n控件树:\n" + tree_text + self.root.after( + 0, + lambda i=info, f=flow, t=body, sn=step_name, a=auto: self._on_pick_ok( + i, f, t, sn, auto=a + ), + ) + except Exception as exc: + err = exc + self.root.after(0, lambda e=err, a=auto: self._on_fail(e, auto=a)) + + threading.Thread(target=work, daemon=True).start() + + def _on_pick_ok( + self, + info: uia.WindowInfo, + flow: dict, + text: str, + step_name: str, + *, + auto: bool = False, + ) -> None: + self._busy = False + self.flow = flow + self.last_window = info.to_dict() + self.sticky_pid = int(info.pid) + self._update_flow_label() + self._refresh_button_states() + self.target_var.set( + f"本步目标: title={info.title!r} class={info.class_name!r} | step=`{step_name}`" + ) + self._set_detail(text) + if not auto: + self._set_status( + f"第 {flow.get('step_count')} 步已写入。可换页继续拾取，或点「结束流程」。" + ) + try: + self.root.lift() + self.root.attributes("-topmost", True) + except Exception: + pass + if self._pending_end: + self._pending_end = False + self._finish_end_flow() + return + + # auto: either pending end, or continue loop after gap + self._continue_auto_loop_after_pick() + + def _on_fail(self, exc: BaseException, *, auto: bool = False) -> None: + self._busy = False + self._refresh_button_states() + self._set_status(f"失败: {exc}") + if not auto: + messagebox.showerror("UI Inspector", str(exc)) + if self._pending_end: + self._pending_end = False + self._finish_end_flow() + return + # Auto: do not pop modal; keep looping after gap (unless ending) + self._continue_auto_loop_after_pick() + + def _on_close(self) -> None: + self._stop_auto_loop() + self._pending_end = False + self.root.destroy() + + def run(self) -> None: + self.captures_root.mkdir(parents=True, exist_ok=True) + self.root.mainloop() + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + p = argparse.ArgumentParser( + description="OpenClaw desktop UI Inspector v0.6（流程文档：开始/拾取/结束/自动串行）" + ) + p.add_argument("--out", type=Path, default=None, help="输出根目录") + p.add_argument("--scene", default="desktop", help="默认流程名") + p.add_argument("--max-depth", type=int, default=16) + p.add_argument("--max-nodes", type=int, default=8000) + return p.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + out = args.out.resolve() if args.out else art.default_captures_root(_HERE) + try: + uia.ensure_uia() + except Exception as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + app = InspectorApp( + captures_root=out, + scene=args.scene, + max_depth=args.max_depth, + max_nodes=args.max_nodes, + ) + app.run() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/ui_inspector/artifacts.py b/tools/ui_inspector/artifacts.py new file mode 100644 index 0000000..f177df9 --- /dev/null +++ b/tools/ui_inspector/artifacts.py @@ -0,0 +1,694 @@ +"""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_/ 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) diff --git a/tools/ui_inspector/desktop_uia.py b/tools/ui_inspector/desktop_uia.py new file mode 100644 index 0000000..d5d7377 --- /dev/null +++ b/tools/ui_inspector/desktop_uia.py @@ -0,0 +1,1597 @@ +"""UI Automation helpers: attach, control-from-point, tree dump.""" + +from __future__ import annotations + +import ctypes +import time +from ctypes import wintypes +from dataclasses import asdict, dataclass, field +from typing import Any, Optional + +user32 = ctypes.windll.user32 + +WM_GETOBJECT = 0x003D +OBJID_WINDOW = 0x00000000 +OBJID_CLIENT = 0xFFFFFFFC +OBJID_SYSMENU = 0xFFFFFFFF +# UIA root object id ((DWORD)-25) — forces UI Automation provider, not just MSAA. +UIA_ROOT_OBJECT_ID = ctypes.c_long(0xFFFFFFE7).value +SW_RESTORE = 9 +SW_SHOW = 5 +SW_MAXIMIZE = 3 +GA_ROOT = 2 +GW_HWNDNEXT = 2 +GW_OWNER = 4 +GWL_EXSTYLE = -20 +WS_EX_TOOLWINDOW = 0x00000080 +WS_EX_NOACTIVATE = 0x08000000 +DWMWA_CLOAKED = 14 + + +@dataclass +class WindowInfo: + hwnd: int + pid: int + title: str + class_name: str + process_name: str = "" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass +class ControlInfo: + name: str = "" + value: str = "" + automation_id: str = "" + control_type: str = "" + class_name: str = "" + framework_id: str = "" + localized_control_type: str = "" + is_enabled: Optional[bool] = None + is_offscreen: Optional[bool] = None + rect: dict[str, int] = field(default_factory=dict) + depth: int = 0 + parent_chain: list[dict[str, str]] = field(default_factory=list) + path_hint: str = "" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def ensure_uia() -> None: + try: + import uiautomation # noqa: F401 + except ImportError as exc: + raise RuntimeError( + "缺少 uiautomation。请先: pip install uiautomation>=2.0.18" + ) from exc + + +def ensure_com_sta() -> None: + """ + UIAutomation is STA-friendly. Worker threads must enter an apartment + or FindAll often returns an empty shell (findall=2~3). + """ + try: + import comtypes + + try: + comtypes.CoInitializeEx(comtypes.COINIT_APARTMENTTHREADED) + except OSError: + # Already initialized on this thread (or MTA) — ignore. + pass + except Exception: + try: + comtypes.CoInitialize() + except Exception: + pass + except Exception: + pass + + +def window_area(hwnd: int) -> int: + rect = _window_rect(int(hwnd)) + if not rect: + return 0 + return max(0, int(rect.get("width") or 0)) * max(0, int(rect.get("height") or 0)) + + +def _force_foreground(hwnd: int) -> bool: + """AttachThreadInput + SetForeground — more reliable than bare SetForeground.""" + hwnd = int(hwnd) + try: + fg = int(user32.GetForegroundWindow() or 0) + tid_fg = int(user32.GetWindowThreadProcessId(fg, None) or 0) if fg else 0 + tid_t = int(user32.GetWindowThreadProcessId(hwnd, None) or 0) + attached = False + if tid_fg and tid_t and tid_fg != tid_t: + attached = bool(user32.AttachThreadInput(tid_fg, tid_t, True)) + try: + user32.BringWindowToTop(hwnd) + ok = bool(user32.SetForegroundWindow(hwnd)) + try: + user32.SetFocus(hwnd) + except Exception: + pass + return ok + finally: + if attached: + user32.AttachThreadInput(tid_fg, tid_t, False) + except Exception: + try: + return bool(user32.SetForegroundWindow(hwnd)) + except Exception: + return False + + +def activate_main_hwnd(hwnd: int, *, maximize: bool = False) -> dict[str, Any]: + """ + Light activation for capture: + - restore only if minimized + - maximize only when explicitly requested (Inspector default: off) + - SetForeground only if not already foreground (avoid flash) + Only call on the chosen main/capture hwnd — never on every child. + """ + hwnd = int(hwnd) + info: dict[str, Any] = { + "hwnd": hwnd, + "iconic": False, + "zoomed_before": False, + "maximized": False, + "foreground": False, + "already_foreground": False, + "maximize_requested": bool(maximize), + } + try: + info["iconic"] = bool(user32.IsIconic(hwnd)) + info["zoomed_before"] = bool(user32.IsZoomed(hwnd)) + except Exception: + pass + + if info["iconic"]: + user32.ShowWindow(hwnd, SW_RESTORE) + time.sleep(0.15) + if maximize and not info["zoomed_before"]: + user32.ShowWindow(hwnd, SW_MAXIMIZE) + time.sleep(0.2) + + try: + already = int(user32.GetForegroundWindow() or 0) == hwnd + except Exception: + already = False + info["already_foreground"] = already + if already: + info["foreground"] = True + else: + info["foreground"] = _force_foreground(hwnd) + time.sleep(0.25 if maximize else 0.15) + try: + info["maximized"] = bool(user32.IsZoomed(hwnd)) + except Exception: + pass + return info + + +def is_provider_cold(hwnd: int, *, findall: Optional[int] = None) -> bool: + """True when WM_GETOBJECT CLIENT is 0 and FindAll is an empty shell.""" + hwnd = int(hwnd) + n = int(findall) if findall is not None else findall_count_for_hwnd(hwnd) + if n < 0: + n = 0 + if n >= 8: + return False + try: + res = int(user32.SendMessageW(hwnd, WM_GETOBJECT, 0, OBJID_CLIENT) or 0) + except Exception: + res = 0 + return res == 0 and n < 8 + + +def is_mainish_window(info: WindowInfo) -> bool: + """Large primary app window (safe to maximize).""" + cls = info.class_name or "" + if cls in {"SearchResultWindow2", "IME", "MSCTFIME UI", "tooltips_class32"}: + return False + if "SearchResult" in cls: + return False + area = window_area(info.hwnd) + if area >= 500 * 400: + return True + if "WeWork" in cls or "Chrome_WidgetWin" in cls: + return area >= 300 * 200 + return bool((info.title or "").strip()) and area >= 600 * 400 + + +def largest_window(candidates: list[WindowInfo]) -> Optional[WindowInfo]: + if not candidates: + return None + return max(candidates, key=lambda c: window_area(c.hwnd)) + + +def _safe_str(value: Any) -> str: + if value is None: + return "" + try: + return str(value).strip() + except Exception: + return "" + + +def _rect_dict(ctrl) -> dict[str, int]: + try: + r = ctrl.BoundingRectangle + return { + "left": int(r.left), + "top": int(r.top), + "right": int(r.right), + "bottom": int(r.bottom), + "width": int(r.width()), + "height": int(r.height()), + } + except Exception: + return {} + + +def _read_value(ctrl) -> str: + # Align with production skills: Value often lives in ValueValueProperty. + try: + import uiautomation as auto + + raw = ctrl.GetPropertyValue(auto.PropertyId.ValueValueProperty) + s = _safe_str(raw) + if s: + return s + except Exception: + pass + try: + pattern = ctrl.GetValuePattern() + if pattern is not None: + s = _safe_str(pattern.Value) + if s: + return s + except Exception: + pass + for attr in ("Value", "LegacyIAccessibleValue", "HelpText"): + try: + s = _safe_str(getattr(ctrl, attr, None)) + if s: + return s + except Exception: + continue + return "" + + +def _control_type_name(ctrl) -> str: + try: + ct = ctrl.ControlTypeName + if ct: + return _safe_str(ct) + except Exception: + pass + try: + return _safe_str(ctrl.ControlType) + except Exception: + return "" + + +def describe_control(ctrl, *, depth: int = 0, parent_chain: Optional[list] = None) -> ControlInfo: + info = ControlInfo( + name=_safe_str(getattr(ctrl, "Name", None)), + value=_read_value(ctrl), + automation_id=_safe_str(getattr(ctrl, "AutomationId", None)), + control_type=_control_type_name(ctrl), + class_name=_safe_str(getattr(ctrl, "ClassName", None)), + framework_id=_safe_str(getattr(ctrl, "FrameworkId", None)), + localized_control_type=_safe_str(getattr(ctrl, "LocalizedControlType", None)), + depth=depth, + parent_chain=list(parent_chain or []), + rect=_rect_dict(ctrl), + ) + try: + info.is_enabled = bool(ctrl.IsEnabled) + except Exception: + info.is_enabled = None + try: + info.is_offscreen = bool(ctrl.IsOffscreen) + except Exception: + info.is_offscreen = None + + bits = [] + if info.control_type: + bits.append(info.control_type) + if info.name: + bits.append(f"Name={info.name!r}") + if info.value: + bits.append(f"Value={info.value!r}") + if info.automation_id: + bits.append(f"AutomationId={info.automation_id!r}") + info.path_hint = " | ".join(bits) + return info + + +def get_toplevel_hwnd(hwnd: int) -> int: + hwnd = int(hwnd) + if not hwnd: + return 0 + root = int(user32.GetAncestor(hwnd, GA_ROOT) or 0) + return root or hwnd + + +def window_info_from_hwnd(hwnd: int) -> WindowInfo: + ensure_uia() + import uiautomation as auto + + hwnd = get_toplevel_hwnd(int(hwnd)) + if not hwnd or not user32.IsWindow(hwnd): + raise RuntimeError(f"无效窗口 HWND={hwnd}") + + title_buf = ctypes.create_unicode_buffer(512) + user32.GetWindowTextW(hwnd, title_buf, 512) + class_buf = ctypes.create_unicode_buffer(256) + user32.GetClassNameW(hwnd, class_buf, 256) + + pid = ctypes.c_ulong(0) + user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid)) + + process_name = "" + try: + root = auto.ControlFromHandle(hwnd) + process_name = _safe_str(getattr(root, "ProcessName", None)) + except Exception: + pass + + return WindowInfo( + hwnd=hwnd, + pid=int(pid.value), + title=_safe_str(title_buf.value), + class_name=_safe_str(class_buf.value), + process_name=process_name, + ) + + +def get_foreground_window_info() -> WindowInfo: + hwnd = int(user32.GetForegroundWindow()) + if not hwnd: + raise RuntimeError("当前没有前台窗口") + return window_info_from_hwnd(hwnd) + + +def is_inspector_window(info: WindowInfo, *, self_hwnd: int = 0) -> bool: + title = info.title or "" + if self_hwnd and int(info.hwnd) == int(self_hwnd): + return True + if self_hwnd and get_toplevel_hwnd(self_hwnd) == int(info.hwnd): + return True + if "UI Inspector" in title or "OpenClaw UI Inspector" in title: + return True + return False + + +def is_system_chrome_window(info: WindowInfo) -> bool: + cls = info.class_name or "" + return cls in { + "Shell_TrayWnd", + "Shell_SecondaryTrayWnd", + "NotifyIconOverflowWindow", + "Progman", + "WorkerW", + } + + +def _window_rect(hwnd: int) -> dict[str, int]: + rect = wintypes.RECT() + if not user32.GetWindowRect(int(hwnd), ctypes.byref(rect)): + return {} + return { + "left": int(rect.left), + "top": int(rect.top), + "right": int(rect.right), + "bottom": int(rect.bottom), + "width": int(rect.right - rect.left), + "height": int(rect.bottom - rect.top), + } + + +def _is_cloaked(hwnd: int) -> bool: + try: + dwmapi = ctypes.windll.dwmapi + cloaked = ctypes.c_int(0) + hr = dwmapi.DwmGetWindowAttribute( + wintypes.HWND(int(hwnd)), + ctypes.c_int(DWMWA_CLOAKED), + ctypes.byref(cloaked), + ctypes.sizeof(cloaked), + ) + return hr == 0 and int(cloaked.value) != 0 + except Exception: + return False + + +def _exstyle(hwnd: int) -> int: + return int(user32.GetWindowLongW(int(hwnd), GWL_EXSTYLE) or 0) + + +def iter_toplevel_hwnds_zorder(): + """Front-to-back top-level windows (generic Win32 Z-order walk).""" + hwnd = int(user32.GetTopWindow(None) or 0) + seen: set[int] = set() + while hwnd: + if hwnd in seen: + break + seen.add(hwnd) + yield hwnd + hwnd = int(user32.GetWindow(hwnd, GW_HWNDNEXT) or 0) + + +def list_capturable_candidates( + *, + exclude_hwnds: Optional[set[int]] = None, + min_width: int = 120, + min_height: int = 80, +) -> list[WindowInfo]: + """Z-order capturable windows; no app-name hardcoding.""" + exclude = {int(h) for h in (exclude_hwnds or set()) if h} + # Also exclude top-level of each excluded hwnd + exclude |= {get_toplevel_hwnd(h) for h in list(exclude)} + + out: list[WindowInfo] = [] + for hwnd in iter_toplevel_hwnds_zorder(): + if hwnd in exclude: + continue + if not user32.IsWindowVisible(hwnd): + continue + if user32.IsIconic(hwnd): + continue + if _is_cloaked(hwnd): + continue + rect = _window_rect(hwnd) + if not rect or rect.get("width", 0) < min_width or rect.get("height", 0) < min_height: + continue + try: + info = window_info_from_hwnd(hwnd) + except Exception: + continue + if "UI Inspector" in (info.title or "") or "OpenClaw UI Inspector" in (info.title or ""): + continue + if is_system_chrome_window(info) and info.class_name in { + "Shell_TrayWnd", + "Shell_SecondaryTrayWnd", + "NotifyIconOverflowWindow", + "Progman", + "WorkerW", + }: + continue + if is_system_chrome_window(info) and not (info.title or "").strip(): + continue + # Skip untitled toolwindows (IME, ghosts) + ex = _exstyle(hwnd) + if (ex & WS_EX_TOOLWINDOW) and not (info.title or "").strip(): + continue + out.append(info) + return out + + +def _owner_root(hwnd: int) -> int: + cur = int(hwnd) + seen: set[int] = set() + while cur and cur not in seen: + seen.add(cur) + owner = int(user32.GetWindow(cur, GW_OWNER) or 0) + if not owner: + break + cur = owner + return cur or int(hwnd) + + +def prefer_last_active_popup(info: WindowInfo) -> WindowInfo: + """If Win32 reports an active owned popup, prefer it (still app-agnostic).""" + try: + root = _owner_root(info.hwnd) + popup = int(user32.GetLastActivePopup(root) or 0) + if not popup or popup == int(info.hwnd): + # also try popup of the candidate itself + popup = int(user32.GetLastActivePopup(int(info.hwnd)) or 0) + if not popup or not user32.IsWindowVisible(popup) or user32.IsIconic(popup): + return info + if _is_cloaked(popup): + return info + prect = _window_rect(popup) + if not prect or prect.get("width", 0) < 80 or prect.get("height", 0) < 60: + return info + popup_info = window_info_from_hwnd(popup) + if is_system_chrome_window(popup_info): + return info + return popup_info + except Exception: + return info + + +def foreground_toplevel_hwnd(*, exclude_hwnds: Optional[set[int]] = None) -> Optional[int]: + """True foreground top-level hwnd (GetForegroundWindow), if capturable.""" + exclude = {int(h) for h in (exclude_hwnds or set()) if h} + exclude |= {get_toplevel_hwnd(h) for h in list(exclude)} + try: + fg = int(user32.GetForegroundWindow() or 0) + except Exception: + fg = 0 + if not fg: + return None + top = get_toplevel_hwnd(fg) + if top in exclude: + return None + if not user32.IsWindowVisible(top) or user32.IsIconic(top) or _is_cloaked(top): + return None + return int(top) + + +def find_capture_target( + *, + exclude_hwnds: Optional[set[int]] = None, + prefer_pid: Optional[int] = None, +) -> tuple[WindowInfo, list[WindowInfo]]: + """ + Pick the frontmost capturable window. + + Priority: + 1) GetForegroundWindow (true frontmost), if capturable + 2) else Z-order top capturable candidate + Optional prefer_pid only filters when sticky (same front app pid); + never jumps to another app just because it is larger. + """ + candidates = list_capturable_candidates(exclude_hwnds=exclude_hwnds) + if not candidates: + raise RuntimeError( + "未找到可拾取的顶层窗口。请确认目标软件窗口可见且未被最小化。" + ) + + by_hwnd = {int(c.hwnd): c for c in candidates} + chosen: Optional[WindowInfo] = None + + fg = foreground_toplevel_hwnd(exclude_hwnds=exclude_hwnds) + if fg and fg in by_hwnd: + # Frontmost always wins — never replace with a larger/sticky other app. + chosen = by_hwnd[fg] + elif fg: + # Foreground exists but filtered from candidates — still try to describe it. + try: + info = window_info_from_hwnd(fg) + if not is_inspector_window(info) and not ( + is_system_chrome_window(info) and not (info.title or "").strip() + ): + chosen = info + except Exception: + chosen = None + + if chosen is None: + # No usable foreground: Z-order front. prefer_pid only as soft filter here. + if prefer_pid: + for c in candidates: + if int(c.pid) == int(prefer_pid): + chosen = c + break + if chosen is None: + chosen = candidates[0] + + # Owned popup of the *same* front window only (e.g. search panel), never another app + popup = prefer_last_active_popup(chosen) + if int(popup.pid) == int(chosen.pid): + chosen = popup + return chosen, candidates + + +def wait_frontmost_target( + *, + exclude_hwnds: Optional[set[int]] = None, + timeout_sec: float = 12.0, + poll_sec: float = 0.35, +) -> tuple[WindowInfo, list[WindowInfo], dict[str, Any]]: + """ + Wait until the true foreground window is a capturable target + (not the Inspector). Frontmost = the capture target. No size ranking. + """ + ensure_com_sta() + ensure_uia() + exclude = {int(h) for h in (exclude_hwnds or set()) if h} + deadline = time.time() + float(timeout_sec) + meta: dict[str, Any] = {"waited": False, "polls": 0} + last_err = "" + while time.time() < deadline: + meta["polls"] += 1 + try: + info, cands = find_capture_target(exclude_hwnds=exclude_hwnds) + except Exception as exc: + last_err = str(exc) + time.sleep(poll_sec) + continue + # If foreground resolved to an excluded hwnd path, candidates[0] is next. + # Reject only when we somehow still landed on Inspector. + if is_inspector_window(info): + time.sleep(poll_sec) + meta["waited"] = True + continue + meta["window"] = info.to_dict() + return info, cands, meta + if last_err: + raise RuntimeError(f"等待最前端目标窗口超时：{last_err}") + raise RuntimeError( + "等待最前端目标窗口超时：请先用鼠标点一下要采集的软件窗口（不要点 Inspector）。" + ) + + +def prepare_frontmost_for_capture( + *, + exclude_hwnds: Optional[set[int]] = None, + timeout_sec: float = 12.0, +) -> dict[str, Any]: + """ + Resolve frontmost target, light-activate if mainish (no force maximize), warmup. + Always returns the frontmost window — findall is informational only, + never a reason to pick another app. + + 企微：先静默调用 kit ``ensure_ui_ready("wecom")``（与技能同路径），再轻量 poke。 + """ + import kit_wake + + info, cands, wait_meta = wait_frontmost_target( + exclude_hwnds=exclude_hwnds, timeout_sec=timeout_sec + ) + act = None + if is_mainish_window(info): + # Keep operator size/position; only restore if minimized. + act = activate_main_hwnd(info.hwnd, maximize=False) + + kit_report = kit_wake.auto_ensure_if_wecom( + class_name=info.class_name or "", + title=info.title or "", + hwnd=info.hwnd, + force=True, + ) + + warm = warmup_uia_provider( + info.hwnd, interact_window=False, allow_foreground_nudge=False + ) + n = int(warm.get("findall_after") or findall_count_for_hwnd(info.hwnd) or 0) + if kit_report.get("ok") and int(kit_report.get("findall") or 0) > n: + n = int(kit_report["findall"]) + return { + "ready": True, # frontmost resolved => ready to capture that window + "window": info.to_dict(), + "candidates": len(cands), + "findall": n, + "uia_sparse": n < 8, + "provider_cold": is_provider_cold(info.hwnd, findall=n), + "activate": act, + "warmup": warm, + "kit_wake": kit_report, + "wait": wait_meta, + "strategy": "foreground_only", + } + + +def enum_child_hwnds(parent: int, *, max_count: int = 80) -> list[int]: + result: list[int] = [] + + @ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM) + def _cb(hwnd, _lparam): + if len(result) >= max_count: + return False + if user32.IsWindowVisible(hwnd): + result.append(int(hwnd)) + return True + + user32.EnumChildWindows(int(parent), _cb, 0) + return result + + +def is_window_minimized(hwnd: int) -> bool: + hwnd = int(hwnd) + if user32.IsIconic(hwnd): + return True + rect = _window_rect(hwnd) + # Windows parks minimized windows near (-32000,-32000) + if rect and int(rect.get("left") or 0) <= -30000 and int(rect.get("top") or 0) <= -30000: + return True + return False + + +def restore_window_if_needed( + hwnd: int, + *, + force_show: bool = False, + bring_foreground: bool = False, +) -> dict[str, Any]: + """ + Only restore when minimized by default. + + Avoid unconditional ShowWindow/SetForeground — repeating those on child/tool + HWNDs (esp. WeCom) spawns/flickers many tiny windows under the process. + """ + hwnd = int(hwnd) + info = { + "hwnd": hwnd, + "was_minimized": False, + "restored": False, + "shown": False, + "foreground": False, + } + if is_window_minimized(hwnd): + info["was_minimized"] = True + user32.ShowWindow(hwnd, SW_RESTORE) + time.sleep(0.2) + info["restored"] = True + elif force_show: + try: + user32.ShowWindow(hwnd, SW_SHOW) + info["shown"] = True + except Exception: + pass + if bring_foreground: + try: + if user32.SetForegroundWindow(hwnd): + info["foreground"] = True + except Exception: + pass + time.sleep(0.05) + return info + + +def findall_count_for_hwnd(hwnd: int) -> int: + try: + from uiautomation.uiautomation import _AutomationClient + + root = control_from_hwnd(int(hwnd)) + ac = _AutomationClient.instance() + arr = root.Element.FindAll(7, ac.IUIAutomation.CreateTrueCondition()) + return int(arr.Length) + except Exception: + return -1 + + +def poke_uia_provider(hwnd: int) -> dict[str, Any]: + """ + Light provider poke: no ShowWindow / no SetForeground. + + Safe to call on child or sibling HWNDs without popping tool windows. + """ + ensure_uia() + hwnd = int(hwnd) + report: dict[str, Any] = {"hwnd": hwnd, "mode": "light", "steps": []} + for obj_id, label in ( + (UIA_ROOT_OBJECT_ID, "UiaRootObjectId"), + (OBJID_CLIENT, "OBJID_CLIENT"), + ): + try: + res = int(user32.SendMessageW(hwnd, WM_GETOBJECT, 0, obj_id) or 0) + report["steps"].append({"wm_getobject": label, "lresult": res}) + except Exception as exc: + report["steps"].append({"wm_getobject": label, "error": str(exc)}) + try: + from uiautomation.uiautomation import _AutomationClient + import uiautomation as auto + + ac = _AutomationClient.instance() + el = ac.IUIAutomation.ElementFromHandle(wintypes.HWND(hwnd)) + report["steps"].append({"ElementFromHandle": el is not None}) + ctrl = auto.ControlFromHandle(hwnd) + try: + arr = ctrl.Element.FindAll(4, ac.IUIAutomation.CreateTrueCondition()) + report["steps"].append({"poke_FindAll_descendants": int(arr.Length)}) + except Exception as exc: + report["steps"].append({"poke_FindAll": str(exc)}) + except Exception as exc: + report["steps"].append({"uia_refresh": str(exc)}) + return report + + +def _msaa_and_uia_poke(hwnd: int, report: dict[str, Any]) -> None: + """Shared MSAA/UIA attach steps (no ShowWindow).""" + hwnd = int(hwnd) + for obj_id, label in ( + (UIA_ROOT_OBJECT_ID, "UiaRootObjectId"), + (OBJID_CLIENT, "OBJID_CLIENT"), + (OBJID_WINDOW, "OBJID_WINDOW"), + ): + try: + res = int(user32.SendMessageW(hwnd, WM_GETOBJECT, 0, obj_id) or 0) + report["steps"].append({"wm_getobject": label, "lresult": res}) + except Exception as exc: + report["steps"].append({"wm_getobject": label, "error": str(exc)}) + + try: + from comtypes import GUID, IUnknown + + iid = GUID("{618736e0-3c3d-11cf-810c-00aa00389b71}") # IID_IAccessible + punk = ctypes.c_void_p() + hr = ctypes.windll.oleacc.AccessibleObjectFromWindow( + wintypes.HWND(hwnd), + ctypes.c_long(OBJID_CLIENT), + ctypes.byref(iid), + ctypes.byref(punk), + ) + report["steps"].append( + { + "AccessibleObjectFromWindow": int(hr), + "has_punk": bool(punk.value), + } + ) + if punk.value: + try: + ctypes.cast(punk, ctypes.POINTER(IUnknown)).Release() + except Exception: + pass + except Exception as exc: + report["steps"].append({"AccessibleObjectFromWindow": str(exc)}) + + try: + from uiautomation.uiautomation import _AutomationClient + import uiautomation as auto + + ac = _AutomationClient.instance() + el = ac.IUIAutomation.ElementFromHandle(wintypes.HWND(hwnd)) + report["steps"].append({"ElementFromHandle": el is not None}) + ctrl = auto.ControlFromHandle(hwnd) + try: + kids = ctrl.GetChildren() + report["steps"].append({"GetChildren_count": len(kids or [])}) + except Exception as exc: + report["steps"].append({"GetChildren": str(exc)}) + try: + arr = ctrl.Element.FindAll(4, ac.IUIAutomation.CreateTrueCondition()) + report["steps"].append({"poke_FindAll_descendants": int(arr.Length)}) + except Exception as exc: + report["steps"].append({"poke_FindAll": str(exc)}) + except Exception as exc: + report["steps"].append({"uia_refresh": str(exc)}) + + +def warmup_uia_provider( + hwnd: int, + *, + interact_window: bool = True, + allow_foreground_nudge: bool = True, + sparse_findall_threshold: int = 8, +) -> dict[str, Any]: + """ + Force accessibility provider initialization (Inspect-like). + + - Restores minimized main window only (no blanket Show on children). + - If still sparse after poke, optionally SetForeground once on THIS hwnd + (never on child/tool HWNDs — that was the popup storm). + """ + ensure_uia() + hwnd = int(hwnd) + report: dict[str, Any] = {"hwnd": hwnd, "mode": "full", "steps": []} + + if interact_window: + report["restore"] = restore_window_if_needed( + hwnd, force_show=False, bring_foreground=False + ) + else: + report["restore"] = {"skipped": True} + + report["findall_before"] = findall_count_for_hwnd(hwnd) + _msaa_and_uia_poke(hwnd, report) + time.sleep(0.15) + report["findall_after"] = findall_count_for_hwnd(hwnd) + + if ( + allow_foreground_nudge + and interact_window + and int(report.get("findall_after") or 0) < int(sparse_findall_threshold) + ): + # One-shot focus nudge on the capture target only. + nudge = restore_window_if_needed( + hwnd, force_show=False, bring_foreground=True + ) + report["foreground_nudge"] = nudge + time.sleep(0.35) + _msaa_and_uia_poke(hwnd, report) + report["findall_after_nudge"] = findall_count_for_hwnd(hwnd) + report["findall_after"] = report["findall_after_nudge"] + + report["woke"] = int(report.get("findall_after") or 0) > max( + 5, int(report.get("findall_before") or 0) + 2 + ) + report["provider_cold"] = is_provider_cold( + hwnd, findall=int(report.get("findall_after") or 0) + ) + return report + + +def list_same_pid_toplevel_hwnds(pid: int) -> list[int]: + found: list[int] = [] + + @ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM) + def _cb(hwnd, _lp): + p = ctypes.c_ulong(0) + user32.GetWindowThreadProcessId(hwnd, ctypes.byref(p)) + if int(p.value) == int(pid): + found.append(int(hwnd)) + return True + + user32.EnumWindows(_cb, 0) + return found + + +def dump_window_tree( + hwnd: int, + *, + max_depth: int = 16, + max_nodes: int = 8000, + only_visible: bool = False, + enrich_child_hwnds: bool = True, + use_findall: bool = True, + warmup: bool = True, + warmup_mode: str = "full", + merge_same_pid_if_sparse: bool = True, +) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any], list[dict[str, Any]]]: + """ + Dump UIA tree + flat index + locatable candidates. + + warmup_mode: + - full: MSAA/UIA poke on target; restore only if minimized + - light: ElementFromHandle poke only (no ShowWindow) + - off: no provider warmup + Child / same-PID merges never call ShowWindow (light poke at most). + """ + import kit_wake + + hwnd = int(hwnd) + mode = (warmup_mode or "full").lower() + if not warmup: + mode = "off" + if mode not in {"full", "light", "off"}: + mode = "full" + + kit_report: dict[str, Any] = {"applicable": False, "skipped": True, "reason": "not_checked"} + try: + winfo_for_kit = window_info_from_hwnd(hwnd) + except Exception: + winfo_for_kit = None + if winfo_for_kit is not None: + hint = findall_count_for_hwnd(hwnd) + kit_report = kit_wake.auto_ensure_if_wecom( + class_name=winfo_for_kit.class_name or "", + title=winfo_for_kit.title or "", + hwnd=hwnd, + force=False, + findall_hint=hint, + ) + + warmup_report: dict[str, Any] = {"mode": mode} + if mode == "full": + try: + winfo = window_info_from_hwnd(hwnd) + except Exception: + winfo = None + if winfo is not None and is_mainish_window(winfo): + # No force maximize — preserve operator window geometry. + act = activate_main_hwnd(hwnd, maximize=False) + warmup_report = warmup_uia_provider( + hwnd, interact_window=False, allow_foreground_nudge=False + ) + warmup_report["activate"] = act + else: + warmup_report = warmup_uia_provider( + hwnd, interact_window=True, allow_foreground_nudge=True + ) + elif mode == "light": + warmup_report = poke_uia_provider(hwnd) + + def _run_dump() -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]: + root_ctrl = control_from_hwnd(hwnd) + t, idx, st = dump_tree( + root_ctrl, + max_depth=max_depth, + max_nodes=max_nodes, + only_visible=only_visible, + ) + st = dict(st) + st["enriched_child_hwnds"] = False + st["findall_count"] = 0 + st["merged_same_pid_hwnds"] = [] + + if enrich_child_hwnds: + children_nodes: list[dict[str, Any]] = list(t.get("children") or []) + seen_hwnd = {int(hwnd)} + for child_hwnd in enum_child_hwnds(hwnd, max_count=80): + if child_hwnd in seen_hwnd: + continue + seen_hwnd.add(child_hwnd) + if int(st.get("nodes_kept") or 0) >= max_nodes: + st["truncated"] = True + break + try: + # Never full-warmup / ShowWindow child HWNDs. + child_root = control_from_hwnd(child_hwnd) + except Exception: + continue + remain = max_nodes - int(st.get("nodes_kept") or 0) + sub_tree, sub_index, sub_stats = dump_tree( + child_root, + max_depth=max(2, max_depth - 1), + max_nodes=max(1, remain), + only_visible=only_visible, + ) + if int(sub_stats.get("nodes_kept") or 0) <= 1 and not ( + (sub_tree.get("name") or "") or (sub_tree.get("value") or "") + ): + if int(sub_stats.get("nodes_kept") or 0) <= 0: + continue + children_nodes.append(sub_tree) + idx.extend(sub_index) + st["nodes_kept"] = int(st.get("nodes_kept") or 0) + int( + sub_stats.get("nodes_kept") or 0 + ) + st["nodes_visited"] = int(st.get("nodes_visited") or 0) + int( + sub_stats.get("nodes_visited") or 0 + ) + st["enriched_child_hwnds"] = True + if sub_stats.get("truncated"): + st["truncated"] = True + if children_nodes: + t = dict(t) + t["children"] = children_nodes + + if use_findall: + found = collect_findall_flat(root_ctrl, max_nodes=max_nodes) + st["findall_count"] = len(found) + existing = {r.get("path_hint") for r in idx} + for row in found: + if row.get("path_hint") in existing: + continue + idx.append(row) + existing.add(row.get("path_hint")) + return t, idx, st + + tree, index, stats = _run_dump() + stats["warmup"] = warmup_report + stats["warmup_mode"] = mode + stats["kit_wake"] = kit_report + stats["walker"] = "GetFirstChild/NextSibling (package default RawViewWalker)" + stats["escalated_warmup"] = False + + locatable = locatable_from_index(index) + stats["locatable_count"] = len(locatable) + stats["uia_sparse"] = int(stats.get("locatable_count") or 0) < 8 + stats["provider_cold"] = is_provider_cold( + hwnd, findall=int(stats.get("findall_count") or 0) + ) + + # 仍冷且目标为企微：强制再 ensure 一次（绕过短时 cache）后重 dump + if stats["provider_cold"] and winfo_for_kit is not None and kit_wake.looks_like_wecom( + class_name=winfo_for_kit.class_name or "", + title=winfo_for_kit.title or "", + ): + kit_report = kit_wake.auto_ensure_if_wecom( + class_name=winfo_for_kit.class_name or "", + title=winfo_for_kit.title or "", + hwnd=hwnd, + force=True, + ) + stats["kit_wake"] = kit_report + if kit_report.get("ok"): + tree, index, stats = _run_dump() + stats["warmup"] = warmup_report + stats["warmup_mode"] = mode + stats["kit_wake"] = kit_report + stats["walker"] = "GetFirstChild/NextSibling (package default RawViewWalker)" + stats["escalated_warmup"] = False + stats["kit_wake_retry"] = True + locatable = locatable_from_index(index) + stats["locatable_count"] = len(locatable) + stats["uia_sparse"] = int(stats.get("locatable_count") or 0) < 8 + stats["provider_cold"] = is_provider_cold( + hwnd, findall=int(stats.get("findall_count") or 0) + ) + + # If light dump is sparse, escalate once to full warmup + re-dump (no child ShowWindow). + if stats["uia_sparse"] and mode == "light": + try: + winfo = window_info_from_hwnd(hwnd) + except Exception: + winfo = None + if winfo is not None and is_mainish_window(winfo): + act = activate_main_hwnd(hwnd, maximize=False) + warmup_report = warmup_uia_provider( + hwnd, interact_window=False, allow_foreground_nudge=False + ) + warmup_report["activate"] = act + else: + warmup_report = warmup_uia_provider( + hwnd, interact_window=True, allow_foreground_nudge=True + ) + tree, index, stats = _run_dump() + stats["warmup"] = warmup_report + stats["warmup_mode"] = "full" + stats["kit_wake"] = kit_report + stats["escalated_warmup"] = True + locatable = locatable_from_index(index) + stats["locatable_count"] = len(locatable) + stats["uia_sparse"] = int(stats.get("locatable_count") or 0) < 8 + stats["provider_cold"] = is_provider_cold( + hwnd, findall=int(stats.get("findall_count") or 0) + ) + + # Sparse fallback: light poke + FindAll on a few large same-pid top-levels. + # Includes WeWorkWindow when current target is a thin popup. + if merge_same_pid_if_sparse and stats["uia_sparse"]: + try: + info = window_info_from_hwnd(hwnd) + merged_n = 0 + # Prefer large main-ish windows first + others: list[tuple[int, int, Any]] = [] + for other in list_same_pid_toplevel_hwnds(info.pid): + if other == hwnd: + continue + rect = _window_rect(other) + if not rect: + continue + area = int(rect.get("width") or 0) * int(rect.get("height") or 0) + if int(rect.get("width") or 0) < 120 or int(rect.get("height") or 0) < 80: + continue + others.append((area, other, rect)) + others.sort(key=lambda x: -x[0]) + for _area, other, _rect in others: + if merged_n >= 8: + break + try: + oinfo = window_info_from_hwnd(other) + except Exception: + continue + if is_inspector_window(oinfo): + continue + if oinfo.class_name in { + "Shell_TrayWnd", + "Shell_SecondaryTrayWnd", + "Progman", + "WorkerW", + "IME", + "MSCTFIME UI", + "tooltips_class32", + }: + continue + # Light poke only — never ShowWindow on siblings + poke_uia_provider(other) + try: + oroot = control_from_hwnd(other) + extra = collect_findall_flat( + oroot, max_nodes=max(100, max_nodes - len(index)) + ) + except Exception: + continue + if len(extra) < 5: + continue + merged_n += 1 + stats.setdefault("merged_same_pid_hwnds", []).append( + { + "hwnd": other, + "title": oinfo.title, + "class_name": oinfo.class_name, + "findall": len(extra), + } + ) + existing = {r.get("path_hint") for r in index} + for row in extra: + row = dict(row) + row["source"] = f"findall_pid:{other}" + if row.get("path_hint") in existing: + continue + index.append(row) + existing.add(row.get("path_hint")) + locatable = locatable_from_index(index) + stats["locatable_count"] = len(locatable) + stats["findall_count"] = max( + int(stats.get("findall_count") or 0), + sum(1 for r in index if str(r.get("source", "")).startswith("findall")), + ) + stats["uia_sparse"] = int(stats.get("locatable_count") or 0) < 8 + except Exception as exc: + stats["same_pid_merge_error"] = str(exc) + + stats["kit_wake"] = kit_report + return tree, index, stats, locatable + + +def collect_findall_flat(root_ctrl, *, max_nodes: int = 8000) -> list[dict[str, Any]]: + """COM FindAll(Subtree) — often finds nodes GetChildren walk misses.""" + out: list[dict[str, Any]] = [] + try: + from uiautomation.uiautomation import Control, _AutomationClient + + ac = _AutomationClient.instance() + cond = ac.IUIAutomation.CreateTrueCondition() + # TreeScope_Subtree = Element|Children|Descendants = 7 + arr = root_ctrl.Element.FindAll(7, cond) + length = int(arr.Length) + for i in range(min(length, max_nodes)): + try: + ele = arr.GetElement(i) + ctrl = Control.CreateControlFromElement(ele) + if ctrl is None: + continue + info = describe_control(ctrl, depth=-1, parent_chain=[]) + out.append( + { + "depth": -1, + "source": "findall", + "name": info.name, + "value": info.value, + "automation_id": info.automation_id, + "control_type": info.control_type, + "class_name": info.class_name, + "is_enabled": info.is_enabled, + "is_offscreen": info.is_offscreen, + "rect": info.rect, + "path_hint": info.path_hint, + } + ) + except Exception: + continue + except Exception: + return out + return out + + +def locatable_from_index( + index: list[dict[str, Any]], *, limit: int = 300 +) -> list[dict[str, Any]]: + """Nodes useful for locator drafting: non-empty Name / Value / AutomationId.""" + rows: list[dict[str, Any]] = [] + seen: set[str] = set() + for r in index: + name = (r.get("name") or "").strip() + value = (r.get("value") or "").strip() + aid = (r.get("automation_id") or "").strip() + if not (name or value or aid): + continue + # Skip ultra-generic noise alone + if name in {"", "pane", "Pane"} and not value and not aid: + continue + key = f"{r.get('control_type')}|{name}|{value}|{aid}" + if key in seen: + continue + seen.add(key) + rows.append(r) + if len(rows) >= limit: + break + # Prefer rows with value or automation_id + rows.sort( + key=lambda x: ( + 0 if (x.get("value") or x.get("automation_id")) else 1, + 0 if x.get("name") else 1, + int(x.get("depth") if x.get("depth") is not None and x.get("depth") >= 0 else 99), + ) + ) + return rows + + +def format_locatable_text(locatable: list[dict[str, Any]], *, limit: int = 120) -> str: + lines = [ + f"可定位候选（Name/Value/AutomationId 非空，共 {len(locatable)}，显示前 {min(limit, len(locatable))}）", + "", + ] + for i, r in enumerate(locatable[:limit]): + lines.append(f"[{i}] {r.get('control_type') or '?'}") + if r.get("name"): + lines.append(f" Name: {r.get('name')}") + if r.get("value"): + lines.append(f" Value: {r.get('value')}") + if r.get("automation_id"): + lines.append(f" AutomationId: {r.get('automation_id')}") + if r.get("class_name"): + lines.append(f" Class: {r.get('class_name')}") + lines.append( + f" offscreen={r.get('is_offscreen')} enabled={r.get('is_enabled')} " + f"src={r.get('source') or 'tree'}" + ) + lines.append("") + if not locatable: + lines.append("（无）UIA 可能过稀；请看截图目录中的 screen.png，或换页后再拾取。") + return "\n".join(lines) + + + +def format_tree_text(tree: dict[str, Any], *, max_lines: int = 2000) -> str: + """Human-readable indented tree for the panel.""" + lines: list[str] = [] + + def walk(node: dict[str, Any], depth: int) -> None: + if len(lines) >= max_lines: + return + indent = " " * depth + ctype = node.get("control_type") or "?" + name = node.get("name") or "" + value = node.get("value") or "" + aid = node.get("automation_id") or "" + bits = [ctype] + if name: + bits.append(f"Name={name!r}") + if value: + bits.append(f"Value={value!r}") + if aid: + bits.append(f"AutomationId={aid!r}") + lines.append(indent + " | ".join(bits)) + for child in node.get("children") or []: + if len(lines) >= max_lines: + lines.append(indent + " …（已截断显示）") + return + walk(child, depth + 1) + + if tree: + walk(tree, 0) + else: + lines.append("(空树)") + if len(lines) >= max_lines: + lines.append(f"… 显示已截断到 {max_lines} 行；完整内容见 tree.json / index.jsonl") + return "\n".join(lines) + + +# Back-compat alias used by older app versions +def resolve_capture_target( + *, + self_hwnd: int = 0, + poll_sec: float = 0.15, + wait_sec: float = 2.0, +) -> WindowInfo: + _ = poll_sec, wait_sec + info, _cands = find_capture_target(exclude_hwnds={int(self_hwnd)} if self_hwnd else None) + return info + + +def hwnd_from_control(ctrl) -> int: + try: + h = int(getattr(ctrl, "NativeWindowHandle", 0) or 0) + if h: + return h + except Exception: + pass + # Walk parents for a native handle + cur = ctrl + for _ in range(24): + try: + h = int(getattr(cur, "NativeWindowHandle", 0) or 0) + if h: + return h + except Exception: + pass + try: + parent = cur.GetParentControl() + except Exception: + break + if parent is None: + break + cur = parent + return 0 + + +def attach_window_by_click(*, timeout_sec: float = 60.0) -> WindowInfo: + """Minimize race: user clicks anywhere on the target window to attach it.""" + x, y = wait_for_left_click(timeout_sec=timeout_sec) + ctrl = control_from_point(x, y) + hwnd = hwnd_from_control(ctrl) + if not hwnd: + # Fallback: window from point via Win32 + hwnd = int(user32.WindowFromPoint(wintypes.POINT(x, y)) or 0) + if not hwnd: + raise RuntimeError(f"点击位置 ({x},{y}) 无法解析到窗口") + info = window_info_from_hwnd(hwnd) + if "UI Inspector" in (info.title or ""): + raise RuntimeError("点到了 Inspector 自己，请点目标软件窗口（如企业微信/添加客户）。") + return info + + +def control_from_hwnd(hwnd: int): + ensure_uia() + import uiautomation as auto + + ctrl = auto.ControlFromHandle(int(hwnd)) + if ctrl is None: + raise RuntimeError(f"无法从 HWND={hwnd} 取得 UIA 根控件") + return ctrl + + +def control_from_point(x: int, y: int): + ensure_uia() + import uiautomation as auto + + ctrl = auto.ControlFromPoint(int(x), int(y)) + if ctrl is None: + raise RuntimeError(f"坐标 ({x},{y}) 处没有 UIA 控件") + return ctrl + + +def build_parent_chain(ctrl, *, max_ancestors: int = 12) -> list[dict[str, str]]: + chain: list[dict[str, str]] = [] + cur = ctrl + for _ in range(max_ancestors): + try: + parent = cur.GetParentControl() + except Exception: + break + if parent is None: + break + chain.append( + { + "name": _safe_str(getattr(parent, "Name", None)), + "value": _read_value(parent), + "automation_id": _safe_str(getattr(parent, "AutomationId", None)), + "control_type": _control_type_name(parent), + "class_name": _safe_str(getattr(parent, "ClassName", None)), + } + ) + cur = parent + chain.reverse() + return chain + + +def dump_tree( + root, + *, + max_depth: int = 16, + max_nodes: int = 8000, + only_visible: bool = False, +) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]: + """Return (tree, flat_index, stats). + + Important: even when a node is offscreen / zero-sized, we still recurse into + its children. Pruning the whole subtree was the main cause of empty dumps. + """ + stats = { + "nodes_visited": 0, + "nodes_kept": 0, + "nodes_skipped_self": 0, + "truncated": False, + "max_depth": max_depth, + "max_nodes": max_nodes, + "only_visible": only_visible, + } + flat: list[dict[str, Any]] = [] + + def walk(ctrl, depth: int, ancestors: list[dict[str, str]]) -> Optional[dict[str, Any]]: + if stats["nodes_kept"] >= max_nodes: + stats["truncated"] = True + return None + stats["nodes_visited"] += 1 + + info = describe_control(ctrl, depth=depth, parent_chain=ancestors) + skip_self = False + if only_visible and depth > 0: + rect = info.rect + if info.is_offscreen is True: + skip_self = True + if rect and rect.get("width", 0) <= 0 and rect.get("height", 0) <= 0: + skip_self = True + + child_ancestors = ancestors + [ + { + "name": info.name, + "value": info.value, + "automation_id": info.automation_id, + "control_type": info.control_type, + "class_name": info.class_name, + } + ] + children_out: list[dict[str, Any]] = [] + if depth < max_depth and not stats["truncated"]: + try: + children = ctrl.GetChildren() + except Exception: + children = [] + for child in children or []: + if stats["truncated"] or stats["nodes_kept"] >= max_nodes: + stats["truncated"] = True + break + child_node = walk(child, depth + 1, child_ancestors) + if child_node is not None: + children_out.append(child_node) + + if skip_self: + stats["nodes_skipped_self"] += 1 + # Pass children up so hierarchy is not lost + if len(children_out) == 1: + return children_out[0] + if children_out: + return { + "control_type": info.control_type or "SkippedContainer", + "name": info.name, + "value": info.value, + "automation_id": info.automation_id, + "class_name": info.class_name, + "skipped_self": True, + "children": children_out, + } + return None + + stats["nodes_kept"] += 1 + node = info.to_dict() + flat.append( + { + "depth": depth, + "source": "tree", + "name": info.name, + "value": info.value, + "automation_id": info.automation_id, + "control_type": info.control_type, + "class_name": info.class_name, + "is_enabled": info.is_enabled, + "is_offscreen": info.is_offscreen, + "rect": info.rect, + "path_hint": info.path_hint, + } + ) + if children_out: + node["children"] = children_out + return node + + tree = walk(root, 0, []) or {} + return tree, flat, stats + + +def wait_for_left_click( + *, + timeout_sec: float = 60.0, + poll_sec: float = 0.05, +) -> tuple[int, int]: + """Block until left mouse button is pressed then released; return cursor pos.""" + VK_LBUTTON = 0x01 + + def down() -> bool: + return bool(user32.GetAsyncKeyState(VK_LBUTTON) & 0x8000) + + deadline = time.time() + timeout_sec + # Wait until button is up first (avoid consuming the GUI click that started pick) + while time.time() < deadline and down(): + time.sleep(poll_sec) + + while time.time() < deadline: + if down(): + while time.time() < deadline and down(): + time.sleep(poll_sec) + point = wintypes.POINT() + user32.GetCursorPos(ctypes.byref(point)) + return int(point.x), int(point.y) + time.sleep(poll_sec) + raise TimeoutError(f"等待鼠标左键点击超时（{timeout_sec:.0f}s）") + + +def pick_control_at_cursor_after_click( + *, + timeout_sec: float = 60.0, + attached_hwnd: Optional[int] = None, +) -> tuple[ControlInfo, dict[str, int]]: + x, y = wait_for_left_click(timeout_sec=timeout_sec) + ctrl = control_from_point(x, y) + info_hwnd = hwnd_from_control(ctrl) or int( + user32.WindowFromPoint(wintypes.POINT(int(x), int(y))) or 0 + ) + top = get_toplevel_hwnd(info_hwnd) if info_hwnd else 0 + + # Taskbar / desktop chrome — almost never what you want for skill RPA + class_buf = ctypes.create_unicode_buffer(256) + if top: + user32.GetClassNameW(top, class_buf, 256) + top_class = class_buf.value or "" + if top_class in {"Shell_TrayWnd", "Shell_SecondaryTrayWnd", "NotifyIconOverflowWindow"}: + raise RuntimeError( + f"点到了任务栏/系统托盘（class={top_class}），不是业务窗口。\n" + f"请点已附着窗口内部的控件，不要点任务栏上的 Python/Cursor 图标。" + ) + + if attached_hwnd: + attached_top = get_toplevel_hwnd(int(attached_hwnd)) + if top and attached_top and top != attached_top: + title_buf = ctypes.create_unicode_buffer(512) + user32.GetWindowTextW(top, title_buf, 512) + raise RuntimeError( + "拾取点不在已附着窗口内。\n" + f"已附着 hwnd={attached_top}；实际点到 title={title_buf.value!r} " + f"class={top_class!r} hwnd={top}。\n" + "请重新「点选附着」到目标软件（如企业微信），再在该窗口内拾取。" + ) + + chain = build_parent_chain(ctrl) + info = describe_control(ctrl, depth=0, parent_chain=chain) + return info, {"x": x, "y": y} diff --git a/tools/ui_inspector/kit_wake.py b/tools/ui_inspector/kit_wake.py new file mode 100644 index 0000000..e35c076 --- /dev/null +++ b/tools/ui_inspector/kit_wake.py @@ -0,0 +1,192 @@ +"""调用 jiangchang_desktop_wake（只调不复制注入逻辑）。 + +- 主窗 WeWorkWindow → ``ensure_ui_ready("wecom")`` +- 搜索浮层 SearchResultWindow2 → ``ensure_ui_ready("wecom_search")`` +未安装 kit 时降级为跳过，不阻断采集。 +""" + +from __future__ import annotations + +import time +from typing import Any, Optional + +_CACHE_TTL_SEC = 45.0 + +# profile -> cache +_caches: dict[str, dict[str, Any]] = {} + +_PROFILE_WECOM = "wecom" +_PROFILE_WECOM_SEARCH = "wecom_search" + +_THRESHOLDS = { + _PROFILE_WECOM: 80, + _PROFILE_WECOM_SEARCH: 20, +} + + +def resolve_wake_profile(*, class_name: str = "", title: str = "") -> Optional[str]: + """返回应调用的 kit profile id；非企微相关则 None。""" + cls = (class_name or "").strip() + title = title or "" + if cls == "SearchResultWindow2": + return _PROFILE_WECOM_SEARCH + if cls == "WeWorkWindow": + return _PROFILE_WECOM + if "WeWork" in cls and ("企业微信" in title or "WeCom" in title or "wecom" in title.lower()): + return _PROFILE_WECOM + return None + + +def looks_like_wecom(*, class_name: str = "", title: str = "") -> bool: + return resolve_wake_profile(class_name=class_name, title=title) is not None + + +def kit_available() -> bool: + try: + from jiangchang_desktop_wake import ensure_ui_ready # noqa: F401 + + return True + except ImportError: + return False + + +def auto_ensure_if_wecom( + *, + class_name: str = "", + title: str = "", + hwnd: Optional[int] = None, + force: bool = False, + findall_hint: Optional[int] = None, +) -> dict[str, Any]: + """若目标像企微主窗/搜索浮层则静默 ensure；否则跳过。""" + profile = resolve_wake_profile(class_name=class_name, title=title) + if profile is None: + return { + "applicable": False, + "skipped": True, + "reason": "not_wecom", + "available": kit_available(), + } + + warm_min = int(_THRESHOLDS.get(profile, 80)) + cache = _caches.setdefault( + profile, {"t": 0.0, "warm": False, "hwnd": 0, "findall": 0} + ) + now = time.monotonic() + hint = int(findall_hint) if findall_hint is not None else -1 + if ( + not force + and cache.get("warm") + and (now - float(cache.get("t") or 0.0)) < _CACHE_TTL_SEC + and (hint < 0 or hint >= warm_min) + ): + return { + "applicable": True, + "available": True, + "skipped": True, + "reason": "cache_warm", + "ok": True, + "warm": True, + "wake_applied": False, + "findall": int(cache.get("findall") or 0), + "hwnd": int(cache.get("hwnd") or 0), + "profile": profile, + } + + try: + from jiangchang_desktop_wake import ensure_ui_ready + from jiangchang_desktop_wake.errors import DesktopWakeError + except ImportError: + return { + "applicable": True, + "available": False, + "skipped": True, + "reason": "kit_not_installed", + "ok": False, + "profile": profile, + "hint": "请安装 jiangchang-platform-kit>=1.2.6（含 wecom_search）后再采企微", + } + + t0 = time.perf_counter() + try: + result = ensure_ui_ready( + profile, + hwnd=int(hwnd) if hwnd else None, + raise_on_cold=True, + ) + except DesktopWakeError as exc: + raw = (getattr(exc, "code", None) or "").strip() + code = raw[6:] if raw.startswith("ERROR:") else (raw or "WAKE_FAILED") + return { + "applicable": True, + "available": True, + "skipped": False, + "ok": False, + "warm": False, + "wake_applied": False, + "error_code": code, + "message": str(getattr(exc, "message", None) or exc), + "elapsed_ms": int(round((time.perf_counter() - t0) * 1000)), + "profile": profile, + } + except Exception as exc: + return { + "applicable": True, + "available": True, + "skipped": False, + "ok": False, + "warm": False, + "error_code": "ENSURE_EXCEPTION", + "message": f"{type(exc).__name__}: {exc}", + "elapsed_ms": int(round((time.perf_counter() - t0) * 1000)), + "profile": profile, + } + + after = result.probe_after + findall = int(after.findall_count) if after is not None else 0 + report = { + "applicable": True, + "available": True, + "skipped": False, + "ok": bool(result.warm), + "warm": bool(result.warm), + "wake_applied": bool(result.wake_applied), + "findall": findall, + "hwnd": int(result.hwnd) if result.hwnd else 0, + "elapsed_ms": int(round((time.perf_counter() - t0) * 1000)), + "profile": profile, + "markers_found": list(after.markers_found) if after is not None else [], + } + if result.warm: + cache["t"] = now + cache["warm"] = True + cache["hwnd"] = report["hwnd"] + cache["findall"] = findall + else: + cache["warm"] = False + return report + + +def format_kit_wake_line(report: Optional[dict[str, Any]]) -> str: + if not report: + return "kit_wake: (none)" + if not report.get("applicable"): + return "kit_wake: n/a (非企微)" + if report.get("reason") == "kit_not_installed": + return "kit_wake: 未安装 platform-kit（企微可能冷树）" + prof = report.get("profile") or "?" + if report.get("skipped") and report.get("reason") == "cache_warm": + return ( + f"kit_wake[{prof}]: cache warm=True findall={report.get('findall')} " + f"hwnd={report.get('hwnd')}" + ) + if report.get("ok"): + return ( + f"kit_wake[{prof}]: warm={report.get('warm')} " + f"wake_applied={report.get('wake_applied')} " + f"findall={report.get('findall')} elapsed_ms={report.get('elapsed_ms')}" + ) + return ( + f"kit_wake[{prof}]: FAILED code={report.get('error_code')} " + f"msg={report.get('message')}" + ) diff --git a/tools/ui_inspector/requirements.txt b/tools/ui_inspector/requirements.txt new file mode 100644 index 0000000..f2da8b2 --- /dev/null +++ b/tools/ui_inspector/requirements.txt @@ -0,0 +1,5 @@ +uiautomation>=2.0.18 +# 可选：导出 screen.png 需要 Pillow；没有也能 dump JSON +Pillow>=10.0.0 +# 企微自动暖树：与技能同款 ensure_ui_ready（请用宿主共享 venv 或本机安装） +# jiangchang-platform-kit>=1.2.5 diff --git a/tools/ui_inspector/run.py b/tools/ui_inspector/run.py new file mode 100644 index 0000000..9023b74 --- /dev/null +++ b/tools/ui_inspector/run.py @@ -0,0 +1,10 @@ +#!/usr/bin/env python3 +"""Entry: python tools/ui_inspector/run.py [--out DIR] [--scene NAME]""" + +from __future__ import annotations + +import runpy +from pathlib import Path + +if __name__ == "__main__": + runpy.run_path(str(Path(__file__).resolve().parent / "app.py"), run_name="__main__") diff --git a/tools/ui_inspector/smoke_check.py b/tools/ui_inspector/smoke_check.py new file mode 100644 index 0000000..f50ad12 --- /dev/null +++ b/tools/ui_inspector/smoke_check.py @@ -0,0 +1,132 @@ +"""Smoke test for flow document API + step labeling + kit_wake helpers (v0.8.0).""" + +from __future__ import annotations + +import sys +import tempfile +from pathlib import Path + +_HERE = Path(__file__).resolve().parent +if str(_HERE) not in sys.path: + sys.path.insert(0, str(_HERE)) + +import artifacts as art +import kit_wake + + +def _assert_kit_wake_detect() -> None: + assert kit_wake.looks_like_wecom(class_name="WeWorkWindow", title="企业微信") + assert kit_wake.resolve_wake_profile(class_name="SearchResultWindow2") == "wecom_search" + assert not kit_wake.looks_like_wecom(class_name="Chrome_WidgetWin_1", title="Google") + r = kit_wake.auto_ensure_if_wecom(class_name="NotWeCom", title="x") + assert r.get("skipped") and r.get("reason") == "not_wecom" + print("OK kit_wake detect") + + +def _assert_label() -> None: + msg_page = [ + {"control_type": "TreeItemControl", "name": "groupbuddyname", "value": "某群"}, + {"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "你好"}, + {"control_type": "TreeItemControl", "name": "con_op", "value": "消息"}, + ] + searching = [ + {"control_type": "TreeItemControl", "name": "groupbuddyname", "value": "徐书建"}, + {"control_type": "TreeControl", "name": "searchedit", "value": "陈德炼"}, + {"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "陈德炼"}, + ] + selected = [ + { + "control_type": "TreeItemControl", + "name": "groupbuddyname", + "value": "陈德炼@微信", + }, + {"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "您好"}, + ] + contacts = [ + {"control_type": "TreeItemControl", "name": "panel_title_label", "value": "RPA之家"}, + {"control_type": "TreeItemControl", "name": "main_chattext", "value": "新的客户"}, + ] + new_customer = [ + {"control_type": "TreeItemControl", "name": "panel_title_label", "value": "新的客户"}, + {"control_type": "TreeItemControl", "name": "ww_btn_text", "value": "添加"}, + ] + add_empty = [ + {"control_type": "TreeItemControl", "name": "title", "value": "添加客户"}, + {"control_type": "TreeControl", "name": "searchedit", "value": ""}, + {"control_type": "WindowControl", "name": "添加客户", "value": ""}, + ] + add_result = [ + {"control_type": "TreeItemControl", "name": "title", "value": "添加客户"}, + {"control_type": "TreeControl", "name": "searchedit", "value": "13316463772"}, + {"control_type": "TreeItemControl", "name": "main_chattext", "value": "Nancy"}, + {"control_type": "TreeItemControl", "name": "ww_btn_text", "value": "添加为"}, + {"control_type": "TreeControl", "name": "search_result", "value": ""}, + ] + + a = art.derive_step_label(window_title="企业微信", locatable=msg_page) + b = art.derive_step_label(window_title="企业微信", locatable=searching) + c = art.derive_step_label(window_title="企业微信", locatable=selected) + d = art.derive_step_label(window_title="企业微信", locatable=contacts) + e = art.derive_step_label(window_title="企业微信", locatable=new_customer) + f = art.derive_step_label(window_title="添加客户", locatable=add_empty) + g = art.derive_step_label(window_title="添加客户", locatable=add_result) + + assert "输入你好" in a or "某群" in a, a + assert "搜索陈德炼" in b, b + assert "陈德炼" in c and "输入您好" in c, c + assert "RPA之家" in d, d + assert "新的客户" in e and "添加" in e, e + assert f.startswith("添加客户"), f + assert "13316463772" in g or "搜索13316463772" in g, g + assert "会话列表" not in a and "会话列表" not in b and "会话列表" not in c + print(f"OK labels:\n {a}\n {b}\n {c}\n {d}\n {e}\n {f}\n {g}") + + +def main() -> int: + _assert_kit_wake_detect() + _assert_label() + with tempfile.TemporaryDirectory(prefix="ui_flow_smoke_") as tmp: + root = Path(tmp) + flow_dir = art.create_flow(root, flow_name="demo-flow") + flow = art.load_flow(flow_dir) + + loc = [ + { + "control_type": "TreeItemControl", + "name": "groupbuddyname", + "value": "陈德炼@微信", + }, + {"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "您好"}, + ] + label = art.resolve_step_label( + flow, window_title="企业微信", window_class="WeWorkWindow", locatable=loc + ) + step_dir, idx, name = art.add_step_dir( + flow_dir, + flow, + window_title="企业微信", + window_class="WeWorkWindow", + locatable=loc, + label=label, + ) + assert idx == 1 + assert "陈德炼" in name, name + art.write_json(step_dir / "meta.json", {"ok": True}) + art.register_step( + flow_dir, + flow, + step_index=idx, + step_name=name, + window={"title": "企业微信", "class_name": "WeWorkWindow", "pid": 1, "hwnd": 2}, + stats={"findall_count": 10, "locatable_count": 5, "nodes_kept": 3, "uia_sparse": False}, + label=label, + ) + flow = art.end_flow(flow_dir, art.load_flow(flow_dir)) + assert flow["status"] == "ended" + print(f"OK flow={flow_dir.name} step={name}") + print("OK ui_inspector flow smoke passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())