""" PregoPal - 营养分析与报告模块 ============================== 分析饮食数据,基于中国官方营养标准输出可视化报告。 当前:基于内置 DEFAULT_NUTRITION_DB(通用知识估算) 后续:替换为中国官方标准(DRIs 2023 / 中国食物成分表) """ import json import datetime import numpy as np import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt from pathlib import Path from config import NUTRITION_DB_FILE, DEFAULT_NUTRITION_DB, REPORTS_DIR from utils import setup_chinese_font # 启动时设置中文字体 _CHINESE_FONT = setup_chinese_font() class NutritionAnalyzer: """分析饮食数据,输出可视化营养报告""" def __init__(self): self.nutrition_db = self._load_nutrition_db() def _load_nutrition_db(self): """加载营养数据库""" if NUTRITION_DB_FILE.exists(): with open(NUTRITION_DB_FILE, 'r', encoding='utf-8') as f: return json.load(f) # 使用默认数据库并保存 self._save_nutrition_db(DEFAULT_NUTRITION_DB) return DEFAULT_NUTRITION_DB def _save_nutrition_db(self, db): with open(NUTRITION_DB_FILE, 'w', encoding='utf-8') as f: json.dump(db, f, ensure_ascii=False, indent=2) def analyze_diet(self, records: list) -> dict: """ 分析一段时间内的饮食记录 Args: records: 饮食记录列表 Returns: 分析结果字典 """ if not records: return {"error": "暂无饮食记录", "score": 0} # 统计各餐次频率 meal_counts = {"早餐": 0, "午餐": 0, "晚餐": 0, "加餐": 0} food_items = [] total_days = len(set(r["date"] for r in records)) for r in records: for meal_time, food in r.get("meals", {}).items(): if meal_time in meal_counts: meal_counts[meal_time] += 1 # 提取食物关键词 food_items.extend(self._extract_foods(food)) # 计算营养覆盖情况 nutrition_coverage = self._calculate_nutrition_coverage(food_items) # 计算饮食多样性评分 diversity_result = self._calculate_diversity_score(meal_counts, total_days) # 生成建议 suggestions = self._generate_suggestions(nutrition_coverage, diversity_result) # 综合评分:多样性(50%) + 营养覆盖(50%) diversity_num = diversity_result.get("score", 0) if isinstance(diversity_result, dict) else 0 covered_nutrients = sum(1 for v in nutrition_coverage.values() if v["covered"]) total_nutrients = max(len(nutrition_coverage), 1) nutrition_score = int(covered_nutrients / total_nutrients * 100) overall_score = int(diversity_num * 0.5 + nutrition_score * 0.5) return { "score": overall_score, "total_days": total_days, "total_records": len(records), "meal_counts": meal_counts, "food_items": list(set(food_items)), "nutrition_coverage": nutrition_coverage, "diversity_score": diversity_num, "diversity_details": diversity_result.get("details", []), "suggestions": suggestions } def _extract_foods(self, food_str: str) -> list: """从餐食描述中提取食物名称""" # 简单分词提取 separators = ['+', '、', ',', ',', '/', ' '] foods = [food_str] for sep in separators: expanded = [] for f in foods: expanded.extend(f.split(sep)) foods = expanded return [f.strip() for f in foods if f.strip()] def _calculate_nutrition_coverage(self, food_items: list) -> dict: """计算营养覆盖情况""" coverage = {} food_text = " ".join(food_items) for nutrient, info in self.nutrition_db.items(): # 检查食物列表中是否包含推荐食物 matched_foods = [f for f in info["foods"] if f in food_text] coverage[nutrient] = { "matched_foods": matched_foods, "covered": len(matched_foods) > 0, "recommended_foods": info["foods"], "benefit": info["benefit"], "daily_recommend": info.get("daily_recommend_mg") or info.get("daily_recommend_g") or info.get("daily_recommend_mcg", ""), "unit": "mg" if "daily_recommend_mg" in info else ("g" if "daily_recommend_g" in info else "mcg") } return coverage def _calculate_diversity_score(self, meal_counts: dict, total_days: int) -> dict: """计算饮食多样性评分""" if total_days == 0: return {"score": 0, "details": "暂无数据"} max_possible = total_days * len(meal_counts) actual = sum(meal_counts.values()) score = min(100, int((actual / max_possible) * 100)) details = [] for meal, count in meal_counts.items(): rate = count / total_days if total_days > 0 else 0 status = "[OK]" if rate >= 0.7 else ("[!]" if rate >= 0.4 else "[!!]") details.append(f"{status} {meal}: {count}/{total_days}天 ({rate:.0%})") return {"score": score, "details": details} def _generate_suggestions(self, nutrition_coverage: dict, diversity: dict) -> list: """生成营养建议""" suggestions = [] # 检查未覆盖的营养素 missing = [n for n, info in nutrition_coverage.items() if not info["covered"]] if missing: suggestions.append(f"[!] 以下营养素摄入不足: {', '.join(missing[:5])}") for n in missing[:3]: info = nutrition_coverage[n] suggestions.append(f" -> 建议补充 {n}({info['benefit']}):可多吃 {', '.join(info['recommended_foods'][:3])}") if diversity["score"] < 60: suggestions.append("[!] 饮食多样性不足,建议增加食物种类") elif diversity["score"] >= 80: suggestions.append("[OK] 饮食多样性良好,继续保持!") suggestions.append(" 建议每天摄入12种以上食物,每周25种以上") suggestions.append(" 保证每天1.5-2L饮水") return suggestions def generate_report_chart(self, analysis: dict) -> plt.Figure: """生成营养报告图表""" if "error" in analysis: fig, ax = plt.subplots(figsize=(8, 4)) ax.text(0.5, 0.5, analysis["error"], ha='center', va='center', fontsize=14) return fig fig = plt.figure(figsize=(14, 10)) # 1. 营养覆盖雷达图 ax1 = fig.add_subplot(2, 2, 1, polar=True) nutrients = list(analysis["nutrition_coverage"].keys())[:8] coverage_values = [1 if analysis["nutrition_coverage"][n]["covered"] else 0 for n in nutrients] angles = np.linspace(0, 2 * np.pi, len(nutrients), endpoint=False).tolist() coverage_values += coverage_values[:1] angles += angles[:1] ax1.plot(angles, coverage_values, 'o-', linewidth=2, color='#FF6B9D') ax1.fill(angles, coverage_values, alpha=0.25, color='#FF6B9D') ax1.set_xticks(angles[:-1]) ax1.set_xticklabels(nutrients, fontsize=9) ax1.set_ylim(0, 1.2) ax1.set_title('营养覆盖雷达图', pad=20, fontsize=13, fontweight='bold') # 2. 各餐次频率柱状图 ax2 = fig.add_subplot(2, 2, 2) meals = list(analysis["meal_counts"].keys()) counts = list(analysis["meal_counts"].values()) colors = ['#FF9AA2', '#FFB7B2', '#FFDAC1', '#E2F0CB'] bars = ax2.bar(meals, counts, color=colors, edgecolor='white', linewidth=1.5) ax2.set_title('各餐次记录频率', fontsize=13, fontweight='bold') ax2.set_ylabel('记录次数') for bar, count in zip(bars, counts): ax2.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.1, str(count), ha='center', va='bottom', fontsize=11) # 3. 饮食多样性评分仪表盘 ax3 = fig.add_subplot(2, 2, 3) score = analysis.get("diversity_score", 0) if isinstance(score, dict): score = score.get("score", 0) ax3.pie([score, 100 - score], startangle=90, colors=['#FF6B9D', '#F0F0F0'], wedgeprops={'width': 0.3, 'edgecolor': 'white'}) ax3.text(0, 0, f'{score}', ha='center', va='center', fontsize=28, fontweight='bold') ax3.text(0, -0.15, '多样性评分', ha='center', va='center', fontsize=10, color='gray') ax3.set_title('饮食多样性评分', fontsize=13, fontweight='bold') # 4. 建议文本 ax4 = fig.add_subplot(2, 2, 4) ax4.axis('off') suggestions = analysis.get("suggestions", []) if suggestions: text = "-- 营养建议 --\n" + "\n".join(f"* {s}" for s in suggestions[:6]) else: text = "[OK] 营养状况良好!" ax4.text(0.05, 0.95, text, transform=ax4.transAxes, fontsize=10, verticalalignment='top', fontfamily='sans-serif', bbox=dict(boxstyle='round,pad=0.5', facecolor='#FFF5F5', edgecolor='#FF6B9D')) plt.tight_layout() return fig def generate_report_text(self, analysis: dict) -> str: """生成文本格式的营养报告""" if "error" in analysis: return f"⚠️ {analysis['error']}" lines = [ "=" * 50, "--- 孕期营养分析报告 ---", "=" * 50, f"分析周期: {analysis['total_days']} 天", f"记录总数: {analysis['total_records']} 条", "", "饮食多样性评分: {}/100".format(analysis.get('diversity_score', 0) if not isinstance(analysis.get('diversity_score'), dict) else analysis['diversity_score'].get('score', 0)), ] lines.append("") lines.append("各餐次记录情况:") for detail in analysis.get('diversity_details', []): lines.append(f" {detail}") lines.append("") lines.append("营养覆盖情况:") for nutrient, info in analysis['nutrition_coverage'].items(): status = "[OK]" if info['covered'] else "[!!]" matched = ", ".join(info['matched_foods']) if info['matched_foods'] else "无" lines.append(f" {status} {nutrient}: 匹配食物 [{matched}]") lines.append(f" -> {info['benefit']}") lines.append("") lines.append("改善建议:") for s in analysis.get("suggestions", []): lines.append(f" {s}") lines.append("") lines.append("=" * 50) lines.append("由 PregoPal 自动生成") return "\n".join(lines) def export_report_markdown(self, analysis: dict, filename: str = None) -> Path: """导出营养报告为 Markdown 文件""" if filename is None: filename = f"营养报告_{datetime.date.today().isoformat()}.md" md_path = REPORTS_DIR / filename content = f"""# 孕期营养分析报告 ## 基本信息 - **分析周期**: {analysis.get('total_days', 0)} 天 - **记录总数**: {analysis.get('total_records', 0)} 条 - **生成时间**: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')} ## 饮食多样性评分 **评分: {analysis.get('diversity_score', 0) if not isinstance(analysis.get('diversity_score'), dict) else analysis['diversity_score'].get('score', 0)}/100** | 餐次 | 记录天数 | 覆盖率 | |------|---------|--------| """ for detail in analysis.get('diversity_details', []): parts = detail.split(': ', 1) if len(parts) == 2: content += f"| {parts[0]} | {parts[1]} |\n" content += """ ## 营养覆盖分析 | 营养素 | 状态 | 匹配食物 | 功效 | |--------|------|---------|------| """ for nutrient, info in analysis.get('nutrition_coverage', {}).items(): status = "OK" if info['covered'] else "!!" matched = ", ".join(info['matched_foods']) if info['matched_foods'] else "-" content += f"| {nutrient} | {status} | {matched} | {info['benefit']} |\n" content += """ ## 改善建议 """ for s in analysis.get('suggestions', []): content += f"- {s}\n" content += """ --- *由 PregoPal 自动生成 | 仅供参考,不构成医疗建议* """ with open(md_path, 'w', encoding='utf-8') as f: f.write(content) return md_path