climperor/notify_site_traffic.py

"""Daily site traffic digest -> Feishu webhook.

Pulls Cloudflare edge analytics for dota2.refining.dev, estimates human
homepage visits (known browser UA on `/`, plus RUM beacons), and posts a
Feishu custom-bot card.

Credentials (env, prefer keyzoo inject names):
  CLOUDFLARE_EMAIL / KEYZOO_ASSET_META_USERNAME
  CLOUDFLARE_API_KEY / KEYZOO_ASSET_SECRET_GLOBAL_API_KEY
  FEISHU_WEBHOOK_URL / KEYZOO_ASSET_SECRET_FEISHU_WEBHOOK_URL

Usage:
  python notify_site_traffic.py              # yesterday (UTC date)
  python notify_site_traffic.py --day 2026-07-27
  python notify_site_traffic.py --dry-run    # print card, do not POST
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from typing import Any

HOST = "dota2.refining.dev"
ZONE_NAME = "refining.dev"
SITE_URL = "https://dota2.refining.dev"

HUMAN_BROWSERS = frozenset(
    {
        "Chrome",
        "Firefox",
        "Safari",
        "Edge",
        "Opera",
        "MobileSafari",
        "ChromeMobile",
        "ChromeMobileWebview",
        "FirefoxMobile",
        "SamsungInternet",
        "EdgeMobile",
    }
)


def _env(*names: str) -> str | None:
    for n in names:
        v = os.environ.get(n)
        if v:
            return v
    return None


def cf_headers() -> dict[str, str]:
    email = _env("CLOUDFLARE_EMAIL", "KEYZOO_ASSET_META_USERNAME")
    key = _env("CLOUDFLARE_API_KEY", "KEYZOO_ASSET_SECRET_GLOBAL_API_KEY")
    if not email or not key:
        raise SystemExit(
            "missing Cloudflare credentials "
            "(CLOUDFLARE_EMAIL + CLOUDFLARE_API_KEY, or keyzoo inject)"
        )
    return {
        "X-Auth-Email": email,
        "X-Auth-Key": key,
        "Content-Type": "application/json",
    }


def feishu_url() -> str:
    url = _env("FEISHU_WEBHOOK_URL", "KEYZOO_ASSET_SECRET_FEISHU_WEBHOOK_URL")
    if not url:
        raise SystemExit(
            "missing Feishu webhook "
            "(FEISHU_WEBHOOK_URL or KEYZOO_ASSET_SECRET_FEISHU_WEBHOOK_URL)"
        )
    return url


def http_json(url: str, *, method: str = "GET", body: dict | None = None, headers: dict | None = None) -> dict:
    data = None if body is None else json.dumps(body).encode()
    req = urllib.request.Request(url, data=data, method=method, headers=headers or {})
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            return json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        raw = e.read().decode(errors="replace")
        raise SystemExit(f"HTTP {e.code} {url}: {raw[:800]}") from e


def gql(query: str, variables: dict) -> dict:
    data = http_json(
        "https://api.cloudflare.com/client/v4/graphql",
        method="POST",
        body={"query": query, "variables": variables},
        headers=cf_headers(),
    )
    if data.get("errors"):
        raise SystemExit("GraphQL errors: " + json.dumps(data["errors"], ensure_ascii=False)[:1500])
    return data


@dataclass
class DayStats:
    day: date
    all_req: int = 0
    all_vis: int = 0
    human_home_req: int = 0
    human_home_vis: int = 0
    other_home_vis: int = 0
    rum_human: int = 0
    rum_other: int = 0
    bytes: int = 0
    top_countries: list[tuple[str, int, int]] = field(default_factory=list)  # name, req, vis
    status: list[tuple[Any, int]] = field(default_factory=list)


def zone_id() -> str:
    d = http_json(
        f"https://api.cloudflare.com/client/v4/zones?name={ZONE_NAME}",
        headers=cf_headers(),
    )
    rows = d.get("result") or []
    if not rows:
        raise SystemExit(f"zone not found: {ZONE_NAME}")
    return rows[0]["id"]


def fetch_day(zid: str, day: date) -> DayStats:
    q = """
    query($zoneTag: string!, $day: Date!, $host: string!) {
      viewer {
        zones(filter: { zoneTag: $zoneTag }) {
          all: httpRequestsAdaptiveGroups(
            limit: 1
            filter: { date: $day, clientRequestHTTPHost: $host }
          ) { count sum { visits edgeResponseBytes } }
          homeByUA: httpRequestsAdaptiveGroups(
            limit: 40
            filter: {
              date: $day
              clientRequestHTTPHost: $host
              clientRequestPath: "/"
            }
            orderBy: [count_DESC]
          ) {
            dimensions { userAgentBrowser }
            count
            sum { visits }
          }
          rumByUA: httpRequestsAdaptiveGroups(
            limit: 20
            filter: {
              date: $day
              clientRequestHTTPHost: $host
              clientRequestPath: "/cdn-cgi/rum"
            }
            orderBy: [count_DESC]
          ) {
            dimensions { userAgentBrowser }
            count
          }
          homeByCountry: httpRequestsAdaptiveGroups(
            limit: 8
            filter: {
              date: $day
              clientRequestHTTPHost: $host
              clientRequestPath: "/"
            }
            orderBy: [count_DESC]
          ) {
            dimensions { clientCountryName }
            count
            sum { visits }
          }
          byStatus: httpRequestsAdaptiveGroups(
            limit: 8
            filter: { date: $day, clientRequestHTTPHost: $host }
            orderBy: [count_DESC]
          ) {
            dimensions { edgeResponseStatus }
            count
          }
        }
      }
    }
    """
    data = gql(q, {"zoneTag": zid, "day": day.isoformat(), "host": HOST})
    z = data["data"]["viewer"]["zones"][0]
    st = DayStats(day=day)

    all_rows = z.get("all") or []
    if all_rows:
        r = all_rows[0]
        st.all_req = r.get("count") or 0
        st.all_vis = (r.get("sum") or {}).get("visits") or 0
        st.bytes = (r.get("sum") or {}).get("edgeResponseBytes") or 0

    for r in z.get("homeByUA") or []:
        ua = (r.get("dimensions") or {}).get("userAgentBrowser")
        c = r.get("count") or 0
        v = (r.get("sum") or {}).get("visits") or 0
        if ua in HUMAN_BROWSERS:
            st.human_home_req += c
            st.human_home_vis += v
        else:
            st.other_home_vis += v

    for r in z.get("rumByUA") or []:
        ua = (r.get("dimensions") or {}).get("userAgentBrowser")
        c = r.get("count") or 0
        if ua in HUMAN_BROWSERS:
            st.rum_human += c
        else:
            st.rum_other += c

    for r in z.get("homeByCountry") or []:
        name = (r.get("dimensions") or {}).get("clientCountryName") or "?"
        v = (r.get("sum") or {}).get("visits") or 0
        st.top_countries.append((name, r.get("count") or 0, v))

    for r in z.get("byStatus") or []:
        code = (r.get("dimensions") or {}).get("edgeResponseStatus")
        st.status.append((code, r.get("count") or 0))

    return st


def fmt_bytes(n: int) -> str:
    x = float(n)
    for unit in ("B", "KB", "MB", "GB"):
        if abs(x) < 1024:
            return f"{x:.1f} {unit}"
        x /= 1024
    return f"{x:.1f} TB"


def build_card(st: DayStats) -> dict:
    """Feishu interactive card (msg_type=interactive)."""
    countries = "、".join(
        f"{name} {vis}访/{req}次" for name, req, vis in st.top_countries[:5]
    ) or "—"
    status_line = "、".join(f"{code}×{n}" for code, n in st.status[:6]) or "—"
    now = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M CST")

    # Free plan: no BotScore. Human ≈ known-browser homepage visits.
    estimate = st.human_home_vis
    rum = st.rum_human

    elements: list[dict] = [
        {
            "tag": "div",
            "text": {
                "tag": "lark_md",
                "content": (
                    f"**真实访问（估）**：约 **{estimate}** 次首页打开\n"
                    f"（已知浏览器 UA 打开 `/`；不含 Unknown/Headless/爬虫）\n\n"
                    f"**RUM 页面浏览**：{rum}（人类浏览器执行 JS）"
                    + (f" + {st.rum_other} 其他" if st.rum_other else "")
                    + "\n"
                    f"**边缘总量**：{st.all_req} 请求 / {st.all_vis} visits / {fmt_bytes(st.bytes)}\n"
                    f"**非人类首页命中**：{st.other_home_vis} visits"
                ),
            },
        },
        {
            "tag": "div",
            "text": {
                "tag": "lark_md",
                "content": f"**首页地区 Top**：{countries}\n**状态码**：{status_line}",
            },
        },
        {
            "tag": "note",
            "elements": [
                {
                    "tag": "plain_text",
                    "content": (
                        f"{HOST} · Free 档无 BotScore，按 UA+路径估算 · {now}"
                    ),
                }
            ],
        },
        {
            "tag": "action",
            "actions": [
                {
                    "tag": "button",
                    "text": {"tag": "plain_text", "content": "打开站点"},
                    "type": "primary",
                    "url": SITE_URL,
                }
            ],
        },
    ]

    return {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {
                    "tag": "plain_text",
                    "content": f"上分帝 Web 日活 · {st.day.isoformat()}",
                },
                "template": "blue" if estimate > 0 else "grey",
            },
            "elements": elements,
        },
    }


def post_feishu(payload: dict) -> dict:
    url = feishu_url()
    return http_json(
        url,
        method="POST",
        body=payload,
        headers={"Content-Type": "application/json"},
    )


def parse_day(s: str | None) -> date:
    if not s:
        # Yesterday in China time (site audience).
        cst = timezone(timedelta(hours=8))
        return (datetime.now(cst) - timedelta(days=1)).date()
    return date.fromisoformat(s)


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--day", help="YYYY-MM-DD (default: yesterday CST)")
    ap.add_argument(
        "--dry-run",
        action="store_true",
        help="print JSON card only, do not call Feishu",
    )
    args = ap.parse_args(argv)

    day = parse_day(args.day)
    zid = zone_id()
    st = fetch_day(zid, day)
    payload = build_card(st)

    print(
        f"{HOST} {day}: human_home_vis={st.human_home_vis} "
        f"rum_human={st.rum_human} all_req={st.all_req}",
        flush=True,
    )

    if args.dry_run:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
        return 0

    resp = post_feishu(payload)
    # Feishu returns {code:0, msg:success} on OK
    code = resp.get("code", resp.get("StatusCode"))
    if code not in (0, None) and code != 200:
        print("Feishu error:", json.dumps(resp, ensure_ascii=False)[:800], file=sys.stderr)
        return 1
    print("feishu ok:", json.dumps(resp, ensure_ascii=False)[:200])
    return 0


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