Python iMessage 虚拟机群发系统:批量发送 + 进度条 + Excel 报告完整实现方案(2026 实测)
🔥 文章简介(平台自动抓取,超吸睛)
TG:@iosxiaoluo
全网最完整、最稳、可直接落地的 iMessage 群发实战教程!多虚拟机分布式发送 + 批量号码导入 + 失败自动重试 + 实时进度条 + Excel 统计报告一键导出。代码完整可运行,环境搭建 + 部署 + 发送 + 排错一步到位,新手照做就能成功!
🔴 正文开始(直接全选复制发布)
大家好,本篇是2026 最新实测可用的 iMessage 虚拟机群发完整实战文档。很多人找了大量教程,却要么代码跑不通,要么功能残缺,要么没有批量导入、没有报表、没有进度条。
这篇直接给你一套完整可落地系统:多虚拟机分流、号码批量导入、失败自动重试、实时发送进度、Excel 报告导出。全程复制粘贴即可运行,保姆级步骤,零基础也能直接上手。
⚠️ 合规声明:本文仅限企业合法通知、会员服务、自身测试等合规场景使用。严禁用于垃圾短信、骚扰营销、诈骗等用途,违者后果自负,Apple ID 会被永久封禁。
📋 一、环境准备(5 分钟搞定)
硬件 / 软件要求
- 主机:Windows /macOS,内存 ≥16G,CPU ≥4 核
- 虚拟机:VMware 17+ / Parallels 18+
- 虚拟机系统:macOS 12.0+
- 账号:正常可用、已开启 iMessage 的 Apple ID
- 网络:主机与虚拟机在同一局域网
依赖安装
虚拟机端执行:
pip3 install pyobjc --upgrade主机端执行:
pip3 install pyobjc tqdm openpyxl --upgrade🖥️ 二、虚拟机端:iMessage 发送服务
新建文件:imessage_server.py
直接复制以下全部代码:
import objc import socket from Foundation import NSURL from Messages import * def load_imessage_framework(): 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_single_imessage(phone_number, message_content): if not phone_number.startswith("+"): return False, "号码格式错误:需带国家码(示例+86138xxxx8888)" if not load_imessage_framework(): return False, "iMessage框架加载失败" try: recipient_url = NSURL.URLWithString_(f"tel:{phone_number}") if not recipient_url: return False, "收件人URL构建失败" message_request = MSMessageRequest.alloc().init() message_request.setRecipients_([recipient_url]) message_request.setMessageText_(message_content) message_request.sendSynchronouslyWithError_(None) return True, "发送成功" except Exception as e: return False, f"发送失败:{str(e)}" def start_socket_server(host="0.0.0.0", port=8888): if not load_imessage_framework(): return server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((host, port)) server_socket.listen(5) print(f"✅ iMessage发送服务已启动,监听:{host}:{port}") while True: try: client_conn, client_addr = server_socket.accept() task_data = client_conn.recv(1024).decode("utf-8").strip() if "|" not in task_data: client_conn.send(b"False|指令格式错误") client_conn.close() continue phone, content = task_data.split("|", 1) send_result, send_msg = send_single_imessage(phone, content) client_conn.send(f"{send_result}|{send_msg}".encode("utf-8")) client_conn.close() print(f"📝 {phone} - {send_msg}") except Exception as e: print(f"❌ 通信异常:{str(e)}") continue if __name__ == "__main__": start_socket_server()启动服务
cd ~/Desktop python3 imessage_server.py出现:
✅ iMessage发送服务已启动,监听:0.0.0.0:8888
即为成功。
🎛️ 三、主机端:群发调度(含进度条 + Excel 导出)
新建文件:sender.py直接复制以下全部代码:
import socket import time import os from typing import List, Tuple, Dict from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from tqdm import tqdm # ====================== 你只需要改这里 ====================== VM_NODES: List[Tuple[str, int]] = [ ("192.168.1.101", 8888), # ("192.168.1.102", 8888), ] PHONE_FILE_PATH: str = "phone_list.txt" MESSAGE_CONTENT: str = "【测试】iMessage群发系统正常发送" SEND_INTERVAL: int = 15 RETRY_MAX: int = 2 RETRY_INTERVAL: int = 60 EXCEL_REPORT_PATH: str = "iMessage发送报告.xlsx" # ========================================================== def read_phones_from_file(file_path: str) -> List[str]: if not os.path.exists(file_path): print(f"❌ 号码文件不存在") return [] with open(file_path, "r", encoding="utf-8") as f: phones = [line.strip() for line in f if line.strip() and line.strip().startswith("+")] print(f"📥 读取有效号码:{len(phones)} 个") return phones def send_task_to_vm(vm_ip: str, vm_port: int, phone: str, content: str) -> Tuple[bool, str]: try: client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.settimeout(10) client_socket.connect((vm_ip, vm_port)) client_socket.send(f"{phone}|{content}".encode()) res = client_socket.recv(1024).decode().strip() client_socket.close() if res.startswith("True"): return True, res.split("|",1)[1] if "|" in res else "成功" else: return False, res.split("|",1)[1] if "|" in res else "失败" except Exception as e: return False, f"异常:{str(e)}" def export_to_excel(send_results: Dict[str, Dict], msg: str): wb = Workbook() ws = wb.active ws.title = "发送明细" h_font = Font(bold=True, color="FFFFFF") h_fill = PatternFill("solid", "366092") s_fill = PatternFill("solid", "E6FFE6") f_fill = PatternFill("solid", "FFE6E6") headers = ["序号", "手机号", "状态", "重试次数", "结果", "内容"] for i, v in enumerate(headers, 1): c = ws.cell(1, i, v) c.font = h_font c.fill = h_fill c.alignment = Alignment(horizontal="center") for row, (p, d) in enumerate(send_results.items(), 2): ws.cell(row,1,row-1) ws.cell(row,2,p) stat = "成功" if d["success"] else "失败" ws.cell(row,3,stat) ws.cell(row,4,d["retry_count"]) ws.cell(row,5,d["msg"]) ws.cell(row,6,msg) c = ws.cell(row,3) c.fill = s_fill if d["success"] else f_fill c.alignment = Alignment(horizontal="center") widths = [8,16,10,12,32,50] for i, w in enumerate(widths,1): ws.column_dimensions[chr(64+i)].width = w st = wb.create_sheet("统计") total = len(send_results) ok = sum(1 for x in send_results.values() if x["success"]) ng = total - ok rate = f"{ok/total*100:.2f}%" if total>0 else "0%" stat_data = [ ["总发送", total],["成功",ok],["失败",ng],["成功率",rate], ["发送内容",msg],["间隔",f"{SEND_INTERVAL}s"], ["最大重试",RETRY_MAX],["时间",time.strftime("%Y-%m-%d %H:%M")] ] for r, (k,v) in enumerate(stat_data,1): st.cell(r,1,k) st.cell(r,2,v) if r==1: st.cell(r,1).font=h_font st.cell(r,1).fill=h_fill st.cell(r,2).font=h_font st.cell(r,2).fill=h_fill st.column_dimensions["A"].width=16 st.column_dimensions["B"].width=50 wb.save(EXCEL_REPORT_PATH) print(f"\n📊 报告已导出:{os.path.abspath(EXCEL_REPORT_PATH)}") def main(): phones = read_phones_from_file(PHONE_FILE_PATH) or ["+8613800138000"] if not phones: print("❌ 无号码") return print("\n🚀 iMessage 批量发送开始") print(f"总号码:{len(phones)} | 虚拟机:{len(VM_NODES)}") res = {p:{"success":False,"retry_count":0,"msg":""} for p in phones} print("\n📤 第一轮发送") for i, p in enumerate(tqdm(phones,desc="发送进度",unit="条")): ip, port = VM_NODES[i % len(VM_NODES)] ok, info = send_task_to_vm(ip, port, p, MESSAGE_CONTENT) res[p]["success"]=ok res[p]["msg"]=info time.sleep(SEND_INTERVAL) retry_list = [p for p in phones if not res[p]["success"]] if RETRY_MAX>0 and retry_list: print(f"\n🔄 失败重试:{len(retry_list)} 条") pb = tqdm(total=len(retry_list)*RETRY_MAX, desc="重试进度",unit="次") while retry_list and res[retry_list[0]]["retry_count"] < RETRY_MAX: for p in retry_list.copy(): cnt = res[p]["retry_count"]+1 if cnt>RETRY_MAX: retry_list.remove(p) pb.update(1) continue i = phones.index(p) % len(VM_NODES) ip,port = VM_NODES[i] ok,info = send_task_to_vm(ip,port,p,MESSAGE_CONTENT) res[p]["retry_count"]=cnt res[p]["msg"]=info if ok: res[p]["success"]=True retry_list.remove(p) tqdm.write(f"✅ {p} 重试成功") pb.update(1) time.sleep(RETRY_INTERVAL) pb.close() total = len(phones) ok_cnt = sum(1 for x in res.values() if x["success"]) ng_cnt = total-ok_cnt print(f"\n🎉 发送完成") print(f"总计:{total} | 成功:{ok_cnt} | 失败:{ng_cnt}") fail_p = [p for p,x in res.items() if not x["success"]] if fail_p: print(f"❌ 失败号码:{', '.join(fail_p)}") export_to_excel(res, MESSAGE_CONTENT) if __name__ == "__main__": main()📄 四、批量号码文件(新建)
新建:phone_list.txt
+8613800138000 +8613900139000 +12025550101🚀 五、启动发送
python3 sender.py你将看到:
- 实时进度条
- 失败自动重试
- 最终统计
- 自动生成 Excel 彩色报告
❌ 六、常见问题速查
- 连不上虚拟机:检查 IP、关闭防火墙
- 发送失败:iMessage 未激活、号码不带国家码
- 依赖报错:重新执行 pip 安装命令
- ID 受限:降低频率,间隔 ≥15 秒,不要发广告
📌 七、总结
这套是2026 全网最完整、最稳、可直接落地的 iMessage 群发方案:
- 多虚拟机分布式发送
- 号码批量导入
- 失败自动重试
- 实时进度条
- Excel 彩色报告导出
代码全部可直接运行,零基础也能一步到位。
