项目概述
一个完整的全栈项目实战:从 iCloud 获取设备信息,存储在 MySQL 数据库,通过 RESTful API 提供数据接口,并打造美观的 Web 监控大屏。
作为 Apple 生态的重度用户,拥有 iPhone、iPad、MacBook 等多台设备。日常使用中,希望能在一个统一的界面上查看所有设备的电量、在线状态等信息。虽然 Apple 提供了"查找"应用,但本项目旨在实现以下目标:
- 私有化部署:数据存储在自己控制的服务器上
- 历史记录:可以查看设备状态的历史变化
- 自定义展示:根据需求定制化的监控界面
- 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)
""" iCloud 设备信息同步脚本
用途:定时拉取设备状态(电量等)存入 MySQL
"""
import os
import sys
import json
import logging
from datetime import datetime
ICLOUD_EMAIL = "[email protected]"
ICLOUD_PASSWORD = "your_app_specific_password"
CHINA_MAINLAND = True
MYSQL_CONFIG = {
"host": "localhost",
"port": 3306,
"user": "your_username",
"password": "your_password",
"database": "your_database",
"charset": "utf8mb4"
}
LOG_FILE = ""
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)])
logger = logging.getLogger(__name__)
def main():
try:
logger.info("▶ 开始同步 iCloud 设备信息...")
from 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()) if hasattr(sys.stdin, 'fileno') else False
if api.requires_2fa:
if not 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 in enumerate(devices, start=1):
logger.info(f" {idx}: {dev}")
choice = input("请选择 FIDO2 设备编号(直接回车使用第一个):")
if not choice:
choice = 1
else:
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}")
if not result:
logger.error("验证码验证失败")
sys.exit(1)
if not api.is_trusted_session:
logger.info("会话未被信任。正在请求信任...")
result = api.trust_session()
logger.info(f"会话信任结果:{result}")
elif api.requires_2sa:
if not IS_INTERACTIVE:
logger.error("❌ 需要两步认证(2SA),但当前运行在非交互模式下")
sys.exit(2)
logger.info("需要进行两步认证(2SA)")
logger.info("你的可信任设备:")
devices = api.trusted_devices
for i, device in enumerate(devices):
device_name = device.get('deviceName', f"SMS to {device.get('phoneNumber','未知')}")
logger.info(f" {i}: {device_name}")
device_choice = input('请选择要使用的设备编号(直接回车使用第一个):')
if not device_choice:
device_choice = 0
else:
device_choice = int(device_choice)
device = devices[device_choice]
if not api.send_verification_code(device):
logger.error("发送验证码失败")
sys.exit(1)
code = input('请输入验证码:')
if not api.validate_verification_code(device, code):
logger.error("验证码验证失败")
sys.exit(1)
logger.info("✓ 认证完成")
import mysql.connector
db = mysql.connector.connect(**MYSQL_CONFIG)
cursor = db.cursor()
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;
""")
try:
devices = api.devices
except PyiCloudServiceUnavailable as e:
logger.error(f"❌ Find My iPhone 服务不可用:{e}")
sys.exit(1)
if not devices:
logger.warning("⚠ 未找到任何设备")
return
count = 0
for 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') or str(dev.id) if hasattr(dev, 'id') else str(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')
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')
if not location and hasattr(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'
if isinstance(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 else None
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:
pass
if __name__ == "__main__":
main()
2. Node.js API 服务
安装依赖
npm install 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 = new Koa();
const router = new Router();
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(, );
ctx.(, );
ctx.(, );
(ctx. === ) {
ctx. = ;
;
}
();
});
app.(());
pool = ;
() {
(!pool) {
pool = mysql.({ ...dbConfig, : , : , : });
}
pool;
}
router.(, (ctx) => {
{
pool = ();
[rows] = pool.();
devices = rows.( {
parsedData = {};
{
(device.) {
parsedData = device. === ? .(device.) : device.;
}
} (e) {
.(, e);
}
parsedLocation = ;
{
(device.) {
parsedLocation = device. === ? .(device.) : device.;
}
} (e) {
.(, e);
}
{ ...device, : parsedData, : parsedLocation };
});
ctx. = { : , : , : devices, : devices., : ().() };
} (error) {
ctx. = ;
ctx. = { : , : , : error. };
}
});
router.(, (ctx) => {
{
pool = ();
[rows] = pool.();
ctx. = { : , : , : rows, : rows. };
} (error) {
ctx. = ;
ctx. = { : , : , : error. };
}
});
router.(, (ctx) => {
{
pool = ();
[rows] = pool.();
ctx. = { : , : , : rows, : rows. };
} (error) {
ctx. = ;
ctx. = { : , : , : error. };
}
});
router.(, (ctx) => {
{
pool = ();
[rows] = pool.();
ctx. = { : , : , : rows, : rows. };
} (error) {
ctx. = ;
ctx. = { : , : , : error. };
}
});
router.(, (ctx) => {
{
pool = ();
[totalRows] = pool.();
[onlineRows] = pool.();
[lowBatteryRows] = pool.();
[avgBatteryRows] = pool.();
[typeRows] = pool.();
ctx. = {
: , : , : {
: totalRows[].,
: onlineRows[].,
: totalRows[]. - onlineRows[].,
: lowBatteryRows[].,
: avgBatteryRows[]. ? (avgBatteryRows[].).() : ,
: typeRows
}
};
} (error) {
ctx. = ;
ctx. = { : , : , : error. };
}
});
router.(, (ctx) => {
{
pool = ();
pool.();
ctx. = { : , : , : ().() };
} (error) {
ctx. = ;
ctx. = { : , : , : error. };
}
});
app.(router.()).(router.());
= process.. || ;
app.(, {
.();
});
3. Web 前端页面
由于 HTML 代码较长,这里提供关键部分。完整代码请根据以下结构自行实现:
核心功能:
- 设备卡片渲染:根据设备类型(手机/平板/电脑)渲染不同的卡片样式
- 电量状态判断:自动识别电量格式(小数/百分比),正确显示状态
- 自动刷新:每 30 秒自动更新设备数据
- 响应式布局:使用 Tailwind CSS 实现美观的界面
关键 JavaScript 函数:
const API_BASE_URL = 'http://your-api-server:3000';
function normalizeBatteryLevel(batteryLevel) {
if (batteryLevel === null || batteryLevel === undefined) return null;
if (batteryLevel > 1) {
return batteryLevel / 100;
}
return batteryLevel;
}
function getBatteryColor(batteryLevel) {
const normalized = normalizeBatteryLevel(batteryLevel);
if (normalized === null) return 'bg-gray-400';
if (normalized >= 0.8) return 'bg-green-500';
if (normalized >= 0.3) return 'bg-orange-500';
return 'bg-red-500';
}
function getStatusBadge(device) {
const batteryLevel = device.battery_level;
const isOnline = device.is_online;
(!isOnline) {
;
}
(batteryLevel === || batteryLevel === ) {
;
}
normalized = (batteryLevel);
(normalized >= ) {
;
} (normalized >= ) {
;
} {
;
}
}
() {
{
response = ();
result = response.();
(result. === && result.) {
devices = result.;
phoneDevices = devices.( d.?.().() || d.?.().());
tabletDevices = devices.( d.?.().() || d.?.().());
computerDevices = devices.( d.?.().() || d.?.().());
.(). = phoneDevices.( (device)).();
(devices);
}
} (error) {
.(, error);
}
}
.(, {
();
(loadDevices, );
});
完整 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
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. 首次运行
python3 update_devices.py
cd api
npm install
npm start
open index.html
2. 设置定时任务
macOS - launchd:
launchctl load ~/Library/LaunchAgents/com.icloud.devices.update.plist
Linux - Crontab:
crontab -e
*/5 * * * * cd /path/to/project && /usr/bin/python3 update_devices.py >> logs/cron.log 2>&1
参考资源