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( """ """, 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( ( "
" f"{label}" f"{value:,}" "
" ), 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." #)