Coverage for health / analytics / weekly_report.py: 0%
219 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-02 17:44 +0800
1"""Weekly health report generator.
3Generates a Slack-formatted weekly health summary covering sleep, steps,
4recovery, body battery, and lifestyle data.
5"""
7from datetime import date, timedelta
8from typing import Any, Optional
10from health.services.manual_log_storage import ManualLogStorage
11from health.services.query import HealthDataQuery
12from health.utils.logging_config import setup_logger
14logger = setup_logger(__name__)
17class WeeklyReportGenerator:
18 """Generates weekly health reports from stored Garmin and manual log data."""
20 def __init__(self) -> None:
21 self.query = HealthDataQuery()
22 self.manual_storage = ManualLogStorage()
24 def generate(self, week_start: date) -> str:
25 """Generate a Slack-formatted weekly health report.
27 Args:
28 week_start: The Monday that starts the report week.
30 Returns:
31 Formatted report string ready for Slack posting.
32 """
33 week_end = week_start + timedelta(days=6)
34 prev_week_start = week_start - timedelta(days=7)
35 prev_week_end = week_start - timedelta(days=1)
37 logger.info(f"Generating weekly report for {week_start} ~ {week_end}")
39 sleep_stats = self._get_sleep_stats(week_start, week_end)
40 steps_stats = self._get_steps_stats(week_start, week_end)
41 activity_stats = self._get_activity_stats(week_start, week_end)
42 intensity_stats = self._get_intensity_stats(week_start, week_end)
43 recovery_stats = self._get_recovery_stats(week_start, week_end)
44 prev_recovery = self._get_recovery_stats(prev_week_start, prev_week_end)
45 battery_stats = self._get_battery_stats(week_start, week_end)
46 lifestyle_stats = self._get_lifestyle_stats(week_start, week_end)
48 all_stats = {
49 "sleep": sleep_stats,
50 "steps": steps_stats,
51 "activity": activity_stats,
52 "intensity": intensity_stats,
53 "recovery": recovery_stats,
54 "battery": battery_stats,
55 "lifestyle": lifestyle_stats,
56 }
57 highlight = self._pick_highlight(all_stats)
59 lines: list[str] = []
61 # 1. Title
62 lines.append(f"📊 *健康周报 {week_start} ~ {week_end}*")
63 lines.append("")
65 # 2. Sleep
66 lines.append("😴 *睡眠*")
67 if sleep_stats.get("count", 0) > 0:
68 lines.append(f" • 平均得分：{sleep_stats['avg_score']:.0f} 分")
69 lines.append(
70 f" • 平均时长：{self._format_duration(sleep_stats['avg_duration_sec'])}"
71 )
72 if sleep_stats.get("best_date"):
73 lines.append(
74 f" • 最佳：{sleep_stats['best_date']} ({sleep_stats['best_score']:.0f} 分)"
75 )
76 if sleep_stats.get("worst_date"):
77 lines.append(
78 f" • 最差：{sleep_stats['worst_date']} ({sleep_stats['worst_score']:.0f} 分)"
79 )
80 else:
81 lines.append(" • 本周无睡眠数据")
82 lines.append("")
84 # 3. Steps & Activity
85 act_count = activity_stats.get("count", 0)
86 lines.append(f"🏃 *运动*{'（共 ' + str(act_count) + ' 次）' if act_count else ''}")
88 # Steps line
89 if steps_stats.get("count", 0) > 0:
90 lines.append(
91 f" • 步数达标（≥10000）：{steps_stats['goal_days']} 天"
92 f" | 平均 {steps_stats['avg_steps']:,.0f} 步"
93 )
94 else:
95 lines.append(" • 本周无步数数据")
97 # Intensity minutes line
98 if intensity_stats.get("has_data"):
99 lines.append(
100 f" • 强度分钟：中等 {intensity_stats['moderate_min']}min"
101 f" + 剧烈 {intensity_stats['vigorous_min']}min"
102 f" = {intensity_stats['total_min']} min"
103 f"（WHO 目标 150 min，完成 {intensity_stats['goal_pct']}%）"
104 )
106 # HR zones
107 zone_seconds = activity_stats.get("zone_seconds", [0, 0, 0, 0, 0])
108 if any(s > 0 for s in zone_seconds):
109 zone_parts = " ".join(
110 f"Z{i + 1} {self._format_duration(zone_seconds[i])}"
111 for i in range(5)
112 )
113 lines.append("")
114 lines.append(" *心率区间（全周累计）*")
115 lines.append(f" • {zone_parts}")
117 # Activity detail list
118 activities = activity_stats.get("activities", [])
119 if activities:
120 lines.append("")
121 lines.append(" *运动明细*")
122 for act in activities:
123 hr_part = ""
124 if act["avg_hr"] or act["max_hr"]:
125 hr_part = f" | ♥ {act['avg_hr']}/{act['max_hr']} bpm"
126 cal_part = f" | {act['calories']}kcal" if act["calories"] else ""
127 lines.append(
128 f" • {act['date']} {act['name']}"
129 f" {act['duration_min']}min{cal_part}{hr_part}"
130 )
131 lines.append("")
133 # 4. Recovery
134 lines.append("❤️ *恢复*")
135 if recovery_stats.get("count", 0) > 0:
136 hrv_line = f" • 平均 HRV：{recovery_stats['avg_hrv']:.0f} ms"
137 if prev_recovery.get("count", 0) > 0:
138 diff = recovery_stats["avg_hrv"] - prev_recovery["avg_hrv"]
139 sign = "+" if diff >= 0 else ""
140 hrv_line += f"（{sign}{diff:.0f} vs 上周）"
141 lines.append(hrv_line)
143 rhr_line = f" • 平均静息心率：{recovery_stats['avg_rhr']:.0f} bpm"
144 if prev_recovery.get("count", 0) > 0:
145 diff = recovery_stats["avg_rhr"] - prev_recovery["avg_rhr"]
146 sign = "+" if diff >= 0 else ""
147 rhr_line += f"（{sign}{diff:.0f} vs 上周）"
148 lines.append(rhr_line)
149 else:
150 lines.append(" • 本周无恢复数据")
151 lines.append("")
153 # 5. Body Battery
154 lines.append("⚡ *Body Battery*")
155 if battery_stats.get("count", 0) > 0:
156 lines.append(f" • 平均充电值：{battery_stats['avg_charged']:.0f}")
157 lines.append(f" • 平均消耗值：{battery_stats['avg_drained']:.0f}")
158 else:
159 lines.append(" • 本周无 Body Battery 数据")
160 lines.append("")
162 # 6. Lifestyle
163 lines.append("🥗 *生活方式*")
164 if lifestyle_stats.get("has_data"):
165 lines.append(f" • 饮酒天数：{lifestyle_stats['alcohol_days']} 天")
166 if lifestyle_stats.get("fasting_modes"):
167 modes = ", ".join(
168 f"{m}（{c}天）" for m, c in lifestyle_stats["fasting_modes"].items()
169 )
170 lines.append(f" • 禁食模式：{modes}")
171 lines.append(f" • 补剂记录天数：{lifestyle_stats['supplement_days']} 天")
172 else:
173 lines.append(" • 本周无手动记录")
174 lines.append("")
176 # 7. Highlight
177 lines.append(f"💡 *本周亮点：* {highlight}")
179 return "\n".join(lines)
181 # ------------------------------------------------------------------
182 # Private helpers – data fetching
183 # ------------------------------------------------------------------
185 def _get_sleep_stats(self, start: date, end: date) -> dict[str, Any]:
186 """Compute sleep statistics for the given date range."""
187 records = self.query.get_metric_range("sleep", start, end)
188 if not records:
189 return {"count": 0}
191 scores = [r["sleep_score"] for r in records if r.get("sleep_score")]
192 durations = [r["total_sleep_seconds"] for r in records if r.get("total_sleep_seconds")]
194 if not scores:
195 return {"count": 0}
197 best = max(records, key=lambda r: r.get("sleep_score", 0))
198 worst = min(records, key=lambda r: r.get("sleep_score", 100))
200 return {
201 "count": len(scores),
202 "avg_score": sum(scores) / len(scores),
203 "avg_duration_sec": int(sum(durations) / len(durations)) if durations else 0,
204 "best_date": best.get("date"),
205 "best_score": best.get("sleep_score", 0),
206 "worst_date": worst.get("date"),
207 "worst_score": worst.get("sleep_score", 0),
208 }
210 def _get_steps_stats(self, start: date, end: date) -> dict[str, Any]:
211 """Compute step statistics for the given date range."""
212 records = self.query.get_metric_range("steps", start, end)
213 if not records:
214 return {"count": 0}
216 step_values = [r["total_steps"] for r in records if r.get("total_steps") is not None]
217 if not step_values:
218 return {"count": 0}
220 return {
221 "count": len(step_values),
222 "avg_steps": sum(step_values) / len(step_values),
223 "goal_days": sum(1 for s in step_values if s >= 10000),
224 }
226 def _get_activity_stats(self, start: date, end: date) -> dict[str, Any]:
227 """Compute activity detail statistics from get_activities_range().
229 Args:
230 start: Start date (inclusive).
231 end: End date (inclusive).
233 Returns:
234 Dict with count, activities list, and cumulative HR zone seconds.
235 """
236 records = self.query.get_activities_range(start, end)
237 if not records:
238 return {"count": 0, "activities": [], "zone_seconds": [0, 0, 0, 0, 0]}
240 activities: list[dict[str, Any]] = []
241 zone_seconds = [0, 0, 0, 0, 0]
243 for r in records:
244 raw = r.get("raw_data") or {}
245 if isinstance(raw, str):
246 import json
247 try:
248 raw = json.loads(raw)
249 except Exception:
250 raw = {}
252 # Accumulate HR zone seconds (Z1~Z5)
253 for i in range(1, 6):
254 val = raw.get(f"hrTimeInZone_{i}")
255 if val is not None:
256 try:
257 zone_seconds[i - 1] += int(val)
258 except (TypeError, ValueError):
259 pass
261 duration_sec = r.get("duration_seconds") or 0
262 activities.append({
263 "date": r.get("date", "")[-5:].replace("-", "-") if r.get("date") else "",
264 "name": r.get("activity_name") or "未知",
265 "duration_min": round(duration_sec / 60) if duration_sec else 0,
266 "calories": r.get("calories") or 0,
267 "avg_hr": r.get("average_heart_rate") or 0,
268 "max_hr": r.get("max_heart_rate") or 0,
269 })
271 return {
272 "count": len(activities),
273 "activities": activities,
274 "zone_seconds": zone_seconds,
275 }
277 def _get_intensity_stats(self, start: date, end: date) -> dict[str, Any]:
278 """Compute intensity minutes vs WHO 150-min target.
280 Args:
281 start: Start date (inclusive).
282 end: End date (inclusive).
284 Returns:
285 Dict with moderate_min, vigorous_min, total_min (WHO-weighted), goal_pct.
286 """
287 records = self.query.get_metric_range("intensity_minutes", start, end)
288 if not records:
289 return {"has_data": False}
291 moderate_min = sum(r.get("moderate_minutes") or 0 for r in records)
292 vigorous_min = sum(r.get("vigorous_minutes") or 0 for r in records)
293 total_min = moderate_min + vigorous_min * 2
294 goal_pct = round(total_min / 150 * 100)
296 return {
297 "has_data": True,
298 "moderate_min": moderate_min,
299 "vigorous_min": vigorous_min,
300 "total_min": total_min,
301 "goal_pct": goal_pct,
302 }
304 def _get_recovery_stats(self, start: date, end: date) -> dict[str, Any]:
305 """Compute HRV and resting heart rate statistics for the given date range."""
306 hrv_records = self.query.get_metric_range("hrv", start, end)
307 hr_records = self.query.get_metric_range("heart_rate", start, end)
309 hrv_values = [
310 r["hrv_value"]
311 for r in hrv_records
312 if r.get("hrv_value") is not None
313 ]
314 rhr_values = [
315 r["resting_heart_rate"]
316 for r in hr_records
317 if r.get("resting_heart_rate")
318 ]
320 if not hrv_values and not rhr_values:
321 return {"count": 0}
323 return {
324 "count": max(len(hrv_values), len(rhr_values)),
325 "avg_hrv": sum(hrv_values) / len(hrv_values) if hrv_values else 0.0,
326 "avg_rhr": sum(rhr_values) / len(rhr_values) if rhr_values else 0.0,
327 }
329 def _get_battery_stats(self, start: date, end: date) -> dict[str, Any]:
330 """Compute Body Battery charge/drain statistics for the given date range."""
331 records = self.query.get_metric_range("body_battery", start, end)
332 if not records:
333 return {"count": 0}
335 charged = [r["charged"] for r in records if r.get("charged") is not None]
336 drained = [r["drained"] for r in records if r.get("drained") is not None]
338 if not charged and not drained:
339 return {"count": 0}
341 return {
342 "count": len(records),
343 "avg_charged": sum(charged) / len(charged) if charged else 0.0,
344 "avg_drained": sum(drained) / len(drained) if drained else 0.0,
345 }
347 def _get_lifestyle_stats(self, start: date, end: date) -> dict[str, Any]:
348 """Compute lifestyle log statistics (alcohol, fasting, supplements)."""
349 logs = self.manual_storage.get_logs_in_range(start, end)
350 if not logs:
351 return {"has_data": False}
353 alcohol_days = sum(1 for log in logs if log.alcohol_entries)
354 supplement_days = sum(1 for log in logs if log.supplement_entries)
356 fasting_modes: dict[str, int] = {}
357 for log in logs:
358 if log.fasting_mode:
359 fasting_modes[log.fasting_mode] = fasting_modes.get(log.fasting_mode, 0) + 1
361 return {
362 "has_data": True,
363 "alcohol_days": alcohol_days,
364 "supplement_days": supplement_days,
365 "fasting_modes": fasting_modes,
366 }
368 # ------------------------------------------------------------------
369 # Private helpers – formatting
370 # ------------------------------------------------------------------
372 def _format_duration(self, seconds: int) -> str:
373 """Format a duration in seconds to 'Xh Ym' string.
375 Args:
376 seconds: Total duration in seconds.
378 Returns:
379 Human-readable duration string, e.g. '7h 23m'.
380 """
381 if seconds <= 0:
382 return "0h 0m"
383 hours = seconds // 3600
384 minutes = (seconds % 3600) // 60
385 return f"{hours}h {minutes}m"
387 def _pick_highlight(self, stats: dict[str, Any]) -> str:
388 """Select the most noteworthy insight from aggregated stats.
390 Args:
391 stats: Dictionary containing all section stats.
393 Returns:
394 A single highlight sentence in Chinese.
395 """
396 highlights: list[tuple[float, str]] = []
398 sleep = stats.get("sleep", {})
399 if sleep.get("count", 0) > 0:
400 score = sleep["avg_score"]
401 if score >= 85:
402 highlights.append((score, f"睡眠质量优秀，平均得分 {score:.0f} 分 🌟"))
403 elif score < 65:
404 highlights.append((100 - score, f"睡眠质量偏低，平均得分 {score:.0f} 分，建议关注"))
406 steps = stats.get("steps", {})
407 if steps.get("count", 0) > 0:
408 goal_days = steps["goal_days"]
409 if goal_days >= 5:
410 highlights.append((goal_days * 10, f"本周 {goal_days} 天达到万步目标，运动坚持度佳 💪"))
411 elif goal_days == 0:
412 highlights.append((60.0, "本周没有达到万步目标，需要增加日常活动量"))
414 intensity = stats.get("intensity", {})
415 if intensity.get("has_data"):
416 pct = intensity["goal_pct"]
417 total_min = intensity["total_min"]
418 if pct >= 100:
419 highlights.append((90.0, f"本周强度分钟达标（{total_min} min），运动量充足 🎯"))
420 elif pct < 50:
421 highlights.append((70.0, f"强度分钟仅 {total_min} min，不足 WHO 目标的一半，建议增加有氧运动"))
423 recovery = stats.get("recovery", {})
424 if recovery.get("count", 0) > 0 and recovery.get("avg_hrv", 0) > 0:
425 hrv = recovery["avg_hrv"]
426 if hrv >= 60:
427 highlights.append((hrv, f"HRV 均值 {hrv:.0f} ms，恢复状态良好 ✅"))
428 elif hrv < 35:
429 highlights.append((80.0, f"HRV 偏低（{hrv:.0f} ms），注意休息与压力管理"))
431 lifestyle = stats.get("lifestyle", {})
432 if lifestyle.get("has_data") and lifestyle.get("alcohol_days", 0) == 0:
433 highlights.append((50.0, "本周零饮酒，生活方式健康 🌿"))
435 if not highlights:
436 return "数据积累中，下周见更多洞察。"
438 highlights.sort(key=lambda x: x[0], reverse=True)
439 return highlights[0][1]