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")

# ---- 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("规则：未退出社区，且邮箱来自 users.email 或手机号注册后在 early_community 留存邮箱。")
    st.metric("参与成员数", f"{len(members_df):,}")
    if members_df.empty:
        st.info("当前没有符合条件的共创社区成员。")
        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("共创社区成员", width="large")
    def show_community_members_dialog():
        _apply_dialog_extra_wide_style()
        _render_community_members_content()
else:
    def show_community_members_dialog():
        st.subheader("共创社区成员")
        _render_community_members_content()

st.title("Lyrria Growth Dashboard")

# ---- Sidebar ----
st.sidebar.header("Filters")

segments = read_df("SELECT DISTINCT user_segment AS seg FROM users WHERE user_segment IS NOT NULL ORDER BY seg")["seg"].dropna().tolist()
orgs = read_df("SELECT DISTINCT org_id_snapshot AS org FROM user_daily_activity WHERE org_id_snapshot IS NOT NULL ORDER BY org")["org"].tolist()

segment_default = [s for s in segments if s != "internal"] if segments else []
segment_sel = st.sidebar.multiselect("users.user_segment", options=segments, default=segment_default)
org_sel = st.sidebar.multiselect("org_id_snapshot (school)", options=orgs, default=[])

exclude_test = st.sidebar.checkbox("Exclude users.is_test_user = 1", 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(
    "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("查看共创社区成员", 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(月内).
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("DAU trend")
trend_start = target_day.replace(day=1)
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="Date", labelAngle=0)),
        y=alt.Y("dau:Q", axis=alt.Axis(title="DAU")),
        tooltip=["day", "dau"],
    )
)
st.altair_chart(chart, use_container_width=True)

# ---- Trend (CURR, by day from curr_daily) ----
st.subheader("CURR trend")
curr_start = target_day.replace(day=1)
curr_df = read_df(
    """
    SELECT activity_date, prev_day_active_count, retained_count, curr_rate_pct, total_active_today
    FROM curr_daily
    WHERE activity_date >= %s AND activity_date <= %s
    ORDER BY activity_date
    """,
    (curr_start.isoformat(), target_day.isoformat()),
)
if not curr_df.empty:
    curr_df["activity_date"] = pd.to_datetime(curr_df["activity_date"]).dt.date

# 补齐日期轴到 target_day：无数据的日期显示 0
curr_days = pd.DataFrame(
    {"activity_date": pd.date_range(start=curr_start, end=target_day, freq="D").date}
)
curr_plot = curr_days.merge(curr_df, on="activity_date", how="left")
for col in ["prev_day_active_count", "retained_count", "curr_rate_pct", "total_active_today"]:
    if col in curr_plot.columns:
        curr_plot[col] = curr_plot[col].fillna(0)
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="Date", labelAngle=0)),
        y=alt.Y("curr_rate_pct:Q", axis=alt.Axis(title="CURR (%)", format=".2f")),
        tooltip=[
            "day",
            alt.Tooltip("curr_rate_pct:Q", title="CURR (%)", format=".2f"),
            alt.Tooltip("prev_day_active_count:Q", title="Prev-day active"),
            alt.Tooltip("retained_count:Q", title="Retained today"),
            alt.Tooltip("total_active_today:Q", title="DAU today"),
        ],
    )
)
st.altair_chart(curr_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(f"Segments on {d0.isoformat()}")

active_buckets = [
    ("New", len(new_users)),
    ("Current", len(current_users)),
    ("Reactivated", len(reactivated_users)),
]
active_total = len(active_today)
inactive_total = max(0, len(cohort_universe) - len(active_today))
inactive_buckets = [
    ("At-risk (7d)", len(at_risk_wau)),
    ("At-risk (30d)", len(at_risk_mau)),
    ("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(f"{pct:.1%} of {denom:,}. {hint}")

# 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("### Active today composition")
    with t2:
        st.metric("DAU", len(active_today))

    st.caption("How today's DAU is composed.")
    for label, val in active_buckets:
        segment_card(label, val, active_total, "Share of DAU.")

with right:
    t1, t2 = st.columns([2, 2], vertical_alignment="center")
    with t1:
        st.markdown("### Not active today")
    with t2:
        st.metric("Inactive / Total", f"{inactive_total:,} / {len(cohort_universe):,}")

    st.caption("Breakdown of users who are not active today (within the current filtered cohort).")
    for label, val in inactive_buckets:
        segment_card(label, val, inactive_total, "Share of users not active today.")


st.markdown("---")
# ---- Drill down ----
st.subheader("Drill-down (first 50 user_ids)")
tab1, tab2, tab3 = st.tabs(["Active today", "At-risk", "Dormant"])
with tab1:
    st.write(sorted(list(active_today))[:50])
with tab2:
    st.write({
        "at_risk_wau": sorted(list(at_risk_wau))[:50],
        "at_risk_mau": sorted(list(at_risk_mau))[:50],
    })
with tab3:
    st.write(sorted(list(dormant_users))[:50])

#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."
#)