定时报表脚本

摘要：日报/周报是把散落的指标汇总成一份「五分钟能看完」的摘要。本文用 Python 采集指标、生成 Markdown/HTML 表格，通过 SMTP 发送，并说明凭证处理、发送失败重试与订阅管理。

适用环境

python3 --version
pip install requests
# 检查本机能否连通 SMTP（587/465）
nc -vz -w 3 smtp.example.com 587 || echo "端口不通，需确认网络与安全组"

操作步骤

一、确定报表内容

必含：总体健康状态、关键指标（QPS/错误率/P95）、昨日告警条数、资源水位（CPU/磁盘）
选含：变更记录、成本环比、Top N 慢接口
原则：一屏看完，异常项排最前，正常项只给数字

二、采集数据

#!/usr/bin/env python3
"""每日运维报表：采集指标 → 渲染表格 → 邮件发送。"""
import json
import os
import smtplib
import ssl
import subprocess
from datetime import datetime, timedelta
from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

METRICS_API = os.environ.get("METRICS_API", "https://api.example.com/v1/metrics")
SMTP_HOST = os.environ.get("SMTP_HOST", "smtp.example.com")
SMTP_PORT = int(os.environ.get("SMTP_PORT", "587"))
SMTP_USER = os.environ.get("SMTP_USER", "")
SMTP_PASS = os.environ.get("SMTP_PASS", "")   # 从环境变量注入，绝不写死
MAIL_TO   = os.environ.get("MAIL_TO", "ops-team@example.com").split(",")


def run(cmd: list[str], timeout: int = 30) -> str:
    p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    return p.stdout.strip() if p.returncode == 0 else f"(失败: {p.stderr.strip()[:80]})"


def collect_host() -> dict:
    """采集本机基础指标"""
    load1, load5, _ = run(["cat", "/proc/loadavg"]).split()[:3]
    disk = run(["bash", "-c", "df -P / | awk 'NR==2{print $5}'"])
    mem = run(["bash", "-c", "free -m | awk '/^Mem:/{printf \"%d%%\", $3*100/$2}'"])
    return {"load1": load1, "load5": load5, "disk_root": disk, "mem": mem}


def fetch_business(day: str) -> dict:
    """从监控接口拉取业务指标（示例：实际替换为你的数据源）"""
    import requests
    try:
        r = requests.get(METRICS_API, params={"day": day}, timeout=(3, 10))
        r.raise_for_status()
        d = r.json().get("data", {})
        return {"qps": d.get("qps", "-"),
                "error_rate": d.get("error_rate", "-"),
                "p95_ms": d.get("p95_ms", "-")}
    except Exception as e:
        return {"qps": "-", "error_rate": "-", "p95_ms": "-", "note": f"采集失败: {e}"}

三、渲染表格

def render_md(day: str, host: dict, biz: dict) -> str:
    lines = [
        f"# 运维日报 {day}",
        "",
        "## 业务指标",
        "",
        "| 指标 | 数值 |",
        "| --- | --- |",
        f"| QPS | {biz.get('qps')} |",
        f"| 错误率 | {biz.get('error_rate')} |",
        f"| P95 延迟 | {biz.get('p95_ms')} ms |",
        "",
        "## 主机水位",
        "",
        "| 项目 | 数值 |",
        "| --- | --- |",
        f"| load1 / load5 | {host['load1']} / {host['load5']} |",
        f"| 根分区使用率 | {host['disk_root']} |",
        f"| 内存使用率 | {host['mem']} |",
        "",
    ]
    if biz.get("note"):
        lines += ["> 注意：业务指标采集异常，请检查数据源。", ""]
    return "\n".join(lines)


def render_html(md: str) -> str:
    """极简 Markdown → HTML（仅处理标题、表格、换行）"""
    out, in_table = [], False
    for line in md.splitlines():
        if line.startswith("|"):
            cells = [c.strip() for c in line.strip("|").split("|")]
            if set("".join(cells)) <= set("- "):
                continue
            if not in_table:
                out.append("<table border=1 cellspacing=0 cellpadding=6>")
                in_table = True
            out.append("<tr>" + "".join(f"<td>{c}</td>" for c in cells) + "</tr>")
            continue
        if in_table:
            out.append("</table>")
            in_table = False
        if line.startswith("# "):
            out.append(f"<h2>{line[2:]}</h2>")
        elif line.startswith("## "):
            out.append(f"<h3>{line[3:]}</h3>")
        elif line.strip():
            out.append(f"<p>{line}</p>")
    if in_table:
        out.append("</table>")
    return "\n".join(out)

四、发送邮件

def send_mail(subject: str, md_body: str, html_body: str) -> None:
    msg = MIMEMultipart("alternative")
    msg["Subject"] = Header(subject, "utf-8")
    msg["From"] = SMTP_USER
    msg["To"] = ", ".join(MAIL_TO)
    msg.attach(MIMEText(md_body, "plain", "utf-8"))
    msg.attach(MIMEText(html_body, "html", "utf-8"))

    ctx = ssl.create_default_context()
    with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as s:
        s.starttls(context=ctx)      # 587 端口用 STARTTLS；465 端口改用 SMTP_SSL
        s.login(SMTP_USER, SMTP_PASS)
        s.sendmail(SMTP_USER, MAIL_TO, msg.as_string())


def main() -> int:
    day = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
    host = collect_host()
    biz = fetch_business(day)
    md = render_md(day, host, biz)
    try:
        send_mail(f"运维日报 {day}", md, render_html(md))
        print("sent")
        return 0
    except Exception as e:
        # 发送失败时把报表落盘，避免内容丢失
        path = f"/var/log/ops/report-{day}.md"
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(md)
        print(f"发送失败已落盘 {path}: {e}", flush=True)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

危险

SMTP 密码属于高敏感凭证，绝不能写在代码里。应通过环境变量或密钥管理服务注入，并且报表脚本所用的发信账号应只具备发信权限，不复用管理员邮箱。

五、定时运行

# 每天 08:30 发送昨日日报
30 8 * * * SMTP_USER=reports@example.com SMTP_PASS="$SMTP_PASS" \
  MAIL_TO=ops@example.com,lead@example.com \
  /usr/bin/python3 /usr/local/bin/daily_report.py >> /var/log/ops/report.log 2>&1

cron 中引用变量要小心：% 在 crontab 里需转义为 \%。凭证建议写在 root 权限 600 的环境文件里再 source。

六、订阅与退订

# 收件人列表放配置文件，避免改代码
cat > /etc/ops/report.conf <<'EOF'
MAIL_TO="ops@example.com,lead@example.com"
EOF

验证

• [ ] 本地运行能打印出完整 Markdown 表格
• [ ] 收到测试邮件，标题、正文、表格渲染正常（含中文不乱码）
• [ ] 故意填错 SMTP 密码，脚本退出码为 1 且报表已落盘
• [ ] 数据源不可用时报表中出现「采集异常」提示而非崩溃

python3 daily_report.py; echo "退出码=$?"

常见坑

• 中文乱码：MIMEText 必须指定 utf-8，主题用 Header(..., "utf-8")。
• 端口与加密方式不匹配：587 用 starttls，465 用 SMTP_SSL，混用会超时或报错。
• cron 里 % 未转义：crontab 中 % 有特殊含义，不转义会导致命令被截断。
• 凭证硬编码：提交到 Git 后必须轮换，代价很高。
• 报表太长没人看：几十项指标会被直接略过，应只保留关键项并把异常置顶。

参考资料

• Python smtplib 文档
• Python email 模块文档
• crontab 手册
• Requests 文档