用 Python + MySQL + Web 打造我的私有 Apple 设备监控面板

优质文章学习记录

— 12 min read
用 Python + MySQL + Web 打造我的私有 Apple 设备监控面板
一个完整的全栈项目实战：从 iCloud 获取设备信息，存储在 MySQL 数据库，通过 RESTful API 提供数据接口，并打造美观的 Web 监控大屏。

作为 Apple 生态的重度用户，我拥有 iPhone、iPad、MacBook 等多台设备。日常使用中，我希望能在一个统一的界面上查看所有设备的电量、在线状态等信息。虽然 Apple 提供了"查找"应用，但我想要：

  1. 私有化部署：数据存储在自己控制的服务器上
  2. 历史记录：可以查看设备状态的历史变化
  3. 自定义展示：根据需求定制化的监控界面
  4. API 接口：方便与其他系统集成

于是，我决定自己动手搭建一个完整的设备监控系统。
仓库地址：
仓库地址
效果预览

项目功能

  • ✅ 自动从 iCloud 获取设备信息（电量、状态、位置等）
  • ✅ 数据持久化存储到 MySQL 数据库
  • ✅ RESTful API 提供灵活的数据查询接口
  • ✅ 美观的 Web 监控大屏，支持实时刷新
  • ✅ 按设备类型分类展示（手机、平板、电脑）
  • ✅ 设备统计信息（总数、在线数、平均电量等）
  • ✅ 定时任务自动更新设备状态

技术栈

  • Python 3 + pyicloud - 数据采集
  • Node.js + Koa2 - API 服务
  • MySQL - 数据存储
  • HTML5 + Tailwind CSS + Vanilla JavaScript - 前端展示

架构设计

iCloud API → Python脚本 → MySQL数据库 → Koa2 API → Web前端

完整代码实现

1. Python 数据采集脚本

安装依赖
pip3 install pyicloud mysql-connector-python
完整代码（update_devices.py）
#!/usr/bin/env python3# -*- coding: utf-8 -*-""" iCloud 设备信息同步脚本 用途：定时拉取设备状态（电量等）存入 MySQL """import os import sys import json import logging from datetime import datetime # === 配置区（请按你的情况修改）=== ICLOUD_EMAIL ="[email protected]"# ← 改成你的 Apple ID ICLOUD_PASSWORD ="your_app_specific_password"# ← 强烈建议用「专用密码」！ CHINA_MAINLAND =True# 如果是中国大陆账户，设为 True；否则设为 False MYSQL_CONFIG ={"host":"localhost",# 数据库主机"port":3306,"user":"your_username","password":"your_password","database":"your_database","charset":"utf8mb4"} LOG_FILE =""# 留空则只输出到 stdout# === 初始化日志 ===if LOG_FILE: log_dir = os.path.dirname(LOG_FILE)if log_dir:try: os.makedirs(log_dir, exist_ok=True)except(OSError, PermissionError)as e:print(f"警告: 无法创建日志目录 {log_dir}: {e}") LOG_FILE =""if LOG_FILE: logging.basicConfig( level=logging.INFO,format='%(asctime)s | %(levelname)s | %(message)s', handlers=[ logging.FileHandler(LOG_FILE, encoding='utf-8'), logging.StreamHandler(sys.stdout)])else: logging.basicConfig( level=logging.INFO,format='%(asctime)s | %(levelname)s | %(message)s', handlers=[logging.StreamHandler(sys.stdout)])else: logging.basicConfig( level=logging.INFO,format='%(asctime)s | %(levelname)s | %(message)s', handlers=[logging.StreamHandler(sys.stdout)]) logger = logging.getLogger(__name__)# === 主逻辑 ===defmain():try: logger.info("▶ 开始同步 iCloud 设备信息...")# 1. 登录 iCloudfrom pyicloud import PyiCloudService from pyicloud.exceptions import PyiCloudFailedLoginException, PyiCloudServiceUnavailable try: api = PyiCloudService(ICLOUD_EMAIL, ICLOUD_PASSWORD, china_mainland=CHINA_MAINLAND)except PyiCloudFailedLoginException as e: logger.error(f"❌ 登录失败: {e}") sys.exit(1)# 检查是否需要二次验证 IS_INTERACTIVE = os.isatty(sys.stdin.fileno())ifhasattr(sys.stdin,'fileno')elseFalseif api.requires_2fa:ifnot IS_INTERACTIVE: logger.error("❌ 需要双重认证（2FA），但当前运行在非交互模式下") logger.error("请先手动运行一次脚本来完成认证") sys.exit(2) logger.info("需要进行两步验证（2FA）") security_key_names = api.security_key_names if security_key_names: logger.info(f"需要安全密钥确认。请插入以下密钥之一: {', '.join(security_key_names)}") devices = api.fido2_devices logger.info("可用的 FIDO2 设备:")for idx, dev inenumerate(devices, start=1): logger.info(f" {idx}: {dev}") choice =input("请选择 FIDO2 设备编号（直接回车使用第一个）: ")ifnot choice: choice =1else: choice =int(choice) selected_device = devices[choice -1] logger.info("请使用安全密钥确认操作") api.confirm_security_key(selected_device)else: logger.info("验证码已发送到你已批准的设备上。") code =input("请输入收到的验证码: ") result = api.validate_2fa_code(code) logger.info(f"验证码验证结果: {result}")ifnot result: logger.error("验证码验证失败") sys.exit(1)# 验证成功后，检查会话是否被信任ifnot api.is_trusted_session: logger.info("会话未被信任。正在请求信任...") result = api.trust_session() logger.info(f"会话信任结果: {result}")elif api.requires_2sa:ifnot IS_INTERACTIVE: logger.error("❌ 需要两步认证（2SA），但当前运行在非交互模式下") sys.exit(2) logger.info("需要进行两步认证（2SA）") logger.info("你的可信任设备:") devices = api.trusted_devices for i, device inenumerate(devices): device_name = device.get('deviceName',f"SMS to {device.get('phoneNumber','未知')}") logger.info(f" {i}: {device_name}") device_choice =input('请选择要使用的设备编号（直接回车使用第一个）: ')ifnot device_choice: device_choice =0else: device_choice =int(device_choice) device = devices[device_choice]ifnot api.send_verification_code(device): logger.error("发送验证码失败") sys.exit(1) code =input('请输入验证码: ')ifnot api.validate_verification_code(device, code): logger.error("验证码验证失败") sys.exit(1) logger.info("✓ 认证完成")# 2. 连接 MySQLimport mysql.connector db = mysql.connector.connect(**MYSQL_CONFIG) cursor = db.cursor()# 3. 确保表存在 cursor.execute(""" CREATE TABLE IF NOT EXISTS `devices` ( `id` INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, `device_id` VARCHAR(100) NOT NULL UNIQUE, `name` VARCHAR(100), `model` VARCHAR(60), `device_class` VARCHAR(20), `battery_level` DECIMAL(3,2), `battery_status` VARCHAR(20), `os_version` VARCHAR(30), `last_location` TEXT, `last_seen` DATETIME, `is_online` BOOLEAN, `raw_data` JSON, `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """)# 4. 获取设备列表try: devices = api.devices except PyiCloudServiceUnavailable as e: logger.error(f"❌ Find My iPhone 服务不可用: {e}") sys.exit(1)ifnot devices: logger.warning("⚠ 未找到任何设备")return# 5. 遍历设备并保存 count =0for dev in devices:try:# 请求完整的设备属性列表 requested_properties =['deviceDisplayName','name','deviceStatus','batteryLevel','batteryStatus','deviceModel','model','modelDisplayName','deviceClass','deviceClassDisplay','deviceType','osVersion','deviceOSType','osVersionDisplay','systemVersion','serialNumber','id','deviceId','location','deviceStatusTime','timestamp','locationTimeStamp','timeStamp']try: status = dev.status(requested_properties)except TypeError: status = dev.status()# 提取设备信息 device_id =(status.get('serialNumber')or status.get('id')or status.get('deviceId')or status.get('deviceDisplayName')orstr(dev.id)ifhasattr(dev,'id')elsestr(dev)) name = status.get('deviceDisplayName')or status.get('name','Unknown') model = status.get('deviceModel')or status.get('model','Unknown') device_class = status.get('deviceClass')or status.get('deviceClassDisplay','Unknown') battery_level = status.get('batteryLevel')# 0-1 之间的小数 battery_status = status.get('batteryStatus','Unknown') os_version =(status.get('osVersion')or status.get('deviceOSType')or status.get('osVersionDisplay')or'Unknown') location = status.get('location')ifnot location andhasattr(dev,'location'):try: location_data = dev.location()if location_data: location = location_data except Exception:pass last_seen =(status.get('deviceStatusTime')or status.get('timestamp')or status.get('locationTimeStamp')) is_online = status.get('deviceStatus')=='200'# 时间转换ifisinstance(last_seen,(int,float)): last_seen = datetime.fromtimestamp(last_seen /1000.0)else: last_seen =None loc_str = json.dumps(location, ensure_ascii=False)if location elseNone# 插入/更新数据库 sql =""" INSERT INTO `devices` ( device_id, name, model, device_class, battery_level, battery_status, os_version, last_location, last_seen, is_online, raw_data ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE name = VALUES(name), model = VALUES(model), battery_level = VALUES(battery_level), battery_status = VALUES(battery_status), os_version = VALUES(os_version), last_location = VALUES(last_location), last_seen = VALUES(last_seen), is_online = VALUES(is_online), raw_data = VALUES(raw_data), updated_at = NOW() """ cursor.execute(sql,( device_id, name, model, device_class, battery_level, battery_status, os_version, loc_str, last_seen, is_online, json.dumps(status, ensure_ascii=False))) count +=1 logger.info(f"✓ 已同步: {name} | 电量: {battery_level or'N/A'}")except Exception as e: logger.error(f"⚠ 设备同步失败 ({dev}): {e}", exc_info=True) db.commit() logger.info(f"✅ 同步完成！共 {count} 台设备")except Exception as e: logger.error(f"❌ 全局错误: {e}", exc_info=True) sys.exit(1)finally:try: cursor.close() db.close()except:passif __name__ =="__main__": main()

2. Node.js API 服务

安装依赖
npminstall koa koa-router koa-bodyparser mysql2 dotenv
完整代码（app.js）
const Koa =require('koa');const Router =require('koa-router');const bodyParser =require('koa-bodyparser');require('dotenv').config();const app =newKoa();const router =newRouter();// 数据库配置const dbConfig ={host: process.env.DB_HOST||'localhost',port: process.env.DB_PORT||3306,user: process.env.DB_USER||'root',password: process.env.DB_PASSWORD||'',database: process.env.DB_NAME||'devices',charset:'utf8mb4'};const mysql =require('mysql2/promise');// 跨域中间件 app.use(async(ctx, next)=>{ ctx.set('Access-Control-Allow-Origin','*'); ctx.set('Access-Control-Allow-Methods','GET, POST, PUT, DELETE, OPTIONS'); ctx.set('Access-Control-Allow-Headers','Content-Type, Authorization, X-Requested-With'); ctx.set('Access-Control-Allow-Credentials','true');if(ctx.method ==='OPTIONS'){ ctx.status =204;return;}awaitnext();}); app.use(bodyParser());// 数据库连接池let pool =null;asyncfunctiongetPool(){if(!pool){ pool = mysql.createPool({...dbConfig,waitForConnections:true,connectionLimit:10,queueLimit:0});}return pool;}// 查询所有设备（完整信息） router.get('/api/devices/all',async(ctx)=>{try{const pool =awaitgetPool();const[rows]=await pool.execute(` SELECT id, device_id, name, model, device_class, battery_level, battery_status, os_version, last_location, last_seen, is_online, raw_data, updated_at FROM devices ORDER BY updated_at DESC `);// 解析 JSON 字段const devices = rows.map(device=>{let parsedData ={};try{if(device.raw_data){ parsedData =typeof device.raw_data ==='string'?JSON.parse(device.raw_data): device.raw_data;}}catch(e){ console.error('解析 raw_data 失败:', e);}let parsedLocation =null;try{if(device.last_location){ parsedLocation =typeof device.last_location ==='string'?JSON.parse(device.last_location): device.last_location;}}catch(e){ console.error('解析 last_location 失败:', e);}return{...device,raw_data: parsedData,last_location: parsedLocation };}); ctx.body ={code:200,message:'success',data: devices,total: devices.length,timestamp:newDate().toISOString()};}catch(error){ ctx.status =500; ctx.body ={code:500,message:'查询失败',error: error.message };}});// 查询所有设备（简化信息） router.get('/api/devices',async(ctx)=>{try{const pool =awaitgetPool();const[rows]=await pool.execute(` SELECT id, device_id, name, model, device_class, battery_level, battery_status, os_version, last_location, last_seen, is_online, updated_at FROM devices ORDER BY updated_at DESC `); ctx.body ={code:200,message:'success',data: rows,total: rows.length };}catch(error){ ctx.status =500; ctx.body ={code:500,message:'查询失败',error: error.message };}});// 查询在线设备 router.get('/api/devices/online',async(ctx)=>{try{const pool =awaitgetPool();const[rows]=await pool.execute(` SELECT * FROM devices WHERE is_online = 1 ORDER BY updated_at DESC `); ctx.body ={code:200,message:'success',data: rows,total: rows.length };}catch(error){ ctx.status =500; ctx.body ={code:500,message:'查询失败',error: error.message };}});// 查询低电量设备（< 30%） router.get('/api/devices/low-battery',async(ctx)=>{try{const pool =awaitgetPool();const[rows]=await pool.execute(` SELECT * FROM devices WHERE battery_level < 0.3 AND battery_level IS NOT NULL ORDER BY battery_level ASC `); ctx.body ={code:200,message:'success',data: rows,total: rows.length };}catch(error){ ctx.status =500; ctx.body ={code:500,message:'查询失败',error: error.message };}});// 设备统计信息 router.get('/api/devices/stats',async(ctx)=>{try{const pool =awaitgetPool();const[totalRows]=await pool.execute('SELECT COUNT(*) as total FROM devices');const[onlineRows]=await pool.execute('SELECT COUNT(*) as count FROM devices WHERE is_online = 1');const[lowBatteryRows]=await pool.execute('SELECT COUNT(*) as count FROM devices WHERE battery_level < 0.3 AND battery_level IS NOT NULL');const[avgBatteryRows]=await pool.execute('SELECT AVG(battery_level) as avg FROM devices WHERE battery_level IS NOT NULL');const[typeRows]=await pool.execute(` SELECT device_class, COUNT(*) as count FROM devices WHERE device_class IS NOT NULL GROUP BY device_class `); ctx.body ={code:200,message:'success',data:{total: totalRows[0].total,online: onlineRows[0].count,offline: totalRows[0].total - onlineRows[0].count,lowBattery: lowBatteryRows[0].count,avgBattery: avgBatteryRows[0].avg ?parseFloat(avgBatteryRows[0].avg).toFixed(2):null,byType: typeRows }};}catch(error){ ctx.status =500; ctx.body ={code:500,message:'查询失败',error: error.message };}});// 健康检查 router.get('/api/health',async(ctx)=>{try{const pool =awaitgetPool();await pool.execute('SELECT 1'); ctx.body ={code:200,message:'服务正常',timestamp:newDate().toISOString()};}catch(error){ ctx.status =500; ctx.body ={code:500,message:'数据库连接失败',error: error.message };}}); app.use(router.routes()).use(router.allowedMethods());constPORT= process.env.PORT||3000; app.listen(PORT,()=>{ console.log(`🚀 服务器运行在 http://localhost:${PORT}`);});

3. Web 前端页面

完整代码（index.html）

由于 HTML 代码较长，这里提供关键部分。完整代码请查看项目仓库或根据以下结构自行实现：

核心功能：

  1. 设备卡片渲染：根据设备类型（手机/平板/电脑）渲染不同的卡片样式
  2. 电量状态判断：自动识别电量格式（小数/百分比），正确显示状态
  3. 自动刷新：每30秒自动更新设备数据
  4. 响应式布局：使用 Tailwind CSS 实现美观的界面

关键 JavaScript 函数：

constAPI_BASE_URL='http://your-api-server:3000';// 标准化电池电量值（统一转换为 0-1 之间的小数）functionnormalizeBatteryLevel(batteryLevel){if(batteryLevel ===null|| batteryLevel ===undefined)returnnull;if(batteryLevel >1){return batteryLevel /100;// 百分比格式转小数}return batteryLevel;// 已经是小数格式}// 获取电池颜色functiongetBatteryColor(batteryLevel){const normalized =normalizeBatteryLevel(batteryLevel);if(normalized ===null)return'bg-gray-400';if(normalized >=0.8)return'bg-green-500';// ≥80%if(normalized >=0.3)return'bg-orange-500';// 30%-79%return'bg-red-500';// <30%}// 获取状态标签functiongetStatusBadge(device){const batteryLevel = device.battery_level;const isOnline = device.is_online;if(!isOnline){return'<span>离线</span>';}if(batteryLevel ===null|| batteryLevel ===undefined){return'<span>在线</span>';}const normalized =normalizeBatteryLevel(batteryLevel);if(normalized >=0.8){return'<span>电量充足</span>';}elseif(normalized >=0.3){return'<span>电量中等</span>';}else{return'<span>电量不足</span>';}}// 加载设备数据asyncfunctionloadDevices(){try{const response =awaitfetch(`${API_BASE_URL}/api/devices/all`);const result =await response.json();if(result.code ===200&& result.data){const devices = result.data;// 按设备类型分类const phoneDevices = devices.filter(d=> d.device_class?.toLowerCase().includes('iphone')|| d.name?.toLowerCase().includes('iphone'));const tabletDevices = devices.filter(d=> d.device_class?.toLowerCase().includes('ipad')|| d.name?.toLowerCase().includes('ipad'));const computerDevices = devices.filter(d=> d.device_class?.toLowerCase().includes('mac')|| d.name?.toLowerCase().includes('mac'));// 渲染到对应区域 document.getElementById('phoneDevices').innerHTML = phoneDevices.map(device=>renderDeviceCard(device)).join('');// ... 其他设备类型// 更新统计信息updateStats(devices);}}catch(error){ console.error('加载设备失败:', error);}}// 页面加载时自动加载，每30秒自动刷新 document.addEventListener('DOMContentLoaded',()=>{loadDevices();setInterval(loadDevices,30000);});

完整 HTML 代码请参考项目中的 web/index.html 文件。

配置说明

环境变量配置

创建 .env 文件：

DB_HOST=localhost DB_PORT=3306 DB_USER=your_username DB_PASSWORD=your_password DB_NAME=your_database PORT=3000

Python 脚本配置

修改 update_devices.py 中的配置：

ICLOUD_EMAIL ="[email protected]" ICLOUD_PASSWORD ="your_app_specific_password"# 建议使用专用密码 CHINA_MAINLAND =True# 中国大陆账户设为 True MYSQL_CONFIG ={"host":"localhost","port":3306,"user":"your_username","password":"your_password","database":"your_database","charset":"utf8mb4"}

关键要点

1. batteryLevel 格式

根据 pyicloud 文档，batteryLevel 是 0-1 之间的小数（如 0.85 表示 85%）。前端代码已兼容两种格式（小数和百分比），会自动识别并转换。

2. 设备状态码

  • deviceStatus = "200" - 设备在线
  • deviceStatus = "201" - 设备离线

3. 离线设备限制

离线时 iCloud API 只返回基本信息（名称、电量），无法获取型号、系统版本等详细信息。这是 iCloud API 的限制，不是代码问题。

4. 会话管理

认证后的会话保存在 .pyicloud/ 目录，有效期约 2 个月。过期后需要重新手动运行脚本完成认证。

使用方法

1. 首次运行

# 运行 Python 脚本（会提示输入验证码） python3 update_devices.py # 启动 API 服务cd api npminstallnpm start # 在浏览器打开前端页面open index.html

2. 设置定时任务

macOS - launchd：

# 创建 plist 文件（每5分钟执行一次） launchctl load ~/Library/LaunchAgents/com.icloud.devices.update.plist

Linux - Crontab：

# 编辑 crontabcrontab -e # 添加任务（每5分钟执行一次） */5 * * * * cd /path/to/project && /usr/bin/python3 update_devices.py >> logs/cron.log 2>&1

参考资源

Could not load content