Python自动化导出小米手环数据：模块化脚本实现与部署指南-尧图网络科技

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数据库。
定时任务：在服务器或常年开机的电脑上，使用系统的cron（Linux/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_email@example.com') # 示例，请替换为环境变量 MI_PASSWORD = os.getenv('MI_PASSWORD', 'your_password') # 示例，请替换为环境变量 # 数据获取时间范围：默认导出昨天一整天的数据 TODAY = datetime.now().date() YESTERDAY = TODAY - timedelta(days=1) 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 = f'mi_band_export_{datetime.now().strftime("%Y%m")}.log' @staticmethod def ensure_dirs(): """确保必要的目录存在""" os.makedirs(Config.OUTPUT_DIR, exist_ok=True) os.makedirs(Config.LOG_DIR, exist_ok=True)

requirements.txt：

pandas>=1.5.0 requests>=2.28.0 schedule>=1.2.0 # 可选，用于脚本内循环 # 假设的第三方库，请根据实际找到的库替换 # miband-toolkit>=0.1.0

3.2 数据获取模块的实现

这是最核心也最可能变化的模块。这里以伪代码展示理想接口，实际实现需依赖你找到的具体库。

data_fetcher.py：

import logging from datetime import date, datetime from typing import Dict, List, Any, Optional import pandas as pd # 假设导入的第三方库 # from miband_toolkit import MiBandClient logging.basicConfig(level=logging.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(f"DataFetcher initialized for user: {username}") def fetch_steps(self, target_date: date) -> pd.DataFrame: """ 获取某一天的步数数据（通常为按小时或分钟聚合）。 返回一个DataFrame，列可能包含：timestamp, steps, distance, calories """ logger.info(f"Fetching 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(f"Fetched {len(df)} records of steps data.") return df def fetch_heart_rate(self, target_date: date) -> pd.DataFrame: """ 获取某一天的心率数据（可能是连续监测的片段或定期测量值）。 返回DataFrame，列可能包含：timestamp, heart_rate """ logger.info(f"Fetching 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(hour=h, minute=m) 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(f"Fetched {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(f"Fetching 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(hour=23, minute=0)], 'sleep_end': [datetime.combine(target_date + timedelta(days=1), datetime.min.time()).replace(hour=7, minute=30)], '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(f"Fetched 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(days=1) 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_index=True) else: all_data[data_type] = pd.DataFrame() # 返回空DataFrame logger.info(f"Finished fetching data from {start_date} to {end_date}.") return all_data

实操心得：在编写DataFetcher类时，务必做好异常处理（try...except）和重试机制。网络请求和第三方库调用非常容易因超时、临时错误导致失败。一个健壮的数据获取模块应该在失败后等待一段时间重试（例如使用tenacity库），并将详细的错误信息记录到日志中，而不是让整个脚本崩溃。

3.3 数据处理与存储模块

获取到原始数据后，通常需要做一些清洗、转换，然后选择合适的方式存储。

data_processor.py：

import 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'}, inplace=True) 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, on='date', how='outer'), summary_list) logger.info(f"Generated 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, index=False, encoding='utf-8-sig') # utf-8-sig支持Excel中文 logger.info(f"Saved {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_exists='append', index=False) elif data_type == 'heart_rate': df_to_save = df[['timestamp', 'heart_rate_bpm']].copy() df_to_save.to_sql('heart_rate_detail', conn, if_exists='append', index=False) elif data_type == 'sleep': df_to_save = df.copy() df_to_save.to_sql('sleep_summary', conn, if_exists='replace', index=False) # 用replace保证日期唯一 conn.commit() logger.info("Data saved to database successfully.") except Exception as e: logger.error(f"Failed 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(f"Updated daily summary for {len(summary_df)} days.") except Exception as e: logger.error(f"Failed to update daily summary: {e}") finally: conn.close()

3.4 主流程与自动化调度

最后，我们用main.py把各个模块串联起来，并加入日志和错误处理。

main.py：

import 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, encoding='utf-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, index=False) 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 2>&1
```
- 0 3 * * *表示分钟0，小时3，每天。
- cd ...确保在项目目录下执行。
- /usr/bin/python3使用绝对路径指定Python解释器（可用which python3查看）。
- >> /path/to/logs/cron.log 2>&1将标准输出和错误都重定向到日志文件。

在Windows上使用任务计划程序：

搜索并打开“任务计划程序”。
创建基本任务，设置触发器为“每天”，时间设为凌晨。
操作选择“启动程序”，程序或脚本填写你的Python解释器全路径（如C:\Python39\python.exe），参数填写main.py的完整路径，起始于填写项目目录。
在条件选项卡，可以取消“只有在计算机使用交流电源时才启动此任务”，确保关机或睡眠后也能执行。

4.2 常见问题与排查技巧实录

即使脚本逻辑正确，在实际运行中你仍会遇到各种问题。以下是我踩过的一些坑和解决方案：

问题1：第三方库登录失败或返回空数据。

可能原因：小米修改了API或登录验证流程；账户需要二次验证（如短信）；库本身有bug。
排查：
1. 检查日志：首先查看脚本输出的错误信息。
2. 手动测试：尝试用库提供的命令行工具（如果有）或写一个最简单的测试脚本，只用你的账号密码进行登录和数据获取，排除脚本其他部分干扰。
3. 更新库：查看该开源项目的GitHub Issues和最新版本，看是否有类似问题和修复。
4. 模拟登录：如果库完全失效，可能需要回到“抓包逆向”的路径。用抓包工具记录一次手机App从登录到拉取数据的完整流程，重点看Cookie、Token和请求签名。

问题2：脚本在定时任务中执行失败，但手动运行成功。

可能原因：环境变量不同（尤其是PATH）；当前工作目录不对；文件权限问题。
排查：
1. 绝对路径：在脚本中所有涉及文件读写的地方（如日志文件、输出目录、数据库），都使用绝对路径，不要用相对路径。
2. 输出日志：确保定时任务配置中，将标准输出和错误重定向到文件，这是最重要的调试信息。
3. 模拟环境：在终端中，切换到定时任务执行时使用的用户（如cron默认是用户自己的精简环境），然后手动执行命令，看是否报错。

问题3：数据库文件越来越大，查询变慢。

解决方案：定期归档旧数据。可以修改storage.py，添加一个清理或归档历史数据的方法。例如，只保留最近365天的详细数据，更早的数据可以按月聚合后存入另一张历史表，然后从详情表中删除。

def archive_old_data(self, days_to_keep=365): conn = sqlite3.connect(self.db_path) cutoff_date = (datetime.now() - timedelta(days=days_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(f"Archived data older than {cutoff_date}") except Exception as e: logger.error(f"Archive failed: {e}") conn.rollback() finally: conn.close()

问题4：想要导出更多类型的数据，如GPS运动轨迹、压力数据等。

解决方案：模块化设计的优势就在这里体现。你只需要：
1. 在config.py的TARGET_DATA列表中添加新类型，如'gps_track'。
2. 在DataFetcher类中实现一个新的方法fetch_gps_track。
3. 在DataProcessor中添加对应的清洗方法。
4. 在DataStorage中创建新的数据库表并实现存储逻辑。
5. 主流程会自动根据TARGET_DATA调用新方法。这种设计使得功能扩展非常清晰。

4.3 进阶优化方向

当基础导出稳定运行后，你可以考虑以下优化，让这个项目更具价值：

数据可视化与报告：使用matplotlib或plotly定期生成图表（如每周步数趋势图、睡眠质量雷达图），并自动保存为图片或HTML报告。甚至可以集成Jupyter Notebook，用papermill参数化运行并输出报告。
异常检测与提醒：在数据处理环节加入规则，例如“连续三天平均步数低于5000”或“夜间静息心率持续高于75”，一旦触发就通过邮件、钉钉或Telegram Bot发送提醒给你。
数据同步到云端：除了本地存储，可以将数据同步到更专业的数据库或数据平台，如InfluxDB（适合时间序列数据）、Google Sheets或Airtable，方便多设备查看和共享。
容器化部署：使用Docker将整个Python环境、脚本和依赖打包成一个镜像。这样可以在任何支持Docker的机器上一键部署，彻底解决环境依赖问题。
加入缓存机制：如果第三方库请求较慢，可以考虑对已成功获取的数据进行缓存（例如用pickle或json保存到本地），避免重复请求相同日期范围的数据，特别是在调试时。