playflow

简体中文 | English

YAML 驱动的 Playwright 自动化引擎。把浏览器操作写成工作流文件：

name: 我的流程
login:
  url: http://10.0.0.1/login
  username: alice
  password: "{{ env.password }}"
  success_url: http://10.0.0.1/home
env:
  password: secret
steps:
  - uses: goto
    with: { url: "http://10.0.0.1/list" }
  - uses: click_text
    with: { text: [签收, 受理], optional: true }
  - uses: pick_radio
    with: { text: 同意 }
  - uses: click_text
    with: { text: [提交, 确定] }

playflow run workflow.yaml          # 执行
playflow validate workflow.yaml     # 只校验

• 一个执行引擎：步骤 + 条件 + 循环 + 变量，页面再复杂也只改 YAML、不改代码。
• 面向真实老平台：文字匹配自动忽略空格（“提 交”→“提交”）、自动遍历所有 iframe、兼容新标签页与同页跳转、穿透开放 Shadow DOM。
• 可选能力：跨流程复用的登录态（state.json）、人工验证暂停点（UKey/短信）、任务队列循环、附件归档、截图与日志。
• 无人值守：断点续跑（--resume）、失败清单重试（--retry-failed）、运行报告（JSON/HTML）、企业微信/钉钉 webhook 通知、异常时保存 Playwright trace。
• 创作工具：probe 扫描页面给出选择器建议与 YAML 草稿；record 录制人工操作生成步骤（--verify 录完回放验证）；heal 在平台改版后检查步骤可达性并给出修复建议；--dry-run 只跑读动作做安全演练。
• 更多任务形态：from_csv / from_xlsx 把表格每行当一次任务；partials + include 复用步骤片段。
• 依赖只有 playwright + PyYAML，驱动系统已装的 Chrome，无需下载浏览器。

安装

pip install playflow                 # PyPI 安装（含 playwright + pyyaml）
# 可选能力：
pip install "playflow[http]"         # request 动作
pip install "playflow[notify]"       # webhook 通知
pip install "playflow[xlsx]"         # from_xlsx 数据源任务

本机需已安装 Chrome（默认 channel: chrome）；只有 Edge 时把 browser.channel 配成 msedge。

源码方式安装（开发）：

pip install -e .[dev]

快速开始（本地演练，不需要内网）

python tests/mock_platform.py     # 启动模拟平台（SSO + UKey + iframe 列表 + 附件 + confirm）

playflow run examples/demo.yaml   # 账号 admin/123456，UKey 环节点页面按钮后回终端按回车

跑完可在 附件/、logs/、shots/ 查看归档、日志与截图。改大 examples/demo.yaml 里的 max_tasks 可批量处理全部 10 条。

真实站点示例：examples/deepseek_usage.yaml（登录 DeepSeek 开放平台，抓当月用量/费用/余额并落盘 JSON）， examples/advanced.yaml（条件、循环、接口、断言）。

工作流结构

name: 流程名称                     # 显示用

settings:                          # 全部可省略
  max_tasks: 1                     # 每个任务处理条数（任务级 max_tasks 可覆盖）
  task_delay_seconds: [1, 3]       # 每条之间随机等待
  screenshot: true                 # 关键动作截图到 shots/
  attachment_dir: 附件              # 附件保存根目录
  dialog: accept                   # 原生弹窗：accept（默认）/ dismiss
  dialog_text: ""                  # dialog=prompt 时的输入文本
  timeout: 20000                   # 元素默认超时（毫秒）
  trace: on_error                  # off / on_error（默认，异常中止时存 trace.zip）/ always

notify:                            # 可选；跑完推送汇总到群机器人
  webhook: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx
  only_on_failure: false           # true 时只有失败/中止才推送

partials:                          # 可选；可复用的步骤片段，见「子流程复用」
  登录后处理: [...]

browser:
  channel: chrome                  # chrome / msedge
  headless: false

login:                             # 可选；不需要登录的流程整段删除
  url: http://sso/login
  username: 工号
  password: "密码"                  # 支持 {{ env.password }}
  username_selector: ""            # 可省略，自动找第一个文本框
  password_selector: ""            # 可省略
  button_text: [登录, 登 录]
  manual_pause: true               # 点登录后暂停等人处理 UKey/短信，回车继续
  pause_hint: 请在 UKey 窗口完成验证
  success_url: http://平台/home     # 登录成功校验地址（也用于登录态归属判断）
  verify_url: ""                   # 校验地址（默认取 success_url）
  # reuse_settle_ms: 3000          # 复用登录态时，到 success_url 后的观察窗口（毫秒）
  state_file: state.json           # 登录态文件，多流程可各用一份
  steps: []                        # 登录页很特殊时，用自定义步骤替代以上简写

env:                               # 自定义变量；也可用 {{ env.xxx }} 引用
  operator: 自动处理

steps:                             # 主流程：登录后、逐任务之前执行一次
  - uses: log
    with: { message: "操作人 {{ operator }}" }

tasks:                             # 任务：列表循环或单页流程
  - name: 待办列表
    url: http://平台/list
    mode: first_row                # first_row（默认）/ once / each_row
    max_tasks: 1
    row_selector: ""               # 省略时自动识别 tr/li/div 内的链接行
    link_selector: ""              # 省略时取行内第一个 <a href>
    label_regex: '[A-Za-z]{1,6}-?\d{2,}'   # 从行文本提取任务号作汇总标签
    steps: [...]
  - name: 批量补录                 # 也可以来自表格数据，见「数据源任务」
    from_csv: 待办.csv
    steps: [...]

步骤通用字段

tasks 三种模式

循环内可用变量：row_text、row_index、label（或 task_no）、category、category_url、index。

动作参考

变量

导航与页面

点击

表单

取值、接口与下载

断言与人工介入

控制流

创作工具

条件与函数

if、while、assert 支持三种写法。

字符串表达式（最常用）：

if: "body_text contains 申请"          # 变量名直接引用，字符串建议不加引号
if: "exists('#submit') or visible('.ok')"
if: "count('table tr') >= 5 and url() contains /list"
if: "title == '任务详情'"               # 带空格的字符串用引号
if: "st.status == 200"

运算符：== != > < >= <=，contains / not contains，matches（正则）， startswith / endswith，in / not in；逻辑 and / or / not 与括号。

条件函数：exists(sel[, frame])、visible、absent、has_text(text)、 count(sel[, frame])、text(sel[, frame])、attr(sel, attr[, frame])、 value(sel[, frame])、page_count()、url()、title()。

结构化映射：

if:
  and:
    - {contains: ["{{ body_text }}", "申请"]}
    - {exists: "#submit"}

变量

• env: 下的键既可直接引用 {{ operator }}，也在 {{ env.operator }} 下。
• 步骤 id: xx 把返回值存入 {{ steps.xx }} 与 {{ steps_xx }}。
• 任务循环提供 {{ row_text }} / {{ row_index }} / {{ label }} / {{ category }}。
• 整串是单个模板时保留原始类型（数字/布尔/字典），如 {{ resp.status }} == 200。

登录态与人工暂停

1. 启动先看 state.json（或 login.state_file）：文件存在不等于直接跳过——先打开 success_url，等掉 JS/延时跳转再判断，确认没有登录/验证特征才复用；cookie 已过期则直接重登。
2. 失效或站点不符则走 login 简写：自动填账密、点登录； manual_pause: true 时暂停等人处理 UKey/短信，回车后回 success_url 校验。
3. 校验通过把登录态写回 state.json（并写 .meta.json 记录站点，防止跨流程误用）。
4. 批量运行中途被踢回登录页：自动重新登录后继续当前任务。
5. 想换账号重登：删除 state.json 与 state.json.meta.json。

login.reuse_settle_ms（默认 3000）控制复用登录态时在 success_url 上的观察窗口， 慢平台/后台轮询多导致跳转更晚时可调大。

断点续跑与失败重试

批量运行时引擎自动把进度写到工作流同目录（相对路径锚定工作流文件目录）：

• <流程名>.progress.json —— 已完成的任务号（按任务名分组），每条任务结束即落盘；
• <流程名>.failed.json —— 失败任务号及原因。

playflow run workflow.yaml --resume         # 跳过 progress 里已完成的任务，从中断处继续
playflow run workflow.yaml --retry-failed   # 只补跑 failed 清单里的任务，其余跳过

说明：first_row 队列里处理成功的行会从列表消失，因此 --resume 天然从第一行未完成的任务继续； 跳过的行用行游标越过，不会重复处理。--retry-failed 对两类任务名分别生效：清单里没有的类别整类跳过。

dry-run：只读演练

playflow run workflow.yaml --dry-run

每个内置动作标注了读/写属性（goto/extract/断言 等是读，click/fill/upload 等是写， request 按 HTTP 方法区分）。dry-run 只执行读动作，写动作跳过并记录 [dry-run] 跳过， 任务之间的随机等待归零、每个列表任务只处理 1 条，用于平台改版后安全验证流程结构（登录仍会真实执行）。

heal：改版后自检与修复建议

playflow heal workflow.yaml --url http://平台/list --fix
playflow heal workflow.yaml --headed --min-score 0.6

登录后打开 --url 页面，逐个检查步骤的 selector / 文字参数能否命中当前页面； 未命中的给出按相似度排序的候选（按钮文字 / id / name / placeholder）。--fix 不改原文件， 把高于阈值（默认 0.55）的建议写入 workflow.healed.yaml，人工核对后再用。 注意 heal 只在“登录后 + --url 这一个页面”上检查，流程中途才出现的元素要分页多次检查。

运行报告与 trace

每次运行结束（含异常中止）自动在 reports/ 写入 run_时间.json 与 run_时间.html （任务清单、成功/失败/跳过、耗时；HTML 单文件可直接发群）。 运行异常中止时，settings.trace: on_error（默认）会把 Playwright trace 存到 shots/trace_中止_*.zip，用 playwright show-trace <文件> 回放操作时间线； always 每次都存，off 关闭。

通知（企业微信 / 钉钉 / 通用 webhook）

notify:
  webhook: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx   # 或 URL 列表
  only_on_failure: false      # true 时只有失败/中止才推送

按域名自动识别消息格式（企业微信 / 钉钉 markdown，其余发 {title, text} JSON）。 需要 pip install "playflow[notify]"；通知失败只记日志，不影响运行结果。

数据源任务（from_csv / from_xlsx）

tasks:
  - name: 批量补录
    from_csv: 待办.csv             # 或 from_xlsx: 待办.xlsx（可加 sheet: 工作表1）
    encoding: gbk                  # 仅 csv；默认 utf-8-sig
    label_column: 单号             # 可选；默认取第一列
    max_tasks: 99
    url: "http://平台/detail?no={{ row.单号 }}"   # 可选；每行打开的页面
    steps:
      - uses: fill
        with: { selector: "#city", text: "{{ row.城市 }}" }

每一行数据执行一次 steps；列名直接作为变量（{{ 单号 }}），整行在 {{ row }} 下， {{ label }} / {{ row_text }} / {{ row_index }} 照常可用。数据源任务不走 mode， 断点续跑/失败重试同样生效。xlsx 需要 pip install "playflow[xlsx]"。

子流程复用（partials / include）

partials:
  归档:                                   # 值是步骤列表
    - uses: download
      with: { optional: true }
    - uses: close_task_page
  公共登录后: { file: common.yaml }        # 或从其它 YAML 文件加载（文件含步骤列表或 {名称: 步骤} 映射）

steps:
  - include: 公共登录后
tasks:
  - name: 待办
    url: http://平台/list
    steps:
      - uses: click_row_link
      - include: 归档

include 可嵌套（有循环引用检测），可在 steps / login.steps / tasks[].steps 及 then/else/do 子步骤中使用；playflow validate 会检查引用是否完整。

命令行

playflow run workflow.yaml [--headed|--headless] [--channel chrome]
playflow run workflow.yaml --dry-run         # 只执行读动作，写动作跳过
playflow run workflow.yaml --resume          # 断点续跑
playflow run workflow.yaml --retry-failed    # 只补跑上次失败的任务
playflow run workflow.yaml --validate        # 只校验
playflow validate workflow.yaml
playflow probe [--workflow workflow.yaml] [--url 网址]
playflow record out.yaml [--workflow workflow.yaml] [--url 网址] [--verify]
playflow heal workflow.yaml [--url 网址] [--fix] [--headed] [--min-score 0.55]
python -m playflow                           # 等价；不带子命令进交互菜单

Python API

from playflow import run_workflow, WorkflowEngine, action

summary = run_workflow("workflow.yaml")
summary = run_workflow(workflow=my_dict, base_dir=".", headless=True)
summary = run_workflow("workflow.yaml", dry_run=True)     # 只读演练
summary = run_workflow("workflow.yaml", resume=True)      # 断点续跑
summary = run_workflow("workflow.yaml", retry_failed=True)

# 自定义动作：函数签名 (engine, params, step)；@action 默认 mode="write"（dry-run 时跳过）
@action("我的动作")
def my_action(engine, params, step):
    engine.get_text(params["selector"])
    engine.logf("自定义动作执行完成", echo=True)

@action("只抓取", mode="read")     # dry-run 下仍会执行
def my_reader(engine, params, step):
    return engine.get_text(params["selector"])

WorkflowEngine 提供 current（当前 Page）、ctx（BrowserContext）、 vars、settings、locator()、frames_of()、selector_exists() 等；注册动作后 即可在 YAML 中 uses: 我的动作。

库内输出走标准 logging（logger 名 playflow）：默认自动挂一个控制台 handler 保证 脚本直跑能看到进度；用 playflow.disable_console_logging() 关掉，或自行配置 handler 后 playflow 不会再附加。文件日志（logs/）不受影响。

产物与目录

相对路径（登录态、附件、上传文件等）都锚定到工作流文件所在目录。

打包 exe / 内网离线

pip install pyinstaller
pyinstaller -F -n playflow --collect-all playwright run_playflow.py

产物 dist\playflow.exe，与 workflow.yaml 放同一目录双击即可。 离线环境用 PyPI 包制作离线 wheels（详见 RELEASING.md）：

pip download playflow -d wheels                                # 外网机
pip install --no-index --find-links=wheels playflow            # 内网机

引擎通过 channel 驱动系统 Chrome，无需 playwright install 下载浏览器。

AI 使用与声明

• 🤖 给 AI 的使用指南：如果你是 AI 代理/编码助手，请先阅读 AI 使用指南—— 安装调用、YAML 书写规则、动作速查、校验与 dry-run 流程一页速览。
• 📝 AI 辅助开发声明：本项目由 @timmycheng 提供需求与验收， 代码/测试/文档/CI 由 ZCode（CLI 编码代理）+ GLM（Z.ai 训练）辅助生成， 以 pytest/selftest/ruff 与 CI 自动验收为合并门槛。详见 docs/ai.md。

开发与测试

pip install -e .[dev]
pytest                        # 单元测试 60+ 项，无需浏览器
python tests/selftest.py      # 端到端验收 60+ 项：自动拉起 mock 平台，需本机 Chrome
ruff check .                  # 代码检查

推送后 GitHub Actions 自动跑 ruff + 单测（Ubuntu/Windows）+ 构建检查 + 端到端。 发布到 PyPI 的步骤见 RELEASING.md，变更记录见 CHANGELOG.md。

目录结构

playflow/            引擎包
  engine.py          工作流引擎：步骤执行、条件、tasks/数据源循环、断点续跑、partials、校验
  actions.py         45 个内置动作（带读/写标注）
  conditions.py      条件表达式求值
  template.py        {{ }} 变量模板
  browser.py         启动/登录/登录态/人工暂停
  dom.py             文字匹配、列表行、单选框、附件下载
  probe.py           页面探测
  recorder.py        操作录制（含回放验证注释）
  heal.py            步骤可达性检查与修复建议
  report.py          运行报告（JSON/HTML）
  notify.py          企业微信/钉钉/通用 webhook 通知
  cli.py             命令行入口（run/validate/probe/record/heal）
examples/            demo（mock 平台）、advanced（高级语法）、deepseek_usage（真实站点）
tests/
  unit/              pytest 单元测试（无浏览器）
  mock_platform.py   本地模拟平台（演练用）
  selftest.py        端到端验收（自动拉起 mock 平台）