1. 项目概述为什么我们需要自动化导出小米手环数据如果你和我一样是个喜欢用数据记录生活的“量化自我”爱好者同时又是个爱折腾的Python程序员那么“小米手环运动数据导出”这个需求大概率会出现在你的待办清单里。小米手环的官方App小米运动或Zepp Life提供了丰富的数据展示但当你想要进行长期趋势分析、跨平台数据整合或者只是想拥有一个本地备份时就会立刻发现它的局限性——数据被牢牢锁在App里导出选项要么缺失要么极其繁琐一次只能导出一小段。这就是我们动手的理由。通过Python自动化我们可以绕过App的限制直接从手环同步到手机的数据源或通过官方/非官方接口中将步数、心率、睡眠、运动记录等数据以结构化的格式如CSV、JSON定期、批量地“解放”出来。这不仅仅是“导出”那么简单它意味着你可以用Pandas做数据分析用Matplotlib绘制个性化图表甚至搭建自己的个人健康数据看板。整个过程我们追求的是稳定、可重复、无需人工干预的自动化流程。2. 核心思路与技术选型解析要实现自动化导出核心在于找到可靠的数据获取途径。目前主流有几种思路每种都有其适用场景和优缺点。2.1 数据获取途径的权衡途径一逆向分析官方App的通信协议这是最直接、理论上数据最全的方法。通过抓包工具如Charles, Fiddler, Mitmproxy分析小米运动App与服务器之间的API请求和响应。一旦破解了登录、鉴权和数据拉取的接口就可以用Python的requests库模拟这些请求。优点能获取到服务器端最完整、最历史的数据。缺点技术门槛高涉及HTTPS抓包、签名算法逆向、Token管理。且官方API一旦变更脚本就容易失效。从合规角度看需严格遵守用户协议仅用于个人数据管理。途径二利用第三方开源库或项目社区中有一些先驱者已经做了部分逆向工作并封装成了Python库例如mi-band相关的非官方项目。这些项目可能通过蓝牙直接与手环通信或者封装了部分云API。优点站在巨人肩膀上开发速度快。缺点项目可能年久失修无法适配新手环型号如小米手环8/9功能可能不完整同样面临官方变更导致失效的风险。途径三导出App本地数据库文件需Root在已Root的Android设备上小米运动App的数据通常存储在SQLite数据库中。你可以通过ADB将数据库文件拉取到电脑然后用Python的sqlite3库直接读取。优点数据来源稳定无需处理网络协议。缺点必须有一台已Root的安卓手机对绝大多数用户来说门槛太高。数据库结构可能随App版本升级而变化。途径四利用官方数据导出功能半自动化部分版本的小米运动或Zepp App提供“数据导出”功能但可能只支持导出单次运动记录为GPX/TCX文件或者以邮件形式发送一份报告通常是PDF或CSV。我们可以用自动化测试工具如uiautomator2,Appium模拟点击操作触发这个导出流程然后自动处理导出的文件。优点完全合规利用官方合法渠道。缺点导出格式和内容可能受限自动化操作手机App本身不稳定容易受界面变化影响无法实现高频、后台静默执行。我的选择与思路 对于大多数希望稳定、长期运行且不想折腾Root和深度逆向的用户我推荐一种“混合策略”优先寻找和维护良好的第三方库途径二作为数据获取核心同时将自动化脚本设计为模块化将数据获取、数据处理、数据存储三个环节解耦。这样即使底层获取方式未来需要从第三方库切换为模拟API途径一也只需更换其中一个模块整体流程不受影响。本项目将主要围绕这种模块化设计思想展开。2.2 技术栈与工具准备基于上述思路我们的技术栈如下核心语言Python 3.8。因其在数据处理、网络请求和自动化方面的强大生态。网络请求requests库用于处理HTTP请求如果第三方库底层需要。数据处理pandas库用于数据清洗、分析和导出为CSV/Excel。数据存储轻量级选择可以用CSV文件长期追踪可以用sqlite3或SQLAlchemy配合SQLite数据库。定时任务在服务器或常年开机的电脑上使用系统的cronLinux/macOS或任务计划程序Windows。在脚本内部也可以用schedule库实现简单的周期循环但不如系统级任务计划可靠。环境管理强烈建议使用venv或conda创建独立的Python虚拟环境避免包冲突。备选手机自动化如果采用途径四需要uiautomator2或Appium。本文暂不深入因其复杂度较高。注意在尝试任何与官方服务器通信的方法前请务必阅读并理解小米的用户协议和隐私政策。自动化脚本应仅用于管理您个人的数据且频率应合理避免对服务器造成不必要的压力这既是法律合规要求也是技术道德。3. 实战构建模块化自动化导出脚本我们假设找到一个名为miband-toolkit此为示例需自行搜索可用开源项目的第三方库它能帮助我们获取数据。我们的脚本结构将分为配置、数据获取、数据处理、数据持久化和主流程控制几个模块。3.1 项目结构与配置管理首先创建项目目录结构mi-band-auto-export/ ├── config.py # 配置文件 ├── data_fetcher.py # 数据获取模块 ├── data_processor.py # 数据处理模块 ├── storage.py # 数据存储模块 ├── main.py # 主程序入口 ├── requirements.txt # 项目依赖 └── logs/ # 日志目录config.py使用配置文件或环境变量来管理敏感信息和可变参数而不是硬编码在脚本里。import os from datetime import datetime, timedelta class Config: # 手环账户信息建议从环境变量读取而非直接写在这里 MI_USERNAME os.getenv(MI_USERNAME, your_emailexample.com) # 示例请替换为环境变量 MI_PASSWORD os.getenv(MI_PASSWORD, your_password) # 示例请替换为环境变量 # 数据获取时间范围默认导出昨天一整天的数据 TODAY datetime.now().date() YESTERDAY TODAY - timedelta(days1) START_DATE YESTERDAY END_DATE YESTERDAY # 如果只需一天起止日期相同 # 导出数据类型 TARGET_DATA [steps, heart_rate, sleep] # 步数心率睡眠 # 存储配置 OUTPUT_DIR ./exported_data DB_PATH ./mi_band_data.db # 日志配置 LOG_DIR ./logs LOG_FILE fmi_band_export_{datetime.now().strftime(%Y%m)}.log staticmethod def ensure_dirs(): 确保必要的目录存在 os.makedirs(Config.OUTPUT_DIR, exist_okTrue) os.makedirs(Config.LOG_DIR, exist_okTrue)requirements.txtpandas1.5.0 requests2.28.0 schedule1.2.0 # 可选用于脚本内循环 # 假设的第三方库请根据实际找到的库替换 # miband-toolkit0.1.03.2 数据获取模块的实现这是最核心也最可能变化的模块。这里以伪代码展示理想接口实际实现需依赖你找到的具体库。data_fetcher.pyimport logging from datetime import date, datetime from typing import Dict, List, Any, Optional import pandas as pd # 假设导入的第三方库 # from miband_toolkit import MiBandClient logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DataFetcher: def __init__(self, username: str, password: str): 初始化获取器进行登录认证。 实际代码取决于你使用的库。 self.username username self.password password self.client None # 示例self.client MiBandClient(username, password) # self.client.login() logger.info(fDataFetcher initialized for user: {username}) def fetch_steps(self, target_date: date) - pd.DataFrame: 获取某一天的步数数据通常为按小时或分钟聚合。 返回一个DataFrame列可能包含timestamp, steps, distance, calories logger.info(fFetching steps data for {target_date}) # 伪代码 # raw_data self.client.get_steps_data(target_date) # df pd.DataFrame(raw_data) # df[date] target_date # return df # 模拟数据实际应替换为库调用 hours list(range(24)) simulated_steps [int(1000 500 * (h/12 - (h/12)**2)) for h in hours] # 模拟一个步数曲线 df pd.DataFrame({ hour: hours, steps: simulated_steps, date: target_date }) df[calories] df[steps] * 0.04 # 非常粗略的估算 logger.info(fFetched {len(df)} records of steps data.) return df def fetch_heart_rate(self, target_date: date) - pd.DataFrame: 获取某一天的心率数据可能是连续监测的片段或定期测量值。 返回DataFrame列可能包含timestamp, heart_rate logger.info(fFetching heart rate data for {target_date}) # 伪代码raw_data self.client.get_heart_rate_data(target_date) # 模拟数据 import random timestamps [datetime.combine(target_date, datetime.min.time()).replace(hourh, minutem) for h in range(24) for m in [0, 30]] # 每半小时一个点 heart_rates [random.randint(60, 100) for _ in timestamps] df pd.DataFrame({ timestamp: timestamps, heart_rate_bpm: heart_rates }) logger.info(fFetched {len(df)} heart rate samples.) return df def fetch_sleep(self, target_date: date) - pd.DataFrame: 获取某一天的睡眠分析数据。 返回DataFrame列可能包含start_time, end_time, deep_sleep_min, light_sleep_min, awake_min, sleep_score logger.info(fFetching sleep data for {target_date}) # 伪代码sleep_summary self.client.get_sleep_data(target_date) # 模拟一个简单的睡眠结构 sleep_data { date: [target_date], sleep_start: [datetime.combine(target_date, datetime.min.time()).replace(hour23, minute0)], sleep_end: [datetime.combine(target_date timedelta(days1), datetime.min.time()).replace(hour7, minute30)], deep_sleep_minutes: [120], light_sleep_minutes: [300], rem_sleep_minutes: [90], awake_minutes: [30], total_sleep_minutes: [510], sleep_score: [85] } df pd.DataFrame(sleep_data) logger.info(fFetched sleep summary for {target_date}) return df def fetch_all(self, start_date: date, end_date: date, data_types: List[str]) - Dict[str, pd.DataFrame]: 批量获取指定日期范围内、指定类型的数据。 all_data {} current_date start_date delta timedelta(days1) while current_date end_date: daily_data {} if steps in data_types: daily_data[steps] self.fetch_steps(current_date) if heart_rate in data_types: daily_data[heart_rate] self.fetch_heart_rate(current_date) if sleep in data_types: # 睡眠数据通常按夜计算日期指向睡眠开始的日期 daily_data[sleep] self.fetch_sleep(current_date) # 按数据类型合并多日数据 for data_type, df in daily_data.items(): if data_type not in all_data: all_data[data_type] [] all_data[data_type].append(df) current_date delta # 将列表合并为单个DataFrame for data_type in all_data: if all_data[data_type]: # 检查列表是否非空 all_data[data_type] pd.concat(all_data[data_type], ignore_indexTrue) else: all_data[data_type] pd.DataFrame() # 返回空DataFrame logger.info(fFinished fetching data from {start_date} to {end_date}.) return all_data实操心得在编写DataFetcher类时务必做好异常处理try...except和重试机制。网络请求和第三方库调用非常容易因超时、临时错误导致失败。一个健壮的数据获取模块应该在失败后等待一段时间重试例如使用tenacity库并将详细的错误信息记录到日志中而不是让整个脚本崩溃。3.3 数据处理与存储模块获取到原始数据后通常需要做一些清洗、转换然后选择合适的方式存储。data_processor.pyimport pandas as pd import logging logger logging.getLogger(__name__) class DataProcessor: staticmethod def clean_steps_data(df: pd.DataFrame) - pd.DataFrame: 清洗步数数据处理缺失值确保数据类型正确 if df.empty: return df df_clean df.copy() # 确保日期列是datetime类型 if date in df_clean.columns: df_clean[date] pd.to_datetime(df_clean[date]).dt.date # 填充可能的步数缺失值例如用0填充 if steps in df_clean.columns: df_clean[steps] df_clean[steps].fillna(0).astype(int) # 计算累计步数等衍生字段可选 # df_clean[cumulative_steps] df_clean[steps].cumsum() logger.debug(Steps data cleaned.) return df_clean staticmethod def clean_heart_rate_data(df: pd.DataFrame) - pd.DataFrame: 清洗心率数据去除异常值如30或200的生理学不可能值 if df.empty: return df df_clean df.copy() if heart_rate_bpm in df_clean.columns: # 简单的异常值过滤 df_clean df_clean[(df_clean[heart_rate_bpm] 30) (df_clean[heart_rate_bpm] 200)] df_clean[heart_rate_bpm] df_clean[heart_rate_bpm].astype(int) logger.debug(Heart rate data cleaned.) return df_clean staticmethod def aggregate_daily_summary(steps_df: pd.DataFrame, hr_df: pd.DataFrame, sleep_df: pd.DataFrame) - pd.DataFrame: 生成每日数据摘要便于快速查看。 例如总步数、平均心率、睡眠时长。 summary_list [] # 这里需要根据实际数据日期进行分组聚合以下为简化示例 # 假设steps_df有date和steps列且steps是每小时数据 if not steps_df.empty and date in steps_df.columns: daily_steps steps_df.groupby(date)[steps].sum().reset_index() daily_steps.rename(columns{steps: total_steps}, inplaceTrue) summary_list.append(daily_steps) if not hr_df.empty and timestamp in hr_df.columns: hr_df[date] pd.to_datetime(hr_df[timestamp]).dt.date daily_hr hr_df.groupby(date)[heart_rate_bpm].agg([mean, min, max]).reset_index() daily_hr.columns [date, avg_heart_rate, min_heart_rate, max_heart_rate] summary_list.append(daily_hr) if not sleep_df.empty and date in sleep_df.columns: # 睡眠数据通常一天一行直接选取所需列 sleep_summary sleep_df[[date, total_sleep_minutes, sleep_score]].copy() summary_list.append(sleep_summary) # 合并所有摘要 from functools import reduce if summary_list: # 使用reduce按date列合并多个DataFrame daily_summary reduce(lambda left, right: pd.merge(left, right, ondate, howouter), summary_list) logger.info(fGenerated daily summary for {len(daily_summary)} days.) return daily_summary else: return pd.DataFrame()storage.py提供多种存储后端这里实现CSV文件和SQLite数据库两种。import pandas as pd import sqlite3 import os import logging from datetime import datetime logger logging.getLogger(__name__) class DataStorage: def __init__(self, output_dir: str, db_path: str): self.output_dir output_dir self.db_path db_path self._init_database() def _init_database(self): 初始化SQLite数据库创建表如果不存在 conn sqlite3.connect(self.db_path) cursor conn.cursor() # 创建步数详情表 cursor.execute( CREATE TABLE IF NOT EXISTS steps_detail ( id INTEGER PRIMARY KEY AUTOINCREMENT, date DATE NOT NULL, hour INTEGER, steps INTEGER, distance_m REAL, calories REAL, fetch_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建心率记录表 cursor.execute( CREATE TABLE IF NOT EXISTS heart_rate_detail ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME NOT NULL, heart_rate_bpm INTEGER, fetch_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建睡眠摘要表 cursor.execute( CREATE TABLE IF NOT EXISTS sleep_summary ( id INTEGER PRIMARY KEY AUTOINCREMENT, date DATE NOT NULL, sleep_start DATETIME, sleep_end DATETIME, deep_sleep_minutes INTEGER, light_sleep_minutes INTEGER, rem_sleep_minutes INTEGER, awake_minutes INTEGER, total_sleep_minutes INTEGER, sleep_score INTEGER, fetch_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(date) -- 保证每天只有一条睡眠记录 ) ) # 创建每日摘要表物化视图便于查询 cursor.execute( CREATE TABLE IF NOT EXISTS daily_summary ( date DATE PRIMARY KEY, total_steps INTEGER DEFAULT 0, avg_heart_rate REAL, min_heart_rate INTEGER, max_heart_rate INTEGER, total_sleep_minutes INTEGER, sleep_score INTEGER, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() logger.info(Database initialized.) def save_to_csv(self, data_dict: dict, date_str: str None): 将数据字典数据类型-DataFrame保存为CSV文件 if date_str is None: date_str datetime.now().strftime(%Y%m%d) for data_type, df in data_dict.items(): if df is not None and not df.empty: filename os.path.join(self.output_dir, f{data_type}_{date_str}.csv) df.to_csv(filename, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel中文 logger.info(fSaved {data_type} data to {filename}) def save_to_database(self, data_dict: dict): 将数据保存到SQLite数据库 conn sqlite3.connect(self.db_path) try: for data_type, df in data_dict.items(): if df is None or df.empty: continue if data_type steps: # 确保列名与表结构匹配 df_to_save df[[date, hour, steps]].copy() # 这里可以添加更多列的处理 df_to_save.to_sql(steps_detail, conn, if_existsappend, indexFalse) elif data_type heart_rate: df_to_save df[[timestamp, heart_rate_bpm]].copy() df_to_save.to_sql(heart_rate_detail, conn, if_existsappend, indexFalse) elif data_type sleep: df_to_save df.copy() df_to_save.to_sql(sleep_summary, conn, if_existsreplace, indexFalse) # 用replace保证日期唯一 conn.commit() logger.info(Data saved to database successfully.) except Exception as e: logger.error(fFailed to save data to database: {e}) conn.rollback() finally: conn.close() def update_daily_summary(self, summary_df: pd.DataFrame): 用新的摘要数据更新或插入daily_summary表 if summary_df.empty: return conn sqlite3.connect(self.db_path) try: for _, row in summary_df.iterrows(): # 使用INSERT OR REPLACE来更新每日摘要 cursor conn.cursor() cursor.execute( INSERT OR REPLACE INTO daily_summary (date, total_steps, avg_heart_rate, min_heart_rate, max_heart_rate, total_sleep_minutes, sleep_score) VALUES (?, ?, ?, ?, ?, ?, ?) , ( row[date], row.get(total_steps, 0), row.get(avg_heart_rate), row.get(min_heart_rate), row.get(max_heart_rate), row.get(total_sleep_minutes), row.get(sleep_score) )) conn.commit() logger.info(fUpdated daily summary for {len(summary_df)} days.) except Exception as e: logger.error(fFailed to update daily summary: {e}) finally: conn.close()3.4 主流程与自动化调度最后我们用main.py把各个模块串联起来并加入日志和错误处理。main.pyimport logging import sys from datetime import datetime, timedelta import traceback from config import Config from data_fetcher import DataFetcher from data_processor import DataProcessor from storage import DataStorage def setup_logging(): 配置日志同时输出到文件和终端 Config.ensure_dirs() log_file_path f{Config.LOG_DIR}/{Config.LOG_FILE} # 创建logger logger logging.getLogger() logger.setLevel(logging.INFO) # 避免重复添加handler if logger.hasHandlers(): logger.handlers.clear() # 文件handler file_handler logging.FileHandler(log_file_path, encodingutf-8) file_formatter logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) file_handler.setFormatter(file_formatter) # 控制台handler console_handler logging.StreamHandler(sys.stdout) console_formatter logging.Formatter(%(asctime)s - %(levelname)s: %(message)s, datefmt%H:%M:%S) console_handler.setFormatter(console_formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger def main(): 主执行函数 logger setup_logging() logger.info(*50) logger.info(小米手环数据自动化导出任务开始) logger.info(f任务日期: {Config.START_DATE} 至 {Config.END_DATE}) fetcher None try: # 1. 初始化数据获取器 # 注意这里需要你填入真实的第三方库初始化逻辑 fetcher DataFetcher(Config.MI_USERNAME, Config.MI_PASSWORD) logger.info(数据获取器初始化成功) # 2. 获取数据 raw_data_dict fetcher.fetch_all(Config.START_DATE, Config.END_DATE, Config.TARGET_DATA) logger.info(f成功获取 {len(raw_data_dict)} 种类型的数据) # 3. 处理数据 processed_data_dict {} if steps in raw_data_dict and not raw_data_dict[steps].empty: processed_data_dict[steps] DataProcessor.clean_steps_data(raw_data_dict[steps]) if heart_rate in raw_data_dict and not raw_data_dict[heart_rate].empty: processed_data_dict[heart_rate] DataProcessor.clean_heart_rate_data(raw_data_dict[heart_rate]) if sleep in raw_data_dict and not raw_data_dict[sleep].empty: processed_data_dict[sleep] raw_data_dict[sleep] # 睡眠数据假设已较干净 # 4. 生成每日摘要 daily_summary DataProcessor.aggregate_daily_summary( processed_data_dict.get(steps, pd.DataFrame()), processed_data_dict.get(heart_rate, pd.DataFrame()), processed_data_dict.get(sleep, pd.DataFrame()) ) # 5. 存储数据 storage DataStorage(Config.OUTPUT_DIR, Config.DB_PATH) # 保存原始/处理后的详细数据 storage.save_to_csv(processed_data_dict) storage.save_to_database(processed_data_dict) # 更新摘要表 if not daily_summary.empty: storage.update_daily_summary(daily_summary) # 也保存一份摘要CSV summary_file f{Config.OUTPUT_DIR}/daily_summary_{datetime.now().strftime(%Y%m%d)}.csv daily_summary.to_csv(summary_file, indexFalse) logger.info(f每日摘要已保存至: {summary_file}) logger.info(数据导出任务完成) except Exception as e: logger.error(f任务执行过程中发生错误: {e}) logger.error(traceback.format_exc()) # 这里可以添加错误通知比如发送邮件或钉钉消息 # send_alert_notification(f小米手环数据导出失败: {e}) finally: if fetcher: # 如果有退出或清理逻辑 # fetcher.logout() pass logger.info(*50) if __name__ __main__: # 直接运行一次 main() # 如果需要脚本内定时循环适用于调试或非服务器环境可以使用schedule库 # import schedule # import time # schedule.every().day.at(02:00).do(main) # 每天凌晨2点执行 # logger.info(定时任务已启动等待执行...) # while True: # schedule.run_pending() # time.sleep(60)4. 部署、调度与进阶优化脚本写好了如何让它真正自动、稳定地跑起来4.1 系统级定时任务部署在个人电脑或服务器上使用系统自带的定时任务工具是最可靠的方式。在Linux/macOS上使用Cron打开终端输入crontab -e编辑当前用户的cron任务。添加一行例如每天凌晨3点执行此时网络和手机同步通常已完成0 3 * * * cd /path/to/your/mi-band-auto-export /usr/bin/python3 /path/to/your/mi-band-auto-export/main.py /path/to/logs/cron.log 210 3 * * *表示分钟0小时3每天。cd ...确保在项目目录下执行。/usr/bin/python3使用绝对路径指定Python解释器可用which python3查看。 /path/to/logs/cron.log 21将标准输出和错误都重定向到日志文件。在Windows上使用任务计划程序搜索并打开“任务计划程序”。创建基本任务设置触发器为“每天”时间设为凌晨。操作选择“启动程序”程序或脚本填写你的Python解释器全路径如C:\Python39\python.exe参数填写main.py的完整路径起始于填写项目目录。在条件选项卡可以取消“只有在计算机使用交流电源时才启动此任务”确保关机或睡眠后也能执行。4.2 常见问题与排查技巧实录即使脚本逻辑正确在实际运行中你仍会遇到各种问题。以下是我踩过的一些坑和解决方案问题1第三方库登录失败或返回空数据。可能原因小米修改了API或登录验证流程账户需要二次验证如短信库本身有bug。排查检查日志首先查看脚本输出的错误信息。手动测试尝试用库提供的命令行工具如果有或写一个最简单的测试脚本只用你的账号密码进行登录和数据获取排除脚本其他部分干扰。更新库查看该开源项目的GitHub Issues和最新版本看是否有类似问题和修复。模拟登录如果库完全失效可能需要回到“抓包逆向”的路径。用抓包工具记录一次手机App从登录到拉取数据的完整流程重点看Cookie、Token和请求签名。问题2脚本在定时任务中执行失败但手动运行成功。可能原因环境变量不同尤其是PATH当前工作目录不对文件权限问题。排查绝对路径在脚本中所有涉及文件读写的地方如日志文件、输出目录、数据库都使用绝对路径不要用相对路径。输出日志确保定时任务配置中将标准输出和错误重定向到文件这是最重要的调试信息。模拟环境在终端中切换到定时任务执行时使用的用户如cron默认是用户自己的精简环境然后手动执行命令看是否报错。问题3数据库文件越来越大查询变慢。解决方案定期归档旧数据。可以修改storage.py添加一个清理或归档历史数据的方法。例如只保留最近365天的详细数据更早的数据可以按月聚合后存入另一张历史表然后从详情表中删除。def archive_old_data(self, days_to_keep365): conn sqlite3.connect(self.db_path) cutoff_date (datetime.now() - timedelta(daysdays_to_keep)).strftime(%Y-%m-%d) try: # 示例归档步数详情 cursor conn.cursor() # 1. 将旧数据聚合后插入历史表需先创建 cursor.execute( INSERT INTO steps_detail_monthly (year_month, total_steps, avg_daily_steps) SELECT strftime(%Y-%m, date) as year_month, SUM(steps) as total_steps, AVG(steps) as avg_daily_steps FROM steps_detail WHERE date ? GROUP BY year_month , (cutoff_date,)) # 2. 删除已归档的旧数据 cursor.execute(DELETE FROM steps_detail WHERE date ?, (cutoff_date,)) conn.commit() logger.info(fArchived data older than {cutoff_date}) except Exception as e: logger.error(fArchive failed: {e}) conn.rollback() finally: conn.close()问题4想要导出更多类型的数据如GPS运动轨迹、压力数据等。解决方案模块化设计的优势就在这里体现。你只需要在config.py的TARGET_DATA列表中添加新类型如gps_track。在DataFetcher类中实现一个新的方法fetch_gps_track。在DataProcessor中添加对应的清洗方法。在DataStorage中创建新的数据库表并实现存储逻辑。主流程会自动根据TARGET_DATA调用新方法。这种设计使得功能扩展非常清晰。4.3 进阶优化方向当基础导出稳定运行后你可以考虑以下优化让这个项目更具价值数据可视化与报告使用matplotlib或plotly定期生成图表如每周步数趋势图、睡眠质量雷达图并自动保存为图片或HTML报告。甚至可以集成Jupyter Notebook用papermill参数化运行并输出报告。异常检测与提醒在数据处理环节加入规则例如“连续三天平均步数低于5000”或“夜间静息心率持续高于75”一旦触发就通过邮件、钉钉或Telegram Bot发送提醒给你。数据同步到云端除了本地存储可以将数据同步到更专业的数据库或数据平台如InfluxDB适合时间序列数据、Google Sheets或Airtable方便多设备查看和共享。容器化部署使用Docker将整个Python环境、脚本和依赖打包成一个镜像。这样可以在任何支持Docker的机器上一键部署彻底解决环境依赖问题。加入缓存机制如果第三方库请求较慢可以考虑对已成功获取的数据进行缓存例如用pickle或json保存到本地避免重复请求相同日期范围的数据特别是在调试时。这个项目的魅力在于它从一个简单的“导出”需求出发可以逐渐演变成你个人健康数据的私有化、自动化管理中枢。每一次脚本的稳定运行都在为你积累一份结构化的数字生命记录。
