"""Weekly health report generator. Generates a Slack-formatted weekly health summary covering sleep, steps, recovery, body battery, and lifestyle data. """ from datetime import date, timedelta from typing import Any, Optional from health.services.manual_log_storage import ManualLogStorage from health.services.query import HealthDataQuery from health.utils.logging_config import setup_logger logger = setup_logger(__name__) class WeeklyReportGenerator: """Generates weekly health reports from stored Garmin and manual log data.""" def __init__(self) -> None: self.query = HealthDataQuery() self.manual_storage = ManualLogStorage() def generate(self, week_start: date) -> str: """Generate a Slack-formatted weekly health report. Args: week_start: The Monday that starts the report week. Returns: Formatted report string ready for Slack posting. """ week_end = week_start + timedelta(days=6) prev_week_start = week_start - timedelta(days=7) prev_week_end = week_start - timedelta(days=1) logger.info(f"Generating weekly report for {week_start} ~ {week_end}") sleep_stats = self._get_sleep_stats(week_start, week_end) steps_stats = self._get_steps_stats(week_start, week_end) activity_stats = self._get_activity_stats(week_start, week_end) intensity_stats = self._get_intensity_stats(week_start, week_end) recovery_stats = self._get_recovery_stats(week_start, week_end) prev_recovery = self._get_recovery_stats(prev_week_start, prev_week_end) battery_stats = self._get_battery_stats(week_start, week_end) lifestyle_stats = self._get_lifestyle_stats(week_start, week_end) all_stats = { "sleep": sleep_stats, "steps": steps_stats, "activity": activity_stats, "intensity": intensity_stats, "recovery": recovery_stats, "battery": battery_stats, "lifestyle": lifestyle_stats, } highlight = self._pick_highlight(all_stats) lines: list[str] = [] # 1. Title lines.append(f"📊 *健康周报 {week_start} ~ {week_end}*") lines.append("") # 2. Sleep lines.append("😴 *睡眠*") if sleep_stats.get("count", 0) > 0: lines.append(f" • 平均得分：{sleep_stats['avg_score']:.0f} 分") lines.append( f" • 平均时长：{self._format_duration(sleep_stats['avg_duration_sec'])}" ) if sleep_stats.get("best_date"): lines.append( f" • 最佳：{sleep_stats['best_date']} ({sleep_stats['best_score']:.0f} 分)" ) if sleep_stats.get("worst_date"): lines.append( f" • 最差：{sleep_stats['worst_date']} ({sleep_stats['worst_score']:.0f} 分)" ) else: lines.append(" • 本周无睡眠数据") lines.append("") # 3. Steps & Activity act_count = activity_stats.get("count", 0) lines.append(f"🏃 *运动*{'（共 ' + str(act_count) + ' 次）' if act_count else ''}") # Steps line if steps_stats.get("count", 0) > 0: lines.append( f" • 步数达标（≥10000）：{steps_stats['goal_days']} 天" f" | 平均 {steps_stats['avg_steps']:,.0f} 步" ) else: lines.append(" • 本周无步数数据") # Intensity minutes line if intensity_stats.get("has_data"): lines.append( f" • 强度分钟：中等 {intensity_stats['moderate_min']}min" f" + 剧烈 {intensity_stats['vigorous_min']}min" f" = {intensity_stats['total_min']} min" f"（WHO 目标 150 min，完成 {intensity_stats['goal_pct']}%）" ) # HR zones zone_seconds = activity_stats.get("zone_seconds", [0, 0, 0, 0, 0]) if any(s > 0 for s in zone_seconds): zone_parts = " ".join( f"Z{i + 1} {self._format_duration(zone_seconds[i])}" for i in range(5) ) lines.append("") lines.append(" *心率区间（全周累计）*") lines.append(f" • {zone_parts}") # Activity detail list activities = activity_stats.get("activities", []) if activities: lines.append("") lines.append(" *运动明细*") for act in activities: hr_part = "" if act["avg_hr"] or act["max_hr"]: hr_part = f" | ♥ {act['avg_hr']}/{act['max_hr']} bpm" cal_part = f" | {act['calories']}kcal" if act["calories"] else "" lines.append( f" • {act['date']} {act['name']}" f" {act['duration_min']}min{cal_part}{hr_part}" ) lines.append("") # 4. Recovery lines.append("❤️ *恢复*") if recovery_stats.get("count", 0) > 0: hrv_line = f" • 平均 HRV：{recovery_stats['avg_hrv']:.0f} ms" if prev_recovery.get("count", 0) > 0: diff = recovery_stats["avg_hrv"] - prev_recovery["avg_hrv"] sign = "+" if diff >= 0 else "" hrv_line += f"（{sign}{diff:.0f} vs 上周）" lines.append(hrv_line) rhr_line = f" • 平均静息心率：{recovery_stats['avg_rhr']:.0f} bpm" if prev_recovery.get("count", 0) > 0: diff = recovery_stats["avg_rhr"] - prev_recovery["avg_rhr"] sign = "+" if diff >= 0 else "" rhr_line += f"（{sign}{diff:.0f} vs 上周）" lines.append(rhr_line) else: lines.append(" • 本周无恢复数据") lines.append("") # 5. Body Battery lines.append("⚡ *Body Battery*") if battery_stats.get("count", 0) > 0: lines.append(f" • 平均充电值：{battery_stats['avg_charged']:.0f}") lines.append(f" • 平均消耗值：{battery_stats['avg_drained']:.0f}") else: lines.append(" • 本周无 Body Battery 数据") lines.append("") # 6. Lifestyle lines.append("🥗 *生活方式*") if lifestyle_stats.get("has_data"): lines.append(f" • 饮酒天数：{lifestyle_stats['alcohol_days']} 天") if lifestyle_stats.get("fasting_modes"): modes = ", ".join( f"{m}（{c}天）" for m, c in lifestyle_stats["fasting_modes"].items() ) lines.append(f" • 禁食模式：{modes}") lines.append(f" • 补剂记录天数：{lifestyle_stats['supplement_days']} 天") else: lines.append(" • 本周无手动记录") lines.append("") # 7. Highlight lines.append(f"💡 *本周亮点：* {highlight}") return "\n".join(lines) # ------------------------------------------------------------------ # Private helpers – data fetching # ------------------------------------------------------------------ def _get_sleep_stats(self, start: date, end: date) -> dict[str, Any]: """Compute sleep statistics for the given date range.""" records = self.query.get_metric_range("sleep", start, end) if not records: return {"count": 0} scores = [r["sleep_score"] for r in records if r.get("sleep_score")] durations = [r["total_sleep_seconds"] for r in records if r.get("total_sleep_seconds")] if not scores: return {"count": 0} best = max(records, key=lambda r: r.get("sleep_score", 0)) worst = min(records, key=lambda r: r.get("sleep_score", 100)) return { "count": len(scores), "avg_score": sum(scores) / len(scores), "avg_duration_sec": int(sum(durations) / len(durations)) if durations else 0, "best_date": best.get("date"), "best_score": best.get("sleep_score", 0), "worst_date": worst.get("date"), "worst_score": worst.get("sleep_score", 0), } def _get_steps_stats(self, start: date, end: date) -> dict[str, Any]: """Compute step statistics for the given date range.""" records = self.query.get_metric_range("steps", start, end) if not records: return {"count": 0} step_values = [r["total_steps"] for r in records if r.get("total_steps") is not None] if not step_values: return {"count": 0} return { "count": len(step_values), "avg_steps": sum(step_values) / len(step_values), "goal_days": sum(1 for s in step_values if s >= 10000), } def _get_activity_stats(self, start: date, end: date) -> dict[str, Any]: """Compute activity detail statistics from get_activities_range(). Args: start: Start date (inclusive). end: End date (inclusive). Returns: Dict with count, activities list, and cumulative HR zone seconds. """ records = self.query.get_activities_range(start, end) if not records: return {"count": 0, "activities": [], "zone_seconds": [0, 0, 0, 0, 0]} activities: list[dict[str, Any]] = [] zone_seconds = [0, 0, 0, 0, 0] for r in records: raw = r.get("raw_data") or {} if isinstance(raw, str): import json try: raw = json.loads(raw) except Exception: raw = {} # Accumulate HR zone seconds (Z1~Z5) for i in range(1, 6): val = raw.get(f"hrTimeInZone_{i}") if val is not None: try: zone_seconds[i - 1] += int(val) except (TypeError, ValueError): pass duration_sec = r.get("duration_seconds") or 0 activities.append({ "date": r.get("date", "")[-5:].replace("-", "-") if r.get("date") else "", "name": r.get("activity_name") or "未知", "duration_min": round(duration_sec / 60) if duration_sec else 0, "calories": r.get("calories") or 0, "avg_hr": r.get("average_heart_rate") or 0, "max_hr": r.get("max_heart_rate") or 0, }) return { "count": len(activities), "activities": activities, "zone_seconds": zone_seconds, } def _get_intensity_stats(self, start: date, end: date) -> dict[str, Any]: """Compute intensity minutes vs WHO 150-min target. Args: start: Start date (inclusive). end: End date (inclusive). Returns: Dict with moderate_min, vigorous_min, total_min (WHO-weighted), goal_pct. """ records = self.query.get_metric_range("intensity_minutes", start, end) if not records: return {"has_data": False} moderate_min = sum(r.get("moderate_minutes") or 0 for r in records) vigorous_min = sum(r.get("vigorous_minutes") or 0 for r in records) total_min = moderate_min + vigorous_min * 2 goal_pct = round(total_min / 150 * 100) return { "has_data": True, "moderate_min": moderate_min, "vigorous_min": vigorous_min, "total_min": total_min, "goal_pct": goal_pct, } def _get_recovery_stats(self, start: date, end: date) -> dict[str, Any]: """Compute HRV and resting heart rate statistics for the given date range.""" hrv_records = self.query.get_metric_range("hrv", start, end) hr_records = self.query.get_metric_range("heart_rate", start, end) hrv_values = [ r["hrv_value"] for r in hrv_records if r.get("hrv_value") is not None ] rhr_values = [ r["resting_heart_rate"] for r in hr_records if r.get("resting_heart_rate") ] if not hrv_values and not rhr_values: return {"count": 0} return { "count": max(len(hrv_values), len(rhr_values)), "avg_hrv": sum(hrv_values) / len(hrv_values) if hrv_values else 0.0, "avg_rhr": sum(rhr_values) / len(rhr_values) if rhr_values else 0.0, } def _get_battery_stats(self, start: date, end: date) -> dict[str, Any]: """Compute Body Battery charge/drain statistics for the given date range.""" records = self.query.get_metric_range("body_battery", start, end) if not records: return {"count": 0} charged = [r["charged"] for r in records if r.get("charged") is not None] drained = [r["drained"] for r in records if r.get("drained") is not None] if not charged and not drained: return {"count": 0} return { "count": len(records), "avg_charged": sum(charged) / len(charged) if charged else 0.0, "avg_drained": sum(drained) / len(drained) if drained else 0.0, } def _get_lifestyle_stats(self, start: date, end: date) -> dict[str, Any]: """Compute lifestyle log statistics (alcohol, fasting, supplements).""" logs = self.manual_storage.get_logs_in_range(start, end) if not logs: return {"has_data": False} alcohol_days = sum(1 for log in logs if log.alcohol_entries) supplement_days = sum(1 for log in logs if log.supplement_entries) fasting_modes: dict[str, int] = {} for log in logs: if log.fasting_mode: fasting_modes[log.fasting_mode] = fasting_modes.get(log.fasting_mode, 0) + 1 return { "has_data": True, "alcohol_days": alcohol_days, "supplement_days": supplement_days, "fasting_modes": fasting_modes, } # ------------------------------------------------------------------ # Private helpers – formatting # ------------------------------------------------------------------ def _format_duration(self, seconds: int) -> str: """Format a duration in seconds to 'Xh Ym' string. Args: seconds: Total duration in seconds. Returns: Human-readable duration string, e.g. '7h 23m'. """ if seconds <= 0: return "0h 0m" hours = seconds // 3600 minutes = (seconds % 3600) // 60 return f"{hours}h {minutes}m" def _pick_highlight(self, stats: dict[str, Any]) -> str: """Select the most noteworthy insight from aggregated stats. Args: stats: Dictionary containing all section stats. Returns: A single highlight sentence in Chinese. """ highlights: list[tuple[float, str]] = [] sleep = stats.get("sleep", {}) if sleep.get("count", 0) > 0: score = sleep["avg_score"] if score >= 85: highlights.append((score, f"睡眠质量优秀，平均得分 {score:.0f} 分 🌟")) elif score < 65: highlights.append((100 - score, f"睡眠质量偏低，平均得分 {score:.0f} 分，建议关注")) steps = stats.get("steps", {}) if steps.get("count", 0) > 0: goal_days = steps["goal_days"] if goal_days >= 5: highlights.append((goal_days * 10, f"本周 {goal_days} 天达到万步目标，运动坚持度佳 💪")) elif goal_days == 0: highlights.append((60.0, "本周没有达到万步目标，需要增加日常活动量")) intensity = stats.get("intensity", {}) if intensity.get("has_data"): pct = intensity["goal_pct"] total_min = intensity["total_min"] if pct >= 100: highlights.append((90.0, f"本周强度分钟达标（{total_min} min），运动量充足 🎯")) elif pct < 50: highlights.append((70.0, f"强度分钟仅 {total_min} min，不足 WHO 目标的一半，建议增加有氧运动")) recovery = stats.get("recovery", {}) if recovery.get("count", 0) > 0 and recovery.get("avg_hrv", 0) > 0: hrv = recovery["avg_hrv"] if hrv >= 60: highlights.append((hrv, f"HRV 均值 {hrv:.0f} ms，恢复状态良好 ✅")) elif hrv < 35: highlights.append((80.0, f"HRV 偏低（{hrv:.0f} ms），注意休息与压力管理")) lifestyle = stats.get("lifestyle", {}) if lifestyle.get("has_data") and lifestyle.get("alcohol_days", 0) == 0: highlights.append((50.0, "本周零饮酒，生活方式健康 🌿")) if not highlights: return "数据积累中，下周见更多洞察。" highlights.sort(key=lambda x: x[0], reverse=True) return highlights[0][1]