#!/usr/bin/env python3
"""Refresh the local all-project DAU dashboard with aggregate, read-only queries."""

from __future__ import annotations

import argparse
import csv
import datetime as dt
import html
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
from typing import Any
from urllib.parse import urlparse


ROOT = Path(__file__).resolve().parents[3]
TASK_DIR = Path(__file__).resolve().parent
REGISTRY = TASK_DIR / "project-registry.csv"
SOURCE_HTML = ROOT / "work_store/product-strategy/20260720-smart-canteen-nutrition-health-platform-direction/index.html"
DEFAULT_STORE = Path("/Users/jack/code/010-cpt/008-zhct/zhct/zhctproject/store")
DEFAULT_OUTPUT = ROOT / "work_store/data-queries/runtime/all-project-dau-dashboard/latest"
SNAPSHOT_DIR = ROOT / "work_store/data-queries/published/2026-07-20-health-product-metrics"
JSR_ENV = ROOT / "config/local/jsr-db.env"
LOCAL_AUTH = ROOT / "config/local/all-project-dashboard-auth.csv"
AUTH_TEMPLATE = TASK_DIR / "local-auth-template.csv"
WE_ANALYSIS_DEFAULT = "https://wedata.weixin.qq.com/mp2/login"

HOST_OVERRIDES = {
    "rds3nh30726o5tko5b9g502.mysql.rds.aliyuncs.com": "rds3nh30726o5tko5b9gro.mysql.rds.aliyuncs.com",
    "rm-2ze9n39s9f40s05ox.mysql.rds.aliyuncs.com": "rm-2ze9n39s9f40s05oxjo.mysql.rds.aliyuncs.com",
    "rm-2zer9po1sbfwqr6p2.mysql.rds.aliyuncs.com": "rm-2zer9po1sbfwqr6p20o.mysql.rds.aliyuncs.com",
}

OUTPUT_FIELDS = [
    "project_key", "project_name", "data_status", "status_detail", "snapshot_at",
    "system_url", "auth_configured", "we_analysis_url", "database_name",
    "registered_users", "all_orders", "all_amount", "valid_orders", "valid_amount",
    "first_meal_date", "last_meal_date", "today_dau", "latest_valid_meal_date",
    "latest_day_dau", "dau7_avg", "dau7_peak", "mau7", "dau30_avg",
    "dau30_peak", "mau30", "orders30", "amount30",
]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Refresh all-project smart-canteen DAU dashboard")
    parser.add_argument("--as-of", default=dt.date.today().isoformat(), help="statistics date, YYYY-MM-DD")
    parser.add_argument("--store-repo", type=Path, default=DEFAULT_STORE)
    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument("--auth-file", type=Path, default=LOCAL_AUTH)
    parser.add_argument("--timeout", type=int, default=20, help="per-query timeout in seconds")
    parser.add_argument("--open-secure", action="store_true", help="open a dedicated Chrome profile with local autofill")
    return parser.parse_args()


def read_env(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    if not path.exists():
        return values
    for raw in path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip().strip("'\"")
    return values


def read_registry() -> list[dict[str, str]]:
    with REGISTRY.open(encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))


def ensure_local_auth(path: Path) -> None:
    """Create the ignored local credential file once, without ever overwriting it."""
    if path.exists():
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    shutil.copyfile(AUTH_TEMPLATE, path)
    path.chmod(0o600)


def read_local_auth(path: Path) -> dict[str, dict[str, str]]:
    ensure_local_auth(path)
    with path.open(encoding="utf-8-sig", newline="") as handle:
        return {row["project_key"]: row for row in csv.DictReader(handle) if row.get("project_key")}


def mysql_binary() -> str:
    candidates = [shutil.which("mysql"), "/usr/local/mysql/bin/mysql", "/opt/homebrew/bin/mysql"]
    for candidate in candidates:
        if candidate and Path(candidate).exists():
            return str(candidate)
    raise RuntimeError("mysql_client_missing")


CONFIG_PATTERN = re.compile(
    r"\$config\['(?P<key>hostname|database|username|password)'\]\s*=\s*'(?P<value>[^']*)'\s*;"
)
DOMAIN_PATTERN = re.compile(r"\$config\['domain_name'\]\s*=\s*'(?P<value>[^']*)'\s*;")


def parse_php_config(content: str) -> dict[str, str]:
    values = {match.group("key"): match.group("value") for match in CONFIG_PATTERN.finditer(content)}
    required = {"hostname", "database", "username", "password"}
    if not required.issubset(values):
        raise RuntimeError("config_fields_missing")
    values["hostname"] = HOST_OVERRIDES.get(values["hostname"], values["hostname"])
    return values


def load_config(store_repo: Path, config_dir: str) -> tuple[dict[str, str] | None, str]:
    relative = Path("application/config/prod") / config_dir / "database.php"
    local_path = store_repo / relative
    if local_path.exists():
        return parse_php_config(local_path.read_text(encoding="utf-8")), "local_config"
    if (store_repo / ".git").exists():
        result = subprocess.run(
            ["git", "-C", str(store_repo), "show", f"origin/master:{relative.as_posix()}"],
            text=True,
            capture_output=True,
            timeout=10,
            check=False,
        )
        if result.returncode == 0 and result.stdout:
            return parse_php_config(result.stdout), "origin_master_config"
    return None, "config_missing"


def read_store_file(store_repo: Path, relative: Path) -> str:
    local_path = store_repo / relative
    if local_path.exists():
        return local_path.read_text(encoding="utf-8")
    if (store_repo / ".git").exists():
        result = subprocess.run(
            ["git", "-C", str(store_repo), "show", f"origin/master:{relative.as_posix()}"],
            text=True, capture_output=True, timeout=10, check=False,
        )
        if result.returncode == 0:
            return result.stdout
    return ""


def inferred_system_url(store_repo: Path, config_dir: str) -> str:
    content = read_store_file(store_repo, Path("application/config/prod") / config_dir / "config.php")
    match = DOMAIN_PATTERN.search(content)
    if not match:
        return ""
    return match.group("value").rstrip("/") + "/dist/#/login"


def project_access(
    project: dict[str, str], local_auth: dict[str, dict[str, str]], store_repo: Path,
) -> dict[str, str]:
    private = local_auth.get(project["project_key"], {})
    system_url = private.get("system_url", "").strip() or inferred_system_url(store_repo, project["config_dir"])
    return {
        "system_url": system_url,
        "username": private.get("username", "").strip(),
        "password": private.get("password", ""),
        "we_analysis_url": private.get("we_analysis_url", "").strip() or WE_ANALYSIS_DEFAULT,
    }


def jsr_readonly_config(database: str) -> dict[str, str] | None:
    env = read_env(JSR_ENV)
    if not all(env.get(key) for key in ("JSR_DB_HOST", "JSR_DB_USER", "JSR_DB_PASSWORD")):
        return None
    return {
        "hostname": env["JSR_DB_HOST"],
        "database": database,
        "username": env["JSR_DB_USER"],
        "password": env["JSR_DB_PASSWORD"],
    }


def run_mysql(binary: str, config: dict[str, str], sql: str, timeout: int) -> str:
    command = [
        binary, "--connect-timeout=8", "--ssl-mode=DISABLED", "--default-character-set=utf8mb4",
        "-h", config["hostname"], "-u", config["username"], config["database"],
        "--batch", "--raw", "--skip-column-names", "-e", sql,
    ]
    env = os.environ.copy()
    env["MYSQL_PWD"] = config["password"]
    result = subprocess.run(command, text=True, capture_output=True, env=env, timeout=timeout, check=False)
    if result.returncode != 0:
        message = (result.stderr or result.stdout).strip()
        message = re.sub(r"for user .*? \(using password", "for user REDACTED (using password", message)
        raise RuntimeError(message or f"mysql_exit_{result.returncode}")
    return result.stdout.strip()


def error_label(message: str) -> str:
    lowered = message.lower()
    if "config_missing" in lowered:
        return "本机缺少项目配置"
    if "timed out" in lowered or "timeout" in lowered:
        return "连接超时，通常为内网不可达"
    if "connection refused" in lowered or "can't connect" in lowered:
        return "数据库地址当前不可达"
    if "access denied" in lowered or "1044" in lowered or "1045" in lowered:
        return "只读账号无该库权限"
    if "mysql_client_missing" in lowered:
        return "本机缺少 mysql 客户端"
    return "只读查询失败"


def as_number(value: str, integer: bool = False) -> int | float:
    if value in ("", "NULL", None):
        return 0 if integer else 0.0
    return int(float(value)) if integer else float(value)


def metric_evidence(valid: str, dau_valid: str, as_of: dt.date, database: str) -> dict[str, dict[str, str]]:
    end = as_of.isoformat()
    start7 = (as_of - dt.timedelta(days=6)).isoformat()
    start30 = (as_of - dt.timedelta(days=29)).isoformat()
    source = f"生产数据库 {database}｜ydy_staff / ydy_meal_order"
    return {
        "today_dau": {
            "label": "当前日 DAU", "period": end,
            "formula": "统计日内满足有效订单口径的 user_id 去重人数。",
            "source": source,
            "sql": f"SELECT COUNT(DISTINCT user_id) FROM ydy_meal_order WHERE {dau_valid} AND meal_date='{end}';",
        },
        "latest_valid_meal_date": {
            "label": "最近有单日", "period": f"截至 {end}",
            "formula": "满足有效订单及 user_id>0 口径的最大 meal_date。", "source": source,
            "sql": f"SELECT MAX(meal_date) FROM ydy_meal_order WHERE {dau_valid};",
        },
        "latest_day_dau": {
            "label": "最近有单日日活", "period": f"截至 {end}",
            "formula": "先找到最近有效用餐日，再对该日 user_id 去重。", "source": source,
            "sql": f"SELECT COUNT(DISTINCT user_id) FROM ydy_meal_order WHERE {dau_valid} AND meal_date=(SELECT MAX(meal_date) FROM ydy_meal_order WHERE {dau_valid});",
        },
        "dau7_avg": {
            "label": "7 日日均 DAU", "period": f"{start7} 至 {end}",
            "formula": "7个自然日的逐日去重人数之和 ÷ 7；无订单日按0计。", "source": source,
            "sql": f"SELECT ROUND(COUNT(DISTINCT CONCAT(meal_date,'#',user_id))/7,2) FROM ydy_meal_order WHERE {dau_valid} AND meal_date BETWEEN '{start7}' AND '{end}';",
        },
        "dau7_peak": {
            "label": "7 日峰值 DAU", "period": f"{start7} 至 {end}",
            "formula": "最近7个自然日逐日去重人数的最大值。", "source": source,
            "sql": f"SELECT MAX(dau) FROM (SELECT meal_date,COUNT(DISTINCT user_id) dau FROM ydy_meal_order WHERE {dau_valid} AND meal_date BETWEEN '{start7}' AND '{end}' GROUP BY meal_date) d;",
        },
        "dau30_avg": {
            "label": "30 日日均 DAU", "period": f"{start30} 至 {end}",
            "formula": "30个自然日的逐日去重人数之和 ÷ 30；无订单日按0计。", "source": source,
            "sql": f"SELECT ROUND(COUNT(DISTINCT CONCAT(meal_date,'#',user_id))/30,2) FROM ydy_meal_order WHERE {dau_valid} AND meal_date BETWEEN '{start30}' AND '{end}';",
        },
        "dau30_peak": {
            "label": "30 日峰值 DAU", "period": f"{start30} 至 {end}",
            "formula": "最近30个自然日逐日去重人数的最大值。", "source": source,
            "sql": f"SELECT MAX(dau) FROM (SELECT meal_date,COUNT(DISTINCT user_id) dau FROM ydy_meal_order WHERE {dau_valid} AND meal_date BETWEEN '{start30}' AND '{end}' GROUP BY meal_date) d;",
        },
        "mau30": {
            "label": "30 日活跃用户", "period": f"{start30} 至 {end}",
            "formula": "最近30个自然日内满足有效订单口径的 user_id 跨日去重人数。", "source": source,
            "sql": f"SELECT COUNT(DISTINCT user_id) FROM ydy_meal_order WHERE {dau_valid} AND meal_date BETWEEN '{start30}' AND '{end}';",
        },
        "registered_users": {
            "label": "累计用户记录", "period": "当前库全量",
            "formula": "人员表 ydy_staff 的记录数；不是跨项目唯一自然人。", "source": source,
            "sql": "SELECT COUNT(*) FROM ydy_staff;",
        },
        "all_orders": {
            "label": "全量订单", "period": "订单表全量历史",
            "formula": "ydy_meal_order 全表记录数，不应用支付或完成状态过滤。", "source": source,
            "sql": "SELECT COUNT(*) FROM ydy_meal_order;",
        },
        "all_amount": {
            "label": "全量流水", "period": "订单表全量历史",
            "formula": "ydy_meal_order.total_price 全量求和；不等于公司确认收入。", "source": source,
            "sql": "SELECT COALESCE(SUM(total_price),0) FROM ydy_meal_order;",
        },
        "valid_orders": {
            "label": "有效订单", "period": f"截至 {end}",
            "formula": "按项目实际存在字段组合已支付、未删除、已完成，并限制统计日。", "source": source,
            "sql": f"SELECT COUNT(*) FROM ydy_meal_order WHERE {valid};",
        },
        "valid_amount": {
            "label": "有效流水", "period": f"截至 {end}",
            "formula": "有效订单范围内 total_price 求和；不等于公司确认收入。", "source": source,
            "sql": f"SELECT COALESCE(SUM(total_price),0) FROM ydy_meal_order WHERE {valid};",
        },
    }


def fill_daily_series(raw: str, start: dt.date, end: dt.date) -> list[dict[str, Any]]:
    observed: dict[str, dict[str, Any]] = {}
    for line in raw.splitlines():
        if not line:
            continue
        values = line.split("\t")
        observed[values[0]] = {
            "date": values[0], "dau": as_number(values[1], True),
            "valid_orders": as_number(values[2], True), "valid_amount": as_number(values[3]),
        }
    rows: list[dict[str, Any]] = []
    current = start
    while current <= end:
        key = current.isoformat()
        rows.append(observed.get(key, {"date": key, "dau": 0, "valid_orders": 0, "valid_amount": 0.0}))
        current += dt.timedelta(days=1)
    return rows


def query_project(
    binary: str,
    project: dict[str, str],
    config: dict[str, str],
    config_source: str,
    as_of: dt.date,
    timeout: int,
) -> dict[str, Any]:
    identity = run_mysql(
        binary,
        config,
        "SET SESSION TRANSACTION READ ONLY; START TRANSACTION; "
        "SELECT IF(CURRENT_USER() IS NULL,0,1),DATABASE(); COMMIT;",
        timeout,
    ).splitlines()[-1].split("\t")
    if len(identity) != 2 or identity[0] != "1" or identity[1] != config["database"]:
        raise RuntimeError("connection_identity_mismatch")

    columns_output = run_mysql(binary, config, "SHOW COLUMNS FROM ydy_meal_order;", timeout)
    columns = {line.split("\t", 1)[0] for line in columns_output.splitlines() if line}
    required = {"meal_date", "user_id", "total_price"}
    if not required.issubset(columns):
        raise RuntimeError("order_schema_missing_required_fields")

    filters = []
    if "pay_status" in columns:
        filters.append("pay_status = 20")
    if "order_status" in columns:
        filters.append("order_status = 30")
    if "is_delete" in columns:
        filters.append("is_delete = 0")
    filters.append(f"meal_date <= '{as_of.isoformat()}'")
    valid = " AND ".join(filters)
    dau_valid = f"{valid} AND user_id > 0"
    start7 = (as_of - dt.timedelta(days=6)).isoformat()
    start30 = (as_of - dt.timedelta(days=29)).isoformat()
    end = as_of.isoformat()

    sql = f"""
SET SESSION TRANSACTION READ ONLY;
START TRANSACTION WITH CONSISTENT SNAPSHOT;
SELECT
  (SELECT COUNT(*) FROM ydy_staff),
  COUNT(*), COALESCE(SUM(total_price),0), MIN(meal_date), MAX(meal_date),
  SUM(CASE WHEN {valid} THEN 1 ELSE 0 END),
  COALESCE(SUM(CASE WHEN {valid} THEN total_price ELSE 0 END),0),
  COUNT(DISTINCT CASE WHEN {dau_valid} AND meal_date='{end}' THEN user_id END),
  MAX(CASE WHEN {dau_valid} THEN meal_date END),
  COALESCE((SELECT COUNT(DISTINCT user_id) FROM ydy_meal_order
            WHERE {dau_valid} AND meal_date=(SELECT MAX(meal_date) FROM ydy_meal_order WHERE {dau_valid})),0),
  ROUND(COUNT(DISTINCT CASE WHEN {dau_valid} AND meal_date BETWEEN '{start7}' AND '{end}'
             THEN CONCAT(meal_date,'#',user_id) END)/7,2),
  COALESCE((SELECT MAX(dau) FROM (SELECT meal_date,COUNT(DISTINCT user_id) dau
            FROM ydy_meal_order WHERE {dau_valid} AND meal_date BETWEEN '{start7}' AND '{end}' GROUP BY meal_date) d7),0),
  COUNT(DISTINCT CASE WHEN {dau_valid} AND meal_date BETWEEN '{start7}' AND '{end}' THEN user_id END),
  ROUND(COUNT(DISTINCT CASE WHEN {dau_valid} AND meal_date BETWEEN '{start30}' AND '{end}'
             THEN CONCAT(meal_date,'#',user_id) END)/30,2),
  COALESCE((SELECT MAX(dau) FROM (SELECT meal_date,COUNT(DISTINCT user_id) dau
            FROM ydy_meal_order WHERE {dau_valid} AND meal_date BETWEEN '{start30}' AND '{end}' GROUP BY meal_date) d30),0),
  COUNT(DISTINCT CASE WHEN {dau_valid} AND meal_date BETWEEN '{start30}' AND '{end}' THEN user_id END),
  SUM(CASE WHEN {valid} AND meal_date BETWEEN '{start30}' AND '{end}' THEN 1 ELSE 0 END),
  COALESCE(SUM(CASE WHEN {valid} AND meal_date BETWEEN '{start30}' AND '{end}' THEN total_price ELSE 0 END),0)
FROM ydy_meal_order;
COMMIT;
"""
    values = run_mysql(binary, config, sql, timeout).splitlines()[-1].split("\t")
    if len(values) != 18:
        raise RuntimeError("unexpected_metric_column_count")
    daily_sql = f"""
SET SESSION TRANSACTION READ ONLY;
SELECT meal_date,COUNT(DISTINCT CASE WHEN user_id > 0 THEN user_id END),COUNT(*),COALESCE(SUM(total_price),0)
FROM ydy_meal_order
WHERE {valid} AND meal_date BETWEEN '{start30}' AND '{end}'
GROUP BY meal_date ORDER BY meal_date;
"""
    daily_output = run_mysql(binary, config, daily_sql, timeout)
    return {
        "project_key": project["project_key"],
        "project_name": project["project_name"],
        "data_status": "live",
        "status_detail": f"生产库实时只读｜{config_source}",
        "snapshot_at": dt.datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %z"),
        "database_name": config["database"],
        "registered_users": as_number(values[0], True),
        "all_orders": as_number(values[1], True),
        "all_amount": as_number(values[2]),
        "first_meal_date": "" if values[3] == "NULL" else values[3],
        "last_meal_date": "" if values[4] == "NULL" else values[4],
        "valid_orders": as_number(values[5], True),
        "valid_amount": as_number(values[6]),
        "today_dau": as_number(values[7], True),
        "latest_valid_meal_date": "" if values[8] == "NULL" else values[8],
        "latest_day_dau": as_number(values[9], True),
        "dau7_avg": as_number(values[10]),
        "dau7_peak": as_number(values[11], True),
        "mau7": as_number(values[12], True),
        "dau30_avg": as_number(values[13]),
        "dau30_peak": as_number(values[14], True),
        "mau30": as_number(values[15], True),
        "orders30": as_number(values[16], True),
        "amount30": as_number(values[17]),
        "_daily": fill_daily_series(daily_output, as_of - dt.timedelta(days=29), as_of),
        "_metric_evidence": metric_evidence(valid, dau_valid, as_of, config["database"]),
    }


def latest_file(pattern: str) -> Path | None:
    files = sorted(SNAPSHOT_DIR.glob(pattern))
    return files[-1] if files else None


def load_snapshot() -> dict[str, dict[str, Any]]:
    path = latest_file("项目汇总_*.csv")
    if not path:
        return {}
    snapshot: dict[str, dict[str, Any]] = {}
    with path.open(encoding="utf-8-sig", newline="") as handle:
        for row in csv.DictReader(handle):
            if row.get("查询状态") != "成功":
                continue
            snapshot[row["配置目录"]] = {
                "registered_users": as_number(row.get("用户数", "0"), True),
                "all_orders": as_number(row.get("order_num", "0"), True),
                "all_amount": as_number(row.get("total_price", "0")),
                "first_meal_date": row.get("first_meal_date", ""),
                "last_meal_date": row.get("last_meal_date", ""),
                "today_dau": 0,
                "latest_valid_meal_date": row.get("最近有订单日期", ""),
                "latest_day_dau": as_number(row.get("最近有单日活人数", "0"), True),
                "dau7_avg": as_number(row.get("近7日平均日活_按自然日", "0")),
                "dau7_peak": as_number(row.get("近7日最高日活", "0"), True),
                "mau7": 0, "dau30_avg": 0.0, "dau30_peak": 0, "mau30": 0,
                "orders30": 0, "amount30": 0.0,
                "valid_orders": "", "valid_amount": "",
            }
    return snapshot


def unavailable_row(project: dict[str, str], detail: str, as_of: dt.date) -> dict[str, Any]:
    row = {field: "" for field in OUTPUT_FIELDS}
    row.update({
        "project_key": project["project_key"], "project_name": project["project_name"],
        "data_status": "unavailable", "status_detail": detail,
        "snapshot_at": as_of.isoformat(),
    })
    return row


def snapshot_evidence(row: dict[str, Any]) -> dict[str, dict[str, str]]:
    labels = {
        "today_dau": "当前日 DAU", "latest_valid_meal_date": "最近有单日",
        "latest_day_dau": "最近有单日日活", "dau7_avg": "7 日日均 DAU",
        "dau7_peak": "7 日峰值 DAU", "dau30_avg": "30 日日均 DAU",
        "dau30_peak": "30 日峰值 DAU", "mau30": "30 日活跃用户",
        "registered_users": "累计用户记录", "all_orders": "全量订单",
        "all_amount": "全量流水", "valid_orders": "有效订单", "valid_amount": "有效流水",
    }
    return {
        key: {
            "label": label,
            "period": str(row.get("snapshot_at", "历史快照")),
            "formula": "来自同源历史快照；当前环境未能重新连接生产库，因此只能核对快照源文件，不能作为实时验证。",
            "source": str(latest_file("项目汇总_*.csv") or "历史项目汇总 CSV"),
            "sql": "-- 当前为历史快照，没有本次实时SQL执行证据。",
        }
        for key, label in labels.items()
    }


def apply_access(row: dict[str, Any], access: dict[str, str]) -> None:
    row["system_url"] = access["system_url"]
    row["we_analysis_url"] = access["we_analysis_url"]
    row["auth_configured"] = "yes" if access["username"] and access["password"] else "no"
    row["_auth"] = {"username": access["username"], "password": access["password"]}


def number(value: Any, decimals: int = 0) -> str:
    if value in ("", None):
        return "—"
    return f"{float(value):,.{decimals}f}" if decimals else f"{int(float(value)):,}"


def money(value: Any) -> str:
    return "—" if value in ("", None) else f"{float(value):,.2f}"


def status_chip(row: dict[str, Any]) -> str:
    status = row["data_status"]
    label = {"live": "实时", "snapshot": "快照", "unavailable": "不可达"}[status]
    return f'<span class="status status-{status}" title="{html.escape(str(row["status_detail"]))}">{label}</span>'


def metric_button(row: dict[str, Any], metric: str, rendered: str) -> str:
    if row.get(metric) in ("", None):
        return "—"
    return (
        f'<button class="metric-link" type="button" data-project="{html.escape(row["project_key"], quote=True)}" '
        f'data-metric="{html.escape(metric, quote=True)}" title="点击查看口径、SQL和明细证据">{rendered}</button>'
    )


def render_html(rows: list[dict[str, Any]], as_of: dt.date, output: Path) -> None:
    source = SOURCE_HTML.read_text(encoding="utf-8")
    live = [row for row in rows if row["data_status"] == "live"]
    available = [row for row in rows if row["data_status"] in ("live", "snapshot")]
    total_users = sum(int(row["registered_users"] or 0) for row in available)
    total_orders = sum(int(row["all_orders"] or 0) for row in available)
    total_valid_amount = sum(float(row["valid_amount"] or 0) for row in available)
    today_dau = sum(int(row["today_dau"] or 0) for row in available)
    avg30 = sum(float(row["dau30_avg"] or 0) for row in available)
    ordered = sum(1 for row in available if int(row["all_orders"] or 0) > 0)
    auth_count = sum(1 for row in rows if row.get("auth_configured") == "yes")
    url_count = sum(1 for row in rows if row.get("system_url"))

    stamp = dt.datetime.now().astimezone().strftime("%Y-%m-%d %H:%M")
    source = re.sub(
        r'<p class="stamp">.*?</p>',
        f'<p class="stamp">数据刷新：{stamp}｜统计日：{as_of.isoformat()}｜17 项目统一只读查询｜交易流水非公司营收</p>',
        source,
        count=1,
        flags=re.S,
    )
    metrics = f"""<div class="metrics">
          <div class="metric"><span>项目总数</span><b>{len(rows)}</b><small>完整项目清单，不隐藏失败项目</small></div>
          <div class="metric"><span>实时查询成功</span><b>{len(live)}</b><small>当前本机可直连生产库</small></div>
          <div class="metric"><span>数据可用项目</span><b>{len(available)}</b><small>实时 + 同源历史快照</small></div>
          <div class="metric"><span>已有订单项目</span><b>{ordered}</b><small>全量订单数大于 0</small></div>
          <div class="metric"><span>当前日 DAU 合计</span><b>{today_dau:,}</b><small>{as_of.isoformat()} 项目内去重后相加</small></div>
          <div class="metric"><span>近 30 日日均 DAU</span><b>{avg30:,.0f}</b><small>各项目自然日日均相加</small></div>
          <div class="metric"><span>累计用户记录</span><b>{total_users:,}</b><small>跨项目可能重复</small></div>
          <div class="metric"><span>累计订单 / 有效流水</span><b>{total_orders:,}</b><small>有效流水 {total_valid_amount/10000:,.2f} 万元</small></div>
        </div>"""
    source = re.sub(r'<div class="metrics">.*?</div>\s*</aside>', metrics + "\n      </aside>", source, count=1, flags=re.S)

    head = """<thead><tr>
                <th>项目 / 状态</th><th>当前日DAU</th><th>最近有单日</th><th>该日日活</th>
                <th>7日日均</th><th>7日峰值</th><th>30日日均</th><th>30日峰值</th><th>30日活跃用户</th>
                <th>累计用户</th><th>全量订单</th><th>全量流水（元）</th><th>有效订单</th><th>有效流水（元）</th>
              </tr></thead>"""
    body_rows = []
    for row in rows:
        system_url = str(row.get("system_url") or "")
        we_url = str(row.get("we_analysis_url") or WE_ANALYSIS_DEFAULT)
        if system_url:
            project_title = (
                f'<a class="project-link" href="{html.escape(system_url, quote=True)}" target="_blank" rel="noopener noreferrer" '
                f'title="打开项目PC管理端">{html.escape(row["project_name"])} <span aria-hidden="true">↗</span></a>'
            )
        else:
            project_title = f'<strong>{html.escape(row["project_name"])}</strong>'
        auth_label = "登录信息" if row.get("auth_configured") == "yes" else "待配账号"
        project_cell = (
            f'{project_title}<br>{status_chip(row)}'
            f'<div class="project-actions"><button type="button" data-login="{html.escape(row["project_key"], quote=True)}">{auth_label}</button>'
            f'<a href="{html.escape(we_url, quote=True)}" target="_blank" rel="noopener noreferrer">We分析 ↗</a></div>'
        )
        body_rows.append(
            "<tr>"
            f"<td>{project_cell}</td><td>{metric_button(row, 'today_dau', number(row['today_dau']))}</td>"
            f"<td>{metric_button(row, 'latest_valid_meal_date', html.escape(str(row['latest_valid_meal_date'] or '—')))}</td>"
            f"<td>{metric_button(row, 'latest_day_dau', number(row['latest_day_dau']))}</td>"
            f"<td>{metric_button(row, 'dau7_avg', number(row['dau7_avg'],2))}</td>"
            f"<td>{metric_button(row, 'dau7_peak', number(row['dau7_peak']))}</td>"
            f"<td>{metric_button(row, 'dau30_avg', number(row['dau30_avg'],2))}</td>"
            f"<td>{metric_button(row, 'dau30_peak', number(row['dau30_peak']))}</td>"
            f"<td>{metric_button(row, 'mau30', number(row['mau30']))}</td>"
            f"<td>{metric_button(row, 'registered_users', number(row['registered_users']))}</td>"
            f"<td>{metric_button(row, 'all_orders', number(row['all_orders']))}</td>"
            f"<td>{metric_button(row, 'all_amount', money(row['all_amount']))}</td>"
            f"<td>{metric_button(row, 'valid_orders', number(row['valid_orders']))}</td>"
            f"<td>{metric_button(row, 'valid_amount', money(row['valid_amount']))}</td></tr>"
        )
    table = head + "\n            <tbody>\n" + "\n".join(body_rows) + "\n            </tbody>"
    source = re.sub(r'<thead>.*?</thead>\s*<tbody>.*?</tbody>', table, source, count=1, flags=re.S)
    toolbar = f"""<div class="verification-toolbar">
        <div><strong>数据可核验</strong><span>点击任意数字查看口径、SQL、来源和30日逐日明细。</span></div>
        <div class="toolbar-actions"><span>{url_count}/17 已配置系统地址</span><span>{auth_count}/17 已配置本机登录</span>
        <a href="{WE_ANALYSIS_DEFAULT}" target="_blank" rel="noopener noreferrer">打开 We分析 ↗</a></div>
      </div>"""
    source = source.replace('<div class="table-card">', toolbar + '\n      <div class="table-card">', 1)
    source = source.replace(
        "日活口径：按最近有效用餐日，对 `ydy_meal_order.user_id` 去重；累计流水为订单表 `total_price` 合计。",
        "日活已扩展为全部 17 个项目：当前日、最近有单日、7/30 日均与峰值、30 日活跃用户；有效流水采用已支付、未删除、已完成（字段存在时）的订单口径。",
    )
    failures = [row for row in rows if row["data_status"] == "unavailable"]
    failure_text = "；".join(f"{row['project_name']}（{row['status_detail']}）" for row in failures) or "无"
    source = re.sub(
        r'<div class="note">.*?</div>',
        f'<div class="note"><strong>全项目状态：</strong>17 个项目均已入表。当前不可达：{html.escape(failure_text)}。不可达不等于 0，恢复网络或配置后重新运行即可刷新。</div>',
        source,
        count=1,
        flags=re.S,
    )
    extra_css = """
    .metric:nth-child(7) { border-top: 4px solid #7c3aed; }
    .metric:nth-child(8) { border-top: 4px solid #475569; }
    .status { display:inline-flex; margin-top:6px; padding:2px 8px; border-radius:999px; font-size:11px; font-weight:800; }
    .status-live { color:#0f6f50; background:#e8f7f1; }
    .status-snapshot { color:#8a5a00; background:#fff4d8; }
    .status-unavailable { color:#9f1239; background:#ffe4e9; }
    td:first-child strong { white-space:normal; }
    .project-link { color:var(--ink); text-decoration:none; font-weight:850; white-space:normal; }
    .project-link:hover { color:var(--blue); text-decoration:underline; }
    .project-actions { display:flex; flex-wrap:wrap; gap:6px; margin-top:7px; }
    .project-actions button, .project-actions a, .toolbar-actions a { border:1px solid var(--line); background:#fff; color:var(--muted); border-radius:999px; padding:3px 8px; font:inherit; font-size:11px; font-weight:750; text-decoration:none; cursor:pointer; }
    .project-actions button:hover, .project-actions a:hover, .toolbar-actions a:hover { border-color:var(--blue); color:var(--blue); }
    .metric-link { border:0; border-bottom:1px dashed #94a3b8; background:transparent; color:var(--ink); font:inherit; font-weight:750; padding:2px 1px; cursor:pointer; white-space:nowrap; }
    .metric-link:hover, .metric-link:focus { color:var(--blue); border-color:var(--blue); outline:none; }
    .verification-toolbar { display:flex; justify-content:space-between; gap:18px; align-items:center; margin:0 0 12px; padding:14px 16px; border:1px solid #bfdbfe; border-radius:14px; background:#eff6ff; }
    .verification-toolbar strong { display:block; margin-bottom:4px; color:#1e3a8a; }
    .verification-toolbar span { color:var(--muted); font-size:12px; }
    .toolbar-actions { display:flex; align-items:center; flex-wrap:wrap; justify-content:flex-end; gap:8px; }
    .drawer-backdrop { position:fixed; inset:0; background:rgba(15,23,42,.32); opacity:0; pointer-events:none; transition:.18s ease; z-index:50; }
    .drawer-backdrop.open { opacity:1; pointer-events:auto; }
    .evidence-drawer { position:fixed; top:0; right:0; width:min(620px,94vw); height:100vh; background:#fff; box-shadow:-24px 0 60px rgba(15,23,42,.18); transform:translateX(105%); transition:.22s ease; z-index:51; overflow:auto; }
    .evidence-drawer.open { transform:translateX(0); }
    .drawer-inner { padding:26px; }
    .drawer-head { display:flex; justify-content:space-between; gap:16px; align-items:flex-start; padding-bottom:16px; border-bottom:1px solid var(--line); }
    .drawer-head h3 { margin:4px 0 0; font-size:22px; }
    .drawer-head p { margin:0; color:var(--muted); font-size:12px; }
    .drawer-close { border:0; background:#eef2f7; border-radius:50%; width:34px; height:34px; cursor:pointer; font-size:18px; }
    .evidence-block { margin-top:18px; }
    .evidence-block h4 { margin:0 0 8px; font-size:13px; color:var(--muted); text-transform:uppercase; letter-spacing:.04em; }
    .evidence-block p { margin:0; line-height:1.75; }
    .evidence-meta { display:grid; grid-template-columns:1fr 1fr; gap:10px; }
    .evidence-meta div { padding:10px 12px; border-radius:10px; background:#f8fafc; }
    .evidence-meta span { display:block; color:var(--muted); font-size:11px; margin-bottom:4px; }
    pre.evidence-sql { margin:0; padding:14px; border-radius:12px; background:#0f172a; color:#dbeafe; overflow:auto; white-space:pre-wrap; word-break:break-word; font-size:12px; line-height:1.65; }
    .copy-btn, .primary-link { display:inline-flex; align-items:center; gap:6px; margin-top:9px; border:1px solid #93c5fd; border-radius:9px; padding:7px 11px; background:#eff6ff; color:#1d4ed8; font:inherit; font-size:12px; font-weight:800; text-decoration:none; cursor:pointer; }
    .daily-table { width:100%; border-collapse:collapse; font-size:12px; }
    .daily-table th, .daily-table td { padding:7px 8px; border-bottom:1px solid var(--line); text-align:right; }
    .daily-table th:first-child, .daily-table td:first-child { text-align:left; }
    .credential-grid { display:grid; grid-template-columns:1fr auto; gap:8px; align-items:center; padding:10px 12px; background:#f8fafc; border-radius:10px; margin-top:8px; }
    .credential-grid code { overflow:hidden; text-overflow:ellipsis; }
    .safe-tip { color:#92400e; background:#fffbeb; border:1px solid #fde68a; padding:10px 12px; border-radius:10px; font-size:12px; line-height:1.6; }
    @media (max-width:760px) { .verification-toolbar { align-items:flex-start; flex-direction:column; } .toolbar-actions { justify-content:flex-start; } .evidence-meta { grid-template-columns:1fr; } }
    """
    source = source.replace("</style>", extra_css + "\n  </style>", 1)

    audit_payload = {
        row["project_key"]: {
            "project_name": row["project_name"], "data_status": row["data_status"],
            "status_detail": row["status_detail"], "snapshot_at": row["snapshot_at"],
            "database_name": row.get("database_name", ""), "system_url": row.get("system_url", ""),
            "we_analysis_url": row.get("we_analysis_url", WE_ANALYSIS_DEFAULT),
            "daily": row.get("_daily", []), "metrics": row.get("_metric_evidence", {}),
        }
        for row in rows
    }
    auth_payload = {
        row["project_key"]: {
            "project_name": row["project_name"], "system_url": row.get("system_url", ""),
            "username": row.get("_auth", {}).get("username", ""),
            "password": row.get("_auth", {}).get("password", ""),
        }
        for row in rows
    }
    drawer = """<div class="drawer-backdrop" id="drawerBackdrop"></div>
    <aside class="evidence-drawer" id="evidenceDrawer" aria-hidden="true">
      <div class="drawer-inner">
        <div class="drawer-head"><div><p id="drawerKicker">数据证据</p><h3 id="drawerTitle">指标核验</h3></div><button class="drawer-close" id="drawerClose" aria-label="关闭">×</button></div>
        <div id="drawerContent"></div>
      </div>
    </aside>"""
    source = source.replace("</main>", "</main>\n  " + drawer, 1)
    payload_json = json.dumps(audit_payload, ensure_ascii=False).replace("</", "<\\/")
    auth_json = json.dumps(auth_payload, ensure_ascii=False).replace("</", "<\\/")
    script = f"""<script>
    const AUDIT_DATA = {payload_json};
    const LOCAL_AUTH = {auth_json};
    const drawer = document.getElementById('evidenceDrawer');
    const backdrop = document.getElementById('drawerBackdrop');
    const title = document.getElementById('drawerTitle');
    const kicker = document.getElementById('drawerKicker');
    const content = document.getElementById('drawerContent');
    const esc = value => String(value ?? '').replace(/[&<>\"']/g, c => ({{'&':'&amp;','<':'&lt;','>':'&gt;','\"':'&quot;',"'":'&#39;'}}[c]));
    function openDrawer() {{ drawer.classList.add('open'); backdrop.classList.add('open'); drawer.setAttribute('aria-hidden','false'); }}
    function closeDrawer() {{ drawer.classList.remove('open'); backdrop.classList.remove('open'); drawer.setAttribute('aria-hidden','true'); }}
    function copyText(value, button) {{
      const done = () => {{ const old=button.textContent; button.textContent='已复制'; setTimeout(()=>button.textContent=old,1200); }};
      if (navigator.clipboard && window.isSecureContext) navigator.clipboard.writeText(value).then(done);
      else {{ const area=document.createElement('textarea'); area.value=value; document.body.appendChild(area); area.select(); document.execCommand('copy'); area.remove(); done(); }}
    }}
    function dailyTable(rows) {{
      if (!rows || !rows.length) return '<p>当前证据没有逐日明细；快照项目需要恢复实时连接后生成。</p>';
      return '<div style="max-height:330px;overflow:auto"><table class="daily-table"><thead><tr><th>日期</th><th>DAU</th><th>有效订单</th><th>有效流水</th></tr></thead><tbody>' + rows.slice().reverse().map(r => `<tr><td>${{esc(r.date)}}</td><td>${{Number(r.dau).toLocaleString()}}</td><td>${{Number(r.valid_orders).toLocaleString()}}</td><td>${{Number(r.valid_amount).toLocaleString(undefined,{{minimumFractionDigits:2,maximumFractionDigits:2}})}}</td></tr>`).join('') + '</tbody></table></div>';
    }}
    function showMetric(projectKey, metricKey) {{
      const project=AUDIT_DATA[projectKey], metric=project?.metrics?.[metricKey];
      if (!project || !metric) return;
      kicker.textContent=project.project_name;
      title.textContent=metric.label;
      content.innerHTML=`<div class="evidence-block evidence-meta"><div><span>数据状态</span><strong>${{esc(project.data_status)}}｜${{esc(project.status_detail)}}</strong></div><div><span>刷新时间</span><strong>${{esc(project.snapshot_at)}}</strong></div><div><span>统计范围</span><strong>${{esc(metric.period)}}</strong></div><div><span>来源</span><strong>${{esc(metric.source)}}</strong></div></div><div class="evidence-block"><h4>计算口径</h4><p>${{esc(metric.formula)}}</p></div><div class="evidence-block"><h4>复核 SQL</h4><pre class="evidence-sql">${{esc(metric.sql)}}</pre><button class="copy-btn" id="copySql">复制 SQL</button></div><div class="evidence-block"><h4>30 日逐日聚合证据</h4>${{dailyTable(project.daily)}}</div>`;
      document.getElementById('copySql').onclick=e=>copyText(metric.sql,e.currentTarget);
      openDrawer();
    }}
    function showLogin(projectKey) {{
      const auth=LOCAL_AUTH[projectKey]; if (!auth) return;
      kicker.textContent='本机私有登录信息'; title.textContent=auth.project_name;
      const configured=auth.username && auth.password;
      content.innerHTML=`<div class="evidence-block"><div class="safe-tip">账号密码来自 Git 已忽略的本机配置，不会写入提交；自动填充只写入登录框，不自动提交。</div></div><div class="evidence-block"><h4>系统地址</h4>${{auth.system_url ? `<a class="primary-link" href="${{esc(auth.system_url)}}" target="_blank" rel="noopener noreferrer">打开项目管理端 ↗</a>` : '<p>尚未配置系统地址。</p>'}}</div><div class="evidence-block"><h4>登录凭据</h4>${{configured ? `<div class="credential-grid"><code>${{esc(auth.username)}}</code><button class="copy-btn" data-copy="username">复制账号</button></div><div class="credential-grid"><code>••••••••••••</code><button class="copy-btn" data-copy="password">复制密码</button></div>` : '<p>尚未在 config/local/all-project-dashboard-auth.csv 配置账号密码。</p>'}}</div>`;
      content.querySelectorAll('[data-copy]').forEach(btn=>btn.onclick=e=>copyText(auth[e.currentTarget.dataset.copy],e.currentTarget));
      openDrawer();
    }}
    document.addEventListener('click', event => {{
      const metric=event.target.closest('[data-metric]'); if (metric) showMetric(metric.dataset.project,metric.dataset.metric);
      const login=event.target.closest('[data-login]'); if (login) showLogin(login.dataset.login);
    }});
    document.getElementById('drawerClose').onclick=closeDrawer; backdrop.onclick=closeDrawer;
    document.addEventListener('keydown', e=>{{if(e.key==='Escape') closeDrawer();}});
    </script>"""
    source = source.replace("</body>", script + "\n</body>", 1)
    output.write_text(source, encoding="utf-8")


def write_csv(rows: list[dict[str, Any]], path: Path) -> None:
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=OUTPUT_FIELDS, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)


def write_metric_evidence(rows: list[dict[str, Any]], path: Path) -> None:
    fields = ["project_key", "project_name", "data_status", "metric", "value", "label", "period", "formula", "source", "sql"]
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        for row in rows:
            for metric, evidence in row.get("_metric_evidence", {}).items():
                writer.writerow({
                    "project_key": row["project_key"], "project_name": row["project_name"],
                    "data_status": row["data_status"], "metric": metric, "value": row.get(metric, ""),
                    **evidence,
                })


def write_autofill_extension(rows: list[dict[str, Any]], output_dir: Path) -> Path:
    extension = output_dir / "local-autofill-extension"
    extension.mkdir(parents=True, exist_ok=True)
    extension.chmod(0o700)
    credentials: dict[str, dict[str, str]] = {}
    matches: list[str] = []
    for row in rows:
        auth = row.get("_auth", {})
        url = str(row.get("system_url") or "")
        if not url or not auth.get("username") or not auth.get("password"):
            continue
        parsed = urlparse(url)
        if parsed.scheme not in ("http", "https") or not parsed.netloc:
            continue
        origin = f"{parsed.scheme}://{parsed.netloc}"
        credentials[origin] = {"username": auth["username"], "password": auth["password"]}
        matches.append(origin + "/*")
    manifest: dict[str, Any] = {
        "manifest_version": 3, "name": "智慧食堂本机登录填充", "version": "1.0.0",
        "description": "仅在本机项目登录页填写本地凭据，不自动提交。",
    }
    if matches:
        manifest["host_permissions"] = sorted(set(matches))
        manifest["content_scripts"] = [{
            "matches": sorted(set(matches)), "js": ["content.js"], "run_at": "document_idle",
        }]
    content_js = """(() => {
  const credentials = __CREDENTIALS__;
  const nativeSetter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, 'value').set;
  function setValue(input, value) {
    if (!input || !value) return;
    nativeSetter.call(input, value);
    input.dispatchEvent(new Event('input', {bubbles: true}));
    input.dispatchEvent(new Event('change', {bubbles: true}));
  }
  function fill() {
    const auth = credentials[location.origin];
    if (!auth) return;
    const loginForm = document.querySelector('.login-form');
    const onLogin = location.hash.includes('/login') || Boolean(loginForm);
    if (!onLogin) return;
    const username = document.querySelector('input[placeholder="请输入用户名"], input[autocomplete="username"], input[name="username"], input[name="account"]');
    const password = document.querySelector('input[placeholder="请输入密码"], input[type="password"], input[autocomplete="current-password"]');
    setValue(username, auth.username);
    setValue(password, auth.password);
  }
  fill();
  const observer = new MutationObserver(fill);
  observer.observe(document.documentElement, {childList: true, subtree: true});
  setTimeout(() => observer.disconnect(), 15000);
})();
""".replace("__CREDENTIALS__", json.dumps(credentials, ensure_ascii=False))
    manifest_path = extension / "manifest.json"
    content_path = extension / "content.js"
    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    content_path.write_text(content_js, encoding="utf-8")
    manifest_path.chmod(0o600)
    content_path.chmod(0o600)
    return extension


def open_secure_chrome(output_dir: Path, extension: Path) -> None:
    chrome = Path("/Applications/Google Chrome.app")
    if not chrome.exists():
        raise RuntimeError("google_chrome_missing")
    profile = output_dir / "secure-chrome-profile"
    profile.mkdir(parents=True, exist_ok=True)
    command = [
        "open", "-na", "Google Chrome", "--args",
        f"--user-data-dir={profile}", f"--disable-extensions-except={extension}",
        f"--load-extension={extension}", (output_dir / "index.html").as_uri(),
    ]
    subprocess.run(command, check=False)


def write_query_plan(rows: list[dict[str, Any]], as_of: dt.date, path: Path) -> None:
    counts = {status: sum(1 for row in rows if row["data_status"] == status) for status in ("live", "snapshot", "unavailable")}
    text = f"""# 全项目 DAU 查询记录

- 统计日：{as_of.isoformat()}
- 项目总数：{len(rows)}
- 实时只读成功：{counts['live']}
- 历史快照补位：{counts['snapshot']}
- 当前不可达：{counts['unavailable']}
- 数据表：`ydy_staff`、`ydy_meal_order`
- 只读门禁：`SET SESSION TRANSACTION READ ONLY`
- 输出不包含账号、密码、姓名、手机号或订单明细。
- 看板中的每个数值可点击查看精确口径、SQL、查询时间和30日聚合明细。
- 登录账号密码仅来自 `config/local/all-project-dashboard-auth.csv`，并只写入 Git 已忽略的运行时页面与本机自动填充扩展。

## DAU 口径

有效订单按项目实际字段组合：`pay_status=20`、`is_delete=0`、`order_status=30`（字段存在时），并限制 `meal_date <= {as_of.isoformat()}`。DAU 按 `meal_date + user_id` 去重。

## 状态边界

不可达项目仍出现在 17 项清单中；不可达不代表无用户、无订单或无流水。历史快照只用于避免把已知的零值项目误报为不可达，页面会明确标记为“快照”。
"""
    path.write_text(text, encoding="utf-8")


def main() -> int:
    args = parse_args()
    try:
        as_of = dt.date.fromisoformat(args.as_of)
    except ValueError:
        print("invalid --as-of, expected YYYY-MM-DD", file=sys.stderr)
        return 2
    args.output_dir.mkdir(parents=True, exist_ok=True)
    snapshot = load_snapshot()
    registry = read_registry()
    local_auth = read_local_auth(args.auth_file)
    rows: list[dict[str, Any]] = []
    try:
        binary = mysql_binary()
    except RuntimeError as exc:
        binary = ""
        global_error = error_label(str(exc))
    else:
        global_error = ""

    for project in registry:
        access = project_access(project, local_auth, args.store_repo)
        config, source = load_config(args.store_repo, project["config_dir"])
        if project["project_key"] == "zhct_rdfz" and not config:
            config, fallback_source = load_config(args.store_repo, "zhct_guoxin")
            if config:
                config["database"] = "zhct_rdfz"
                source = f"same_rds_config_fallback:{fallback_source}"
        if project["project_key"] == "zhct_jsr" and config:
            config = jsr_readonly_config(config["database"])
            source = "jsr_readonly_env"
        try:
            if global_error:
                raise RuntimeError(global_error)
            if not config:
                raise RuntimeError(source)
            row = query_project(binary, project, config, source, as_of, args.timeout)
        except (RuntimeError, subprocess.TimeoutExpired) as exc:
            cached = snapshot.get(project["project_key"])
            if cached:
                row = unavailable_row(project, "同源历史快照，非本次实时查询", as_of)
                row.update(cached)
                row["data_status"] = "snapshot"
                row["status_detail"] = "同源历史快照，非本次实时查询"
                row["_daily"] = []
                row["_metric_evidence"] = snapshot_evidence(row)
            else:
                row = unavailable_row(project, error_label(str(exc)), as_of)
                row["_daily"] = []
                row["_metric_evidence"] = {}
        apply_access(row, access)
        rows.append(row)

    write_csv(rows, args.output_dir / "all-project-dau.csv")
    render_html(rows, as_of, args.output_dir / "index.html")
    write_metric_evidence(rows, args.output_dir / "metric-evidence.csv")
    write_query_plan(rows, as_of, args.output_dir / "query-plan.md")
    extension = write_autofill_extension(rows, args.output_dir)
    print(f"projects={len(rows)}")
    print(f"live={sum(1 for row in rows if row['data_status']=='live')}")
    print(f"snapshot={sum(1 for row in rows if row['data_status']=='snapshot')}")
    print(f"unavailable={sum(1 for row in rows if row['data_status']=='unavailable')}")
    print(f"html={args.output_dir / 'index.html'}")
    print(f"csv={args.output_dir / 'all-project-dau.csv'}")
    print(f"evidence={args.output_dir / 'metric-evidence.csv'}")
    print(f"auth_file={args.auth_file}")
    print(f"autofill_configured={sum(1 for row in rows if row['auth_configured']=='yes')}")
    if args.open_secure:
        open_secure_chrome(args.output_dir, extension)
    return 0


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