"""Daily site digest -> Feishu webhook.

1) Cloudflare edge analytics for dota2.refining.dev (human homepage estimate)
2) Auto-refresh status: Gitea Actions (web-daily / web-weekly / web-patch)
   + Cloudflare Pages production deploys that day

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
  GITEA_TOKEN / KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN_GITEA_1  (optional;
    without it the refresh section shows \"未配置 token\")

Usage:
  python notify_site_traffic.py              # yesterday CST
  python notify_site_traffic.py --day 2026-07-27
  python notify_site_traffic.py --dry-run
"""

from __future__ import annotations

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

HOST = "dota2.refining.dev"
ZONE_NAME = "refining.dev"
SITE_URL = "https://dota2.refining.dev"
PAGES_PROJECT = "climperor-relations"
GITEA_URL_DEFAULT = "https://gitea.refining.dev"
GITEA_OWNER = "refining"
GITEA_REPO = "climperor"
CST = timezone(timedelta(hours=8))

# (workflow file basename, short label)
REFRESH_WORKFLOWS: tuple[tuple[str, str], ...] = (
    ("web-daily.yml", "每日数据"),
    ("web-weekly.yml", "每周 Meta"),
    ("web-patch.yml", "版本检测"),
)

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 gitea_token() -> str | None:
    return _env(
        "GITEA_TOKEN",
        "KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN_GITEA_1",
        "KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN__GITEA_1",
        "KEYZOO_ASSET_TOKEN",
    )


def gitea_base() -> str:
    return (
        _env("GITEA_URL", "KEYZOO_ASSET_META_URL") or GITEA_URL_DEFAULT
    ).rstrip("/")


def http_json(
    url: str,
    *,
    method: str = "GET",
    body: dict | None = None,
    headers: dict | None = None,
) -> dict | list:
    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:
            raw = r.read().decode()
            if not raw:
                return {}
            return json.loads(raw)
    except urllib.error.HTTPError as e:
        raw = e.read().decode(errors="replace")
        raise SystemExit(f"HTTP {e.code} {url}: {raw[:800]}") from e


def http_json_soft(
    url: str,
    *,
    method: str = "GET",
    body: dict | None = None,
    headers: dict | None = None,
) -> tuple[int, Any]:
    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:
            raw = r.read().decode()
            if not raw:
                return r.status, {}
            return r.status, json.loads(raw)
    except urllib.error.HTTPError as e:
        raw = e.read().decode(errors="replace")
        try:
            return e.code, json.loads(raw)
        except json.JSONDecodeError:
            return e.code, raw


def http_response_soft(
    url: str,
    *,
    headers: dict | None = None,
) -> tuple[int, bytes, str, dict[str, str]]:
    request_headers = {"User-Agent": "climperor-monitor", "Accept": "*/*"}
    request_headers.update(headers or {})
    req = urllib.request.Request(url, headers=request_headers)
    try:
        with urllib.request.urlopen(req, timeout=60) as response:
            return (
                response.status,
                response.read(),
                response.headers.get_content_type(),
                {k.lower(): v for k, v in response.headers.items()},
            )
    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as exc:
        if isinstance(exc, urllib.error.HTTPError):
            return (
                exc.code,
                exc.read(),
                exc.headers.get_content_type(),
                {k.lower(): v for k, v in exc.headers.items()},
            )
        return 0, str(exc).encode("utf-8", errors="replace"), "", {}


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 isinstance(data, dict) and data.get("errors"):
        raise SystemExit(
            "GraphQL errors: " + json.dumps(data["errors"], ensure_ascii=False)[:1500]
        )
    assert isinstance(data, dict)
    return data


def read_site_version() -> str:
    path = Path(__file__).resolve().parent / "frontend" / "config.js"
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        return "?"
    m = re.search(r'SITE_VERSION\s*=\s*"([^"]+)"', text)
    return m.group(1) if m else "?"


@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)
    status: list[tuple[Any, int]] = field(default_factory=list)


@dataclass
class WorkflowDayStatus:
    file: str
    label: str
    runs: list[dict] = field(default_factory=list)  # status/conclusion/started_at
    note: str = ""  # e.g. missing token / none

    @property
    def ok_count(self) -> int:
        return sum(1 for r in self.runs if r.get("conclusion") == "success")

    @property
    def fail_count(self) -> int:
        return sum(
            1
            for r in self.runs
            if r.get("conclusion") not in (None, "success")
            and r.get("status") == "completed"
        )

    @property
    def running_count(self) -> int:
        return sum(1 for r in self.runs if r.get("status") in ("queued", "in_progress", "waiting"))


@dataclass
class PagesDeploy:
    created_on: datetime
    status: str
    commit: str
    url: str


@dataclass
class DigestExtras:
    site_version: str
    workflows: list[WorkflowDayStatus] = field(default_factory=list)
    deploys: list[PagesDeploy] = field(default_factory=list)
    gitea_note: str = ""
    refresh_summaries: dict[str, dict] = field(default_factory=dict)
    production_health: "ProductionHealth | None" = None


@dataclass
class ProductionHealth:
    site_version: str = "?"
    ages_hours: dict[str, float | None] = field(default_factory=dict)
    stale: list[str] = field(default_factory=list)
    live_api_ok: bool = False
    refresh_run_id: str = ""
    note: str = ""


def zone_id() -> str:
    d = http_json(
        f"https://api.cloudflare.com/client/v4/zones?name={ZONE_NAME}",
        headers=cf_headers(),
    )
    assert isinstance(d, dict)
    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 day_window_utc(day: date) -> tuple[datetime, datetime]:
    """Inclusive start / exclusive end of CST calendar day, as aware UTC datetimes."""
    start_cst = datetime(day.year, day.month, day.day, tzinfo=CST)
    end_cst = start_cst + timedelta(days=1)
    return start_cst.astimezone(timezone.utc), end_cst.astimezone(timezone.utc)


def parse_gitea_time(s: str | None) -> datetime | None:
    if not s:
        return None
    try:
        # Gitea may return +08:00 or Z
        if s.endswith("Z"):
            return datetime.fromisoformat(s.replace("Z", "+00:00"))
        return datetime.fromisoformat(s)
    except ValueError:
        return None


def fetch_workflow_statuses(day: date) -> tuple[list[WorkflowDayStatus], str]:
    tok = gitea_token()
    if not tok:
        return (
            [
                WorkflowDayStatus(file=f, label=lab, note="未配置 GITEA_TOKEN")
                for f, lab in REFRESH_WORKFLOWS
            ],
            "未配置 GITEA_TOKEN，跳过 Actions 查询",
        )

    headers = {
        "Authorization": f"token {tok}",
        "Accept": "application/json",
        "Content-Type": "application/json",
    }
    start_utc, end_utc = day_window_utc(day)
    code, payload = http_json_soft(
        f"{gitea_base()}/api/v1/repos/{GITEA_OWNER}/{GITEA_REPO}/actions/runs?limit=50",
        headers=headers,
    )
    if code != 200:
        return (
            [
                WorkflowDayStatus(file=f, label=lab, note=f"API {code}")
                for f, lab in REFRESH_WORKFLOWS
            ],
            f"Gitea runs API HTTP {code}",
        )

    if isinstance(payload, dict):
        items = payload.get("workflow_runs") or payload.get("runs") or []
    elif isinstance(payload, list):
        items = payload
    else:
        items = []

    by_file: dict[str, list[dict]] = {f: [] for f, _ in REFRESH_WORKFLOWS}
    for x in items:
        path = str(x.get("path") or "")
        # path like "web-daily.yml@refs/heads/main"
        base = path.split("@", 1)[0]
        if base not in by_file:
            continue
        ts = parse_gitea_time(x.get("started_at") or x.get("completed_at"))
        if ts is None:
            continue
        ts_utc = ts.astimezone(timezone.utc)
        if not (start_utc <= ts_utc < end_utc):
            continue
        by_file[base].append(
            {
                "id": x.get("id"),
                "status": x.get("status"),
                "conclusion": x.get("conclusion"),
                "event": x.get("event"),
                "started_at": x.get("started_at"),
                "completed_at": x.get("completed_at"),
                "html_url": x.get("html_url"),
            }
        )

    out: list[WorkflowDayStatus] = []
    for f, lab in REFRESH_WORKFLOWS:
        runs = by_file.get(f) or []
        note = "" if runs else "当日无运行"
        out.append(WorkflowDayStatus(file=f, label=lab, runs=runs, note=note))
    return out, ""


def fetch_refresh_summaries(workflows: list[WorkflowDayStatus]) -> tuple[dict[str, dict], str]:
    """Download summary.json artifacts for the latest visible refresh runs."""
    token = gitea_token()
    if not token:
        return {}, "无 Gitea token，未读取刷新摘要"
    headers = {"Authorization": f"token {token}", "Accept": "application/json"}
    code, payload = http_json_soft(
        f"{gitea_base()}/api/v1/repos/{GITEA_OWNER}/{GITEA_REPO}/actions/artifacts?limit=100",
        headers=headers,
    )
    if code != 200 or not isinstance(payload, dict):
        return {}, f"刷新摘要 API HTTP {code}"
    artifacts = payload.get("artifacts") or []
    by_run: dict[int, dict] = {}
    for artifact in artifacts:
        if not isinstance(artifact, dict) or artifact.get("expired"):
            continue
        run = artifact.get("workflow_run") or {}
        try:
            run_id = int(run.get("id") or artifact.get("workflow_run_id") or 0)
        except (TypeError, ValueError):
            continue
        if run_id:
            by_run[run_id] = artifact

    summaries: dict[str, dict] = {}
    for workflow in workflows:
        latest = None
        for run in workflow.runs:
            try:
                run_id = int(run.get("id") or 0)
            except (TypeError, ValueError):
                continue
            if run.get("status") == "completed" and run_id in by_run:
                latest = run
                break
        if not latest:
            continue
        artifact = by_run[int(latest["id"])]
        artifact_id = artifact.get("id")
        if not artifact_id:
            continue
        status, raw, _, _ = http_response_soft(
            f"{gitea_base()}/api/v1/repos/{GITEA_OWNER}/{GITEA_REPO}/actions/artifacts/"
            f"{artifact_id}/zip",
            headers=headers,
        )
        if status != 200:
            continue
        try:
            with zipfile.ZipFile(io.BytesIO(raw)) as archive:
                member = next(
                    name for name in archive.namelist() if name.endswith("summary.json")
                )
                summary = json.loads(archive.read(member).decode("utf-8"))
            if isinstance(summary, dict):
                summaries[workflow.file] = summary
        except (ValueError, KeyError, StopIteration, zipfile.BadZipFile, json.JSONDecodeError):
            continue
    return summaries, ""


def _nested_timestamp(payload: dict, *path: str) -> datetime | None:
    value: Any = payload
    for key in path:
        if not isinstance(value, dict):
            return None
        value = value.get(key)
    return parse_gitea_time(value)


def _summary_proves_refresh(summary: dict, label: str, step: str) -> bool:
    if not summary.get("ok") or summary.get("skipped") or summary.get("dry_run"):
        return False
    step_ok = any(
        isinstance(row, dict)
        and row.get("step") == step
        and row.get("ok")
        for row in (summary.get("steps") or [])
    )
    if not step_ok:
        return False
    health = summary.get("health") or {}
    if label == "主播资料":
        return not (
            int(health.get("streamer_profile_missing") or 0)
            or int(health.get("streamer_live_probe_missing") or 0)
        )
    if label == "STRATZ Meta":
        return not health.get("stratz_meta_stale")
    if label == "STRATZ 对位":
        return int(health.get("stratz_matchup_stale") or 0) == 0
    return True


def fetch_production_health(
    now: datetime | None = None,
    refresh_summaries: dict[str, dict] | None = None,
) -> ProductionHealth:
    """Probe production payload freshness and the live-status Function."""
    current = now or datetime.now(timezone.utc)
    health = ProductionHealth()
    status, config_raw, _, _ = http_response_soft(f"{SITE_URL}/config.js")
    if status == 200:
        match = re.search(
            rb'SITE_VERSION\s*=\s*"([^"]+)"',
            config_raw,
        )
        if match:
            health.site_version = match.group(1).decode("utf-8", errors="replace")

    status, data_raw, _, _ = http_response_soft(f"{SITE_URL}/data.json")
    if status != 200:
        health.note = f"生产 data.json HTTP {status}"
        health.stale.append("data.json")
        return health
    try:
        payload = json.loads(data_raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        health.note = "生产 data.json 无法解析"
        health.stale.append("data.json")
        return health

    health.refresh_run_id = str((payload.get("meta") or {}).get("refresh_run_id") or "")
    summaries = refresh_summaries or {}
    checks = {
        "英雄统计": (
            ("hero_stats", "fetched_at"),
            36.0,
            "web-daily.yml",
            "fetch_hero_stats.py",
        ),
        "排行榜": (
            ("leaderboards", "fetched_at"),
            36.0,
            "web-daily.yml",
            "fetch_leaderboards.py",
        ),
        "近期比赛": (
            ("hero_matches", "meta", "fetched_at"),
            36.0,
            "web-daily.yml",
            "fetch_hero_matches.py",
        ),
        "明星比赛": (
            ("pro_matches", "meta", "fetched_at"),
            36.0,
            "web-daily.yml",
            "fetch_pro_matches.py",
        ),
        "主播资料": (
            ("streamers", "fetched_at"),
            36.0,
            "web-daily.yml",
            "fetch_streamers.py",
        ),
        "STRATZ Meta": (
            ("stratz_hero_meta", "fetched_at"),
            216.0,
            "web-weekly.yml",
            "fetch_stratz_meta.py",
        ),
        "STRATZ 对位": (
            ("stratz_matchup_tops", "fetched_at"),
            216.0,
            "web-weekly.yml",
            "fetch_stratz_meta.py",
        ),
        "英雄装备": (
            ("hero_items", "meta", "fetched_at"),
            216.0,
            "web-weekly.yml",
            "fetch_hero_items.py",
        ),
    }
    for label, (path, max_age, workflow_file, step) in checks.items():
        timestamp = _nested_timestamp(payload, *path)
        age = None
        if timestamp is not None:
            age = max(0.0, (current - timestamp.astimezone(timezone.utc)).total_seconds() / 3600)
        health.ages_hours[label] = age
        latest_refresh_ok = _summary_proves_refresh(
            summaries.get(workflow_file) or {},
            label,
            step,
        )
        if (age is None or age > max_age) and not latest_refresh_ok:
            health.stale.append(label)

    live_status, live_raw, live_type, live_headers = http_response_soft(
        f"{SITE_URL}/api/live-status"
    )
    try:
        live_payload = json.loads(live_raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        live_payload = None
    health.live_api_ok = (
        live_status == 200
        and live_type == "application/json"
        and isinstance(live_payload, dict)
        and live_headers.get("x-live-cache") not in ("error", "stale-override")
    )
    if not health.live_api_ok:
        health.stale.append("直播状态 API")
    return health


def cf_account_id() -> str:
    d = http_json(
        "https://api.cloudflare.com/client/v4/accounts",
        headers=cf_headers(),
    )
    assert isinstance(d, dict)
    rows = d.get("result") or []
    if not rows:
        raise SystemExit("no Cloudflare accounts")
    return rows[0]["id"]


def fetch_pages_deploys(day: date) -> list[PagesDeploy]:
    aid = cf_account_id()
    d = http_json(
        f"https://api.cloudflare.com/client/v4/accounts/{aid}/pages/projects/"
        f"{PAGES_PROJECT}/deployments",
        headers=cf_headers(),
    )
    assert isinstance(d, dict)
    start_utc, end_utc = day_window_utc(day)
    out: list[PagesDeploy] = []
    for x in d.get("result") or []:
        created = parse_gitea_time(x.get("created_on"))
        if created is None:
            continue
        created_utc = created.astimezone(timezone.utc)
        if not (start_utc <= created_utc < end_utc):
            continue
        if (x.get("environment") or "") != "production":
            continue
        stage = x.get("latest_stage") or {}
        trig = x.get("deployment_trigger") or {}
        meta = trig.get("metadata") or {}
        commit = str(meta.get("commit_hash") or "")[:10]
        out.append(
            PagesDeploy(
                created_on=created.astimezone(CST),
                status=str(stage.get("status") or "?"),
                commit=commit,
                url=str(x.get("url") or ""),
            )
        )
    out.sort(key=lambda p: p.created_on, reverse=True)
    return out


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 _fmt_workflow_line(w: WorkflowDayStatus) -> str:
    if w.note and not w.runs:
        return f"- **{w.label}**：{w.note}"
    parts: list[str] = []
    if w.ok_count:
        parts.append(f"成功 {w.ok_count}")
    if w.fail_count:
        parts.append(f"失败 {w.fail_count}")
    if w.running_count:
        parts.append(f"进行中 {w.running_count}")
    if not parts:
        parts.append(w.note or "无结论")
    # Show latest conclusion time if any
    latest = ""
    if w.runs:
        r0 = w.runs[0]
        t = r0.get("completed_at") or r0.get("started_at") or ""
        if t:
            dt = parse_gitea_time(t)
            if dt:
                latest = " · 最近 " + dt.astimezone(CST).strftime("%H:%M")
    return f"- **{w.label}**：{' / '.join(parts)}{latest}"


def _fmt_refresh_summary_result(summary: dict) -> str:
    """Human label for a refresh_web.py summary.json payload."""
    if summary.get("skipped"):
        reason = str(summary.get("skip_reason") or "").strip()
        return f"已跳过（{reason}）" if reason else "已跳过"
    if not summary.get("ok"):
        return "刷新失败"
    if summary.get("deployed"):
        return "已部署"
    return "无需部署"


def build_card(st: DayStats, extras: DigestExtras) -> 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(CST).strftime("%Y-%m-%d %H:%M CST")

    estimate = st.human_home_vis
    rum = st.rum_human

    # Human estimate = known-browser UA hits on `/` (SPA hash routes are not paths).
    traffic_md = (
        f"**真实访问（估）**：约 **{estimate}** 次首页打开\n"
        f"**RUM 页面浏览**：{rum}"
        + (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\n"
        f"**地区 Top**：{countries}\n"
        f"**状态码**：{status_line}"
    )

    wf_lines = [_fmt_workflow_line(w) for w in extras.workflows]
    if extras.gitea_note:
        wf_lines.append(f"- _{extras.gitea_note}_")
    for workflow_file, summary in extras.refresh_summaries.items():
        label = next(
            (label for file, label in REFRESH_WORKFLOWS if file == workflow_file),
            workflow_file,
        )
        result = _fmt_refresh_summary_result(summary)
        stale_count = int(
            (summary.get("health") or {}).get("stratz_matchup_stale") or 0
        )
        live_missing = int(
            (summary.get("health") or {}).get("streamer_live_probe_missing") or 0
        )
        wf_lines.append(
            f"- **{label}摘要**：{result}"
            + (f" · 对位陈旧 {stale_count}" if stale_count else "")
            + (f" · 直播探测缺失 {live_missing}" if live_missing else "")
        )

    ok_deploys = [p for p in extras.deploys if p.status == "success"]
    fail_deploys = [p for p in extras.deploys if p.status != "success"]
    if extras.deploys:
        deploy_lines = [
            f"**Pages**：成功 **{len(ok_deploys)}**"
            + (f" / 异常 {len(fail_deploys)}" if fail_deploys else "")
        ]
        for p in extras.deploys[:5]:
            mark = "ok" if p.status == "success" else "!"
            deploy_lines.append(
                f"- [{mark}] {p.created_on.strftime('%H:%M')} "
                f"`{p.commit or '?'}` {p.status}"
            )
        deploy_md = "\n".join(deploy_lines)
    else:
        deploy_md = "**Pages**：当日无新部署"

    health = extras.production_health
    production_lines: list[str] = []
    if health:
        newest = [
            f"{label} {age:.1f}h"
            for label, age in health.ages_hours.items()
            if age is not None
        ]
        production_lines.append(
            f"**生产探针**：{'正常' if not health.stale else '异常'}"
            f" · live API {'正常' if health.live_api_ok else '失败'}"
        )
        if newest:
            production_lines.append("- " + " / ".join(newest))
        if health.stale:
            production_lines.append("- **超时/异常**：" + "、".join(health.stale))
        if health.refresh_run_id:
            production_lines.append(f"- run `{health.refresh_run_id}`")

    refresh_md = (
        f"**版本**：`v{extras.site_version}`\n"
        f"**Actions**\n"
        + "\n".join(wf_lines)
        + "\n"
        + deploy_md
        + ("\n" + "\n".join(production_lines) if production_lines else "")
    )

    any_refresh_ok = any(w.ok_count > 0 for w in extras.workflows) or bool(ok_deploys)
    any_refresh_fail = any(w.fail_count > 0 for w in extras.workflows) or bool(
        fail_deploys
    )
    any_refresh_fail = any_refresh_fail or bool(health and health.stale)

    if any_refresh_fail:
        header_color = "red"
    elif estimate > 0 or any_refresh_ok:
        header_color = "blue"
    else:
        header_color = "grey"

    elements: list[dict] = [
        {
            "tag": "div",
            "text": {"tag": "lark_md", "content": "**访问**\n" + traffic_md},
        },
        {
            "tag": "div",
            "text": {"tag": "lark_md", "content": "**刷新与部署**\n" + refresh_md},
        },
        {
            "tag": "note",
            "elements": [
                {
                    "tag": "plain_text",
                    "content": (
                        f"{HOST} · 估数=已知浏览器 UA 打开 / · {now}"
                    ),
                }
            ],
        },
        {
            "tag": "action",
            "actions": [
                {
                    "tag": "button",
                    "text": {"tag": "plain_text", "content": "打开站点"},
                    "type": "primary",
                    "url": SITE_URL,
                },
                {
                    "tag": "button",
                    "text": {"tag": "plain_text", "content": "Gitea Actions"},
                    "type": "default",
                    "url": f"{gitea_base()}/{GITEA_OWNER}/{GITEA_REPO}/actions",
                },
            ],
        },
    ]

    return {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {
                    "tag": "plain_text",
                    "content": f"上分帝 Web 日报 · {st.day.isoformat()}",
                },
                "template": header_color,
            },
            "elements": elements,
        },
    }


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


def parse_day(s: str | None) -> date:
    if not s:
        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)
    refresh_day = datetime.now(CST).date()
    zid = zone_id()
    st = fetch_day(zid, day)
    workflows, gitea_note = fetch_workflow_statuses(refresh_day)
    refresh_summaries, summary_note = fetch_refresh_summaries(workflows)
    if summary_note:
        gitea_note = "；".join(x for x in (gitea_note, summary_note) if x)
    deploys = fetch_pages_deploys(day)
    production_health = fetch_production_health(refresh_summaries=refresh_summaries)
    extras = DigestExtras(
        site_version=production_health.site_version,
        workflows=workflows,
        deploys=deploys,
        gitea_note=gitea_note,
        refresh_summaries=refresh_summaries,
        production_health=production_health,
    )
    payload = build_card(st, extras)

    print(
        f"{HOST} {day}: human_home_vis={st.human_home_vis} "
        f"rum_human={st.rum_human} all_req={st.all_req} "
        f"deploys={len(deploys)} "
        f"wf_ok={sum(w.ok_count for w in workflows)} "
        f"wf_fail={sum(w.fail_count for w in workflows)} "
        f"v={extras.site_version}",
        flush=True,
    )

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

    resp = post_feishu(payload)
    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())