一个全栈项目，通过 Python 脚本从 iCloud 获取 Apple 设备信息（电量、状态等），存入 MySQL 数据库。后端使用 Node.js (Koa2) 提供 RESTful API，前端采用 HTML5 + Tailwind CSS 构建监控大屏。支持实时刷新、设备分类展示及统计信息。实现了私有化部署、历史记录查看及自定义界面，解决了原生“查找”应用功能局限的问题。包含详细代码实现、环境配置及定时任务设置指南。
一个完整的全栈项目实战：从 iCloud 获取设备信息，存储在 MySQL 数据库，通过 RESTful API 提供数据接口，并打造美观的 Web 监控大屏。
作为 Apple 生态的重度用户，拥有 iPhone、iPad、MacBook 等多台设备。日常使用中，希望能在一个统一的界面上查看所有设备的电量、在线状态等信息。虽然 Apple 提供了"查找"应用，但本项目旨在实现以下目标：
于是，决定自己动手搭建一个完整的设备监控系统。
iCloud API → Python 脚本 → MySQL 数据库 → Koa2 API → Web 前端
pip3 install pyicloud mysql-connector-python
#!/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)])
logger = logging.getLogger(__name__)
# === 主逻辑 ===
def main():
try:
logger.info("▶ 开始同步 iCloud 设备信息...")
# 1. 登录 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("✓ 认证完成")
# 2. 连接 MySQL
import 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)
if not devices:
logger.warning("⚠ 未找到任何设备")
return
# 5. 遍历设备并保存
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') # 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')
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()
npm install koa koa-router koa-bodyparser mysql2 dotenv
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.(, {
.();
});
由于 HTML 代码较长，这里提供关键部分。完整代码请根据以下结构自行实现：
核心功能：
关键 JavaScript 函数：
const API_BASE_URL = 'http://your-api-server:3000';
// 标准化电池电量值（统一转换为 0-1 之间的小数）
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'; // ≥80%
if (normalized >= 0.3) return 'bg-orange-500'; // 30%-79%
return 'bg-red-500'; // <30%
}
// 获取状态标签
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
修改 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"
}
根据 pyicloud 文档，batteryLevel 是 0-1 之间的小数（如 0.85 表示 85%）。前端代码已兼容两种格式（小数和百分比），会自动识别并转换。
deviceStatus = "200" - 设备在线deviceStatus = "201" - 设备离线离线时 iCloud API 只返回基本信息（名称、电量），无法获取型号、系统版本等详细信息。这是 iCloud API 的限制，不是代码问题。
认证后的会话保存在 .pyicloud/ 目录，有效期约 2 个月。过期后需要重新手动运行脚本完成认证。
# 运行 Python 脚本（会提示输入验证码）
python3 update_devices.py
# 启动 API 服务
cd api
npm install
npm start
# 在浏览器打开前端页面
open index.html
macOS - launchd：
# 创建 plist 文件（每 5 分钟执行一次）
launchctl load ~/Library/LaunchAgents/com.icloud.devices.update.plist
Linux - Crontab：
# 编辑 crontab
crontab -e
# 添加任务（每 5 分钟执行一次）
*/5 * * * * cd /path/to/project && /usr/bin/python3 update_devices.py >> logs/cron.log 2>&1
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML 转 Markdown 互为补充。 在线工具，Markdown 转 HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML 转 Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online