数据截至 2026-08-18(上游 commit 2689884a6257)

legacy 录制管线

这一章讲什么: 人在电脑前操作时,同时有三股信息在变:输入、画面、当前窗口。要把它们录成能回放的数据,难点不在「录」,在**「对齐」**。这一章讲 OpenAdapt 怎么对齐,以及它为什么要用线程 + 进程两种并发。

源码总入口:legacy/openadapt/record.py:1269(record),1653 行,是整个 legacy 里最长的文件。

1. 它要解决的小问题

一次点击要能回放,光知道「在 (832, 419) 按了左键」是不够的。你还得知道:

按下的那一刻屏幕长什么样(否则回放时没法判断该不该点)。
按下的那一刻哪个窗口是活动的、它在屏幕上的位置和大小(否则窗口挪了就全错)。
如果是浏览器,最好还知道点到了哪个 DOM 元素。

三股信息各有各的采集频率,而且都不便宜:截屏慢、读无障碍树更慢、写数据库最慢。如果串行做,人还没点完第二下,第一下的截图还没存完。

2. 思路:线程采集、队列汇合、进程落库

 【线程】读取端                【线程】汇合端                   【进程】写入端

 read_screen_events ──────┐
 read_window_events ──────┤
 read_keyboard_events ────┼──→ event_q ──→ process_events ──┬─→ screen_write_q  ─→ 写进程 ─┐
 read_mouse_events ───────┤    (有序)      (贴标签/配对)    ├─→ action_write_q  ─→ 写进程 ─┼→ SQLite
 run_browser_event_server ┘                                 ├─→ window_write_q  ─→ 写进程 ─┤
                                                            ├─→ browser_write_q ─→ 写进程 ─┤
                                                            └─→ video_write_q   ─→ 写进程 ─┘

为什么读用线程、写用进程?

阶段	用什么	原因
读取端	`threading.Thread`	主要是 I/O 等待(等钩子回调、等截屏返回),GIL 影响小;而且 `pynput` 的监听器天然是回调式的
汇合端	`threading.Thread`	纯内存操作,必须和读取端共享同一个 `queue.Queue`
写入端	`multiprocessing.Process`	PNG 编码、视频编码、SQLite 写入是 CPU 密集,放进程才能真正并行

启动代码见 legacy/openadapt/record.py:1325-1507:窗口/浏览器/屏幕/键盘/鼠标五个读线程 + 一个 event_processor 线程,然后是 screen/browser/action/window/video 五个写进程。其中两条受配置开关控制:浏览器的读线程和写进程都要 config.RECORD_BROWSER_EVENTS(:1337、:1435),视频写进程要 config.RECORD_VIDEO(:1490)。

跨进程的队列不是标准 multiprocessing.Queue,而是 legacy/openadapt/extensions/synchronized_queue.py 里的 SynchronizedQueue——因为写进程需要一个可靠的 qsize() 来做收尾判断,而标准库的 qsize() 在 macOS 上不可用。

3. 核心机制:动作和屏幕/窗口怎么配对

直觉先行

配对规则一句话:一个动作,配上「在它之前最近的那张截图」和「在它之前最近的那个窗口状态」。

再加一条去重规则:同一张截图只在第一次被引用时写库,后面的动作引用同一个时间戳即可。

原理演示

# 示意,非源码:配对的核心就这几行
prev_screen = None
prev_window = None
last_saved_screen_ts = 0

for event in event_queue:
    if event.type == "screen":
        prev_screen = event          # 只记住,先不写库
    elif event.type == "window":
        prev_window = event          # 同上
    elif event.type == "action":
        event.data["screenshot_timestamp"] = prev_screen.timestamp   # 贴标签
        event.data["window_event_timestamp"] = prev_window.timestamp
        write(event)
        if last_saved_screen_ts < prev_screen.timestamp:   # 这张图还没存过
            write(prev_screen)
            last_saved_screen_ts = prev_screen.timestamp

重点看:截图不是被动作触发才拍的,而是一直在拍;但只有被动作引用到的那些才落库。 这样磁盘上只留有用的帧。

真实实现

process_events(legacy/openadapt/record.py:133-279)就是上面那段的完整版:

:227 和 :233 分别贴上 screenshot_timestamp 和 window_event_timestamp。
:245-254 是截图的「首次引用才写」判断,:265-274 是窗口事件的同款判断。
:223-231:如果一个动作出现在任何截图或窗口事件之前,直接丢弃并打 warning。启动瞬间会有这种事件。

一个诚实的缺陷

:186-198 有一段断言时间戳单调递增的代码,断言失败时只打 error 然后继续:

# 真实源码节选,legacy/openadapt/record.py:196-198
logger.error(f"{delta=} {log_prev_event=} {log_event=}")
# behavior undefined, swallow for now
# XXX TODO: mitigate

也就是说:多个线程往同一个队列塞事件,时间戳偶尔会乱序,代码知道这件事但没有解决。这会直接影响下游的归并逻辑(见 03 章)。

4. 各个采集端分别在做什么

4.1 屏幕:最朴素的忙循环

read_screen_events(legacy/openadapt/record.py:702-733)就是 while not terminate: 截图; 塞队列。没有节流,没有帧率控制——文件里留着 # TODO: throttle 和被注释掉的 CPU/内存上限参数(:707-710)。

4.2 窗口:轮询 + 变化才入队

read_window_events(:737-787)每轮取一次活动窗口数据,只有和上一次不同才入队(:778)。这是唯一一个自带去重的读取端。

窗口数据从哪来?legacy/openadapt/window/__init__.py:12-19 按平台分派:

平台	实现文件	底层 API
macOS	`window/_macos.py`	Quartz `CGWindowListCopyWindowInfo` + ApplicationServices `AXUIElement`
Windows	`window/_windows.py`	UI Automation
Linux	`window/_linux.py`	AT-SPI

macOS 的实现里有两个值得记的细节:

不用 pywinctl。window/_macos.py:24-25 的注释直说 pywinctl 在 macOS 上「性能不可用」,并附了 issue 链接,所以改成直接调 Quartz。
无障碍树可能序列化不了。get_active_window_state(window/_macos.py:17-59)拿到整棵树后,先试着 pickle.dumps 一遍,失败就把 data 字段丢掉再返回。因为这份数据要跨进程传给写进程,不可 pickle 的对象会让整条管线炸掉。这是一条很实在的防御。

4.3 键鼠:钩子 + 「不录自己注入的事件」

三个回调 on_move/on_click/on_scroll(record.py:578、:598、:633)结构一致,都先判 if not injected。injected 是 pynput 给出的标志,表示这个事件是程序合成的而不是人按的——没有这个判断,回放时注入的动作会被自己录进去。

每个回调最终都走 trigger_action_event(:555-575),它在入队前多做一件事:

# 真实源码节选,legacy/openadapt/record.py:569-574
if x is not None and y is not None:
    if config.RECORD_READ_ACTIVE_ELEMENT_STATE:
        element_state = window.get_active_element_state(x, y)
    else:
        element_state = {}
    action_event_args["element_state"] = element_state

按坐标去问操作系统「这里是什么控件」,把结果一起存下来。这是默认关着的开关(config.RECORD_READ_ACTIVE_ELEMENT_STATE),因为很慢。

4.4 停止录制:键序列状态机

录制时鼠标键盘都被占用,得有一个不打断操作的退出方式。read_keyboard_events(:921-990)里维护了一组索引 stop_sequence_indices,每个停止序列一个:

对每个停止序列 s:
   按下的键 == s[idx] ?  ── 是 ──→ idx += 1
                         └─ 否 ──→ idx = 0(整条重来)
   idx == len(s) ? ──→ 触发停止

默认序列在 legacy/openadapt/config.py:36:SPECIAL_CHAR_STOP_SEQUENCES = [["ctrl", "ctrl", "ctrl"]],也就是连按三下 Ctrl。比对用的不是原始按键,而是 canonical key——按下时先取一次规范化结果(record.py:959),再拿它和序列里当前那个键比(:974-982),这样不同键盘布局下同一个物理键能对上。

4.5 浏览器:WebSocket 服务端

run_browser_event_server(:1214)起一个 websockets.sync.server,接收浏览器扩展推来的 DOM 事件。这些事件带自己的时间戳和 screenX/screenY,后面要专门做对齐(见 03 章第 5 节)。

5. 落地成什么:数据模型

五张主表,全部在 legacy/openadapt/models.py:

表	类	关键字段	行
`recording`	`Recording`	`task_description`、`monitor_width/height`、`double_click_interval_seconds`、`config`	`:46`
`action_event`	`ActionEvent`	`name`、`mouse_x/y`、`key_name/char/vk`、`parent_id`、`element_state`	`:129`
`window_event`	`WindowEvent`	`title`、`left/top/width/height`、`state`(整棵 a11y 树)	`:629`
`browser_event`	`BrowserEvent`	`message`(原始 JSON)	`:755`
`screenshot`	`Screenshot`	`png_data`、`png_diff_data`	`:924`

有两个建模决定值得单独说。

决定一:关联用时间戳,不只用外键

ActionEvent 同时有 screenshot_timestamp 和 screenshot_id(models.py:142-147)。录制期只有时间戳(那时截图还没写进库、没有 id),外键是后来补的。这让写入端可以完全并行——动作和截图各写各的,不用互相等 id。

决定二:录制期把关键系统参数快照下来

Recording 存了 double_click_interval_seconds 和 double_click_distance_pixels(models.py:56-57)。这两个值是录制那台机器的系统设置。为什么要存?因为「两次点击算不算双击」必须按录制时的标准判定,而不是按分析时那台机器的标准。归并逻辑会去读它(见 03 章)。

6. 关键细节与坑

录制要先抢数据库锁。 record 的第一件事就是 crud.acquire_db_lock(),失败直接返回(record.py:1297-1299)。同一时间只允许一个录制。
模块导入时就截了一张屏。 record.py:68:monitor_width, monitor_height = utils.take_screenshot().size,旁边写着 # TODO XXX replace with utils.get_monitor_dims() once fixed。这就是 01 章提到的「import 有副作用」的同类问题。
视频和图片是两条并行的路。 config.RECORD_VIDEO / RECORD_IMAGES 至少要开一个(record.py:1292-1295);RECORD_FULL_VIDEO 决定是每帧都进视频,还是只有被动作引用的帧才进(record.py:201-210 对 :255-264)。视频那条路自带一个独立写进程,只在 RECORD_VIDEO 打开时才起(record.py:1490-1507)。

7. 代码地图

主题	文件路径	符号名
录制总入口与并发编排	`legacy/openadapt/record.py`	`record`
动作/截图/窗口配对	`legacy/openadapt/record.py`	`process_events`
键鼠钩子与注入过滤	`legacy/openadapt/record.py`	`on_move`、`on_click`、`on_scroll`、`handle_key`、`trigger_action_event`
停止键序列状态机	`legacy/openadapt/record.py`	`read_keyboard_events`
各类型读取循环	`legacy/openadapt/record.py`	`read_screen_events`、`read_window_events`、`run_browser_event_server`
写进程模板	`legacy/openadapt/record.py`	`write_events`、`write_action_event`、`write_screen_event`、`write_video_event`
跨平台窗口分派	`legacy/openadapt/window/__init__.py`	`get_active_window_data`、`get_active_element_state`
macOS 无障碍树抓取	`legacy/openadapt/window/_macos.py`	`get_active_window_state`、`get_active_window_meta`、`dump_state`
数据模型	`legacy/openadapt/models.py`	`Recording`、`ActionEvent`、`WindowEvent`、`Screenshot`、`BrowserEvent`
跨进程队列	`legacy/openadapt/extensions/synchronized_queue.py`	`SynchronizedQueue`

1. 它要解决的小问题​

2. 思路:线程采集、队列汇合、进程落库​

3. 核心机制:动作和屏幕/窗口怎么配对​

直觉先行​

原理演示​

真实实现​

一个诚实的缺陷​

4. 各个采集端分别在做什么​

4.1 屏幕:最朴素的忙循环​

4.2 窗口:轮询 + 变化才入队​

4.3 键鼠:钩子 + 「不录自己注入的事件」​

4.4 停止录制:键序列状态机​

4.5 浏览器:WebSocket 服务端​

5. 落地成什么:数据模型​

决定一:关联用时间戳,不只用外键​

决定二:录制期把关键系统参数快照下来​

6. 关键细节与坑​

7. 代码地图​