行业资讯
📅 2026/8/12 12:39:33
Python自动化抓取小米手环运动数据:从API逆向到定时导出完整方案
1. 项目缘起为什么需要自动化导出运动数据作为一名长期佩戴小米手环的运动爱好者兼数据分析爱好者我遇到了一个很实际的问题小米运动App现在叫Zepp Life确实提供了丰富的数据图表但当我想要进行更深度的分析时比如对比不同月份的运动趋势、结合天气数据研究运动表现或者只是单纯地想拥有一个自己完全掌控的本地数据备份时App内有限的数据导出功能就显得捉襟见肘了。官方通常只支持导出单次运动的详细报告如GPX轨迹或者以周/月为单位的汇总数据表格格式固定且历史数据导出非常麻烦。更关键的是这些数据散落在手机App里无法与我的其他数据源如本地健康数据库、Notion日记便捷地联动。手动截图、整理那太不“极客”了。于是一个念头自然产生能否用Python写个脚本自动、定期地将手环里的运动数据包括步数、心率、睡眠、具体运动记录等抓取下来并保存成结构化的CSV或数据库供我自由分析这不仅是数据备份更是构建个人量化生活Quantified Self体系的关键一步。经过一番摸索和实践我成功搭建了一套稳定运行的自动化方案。它不涉及任何硬件破解或逆向工程完全基于官方开放的接口虽然未公开文档通过模拟App通信协议来实现。下面我就把这套从原理分析、环境搭建、核心代码解析到部署优化的完整流程分享出来你可以直接“抄作业”。2. 核心原理数据从手环到我们手中的旅程在动手写代码之前我们必须先搞清楚数据是怎么流转的。这决定了我们的代码应该“拦截”在哪一环。整个链路可以简化为小米手环硬件 - 蓝牙 - 手机端Zepp Life App - 互联网 - 小米云端服务器。我们的目标数据最终存储在小米的云端。因此最可行的方案不是直接从手环蓝牙抓取那需要复杂的蓝牙协议逆向而是模拟手机App去云端获取我们已经同步上去的数据。这本质上是一个“爬虫”或“自动化客户端”的思路但对象是某个App的私有API。2.1 关键接口与认证机制Zepp Life App与服务器通信使用的是HTTPS协议。通过抓包分析可以使用Charles、Fiddler或mitmproxy等工具需在手机上配置代理并安装证书我们可以观察到几个核心的API端点登录认证接口用于获取访问令牌Token。通常需要用户名手机号/邮箱和密码服务器返回一个token和user_id。这是所有后续请求的敲门砖。设备列表接口获取当前账号绑定的所有小米穿戴设备手环、手表列表从中取得目标手环的device_id。运动数据汇总接口按日、周、月获取步数、距离、消耗卡路里等汇总数据。心率、睡眠等详细数据接口获取以分钟或更细粒度记录的心率、睡眠阶段数据。运动记录列表接口获取每次GPS或室内运动的记录概要开始时间、类型、时长、消耗等。单次运动详情接口根据运动记录ID获取详细的轨迹点经纬度、海拔、心率、配速等。这些接口的请求和响应格式通常是JSON。一个重要的发现是许多请求需要在Header中携带签名sign这个签名通常由请求参数、一个固定盐值salt和时间戳等通过某种算法如MD5生成用于防止简单的重放攻击。逆向这个签名算法是整个过程中最具技术挑战的一环。2.2 为什么选择PythonPython在这个任务中具有天然优势丰富的网络库requests、aiohttp可以轻松处理HTTP请求。强大的数据处理库pandas、numpy可以方便地对获取的JSON数据进行清洗、转换和导出为CSV/Excel。成熟的调度库schedule、APScheduler或操作系统级的cronLinux/Task SchedulerWindows可以完美实现定时自动化。活跃的社区事实上已经有一些开源项目如xiaomi-watch-miband相关的非官方SDK部分实现了与小米穿戴设备的交互我们可以借鉴其思路但我们的目标是云端数据更稳定且不依赖蓝牙连接。3. 环境准备与关键工具链搭建工欲善其事必先利其器。我们的开发环境不需要很复杂但几个关键工具必须准备好。3.1 Python基础环境我推荐使用Python 3.8或以上版本。使用虚拟环境venv来隔离项目依赖是一个好习惯。# 创建项目目录并进入 mkdir mi_band_auto_export cd mi_band_auto_export # 创建虚拟环境 python3 -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate3.2 依赖包安装核心依赖包并不多我们通过requirements.txt来管理。requests2.28.0 # 用于发送HTTP请求 pandas1.5.0 # 数据处理与分析导出CSV schedule1.2.0 # 轻量级定时任务调度可选 python-dotenv0.21.0 # 管理敏感配置如账号密码使用pip安装pip install -r requirements.txt3.3 抓包工具配置用于逆向分析这是最核心、最耗时但也最有趣的一步。我们需要知道API的具体地址、参数和签名算法。工具选择我强烈推荐使用mitmproxy。它是命令行工具功能强大支持脚本化对HTTPS抓包的支持很好。Charles图形化界面更友好但收费。手机配置电脑和手机处于同一Wi-Fi网络。在电脑上运行mitmproxy或Charles记下代理地址如192.168.1.100:8888。在手机Wi-Fi设置中配置手动代理填入上述地址和端口。在手机浏览器访问mitm.it或chls.pro/ssl下载并安装对应的CA证书。对于Android还需要将安装的证书移至“系统信任的凭据”中设置-安全-加密与凭据-安装证书-CA证书。对于iOS安装描述文件后需要在“设置-通用-关于本机-证书信任设置”中完全信任该根证书。抓包过程配置好后在手机上打开Zepp Life App进行登录、查看步数、查看心率图表等操作。此时mitmproxy的控制台或Charles的会话列表里就会滚动出现所有网络请求。我们需要仔细筛选主机名host包含mi、zepp、huami等关键词的请求重点关注POST请求查看其Request Body和Response Body。注意此过程仅用于个人学习与研究API调用模式。请勿高频请求对服务器造成压力也请妥善保管自己的账号信息不要将含有个人令牌的代码公开。3.4 项目目录结构一个清晰的结构有助于后续维护。mi_band_auto_export/ ├── config/ # 配置文件目录 │ └── .env # 存储账号密码等敏感信息务必加入.gitignore ├── src/ # 源代码目录 │ ├── __init__.py │ ├── auth.py # 认证模块处理登录和Token管理 │ ├── api_client.py # 核心API请求客户端封装签名和请求 │ ├── data_fetchers.py # 各种数据获取函数步数、心率、睡眠等 │ └── exporters.py # 数据导出器CSV、数据库等 ├── main.py # 主程序入口 ├── scheduler.py # 定时任务调度脚本 ├── requirements.txt └── README.md4. 核心代码实现构建小米运动API客户端一切准备就绪我们开始编写最核心的代码。我将分模块讲解并附上关键代码段。4.1 认证模块auth.py这个模块负责登录并维护有效的访问令牌。小米的登录接口通常需要经过多个步骤包括获取非对称加密密钥、加密密码等。这里我展示一个简化后的逻辑流程。# src/auth.py import hashlib import time import hmac from typing import Optional, Dict, Any import requests from dotenv import load_dotenv import os import json load_dotenv() # 从 .env 文件加载环境变量 class MiBandAuth: 小米运动认证管理器 def __init__(self): self.username os.getenv(MI_USERNAME) # 从环境变量读取 self.password os.getenv(MI_PASSWORD) self.base_url https://api-mifit.huami.com # 示例基地址实际需抓包确认 self.session requests.Session() self.session.headers.update({ User-Agent: Zepp/5.0 (iPhone; iOS 16.0; Scale/3.00), # 模拟App UA Content-Type: application/x-www-form-urlencoded, }) self.token: Optional[str] None self.user_id: Optional[str] None def _generate_signature(self, params: Dict[str, Any], salt: str some_salt) - str: 生成请求签名关键逆向部分。 这是一个示例逻辑实际算法需要通过抓包逆向得出。 通常是对排序后的参数键值对拼接加上盐值和时间戳再进行MD5或HMAC。 # 示例将参数按key排序拼接成k1v1k2v2格式末尾加盐 param_str .join([f{k}{v} for k, v in sorted(params.items())]) sign_str param_str salt # 可能是MD5也可能是其他哈希 return hashlib.md5(sign_str.encode(utf-8)).hexdigest() def login(self) - bool: 执行登录流程获取token和user_id if not self.username or not self.password: raise ValueError(用户名或密码未在环境变量中设置) # 步骤1可能先获取一个登录所需的临时token或nonce # init_url f{self.base_url}/path/to/init # init_resp self.session.post(init_url, data{...}) # nonce init_resp.json()[nonce] # 步骤2构造登录参数密码可能需要RSA加密 login_params { client_id: HuaMi, password: self.password, # 注意真实情况可能是加密后的密码 username: self.username, grant_type: password, timestamp: int(time.time() * 1000), # ... 其他必要参数 } # 生成签名并添加到参数或Header中 sign self._generate_signature(login_params) login_params[sign] sign login_url f{self.base_url}/path/to/login # 需替换为真实URL try: resp self.session.post(login_url, datalogin_params) resp.raise_for_status() data resp.json() if data.get(code) 0 or data.get(token): # 成功判断条件 self.token data[token] self.user_id data[user_id] print(f登录成功user_id: {self.user_id}) # 更新session的默认headers后续请求携带token self.session.headers.update({Authorization: fBearer {self.token}}) return True else: print(f登录失败: {data}) return False except requests.exceptions.RequestException as e: print(f登录请求出错: {e}) return False def get_auth_headers(self) - Dict[str, str]: 获取包含认证信息的请求头 if not self.token: self.login() return {Authorization: fBearer {self.token}}实操心得1签名算法是最大的拦路虎。它可能隐藏在App的Native代码SO库里。一个取巧的方法是在抓包时同一个操作重复几次对比参数中timestamp和sign的变化尝试推断算法。有时开源社区已有逆向成果可以谨慎参考。4.2 API客户端模块api_client.py这个模块封装所有对小米运动API的请求统一处理签名、错误重试等。# src/api_client.py import time import hashlib from typing import Dict, Any, Optional import requests from .auth import MiBandAuth class MiBandAPIClient: 小米运动API客户端 def __init__(self, auth: MiBandAuth): self.auth auth self.base_url https://api-mifit.huami.com def _request(self, method: str, endpoint: str, params: Optional[Dict]None, data: Optional[Dict]None) - Optional[Dict]: 发送带签名的请求 url f{self.base_url}{endpoint} all_params params or {} # 添加公共参数如时间戳、token等 all_params.update({ timestamp: int(time.time() * 1000), userid: self.auth.user_id, # ... 其他公共参数 }) # 移除空值参数 all_params {k: v for k, v in all_params.items() if v is not None} # 生成签名签名算法可能需要特定的参数排序和拼接方式 sign self._generate_sign_for_request(all_params, endpoint) all_params[sign] sign headers self.auth.get_auth_headers() try: if method.upper() GET: resp self.auth.session.get(url, paramsall_params, headersheaders) else: # POST # 注意有些API是form-data有些是x-www-form-urlencoded需根据抓包确定 headers[Content-Type] application/x-www-form-urlencoded resp self.auth.session.post(url, dataall_params, headersheaders) resp.raise_for_status() result resp.json() # 检查API返回的业务码 if result.get(code) ! 0: print(fAPI请求业务错误: {result.get(message)}, endpoint: {endpoint}) return None return result.get(data) # 通常有效数据在data字段 except requests.exceptions.RequestException as e: print(f网络请求失败 [{method} {endpoint}]: {e}) return None def _generate_sign_for_request(self, params: Dict, endpoint: str) - str: 为特定请求生成签名。 这是一个更复杂的示例实际算法需要逆向。 可能涉及请求路径 排序参数 盐值 token等。 # 示例拼接路径和参数字符串 param_str .join([f{k}{v} for k, v in sorted(params.items())]) string_to_sign f{endpoint}?{param_str}some_secret_salt{self.auth.token} return hashlib.md5(string_to_sign.encode(utf-8)).hexdigest() # 以下是具体的API方法封装 def get_device_list(self): 获取绑定的设备列表 return self._request(GET, /v2/device/list) def get_daily_summary(self, date: str): 获取某日的运动汇总数据 Args: date: 格式 2023-10-27 params {date: date, device_type: band} # device_type需根据实际设备调整 return self._request(GET, /v1/sport/summary, paramsparams) def get_heart_rate_detail(self, date: str, device_id: str): 获取某日的心率详细数据每分钟一个点 params { date: date, device_id: device_id, data_type: heart_rate, page: 1, page_size: 1440 # 一天最多1440分钟 } return self._request(GET, /v1/data/list, paramsparams) def get_sleep_detail(self, date: str, device_id: str): 获取某日的睡眠详细数据 params { date: date, device_id: device_id, data_type: sleep } return self._request(GET, /v1/data/list, paramsparams) def get_sport_records(self, start_date: str, end_date: str, limit: int 100): 获取一段时间内的运动记录列表 params { start_time: start_date, end_time: end_date, limit: limit, type: all # 或指定跑步、骑行等 } return self._request(GET, /v1/sport/record/list, paramsparams)4.3 数据获取与导出模块data_fetchers.py exporters.py有了客户端我们就可以组织业务逻辑了。# src/data_fetchers.py from typing import List, Dict, Any from datetime import datetime, timedelta import pandas as pd from .api_client import MiBandAPIClient class DataFetcher: def __init__(self, api_client: MiBandAPIClient): self.client api_client self.device_id None def init_device(self): 初始化获取主要设备ID devices self.client.get_device_list() if devices and len(devices) 0: # 假设第一个设备是手环 self.device_id devices[0][device_id] print(f使用设备: {devices[0][device_name]} (ID: {self.device_id})) return True return False def fetch_daily_data_range(self, start_date: str, end_date: str) - List[Dict]: 获取一个日期范围内的每日汇总数据 daily_data [] current datetime.strptime(start_date, %Y-%m-%d) end datetime.strptime(end_date, %Y-%m-%d) while current end: date_str current.strftime(%Y-%m-%d) print(f正在获取 {date_str} 的数据...) data self.client.get_daily_summary(date_str) if data: data[date] date_str # 添加日期字段 daily_data.append(data) else: print(f 警告{date_str} 数据获取为空或失败) current timedelta(days1) time.sleep(0.5) # 礼貌性延迟避免请求过快 return daily_data def fetch_heart_rate_for_date(self, date: str) - Optional[pd.DataFrame]: 获取某日详细心率数据并转换为DataFrame if not self.device_id: print(未初始化设备ID) return None data self.client.get_heart_rate_detail(date, self.device_id) if data and items in data: df pd.DataFrame(data[items]) # 解析时间戳等字段 if timestamp in df.columns: df[time] pd.to_datetime(df[timestamp], unitms) return df return None# src/exporters.py import pandas as pd import json from pathlib import Path from typing import List, Dict, Any class DataExporter: 数据导出器 staticmethod def to_csv(data_list: List[Dict], filename: str): 将字典列表导出为CSV文件 if not data_list: print(数据为空不导出) return df pd.DataFrame(data_list) # 确保输出目录存在 Path(output).mkdir(exist_okTrue) filepath Path(output) / filename df.to_csv(filepath, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel中文 print(f数据已导出至: {filepath}) return filepath staticmethod def to_json(data, filename: str, indent2): 导出为JSON文件适用于嵌套结构数据 filepath Path(output) / filename with open(filepath, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indentindent) print(fJSON数据已导出至: {filepath}) staticmethod def to_sqlite(df: pd.DataFrame, table_name: str, db_pathoutput/miband_data.db): 导出到SQLite数据库便于历史数据累积和查询 import sqlite3 Path(output).mkdir(exist_okTrue) conn sqlite3.connect(db_path) try: df.to_sql(table_name, conn, if_existsappend, indexFalse) print(f数据已追加到数据库表 {table_name}) except Exception as e: print(f写入数据库失败: {e}) finally: conn.close()5. 主程序与自动化调度将各个模块组合起来形成一个完整的脚本。# main.py import sys from datetime import datetime, timedelta from src.auth import MiBandAuth from src.api_client import MiBandAPIClient from src.data_fetchers import DataFetcher from src.exporters import DataExporter def main(): 主函数获取最近7天的数据并导出 print( 小米手环数据自动化导出脚本 ) # 1. 初始化认证和客户端 auth MiBandAuth() if not auth.login(): print(登录失败程序退出) sys.exit(1) client MiBandAPIClient(auth) fetcher DataFetcher(client) # 2. 初始化设备 if not fetcher.init_device(): print(获取设备列表失败) sys.exit(1) # 3. 定义要获取的日期范围例如最近7天 end_date datetime.now().strftime(%Y-%m-%d) start_date (datetime.now() - timedelta(days6)).strftime(%Y-%m-%d) print(f获取数据范围: {start_date} 至 {end_date}) # 4. 获取每日汇总数据 print(\n 正在获取每日汇总数据...) daily_summaries fetcher.fetch_daily_data_range(start_date, end_date) if daily_summaries: DataExporter.to_csv(daily_summaries, fdaily_summary_{start_date}_to_{end_date}.csv) # 也可以导出到数据库 # df_daily pd.DataFrame(daily_summaries) # DataExporter.to_sqlite(df_daily, daily_summary) # 5. 获取某一天例如昨天的详细心率数据 print(\n 正在获取详细心率数据...) yesterday (datetime.now() - timedelta(days1)).strftime(%Y-%m-%d) hr_df fetcher.fetch_heart_rate_for_date(yesterday) if hr_df is not None and not hr_df.empty: DataExporter.to_csv(hr_df.to_dict(records), fheart_rate_detail_{yesterday}.csv) print(f心率数据示例:\n{hr_df.head()}) # 6. 获取运动记录列表例如最近30天 print(\n 正在获取运动记录列表...) sport_start (datetime.now() - timedelta(days30)).strftime(%Y-%m-%d) sport_records client.get_sport_records(sport_start, end_date, limit50) if sport_records: DataExporter.to_json(sport_records, fsport_records_{sport_start}_to_{end_date}.json) print(\n 数据导出任务完成 ) if __name__ __main__: main()5.1 实现定时自动化运行我们不想每次都手动运行脚本。在Linux/Mac上可以使用系统的cron在Windows上可以使用任务计划程序。但用Python的schedule库可以写出跨平台的定时脚本。# scheduler.py import schedule import time from main import main import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def job(): logging.info(开始执行定时数据导出任务...) try: main() logging.info(定时任务执行成功) except Exception as e: logging.error(f定时任务执行失败: {e}) if __name__ __main__: # 每天凌晨2点执行一次 schedule.every().day.at(02:00).do(job) # 也可以每6小时执行一次 # schedule.every(6).hours.do(job) logging.info(定时任务调度器已启动等待执行...) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次你可以让这个scheduler.py在后台长期运行例如使用nohup或pm2或者更专业的做法是将其配置为系统服务Systemd service。6. 避坑指南与进阶优化在实际操作中你肯定会遇到比我这里提到的更多问题。以下是我踩过的一些坑和对应的解决方案。6.1 签名算法逆向的困境这是最大的技术难点。如果完全无法逆向可以尝试以下思路寻找开源项目在GitHub上搜索mi band api、zepp api、huami等关键词有些开源项目可能已经破解了部分接口可以参考其实现。务必注意开源协议并尊重原作者的劳动成果。使用“半自动”方案如果只是偶尔导出可以结合抓包工具如mitmdump的脚本功能在手机正常使用App时自动拦截并保存特定的API响应数据到本地文件。这样无需关心签名但需要手机和电脑配合无法全自动。官方数据导出定期手动从Zepp Life App的“我的”-“数据导出”功能中申请完整数据包通常需要1-3天处理然后编写Python脚本解析下载的ZIP或CSV文件。这是最合规但最不实时的方法。6.2 Token过期与刷新登录获取的token通常有有效期如30天。我们的脚本需要处理token过期的情况。可以在api_client的_request方法中加入错误码判断如果返回token expired之类的错误则自动调用auth.login()刷新token并重试原请求。6.3 请求频率限制小米的API肯定有频率限制。我们的代码中已经加入了time.sleep(0.5)这样的礼貌延迟。在批量获取历史数据时建议将延迟设置得更大一些如1-2秒并且避免在短时间内发起大量请求否则IP或账号可能会被临时限制。6.4 数据解析与字段映射API返回的JSON字段名可能是英文缩写或拼音需要仔细对照App显示的数据进行映射。例如steps是步数dis是距离公里cal是卡路里大卡。心率数据中的hr字段代表心率值timestamp是毫秒时间戳。最好将字段映射关系写在代码注释或配置文件中。6.5 数据存储的进阶选择SQLite对于个人使用SQLite简单轻量无需安装数据库服务。可以用pandas的to_sql写入用SQL查询非常灵活。InfluxDB如果你对时间序列数据如每分钟心率有复杂的查询或可视化需求InfluxDB是专业选择。本地文件版本控制将每日导出的CSV文件用Git管理可以清晰看到历史变化。结合dvcData Version Control可以管理更大的数据文件。6.6 错误处理与日志生产环境运行的脚本必须有完善的错误处理和日志记录。我们使用了try...except捕获网络异常并使用logging模块记录信息。可以将日志同时输出到文件和控制台方便日后排查问题。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(miband_export.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__)6.7 扩展方向数据可视化使用matplotlib、plotly或seaborn库将导出的数据生成每日/每周/每月趋势图自动发送到邮箱或生成HTML报告。多平台集成将数据同步到Notion、Google Sheets或Obsidian构建个人仪表盘。健康洞察结合睡眠数据和运动数据用简单的规则或机器学习模型如scikit-learn评估睡眠质量与运动量的关系。消息通知使用requests调用钉钉、企业微信或pushover的API在数据导出完成或失败时发送通知。整个过程从逆向分析到稳定运行需要一定的耐心和调试。但一旦跑通看着自己的运动数据源源不断地流入本地数据库并能用SQL或Python进行任意分析时那种对数据的掌控感和成就感是非常棒的。这套框架的核心思路——模拟App客户端与服务器交互——同样可以借鉴到其他有App但无开放API的平台的数据获取上算是掌握了一项实用的“数字生活”技能。