自动化：按日期生成可重放报告

本篇以日期报告任务为例；前置是批次日期、专用 state 目录与执行身份。源码与输入与下文配置共用同一份文件，示例未执行。

教学版本族：Python 3.12 标准库；cron 常见五字段模型、launchd 与 Windows Task Scheduler 2.0 仅讲调度语义，不注册或运行任务。核对日期2026-10-17，不称为2026最新版。示例未执行，配置与命令需按目标环境验收。

完整日期报告项目

自动化日期报告示例未执行。Python3.12标准库实现，输入是一个合法ISO日期，输出为专用 state 子目录的 report-日期.json；没有HTTP、服务或外部命令。报告内容固定、不包含实时时间，确保重跑可以比较。参数错误退出2，锁冲突3，已有内容冲突4，文件系统失败1。

完整源码与配置

自动化日期报告完整文件含 job.py、task.json、requirements.txt 与 README。配置是明确的版本1、相对目录state，脚本不接受可配置任意路径；JSON读取失败停止。持锁后先核对目标，再在同目录固定临时文件写入、flush并fsync，最后 os.replace 发布；这提供正常本地文件系统下的名称替换边界，不承诺断电持久性或网络文件系统一致性。

python

"""Deterministic per-date report. Documentation specimen; not executed."""
import argparse
import datetime
import json
import os
from pathlib import Path
import stat
import sys


def reject_links(path):
    """Reject existing symlinks and Windows reparse points, including parents."""
    for part in (path, *path.parents):
        try:
            info = part.lstat()
        except FileNotFoundError:
            continue
        if stat.S_ISLNK(info.st_mode) or getattr(info, "st_file_attributes", 0) & 0x400:
            raise ValueError("link or reparse point not supported")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("date", help="explicit ISO date, YYYY-MM-DD")
    args = parser.parse_args()
    try:
        day = datetime.date.fromisoformat(args.date)
        if day.isoformat() != args.date:
            raise ValueError("date must use YYYY-MM-DD")
    except ValueError:
        print("invalid ISO date", file=sys.stderr)
        return 2
    # Absolute lexical path before resolving, so parent links can be rejected.
    base = Path(os.path.abspath(__file__)).parent
    state = base / "state"
    lock = state / ".lock"
    pending = state / ".pending"
    target = state / ("report-" + day.isoformat() + ".json")
    owned_lock = False
    try:
        reject_links(base)
        config = json.loads((base / "task.json").read_text(encoding="utf-8"))
        if config != {"schemaVersion": 1, "stateDirectory": "state"}:
            raise ValueError("unsupported task configuration")
        reject_links(state)
        state.mkdir(mode=0o700, exist_ok=True)
        reject_links(lock)
        try:
            lock.mkdir(mode=0o700)
            owned_lock = True
        except FileExistsError:
            print("busy or stale lock; manual inspection required", file=sys.stderr)
            return 3
        reject_links(target)
        reject_links(pending)
        content = json.dumps({"batchDate": day.isoformat(), "schemaVersion": 1},
                             sort_keys=True, separators=(",", ":")) + "\n"
        if target.exists():
            if not target.is_file():
                raise ValueError("target must be a regular file")
            if target.read_text(encoding="utf-8") == content:
                print("unchanged " + day.isoformat())
                return 0
            print("conflicting report; preserved", file=sys.stderr)
            return 4
        if pending.exists() and not pending.is_file():
            raise ValueError("pending must be a regular file")
        with pending.open("w", encoding="utf-8", newline="\n") as stream:
            stream.write(content)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(pending, target)
        print("created " + day.isoformat())
        return 0
    except (OSError, ValueError) as error:
        print("report failed: " + type(error).__name__, file=sys.stderr)
        return 1
    finally:
        if owned_lock:
            try:
                lock.rmdir()
            except OSError:
                print("could not release lock; inspect before retry", file=sys.stderr)


if __name__ == "__main__":
    sys.exit(main())

json

{
  "schemaVersion": 1,
  "stateDirectory": "state"
}

text

# Python 3.12 standard library only; no pip install required.
# Local private stable directory; no network filesystem or shared account guarantee.

完整项目位于仓库 examples/m5-next-automation/；README给出全部未来自跑命令与精确清理步骤，正文包含全部源码和输入。

markdown

# 日期报告与幂等重放（未执行）

完整文件：job.py、task.json、requirements.txt、本说明。已有Python3.12标准库，零安装、零网络、零服务、零自动注册调度器。运行在本例的私有稳定目录，所有父目录不可由不可信用户改写；不用于共享账户、特权任务、网络文件系统或多主机并发。

输入为明确ISO日期，不猜测时区和当前业务日。配置必须完全等于版本1和stateDirectory=state，不接受任意输出路径。单进程mkdir锁覆盖合作调用者，非安全隔离边界；链接/Windows重解析点检查不能防御同身份恶意替换；硬链接也应由私有目录前置条件排除。目录mode不配置Windows ACL，读者须预先核对ACL。

未来隔离自跑（本轮不执行）：

```text
python job.py 2026-10-17
python job.py 2026-10-17
python job.py 2026-02-30
```

首次预期 `created 2026-10-17`，第二次 `unchanged 2026-10-17`，均退出0，最终文件为state/report-2026-10-17.json：

```json
{"batchDate":"2026-10-17","schemaVersion":1}
```

非法日期退出2，尚未创建state。若先手动在专用state内建立空.lock目录，下次退出3；若改动已有报告内容，下次退出4且保留文件。I/O、配置或链接错误退出1，不输出created。普通异常在finally释放自己获取的锁；强杀/掉电可能留锁。释放锁失败会打印摘要，成功报告已提交时仍返回0，调用者必须检查锁遗留告警。

本地同目录replace只作为名称发布边界，不保证目录fsync、断电持久性或跨文件事务。临时文件已fsync，但Windows/POSIX与网络文件系统行为需未来实测。没有发邮件、HTTP或外部副作用，不能把此模式用于“exactly once付款”。

故障恢复：先停外部触发，确认没有活跃实例，保存报告/.pending/.lock证据；内容一致可重跑，内容冲突先人工核对，不自动覆盖。确认无实例后仅移除空.lock；临时文件为普通私有文件时可由下次重跑覆盖，不得把它当成功报告。更改日期只补跑明确缺失批次，不根据墙钟猜测遗漏。

清理：先解析并核对绝对路径恰为本例/state，无链接/重解析点且无活跃实例。只逐一删除本例拥有的report-日期.json和.pending，再删除空.lock与空state；不使用递归删除、不清理父目录。源码和配置若要撤回，仅删用户复制的本例副本。没有注册任何定时任务，因此无需调用调度产品CLI。

成功失败与恢复预期

自动化日期报告预期首次指定日期得到 created，第二次 unchanged，且字节相同；日期非法不创建state。锁存在时退出3，目标内容被改动时退出4且保留原文件。失败后的临时文件可留作诊断但不作为成功产物，下次持锁并核对普通文件后可覆盖临时内容。

重放练习与解答

题目：如果先写done标记再写报告，崩溃后会怎样？如果按“锁超时十分钟”自动删除又如何？

参考解答：可能标记成功却没有完整产物；慢任务仍活着时删锁会允许两实例并发。当前项目把最终报告作为幂等证据，冲突拒绝、过期锁人工确认；不把时间窗口当身份与活性证明。

核对资料

继续阅读

知识专题
自动化任务：发布、排障与恢复练习
知识专题
自动化任务：基础对象与机制

自动化：按日期生成可重放报告 ​

完整日期报告项目 ​

完整源码与配置 ​

成功失败与恢复预期 ​

重放练习与解答 ​

核对资料 ​