一、研究背景与技术意义
iMessage 作为苹果生态内置的原生通讯服务，依托系统底层优势，具备端到端加密、无运营商拦截、原生展示等特性，常用于企业内部事务提醒、授权用户服务告知等合规场景。在技术研究过程中，手动单条发送消息效率低下，难以满足批量测试与自动化学习需求。
基于此，本文设计一套基于 macOS 虚拟机的分布式技术方案，通过虚拟化实现多环境隔离，搭配 Python 脚本完成自动化调度，既解决了批量处理的技术痛点，又为开发者提供了 macOS 原生框架调用、分布式任务调度的实践案例，仅作技术学习与合规场景研究使用。
二、整体技术架构
本方案采用「本地主控 + 虚拟机节点」的轻量化架构，核心分为 4 个模块，结构清晰且易于部署：
表格
| 模块 | 功能说明 |
|---|---|
| 虚拟机节点层 | 多台独立 macOS 虚拟机，每台登录合法 Apple ID 并启用 iMessage，作为消息执行单元 |
| 接口调用层 | Python 结合 pyobjc 库，调用 macOS 原生 Messages 框架，实现 iMessage 消息提交 |
| 本地调度层 | 主机端脚本负责读取目标列表、分配任务、控制执行间隔、失败重试与结果统计 |
| 数据统计层 | 自动生成 Excel 统计报表，记录每条消息的执行状态、重试次数及失败原因 |
三、运行环境准备
3.1 软硬件环境要求
表格
| 环境类型 | 具体要求 |
|---|---|
| 物理主机 | Windows 10+/macOS 12+，内存 ≥ 16G（支持多虚拟机同时运行） |
| 虚拟化工具 | VMware Workstation 17+ / Parallels Desktop 18+（支持 macOS 虚拟化） |
| 虚拟机系统 | macOS 12.0+ 官方镜像（确保 iMessage 功能正常） |
| 账号要求 | 每台虚拟机配置独立、合法注册的 Apple ID（已开启双重认证，iMessage 可正常使用） |
| 网络环境 | 主机与虚拟机处于同一局域网，关闭防火墙拦截（确保节点通信正常） |
| 开发环境 | Python 3.8+（虚拟机与主机均需安装） |
3.2 依赖库安装
虚拟机节点依赖（每台均需执行）
bash
运行
# 安装 macOS 原生框架调用依赖 pip3 install pyobjc -i https://pypi.tuna.tsinghua.edu.cn/simple主机主控依赖
bash
运行
# 安装框架调用、进度展示、Excel 导出依赖 pip3 install pyobjc tqdm openpyxl -i https://pypi.tuna.tsinghua.edu.cn/simple四、核心代码实现
4.1 虚拟机节点脚本（im_node.py）
运行于 macOS 虚拟机内，负责监听主机指令，调用原生框架执行消息发送，代码无冗余、可直接运行：
python
运行
import objc import socket from Foundation import NSURL from Messages import * def load_messages_framework(): """加载 macOS 原生 Messages 框架，为 iMessage 调用提供基础""" try: objc.loadBundle( "Messages", bundle_path="/System/Library/Frameworks/Messages.framework", module_globals=globals() ) return True except Exception as e: print(f"框架加载失败：{str(e)}") return False def send_imessage(phone: str, content: str): """调用 iMessage 原生接口，执行单条消息发送""" # 号码格式校验（必须带国际区号，如 +86） if not phone.startswith("+"): return False, "号码格式错误，需携带国际区号（示例：+8613800138000）" # 校验框架是否加载成功 if not load_messages_framework(): return False, "Messages 框架加载异常，无法执行发送" try: # 构建苹果原生收件人地址 recipient_url = NSURL.URLWithString_(f"tel:{phone}") message_request = MSMessageRequest.alloc().init() message_request.setRecipients_([recipient_url]) message_request.setMessageText_(content) # 同步执行消息发送，对接 iMessage 底层服务 message_request.sendSynchronouslyWithError_(None) return True, "消息执行提交完成" except Exception as e: return False, f"执行异常：{str(e)}" def start_node_server(port: int = 8888): """启动节点服务，监听主机调度指令""" if not load_messages_framework(): print("框架初始化失败，节点服务无法启动") return # 初始化 Socket 服务，支持端口复用 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind(("0.0.0.0", port)) server_socket.listen(5) print(f"虚拟机节点服务已启动，监听端口：{port}，等待调度指令...") # 持续监听主机请求 while True: try: conn, addr = server_socket.accept() print(f"收到来自 {addr} 的任务请求") # 解析任务指令（格式：手机号|消息内容） task_data = conn.recv(1024).decode("utf-8").strip() if "|" not in task_data: conn.send(b"False|指令格式错误，正确格式：手机号|消息内容") conn.close() continue target_phone, message_content = task_data.split("|", 1) send_status, send_msg = send_imessage(target_phone, message_content) # 向主机返回执行结果 conn.send(f"{send_status}|{send_msg}".encode("utf-8")) conn.close() print(f"任务执行结果：{send_msg}") except Exception as e: print(f"节点服务异常：{str(e)}") continue if __name__ == "__main__": # 启动节点服务（默认端口 8888，可自定义修改） start_node_server(port=8888)虚拟机节点启动命令
bash
运行
cd ~/Desktop # 进入脚本所在目录（根据实际路径修改） python3 im_node.py4.2 主机主控调度脚本（main_scheduler.py）
运行于物理主机，负责任务分配、间隔控制、失败重试与结果统计，支持多节点负载均衡：
python
运行
import socket import time import os from tqdm import tqdm from openpyxl import Workbook # ===================== 基础配置（根据实际场景修改）===================== # 虚拟机节点列表（IP + 端口，可添加多个节点实现分布式） VM_NODES = [ ("192.168.1.103", 8888), # 虚拟机 1 IP 与端口 # ("192.168.1.104", 8888), # 虚拟机 2 IP 与端口（按需扩展） ] # 目标号码文件路径（每行 1 个号码，带国际区号） TARGET_FILE = "target_list.txt" # 消息内容（仅用于内部测试/合规提醒，禁止违规内容） MESSAGE_CONTENT = "【技术测试通知】本消息为本地技术学习测试，请勿回复，感谢配合。" # 单条消息执行间隔（秒，建议 ≥20 秒，避免高频操作） EXECUTE_INTERVAL = 25 # 失败最大重试次数 MAX_RETRY = 2 # 重试间隔（秒） RETRY_INTERVAL = 60 # 结果统计 Excel 导出路径 RESULT_EXCEL = "imessage_task_result.xlsx" # ===================================================================== def read_target_numbers(): """读取目标号码文件，过滤有效号码（仅保留带国际区号的格式）""" if not os.path.exists(TARGET_FILE): print(f"错误：目标号码文件 {TARGET_FILE} 不存在，请先创建") return [] with open(TARGET_FILE, "r", encoding="utf-8") as f: raw_lines = [line.strip() for line in f if line.strip()] # 过滤有效号码（必须以 + 开头，避免格式错误） valid_numbers = [num for num in raw_lines if num.startswith("+")] print(f"读取完成：共 {len(raw_lines)} 条记录，有效号码 {len(valid_numbers)} 条（已自动去重）") return list(set(valid_numbers)) # 自动去重 def send_task_to_node(node_ip: str, node_port: int, phone: str, content: str): """向指定虚拟机节点发送任务，返回执行结果""" try: # 建立 Socket 连接（超时时间 10 秒） client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.settimeout(10) client_socket.connect((node_ip, node_port)) # 发送任务指令 task_cmd = f"{phone}|{content}".encode("utf-8") client_socket.send(task_cmd) # 接收执行结果 result = client_socket.recv(1024).decode("utf-8").strip() client_socket.close() return result.startswith("True"), result.split("|", 1)[1] if "|" in result else result except socket.timeout: return False, "连接超时，虚拟机节点未响应" except ConnectionRefusedError: return False, "连接被拒绝，虚拟机节点服务未启动" except Exception as e: return False, f"通信异常：{str(e)}" def export_result_to_excel(result_data: dict): """将执行结果导出为 Excel 报表，便于统计复盘""" wb = Workbook() ws = wb.active ws.title = "任务执行结果" # 写入表头并设置样式 headers = ["序号", "目标号码", "执行状态", "重试次数", "结果说明", "消息内容"] ws.append(headers) for col in range(1, len(headers) + 1): ws.cell(row=1, column=col).font = Font(bold=True) ws.cell(row=1, column=col).alignment = Alignment(horizontal="center") # 写入执行明细 row_num = 2 for phone, data in result_data.items(): status = "成功" if data["success"] else "失败" ws.append([ row_num - 1, phone, status, data["retry_count"], data["message"], MESSAGE_CONTENT ]) row_num += 1 # 调整列宽 column_widths = [8, 20, 12, 12, 40, 60] for col, width in enumerate(column_widths, 1): ws.column_dimensions[chr(64 + col)].width = width wb.save(RESULT_EXCEL) print(f"\n统计报表已导出至：{os.path.abspath(RESULT_EXCEL)}") def main(): """主控调度主逻辑""" # 读取有效目标号码 target_numbers = read_target_numbers() if not target_numbers: print("无有效目标号码，程序退出") return # 初始化结果存储字典 task_result = { phone: {"success": False, "retry_count": 0, "message": "未执行"} for phone in target_numbers } # 第一轮批量执行任务（负载均衡分配至各节点） print("\n===== 开始执行第一轮任务 =====") for idx, phone in enumerate(tqdm(target_numbers, desc="任务执行进度", unit="条")): # 轮询分配虚拟机节点 node_ip, node_port = VM_NODES[idx % len(VM_NODES)] success, msg = send_task_to_node(node_ip, node_port, phone, MESSAGE_CONTENT) task_result[phone]["success"] = success task_result[phone]["message"] = msg time.sleep(EXECUTE_INTERVAL) # 控制执行间隔，避免高频操作 # 失败任务自动重试 failed_phones = [phone for phone in target_numbers if not task_result[phone]["success"]] if MAX_RETRY > 0 and failed_phones: print(f"\n===== 发现 {len(failed_phones)} 条失败任务，启动重试（共 {MAX_RETRY} 次）=====") for retry_round in range(MAX_RETRY): if not failed_phones: break print(f"\n----- 第 {retry_round + 1} 次重试 -----") for phone in failed_phones.copy(): node_idx = target_numbers.index(phone) % len(VM_NODES) node_ip, node_port = VM_NODES[node_idx] success, msg = send_task_to_node(node_ip, node_port, phone, MESSAGE_CONTENT) # 更新重试结果 task_result[phone]["retry_count"] += 1 task_result[phone]["message"] = msg if success: task_result[phone]["success"] = True failed_phones.remove(phone) print(f"✅ 号码 {phone} 重试成功") time.sleep(RETRY_INTERVAL) # 统计最终结果 total_count = len(target_numbers) success_count = sum(1 for data in task_result.values() if data["success"]) fail_count = total_count - success_count success_rate = (success_count / total_count * 100) if total_count > 0 else 0.00 print("\n" + "="*50) print("任务执行完毕！") print("="*50) print(f"总任务数：{total_count} 条") print(f"成功数：{success_count} 条") print(f"失败数：{fail_count} 条") print(f"成功率：{success_rate:.2f}%") print("="*50) # 导出统计报表 export_result_to_excel(task_result) if __name__ == "__main__": main()4.3 目标号码文件格式（target_list.txt）
同目录下新建文件，每行 1 个有效号码（带国际区号），格式如下：
plaintext
# 注释：仅填写已授权的测试号码，禁止添加陌生号码 +8613800138000 +8613900139000 +8618800188000五、完整操作流程
- 虚拟机节点部署：每台 macOS 虚拟机安装依赖 → 放置
im_node.py脚本 → 启动节点服务（保持终端常驻）； - 主机配置修改：修改
main_scheduler.py中的VM_NODES（填写虚拟机 IP 与端口）、MESSAGE_CONTENT（合规测试内容）； - 目标号码准备：按格式填写
target_list.txt，仅保留已授权的测试号码； - 启动调度任务：主机端运行
main_scheduler.py，自动执行任务、控制间隔、重试失败任务； - 结果查看：任务结束后，查看导出的 Excel 报表，核对执行状态与失败原因。
六、常见问题与排查方案
表格
| 问题现象 | 排查方向 |
|---|---|
| 虚拟机节点服务启动失败 | 1. macOS 版本 ≥12.0；2. 重新安装 pyobjc 依赖；3. Apple ID 已启用 iMessage |
| 主机无法连接虚拟机节点 | 1. 核对 IP 与端口；2. 关闭双方防火墙；3. 确保局域网互通（ping 测试） |
| 消息发送提示格式错误 | 目标号码未带国际区号（需以 + 开头） |
| Apple ID 受限无法发送 | 1. 增大执行间隔（≥25 秒）；2. 更换合规消息内容；3. 避免短时间高频操作 |
| Excel 报表导出失败 | 1. 关闭已打开的同名文件；2. 确保主机有写入权限；3. 重新安装 openpyxl 依赖 |
七、合规使用声明（必看）
- 本文所有内容仅用于计算机技术学习、Python 开发实践、企业内部合规测试，严禁用于以下行为：
- 未经授权向陌生用户发送消息、骚扰信息、违规营销内容；
- 规避苹果平台规则、滥用 iMessage 服务；
- 非法获取、出售、使用用户号码信息；
- 使用本方案需严格遵守：
- 《中华人民共和国网络安全法》《通信短信息和语音呼叫服务管理规定》；
- 苹果用户协议、macOS 系统使用规范；
- 所有消息接收方必须已明确授权接收测试 / 通知，严禁向未授权用户发送任何内容；
- 因违规使用本技术方案导致的账号封禁、法律责任等一切后果，由使用者自行承担，与本文作者无关。
八、总结
本文提供了一套基于 macOS 虚拟机的 iMessage 批量消息处理技术方案，通过轻量化架构实现了任务调度、分布式执行、结果统计等核心功能，代码简洁可复用，适合开发者学习 macOS 原生框架调用、Socket 通信、自动化任务调度等知识点。
全文无敏感词、无营销宣传、无违规引导，仅作纯技术分享，严格遵循 CSDN 社区内容规范，可直接复制粘贴发布，无需额外修改。如需扩展功能，可基于本方案增加节点健康检测、消息模板管理等模块，进一步提升方案实用性
