Tolan/dashboard/streamlit_growth_dashboard.py

import configparser
import os
from urllib.parse import quote_plus

from datetime import date, timedelta

import altair as alt
import pandas as pd
import streamlit as st

try:
    from sqlalchemy import create_engine, text
    from sqlalchemy.exc import OperationalError
except ImportError:
    st.set_page_config(page_title="Growth Segmentation Dashboard", layout="wide")
    st.error("请安装: pip install sqlalchemy pymysql")
    st.stop()

st.set_page_config(page_title="Growth Segmentation Dashboard", layout="wide")

# ---- UI i18n ----
I18N = {
    "zh": {
        "dashboard_title": "Lyrria 增长看板",
        "sidebar_lang": "Language / 语言",
        "sidebar_filters": "筛选器",
        "sidebar_segment": "用户分层（users.user_segment）",
        "sidebar_school": "学校（org_name）",
        "sidebar_exclude_test": "排除 users.is_test_user = 1",
        "sidebar_target_date": "目标日期",
        "sidebar_view_members": "查看共创社区成员",
        "community_dialog_title": "共创社区成员",
        "community_subheader": "共创社区成员",
        "community_rule": "规则：未退出社区，且邮箱来自 users.email 或手机号注册后在 early_community 留存邮箱。",
        "community_count": "参与成员数",
        "community_empty": "当前没有符合条件的共创社区成员。",
        "dau_trend": "DAU 趋势",
        "curr_trend": "CURR 趋势",
        "axis_date": "日期",
        "axis_dau": "DAU",
        "axis_curr": "CURR（%）",
        "tooltip_curr": "CURR（%）",
        "tooltip_prev_day_active": "昨日活跃",
        "tooltip_retained_today": "今日留存",
        "tooltip_dau_today": "今日 DAU",
        "trend3_caption_1": "30 天走势：左轴=每日用户对话总时长（秒），右轴=页面浏览量（次）。",
        "trend3_caption_2": "图例说明：蓝线=对话时长（秒）；橙线=Lumi 商店 PV；绿线=Lumi 装扮 PV。",
        "line_duration": "对话时长（秒）",
        "line_shop_pv": "Lumi 商店 PV",
        "line_outfits_pv": "Lumi 装扮 PV",
        "legend_lines": "曲线",
        "axis_duration": "对话时长（秒）",
        "axis_pv_count": "页面浏览量（次）",
        "tooltip_pv_type": "PV 类型",
        "tooltip_pv": "PV",
        "segments_on": "分群（日期：{date}）",
        "segment_new": "新用户",
        "segment_current": "持续活跃",
        "segment_reactivated": "重新活跃",
        "segment_at_risk_7d": "风险流失（7 天）",
        "segment_at_risk_30d": "风险流失（30 天）",
        "segment_dormant_30d": "沉睡（>=30 天）",
        "active_today_composition": "今日活跃构成",
        "not_active_today": "今日未活跃",
        "metric_inactive_total": "未活跃 / 总用户",
        "caption_active_comp": "今日 DAU 的构成。",
        "caption_inactive_comp": "当前筛选人群中，今日未活跃用户的分布。",
        "hint_share_dau": "占 DAU 的比例。",
        "hint_share_inactive": "占今日未活跃用户的比例。",
        "caption_of": "占 {denom} 的 {pct}。{hint}",
        "drilldown_title": "明细下钻（前 50 位用户）",
        "drilldown_caption": "邮箱、手机号为 `users` 表字段；未录入则为空。",
        "tab_active_today": "今日活跃",
        "tab_at_risk": "风险流失",
        "tab_dormant": "沉睡",
    },
    "en": {
        "dashboard_title": "Lyrria Growth Dashboard",
        "sidebar_lang": "Language / 语言",
        "sidebar_filters": "Filters",
        "sidebar_segment": "users.user_segment",
        "sidebar_school": "School (org_name)",
        "sidebar_exclude_test": "Exclude users.is_test_user = 1",
        "sidebar_target_date": "Target date",
        "sidebar_view_members": "View Community Members",
        "community_dialog_title": "Community Members",
        "community_subheader": "Community Members",
        "community_rule": "Rule: not opted out, and email comes from users.email or from early_community after phone sign-up.",
        "community_count": "Members Count",
        "community_empty": "No community members match the current rule.",
        "dau_trend": "DAU trend",
        "curr_trend": "CURR trend",
        "axis_date": "Date",
        "axis_dau": "DAU",
        "axis_curr": "CURR (%)",
        "tooltip_curr": "CURR (%)",
        "tooltip_prev_day_active": "Prev-day active",
        "tooltip_retained_today": "Retained today",
        "tooltip_dau_today": "DAU today",
        "trend3_caption_1": "30-day trend: left axis = total daily conversation duration (s), right axis = page views.",
        "trend3_caption_2": "Legend: blue = Conversation duration (s); orange = Lumi shop PV; green = Lumi outfits PV.",
        "line_duration": "Conversation duration (s)",
        "line_shop_pv": "Lumi shop PV",
        "line_outfits_pv": "Lumi outfits PV",
        "legend_lines": "Lines",
        "axis_duration": "Conversation duration (s)",
        "axis_pv_count": "Page views (count)",
        "tooltip_pv_type": "PV type",
        "tooltip_pv": "PV",
        "segments_on": "Segments on {date}",
        "segment_new": "New",
        "segment_current": "Current",
        "segment_reactivated": "Reactivated",
        "segment_at_risk_7d": "At-risk (7d)",
        "segment_at_risk_30d": "At-risk (30d)",
        "segment_dormant_30d": "Dormant (>=30d)",
        "active_today_composition": "Active today composition",
        "not_active_today": "Not active today",
        "metric_inactive_total": "Inactive / Total",
        "caption_active_comp": "How today's DAU is composed.",
        "caption_inactive_comp": "Breakdown of users who are not active today (within the current filtered cohort).",
        "hint_share_dau": "Share of DAU.",
        "hint_share_inactive": "Share of users not active today.",
        "caption_of": "{pct} of {denom}. {hint}",
        "drilldown_title": "Drill-down (first 50 users)",
        "drilldown_caption": "Email and phone come from the `users` table; blank means missing.",
        "tab_active_today": "Active today",
        "tab_at_risk": "At-risk",
        "tab_dormant": "Dormant",
    },
}


def get_lang() -> str:
    selected = st.sidebar.selectbox(I18N["en"]["sidebar_lang"], options=["中文", "English"], index=1)
    return "zh" if selected == "中文" else "en"


lang = get_lang()


def t(key: str, **kwargs) -> str:
    text = I18N.get(lang, I18N["en"]).get(key, key)
    return text.format(**kwargs) if kwargs else text

# ---- MySQL config from setting.conf ----
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DEFAULT_CONF = os.path.join(SCRIPT_DIR, "setting.conf")


def load_config(conf_path: str) -> configparser.ConfigParser | None:
    if not os.path.isfile(conf_path):
        return None
    cfg = configparser.ConfigParser(interpolation=None)
    cfg.read(conf_path, encoding="utf-8")
    return cfg


def get_mysql_config(cfg: configparser.ConfigParser) -> dict:
    return {
        "host": cfg.get("mysql", "host", fallback="127.0.0.1"),
        "port": cfg.getint("mysql", "port", fallback=3306),
        "user": cfg.get("mysql", "user", fallback="root"),
        "password": cfg.get("mysql", "password", fallback=""),
        "database": cfg.get("mysql", "database", fallback="Lyrria"),
        "charset": cfg.get("mysql", "charset", fallback="utf8mb4"),
    }


cfg = load_config(os.getenv("DASHBOARD_CONF", DEFAULT_CONF))
if not cfg or not cfg.has_section("mysql"):
    st.error(f"配置文件不存在或缺少 [mysql] 节: {DEFAULT_CONF}")
    st.stop()

mysql_kw = get_mysql_config(cfg)


def _make_engine_url() -> str:
    user = quote_plus(mysql_kw["user"])
    password = quote_plus(mysql_kw["password"])
    host = mysql_kw["host"]
    port = mysql_kw["port"]
    database = mysql_kw["database"]
    charset = mysql_kw["charset"]
    return f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset={quote_plus(charset)}"


@st.cache_resource
def _get_engine():
    # 使用连接保活与周期回收，减少 MySQL 空闲断连(2013/2006)问题
    engine = create_engine(
        _make_engine_url(),
        pool_pre_ping=True,
        pool_recycle=1800,
        pool_size=5,
        max_overflow=10,
        pool_timeout=30,
        connect_args={
            "connect_timeout": 10,
            "read_timeout": 120,
            "write_timeout": 120,
        },
    )
    with engine.connect() as c:
        c.execute(text("SELECT 1"))
    return engine


try:
    engine = _get_engine()
except Exception as e:
    st.error(f"MySQL 连接失败: {e}")
    st.stop()


def read_df(sql: str, params: tuple = ()) -> pd.DataFrame:
    try:
        return pd.read_sql_query(sql, engine, params=params)
    except OperationalError:
        # 连接中断时主动释放连接池并重试一次
        try:
            engine.dispose()
        except Exception:
            pass
        return pd.read_sql_query(sql, engine, params=params)


@st.cache_data(ttl=300)
def read_community_members_df() -> pd.DataFrame:
    """读取参与共创社区的成员列表（包含邮箱来源）"""
    return read_df(
        """
        SELECT
            u.user_id,
            COALESCE(NULLIF(TRIM(u.email), ''), NULLIF(TRIM(uec.email), '')) AS community_email,
            CASE
                WHEN u.email IS NOT NULL AND TRIM(u.email) <> '' THEN 'users.email'
                ELSE 'user_early_community.email'
            END AS email_source,
            u.phone,
            COALESCE(ucc.opted_out, 0) AS opted_out,
            uec.joined_at
        FROM users u
        LEFT JOIN user_community_choice ucc
            ON ucc.user_id = u.user_id
        LEFT JOIN user_early_community uec
            ON uec.user_id = u.user_id
        WHERE
            COALESCE(ucc.opted_out, 0) = 0
            AND (
                (u.email IS NOT NULL AND TRIM(u.email) <> '')
                OR (
                    (u.email IS NULL OR TRIM(u.email) = '')
                    AND u.phone IS NOT NULL AND TRIM(u.phone) <> ''
                    AND uec.email IS NOT NULL AND TRIM(uec.email) <> ''
                )
            )
        ORDER BY community_email, u.user_id
        """
    )


def _render_community_members_content():
    members_df = read_community_members_df()
    st.caption(t("community_rule"))
    st.metric(t("community_count"), f"{len(members_df):,}")
    if members_df.empty:
        st.info(t("community_empty"))
        return
    st.dataframe(
        members_df,
        use_container_width=True,
        hide_index=True,
    )


def _apply_dialog_extra_wide_style():
    st.markdown(
        """
        <style>
        div[data-testid="stDialog"] div[role="dialog"] {
            width: min(96vw, 1600px) !important;
            max-width: min(96vw, 1600px) !important;
        }
        </style>
        """,
        unsafe_allow_html=True,
    )


if hasattr(st, "dialog"):
    @st.dialog(t("community_dialog_title"), width="large")
    def show_community_members_dialog():
        _apply_dialog_extra_wide_style()
        _render_community_members_content()
else:
    def show_community_members_dialog():
        st.subheader(t("community_subheader"))
        _render_community_members_content()

st.title(t("dashboard_title"))

# ---- Sidebar ----
st.sidebar.header(t("sidebar_filters"))

segments = read_df("SELECT DISTINCT user_segment AS seg FROM users WHERE user_segment IS NOT NULL ORDER BY seg")["seg"].dropna().tolist()
org_df = read_df(
    """
    SELECT DISTINCT uda.org_id_snapshot AS org_id, o.org_name
    FROM user_daily_activity uda
    LEFT JOIN organizations o
        ON o.org_id COLLATE utf8mb4_unicode_ci = uda.org_id_snapshot COLLATE utf8mb4_unicode_ci
    WHERE uda.org_id_snapshot IS NOT NULL
    ORDER BY
        COALESCE(
            o.org_name COLLATE utf8mb4_unicode_ci,
            uda.org_id_snapshot COLLATE utf8mb4_unicode_ci
        ),
        uda.org_id_snapshot COLLATE utf8mb4_unicode_ci
    """
)
org_df["org_id"] = org_df["org_id"].astype(str)
org_map = {
    row["org_id"]: (str(row["org_name"]).strip() if pd.notna(row["org_name"]) and str(row["org_name"]).strip() else row["org_id"])
    for _, row in org_df.iterrows()
}
orgs = org_df["org_id"].tolist()

segment_default = [s for s in segments if s != "internal"] if segments else []
segment_sel = st.sidebar.multiselect(t("sidebar_segment"), options=segments, default=segment_default)
org_sel = st.sidebar.multiselect(
    t("sidebar_school"),
    options=orgs,
    default=[],
    format_func=lambda org_id: org_map.get(org_id, org_id),
)

exclude_test = st.sidebar.checkbox(t("sidebar_exclude_test"), value=True)

# Date bounds
max_day = pd.to_datetime(read_df("SELECT MAX(activity_date) AS d FROM user_daily_activity")["d"].iloc[0]).date()
min_day = pd.to_datetime(read_df("SELECT MIN(activity_date) AS d FROM user_daily_activity")["d"].iloc[0]).date()
today = date.today()

# UX: 默认选择今天，允许选择今天之前所有日期（若库里最小日期晚于今天，则回退为该最小日期）
target_day = st.sidebar.date_input(
    t("sidebar_target_date"),
    value=today if today >= min_day else min_day,
    min_value=min_day,
    max_value=today if today >= min_day else min_day,
)

st.sidebar.markdown("---")
if st.sidebar.button(t("sidebar_view_members"), use_container_width=True):
    show_community_members_dialog()

# ---- Load users (for universe + test filter) ----
users = read_df("SELECT user_id, is_test_user, phone, email, user_segment FROM users")
if exclude_test:
    users = users[users["is_test_user"].fillna(0).astype(int) == 0].copy()
# Base eligibility: phone/email at least one non-empty.
phone_non_empty = users["phone"].fillna("").astype(str).str.strip() != ""
email_non_empty = users["email"].fillna("").astype(str).str.strip() != ""
users = users[phone_non_empty | email_non_empty].copy()
if segment_sel:
    users = users[users["user_segment"].isin(segment_sel)].copy()
universe_users = set(users["user_id"].astype(str).tolist())

# ---- Pull daily activity for trend window + recent history (for 7/30d sets) ----
# We only need recent history for rolling-window sets (7/30d) and trend chart(最近30天).
# Keep a small buffer beyond 30 days to avoid off-by-one and sparse-data gaps.
history_days = 35
start_recent = target_day - timedelta(days=history_days)

daily = read_df(
    """
    SELECT activity_date, user_id, active_flag, session_count, message_count,
           first_seen_date, user_segment_snapshot, org_id_snapshot
    FROM user_daily_activity
    WHERE activity_date >= %s AND activity_date <= %s
    """,
    (start_recent.isoformat(), target_day.isoformat()),
)

daily["activity_date"] = pd.to_datetime(daily["activity_date"]).dt.date
daily["first_seen_date"] = pd.to_datetime(daily["first_seen_date"]).dt.date

# Apply snapshot filters
if org_sel:
    daily = daily[daily["org_id_snapshot"].isin(org_sel)]

# Apply user universe filter (test users removed)
daily = daily[daily["user_id"].isin(universe_users)]

# Cohort base comes from users table (test/contact/segment filters).
# If org filter is selected, intersect with users ever seen in that org (full history).
if org_sel:
    org_placeholders = ",".join(["%s"] * len(org_sel))
    org_df = read_df(
        f"""
        SELECT DISTINCT user_id
        FROM user_daily_activity
        WHERE org_id_snapshot IN ({org_placeholders})
        """,
        tuple(org_sel),
    )
    org_users = set(org_df["user_id"].astype(str).tolist())
    cohort_universe = universe_users & org_users
else:
    cohort_universe = universe_users

# ---- Trend (DAU, by day) ----
st.subheader(t("dau_trend"))
# 最近 30 天（含 target_day）；若早于库内最小日期则截断
trend_start = max(min_day, target_day - timedelta(days=29))
dau_raw = (
    daily[
        (daily["activity_date"] >= trend_start)
        & (daily["activity_date"] <= target_day)
        & (daily["active_flag"] == 1)
    ]
    .groupby("activity_date")["user_id"]
    .nunique()
    .reset_index(name="dau")
)
# 补齐日期轴到 target_day：无数据的日期显示 0
trend_days = pd.DataFrame(
    {"activity_date": pd.date_range(start=trend_start, end=target_day, freq="D").date}
)
trend = trend_days.merge(dau_raw, on="activity_date", how="left")
trend["dau"] = trend["dau"].fillna(0).astype(int)
trend["day"] = trend["activity_date"].astype(str)

chart = (
    alt.Chart(trend)
    .mark_line(point=True)
    .encode(
        x=alt.X("day:N", axis=alt.Axis(title=t("axis_date"), labelAngle=0)),
        y=alt.Y("dau:Q", axis=alt.Axis(title=t("axis_dau"))),
        tooltip=["day", "dau"],
    )
)
st.altair_chart(chart, use_container_width=True)

# ---- Trend (CURR, by day from curr_daily) ----
st.subheader(t("curr_trend"))
curr_start = max(min_day, target_day - timedelta(days=29))
def _compute_curr_from_daily(daily_df: pd.DataFrame, start_day: date, end_day: date) -> pd.DataFrame:
    """基于已筛选口径的 user_daily_activity 现算 CURR（避免使用全量汇总表导致筛选不联动）。"""
    df = daily_df.copy()
    if df.empty:
        out = pd.DataFrame({"activity_date": pd.date_range(start=start_day, end=end_day, freq="D").date})
        out["prev_day_active_count"] = 0
        out["retained_count"] = 0
        out["total_active_today"] = 0
        out["curr_rate_pct"] = 0.0
        return out

    df = df[df["active_flag"] == 1][["activity_date", "user_id"]].copy()
    df["user_id"] = df["user_id"].astype(str)

    # 构造 day -> active_users(set) 的映射，后续按天 O(1) 取集合
    active_by_day = df.groupby("activity_date")["user_id"].apply(lambda s: set(s.tolist())).to_dict()

    days = pd.date_range(start=start_day, end=end_day, freq="D").date
    rows: list[dict] = []
    for d in days:
        prev_d = d - timedelta(days=1)
        prev_active = active_by_day.get(prev_d, set())
        today_active = active_by_day.get(d, set())
        prev_cnt = len(prev_active)
        retained_cnt = len(prev_active & today_active) if prev_cnt else 0
        total_today = len(today_active)
        curr_rate = (retained_cnt / prev_cnt * 100.0) if prev_cnt else 0.0
        rows.append(
            {
                "activity_date": d,
                "prev_day_active_count": prev_cnt,
                "retained_count": retained_cnt,
                "curr_rate_pct": curr_rate,
                "total_active_today": total_today,
            }
        )
    return pd.DataFrame(rows)


curr_plot = _compute_curr_from_daily(daily, curr_start, target_day)
curr_plot["day"] = curr_plot["activity_date"].astype(str)

curr_chart = (
    alt.Chart(curr_plot)
    .mark_line(point=True, color="#f59e0b")
    .encode(
        x=alt.X("day:N", axis=alt.Axis(title=t("axis_date"), labelAngle=0)),
        y=alt.Y("curr_rate_pct:Q", axis=alt.Axis(title=t("axis_curr"), format=".2f")),
        tooltip=[
            "day",
            alt.Tooltip("curr_rate_pct:Q", title=t("tooltip_curr"), format=".2f"),
            alt.Tooltip("prev_day_active_count:Q", title=t("tooltip_prev_day_active")),
            alt.Tooltip("retained_count:Q", title=t("tooltip_retained_today")),
            alt.Tooltip("total_active_today:Q", title=t("tooltip_dau_today")),
        ],
    )
)
st.altair_chart(curr_chart, use_container_width=True)

# ---- Trend (Shop/Outfits PV + Conversation duration, 30d; dual-axis, 3 lines) ----
# 数据源均来自 user_events（已登录用户）：
# - page.lumi_shop__enter: 家具商店页面浏览（次）
# - page.lumi_outfits__enter: Lumi 装扮页面浏览（次）
# - conversation.chatbot__end + properties.duration_seconds: 对话时长（秒）
trend3_start = curr_start
trend3_params: list = [trend3_start.isoformat(), target_day.isoformat()]
trend3_user_filters: list[str] = []

# users.is_test_user
if exclude_test:
    trend3_user_filters.append("COALESCE(u.is_test_user, 0) = 0")

# phone/email 至少一个非空（与上方 users 口径一致）
trend3_user_filters.append("(TRIM(COALESCE(u.phone, '')) <> '' OR TRIM(COALESCE(u.email, '')) <> '')")

# users.user_segment（来自 users 表）
if segment_sel:
    seg_placeholders = ",".join(["%s"] * len(segment_sel))
    trend3_user_filters.append(f"u.user_segment IN ({seg_placeholders})")
    trend3_params += list(segment_sel)

# org filter：与 cohort_universe 的“用户曾出现在该 org”口径一致
if org_sel:
    org_placeholders = ",".join(["%s"] * len(org_sel))
    trend3_user_filters.append(
        f"""
        EXISTS (
            SELECT 1
            FROM user_daily_activity uda2
            WHERE uda2.user_id = e.user_id
              AND uda2.org_id_snapshot IN ({org_placeholders})
        )
        """
    )
    trend3_params += list(org_sel)

trend3_user_where = " AND " + " AND ".join(f"({c})" for c in trend3_user_filters) if trend3_user_filters else ""

trend3_df = read_df(
    f"""
    SELECT
        DATE(e.event_time) AS activity_date,
        SUM(CASE WHEN e.event_name = 'page.lumi_shop__enter' THEN 1 ELSE 0 END) AS lumi_shop_pv,
        SUM(CASE WHEN e.event_name = 'page.lumi_outfits__enter' THEN 1 ELSE 0 END) AS lumi_outfits_pv,
        SUM(
            CASE
                WHEN e.event_name <> 'conversation.chatbot__end' THEN 0
                WHEN JSON_EXTRACT(e.properties, '$.duration_seconds') IS NULL THEN 0
                ELSE CAST(JSON_EXTRACT(e.properties, '$.duration_seconds') AS DECIMAL(18, 3))
            END
        ) AS total_conversation_duration_seconds
    FROM user_events e
    INNER JOIN users u
        ON u.user_id = e.user_id
    WHERE e.event_time >= %s
      AND e.event_time < DATE_ADD(%s, INTERVAL 1 DAY)
      AND e.event_name IN (
        'page.lumi_shop__enter',
        'page.lumi_outfits__enter',
        'conversation.chatbot__end'
      )
      {trend3_user_where}
    GROUP BY DATE(e.event_time)
    ORDER BY activity_date
    """,
    tuple(trend3_params),
)

if not trend3_df.empty:
    trend3_df["activity_date"] = pd.to_datetime(trend3_df["activity_date"]).dt.date

# 补齐日期轴到 target_day：无数据的日期显示 0
trend3_days = pd.DataFrame(
    {"activity_date": pd.date_range(start=trend3_start, end=target_day, freq="D").date}
)
trend3_plot = trend3_days.merge(trend3_df, on="activity_date", how="left")
for col in ["lumi_shop_pv", "lumi_outfits_pv", "total_conversation_duration_seconds"]:
    if col in trend3_plot.columns:
        trend3_plot[col] = trend3_plot[col].fillna(0)
trend3_plot["day"] = trend3_plot["activity_date"].astype(str)

st.caption(t("trend3_caption_1"))
st.caption(t("trend3_caption_2"))

duration_label = t("line_duration")
shop_label = t("line_shop_pv")
outfits_label = t("line_outfits_pv")

duration_line = (
    alt.Chart(trend3_plot)
    .transform_calculate(line_type=f"'{duration_label}'")
    .mark_line(point=True, color="#3b82f6")
    .encode(
        x=alt.X("day:N", axis=alt.Axis(title=t("axis_date"), labelAngle=0)),
        y=alt.Y(
            "total_conversation_duration_seconds:Q",
            axis=alt.Axis(title=t("axis_duration")),
            scale=alt.Scale(domainMin=0),
        ),
        color=alt.Color(
            "line_type:N",
            scale=alt.Scale(
                domain=[duration_label, shop_label, outfits_label],
                range=["#3b82f6", "#f59e0b", "#10b981"],
            ),
            legend=alt.Legend(title=t("legend_lines")),
        ),
        tooltip=[
            "day",
            alt.Tooltip("total_conversation_duration_seconds:Q", title=duration_label, format=",.0f"),
            alt.Tooltip("lumi_shop_pv:Q", title=shop_label, format=",d"),
            alt.Tooltip("lumi_outfits_pv:Q", title=outfits_label, format=",d"),
        ],
    )
)

pv_lines = (
    alt.Chart(trend3_plot)
    .transform_fold(
        ["lumi_shop_pv", "lumi_outfits_pv"],
        as_=["pv_type", "pv_count"],
    )
    .transform_calculate(
        line_type=f"""
        datum.pv_type === 'lumi_shop_pv' ? '{shop_label}' : '{outfits_label}'
        """
    )
    .mark_line(point=True)
    .encode(
        x=alt.X("day:N"),
        y=alt.Y(
            "pv_count:Q",
            axis=alt.Axis(title=t("axis_pv_count"), orient="right"),
            scale=alt.Scale(domainMin=0),
        ),
        color=alt.Color(
            "line_type:N",
            scale=alt.Scale(
                domain=[duration_label, shop_label, outfits_label],
                range=["#3b82f6", "#f59e0b", "#10b981"],
            ),
            legend=alt.Legend(title=t("legend_lines")),
        ),
        tooltip=[
            "day",
            alt.Tooltip("pv_type:N", title=t("tooltip_pv_type")),
            alt.Tooltip("pv_count:Q", title=t("tooltip_pv"), format=",d"),
            alt.Tooltip("total_conversation_duration_seconds:Q", title=duration_label, format=",.0f"),
        ],
    )
)

trend3_chart = alt.layer(duration_line, pv_lines).resolve_scale(y="independent")
st.altair_chart(trend3_chart, use_container_width=True)

# ---- Segment computation on target_day ----
d0 = target_day
d7 = d0 - timedelta(days=7)
d30 = d0 - timedelta(days=30)

# Active sets (from recent daily window)
active_today = set(daily.loc[(daily["activity_date"] == d0) & (daily["active_flag"] == 1), "user_id"])
active_past_7 = set(daily.loc[(daily["activity_date"] < d0) & (daily["activity_date"] >= d7) & (daily["active_flag"] == 1), "user_id"])
active_past_30 = set(daily.loc[(daily["activity_date"] < d0) & (daily["activity_date"] >= d30) & (daily["active_flag"] == 1), "user_id"])

# --- IMPORTANT: Dormant requires "last active before today" using FULL history ---
# We compute:
# prev_last_active_date(user) = max(activity_date) where active_flag=1 and activity_date < target_day
#
# This query uses the daily table (fast) and avoids limiting to a 35-day window.
# Note: For segment/org filters, we still compute last_active across all time, but only for cohort_universe users.
params = [d0.isoformat()]
user_filter_sql = ""
if cohort_universe:
    # MySQL 单次 IN 变量过多时可改为临时表
    placeholders = ",".join(["%s"] * len(cohort_universe))
    user_filter_sql = f" AND user_id IN ({placeholders}) "
    params += list(cohort_universe)

last_active_df = read_df(
    f"""
    SELECT user_id, MAX(activity_date) AS last_active_date
    FROM user_daily_activity
    WHERE active_flag = 1
      AND activity_date < %s
      {user_filter_sql}
    GROUP BY user_id
    """,
    tuple(params),
)
if not last_active_df.empty:
    last_active_df["last_active_date"] = pd.to_datetime(last_active_df["last_active_date"]).dt.date
prev_last_active = dict(zip(last_active_df["user_id"], last_active_df["last_active_date"])) if not last_active_df.empty else {}

def prev_last_active_date(uid: str):
    return prev_last_active.get(uid, None)

# New users: use daily.first_seen_date (which you specified is users.created_at, written once and never overwritten)
# Definition: first_seen_date == target_day AND active today
# (If you want "registered today regardless of active", remove the active_today constraint.)
first_seen_map = daily.groupby("user_id")["first_seen_date"].min().to_dict()
new_users = {u for u in active_today if first_seen_map.get(u) == d0}

# Current / Reactivated
current_users = {u for u in active_today if u in active_past_7}
reactivated_users = {u for u in active_today if (u in active_past_30) and (u not in active_past_7)}

# Inactive segments
at_risk_wau = {u for u in active_past_7 if u not in active_today}
at_risk_mau = {u for u in active_past_30 if (u not in active_past_7) and (u not in active_today)}

# Dormant: users not active today AND (never active before today OR previous last active < d0-30)
dormant_users = {
    u for u in cohort_universe
    if (u not in active_today) and ((prev_last_active_date(u) is None) or (prev_last_active_date(u) < d30))
}

# ---- Segments visualization (Duolingo-like, no tables) ----
st.subheader(t("segments_on", date=d0.isoformat()))

active_buckets = [
    ("segment_new", len(new_users)),
    ("segment_current", len(current_users)),
    ("segment_reactivated", len(reactivated_users)),
]
active_total = len(active_today)
inactive_total = max(0, len(cohort_universe) - len(active_today))
inactive_buckets = [
    ("segment_at_risk_7d", len(at_risk_wau)),
    ("segment_at_risk_30d", len(at_risk_mau)),
    ("segment_dormant_30d", len(dormant_users)),
]

def segment_card(label: str, value: int, denom: int, hint: str):
    pct = (value / denom) if denom > 0 else 0.0
    st.markdown(
        (
            "<div style='display:flex; align-items:baseline; gap:8px;'>"
            f"<span style='font-weight:700;'>{label}</span>"
            f"<span style='font-size:30px; font-weight:800;'>{value:,}</span>"
            "</div>"
        ),
        unsafe_allow_html=True,
    )
    st.progress(min(max(pct, 0.0), 1.0))
    st.caption(t("caption_of", pct=f"{pct:.1%}", denom=f"{denom:,}", hint=hint))


def drilldown_table(user_ids: set, users_df: pd.DataFrame, limit: int = 50) -> pd.DataFrame:
    """Sorted user_ids (first `limit`), joined to users.email / users.phone."""
    ids = sorted(str(u) for u in user_ids)[:limit]
    if not ids:
        return pd.DataFrame(columns=["user_id", "email", "phone"])
    id_df = pd.DataFrame({"user_id": ids})
    u = users_df.copy()
    u["user_id"] = u["user_id"].astype(str)
    return id_df.merge(u[["user_id", "email", "phone"]], on="user_id", how="left")

# KPI tiles (top row)
#k1, k2, k3, k4, k5 = st.columns(5)
#k1.metric("DAU", len(active_today))
#k2.metric("New", len(new_users))
#k3.metric("Current", len(current_users))
#k4.metric("Reactivated", len(reactivated_users))
#k5.metric("Resurrected", len(resurrected_users))
#
#k6, k7, k8 = st.columns(3)
#k6.metric("At-risk WAU", len(at_risk_wau))
#k7.metric("At-risk MAU", len(at_risk_mau))
#k8.metric("Dormant", len(dormant_users))

st.markdown("")


left, right = st.columns(2, gap="large")

with left:
    t1, t2 = st.columns([3, 1], vertical_alignment="center")
    with t1:
        st.markdown(f"### {t('active_today_composition')}")
    with t2:
        st.metric("DAU", len(active_today))

    st.caption(t("caption_active_comp"))
    for label_key, val in active_buckets:
        segment_card(t(label_key), val, active_total, t("hint_share_dau"))

with right:
    t1, t2 = st.columns([2, 2], vertical_alignment="center")
    with t1:
        st.markdown(f"### {t('not_active_today')}")
    with t2:
        st.metric(t("metric_inactive_total"), f"{inactive_total:,} / {len(cohort_universe):,}")

    st.caption(t("caption_inactive_comp"))
    for label_key, val in inactive_buckets:
        segment_card(t(label_key), val, inactive_total, t("hint_share_inactive"))


st.markdown("---")
# ---- Drill down ----
st.subheader(t("drilldown_title"))
st.caption(t("drilldown_caption"))
tab1, tab2, tab3 = st.tabs([t("tab_active_today"), t("tab_at_risk"), t("tab_dormant")])
with tab1:
    st.dataframe(
        drilldown_table(active_today, users),
        use_container_width=True,
        hide_index=True,
    )
with tab2:
    c_wau, c_mau = st.columns(2)
    with c_wau:
        st.markdown("**at_risk_wau**")
        st.dataframe(
            drilldown_table(at_risk_wau, users),
            use_container_width=True,
            hide_index=True,
        )
    with c_mau:
        st.markdown("**at_risk_mau**")
        st.dataframe(
            drilldown_table(at_risk_mau, users),
            use_container_width=True,
            hide_index=True,
        )
with tab3:
    st.dataframe(
        drilldown_table(dormant_users, users),
        use_container_width=True,
        hide_index=True,
    )

#st.caption(
#    "Notes: Dormant/Resurrected use prev_last_active_date computed from full user_daily_activity history "
#    "(MAX(activity_date) where active_flag=1 and activity_date < target_day). "
#    "If your cohort is very large, replace the IN (...) filter with a temp table join."
#)