Python批量远程关机:Windows局域网运维脚本实战
简介:这份资源面向需要在Windows环境下管理多台设备的IT运维人员与Python学习者,提供一套自动远程批量关机的脚本方案,解决逐台手动关机效率低、易遗漏的问题。压缩包共3个文件,包含1个py脚本与2个txt说明文档,整体仅923B,轻量易取,脚本可直接参考或改造使用。资源围绕WMI远程管理展开,涉及通过win32com.client建立远程连接、调用Win32_ComputerSystem的Shutdown方法执行软关机或强制关机,并给出IP列表遍历实现批量处理的思路,同时提示管理员权限、目标机远程管理配置及账号密码安全存储等注意事项。已有897人学习下载,适合希望快速掌握远程批量关机实现方式、并在此基础上扩展重试与超时等健壮性处理的读者参考借鉴。
1. 批量远程关机:从手工点到脚本一键收工
机房巡检完,几十台 Windows 机器还亮着,一台台点开始菜单关机,手都酸了。这个项目就是解决这个场景的:用 Python 在 Windows 下批量远程关闭局域网内的机器,一条命令搞定整个机架。它适合运维、实验室管理、测试环境维护的人,也适合想学 Python 网络编程和 Windows 远程管理的开发者。核心思路不复杂——通过 Windows 自带的远程关机接口,配合 Python 的并发能力,把逐台操作变成批量执行。下面从技术选型、目录结构、核心代码到二次开发,完整拆一遍。
2. 技术选型与目录结构:为什么用 Python 调系统接口
2.1 为什么选 Python 而不是批处理
批处理也能调shutdown命令,但批量场景下它有三个硬伤:一是并发控制弱,只能串行等待;二是错误处理粗糙,某台机器失败后很难拿到具体原因;三是配置管理麻烦,机器列表、超时、重试策略散在脚本里不好维护。Python 的优势在于能用subprocess调系统命令的同时,用concurrent.futures做并发,用logging记录每台机器的执行结果,用配置文件管理目标列表。常见做法是把机器列表放在一个文本文件或 YAML 里,脚本读取后并发执行,最后汇总成功和失败清单。
2.2 目录结构长什么样
一个能直接跑的版本,目录大概是这样:
remote_shutdown/ ├── config/ │ └── hosts.txt # 目标机器列表,每行一个 IP 或主机名 ├── logs/ │ └── shutdown_20250101.log ├── src/ │ ├── __init__.py │ ├── shutdown.py # 核心关机逻辑 │ ├── config_loader.py # 读取配置和机器列表 │ └── logger.py # 日志配置 ├── main.py # 入口 └── requirements.txthosts.txt里就是纯文本,一行一个目标:
192.168.1.101 192.168.1.102 192.168.1.103这种结构的好处是机器列表和代码分离,换环境只改hosts.txt,不用动代码。logs目录按日期生成日志文件,出问题能回溯是哪台机器、什么时间、报了什么错。
2.3 核心依赖和前置条件
Python 侧不需要额外装第三方库,subprocess、concurrent.futures、logging都是标准库。但 Windows 侧有几个前置条件必须满足,否则脚本跑不通:
- 目标机器必须开启远程注册表服务或远程关机权限。Windows 默认允许管理员远程关机,但普通用户不行。
- 执行脚本的机器和目标机器要在同一局域网,或者能路由到。
- 如果目标机器开了防火墙,需要放行远程管理相关的端口。常见做法是在域环境里用组策略统一配置,非域环境就手动在目标机器上执行一次授权命令。
授权命令在目标机器上执行一次即可:
net use \\192.168.1.101\ipc$ /user:Administrator这条命令建立 IPC 连接,后续shutdown命令才能远程生效。参数说明:\\IP\ipc$是目标机器的 IPC 共享,/user:后面跟管理员账号。执行后会提示输入密码,输入正确即建立连接。
注意:如果目标机器是 Windows 家庭版,远程关机接口可能不可用,建议先在一台机器上手动测试
shutdown /s /m \\IP /t 0是否能成功。
3. 核心模块拆解:从读取机器列表到并发关机
3.1 读取机器列表和配置
config_loader.py负责把hosts.txt读成一个列表,同时做一些基础校验,比如去掉空行、去掉注释行、检查 IP 格式是否合法。
# config_loader.py import re def load_hosts(filepath): """读取机器列表,返回去重后的目标列表""" hosts = [] with open(filepath, 'r', encoding='utf-8') as f: for line in f: line = line.strip() # 跳过空行和以 # 开头的注释行 if not line or line.startswith('#'): continue # 简单校验:IP 或主机名格式 if re.match(r'^(\d{1,3}\.){3}\d{1,3}$', line) or re.match(r'^[a-zA-Z0-9\-\.]+$', line): hosts.append(line) else: print(f"跳过非法目标: {line}") # 去重但保持顺序 seen = set() unique_hosts = [] for h in hosts: if h not in seen: seen.add(h) unique_hosts.append(h) return unique_hosts逻辑说明:逐行读取,strip()去掉首尾空白,跳过空行和注释。正则校验部分,IP 格式用四段数字匹配,主机名允许字母数字和连字符。去重时用set记录已见过的目标,但最终返回的列表保持原始顺序,方便日志对照。参数filepath是hosts.txt的路径,一般从main.py传进来。
3.2 单台机器的关机函数
核心关机动作封装成一个函数,接收目标地址和超时时间,返回执行结果。这里用subprocess.run调 Windows 的shutdown命令。
# shutdown.py import subprocess def shutdown_single(host, timeout=10): """ 关闭单台远程机器 host: IP 或主机名 timeout: 命令超时秒数 返回: (host, success, message) """ cmd = ['shutdown', '/s', '/m', f'\\\\{host}', '/t', '0', '/f'] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout ) if result.returncode == 0: return (host, True, '关机指令已发送') else: return (host, False, result.stderr.strip() or '未知错误') except subprocess.TimeoutExpired: return (host, False, f'超时({timeout}秒)') except Exception as e: return (host, False, str(e))逻辑说明:shutdown命令的参数含义——/s表示关机,/m \\host指定远程目标,/t 0表示立即执行不延迟,/f强制关闭正在运行的程序不弹窗。capture_output=True捕获标准输出和错误,text=True让输出以字符串返回而不是字节。timeout防止某台机器无响应导致整个脚本卡死。返回值统一成三元组,方便后续汇总。
参数怎么改:如果希望给用户留保存时间,把/t 0改成/t 60,表示 60 秒后关机。如果不想强制关闭程序,去掉/f,但这样有程序未保存时会弹窗阻塞,远程场景下不推荐。
3.3 并发调度和结果汇总
单台能关,接下来用线程池并发执行。concurrent.futures.ThreadPoolExecutor适合这种 IO 密集型任务,因为主要时间花在等待远程命令返回,不是 CPU 计算。
# main.py import concurrent.futures from src.config_loader import load_hosts from src.shutdown import shutdown_single from src.logger import setup_logger def main(): logger = setup_logger() hosts = load_hosts('config/hosts.txt') if not hosts: logger.warning('机器列表为空,退出') return logger.info(f'开始批量关机,共 {len(hosts)} 台目标') success, failed = [], [] # 最大并发数设为 10,避免瞬间大量连接被防火墙拦截 with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: future_map = { executor.submit(shutdown_single, host): host for host in hosts } for future in concurrent.futures.as_completed(future_map): host, ok, msg = future.result() if ok: success.append(host) logger.info(f'[成功] {host} - {msg}') else: failed.append(host) logger.error(f'[失败] {host} - {msg}') logger.info(f'执行完成:成功 {len(success)} 台,失败 {len(failed)} 台') if failed: logger.info(f'失败列表:{", ".join(failed)}') if __name__ == '__main__': main()逻辑说明:ThreadPoolExecutor的max_workers=10控制同时执行的任务数。为什么不设更大?因为 Windows 远程管理对并发连接数有隐性限制,瞬间几十个连接可能被目标机器的防火墙或安全策略拦截,反而导致大面积失败。as_completed按完成顺序返回结果,不用等所有任务都结束才处理。成功和失败分别记录,最后汇总输出。
参数怎么改:max_workers根据网络质量和目标机器数量调整。局域网内 50 台以内,10 到 20 比较稳。跨网段或网络质量差,降到 5 以下。timeout在shutdown_single里默认 10 秒,如果目标机器响应慢,可以调到 20 或 30。
3.4 日志模块
日志不是可有可无的装饰,批量操作出问题时,日志是唯一的后悔药。
# logger.py import logging import os from datetime import datetime def setup_logger(): log_dir = 'logs' os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, f'shutdown_{datetime.now().strftime("%Y%m%d")}.log') logger = logging.getLogger('remote_shutdown') logger.setLevel(logging.INFO) # 文件处理器:记录所有级别 fh = logging.FileHandler(log_file, encoding='utf-8') fh.setLevel(logging.INFO) # 控制台处理器:只显示 INFO 以上 ch = logging.StreamHandler() ch.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s') fh.setFormatter(formatter) ch.setFormatter(formatter) logger.addHandler(fh) logger.addHandler(ch) return logger逻辑说明:日志同时输出到文件和控制台。文件按日期命名,每天一个文件,方便回溯。encoding='utf-8'避免中文路径或消息乱码。格式里带时间戳和级别,排查时能精确到秒。常见做法是再加一个RotatingFileHandler做日志轮转,防止单文件过大,但小规模场景下按日期分文件已经够用。
4. 避坑与排查:远程关机失败的常见原因
4.1 提示“拒绝访问”或“找不到网络路径”
现象:脚本返回失败,错误信息是“拒绝访问”或“系统找不到指定的网络路径”。原因通常是执行脚本的账号在目标机器上没有关机权限,或者 IPC 连接没建立。解决:先在命令行手动执行net use \\目标IP\ipc$ /user:管理员账号,输入密码建立连接,再跑脚本。如果是域环境,用域管理员账号执行脚本即可。非域环境建议在目标机器上把执行账号加入 Administrators 组。
4.2 部分机器成功部分失败
现象:同一批机器,有的能关,有的报超时。原因一般是网络不通、目标机器防火墙拦截、或者目标机器已经处于关机状态。解决:先用ping确认目标是否在线,再用shutdown /s /m \\IP /t 0手动测一台失败的机器,看具体报错。如果手动也失败,检查目标机器的防火墙入站规则,放行“远程管理”相关规则。常见做法是在目标机器上执行一次netsh advfirewall firewall set rule group="remote administration" new enable=yes。
4.3 脚本卡住不返回
现象:执行后长时间无输出,也不结束。原因通常是某台机器网络不可达,subprocess在等待超时。解决:确保shutdown_single里的timeout参数生效,并且ThreadPoolExecutor的max_workers不要设得过大。如果还是卡,可以在subprocess.run外面再包一层信号超时,但标准库的timeout参数在 Windows 上对shutdown命令是有效的,一般不需要额外处理。
4.4 关机后机器又自动重启
现象:机器确实关了,但过一会儿又自己开机。原因可能是目标机器开了网络唤醒(Wake-on-LAN),或者有定时任务触发。解决:检查目标机器的 BIOS 和网卡设置,关闭网络唤醒。如果是虚拟机,检查宿主机的电源管理策略。这个坑比较隐蔽,日志里看不出问题,需要到目标机器上排查。
4.5 中文主机名或路径导致乱码
现象:日志里中文显示成乱码,或者读取hosts.txt时报编码错误。原因:Windows 默认编码可能是 GBK,而脚本用 UTF-8 读取。解决:open时显式指定encoding='utf-8',日志的FileHandler也指定encoding='utf-8'。如果hosts.txt本身是 GBK 编码,用记事本另存为 UTF-8 即可。
5. 二次开发:从批量关机到远程任务调度
5.1 扩展成远程重启或注销
关机逻辑改一个参数就能变成重启。shutdown命令的/s换成/r是重启,换成/l是注销。把shutdown_single里的命令参数抽成变量,就能支持多种操作:
def remote_action(host, action='shutdown', timeout=10): action_map = { 'shutdown': ['/s'], 'restart': ['/r'], 'logoff': ['/l'], } if action not in action_map: return (host, False, f'不支持的操作: {action}') cmd = ['shutdown'] + action_map[action] + ['/m', f'\\\\{host}', '/t', '0', '/f'] # 后续 subprocess 调用同上这样改完,入口传一个action参数就能切换行为。注意/l注销不支持/m远程参数,只能在本地用,远程注销需要其他方案。
5.2 加入重试机制
网络抖动导致单次失败很常见,加一个简单的重试能显著提高成功率:
def shutdown_with_retry(host, retries=2, timeout=10): for i in range(retries + 1): result = shutdown_single(host, timeout) if result[1]: return result if i < retries: time.sleep(2 ** i) # 指数退避:2秒、4秒 return result逻辑说明:retries=2表示最多重试两次,加上首次执行共三次。time.sleep(2 ** i)做指数退避,第一次等 2 秒,第二次等 4 秒,避免密集重试加重网络负担。这个模式在远程调用场景里很实用,我一般会把它作为默认配置。
5.3 用配置文件管理多组目标
如果机器分多个机架或用途,可以在config下按组建多个文件,比如hosts_web.txt、hosts_db.txt,入口根据参数选择加载哪个。再进一步,用 YAML 配置文件把目标、超时、并发数、重试次数都写进去:
# config/settings.yaml groups: web: hosts_file: config/hosts_web.txt max_workers: 10 timeout: 15 retries: 2 db: hosts_file: config/hosts_db.txt max_workers: 5 timeout: 30 retries: 3读取 YAML 需要装pyyaml,但换来的是不用改代码就能调整策略。对于经常要批量操作的场景,这个投入值得。
5.4 验证改造是否生效
改完代码后,不要直接对生产机器跑。先在一台测试机上验证:把hosts.txt里只留一个测试 IP,跑一遍看日志输出是否符合预期。确认单台没问题后,再加到三台,观察并发和重试是否按配置执行。最后才全量跑。我自己的习惯是每次改完关机脚本,都先拿一台不重要的机器走一遍完整流程,确认日志里成功和失败的分支都覆盖到了,再动真格。从那以后我每次改批量操作脚本,都强制走一遍“单台→三台→全量”的验证流程,省了不少半夜爬起来处理故障的时间。希望帮到你。
本文还有配套的精品资源,点击获取