简介：这份资源是一套基于Django框架搭建的运维平台源码，面向具备Python基础、希望了解Web运维系统开发的IT从业者与学习者。平台采用前后端不分离架构，将主机管理、服务监控、日志查看、任务调度、权限控制、资源统计、脚本执行、故障告警、审计追踪与知识库等功能整合于同一项目，适合作为小型运维工具原型或Django实战练手项目。压缩包共164个文件，约3.63MB，以39个py源码文件为核心，配合38个js、23个css及11个html构成前端界面，另有png、svg、ttf等图片与字体资源，以及少量日志、配置与说明文档，目录结构完整，便于按模块阅读。目前已有2716人学习下载。通过研读模型定义、视图函数与模板文件，读者可掌握Django项目组织方式、后台管理界面定制及前后端协作思路，并理解Linux运维场景下的功能设计逻辑，为后续二次开发或搭建自有运维平台提供可复用的参考。
1. 从一堆散落的脚本说起：这套运维平台到底要解决什么
很多团队早期的运维状态是这样的：十几台机器，每台上面散落着几十个 shell 脚本，部署靠手动 scp，重启服务靠 SSH 上去敲命令，日志要一台台登录去看。人少的时候还能扛，一旦机器数量上到三五十台，或者来了新人接手，这套东西立刻变成黑匣子——没人说得清哪台机器上跑的是什么版本，也没人敢随便动那些脚本。这个项目要解决的，就是把这种「脚本游击队」状态收敛成一个统一的 Web 运维平台，用 Django 做后端，把主机管理、任务下发、部署发布、日志查看这几件事收进一个界面里。它适合两类人：一类是正在做 Python 方向毕设或课程设计、需要一套结构完整可交付系统的学生；另一类是小团队里想自己搭一套轻量运维工具的开发者，不想上重型平台，但又受够了手工操作。
2. 需求拆解与技术选型：为什么是 Django 而不是 Flask
2.1 先把功能边界划清楚
做这类平台最容易翻车的地方，是一开始什么都想要，最后什么都没做完。我一般会把需求压到四个核心模块，其余全部砍掉或放到二期：
- 主机管理：维护被管机器的 IP、SSH 端口、登录凭据、分组标签，支持增删改查和批量导入。
- 任务执行：选一批主机，下发一条命令或一段脚本，实时或准实时拿到每台机器的执行结果。
- 应用部署：把「拉代码 → 装依赖 → 重启服务」这套流程固化成模板，一键对指定主机组执行。
- 操作审计：谁在什么时间对哪些机器做了什么操作，全部落库，出问题能回溯。
这四个模块覆盖了日常运维 80% 的高频动作，剩下的监控告警、CMDB 自动发现这些，属于锦上添花，毕设或课设阶段不建议铺开。
2.2 技术选型：Django 的取舍
选 Django 而不是 Flask，核心理由是这类平台天然需要一套完整的后台能力：用户认证、权限控制、ORM、Admin 后台、表单校验，Django 全都自带。用 Flask 你得自己拼 SQLAlchemy、Flask-Login、WTForms，拼完发现工作量比 Django 还大，而且权限模型要自己设计。Django 的 Admin 还能在开发阶段直接当临时管理界面用，省掉大量前端联调时间。
具体技术栈我一般这样定：
| 层次 | 选型 | 说明 |
|---|---|---|
| Web 框架 | Django 4.x | 自带 ORM、认证、Admin |
| 异步任务 | Celery + Redis | 命令下发不能阻塞请求 |
| 远程执行 | Paramiko | 纯 Python 实现 SSH，无需装客户端 |
| 前端 | Bootstrap + 少量 jQuery | 不引入重型前端框架，降低门槛 |
| 数据库 | MySQL 或 PostgreSQL | 生产用 PG，课设用 MySQL 也行 |
| 部署 | Gunicorn + Nginx | 标准组合 |
这里有个关键决策：命令执行必须异步化。如果直接在 Django 视图里同步跑 Paramiko，一条命令卡 30 秒，整个请求就挂 30 秒，并发一上来直接崩。所以用 Celery 把执行任务丢到队列里，视图只负责创建任务并返回任务 ID，前端轮询或走 WebSocket 拿结果。
提示：课设环境如果不想装 Redis，可以用 Django 自带的
django.core.mail那种思路，把 Celery 换成线程池，但生产环境不要这么干，线程池扛不住并发也管不了任务状态。
2.3 数据模型设计
模型设计直接决定后面写起来顺不顺。核心表我一般这样建：
# hosts/models.py from django.db import models class HostGroup(models.Model): """主机分组，用于批量操作""" name = models.CharField(max_length=64, unique=True) description = models.TextField(blank=True) def __str__(self): return self.name class Host(models.Model): """被管主机""" hostname = models.CharField(max_length=128) ip = models.GenericIPAddressField() ssh_port = models.IntegerField(default=22) username = models.CharField(max_length=64) # 密码和密钥不存明文，存加密后的密文 credential = models.TextField(help_text="加密后的登录凭据") group = models.ForeignKey(HostGroup, on_delete=models.SET_NULL, null=True, related_name="hosts") is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: unique_together = ("ip", "ssh_port") class TaskRecord(models.Model): """任务执行记录，审计用""" STATUS_CHOICES = [ ("pending", "待执行"), ("running", "执行中"), ("success", "成功"), ("failed", "失败"), ] operator = models.CharField(max_length=64) command = models.TextField() status = models.CharField(max_length=16, choices=STATUS_CHOICES, default="pending") created_at = models.DateTimeField(auto_now_add=True) finished_at = models.DateTimeField(null=True, blank=True)Host表里凭据字段用TextField存密文，加密密钥放环境变量，不要写死在代码里。TaskRecord是审计的核心，每次下发命令都落一条，operator字段记录操作人。unique_together防止同一台机器重复录入。
3. 核心模块实现：从 SSH 连接到任务下发
3.1 用 Paramiko 封装一个可靠的执行器
Paramiko 用起来简单，但坑不少。最典型的是连接超时和命令执行超时不分，导致一个卡死的命令把整个 worker 拖住。我一般会封装一个执行器类，把连接和执行分开控制：
# executor/ssh_client.py import paramiko from paramiko import SSHClient, AutoAddPolicy class SSHExecutor: def __init__(self, ip, port, username, password=None, key=None, connect_timeout=10, exec_timeout=60): self.ip = ip self.port = port self.username = username self.password = password self.key = key self.connect_timeout = connect_timeout self.exec_timeout = exec_timeout def run(self, command): """执行单条命令，返回 (exit_code, stdout, stderr)""" client = SSHClient() # 生产环境应加载 known_hosts 做校验，课设可先用 AutoAddPolicy client.set_missing_host_key_policy(AutoAddPolicy()) try: client.connect( hostname=self.ip, port=self.port, username=self.username, password=self.password, key_filename=self.key, timeout=self.connect_timeout, banner_timeout=self.connect_timeout, ) stdin, stdout, stderr = client.exec_command( command, timeout=self.exec_timeout ) exit_code = stdout.channel.recv_exit_status() out = stdout.read().decode("utf-8", errors="ignore") err = stderr.read().decode("utf-8", errors="ignore") return exit_code, out, err except Exception as e: return -1, "", str(e) finally: client.close()几个参数要重点说：connect_timeout控制 TCP 握手和认证阶段，exec_timeout控制命令执行阶段，两者分开设。recv_exit_status()会阻塞直到命令结束，所以exec_timeout必须设，否则一个tail -f就能把 worker 挂死。errors="ignore"是防止某些机器返回非 UTF-8 编码导致解码报错。
3.2 Celery 任务：把执行结果落库
执行器写好了，接下来是把它接到 Celery 里。任务函数要做三件事：更新任务状态为 running、逐台执行、把结果写回数据库。
# executor/tasks.py from celery import shared_task from django.utils import timezone from .models import TaskRecord, Host from .ssh_client import SSHExecutor @shared_task(bind=True, max_retries=2) def run_command_task(self, task_id, host_ids, command): task = TaskRecord.objects.get(id=task_id) task.status = "running" task.save(update_fields=["status"]) results = [] for host in Host.objects.filter(id__in=host_ids, is_active=True): executor = SSHExecutor( ip=host.ip, port=host.ssh_port, username=host.username, password=decrypt(host.credential), ) code, out, err = executor.run(command) results.append({ "host": host.hostname, "ip": host.ip, "exit_code": code, "stdout": out[-2000:], # 只留尾部，防止结果过大 "stderr": err[-2000:], }) task.status = "success" if all(r["exit_code"] == 0 for r in results) else "failed" task.finished_at = timezone.now() task.save(update_fields=["status", "finished_at"]) return resultsbind=True让任务能访问self，方便做重试。max_retries=2是防止网络抖动导致任务直接失败。out[-2000:]截断输出，因为有些命令输出几十兆，全存进数据库会把表撑爆。decrypt是自定义的解密函数，和存凭据时的加密函数对应。
3.3 部署模板：把流程固化下来
部署模块的本质是把一串命令按顺序执行，中间任何一步失败就中断。我一般用一个简单的模板表存命令列表，执行时按顺序跑：
# deploy/models.py class DeployTemplate(models.Model): name = models.CharField(max_length=128) # 每行一条命令，按顺序执行 steps = models.TextField(help_text="每行一条命令") target_group = models.ForeignKey("hosts.HostGroup", on_delete=models.CASCADE) def get_steps(self): return [s.strip() for s in self.steps.splitlines() if s.strip()]执行时遍历get_steps()，每条命令调一次SSHExecutor.run，exit_code != 0就 break 并记录失败步骤。这样设计的好处是模板可编辑、可复用，新人不用懂脚本也能操作。
4. 避坑与常见问题：那些年踩过的 SSH 和 Celery 的坑
4.1 现象：任务一直卡在 running，永远不结束
原因：最常见的是 Paramiko 的exec_timeout没设，或者设了但命令本身在等输入（比如sudo要密码、read等用户输入）。另一种是 Celery worker 数量不够，任务排队但没 worker 消费。
解决：所有exec_command必须带timeout参数；命令里避免交互式操作，sudo用-S从 stdin 读密码或配免密；Celery 启动时用-c指定并发数，监控队列长度，积压了就加 worker。
4.2 现象：部分机器连接成功但命令返回空
原因：目标机器的 shell 环境变量没加载。Paramiko 执行命令时默认不走 login shell，/etc/profile里的 PATH 不生效，导致java、node这些命令找不到。
解决：命令前加source /etc/profile &&，或者用bash -l -c "your command"走 login shell。我一般直接在执行器里包一层，统一处理。
4.3 现象：Celery 任务重复执行
原因：acks_late配置和 worker 崩溃的组合。默认 Celery 在任务开始前就 ack，worker 崩了任务丢失；开了acks_late后任务执行完才 ack，但 worker 执行到一半崩了，任务会被重新投递，导致重复执行。
解决：任务函数必须幂等。比如部署任务，重复执行前先检查当前版本，已经是目标版本就跳过。或者用 Redis 锁，任务开始时抢锁，抢不到直接返回。
4.4 现象：凭据加密后解密失败
原因：加密密钥换了，或者加密时用的库版本和解密时不一致。Fernet 这类库对密钥格式敏感，密钥变了旧数据全解不开。
解决：密钥放环境变量并做好备份，不要硬编码。加密库锁定版本，升级前先测试解密旧数据。生产环境建议引入密钥管理服务，课设阶段至少把密钥写进.env并加入.gitignore。
4.5 现象：批量执行时部分机器超时拖垮整体
原因：串行执行，一台机器卡 60 秒，十台就是 10 分钟。用户等不及刷新页面，以为系统挂了。
解决：用 Celery 的group或chord做并行执行，每台机器一个子任务，最后汇总结果。并行度受 worker 数量限制，但比串行快得多。前端用轮询任务状态，不要同步等待。
5. 验收与答辩：怎么证明这套平台真的能用
验收这件事，很多人栽在「功能演示很流畅，一问细节就露馅」。我一般会准备三层验证，从功能到边界到异常，逐层过。
第一层：功能闭环验证。准备三台测试机（虚拟机就行），录入平台，建一个分组。然后走一遍完整流程：下发uptime命令，确认三台都返回结果；建一个部署模板，内容是echo "deploy step 1"和echo "deploy step 2"，执行后确认两步都成功；最后查审计记录，确认操作人、时间、命令都对得上。这一层证明系统能跑通。
第二层：边界与异常验证。这是答辩时拉开差距的地方。故意关掉一台测试机的 SSH 服务，再下发命令，看平台是否优雅地返回「连接失败」而不是抛 500；故意下发一条sleep 120的命令，看是否在exec_timeout后超时返回；故意下发一条不存在的命令，看exit_code是否非零且 stderr 有内容。把这些异常场景的结果截图放进答辩 PPT，比只演示成功路径有说服力得多。
第三层：性能与并发验证。如果时间允许，用脚本模拟 20 台机器同时下发命令，观察 Celery 队列和任务完成时间。不需要做到生产级压测，但至少证明「批量操作不会把系统卡死」。可以用locust或简单的 Python 脚本发请求。
答辩时还有一个高频问题：「你的平台和 Ansible 有什么区别？」这个问题不要回避，直接说定位不同——Ansible 是命令行工具，面向熟练运维；这套平台是 Web 界面，面向团队协作和审计，把操作门槛降下来，同时留下操作记录。两者不冲突，平台底层甚至可以调 Ansible 做执行引擎，只是这个项目为了教学完整性自己实现了 SSH 层。
注意：答辩演示前一定要把测试环境跑一遍，尤其是 Celery 和 Redis 是否启动、测试机是否可达。我见过太多现场演示时 Redis 没起来、任务一直 pending 的翻车现场。
从那以后我每次做这类平台，都会在验收前强制走一遍「关服务、断网络、发错命令」这三步异常测试，确认系统不会崩、不会卡、不会丢数据。这套习惯帮我省掉了无数次答辩现场的尴尬。希望帮到你。
本文还有配套的精品资源，点击获取
