一次性优化能省 30%，半年后又涨回去了

"""成本异动检测与告警。

每天凌晨执行，对比昨日与前 7 天同期的日均成本，
按业务线维度找出异动并推送到企业微信/钉钉。
"""

from __future__ import annotations

import json
import os
from datetime import date, timedelta

import boto3
import urllib3

ce = boto3.client("ce", region_name="us-east-1")   # Cost Explorer 只在 us-east-1
http = urllib3.PoolManager()

WEBHOOK = os.environ["ALERT_WEBHOOK"]
SPIKE_PCT = float(os.environ.get("SPIKE_PCT", "20"))
MIN_DELTA_USD = float(os.environ.get("MIN_DELTA_USD", "50"))


def daily_cost_by_tag(start: date, end: date, tag_key: str = "CostCenter") -> dict[str, dict[str, float]]:
    """返回 {日期: {业务线: 成本}}。"""
    resp = ce.get_cost_and_usage(
        TimePeriod={"Start": start.isoformat(), "End": end.isoformat()},
        Granularity="DAILY",
        Metrics=["UnblendedCost"],
        GroupBy=[{"Type": "TAG", "Key": tag_key}],
    )
    out: dict[str, dict[str, float]] = {}
    for result in resp["ResultsByTime"]:
        day = result["TimePeriod"]["Start"]
        out[day] = {}
        for group in result["Groups"]:
            raw = group["Keys"][0]
            name = raw.split("$", 1)[1] or "__UNTAGGED__"
            out[day][name] = float(group["Metrics"]["UnblendedCost"]["Amount"])
    return out


def detect_spikes() -> list[dict]:
    today = date.today()
    yesterday = today - timedelta(days=1)
    baseline_start = today - timedelta(days=8)

    data = daily_cost_by_tag(baseline_start, today)
    y_key = yesterday.isoformat()
    if y_key not in data:
        return []

    yesterday_cost = data[y_key]
    baseline_days = [d for d in data if d != y_key]

    spikes = []
    for cc, cost in yesterday_cost.items():
        history = [data[d].get(cc, 0.0) for d in baseline_days]
        if not history:
            continue
        avg = sum(history) / len(history)
        if avg <= 0:
            continue
        delta = cost - avg
        pct = 100.0 * delta / avg
        if pct >= SPIKE_PCT and delta >= MIN_DELTA_USD:
            spikes.append({
                "cost_center": cc,
                "yesterday": round(cost, 2),
                "baseline_avg": round(avg, 2),
                "delta": round(delta, 2),
                "pct": round(pct, 1),
            })

    return sorted(spikes, key=lambda s: s["delta"], reverse=True)


def notify(spikes: list[dict]) -> None:
    if not spikes:
        return
    lines = ["## 成本异动提醒",", f"检测日期：{date.today() - timedelta(days=1)}","]
    for s in spikes:
        lines.append(
            f"- **{s['cost_center']}**：${s['yesterday']} "
            f"（7 日均值 ${s['baseline_avg']}，↑{s['pct']}%，"
            f"多支出 ${s['delta']}）"
        )
    lines += ["", "请对应负责人在今日内确认原因。"]

    payload = {"msgtype": "markdown", "markdown": {"content": "\n".join(lines)}}
    http.request(
        "POST", WEBHOOK,
        body=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
    )


def lambda_handler(event, context):  # noqa: ARG001
    spikes = detect_spikes()
    notify(spikes)
    return {"spike_count": len(spikes), "spikes": spikes}