|
|
@@ -0,0 +1,344 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+"""
|
|
|
+综合按日统计流水线脚本：
|
|
|
+1) 从 user_events、messages、users 汇总到 user_daily_activity
|
|
|
+2) 基于 user_daily_activity 计算 CURR，写入 curr_daily
|
|
|
+
|
|
|
+配置：dailyCount/setting.conf（MySQL、interval_minutes、active_events）
|
|
|
+用法：
|
|
|
+ python run_daily_stats_pipeline.py # 默认守护模式，按 interval_minutes 周期执行
|
|
|
+ python run_daily_stats_pipeline.py --once # 仅执行一次（只跑当天）
|
|
|
+ python run_daily_stats_pipeline.py --once --days 7 # 补跑最近 7 天（含今天），依次跑 user_daily_activity 与 curr_daily
|
|
|
+"""
|
|
|
+
|
|
|
+import argparse
|
|
|
+import configparser
|
|
|
+import os
|
|
|
+import sys
|
|
|
+from datetime import date, datetime, timedelta, timezone
|
|
|
+
|
|
|
+try:
|
|
|
+ import pymysql
|
|
|
+except ImportError:
|
|
|
+ print("请安装 pymysql: pip install pymysql", file=sys.stderr)
|
|
|
+ sys.exit(1)
|
|
|
+
|
|
|
+
|
|
|
+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:
|
|
|
+ if not os.path.isfile(conf_path):
|
|
|
+ print(f"配置文件不存在: {conf_path}", file=sys.stderr)
|
|
|
+ sys.exit(1)
|
|
|
+ # 关闭默认的 % 插值功能，避免像 "page.%__enter" 这样的值触发插值错误
|
|
|
+ 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": "utf8mb4",
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def parse_active_events(active_events_str: str) -> tuple[str, list]:
|
|
|
+ """
|
|
|
+ 解析 active_events 配置。
|
|
|
+ 返回 (sql_condition_type, args):
|
|
|
+ - ("like", [pattern]) 表示 event_name LIKE %s
|
|
|
+ - ("in", [name1, name2, ...]) 表示 event_name IN (%s, %s, ...)
|
|
|
+ """
|
|
|
+ raw = (active_events_str or "").strip()
|
|
|
+ if not raw:
|
|
|
+ return "in", []
|
|
|
+ parts = [p.strip() for p in raw.split(",") if p.strip()]
|
|
|
+ if not parts:
|
|
|
+ return "in", []
|
|
|
+ if any("%" in p for p in parts):
|
|
|
+ return "like", [parts[0]]
|
|
|
+ return "in", parts
|
|
|
+
|
|
|
+
|
|
|
+def build_active_events_where(active_events_str: str) -> tuple[str, list]:
|
|
|
+ """返回 (WHERE 片段, 参数列表)，用于 user_events 的活跃事件过滤。"""
|
|
|
+ cond_type, args = parse_active_events(active_events_str)
|
|
|
+ if cond_type == "like":
|
|
|
+ return " AND event_name LIKE %s", args
|
|
|
+ if cond_type == "in" and args:
|
|
|
+ placeholders = ", ".join(["%s"] * len(args))
|
|
|
+ return f" AND event_name IN ({placeholders})", args
|
|
|
+ return " AND 1=0", []
|
|
|
+
|
|
|
+
|
|
|
+def run_user_daily_activity_once(
|
|
|
+ conn, activity_date: date, active_events_where: str, active_events_args: list
|
|
|
+) -> int:
|
|
|
+ """对给定 UTC 日期执行一次 user_daily_activity 聚合，返回写入/更新的行数。"""
|
|
|
+ date_str = activity_date.isoformat()
|
|
|
+ cursor = conn.cursor()
|
|
|
+
|
|
|
+ cursor.execute("SET SESSION time_zone = '+00:00'")
|
|
|
+
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ SELECT user_id FROM user_events WHERE DATE(event_time) = %s
|
|
|
+ UNION
|
|
|
+ SELECT user_id FROM messages WHERE DATE(created_at) = %s AND role = 'user'
|
|
|
+ """,
|
|
|
+ (date_str, date_str),
|
|
|
+ )
|
|
|
+ user_ids = [row[0] for row in cursor.fetchall()]
|
|
|
+ if not user_ids:
|
|
|
+ cursor.close()
|
|
|
+ return 0
|
|
|
+
|
|
|
+ active_where = " DATE(event_time) = %s " + active_events_where
|
|
|
+ active_params = [date_str] + active_events_args
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ SELECT user_id,
|
|
|
+ LEAST(1, COUNT(*)) AS active_flag,
|
|
|
+ COUNT(DISTINCT session_id) AS session_count
|
|
|
+ FROM user_events
|
|
|
+ WHERE """
|
|
|
+ + active_where
|
|
|
+ + """
|
|
|
+ GROUP BY user_id
|
|
|
+ """,
|
|
|
+ active_params,
|
|
|
+ )
|
|
|
+ active_map = {
|
|
|
+ row[0]: {"active_flag": int(row[1]), "session_count": int(row[2])}
|
|
|
+ for row in cursor.fetchall()
|
|
|
+ }
|
|
|
+
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ SELECT user_id, COUNT(*) AS message_count
|
|
|
+ FROM messages
|
|
|
+ WHERE DATE(created_at) = %s AND role = 'user'
|
|
|
+ GROUP BY user_id
|
|
|
+ """,
|
|
|
+ (date_str,),
|
|
|
+ )
|
|
|
+ message_map = {row[0]: int(row[1]) for row in cursor.fetchall()}
|
|
|
+
|
|
|
+ placeholders = ", ".join(["%s"] * len(user_ids))
|
|
|
+ cursor.execute(
|
|
|
+ f"""
|
|
|
+ SELECT user_id,
|
|
|
+ DATE(created_at) AS first_seen_date,
|
|
|
+ user_segment,
|
|
|
+ org_id
|
|
|
+ FROM users
|
|
|
+ WHERE user_id IN ({placeholders})
|
|
|
+ """,
|
|
|
+ user_ids,
|
|
|
+ )
|
|
|
+ user_snapshot = {
|
|
|
+ row[0]: {
|
|
|
+ "first_seen_date": row[1],
|
|
|
+ "user_segment": row[2],
|
|
|
+ "org_id": row[3],
|
|
|
+ }
|
|
|
+ for row in cursor.fetchall()
|
|
|
+ }
|
|
|
+
|
|
|
+ rows = []
|
|
|
+ for uid in user_ids:
|
|
|
+ active = active_map.get(uid, {"active_flag": 0, "session_count": 0})
|
|
|
+ msg_count = message_map.get(uid, 0)
|
|
|
+ snap = user_snapshot.get(uid, {})
|
|
|
+ first_seen = snap.get("first_seen_date")
|
|
|
+ first_seen_str = first_seen.isoformat() if first_seen else None
|
|
|
+ seg = snap.get("user_segment") or None
|
|
|
+ org = snap.get("org_id") or None
|
|
|
+ rows.append(
|
|
|
+ (
|
|
|
+ date_str,
|
|
|
+ uid,
|
|
|
+ active["active_flag"],
|
|
|
+ active["session_count"],
|
|
|
+ msg_count,
|
|
|
+ first_seen_str,
|
|
|
+ seg,
|
|
|
+ org,
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+ insert_sql = """
|
|
|
+ INSERT INTO user_daily_activity
|
|
|
+ (activity_date, user_id, active_flag, session_count, message_count,
|
|
|
+ first_seen_date, user_segment_snapshot, org_id_snapshot)
|
|
|
+ VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
|
+ ON DUPLICATE KEY UPDATE
|
|
|
+ active_flag = VALUES(active_flag),
|
|
|
+ session_count = VALUES(session_count),
|
|
|
+ message_count = VALUES(message_count),
|
|
|
+ user_segment_snapshot = VALUES(user_segment_snapshot),
|
|
|
+ org_id_snapshot = VALUES(org_id_snapshot),
|
|
|
+ first_seen_date = IFNULL(first_seen_date, VALUES(first_seen_date)),
|
|
|
+ updated_at = CURRENT_TIMESTAMP
|
|
|
+ """
|
|
|
+ cursor.executemany(insert_sql, rows)
|
|
|
+ conn.commit()
|
|
|
+ count = cursor.rowcount if cursor.rowcount else len(rows)
|
|
|
+ cursor.close()
|
|
|
+ return count
|
|
|
+
|
|
|
+
|
|
|
+def run_curr_daily_once(conn, activity_date: date) -> bool:
|
|
|
+ """
|
|
|
+ 对给定日期计算 CURR 并写入 curr_daily。
|
|
|
+ CURR(activity_date) = 昨日活跃且今日仍活跃人数 / 昨日活跃人数 * 100。
|
|
|
+ 返回是否写入了该日。
|
|
|
+ """
|
|
|
+ cursor = conn.cursor()
|
|
|
+ date_str = activity_date.isoformat()
|
|
|
+ prev_date = activity_date - timedelta(days=1)
|
|
|
+ prev_str = prev_date.isoformat()
|
|
|
+
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ SELECT COUNT(DISTINCT user_id) FROM user_daily_activity
|
|
|
+ WHERE activity_date = %s AND active_flag = 1
|
|
|
+ """,
|
|
|
+ (prev_str,),
|
|
|
+ )
|
|
|
+ prev_count = cursor.fetchone()[0] or 0
|
|
|
+
|
|
|
+ if prev_count == 0:
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ SELECT COUNT(DISTINCT user_id) FROM user_daily_activity
|
|
|
+ WHERE activity_date = %s AND active_flag = 1
|
|
|
+ """,
|
|
|
+ (date_str,),
|
|
|
+ )
|
|
|
+ total_today = cursor.fetchone()[0] or 0
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ INSERT INTO curr_daily
|
|
|
+ (activity_date, prev_day_active_count, retained_count, curr_rate_pct, total_active_today)
|
|
|
+ VALUES (%s, 0, 0, NULL, %s)
|
|
|
+ ON DUPLICATE KEY UPDATE
|
|
|
+ prev_day_active_count = 0,
|
|
|
+ retained_count = 0,
|
|
|
+ curr_rate_pct = NULL,
|
|
|
+ total_active_today = VALUES(total_active_today),
|
|
|
+ updated_at = CURRENT_TIMESTAMP
|
|
|
+ """,
|
|
|
+ (date_str, total_today),
|
|
|
+ )
|
|
|
+ conn.commit()
|
|
|
+ cursor.close()
|
|
|
+ return True
|
|
|
+
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ SELECT COUNT(DISTINCT a.user_id)
|
|
|
+ FROM user_daily_activity a
|
|
|
+ INNER JOIN user_daily_activity b ON a.user_id = b.user_id AND b.activity_date = %s AND b.active_flag = 1
|
|
|
+ WHERE a.activity_date = %s AND a.active_flag = 1
|
|
|
+ """,
|
|
|
+ (date_str, prev_str),
|
|
|
+ )
|
|
|
+ retained = cursor.fetchone()[0] or 0
|
|
|
+
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ SELECT COUNT(DISTINCT user_id) FROM user_daily_activity
|
|
|
+ WHERE activity_date = %s AND active_flag = 1
|
|
|
+ """,
|
|
|
+ (date_str,),
|
|
|
+ )
|
|
|
+ total_today = cursor.fetchone()[0] or 0
|
|
|
+
|
|
|
+ curr_pct = round(100.0 * retained / prev_count, 2) if prev_count else None
|
|
|
+
|
|
|
+ cursor.execute(
|
|
|
+ """
|
|
|
+ INSERT INTO curr_daily
|
|
|
+ (activity_date, prev_day_active_count, retained_count, curr_rate_pct, total_active_today)
|
|
|
+ VALUES (%s, %s, %s, %s, %s)
|
|
|
+ ON DUPLICATE KEY UPDATE
|
|
|
+ prev_day_active_count = VALUES(prev_day_active_count),
|
|
|
+ retained_count = VALUES(retained_count),
|
|
|
+ curr_rate_pct = VALUES(curr_rate_pct),
|
|
|
+ total_active_today = VALUES(total_active_today),
|
|
|
+ updated_at = CURRENT_TIMESTAMP
|
|
|
+ """,
|
|
|
+ (date_str, prev_count, retained, curr_pct, total_today),
|
|
|
+ )
|
|
|
+ conn.commit()
|
|
|
+ cursor.close()
|
|
|
+ return True
|
|
|
+
|
|
|
+
|
|
|
+def main() -> None:
|
|
|
+ parser = argparse.ArgumentParser(
|
|
|
+ description="用户日活跃 + CURR 综合按日统计流水线 -> user_daily_activity & curr_daily"
|
|
|
+ )
|
|
|
+ parser.add_argument("--config", default=DEFAULT_CONF, help="setting.conf 路径")
|
|
|
+ parser.add_argument(
|
|
|
+ "--days",
|
|
|
+ type=int,
|
|
|
+ default=0,
|
|
|
+ help="补跑最近 N 天（含今天），0 表示仅今天",
|
|
|
+ )
|
|
|
+ parser.add_argument(
|
|
|
+ "--once",
|
|
|
+ action="store_true",
|
|
|
+ help="仅执行一次（默认按 interval_minutes 守护循环执行）",
|
|
|
+ )
|
|
|
+ args = parser.parse_args()
|
|
|
+
|
|
|
+ cfg = load_config(args.config)
|
|
|
+ mysql_kw = get_mysql_config(cfg)
|
|
|
+ active_events_str = cfg.get(
|
|
|
+ "user_daily_activity", "active_events", fallback="page.%__enter"
|
|
|
+ )
|
|
|
+ interval_minutes = cfg.getint("schedule", "interval_minutes", fallback=60)
|
|
|
+
|
|
|
+ active_where, active_args = build_active_events_where(active_events_str)
|
|
|
+
|
|
|
+ def do_run() -> None:
|
|
|
+ conn = pymysql.connect(**mysql_kw)
|
|
|
+ try:
|
|
|
+ today_utc = datetime.now(timezone.utc).date()
|
|
|
+ if args.days > 0:
|
|
|
+ dates = [today_utc - timedelta(days=i) for i in range(args.days)]
|
|
|
+ else:
|
|
|
+ dates = [today_utc]
|
|
|
+ for d in dates:
|
|
|
+ n = run_user_daily_activity_once(conn, d, active_where, active_args)
|
|
|
+ print(f"[pipeline] [user_daily_activity] {d.isoformat()} 写入/更新 {n} 条")
|
|
|
+ run_curr_daily_once(conn, d)
|
|
|
+ print(f"[pipeline] [curr_daily] {d.isoformat()} 已更新")
|
|
|
+ finally:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ if args.once:
|
|
|
+ do_run()
|
|
|
+ else:
|
|
|
+ import time
|
|
|
+
|
|
|
+ print(
|
|
|
+ f"[pipeline] 守护模式，每 {interval_minutes} 分钟依次执行 user_daily_activity 与 curr_daily"
|
|
|
+ )
|
|
|
+ while True:
|
|
|
+ do_run()
|
|
|
+ time.sleep(interval_minutes * 60)
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ main()
|
|
|
+
|
