当 CEO 说"用 AI 替代度量者"，你的岗位离自动化还有多远

"""
weekly_metrics_report.py
自动拉取指标、标注异常、生成中文周报。
依赖：pip install httpx openai
运行前设置环境变量：
  export OPENAI_API_KEY="sk-..."
  export SERVICE_API_BASE="https://api.example.com"  # 你的指标服务地址
"""

import os
import json
from datetime import datetime, timedelta
import httpx
from openai import OpenAI

# ---------- 1. 拉取结构化指标 ----------

def fetch_service_metrics(service_api_base: str, days: int = 7) -> dict:
    """从内部 API 拉取最近 N 天的核心指标"""
    end = datetime.utcnow()
    start = end - timedelta(days=days)
    params = {"start": start.isoformat(), "end": end.isoformat()}

    endpoints = {
        "availability": "/metrics/availability",
        "latency_p99": "/metrics/latency",
        "error_rate": "/metrics/errors",
        "request_count": "/metrics/requests",
    }

    results = {}
    with httpx.Client(base_url=service_api_base, timeout=30) as client:
        for key, path in endpoints.items():
            try:
                resp = client.get(path, params=params)
                resp.raise_for_status()
                results[key] = resp.json()
            except httpx.HTTPError as e:
                results[key] = {"error": str(e)}

    return results


# ---------- 2. 定义异常阈值 ----------

THRESHOLDS = {
    "availability": 0.995,      # 低于 99.5% 标红
    "latency_p99": 500,         # P99 超过 500ms 标红
    "error_rate": 0.01,         # 错误率超过 1% 标红
}


def annotate_anomalies(metrics: dict) -> dict:
    """对照阈值标注异常"""
    annotated = {}
    for key, threshold in THRESHOLDS.items():
        data = metrics.get(key, {})
        if "error" in data:
            annotated[key] = {"value": None, "status": "数据获取失败", "threshold": threshold}
            continue
        # 假设 API 返回 {"value": 0.998, "unit": "ratio"} 或 {"value": 320, "unit": "ms"}
        value = data.get("value")
        violated = value < threshold if key != "latency_p99" else value > threshold
        annotated[key] = {
            "value": value,
            "unit": data.get("unit", ""),
            "status": "异常 ⚠️" if violated else "正常 ✅",
            "threshold": threshold,
        }
    # 不设阈值的指标直接记录
    annotated["request_count"] = metrics.get("request_count", {})
    return annotated


# ---------- 3. 用 LLM 生成可读周报 ----------

def generate_report(annotated: dict) -> str:
    """把标注后的指标交给 LLM，生成结构化中文周报"""
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

    prompt = f"""你是一名基础设施团队的周报撰写助手。
以下是本周服务指标的标注结果（JSON 格式），其中已标注异常项。
请生成一份简洁的中文周报，包含：
1. 总体健康评估（一句话）
2. 异常项说明和建议动作
3. 下周关注重点

指标数据：
{json.dumps(annotated, ensure_ascii=False, indent=2)}
"""

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
    )
    return resp.choices[0].message.content


# ---------- 4. 主流程 ----------

def main():
    api_base = os.environ.get("SERVICE_API_BASE", "https://api.example.com")
    raw_metrics = fetch_service_metrics(api_base)
    annotated = annotate_anomalies(raw_metrics)
    report = generate_report(annotated)

    print("=" * 60)
    print(f"服务周报 — {datetime.utcnow().strftime('%Y-%m-%d')}")
    print("=" * 60)
    print(report)


if __name__ == "__main__":
    main()

# 安装依赖
pip install httpx openai

# 设置环境变量
export OPENAI_API_KEY="sk-your-key-here"
export SERVICE_API_BASE="https://your-internal-api.example.com"

# 执行
python weekly_metrics_report.py