|
|
@@ -10,6 +10,7 @@ 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")
|
|
|
@@ -61,7 +62,20 @@ def _make_engine_url() -> str:
|
|
|
|
|
|
@st.cache_resource
|
|
|
def _get_engine():
|
|
|
- engine = create_engine(_make_engine_url())
|
|
|
+ # 使用连接保活与周期回收，减少 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
|
|
|
@@ -75,38 +89,137 @@ except Exception as e:
|
|
|
|
|
|
|
|
|
def read_df(sql: str, params: tuple = ()) -> pd.DataFrame:
|
|
|
- return pd.read_sql_query(sql, engine, params=params)
|
|
|
+ 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_snapshot AS seg FROM user_daily_activity ORDER BY seg")["seg"].dropna().tolist()
|
|
|
+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("user_segment_snapshot", options=segments, default=segment_default)
|
|
|
+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 (from daily table)
|
|
|
+# 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,
|
|
|
+)
|
|
|
|
|
|
-target_day = st.sidebar.date_input("Target date", value=max_day, min_value=min_day, max_value=max_day)
|
|
|
-lookback_days = st.sidebar.selectbox("Trend window (days)", [14, 30, 60, 90], index=1)
|
|
|
+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 FROM users")
|
|
|
+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 for rolling-window sets & trend chart.
|
|
|
-start_recent = target_day - timedelta(days=max(lookback_days, 35))
|
|
|
+# 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(
|
|
|
"""
|
|
|
@@ -122,35 +235,48 @@ 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 segment_sel:
|
|
|
- daily = daily[daily["user_segment_snapshot"].isin(segment_sel)]
|
|
|
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)]
|
|
|
|
|
|
-# If segment/org filters are applied, universe becomes "users within that cohort".
|
|
|
-# We derive cohort universe from snapshots (all users who ever appeared in daily within recent window).
|
|
|
-# In production, if you need cohort-universe across all time, build a cohort-user table or widen this query.
|
|
|
-if segment_sel or org_sel:
|
|
|
- cohort_universe = set(daily["user_id"].unique().tolist())
|
|
|
+# 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 = (
|
|
|
+trend_start = target_day.replace(day=1)
|
|
|
+dau_raw = (
|
|
|
daily[
|
|
|
- (daily["activity_date"] >= (target_day - timedelta(days=lookback_days)))
|
|
|
+ (daily["activity_date"] >= trend_start)
|
|
|
& (daily["activity_date"] <= target_day)
|
|
|
+ & (daily["active_flag"] == 1)
|
|
|
]
|
|
|
.groupby("activity_date")["user_id"]
|
|
|
.nunique()
|
|
|
.reset_index(name="dau")
|
|
|
)
|
|
|
-# 使用字符串日期 + Altair，控制 X 轴标签为横向
|
|
|
-trend["activity_date"] = pd.to_datetime(trend["activity_date"]).dt.date
|
|
|
+# 补齐日期轴到 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 = (
|
|
|
@@ -164,6 +290,48 @@ chart = (
|
|
|
)
|
|
|
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)
|
|
|
@@ -174,7 +342,7 @@ active_today = set(daily.loc[(daily["activity_date"] == d0) & (daily["active_fla
|
|
|
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/Resurrected require "last active before today" using FULL history ---
|
|
|
+# --- 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
|
|
|
#
|
|
|
@@ -216,12 +384,6 @@ new_users = {u for u in active_today if first_seen_map.get(u) == d0}
|
|
|
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)}
|
|
|
|
|
|
-# Resurrected: active today AND previous last active < d0-30 (strictly before last 30 days)
|
|
|
-resurrected_users = {
|
|
|
- u for u in active_today
|
|
|
- if (prev_last_active_date(u) is not None) and (prev_last_active_date(u) < d30)
|
|
|
-}
|
|
|
-
|
|
|
# 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)}
|
|
|
@@ -235,15 +397,10 @@ dormant_users = {
|
|
|
# ---- Segments visualization (Duolingo-like, no tables) ----
|
|
|
st.subheader(f"Segments on {d0.isoformat()}")
|
|
|
|
|
|
-# Active buckets (composition of DAU). We add an "Other active" bucket as a safe remainder.
|
|
|
-other_active = active_today - new_users - current_users - reactivated_users - resurrected_users
|
|
|
-
|
|
|
active_buckets = [
|
|
|
("New", len(new_users)),
|
|
|
("Current", len(current_users)),
|
|
|
("Reactivated", len(reactivated_users)),
|
|
|
- ("Resurrected", len(resurrected_users)),
|
|
|
- ("Other active", len(other_active)),
|
|
|
]
|
|
|
active_total = len(active_today)
|
|
|
inactive_total = max(0, len(cohort_universe) - len(active_today))
|
|
|
@@ -256,8 +413,12 @@ inactive_buckets = [
|
|
|
def segment_card(label: str, value: int, denom: int, hint: str):
|
|
|
pct = (value / denom) if denom > 0 else 0.0
|
|
|
st.markdown(
|
|
|
- f"**{label}**"
|
|
|
- f"<span style='font-size:30px; font-weight:800;'>{value:,}</span>",
|
|
|
+ (
|
|
|
+ "<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))
|
|
|
@@ -288,10 +449,8 @@ with left:
|
|
|
with t2:
|
|
|
st.metric("DAU", len(active_today))
|
|
|
|
|
|
- st.caption("How today's DAU is composed (buckets sum to DAU, with 'Other active' as remainder).")
|
|
|
+ st.caption("How today's DAU is composed.")
|
|
|
for label, val in active_buckets:
|
|
|
- if val == 0:
|
|
|
- continue
|
|
|
segment_card(label, val, active_total, "Share of DAU.")
|
|
|
|
|
|
with right:
|
|
|
@@ -303,8 +462,6 @@ with right:
|
|
|
|
|
|
st.caption("Breakdown of users who are not active today (within the current filtered cohort).")
|
|
|
for label, val in inactive_buckets:
|
|
|
- if val == 0:
|
|
|
- continue
|
|
|
segment_card(label, val, inactive_total, "Share of users not active today.")
|
|
|
|
|
|
|
