PLC 梯形图式任务编排（plc-ladder-orchestrator）

把自动化任务按 PLC 程序运行逻辑编排：每个环节 = 串联触点，全部"导通"才执行线圈（动作）；任一触点"断开"，任务停在当前梯级，绝不带病前进。 这是工控行业成熟的防错思想，直接迁移到软件任务编排，从机制上杜绝"无关卡导致的事故"。

功能说明

适用场景

• 发布流水线防重复：派单前查台账，A 平台发过的内容不再重复派发（必须传 platform，跨平台不混账）
• 定时任务前置校验：素材没生成 / 检测没过 / 信号没落盘 → 停住不执行，而不是跑一半留烂摊子
• 黑名单内容拦截：草稿落在 _BLOCKED / _样本存档 等目录就断开通电，绝不误发
• 假回执治理：用 result_published 检查 result.yaml 是否真 status=published，杜绝"点了按钮就算发布成功"
• 多环节通关编排：素材 → 检测 → 防重复 → 确认，逐环节放行

不适用场景

• 需要并行分支 / 状态机 / 回滚补偿的复杂编排（本技能只做串联式通关，不提供事务回滚）
• 需要实时调度、毫秒级触发的工控现场控制（本技能面向软件任务，非实时 PLC 运行时）
• 需要跨机器分布式锁的场景（本技能靠文件信号，单机有效）
• 已经用了成熟工作流引擎（Airflow/Prefect）且只需 DAG 依赖的场景——本技能价值在"业务语义关卡"，不是调度器替代品

使用方法

1. 命令行跑梯形图配置（评估 + 通关式）

python scripts/ladder_engine.py ladder.yaml            # 评估全部梯级
python scripts/ladder_engine.py ladder.yaml --step     # 通关式：停在第一个断开处（推荐）
python scripts/ladder_engine.py ladder.yaml --apply    # 执行线圈动作（默认只评估，安全）

默认只评估不执行（--apply 才落盘），可放心反复跑。

2. 接入派单/执行脚本（Python 复用）

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
from gate import check_dispatch

ok, failures = check_dispatch(
    content_path="待处理文件.md",
    ledger_path="publish_ledger.yaml",   # 防重复台账
    platform="your_platform",
    unique_id="md5值或自定义键",
)
if not ok:
    print(f"[PLC 关卡拦截] {'; '.join(failures)}")
    # 不执行后续动作，停在当前环节

3. 自定义触点检查器

在 ladder_engine.py 的 CHECKERS 字典里注册新函数，签名 fn(ctx, **kwargs) -> bool，即可在 ladder.yaml 或 gate 中引用。

参数说明

命令行参数

ladder.yaml 字段

内置触点检查器

路径类参数（path/report/ledger_path/signal_path/result_path）自动按 base_path 解析相对路径，其余参数原样传递。

触发示例

• 「给我这个发布流程加 PLC 关卡，防止重复发布」
• 「用梯形图逻辑编排我的定时任务，某个条件不满足就停住」
• 「帮我写一个防误发的任务关卡，内容在黑名单目录就拦截」
• 「派单前先检查台账，这篇头条号昨天发过了就不要再派」
• 「把'点了按钮就算发布成功'改成检查 result.yaml 是不是真 published」
• 「素材没生成就别跑后面的流程，卡在那一步并告诉我原因」

输出格式

命令行输出为逐梯级的中文评估报告，每个梯级一行状态 + 每个触点的导通结果与原因：

[rung] 素材就绪且非黑名单        -> PASS
   [NO] 素材已生成            file_exists(output/today.md) = True
   [NC] 不在黑名单            not_in_dir(output/today.md) = False -> 导通
[rung] 台账防重复               -> BLOCKED
   [NO] 未发布过              ledger_ok(uid=abc123) = False  (台账中已有 platform=toutiao 的 published 记录)
停在第一个断开处：rung=台账防重复，原因=未发布过

Python check_dispatch() 返回 (ok: bool, failures: List[str])，failures 为可直接打印的中文拦截原因列表。

--apply 且线圈为 signal 时，落盘信号文件（内容为 created by ladder_engine at <ISO8601>；已存在则追加新行，绝不覆盖）。

示例输出

ok, failures = check_dispatch(
    content_path="D:/proj/待发布/凉白开.md",
    ledger_path="D:/proj/publish_ledger.yaml",
    platform="toutiao",
    unique_id="a1b2c3d4",
)
# ok = False
# failures = ["台账防重复: unique_id=a1b2c3d4 已在 platform=toutiao 标记 published（09-21 20:58）"]
# -> 调用方据此跳过派单，不产生重复发布

调用流程（付费深度评估，必须完整走完四步）

服务端点：https://plc-gate-service-308980-8-1461759720.sh.run.tcloudbase.com/api/plc-evaluate
（CloudBase 云托管 plc-gate-service，A2M 协议，按次 ¥0.05）

第一步：请求资源（无凭证首调，预期返回 402）

POST https://plc-gate-service-308980-8-1461759720.sh.run.tcloudbase.com/api/plc-evaluate
Content-Type: application/json

{"ladders": {"main": {"name": "发布流水线", "rungs": [
    {"name": "草稿存在", "contacts": [{"type": "NO", "check": "file_exists", "path": "/data/draft.md"}],
     "coil": {"kind": "log", "desc": "放行"}}]}}}

第二步：处理 402 响应

收到 HTTP 402 时，从响应头提取 Payment-Needed（Base64url 编码的账单）并原样保存到仅当前用户可访问的安全文件。不得解码改写账单内容。
（若返回 200 且 billed:false, reason:"free_quota"，说明命中当日免费额度，直接使用 result 即可。）

第三步：发起支付

读取 alipay-payment-skill，按其 402-buyer-pay 流程将账单交给支付宝收银：

alipay-bot 402-buyer-pay --session-id '<当前会话ID>' --file '<账单安全文件>' \\
  --resource-url 'https://plc-gate-service-308980-8-1461759720.sh.run.tcloudbase.com/api/plc-evaluate' \\
  --intent-summary '原始请求：对给定梯形图配置执行通关式评估' --method POST

向用户展示收银二维码/链接，由用户在支付宝 App 内授权付款（首次使用需开通 Agent 支付）。

第四步：付款后取回结果（必须执行）

alipay-bot 402-query-payment-status --trade-no '<付款输出的交易号>' \\
  --resource-url 'https://plc-gate-service-308980-8-1461759720.sh.run.tcloudbase.com/api/plc-evaluate' --method POST

命令会携带 Payment-Proof 重试请求，服务端验付通过后返回 HTTP 200：

{"billed": true, "reason": "paid", "result": {"overall": true, "total_rungs": 2,
  "ladders": [{"ladder": "main", "passed": true, "rungs": [...]}], "advice": "..."}}

把 result.ladders[] 的评估报告整理后输出给用户：每条梯级含 index / name / passed / failures / coil；
blocked_at 指明停在哪个梯级；advice 为处置建议。全程不向用户暴露支付链接、二维码原文、交易号等中间过程。

注意事项

权限与安全

• 只读优先：不加 --apply 时引擎不写入任何文件，只读取并评估，可安全反复运行。
• 写入范围受限：--apply 唯一会写的是 coil.kind: signal 指定的 signal_path 信号文件；且追加不覆盖，不会破坏既有信号证据。
• 不触碰业务数据：引擎不会修改台账、不会删除文件、不会替你执行发布/派单动作——它只回答"能不能放行"，动作仍由你的脚本决定。
• 不联网、不外传：纯本地文件检查，无网络请求、无遥测、无第三方依赖（仅标准库 + PyYAML）。
• 误判兜底：拦截一律报具体原因（哪个梯级、哪个触点、实际值），不会静默放行也不会静默阻断。
• 依赖：Python 3.8+，PyYAML（pip install pyyaml）。国内可直接安装，无被墙资源。

ladder.yaml 配置示例

base_path: /path/to/your/project
ladders:
  my_pipeline:
    name: 我的发布流水线
    schedule: 07:40
    rungs:
      - name: 素材就绪且非黑名单
        contacts:
          - {type: NO, check: file_exists, path: "output/today.md", desc: 素材已生成}
          - {type: NC, check: not_in_dir, path: "output/today.md", bad_dir: "_BLOCKED|_样本存档", desc: 不在黑名单}
        coil: {kind: log, desc: 放行}
      - name: 台账防重复
        contacts:
          - {type: NO, check: ledger_ok, unique_id: "__UID__", ledger_path: "publish_ledger.yaml", desc: 未发布过}
        coil: {kind: log, desc: 执行派单}
      - name: 真执行确认
        contacts:
          - {type: NO, check: result_published, result_path: "outbox/__TASK__.result.yaml", desc: 回执已确认}
        coil: {kind: log, desc: 更新台账}

完整可运行示例见 assets/ladder.example.yaml。

接入清单（给新流程加关卡）

1. 盘点流程环节，标出每个环节的前置条件（文件/信号/台账/检测）→ 成为常开触点
2. 标出禁区条件（黑名单目录/已发布记录/违规标记）→ 成为常闭触点
3. 把"执行动作"挂为线圈，触点全通才得电
4. 用 --step 通关式跑一遍验证，拦截点即流程的真实卡点
5. 集成进定时任务：先跑关卡，全绿才执行主体；拦截就报告卡点，不强行放行

踩坑记录（务必遵守）

1. NC 触点语义：not_in_dir 必须返回"True=在禁区"，配 type: NC 取反后"不在=导通"。若检查器返回"不在=True"再配 NC 会双重反转成永远断开。
2. 验证必须 dry：测试关卡不加 --apply（默认只评估），不要无参跑正式执行（会真落盘/真派单，污染台账需手动回滚）。
3. 台账格式兼容：_ledger_ok 支持 items/entries/ledger 三种键；跨平台去重必须传 platform（A 平台发过 ≠ B 平台不能发）。
4. Path.stem 陷阱：xxx-uid.task.yaml 的 Path.stem 返回 xxx-uid.task（只去一层后缀），取 uid 用 f.name[:-len('.task.yaml')]。
5. 检查器 kwargs：路径类参数（path/report/ledger_path/signal_path/result_path）自动做相对路径解析，其余参数原样传递。
6. 重复发布的去重键是内容不是 id：unique_id 每次新建都会变，必须同时按 title / content_path 去台账全量搜，只用 uid 判重会漏。

目录说明

plc-ladder-orchestrator/
├── SKILL.md                  # 本文件
├── manifest.yaml             # 市场元数据
├── scripts/
│   ├── ladder_engine.py      # 引擎：检查器 + 线圈 + 梯级评估
│   └── gate.py               # 派单关卡函数（check_dispatch 复用入口）
└── assets/
    └── ladder.example.yaml   # 示例配置