用 Sentry Crons 监控每晚 02:30（北京时间）运行的订单日报脚本

用 Sentry Crons 监控每晚 02:30（北京时间）运行的订单日报脚本：任务开始和结束各发一次 check-in，没按时启动、运行超过 20 分钟、执行报错，Sentry 都会创建问题并告警。监控配置写在代码里，首次运行自动创建。

先看清成品

本机用 sentry-sdk 2.72.0 跑了 2 个测试，全部通过：正常运行时截获 in_progress → ok 两个 check-in（同一个 check_in_id，带时区和时长），出错时截获 in_progress → error；脚本实际生成了日报文件。没有发送到真实 Sentry，也没有配置真实的 cron 定时任务。

注意：Sentry 有免费和付费计划，额度与价格以官方价格页为准。错误数据里可能包含用户个人信息，上线前确认采集范围符合你所在地区的个人信息保护法规（如《个人信息保护法》），并在隐私政策中说明。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

25-sentry-C-src/
├── nightly_report.py
├── orders_sample.csv
└── test_nightly.py

第 1 步：脚本与监控配置

nightly_report.py：

"""每天 02:30（北京时间）汇总前一天订单，生成日报文件。用 Sentry Crons 监控：没按时跑、跑太久、报错都会告警。"""
import csv, os, sys
from datetime import date, timedelta
from pathlib import Path
import sentry_sdk
from sentry_sdk.crons import monitor

MONITOR_CONFIG = {
    "schedule": {"type": "crontab", "value": "30 2 * * *"},
    "timezone": "Asia/Shanghai",
    "checkin_margin": 15,         # 02:45 前没开始算“错过”
    "max_runtime": 20,            # 超过 20 分钟算失败
    "failure_issue_threshold": 1, # 失败 1 次就建问题
    "recovery_threshold": 1,
}

def summarize(rows):
    paid = [r for r in rows if r["status"] == "paid"]
    if not rows:
        raise RuntimeError("订单导出为空，可能上游导出失败")
    return {"orders": len(paid), "revenue": round(sum(float(r["amount"]) for r in paid), 2)}

@monitor(monitor_slug="nightly-order-report", monitor_config=MONITOR_CONFIG)
def run(src: Path, outdir: Path, day: date):
    with open(src, encoding="utf-8-sig", newline="") as f:
        stats = summarize(list(csv.DictReader(f)))
    outdir.mkdir(exist_ok=True)
    out = outdir / f"report_{day.isoformat()}.txt"
    out.write_text(f"{day} 已支付订单 {stats['orders']} 笔，金额 {stats['revenue']:.2f} 元\n", encoding="utf-8")
    return out

if __name__ == "__main__":
    sentry_sdk.init(dsn=os.environ.get("SENTRY_DSN"), environment=os.environ.get("APP_ENV", "production"))
    day = date.today() - timedelta(days=1)
    print("写出", run(Path(sys.argv[1]), Path("reports"), day))

orders_sample.csv：

order_id,status,amount
1001,paid,59.90
1002,refunded,89.00
1003,paid,15.00

要点：

• schedule 与 crontab 写法相同；timezone 设为 Asia/Shanghai，Sentry 按北京时间判断是否错过。
• 导出为空时主动抛错，让“数据不对”也算失败，而不是生成一份 0 元日报。

第 2 步：测试

test_nightly.py：

from datetime import date
from pathlib import Path
import pytest, sentry_sdk
from sentry_sdk.transport import Transport
import nightly_report as m

class Capture(Transport):
    def __init__(self, options=None): super().__init__(options); self.items = []
    def capture_envelope(self, env):
        for it in env.items: self.items.append((it.type, it.payload.json))

@pytest.fixture
def cap():
    t = Capture(); sentry_sdk.init(dsn="https://public@o0.ingest.sentry.io/0", transport=t); yield t

def checkins(t): return [p for typ, p in t.items if typ == "check_in"]

def test_ok(cap, tmp_path):
    out = m.run(Path("orders_sample.csv"), tmp_path, date(2026, 10, 20)); sentry_sdk.flush()
    assert out.read_text(encoding="utf-8") == "2026-10-20 已支付订单 2 笔，金额 74.90 元\n"
    c = checkins(cap); assert [x["status"] for x in c] == ["in_progress", "ok"]
    assert c[0]["monitor_slug"] == "nightly-order-report" and c[0]["monitor_config"]["timezone"] == "Asia/Shanghai"
    assert c[1]["check_in_id"] == c[0]["check_in_id"] and c[1]["duration"] >= 0

def test_error(cap, tmp_path):
    empty = tmp_path / "e.csv"; empty.write_text("order_id,status,amount\n")
    with pytest.raises(RuntimeError): m.run(empty, tmp_path, date(2026, 10, 20))
    sentry_sdk.flush()
    assert [x["status"] for x in checkins(cap)] == ["in_progress", "error"]

pip install sentry-sdk pytest
python -m pytest -q
python nightly_report.py orders_sample.csv

检查结果（本机实际输出）：

2 passed
写出 reports/report_2026-10-20.txt

文件内容：2026-10-20 已支付订单 2 笔，金额 74.90 元。

第 3 步：配置定时任务（本文未执行）

服务器时区若为北京时间，crontab：

CRON_TZ=Asia/Shanghai
30 2 * * * cd /srv/report && SENTRY_DSN=https://xxx@oNNN.ingest.sentry.io/NNN /usr/bin/python3 nightly_report.py /srv/export/orders.csv >> report.log 2>&1

CRON_TZ 只有部分 cron 实现支持；不支持时按服务器时区换算。DSN 也可以写进只有运行用户可读的环境文件。

第 4 步：告警（本文未执行）

首次运行后，Sentry Crons 页面出现 nightly-order-report。在 Alerts → Create Alert → Issues 中添加筛选条件 monitor.slug equals nightly-order-report，通知到值班邮箱。

预期结果：

调试示例

测试 test_error 中传入只有表头的 CSV，summarize 抛出 RuntimeError: 订单导出为空，可能上游导出失败，monitor 自动把第二个 check-in 设为 error。截获的状态序列为：

['in_progress', 'error']

这说明错误不需要手动上报 check-in，装饰器会处理。

出错了怎么办？

模板规则

• 每个定时任务一个 monitor_slug，配置写在代码里。
• 数据异常主动抛错。
• 时区写明，和定时器保持一致。

项目资源

定时任务最常见的故障是“悄悄没跑”。有了 check-in，没跑、跑太久、跑出错都会有人知道，而不是等到第二天有人问日报去哪了。