zrd-nb666/campus_health_app/analytics/analyze.py

"""
体重数据分析 CLI
=================

两个子命令：
  python analyze.py generate   生成 180 天合成数据（synthetic-user-001）写入 MySQL
  python analyze.py report     从库里读数据，输出终端报告 + 一张趋势 PNG

设计原则：
- 合成数据有真实的减重轨迹 + 测量噪声 + 标签事件 + 若干异常点，
  方便直接演示"数据洞察"，不需要真实用户数据。
- 所有 SQL 都写出来，方便面试官 review。
"""
from __future__ import annotations

import sys
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from uuid import uuid4

import matplotlib

matplotlib.use("Agg")  # 无 GUI 环境也能出图
import matplotlib.pyplot as plt

# Windows 中文字体，避免图里中文变方框
plt.rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
import numpy as np
import pandas as pd
import pymysql
import typer
from rich.console import Console
from rich.table import Table

app = typer.Typer(help="体重数据合成与分析 CLI", no_args_is_help=True)
console = Console()

# ---------------------------------------------------------------------------
# 数据库连接配置（与 backend/.env 保持一致）
# ---------------------------------------------------------------------------
import os

DB_CONFIG = dict(
    host=os.getenv("DB_HOST", "localhost"),
    port=int(os.getenv("DB_PORT", "3306")),
    user=os.getenv("DB_USERNAME", "analytics"),
    password=os.getenv("DB_PASSWORD", "123456"),
    database=os.getenv("DB_DATABASE", "campus_health"),
    charset="utf8mb4",
    autocommit=False,
)

SYNTH_USER_ID = "synthetic-user-001"
SYNTH_USER_PHONE = "13900000001"
SYNTH_USER_NICKNAME = "合成用户·小A"

OUTPUT_DIR = Path(__file__).parent / "output"


def get_conn(password: str | None = None) -> pymysql.Connection:
    cfg = dict(DB_CONFIG)
    if password:
        cfg["password"] = password
    try:
        return pymysql.connect(**cfg)
    except pymysql.err.OperationalError as e:
        if e.args[0] == 1045 and not password:
            # 密码错误或未提供，交互式问一次
            pwd = typer.prompt("MySQL 密码", hide_input=True)
            cfg["password"] = pwd
            return pymysql.connect(**cfg)
        console.print(f"[red]数据库连接失败：{e}[/red]")
        console.print(
            "请确认：1) MySQL 服务已启动  2) 已执行 backend/database/init.sql 和 schema.sql"
        )
        sys.exit(1)


# ---------------------------------------------------------------------------
# 数据生成器
# ---------------------------------------------------------------------------
@app.command()
def generate(
    days: int = typer.Option(180, help="生成多少天的数据"),
    start_weight: float = typer.Option(75.0, help="初始体重 kg"),
    target_weight: float = typer.Option(66.0, help="目标体重 kg"),
    seed: int = typer.Option(42, help="随机种子，保证可复现"),
):
    """生成 180 天合成体重数据并写入 campus_health.weight_records"""
    rng = np.random.default_rng(seed)
    end_date = date.today()
    start_date = end_date - timedelta(days=days - 1)

    conn = get_conn()
    with conn.cursor() as cur:
        # 幂等：先清掉这个用户的旧数据
        cur.execute("DELETE FROM weight_record_tags WHERE weight_record_id IN (SELECT id FROM weight_records WHERE user_id=%s)", (SYNTH_USER_ID,))
        cur.execute("DELETE FROM weight_records WHERE user_id=%s", (SYNTH_USER_ID,))
        cur.execute("DELETE FROM weight_goals WHERE user_id=%s", (SYNTH_USER_ID,))
        cur.execute("DELETE FROM anomaly_logs WHERE user_id=%s", (SYNTH_USER_ID,))
        cur.execute(
            "INSERT INTO users (id, phone, nickname, grade) VALUES (%s,%s,%s,%s) "
            "ON DUPLICATE KEY UPDATE nickname=VALUES(nickname)",
            (SYNTH_USER_ID, SYNTH_USER_PHONE, SYNTH_USER_NICKNAME, "大三"),
        )

        # ---- 目标 ----
        cur.execute(
            "INSERT INTO weight_goals (id,user_id,target_weight,target_body_fat,target_date,"
            "start_weight,start_body_fat,start_date,weekly_target,status) "
            "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'active')",
            (
                str(uuid4()), SYNTH_USER_ID, target_weight, 21.0, end_date + timedelta(days=30),
                start_weight, 28.0, start_date, 0.5,
            ),
        )

        # ---- 标签事件（相对天数偏移）----
        tag_events = {
            0: "开始运动",
            30: "饮食改变",
            60: "生病",
            90: "假期",
            120: "目标调整",
            150: "压力期",
        }

        # 取系统标签 ID
        cur.execute("SELECT id, tag_name FROM tags WHERE user_id IS NULL")
        tag_id_by_name = {name: tid for tid, name in cur.fetchall()}

        # ---- 逐日生成 ----
        records = []  # (record_id, date, weight, body_fat, muscle, condition, notes)
        total_drop = start_weight - target_weight
        for i in range(days):
            d = start_date + timedelta(days=i)
            # 周末 20% 概率不称
            if d.weekday() >= 5 and rng.random() < 0.35:
                continue

            # 基线减重曲线：线性，但在标签事件点有折点
            progress = i / days
            baseline = start_weight - total_drop * progress

            # 事件扰动
            if 60 <= i < 67:           # 生病：急性脱水，体重骤降但体脂反升
                baseline -= 1.8
            elif 90 <= i < 98:         # 假期：饮食放纵，反弹
                baseline += 1.5
            elif 120 <= i < 140:       # 压力期：皮质醇高，平台
                baseline += 0.4

            # 测量噪声
            noise = rng.normal(0, 0.25)
            # 测量条件：80% 早晨空腹，20% 餐后
            if rng.random() < 0.8:
                condition = "fasting"
                time_str = "07:45:00"
                condition_offset = 0.0
            else:
                condition = "after_meal"
                time_str = "12:30:00"
                condition_offset = rng.uniform(0.8, 1.4)  # 餐后称重偏高

            weight = round(baseline + noise + condition_offset, 2)
            # 体脂：跟随体重下降，但生病周会反向
            bf_base = 28.0 - 7.0 * progress
            if 60 <= i < 67:
                bf_base += 1.5
            body_fat = round(np.clip(bf_base + rng.normal(0, 0.2), 5, 60), 2)
            # 肌肉量：缓慢上升
            muscle = round(26.0 + 1.2 * progress + rng.normal(0, 0.15), 2)

            notes = None
            if i in tag_events:
                notes = f"事件：{tag_events[i]}"

            records.append((str(uuid4()), d, time_str, weight, body_fat, muscle, condition, notes))

        cur.executemany(
            "INSERT INTO weight_records (id,user_id,record_date,measurement_time,weight,body_fat,muscle_mass,measurement_condition,notes) "
            "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)",
            [(r[0], SYNTH_USER_ID, r[1], r[2], r[3], r[4], r[5], r[6], r[7]) for r in records],
        )

        # 关联标签到事件当天的记录
        for offset, tag_name in tag_events.items():
            d = start_date + timedelta(days=offset)
            cur.execute(
                "SELECT id FROM weight_records WHERE user_id=%s AND record_date=%s LIMIT 1",
                (SYNTH_USER_ID, d),
            )
            row = cur.fetchone()
            if row and tag_name in tag_id_by_name:
                cur.execute(
                    "INSERT INTO weight_record_tags (id,weight_record_id,tag_id) VALUES (%s,%s,%s) "
                    "ON DUPLICATE KEY UPDATE id=id",
                    (str(uuid4()), row[0], tag_id_by_name[tag_name]),
                )

        # 埋一个 anomaly_log 示例
        cur.execute(
            "INSERT INTO anomaly_logs (id,user_id,anomaly_type,severity,message,detected_date) "
            "VALUES (%s,%s,'rapid_change','warning','近7天体重骤降2.1kg（脱水/生病期），请结合体脂与症状判断',%s)",
            (str(uuid4()), SYNTH_USER_ID, start_date + timedelta(days=63)),
        )

    conn.commit()
    conn.close()
    console.print(f"[green]✓ 已写入 {len(records)} 条记录，{len(tag_events)} 个标签事件[/green]")
    console.print(f"  日期范围：{start_date} → {end_date}")
    console.print(f"  体重区间：{start_weight}kg → {target_weight}kg（线性减重曲线 + 事件扰动）")


# ---------------------------------------------------------------------------
# 分析报告
# ---------------------------------------------------------------------------
def load_data() -> pd.DataFrame:
    sql = """
        SELECT wr.record_date, wr.weight, wr.body_fat, wr.muscle_mass,
               wr.measurement_condition, wr.notes,
               (SELECT GROUP_CONCAT(t.tag_name)
                  FROM weight_record_tags wrt JOIN tags t ON t.id = wrt.tag_id
                 WHERE wrt.weight_record_id = wr.id) AS tags
        FROM weight_records wr
        WHERE wr.user_id = %s AND wr.deleted_at IS NULL
        ORDER BY wr.record_date
    """
    conn = get_conn()
    df = pd.read_sql(sql, conn, params=[SYNTH_USER_ID])
    conn.close()
    df["record_date"] = pd.to_datetime(df["record_date"])
    return df


def report_overview(df: pd.DataFrame) -> None:
    table = Table(title="总体进度", show_header=True, header_style="bold cyan")
    table.add_column("指标")
    table.add_column("数值", justify="right")
    current = df.iloc[-1]
    initial = df.iloc[0]
    total_change = current["weight"] - initial["weight"]
    days = (df["record_date"].iloc[-1] - df["record_date"].iloc[0]).days
    weekly_avg = total_change / days * 7

    # 最近 4 周趋势
    recent = df[df["record_date"] >= df["record_date"].max() - pd.Timedelta(days=28)]
    recent_weekly = (recent["weight"].iloc[-1] - recent["weight"].iloc[0]) / 4

    table.add_row("记录天数", str(len(df)))
    table.add_row("起始体重", f"{initial['weight']:.2f} kg")
    table.add_row("当前体重", f"{current['weight']:.2f} kg")
    table.add_row("累计变化", f"{total_change:+.2f} kg")
    table.add_row("周均变化（全程）", f"{weekly_avg:+.3f} kg/周")
    table.add_row("周均变化（近4周）", f"{recent_weekly:+.3f} kg/周")
    table.add_row("体脂变化", f"{current['body_fat'] - initial['body_fat']:+.2f} %")
    console.print(table)


def report_condition_bias(df: pd.DataFrame) -> None:
    """空腹 vs 餐后的称重偏差——这是个真实且有价值的洞察"""
    fasting = df[df["measurement_condition"] == "fasting"]["weight"]
    after = df[df["measurement_condition"] == "after_meal"]["weight"]
    if len(fasting) < 3 or len(after) < 3:
        console.print("[yellow]样本不足，跳过测量偏差分析[/yellow]")
        return

    table = Table(title="测量条件偏差（餐后 vs 空腹）", show_header=True, header_style="bold magenta")
    table.add_column("条件")
    table.add_column("样本数", justify="right")
    table.add_column("均值 (kg)", justify="right")
    table.add_column("标准差", justify="right")
    table.add_row("空腹", str(len(fasting)), f"{fasting.mean():.2f}", f"{fasting.std():.2f}")
    table.add_row("餐后", str(len(after)), f"{after.mean():.2f}", f"{after.std():.2f}")
    table.add_row("差值", "—", f"{after.mean() - fasting.mean():+.2f}", "—")
    console.print(table)
    console.print(
        f"[dim]→ 餐后称重比空腹平均高 {after.mean()-fasting.mean():.2f} kg，"
        "跨条件比较时必须先按条件分层，否则会把'刚吃完饭'误读成'长胖了'。[/dim]"
    )


def report_tag_impact(df: pd.DataFrame) -> None:
    """每个标签事件前后 7 天的周减重速度对比"""
    events = df[df["tags"].notna() & (df["tags"] != "")]
    if events.empty:
        return

    table = Table(title="标签事件影响（事件日 ±7 天的周减重速度）", show_header=True, header_style="bold green")
    table.add_column("标签")
    table.add_column("事件日")
    table.add_column("前7天周速 (kg/周)")
    table.add_column("后7天周速 (kg/周)")
    table.add_column("变化")

    for _, row in events.iterrows():
        tags = str(row["tags"]).split(",")
        for tag in tags:
            event_date = row["record_date"]
            pre = df[(df["record_date"] < event_date) & (df["record_date"] >= event_date - pd.Timedelta(days=7))]
            post = df[(df["record_date"] > event_date) & (df["record_date"] <= event_date + pd.Timedelta(days=7))]
            if len(pre) < 2 or len(post) < 2:
                continue
            pre_speed = (pre["weight"].iloc[-1] - pre["weight"].iloc[0]) / max(len(pre) - 1, 1) * 7
            post_speed = (post["weight"].iloc[-1] - post["weight"].iloc[0]) / max(len(post) - 1, 1) * 7
            delta = post_speed - pre_speed
            table.add_row(
                tag,
                event_date.strftime("%Y-%m-%d"),
                f"{pre_speed:+.2f}",
                f"{post_speed:+.2f}",
                f"{delta:+.2f}",
            )
    console.print(table)


def report_anomaly(df: pd.DataFrame) -> None:
    """简单的异常检测：周变化 > 2kg 或 体重降但体脂升"""
    df = df.sort_values("record_date").reset_index(drop=True)
    table = Table(title="自动异常扫描", show_header=True, header_style="bold red")
    table.add_column("日期")
    table.add_column("类型")
    table.add_column("说明")

    for i in range(7, len(df)):
        week_ago = df.iloc[i - 7]
        now = df.iloc[i]
        dw = now["weight"] - week_ago["weight"]
        dbf = now["body_fat"] - week_ago["body_fat"]
        if abs(dw) > 2.0:
            table.add_row(
                now["record_date"].strftime("%Y-%m-%d"),
                "rapid_change",
                f"7天体重变化 {dw:+.2f} kg（阈值 ±2.0）",
            )
        if dw < -0.5 and dbf > 0.5:
            table.add_row(
                now["record_date"].strftime("%Y-%m-%d"),
                "body_fat_anomaly",
                f"体重 {dw:+.2f} kg 但体脂 {dbf:+.2f}%——疑似脱水而非减脂",
            )
    if table.row_count == 0:
        console.print("[green]未发现异常[/green]")
    else:
        console.print(table)


def draw_chart(df: pd.DataFrame) -> Path:
    OUTPUT_DIR.mkdir(exist_ok=True)
    fig, ax = plt.subplots(figsize=(11, 5.5))
    fasting = df[df["measurement_condition"] == "fasting"]
    after = df[df["measurement_condition"] == "after_meal"]
    ax.scatter(after["record_date"], after["weight"], s=18, alpha=0.5, color="#f59e0b", label="餐后")
    ax.plot(fasting["record_date"], fasting["weight"], color="#2563eb", linewidth=1.5, label="空腹趋势")

    # 标签事件竖线
    events = df[df["tags"].notna() & (df["tags"] != "")]
    for _, row in events.iterrows():
        ax.axvline(row["record_date"], color="#9ca3af", linestyle="--", alpha=0.6)
        ax.text(row["record_date"], ax.get_ylim()[1], row["tags"].split(",")[0], rotation=90,
                fontsize=8, va="top", ha="right", color="#374151")

    ax.set_title("Synthetic User — Weight Trajectory")
    ax.set_xlabel("Date")
    ax.set_ylabel("Weight (kg)")
    ax.legend(loc="upper right")
    ax.grid(alpha=0.3)
    fig.tight_layout()
    out = OUTPUT_DIR / "weight_trend.png"
    fig.savefig(out, dpi=130)
    plt.close(fig)
    return out


@app.command()
def report():
    """从库里读数据，输出分析报告 + 趋势图"""
    df = load_data()
    if df.empty:
        console.print("[red]没有数据，请先运行：python analyze.py generate[/red]")
        raise typer.Exit(1)

    console.print(f"\n[bold]数据窗口：[/bold]{df['record_date'].iloc[0].date()} → {df['record_date'].iloc[-1].date()}，共 {len(df)} 条\n")
    report_overview(df)
    console.print()
    report_condition_bias(df)
    console.print()
    report_tag_impact(df)
    console.print()
    report_anomaly(df)
    chart_path = draw_chart(df)
    console.print(f"\n[green]✓ 趋势图已保存：{chart_path}[/green]")


if __name__ == "__main__":
    app()