任务自动化编排·免费版
红线
- 先算收益再动手：一年做三次的事，不值得花两小时写脚本。
- 破坏性操作必须有干跑（dry-run）：删除、覆盖、批量发布前先看会做什么。
- 幂等优先：同一个脚本跑两次，结果应该一样，不能产生重复数据。
- 先在安全环境试：不要拿生产数据跑第一遍。
- 不引用任何付费端点/专有实现。
何时使用
- 同一个操作重复做了三次以上。
- 需要批量处理大量同类任务。
- 流程有明确步骤，但人工执行易出错或耗时长。
- 需要定时/周期执行某件事。
不要使用：
- 一年只做一两次的事。
- 流程本身还没稳定（先把流程定下来再自动化）。
- 需要大量主观判断的环节（这类应该保留人工）。
职责边界
- 做：收益评估、流程拆解、幂等设计、脚本骨架、调度方案、风险控制。
- 不做：不替用户决定业务规则；不在未经确认下执行破坏性操作。
主流程
第 1 步：收益评估（质检闸，先算账）
单次手动耗时 × 年执行次数 = 年耗时（A）
写脚本 + 调试 + 维护耗时 = 投入（B）
判断：
| 情况 | 结论 |
|---|---|
| A 远大于 B（如 A=20h，B=2h） | 值得自动化 |
| A 与 B 接近 | 不值得，或只做半自动 |
| A 小于 B | 不要自动化，写个操作清单就够 |
另外三条硬判断：
- 流程稳定吗？ 流程还在变 → 先别自动化
- 出错代价高吗？ 不可逆操作 → 必须干跑 + 人工确认
- 有边界情况吗？ 边界多到写不完 → 保留人工环节
> 质检闸：三项中有两项不满足，就只做「半自动」（脚本生成待办清单，人工逐条确认）。
第 2 步：流程拆解
把流程拆成「步骤 + 每步的输入输出 + 失败处理」：
| 步骤 | 输入 | 输出 | 失败怎么办 |
|---|---|---|---|
| 1. 获取列表 | 目录/API | 待处理清单 | 重试 3 次 |
| 2. 逐项处理 | 单项 | 处理结果 | 记录失败项，跳过继续 |
| 3. 汇总 | 全部结果 | 报告 | — |
关键：每步都要能单独重跑。
第 3 步：幂等设计（核心）
让脚本可以安全地重复执行：
| 手段 | 做法 | 适用 |
|---|---|---|
| 状态标记 | 处理过的项打标记，跳过已处理 | 批量处理 |
| 输出即证据 | 先检查产物是否存在，存在就跳过 | 文件生成类 |
| 唯一键 | 用业务唯一键去重，避免重复插入 | 数据写入类 |
| 事务/原子写 | 先写临时文件再原子替换 | 文件覆盖类 |
反例（不幂等）：
# 错：每跑一次追加一份，越跑越多
with open("log.txt","a") as f: f.write(item)
正例（幂等）：
# 对：先判断，已存在则跳过
if not os.path.exists(out_path):
process(item, out_path)
第 4 步：脚本骨架
#!/usr/bin/env python3
"""任务自动化骨架：幂等 + 可续跑 + 干跑 + 报告"""
import os, sys, json, time
DRY_RUN = "--dry-run" in sys.argv # 干跑：只打印将做什么
STATE_FILE = ".state.json" # 断点续跑状态
def load_state():
return json.load(open(STATE_FILE)) if os.path.exists(STATE_FILE) else {"done": {}}
def save_state(s):
tmp = STATE_FILE + ".tmp"
json.dump(s, open(tmp, "w"), ensure_ascii=False)
os.replace(tmp, STATE_FILE) # 原子写，避免写一半中断
def main():
state = load_state()
items = get_items() # 第 1 步：获取待处理清单
ok = fail = skip = 0
for it in items:
key = item_key(it)
if key in state["done"]: # 幂等：已处理跳过
skip += 1; continue
if DRY_RUN:
print(f"[DRY] would process: {key}"); continue
try:
result = process_one(it)
state["done"][key] = {"at": time.time(), "result": result}
save_state(state) # 每步存盘，中断可续跑
ok += 1
except Exception as e:
print(f"[FAIL] {key}: {e}")
fail += 1
# 失败不中断，继续下一项
print(f"\n完成: 成功{ok} 失败{fail} 跳过{skip}")
if fail: sys.exit(1)
if __name__ == "__main__":
main()
四个必备特性：干跑 / 幂等 / 断点续跑 / 失败不中断且汇总报告。
第 5 步：外部调用的节流与重试
调用外部服务（API、发布接口、网络请求）时：
| 机制 | 做法 |
|---|---|
| 间隔 | 每次调用间留 sleep(N)，避免触发限流 |
| 指数退避 | 失败后等待 1s → 2s → 4s 再重试 |
| 上限 | 最多重试 3 次，超过则记录失败 |
| 分类处理 | 「限流」可重试；「参数错误/冲突」不重试，直接报 |
for attempt in range(3):
try:
return call_api(item)
except RateLimited:
time.sleep(2 ** attempt) # 1, 2, 4 秒
except Conflict as e:
log(f"冲突跳过: {e}"); return None # 不可重试
log(f"最终失败: {item}")
第 6 步：调度方案
| 频率 | 方案 |
|---|---|
| 一次性 | 手动跑，或挂后台任务 |
| 每小时/每天 | 系统定时任务（cron / 任务计划程序） |
| 事件触发 | 文件监听 / webhook |
调度注意：
- 脚本要能在无人值守下运行（不依赖交互输入）
- 输出日志到文件，便于事后排查
- 加超时保护，避免挂死
第 7 步：风险控制（质检闸）
首次运行前必须确认：
- [ ] 有干跑模式，且已跑过一遍看输出是否正确
- [ ] 在测试数据/副本上试跑过
- [ ] 破坏性操作（删/覆盖/发布）有确认或可回滚
- [ ] 有状态文件，中断后能续跑
- [ ] 失败会记录并汇总，不会静默吞掉
- [ ] 有超时与重试上限
输出规范
- 先给收益评估结论（值不值得做），再给方案。
- 脚本要能直接跑，不要只给伪代码。
- 必须在脚本里体现干跑、幂等、续跑、报告四件事。
- 涉及外部调用必须带节流与重试。
失败降级
- 流程不稳定：只自动化其中稳定的子步骤，其余保留人工。
- 边界情况多：脚本生成「待确认清单」，人工确认后批量执行（半自动）。
- 出错代价高：强制干跑 + 单次确认，不做全自动。
依赖
无强制依赖。可用任意通用脚本语言实现；示例基于 Python 标准库，无需安装任何包。不引用任何付费服务。
致谢
方法论为原创整理，融合了公开的软件工程实践（幂等性、指数退避、原子写、干跑模式）。本技能为免费版，MIT 许可，不引用任何付费端点。
