"""数据导出/导入模块（JSON/CSV）""" from __future__ import annotations import csv import json from dataclasses import asdict, is_dataclass from datetime import date, time from pathlib import Path from typing import Literal, Optional from . import database as db from .models import Exercise, Meal, Sleep, UserConfig, Weight ExportType = Literal["exercise", "meal", "sleep", "weight"] def _json_default(value): """json.dumps 默认序列化器：处理 date/time/dataclass""" if isinstance(value, (date, time)): return value.isoformat() if is_dataclass(value): return asdict(value) raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable") def export_all_data_json(output_path: Path, user_id: int = 1) -> None: """导出所有数据为 JSON""" data = { "version": "1.0", "export_date": date.today().isoformat(), "exercises": [e.to_dict() for e in db.get_exercises()], "meals": [m.to_dict() for m in db.get_meals()], "sleep": [s.to_dict() for s in db.get_sleep_records()], "weight": [w.to_dict() for w in db.get_weight_records()], "config": db.get_config(user_id).to_dict(), } output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=_json_default), encoding="utf-8") def export_to_csv( data_type: ExportType, output_path: Path, start_date: Optional[date] = None, end_date: Optional[date] = None, ) -> None: """导出指定类型数据为 CSV（字段与导入器兼容）""" output_path.parent.mkdir(parents=True, exist_ok=True) if data_type == "exercise": records = db.get_exercises(start_date=start_date, end_date=end_date) fieldnames = ["date", "type", "duration", "calories", "distance", "heart_rate_avg", "notes"] rows = [ { "date": r.date.isoformat(), "type": r.type, "duration": r.duration, "calories": r.calories, "distance": r.distance or "", "heart_rate_avg": r.heart_rate_avg or "", "notes": r.notes or "", } for r in records ] elif data_type == "meal": records = db.get_meals(start_date=start_date, end_date=end_date) fieldnames = ["date", "meal_type", "description", "calories", "protein", "carbs", "fat"] rows = [ { "date": r.date.isoformat(), "meal_type": r.meal_type, "description": r.description, "calories": r.calories, "protein": r.protein or "", "carbs": r.carbs or "", "fat": r.fat or "", } for r in records ] elif data_type == "sleep": records = db.get_sleep_records(start_date=start_date, end_date=end_date) fieldnames = ["date", "bedtime", "wake_time", "duration", "quality", "deep_sleep_mins", "notes"] rows = [ { "date": r.date.isoformat(), "bedtime": r.bedtime.isoformat() if r.bedtime else "", "wake_time": r.wake_time.isoformat() if r.wake_time else "", "duration": r.duration, "quality": r.quality, "deep_sleep_mins": r.deep_sleep_mins or "", "notes": r.notes or "", } for r in records ] elif data_type == "weight": records = db.get_weight_records(start_date=start_date, end_date=end_date) fieldnames = ["date", "weight_kg", "body_fat_pct", "muscle_mass", "notes"] rows = [ { "date": r.date.isoformat(), "weight_kg": r.weight_kg, "body_fat_pct": r.body_fat_pct or "", "muscle_mass": r.muscle_mass or "", "notes": r.notes or "", } for r in records ] else: raise ValueError(f"不支持的数据类型: {data_type}") with open(output_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow(row) def import_from_json(input_path: Path, user_id: int = 1) -> dict[str, int]: """从 JSON 导入数据（最小实现：覆盖性导入，不做去重）""" data = json.loads(input_path.read_text(encoding="utf-8")) stats = {"exercise": 0, "meal": 0, "sleep": 0, "weight": 0} # config（可选） config = data.get("config") if isinstance(config, dict): # 仅持久化可写字段 db.save_config( user_id, UserConfig( age=config.get("age"), gender=config.get("gender"), height=config.get("height"), weight=config.get("weight"), activity_level=config.get("activity_level", "moderate"), goal=config.get("goal", "maintain"), ) ) # exercises for item in data.get("exercises", []): try: ex = Exercise( date=date.fromisoformat(item["date"]), type=item.get("type", "其他"), duration=int(item.get("duration", 0)), calories=int(item.get("calories", 0)), distance=float(item["distance"]) if item.get("distance") not in (None, "") else None, heart_rate_avg=int(item["heart_rate_avg"]) if item.get("heart_rate_avg") not in (None, "") else None, source=item.get("source", "手动"), notes=item.get("notes"), ) db.add_exercise(ex) stats["exercise"] += 1 except Exception: continue # meals for item in data.get("meals", []): try: meal = Meal( date=date.fromisoformat(item["date"]), meal_type=item.get("meal_type", "午餐"), description=item.get("description", ""), calories=int(item.get("calories", 0)), protein=float(item["protein"]) if item.get("protein") not in (None, "") else None, carbs=float(item["carbs"]) if item.get("carbs") not in (None, "") else None, fat=float(item["fat"]) if item.get("fat") not in (None, "") else None, photo_path=item.get("photo_path"), ) db.add_meal(meal) stats["meal"] += 1 except Exception: continue # sleep for item in data.get("sleep", []): try: s = Sleep( date=date.fromisoformat(item["date"]), bedtime=time.fromisoformat(item["bedtime"]) if item.get("bedtime") else None, wake_time=time.fromisoformat(item["wake_time"]) if item.get("wake_time") else None, duration=float(item.get("duration", 0)), quality=int(item.get("quality", 3)), deep_sleep_mins=int(item["deep_sleep_mins"]) if item.get("deep_sleep_mins") not in (None, "") else None, source=item.get("source", "手动"), notes=item.get("notes"), ) db.add_sleep(s) stats["sleep"] += 1 except Exception: continue # weight for item in data.get("weight", []): try: w = Weight( date=date.fromisoformat(item["date"]), weight_kg=float(item.get("weight_kg", 0)), body_fat_pct=float(item["body_fat_pct"]) if item.get("body_fat_pct") not in (None, "") else None, muscle_mass=float(item["muscle_mass"]) if item.get("muscle_mass") not in (None, "") else None, notes=item.get("notes"), ) db.add_weight(w) stats["weight"] += 1 except Exception: continue return stats