• 工作流自动化
• 任务调度
• 桌面应用

AUTO-MAS 是一个多脚本多配置统一管理与自动化工具，而log_box正是它背后的日志采集管线：自动 tail 脚本日志、按规则匹配提取关键信息、清洗过滤后再推送到通知渠道。这篇指南带你在 10 分钟看懂「采集 → 前置处理 → 规则匹配 → 后置处理 → 推送」的完整流程，无需阅读源码也能掌握它的每一步。

1 分钟认识 log_box：它负责什么、不负责什么

log_box是 AUTO-MAS 中与具体游戏脚本解耦的通用日志组件（位于 app/log_box/）。它的定位可以用一句话概括：

调用方只提供「日志源 + 规则 + 处理器」，剩下的采集、匹配、清洗、推送全部由 log_box 在内部完成。

它不关心你跑的是哪个脚本，只关心日志本身；而采集结果的最终落点（推送报告、调度台）则由各个专项（OK-WW、通用脚本等）决定。

整个管线可以抽象成一条 5 步流水线：

日志文件 ──> ① 采集(LogSource) ──> ② 前置处理(hooks/翻译) ──> ③ 规则匹配($()提取) ──> ④ 后置处理(去重/规整) ──> ⑤ 推送(sink 或 @@LOGBOX@@ 标记)

第一步：日志采集——tail 读取与轮转补偿

采集由 app/log_box/sources.py 中的LogSource完成，它是典型的主动拉取式 tail：

• 只读增量：open()时记录起始位置，默认start_from_end=True，即只采集会话开启之后新增的内容；
• 按完整行读取：未闭合的半行会留到下次读取，不会把一行日志切成两段；
• 轮转补偿（亮点）：日志滚动时文件身份（inode）会变化，log_box 会按 inode 在同目录找回被重命名的旧日志，从原位置续读，避免轮转前的内容静默丢失；在不提供 inode 的文件系统（如 FAT32）上，则按rotated_name模板（如ok-script.%Y-%m-%d.log）或.bak约定探测；
• 截断处理：文件变小则重置到文件头重读。

这意味着无论脚本把日志滚动成日期文件还是直接清空重建，采集都不会漏、不会错。

第二步：前置处理——hooks 规则先于一切

日志进入规则匹配之前，会先经过 app/log_box/hooks.py 的钩子层，支持两类规则：

规则以 JSON 数组形式配置，例如：

[{"type": "drop", "match": "heartbeat"}, {"type": "replace", "match": "IP地址", "replace": "***"}]

关键语义：被丢弃的行不进入任务日志、推送报告与成功/失败判定；改写后的文本才是下游匹配的依据。配置drop时请务必避开成功/失败标志所在的行，否则会误判任务状态。

此外，前置处理还常用于日志翻译——如 OK-WW 适配器用.po词条把英文日志翻成中文后再匹配，规则只需按中文书写（见 app/task/Okww/AutoProxy.py）。

第三步：规则匹配——正则 +$()表达式提取

匹配与提取能力来自 app/utils/LogPatternExtractor.py，每条采集规则由三部分组成：匹配正则 + 提取表达式 + 日志类型。

• 匹配正则为空则该规则不生效；正则非法会fail-fast直接报错，尽早暴露配置错误；
• 提取表达式使用$()语法，支持+同行拼接、;换行拼接，以及cut/get/sub/upper/lower等函数链，可从一行日志中精确抠出想要的字段；
• 除单行规则外还支持多行聚合（multiline）：由起始/结束正则划定窗口，从跨行内容中提取并拼接；
• 每条规则命中即产出(日志类型, 文本, 时间戳)三元组，时间戳保留「规则命中的那一刻」，供推送报告加时间前缀。

日志类型只有两种（见 app/log_box/logtype.py）：

• 普通（NORMAL）：任何推送报告都会包含；
• 失败（FAIL）：仅当任务中存在未完成用户时才纳入报告。

第四步：结果推送——两种宿主、两条通道

log_box 的结果落点由「宿主」决定，这是它最巧妙的设计（见 app/log_box/collect.py 与 app/log_box/markers.py）：

① MAS 进程宿主（注入 sink）

运行在 MAS 主进程内的专项（如 OK-WW）通过log_box.get_collect(paths=..., sink=...)创建采集会话，命中结果直接经 sink 回调写入当前用户的push_log，零 IPC 开销。

② 脚本子进程宿主（@@LOGBOX@@标记回传）

通用脚本以子进程运行，无法直接写 MAS 内存。此时 log_box 把结果渲染为受控的 stdout 标记行回传：

@@LOGBOX@@{"op":"push","type":"普通","text":"✅ 成功: 节点A"} @@LOGBOX@@{"op":"flush"}

MAS 侧在check_log回调中单行嗅探该前缀（见 app/task/general/AutoProxy.py），识别后写入用户推送日志；非标记行则按普通日志处理。子进程正常退出时由atexit自动冲刷残留并完成最终推送，保证收尾不遗漏。

第五步：采集结果如何变成推送报告

进入push_log的条目最终由 app/tools/push_log.py 按用户的节点详情推送模式渲染：

• 关闭：不输出节点详情，只有结果行；
• 逐条：每条独占一行，带HH:MM采集时间前缀；
• 汇总：按状态（成功/跳过/失败）聚合成一行，多账号自动分块。

报告还会镜像到调度台日志，未配置推送的用户同样能在调度中心看到完整的节点详情。

如何接入 log_box：一个最小示例

各专项都是「喂参数方」——只提供路径、规则与处理器，核心逻辑全部复用。以 MaaEnd 为例（app/task/MaaEnd/push_log.py），接入四步：

1. 创建会话：log_box.get_collect(paths=[日志路径], sink=回调, rotated_name=轮转模板)；
2. 挂前置处理：log_collect.open(处理器函数)，如日志翻译；
3. 声明规则：log_collect.collect(匹配正则, 提取表达式, 日志类型)，可注册多条；
4. 收尾推送：任务结束时调用close()，自动冲刷多行残留、应用后置处理器并完成推送（open()/close()均幂等，可安全重复调用）。

工厂入口只有一个：from app.log_box import log_box（见 app/log_box/factory.py）。路径参数为None时会回退读取环境变量MAS_SCRIPT_LOG_PATH，方便脚本子进程宿主使用。

小结：为什么这条管线值得学

• 职责单一：只管日志，与具体脚本解耦，新增专项零重复代码；
• 不漏日志：inode 轮转找回 + 截断重读 + 完整行读取三重保险；
• 先清洗后匹配：hooks 丢弃/改写先于判定，噪声不干扰成功/失败标志；
• 双宿主通道：进程内 sink 直写、子进程@@LOGBOX@@标记嗅探，一条管线通吃两种运行形态。

想深入细节，可以直接阅读 app/log_box/init.py 的模块说明，或运行一次带推送日志配置的任务，在调度台观察逐条/汇总两种模式的渲染效果——这就是整条管线的「用户视角」验证方式。

• 工作流自动化
• 任务调度
• 桌面应用