用量统计

curl 'https://api.example.com/api/v1/proxy/usage/summary' \
  -H 'Authorization: Bearer at_YOUR_ACCESS_TOKEN'

curl 'https://api.example.com/api/v1/proxy/usage/daily?days=30' \
  -H 'Authorization: Bearer at_YOUR_ACCESS_TOKEN'

curl 'https://api.example.com/api/v1/proxy/usage/domains' \
  -H 'Authorization: Bearer at_YOUR_ACCESS_TOKEN'

import csv
import json
from datetime import date, timedelta


def export_daily_usage(client, days: int = 90, path: str = "usage.csv") -> None:
    """导出逐日用量。

    小时级明细只保留 60 天，日汇总是长期的，
    但仍建议定期导出——接口一次最多给 90 天。
    """
    payload = client.get("/api/v1/proxy/usage/daily", days=days)

    with open(path, "w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=["date", "uploadBytes", "downloadBytes", "bytesBilled"],
        )
        writer.writeheader()
        writer.writerows(payload["items"])


def export_hourly_snapshot(client, path: str = "hourly.jsonl") -> None:
    """导出小时级明细。这类数据 60 天后消失，需要定期跑。"""
    payload = client.get("/api/v1/proxy/usage/series", granularity="hour")

    with open(path, "a", encoding="utf-8") as handle:
        for point in payload["items"]:
            handle.write(json.dumps(point, ensure_ascii=False) + "\n")


def export_domains(client, path: str = "domains.csv") -> None:
    """域名明细最多 500 个，且是"最近访问"的滚动窗口。"""
    payload = client.get("/api/v1/proxy/usage/domains")

    with open(path, "w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=["domain", "bytesBilled"])
        writer.writeheader()
        writer.writerows(payload["items"])

def daily_check(client, *, warn_gb: float = 20.0, spike_ratio: float = 2.0) -> None:
    """每天跑一次：余额告警 + 消耗突增检测。"""
    summary = client.get("/api/v1/proxy/usage/summary")
    remaining_gb = summary["trafficRemainBytes"] / 1024 ** 3

    if remaining_gb < warn_gb:
        alert(f"流量余额 {remaining_gb:.1f} GB，低于 {warn_gb} GB")

    daily = client.get("/api/v1/proxy/usage/daily", days=8)["items"]
    if len(daily) < 8:
        return

    # 拿最近一天和前七天均值比，突增说明有异常任务
    recent = daily[-1]["bytesBilled"]
    baseline = sum(item["bytesBilled"] for item in daily[-8:-1]) / 7

    if baseline > 0 and recent > baseline * spike_ratio:
        domains = client.get("/api/v1/proxy/usage/domains")["items"][:5]
        top = ", ".join(f"{d['domain']}" for d in domains)
        alert(
            f"用量突增：{recent / 1024 ** 3:.1f} GB vs 均值 "
            f"{baseline / 1024 ** 3:.1f} GB。消耗最高的域名：{top}"
        )